一、理解 Agent:什么是智能体?

通用人工智能(AGI)是 AI 的终极形态,这几乎已成为业界共识。而在工程应用层面,构建智能体(Agent)就是当下的"终极形态",它是大模型应用开发的核心。

1.1 定义

在大模型应用开发中,智能体通常指一种以大语言模型为推理与决策核心,结合记忆、工具调用与环境交互能力,能够进行规划决策并执行复杂任务以达成目标的软件系统。

Agent 的关键能力可拆解为几个问题:

  • 如何理解用户问题

  • 如何拆解任务

  • 如何判断是否需要工具

  • 需要调用哪些工具

  • 如何利用好工具结果生成回答、推进任务

1.2 核心组件

AI Agent 的几个组成要素,在实际开发中并不需要同时出现:

要素

重要程度

行动(Action)

必须的

工具(Tool)

几乎总是存在的

规划决策(Planning)

有条件存在

记忆(Memory)

最容易被省略

1.3 创建与调用的演进:从"碎片化"到"统一入口"

① LangChain 0.x 时代的"碎片化"

旧版的设计理念是"针对场景设计特定 Agent",于是有了三个互相割裂的函数:

  • 实现思维链推理(ReAct)→ create_react_agent

  • 需要结构化输出 → create_structured_chat_agent

  • 要工具调用 → create_tool_calling_agent

一个典型的 v0.x 调用需要 初始化模型 → 建提示词模板 → 创建 agent → 创建 executor → 调用,整整五个步骤:

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.prompts import PromptTemplate
​
model = ChatOpenAI(model="gpt-4o-mini")
prompt = PromptTemplate.from_template("""
You are a helpful assistant.
Tools: {tools}
Tool Names: {tool_names}
{agent_scratchpad}
""")
agent = create_react_agent(llm=model, tools=tools, prompt=prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({"input": "问题"})

这种方式虽然灵活,却带来三个明显问题:

  1. 心智负担高 —— 每种 Agent 都要单独记忆 API 与参数;

  2. 可组合性差 —— 多个 Agent 之间无法统一调度;

  3. 生态碎片化 —— 不同模块难以复用或协同演化。

② LangChain 1.0 以后的"统一入口"

1.0 版本做了彻底重构,将所有 Agent 创建方式统一为一个函数 create_agent(),一行代码即可创建任何类型的智能体:

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
​
model = init_chat_model("gpt-4o-mini", model_provider="openai")
agent = create_agent(
    model=model,
    tools=[tool1, tool2],
    system_prompt="Agent 的行为指令"  # 可选
)
result = agent.invoke({"messages": [{"role": "user", "content": "问题"}]})

底层通过中间件机制(Middleware)和标准模型接口(invoke / stream)实现全局统一,让框架更轻、更稳、更易被集成。


二、基本用法①:模型的传入方式

在 LangChain 1.2 中,create_agent 底层基于 LangGraph 实现。它的完整参数如下:

agent = create_agent(
    model,                    # 必需:聊天模型(str 或 BaseChatModel)
    tools,                    # 必需:工具列表
    *,
    system_prompt="",         # 系统提示词
    middleware=(),            # 中间件
    interrupt_before=None,    # 在某些工具前暂停(人机协作)
    interrupt_after=None,     # 在某些工具后暂停
    debug=False,              # 调试模式
    name=None,                # 设置模型名称
)

模型是 Agent 的"大脑",负责决策和推理。传入方式有两种:

2.1 传入模型字符串

Agent 会根据传入的字符串自主创建模型对象:

from langchain.agents import create_agent
from dotenv import load_dotenv
​
load_dotenv(override=True)
​
agent = create_agent("deepseek-v4-flash")
print(type(agent))  # <class 'langgraph.graph.state.CompiledStateGraph'>

关键洞察:agent 本质上是 LangGraph 的 CompiledStateGraph 实例,底层是一张图结构。可以用 agent.get_graph().draw_mermaid_png() 可视化这张图。

2.2 传入模型对象

更灵活的方式是显式构造模型对象,例如用 init_chat_model:

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
import os
​
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)
agent = create_agent(model)

三、基本用法②:如何调用 Agent

