2-LangChain Agent示例代码

雨师@

234人浏览 · 2026-06-22 18:37:49

雨师@ · 2026-06-22 18:37:49 发布

一、Agent 基础定义与核心原理

1.1 什么是 Agent

Agent 是 LangChain 最核心组件，拥有自主决策能力，能够自动判断是否调用外部工具，获取工具返回数据后持续迭代思考，循环执行直至完整解决用户任务。

普通 LLM/Chain 属于被动固定流程：用户提问 → 模型单次回答 → 流程结束； Agent 属于主动动态闭环：模型自主判断、自选工具、循环执行，自动拆解复杂任务。

1.2 Agent 标准工作循环（ReAct）

遵循「思考 - 行动 - 观察」循环逻辑：

思考：分析用户需求，判断是否需要工具、选用哪个工具、需要哪些参数
行动：调用对应工具函数，执行查询、计算等外部操作
观察：接收工具返回的实时、真实数据
迭代推理：结合新数据重新思考，按需重复调用工具
输出答案：信息充足后生成最终回复，终止循环

1.3 普通大模型 VS Agent 对比

表格

对比项	普通大模型	LangChain Agent
执行流程	一问一答，单次结束	思考 - 行动 - 观察循环，多轮迭代
数据来源	仅依赖训练静态知识库，信息滞后	可调用工具获取实时外部数据
任务处理	无法自主拆分多步骤复杂问题	自动拆解任务，自主多步骤执行
扩展能力	无外部工具调用能力	可对接搜索、接口、数据库、计算器等工具

1.4 业务示例

提问：今天杭州天气怎么样？

普通模型：只能输出训练截止前的过时历史天气，结果不准确；
Agent：自动识别需要实时数据 → 调用天气查询工具 → 获取当日真实气温、湿度 → 整理实时结果回复。

1.5 Agent 核心价值

弥补大模型知识库滞后缺陷，支持实时获取外部信息；
自动拆解复杂多步骤任务，无需人工拆分执行流程；
无限拓展 AI 能力边界，对接任意自定义工具；
全流程自主运行，无需人工干预中间步骤。

1.6 Agent 与 Chain 本质区别

Chain：固定预设流程，人为规定执行顺序，无法动态调整步骤；
Agent：动态自适应流程，仅下发任务，模型自主选择工具与执行顺序。

二、创建第一个 Agent（三大标准步骤）

搭建基础 Agent 仅需 3 步，同时支持同步、异步两种运行模式，适配不同开发场景。

定义工具：使用 @tool 装饰器封装可调用函数；
构建 Agent：传入大模型 + 工具列表完成初始化；
运行 Agent：分为同步 invoke()、异步 ainvoke() 两种调用方式。

2.1 步骤 1：定义工具函数

通过 @tool 装饰器将普通 Python 函数转为 Agent 可识别工具：

自动读取函数名、参数、文档注释作为工具描述；
大模型依靠描述判断工具用途、调用时机、入参格式；
是 Agent 扩展外部能力的唯一入口。

2.2 步骤 2：创建 Agent

初始化依赖两大核心组件：

LLM 大模型：Agent 的决策大脑，负责推理、判断、输出指令；
tools 工具列表：配置所有可供模型调用的外部工具。框架自动封装提示词、循环执行逻辑，无需手动编写思考规则。

2.3 步骤 3：两种运行模式

同步运行 invoke ()
- 特点：阻塞式顺序执行；
- 适用场景：本地脚本、简单测试、离线工具。
异步运行 ainvoke ()
- 特点：非阻塞，不占用主线程；
- 适用场景：Web 后端、接口服务、高并发线上项目。

2.4 Agent 完整执行链路

接收用户提问消息；
LLM 分析需求，判断是否需要调用工具；
如需工具，自动匹配工具、生成参数并执行；
接收工具返回观测结果；
结合结果二次推理，循环判断；
信息完备，输出最终回答，结束流程。

三、全套可运行代码示例

示例 1：基础同步 Agent（打印完整消息链路）

适合新手入门，完整展示工具调用全过程消息记录

python

运行

import os
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

# 加载环境变量
loaded = load_dotenv()
if not loaded:
    print("警告：.env 文件未找到")

from langchain.tools import tool
# 步骤1：使用@tool装饰器定义工具
@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气情况。
    Args:
        city: 城市名称，如 "杭州"、"北京"
    """
    weather_data = {
        "杭州": "晴，25°C，湿度 60%",
        "北京": "多云，18°C，湿度 45%",
        "上海": "小雨，22°C，湿度 80%",
    }
    return weather_data.get(city, f"未找到 {city} 的天气数据")

@tool
def calculate(expression: str) -> str:
    """执行数学计算。支持加减乘除等基本运算。
    Args:
        expression: 数学表达式，如 "3 * 7 + 2"
    """
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"计算结果: {expression} = {result}"
    except Exception as e:
        return f"计算错误: {e}"

# 步骤2：创建Agent
from langchain.agents import create_agent
from langchain.messages import HumanMessage

# 初始化大模型
model = ChatOpenAI(
    model="hunyuan-lite",
    api_key=os.getenv("HUNYUAN_API_KEY"),
    base_url="https://api.hunyuan.cloud.tencent.com/v1",
    temperature=0.3,
    streaming=False,
)

# 组装Agent
agent = create_agent(
    model=model,
    tools=[get_weather, calculate],
    system_prompt="你是一个乐于助人的助手，会使用工具来回答问题。",
)

# 步骤3：同步运行Agent
inputs = {"messages": [HumanMessage(content="杭州今天天气怎么样？")]}
result = agent.invoke(inputs)

# 打印完整消息流转记录
print("=== 完整消息历史 ===")
for msg in result["messages"]:
    print(f"[{msg.type}] {msg.content[:100]}")

print("\n=== 最终回复 ===")
print(result["messages"])

示例 2：封装通用同步调用函数（批量测试多问题）

封装统一提问方法，支持批量测试单一 / 多工具、无工具类问题

python

运行

import os
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
load_dotenv()

from langchain.tools import tool
from langchain.agents import create_agent
from langchain.messages import HumanMessage

# 定义工具
@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气情况。
    Args:
        city: 城市名称，如 "杭州"、"北京"
    """
    weather_data = {
        "杭州": "晴，25°C，湿度 60%",
        "北京": "多云，18°C，湿度 45%",
        "上海": "小雨，22°C，湿度 80%",
    }
    return weather_data.get(city, f"未找到 {city} 的天气数据")