agent.invoke() 是 Agent 最基本的同步调用方法,会阻塞直到返回最终结果。

  • 输入:字典类型,通过 messages 字段传递消息列表,即 {"messages": [{"role": "...", "content": "..."}]}

  • 输出:底层可能经历多轮交互,返回的是完整的消息列表,封装在字典的 messages 字段里

response = agent.invoke({"messages": [...]})
​
# response 结构示意
{
    "messages": [
        HumanMessage(...),    # 用户问题
        AIMessage(...),       # AI 工具调用
        ToolMessage(...),     # 工具返回结果
        AIMessage(...)        # 最终回答 ← 通常取这个
    ]
}
​
final_answer = response['messages'][-1].content

也可以在消息列表开头加一条 system 角色消息来定义 Agent 行为:

resp = agent.invoke({
    "messages": [
        {"role": "system", "content": "你是一个小学数学老师,耐心,幽默,讲解深入浅出"},
        {"role": "user", "content": "100加上50等于多少?"}
    ]
})

四、基本用法③:绑定工具

只有接入工具,create_agent 才算完整。Agent 支持静态和动态绑定工具(动态绑定需借助中间件)。

4.1 绑定自定义工具

用 @tool 装饰器即可把一个普通函数变成工具,parse_docstring=True 表示从 docstring 自动解析工具描述与参数说明:

from langchain.agents import create_agent
from langchain.tools import tool
​
@tool(parse_docstring=True)
def get_weather(city: str) -> str:
    """天气查询工具
​
    Args:
        city: 城市名称
    """
    return f"{city}的天气为晴朗,25°C。"
​
agent = create_agent(model=model, tools=[get_weather])

4.2 绑定内置工具(如 Tavily 搜索)

LangChain 生态内置了大量工具,可以直接拿来用,比如 Tavily 网络搜索:

from langchain_tavily import TavilySearch
from langchain.agents import create_agent
​
web_search = TavilySearch(max_results=2)
agent = create_agent(model=model, tools=[web_search])
​
result = agent.invoke({"messages": [{"role": "user", "content": "2024年诺贝尔物理学奖得主是谁?"}]})
print(result['messages'][-1].content)

内置工具列表:https://docs.langchain.com/oss/python/integrations/tools

4.3 一次完整的 Function Calling 流程

观察运行日志可以发现,一次工具调用实际是标准的 Function Calling 流程,包含四条消息:

human message → ai message(function call)→ tool message(function response)→ ai message(final response)

4.4 绑定多个工具

把多个工具放进列表即可,Agent 会根据用户问题自动判断调用哪个(甚至并行调用多个):

agent = create_agent(model, tools=[get_weather, get_news])
response = agent.invoke({"messages": ["你好,杭州今天的天气如何?今天有哪些新闻?"]})

4.5 常见问题排查(FAQ)

问题

原因

解决

选错工具

多个工具描述相似 / 工具太多

只给必要工具(2~5 个最佳)、描述区分明确、system_prompt 中说明使用场景

如何判断完成

—

当 AIMessage 不含 tool_calls 时即完成

能调用多少次工具

默认无限制

可用 config={"recursion_limit": 5} 限制步数


五、高级用法①:设置 Agent 名称(name)

创建 Agent 时可以用 name 参数为其命名:

agent = create_agent(model=model, name="chat_assistant")

设置后,Agent 产生的 AIMessage 会携带 name 信息,在 Multi-Agent 场景中用于区分不同的 Agent。它的作用远不止多 Agent 编排:

  1. 流式输出归因 —— 标识当前输出来自哪个 Agent;

  2. 消息身份标记 —— 会话记录、审计日志中明确消息生成者;

  3. 调试与 trace 可读性 —— 快速定位当前执行的是哪个 Agent;

  4. 组件化封装 —— 维护一致的身份标识;

  5. 前端展示与可观测性 —— 作为运行时展示标识;

  6. 稳定的运行时身份 ID —— 便于日志检索、监控统计、链路分析。

生产环境建议显式设置 name,而不是依赖默认行为。


六、高级用法②:系统提示词(system_prompt)

系统指令(SystemMessage)通过 system_prompt 设置,为 Agent 提供任务背景、行为准则和操作指南。这个参数可以是 str,也可以是 SystemMessage 类型。

使用建议:

  • 明确说明 Agent 的角色

  • 定义输出格式

  • 说明何时使用工具

agent = create_agent(
    model=model,
    tools=[get_weather],
    system_prompt="""你是天气助手。
​
工作流程:
1. 理解用户的城市查询
2. 使用 get_weather 工具获取数据
3. 简洁清晰地回答
​
输出格式:
- 天气状况
- 温度
- 注意事项(如有)
"""
)

提示词设置分静态和动态两种,动态设置需借助中间件。


七、高级用法③:结构化输出

结构化输出是 Agent 的核心功能之一,允许 Agent 以可预测的格式返回数据(Pydantic 模型、JSON 对象、数据类等),让程序直接使用,无需复杂解析。

7.1 模型 vs Agent 的结构化输出

维度

模型的结构化输出

Agent 结构化输出

操作对象

大模型对象

Agent

解析时机

每次模型调用生成 AIMessage 时解析

仅在 Agent 决定"任务结束"输出最终答案时解析

数据流转

模型 → 结构化对象

模型 → 工具 → 反思 → … → 结构化对象

绑定方式

with_structured_output

response_format 参数

适用场景

单次、确定性的任务(提取字段、翻译、分类)

多步、复杂推理的任务(查文档后汇总报表)

7.2 四种策略

通过 response_format 参数设置,支持四种策略:

① ProviderStrategy(原生结构化输出)

使用模型提供商 API 原生能力强制保证格式,在源头确保结构准确性。适用于 OpenAI、Anthropic Claude、xAI Grok 等支持原生结构化输出的模型。

from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.agents.structured_output import ProviderStrategy
​
class ContactInfo(BaseModel):
    """用户的联系方式"""
    name: str = Field(description="用户姓名")
    email: str = Field(description="用户邮箱地址")
    phone: str = Field(description="用户的手机号")
​
agent = create_agent(model=model, response_format=ProviderStrategy(ContactInfo))

② ToolStrategy(工具调用实现)

对不支持原生结构化输出的模型,LangChain 动态创建一个"虚拟工具",其输入参数对应期望的数据结构。模型生成最终答案时会被引导"调用"这个虚拟工具,间接产生结构化数据。这是兼容性最强的策略,适用于任何支持工具调用的现代模型。

from langchain.agents.structured_output import ToolStrategy
​
agent = create_agent(
    model=model,
    tools=[search_tool],
    response_format=ToolStrategy(ContactInfo)
)
result = agent.invoke({"messages": [{"role": "user", "content": "联系方式:John Doe,john@atguigu.com"}]})
print(result["structured_response"])

③ type / AutoStrategy(自动选择)

直接传入类型时,LangChain 自动包装为 AutoStrategy:模型支持原生结构化输出则用 ProviderStrategy,否则用 ToolStrategy。

⚠️ 注意:LangChain 1.0+ 中直接传 response_format=ContactInfo 的方式曾不支持,必须显式用 ToolStrategy 或 ProviderStrategy(1.2 版本经测试仍可用)。

④ None(默认)

不结构化输出,以自然语言响应。

总结:实际 Agent 开发中若用到结构化输出,推荐使用 ToolStrategy。

7.3 ToolStrategy 详解

ToolStrategy 实际通过 Function Calling 实现结构化输出,LangChain 会在消息列表末尾追加一条伪造的 ToolMessage(并无真实工具执行)。它有三个参数:

class ToolStrategy(Generic[SchemaT]):
    schema: type[SchemaT]                                    # 必需
    tool_message_content: str | None                         # 可选:自定义成功提示
    handle_errors: Union[bool, str, type, tuple, Callable]   # 可选:错误处理策略,默认 True

7.3.1 schema 支持的四种类型

  1. Pydantic 模型(推荐,支持数据验证)

  2. TypedDict(带类型提示的字典,不支持运行时验证)

  3. JSON Schema

  4. 数据类(@dataclass)

还支持联合类型 Union[类型1, 类型2],允许模型根据输入内容选择最匹配的数据结构。