@tool
def calculate(expression: str) -> str:
    """执行数学计算。支持加减乘除等基本运算。
    Args:
        expression: 数学表达式，如 "3 * 7 + 2"
    """
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"计算结果: {expression} = {result}"
    except Exception as e:
        return f"计算错误: {e}"

# 初始化模型与Agent
model = ChatOpenAI(
    model="hunyuan-lite",
    api_key=os.getenv("HUNYUAN_API_KEY"),
    base_url="https://api.hunyuan.cloud.tencent.com/v1",
    temperature=0.3,
    streaming=False,
)

agent = create_agent(
    model=model,
    tools=[get_weather, calculate],
    system_prompt="你是一个乐于助人的助手，会使用工具来回答问题。",
)

# 封装通用提问方法
def ask(question: str):
    inputs = {"messages": [HumanMessage(content=question)]}
    result = agent.invoke(inputs)
    print(f"问题: {question}")
    print(f"回答: {result['messages'][-1].content}")
    print("-" * 50)
    return result

# 批量测试场景
ask("杭州今天天气怎么样？")
ask("杭州和北京今天温差多少度？")
ask("c语言有必要学习吗？")

示例 3：同步 + 异步完整代码（线上服务专用）

同时实现同步、异步调用，适配 Web、高并发项目

python

运行

import os
import asyncio
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

from langchain.tools import tool
from langchain.agents import create_agent
from langchain.messages import HumanMessage

# 定义工具
@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气情况。
    Args:
        city: 城市名称，如 "杭州"、"北京"
    """
    weather_data = {
        "杭州": "晴，25°C，湿度 60%",
        "北京": "多云，18°C，湿度 45%",
        "上海": "小雨，22°C，湿度 80%",
    }
    return weather_data.get(city, f"未找到 {city} 的天气数据")

@tool
def calculate(expression: str) -> str:
    """执行数学计算。支持加减乘除等基本运算。
    Args:
        expression: 数学表达式，如 "3 * 7 + 2"
    """
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"计算结果: {expression} = {result}"
    except Exception as e:
        return f"计算错误: {e}"

# 初始化模型与Agent
model = ChatOpenAI(
    model="hunyuan-lite",
    api_key=os.getenv("HUNYUAN_API_KEY"),
    base_url="https://api.hunyuan.cloud.tencent.com/v1",
    temperature=0.3,
    streaming=False,
)

agent = create_agent(
    model=model,
    tools=[get_weather, calculate],
    system_prompt="你是一个乐于助人的助手，会使用工具来回答问题。",
)

# 同步调用函数
def ask(question: str):
    inputs = {"messages": [HumanMessage(content=question)]}
    result = agent.invoke(inputs)
    print(f"[同步] 问题: {question}")
    print(f"[同步] 回答: {result['messages'][-1].content}")
    print("-" * 50)
    return result

# 异步调用函数
async def ask_async(question: str):
    inputs = {"messages": [HumanMessage(content=question)]}
    result = await agent.ainvoke(inputs)
    print(f"[异步] 问题: {question}")
    print(f"[异步] 回答: {result['messages'][-1].content}")
    print("-" * 50)
    return result

# 异步执行入口
async def main():
    await ask_async("杭州今天天气怎么样？")
    await ask_async("杭州和北京今天温差多少度？")
    await ask_async("c语言有必要学习吗？")

if __name__ == "__main__":
    # 同步测试
    print("=" * 50)
    print("同步调用测试")
    print("=" * 50)
    ask("杭州今天天气怎么样？")
    ask("杭州和北京今天温差多少度？")

    # 异步测试
    print("=" * 50)
    print("异步调用测试")
    print("=" * 50)
    asyncio.run(main())

AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

零成本在终端里搭建 AI 编程智能体（Open Interpreter + 智谱免费模型）

作为嵌入式开发者，我一直在寻找一种能的 AI 工具。Claude Code 虽然强大，但需要付费 API 且国内网络不稳定；GitHub Copilot 依赖 IDE，不够灵活。后来我发现了——一个开源的终端 AI Agent，支持多种大模型后端。配合智谱 AI 的免费模型，我实现了，能读写文件、执行 Shell 命令、管理 QEMU 虚拟机，极大提升了内核和驱动学习效率。这篇文章记录完整的搭建过

AI Agent技术社区

如何成为一名 AI Agent 工程师：从入门到精通的完整指南

AI Agent 工程师是专注于设计、开发和部署智能代理（Intelligent Agent）的专业技术人员。AI Agent 是一种能够感知环境、自主决策并执行行动以实现特定目标的软件实体。与传统的 AI 模型（如分类器、生成器）不同，AI Agent 强调在动态环境中的持续交互、规划与学习能力。自动化工作流：构建能够处理复杂、多步骤任务的自动化助手，如数据分析、报告生成、客户服务等。游戏与模拟