7.3.2 错误处理:handle_errors 参数

模型输出可能不符合格式要求,handle_errors 提供五种策略:

取值

行为

True(默认)

捕获所有异常,用内置错误模板提示模型重试

False

关闭自动重试,异常直接抛出中断程序

"自定义字符串"

捕获异常后用固定字符串作为错误消息

ExceptionType/元组

仅捕获指定异常类型重试,其余直接抛出

callable

用自定义函数处理异常,最灵活

LangChain 默认处理两类异常:MultipleStructuredOutputsError(多结构化输出错误)和 StructuredOutputValidationError(输出验证错误),默认都会拦截并自动重试。


八、高级用法④:流式输出及模式

invoke 调用时内部可能经历多次模型调用,长时间看不到进展,体验不佳。流式调用(渐进式显示输出) 能实时显示 Agent 运行过程,大幅降低用户等待焦虑。

通过 agent.stream(stream_mode=指定模式) 设置,共七种模式:

模式

输出内容

使用场景

values

每步执行后输出完整状态信息

每步都要获取完整状态、状态持久化

updates(默认)

每步只增量更新变化内容

监控 Agent 执行进度

messages

流式 Token + 元数据(来自哪个节点)

ChatGPT 打字机效果,实时对话

tasks

task 开始/结束时间 + 结果/错误

监控任务生命周期

debug

比 tasks 多任务步骤、时间戳、类型

调试、监控生命周期

checkpoints

检查点创建时触发输出

状态持久化、工作流恢复、分布式跟踪

custom

通过 get_stream_writer 自定义发送数据

输出业务进度信息、自定义日志

选型建议:

  • 实时对话交互 → messages

  • 观察思考与执行步骤 → updates

  • 查看每步状态 → values / tasks / debug

  • 工具执行时输出自定义业务日志 → custom

这些模式还可以组合使用,例如 stream_mode=["tasks", "updates"]:

for stream_mode, chunk in customer_service_agent.stream(
    {"messages": [...]},
    stream_mode=["tasks", "updates"]
):
    print(f"当前流模式: {stream_mode}, 当前数据: {chunk}")

九、实战:多功能智能助手

综合运用上述知识,开发一个支持天气查询、数学计算、时间查询、货币转换、信息搜索的多功能智能助手。

9.1 工具定义

用 @tool 定义五个工具(这里以天气为例,其余工具同理):

@tool
def get_weather(city: str) -> str:
    """获取指定城市的实时天气信息"""
    weather_db = {
        "北京": "多云,15-22℃,空气质量良,湿度 45%",
        "上海": "晴天,18-25℃,空气质量优,湿度 60%",
        # ...
    }
    return weather_db.get(city, f"暂不支持查询{city}的天气")

9.2 Agent 创建(封装为类)

class SmartAssistant:
    def __init__(self):
        self.tools = [get_weather, calculator, get_time_info, convert_currency, search_info]
        system_prompt = """你是一个多功能智能助手,可以帮助用户:
🌤 查询天气  🔢 数学计算  ⏰ 时间查询  💱 货币转换  🔍 信息搜索
请始终使用中文回答。"""
        self.agent = create_agent(model=self.model, tools=self.tools, system_prompt=system_prompt)
        self.messages = []
​
    def chat(self, user_input: str) -> str:
        self.messages.append({"role": "user", "content": user_input})
        result = self.agent.invoke({"messages": self.messages})
        self.messages = result["messages"]
        for msg in reversed(self.messages):
            if msg.type == "ai" and msg.content:
                return msg.content
        return "抱歉,我无法处理这个请求。"
​
    def reset(self):
        self.messages = []

9.3 运行效果

👤 北京今天天气怎么样?
🤖 北京今天天气是:多云,15–22℃,空气质量良,湿度 45%。
​
👤 帮我算一下 (25 + 17) * 3
🤖 计算结果是:126。
​
👤 现在几点了?
🤖 现在是:2026年06月03日 19:45:40。
​
👤 100 美元等于多少人民币?
🤖 100 美元(USD)约等于 714.29 人民币(CNY)。

十、总结:一张图记住本章要点