Skip to content

Agent 智能体

分类:LangChain
标签:LangChain
创建时间:2026年10月26日 22:20:48

理解 Agent ​

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

Agent的关键能力

  • 理解用户问题
  • 如何拆解任务
  • 判断是否需要工具
  • 需要调用哪些工具
  • 如何利用好工具结果生成回答&推进任务

Agent的核心组件 ​

  • 必须的:行动(Action)
  • 几乎总是存在的:工具(Tool)
  • 有条件存在的:规划决策(Planning)
  • 最容易被省略的:记忆(Memory)

Agent创建与调用 ​

在 LangChain 0.x 时代,框架内的 Agent 系统经历了“碎片化”阶段。

  • 如果你要实现思维链推理(ReAct),就用 create_react_agent。
  • 如果需要结构化输出,就用 create_structured_chat_agent。
  • 要工具调用,则用 create_tool_calling_agent。

存在的问题:

  • 心智负担高——每种 Agent 都要单独记忆 API 与参数。
  • 可组合性差——多个 Agent 之间无法统一调度。
  • 生态碎片化——不同模块难以复用或协同演化。

LangChain 在 1.0 版本后,将所有 Agent 的创建方式统一为一个入口: create_agent()。

python
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent

# 1. 初始化模型
model = init_chat_model("gpt-4o-mini", model_provider="openai")

# 2. 创建 agent(一步完成)
agent = create_agent(
    model=model,
    tools=[tool1, tool2],
    system_prompt="Agent 的行为指令"  # 可选
)

# 3. 调用
result = agent.invoke({
    "messages": [{"role": "user", "content": "问题"}]
})

模型的传入方式 ​

传入字符串 ​

python
from langchain.agents import create_agent

agent = create_agent(
    model="ollama:qwen3.5"
)

print(type(agent))

""" 输出结果:
<class 'langgraph.graph.state.CompiledStateGraph'>
"""

传入模型对象 ​

python
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

model = init_chat_model(
    model="ollama:qwen3.5"
)

agent = create_agent(model)

print(type(agent))

""" 输出结果:
<class 'langgraph.graph.state.CompiledStateGraph'>
"""

如何调用 Agent ​

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

  • 输入:传入的参数为字典类型,字典内通过 messages 字段传递消息列表。
  • 输出:通过 invoke 调用 Agent,底层可能会经历多轮交互,返回的是完整的消息列表,被封装在字典中,是 messages 字段的值。
python
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage
from rich import print as rprint

load_dotenv(verbose=True)

agent = create_agent(
    model=init_chat_model(
        model="deepseek:deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    )
)

messages = [
    SystemMessage("你是一个优秀的数学老师"),
    HumanMessage("1 + 2 = ?")
]

res = agent.invoke({"messages": messages})

rprint(res)

""" 输出结果:
{
    'messages': [
        SystemMessage(
            content='你是一个优秀的数学老师',
            additional_kwargs={},
            response_metadata={},
            id='2e9bc7b5-e22b-469d-a421-9f3544b41bbe'
        ),
        HumanMessage(
            content='1 + 2 = ?',
            additional_kwargs={},
            response_metadata={},
            id='2f77e1d3-196b-4579-82da-aba0af509acd'
        ),
        AIMessage(
            content='1 + 2 = 3',
            additional_kwargs={'refusal': None},
            response_metadata={
                'token_usage': {
                    'completion_tokens': 7,
                    'prompt_tokens': 16,
                    'total_tokens': 23,
                    'completion_tokens_details': None,
                    'prompt_tokens_details': {
                        'audio_tokens': None,
                        'cached_tokens': 0
                    },
                    'prompt_cache_hit_tokens': 0,
                    'prompt_cache_miss_tokens': 16
                },
                'model_provider': 'deepseek',
                'model_name': 'deepseek-flash',
                'system_fingerprint': 'e72d3c7f663d3e1383b00b3884cb4501',
                'id': '36f7cb13-dd1f-414a-8e82-cd98ed6b6f35',
                'finish_reason': 'stop',
                'logprobs': None
            },
            id='lc_run--01a1791e-80e5-7161-b48d-5a468b5d3c3d-0',
            tool_calls=[],
            invalid_tool_calls=[],
            usage_metadata={
                'input_tokens': 16,
                'output_tokens': 7,
                'total_tokens': 23,
                'input_token_details': {'cache_read': 0},
                'output_token_details': {}
            }
        )
    ]
}
"""

工具的调用 ​

只有接入了一些工具,create_agent 完成 Agent 创建才算完整。

Agent支持 静态 和 动态 绑定工具,后者需要用到中间件。

之前我们定义工具需要用 invoke 方式调用,但是在 Anget 中我们可用通过 tools 配置项绑定工具,Agent 会在需要调用工具时自动调用。

调用自定义工具 ​

python
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from rich import print as rprint

load_dotenv(verbose=True)

@tool(parse_docstring=True)
def get_weather(city: str):
    """天气查询工具

    Args:
        city: 城市
    """
    return f"{city}天气晴朗,温度15摄氏度"

agent = create_agent(
    model=init_chat_model(
        model="deepseek:deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    tools=[get_weather]
)

res = agent.invoke({
    "messages": ["帮我查一下北京今天的天气情况"]
})
rprint(res)

""" 输出结果:
{
    'messages': [
        HumanMessage(
            content='帮我查一下北京今天的天气情况',
            additional_kwargs={},
            response_metadata={},
            id='5b1f735a-d175-4604-8279-26b4ef18fdbc'
        ),
        AIMessage(
            content='我来帮您查询北京今天的天气。',
            additional_kwargs={'refusal': None},
            response_metadata={
                'token_usage': {
                    'completion_tokens': 45,
                    'prompt_tokens': 275,
                    'total_tokens': 320,
                    'completion_tokens_details': None,
                    'prompt_tokens_details': {
                        'audio_tokens': None,
                        'cached_tokens': 0
                    },
                    'prompt_cache_hit_tokens': 0,
                    'prompt_cache_miss_tokens': 275
                },
                'model_provider': 'deepseek',
                'model_name': 'deepseek-flash',
                'system_fingerprint': 'e72d3c7f663d3e1383b00b3884cb4501',
                'id': '348d3bfd-6fd4-4e74-96bc-2b5dc670b780',
                'finish_reason': 'tool_calls',
                'logprobs': None
            },
            id='lc_run--01a17bf7-c783-7621-b4c1-639b8f14bdd4-0',
            tool_calls=[
                {
                    'name': 'get_weather',
                    'args': {'city': '北京'},
                    'id': 'call_00_hruOLvhkeDVDkuKYc1B86560',
                    'type': 'tool_call'
                }
            ],
            invalid_tool_calls=[],
            usage_metadata={
                'input_tokens': 275,
                'output_tokens': 45,
                'total_tokens': 320,
                'input_token_details': {'cache_read': 0},
                'output_token_details': {}
            }
        ),
        ToolMessage(
            content='北京天气晴朗,温度15摄氏度',
            name='get_weather',
            id='545089ba-ea29-4ad8-b297-0028de364864',
            tool_call_id='call_00_hruOLvhkeDVDkuKYc1B86560'
        ),
        AIMessage(
            content='北京今天的天气情况如下:\n\n- **天气状况**:晴朗 ☀️\n- 
**温度**:15°C\n\n今天是个大晴天,气温15摄氏度,体感比较舒适。适合外出活动和户外
安排,不过早晚温差可能较大,建议适当增减衣物。祝您今天愉快!',
            additional_kwargs={'refusal': None},
            response_metadata={
                'token_usage': {
                    'completion_tokens': 62,
                    'prompt_tokens': 339,
                    'total_tokens': 401,
                    'completion_tokens_details': None,
                    'prompt_tokens_details': {
                        'audio_tokens': None,
                        'cached_tokens': 128
                    },
                    'prompt_cache_hit_tokens': 128,
                    'prompt_cache_miss_tokens': 211
                },
                'model_provider': 'deepseek',
                'model_name': 'deepseek-flash',
                'system_fingerprint': 'e72d3c7f663d3e1383b00b3884cb4501',
                'id': '01a5212c-9656-4226-bd20-8e5f54dab939',
                'finish_reason': 'stop',
                'logprobs': None
            },
            id='lc_run--01a17bf7-cb8e-76ac-81b0-b9ee71418614-0',
            tool_calls=[],
            invalid_tool_calls=[],
            usage_metadata={
                'input_tokens': 339,
                'output_tokens': 62,
                'total_tokens': 401,
                'input_token_details': {'cache_read': 128},
                'output_token_details': {}
            }
        )
    ]
}
"""

接入内置工具 ​

LangChain 内置工具列表:中文文档 英文文档

绑定内置的 TavilySearch 搜索工具,可以借助 Tavily 进行网络搜索和信息爬取。

这里我们需要先在 tavily 官网注册并获得API-KEY(每月有免费额度):https://www.tavily.com/

python
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_tavily import TavilySearch
from rich import print as rprint

load_dotenv(verbose=True)

agent = create_agent(
    model=init_chat_model(
        model="deepseek:deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    tools=[TavilySearch(max_results=2)]
)

res = agent.invoke({
    "messages": ["帮我查一下荣获 2026 年金球奖的球员信息"]
})

工具调用流程分析 ​

LangChain 的 Agent 会将 模型 与 工具 结合起来,在实现上由一个基于 LangGraph 的图结构来编排执行流程,如下所示。

这与前文得到的 Agent 图结构是一致的,本质上就是经典的 ReAct 结构:一个具备 “思考-行动-观察” 不断循环的自主工作者。

重试机制 ​

Agent可以在工具调用结果不满足要求时,自主重试。

python
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool

load_dotenv(verbose=True)

flag = 0

@tool(parse_docstring=True)
def get_weather(city: str):
    """天气查询工具

    Args:
        city: 城市
    """
    global flag
    flag += 1

    if flag < 3:
        return "TEMP_UNAVAILABLE"

    return f"{city}天气晴朗,温度15摄氏度"

agent = create_agent(
    model=init_chat_model(
        model="deepseek:deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    tools=[get_weather]
)

res = agent.invoke({
    "messages": [
        SystemMessage("你是一个天气助手。当工具返回以 'TEMP_UNAVAILABLE:' 开头的结果时,说明是临时故障,不要立即放弃;你应再次调用同一个工具,最多重试 3 次。如果 3 次后仍失败,再向用户说明服务暂时不可用"),
        HumanMessage("帮我查一下北京今天的天气情况")
    ]
})

for msg in res["messages"]:
    msg.pretty_print()

""" 输出结果:
================================ System Message ================================

你是一个天气助手。当工具返回以 'TEMP_UNAVAILABLE:' 开头的结果时,说明是临时故障,不要立即放弃;你应再次调用同一个工具,最多重试 3 次。如果 3 次后仍失败,再向用户说明服务暂时不可用
================================ Human Message =================================

帮我查一下北京今天的天气情况
================================== Ai Message ==================================

I'll check the weather in Beijing for you.
Tool Calls:
  get_weather (call_00_DRpil6AFzmcL0XTbIrv09025)
 Call ID: call_00_DRpil6AFzmcL0XTbIrv09025
  Args:
    city: 北京
================================= Tool Message =================================
Name: get_weather

TEMP_UNAVAILABLE
================================== Ai Message ==================================

The service returned a temporary error. Let me retry.
Tool Calls:
  get_weather (call_00_q329VcBRsKgcC1M1bGpc6191)
 Call ID: call_00_q329VcBRsKgcC1M1bGpc6191
  Args:
    city: 北京
================================= Tool Message =================================
Name: get_weather

TEMP_UNAVAILABLE
================================== Ai Message ==================================

Still failing. Let me try one more time.
Tool Calls:
  get_weather (call_00_GV1MKrgze5t0rDmFnFun6237)
 Call ID: call_00_GV1MKrgze5t0rDmFnFun6237
  Args:
    city: 北京
================================= Tool Message =================================
Name: get_weather

北京天气晴朗,温度15摄氏度
================================== Ai Message ==================================

刚才服务出现了两次临时故障,第三次尝试成功了。以下是北京今天的天气情况:

- **天气**:晴朗 ☀️
- **温度**:15°C

天气不错,适合外出活动,不过早晚温差可能较大,建议适当增减衣物。
"""

常见问题 ​

面试题:Agent 如何选择工具?

依据工具的 docstring。AI 会根据问题内容和每个工具的描述自动选择最匹配的工具

面试题:Agent 为什么没有调用工具?

  1. 工具的 docstring 不清晰
  2. 问题表述不明确
  3. 模型认为不需要工具

面试题:Agent 选错工具?

原因:

  1. 多个工具的功能描述相似
  2. 工具太多导致混淆

解决:

  1. 只给必要的工具
  2. 工具描述要有明确区分
  3. 在 system_prompt 中说明工具使用场景

面试题:如何知道 Agent 何时完成?

当 AIMessage 不包含 tool_calls 时视为完成。

面试题:Agent 可以调用多少次工具?

默认没有限制,直到得到最终答案。但可能会:

  1. 超时
  2. 达到 token 限制
  3. 模型决定停止

面试题:如何限制工具调用次数?

LangChain 1.0 的 create_agent 默认使用 LangGraph,可以通过配置限制:

python
config = {
    "recursion_limit": 5  # 最多 5 步
}
res = agent.invoke(input, config=config)

设置 Agent 名称 ​

创建 Agent 时可以通过 name 属性配置 Agent 名称。

使用场景:

  1. 多 Agent 交互:name 会作为 Agent 的唯一标识。
  2. 流式输出归因;在启用流式输出时,name 可用于标识当前输出内容来自哪个Agent。
  3. 消息身份标记:设置 name 后,Agent 产生的 AIMessage 会携带对应的name信息。
  4. 调试与trace可读性。
  5. 组件化封装。
  6. 前端展示与运行态可观测性。
  7. 作为稳定的运行时身份标识。

设置系统提示词 ​

创建 Agent 时可以通过 system_prompt 属性配置系统提示词,提示词不会出现在最终的结果里。

使用建议:

  • 明确说明 Agent 的角色
  • 定义输出格式
  • 说明何时使用工具

结构化输出 ​

创建 Agent 时通过 response_format 属性配置 Agent 的输出结构。

当模型生成结构化数据时,系统会自动捕获、验证并将结果存储在Agent状态的 structured_response 键中。

模型与 Agent 结构化输出的区别 ​

模型Agent
操作对象作用于大模型对象作用于 Agent
解析时机每次模型调用生成 AIMessage 时进行解析仅在 Agent 决定结束任务并输出最终将答案时解析
数据流转模型 -> 结构化对象模型 -> 工具 -> 反思 -> ... -> 结构化对象
绑定方式使用 with_structured_output使用 response_format 参数
适用场景单次、确定性的任务多步、复杂推理的任务

ProviderStrategy 策略 ​

使用模型提供商的 原生结构化输出功能 实现结构化输出。原生结构化输出 指的是大语言模型(LLM)提供商通过其API直接提供的、在模型响应阶段就强制保证输出格式符合预定规范的能力,这种能力能够在模型生成内容的源头确保结构化准确性。

ToolStrategy 策略 ​

对于不支持原生结构化输出的模型,LangChain采用“ToolStrategy”工具调用的方式实现结构化输出。

此策略兼容绝大多数 支持工具调用 的现代模型,其核心原理是动态创建一个 "虚拟工具",该工具的输入参数对应着期望的数据结构。

当模型需要生成最终答案时,系统会引导模型 "调用"这个虚拟工具 ,从而间接产生符合要求的结构化数据。

python
import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.structured_output import ProviderStrategy, ToolStrategy
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field

load_dotenv(verbose=True)

class UserInfo(BaseModel):
    """用户信息结构体"""
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    work: str = Field(description="工作")

agent = create_agent(
    model=init_chat_model(
        model="deepseek:deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    response_format=ToolStrategy(UserInfo)
)

res = agent.invoke({
    "messages": [
        HumanMessage("请在下面文本中提取用户信息:小明是一个30岁的程序员")
    ]
})

for msg in res["messages"]:
    msg.pretty_print()

print(f"\nstructured_response = {res["structured_response"]}")

""" 输出结果:
================================ Human Message =================================

请在下面文本中提取用户信息:小明是一个30岁的程序员
================================== Ai Message ==================================
Tool Calls:
  UserInfo (call_00_5Sxj89LyT8wiPCUaFzT92581)
 Call ID: call_00_5Sxj89LyT8wiPCUaFzT92581
  Args:
    name: 小明
    age: 30
    work: 程序员
================================= Tool Message =================================
Name: UserInfo

Returning structured response: name='小明' age=30 work='程序员'

structured_response = name='小明' age=30 work='程序员'
"""

AutoStrategy 策略 ​

使用 AutoStrategy 包装时会触发 自动选择策略,模型若支持 原生结构化输出功能 则优先使用 ProviderStrategy 策略,若不支持则使用 ToolStrategy 策略。

None 策略 ​

默认配置,表示不以结构化输出,以 自然语言 响应用户问题。

ToolStrategy 策略详解 ​

上一节结构化输出中提到了四种输出模式Pydantic、 TypedDict、 JSON Schema、dataclass

ToolStrategy 参数

  • schema(必须):支持四种输出模式,同时支持联合类型 Union[类型1, 类型2](允许模型根据输入内容选择最匹配的数据结构)
  • tool_message_content(可选参数):用于自定义生成结构化输出时,会话历史中记录的提示信息。默认使用展示输出数据的标准响应语句。
  • handle_errors(可选参数):用于指定数据校验失败时的重试策略,默认值为True。

schema 联合类型 ​

python
import os
from typing import Union
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field
from typing_extensions import TypedDict, Annotated

load_dotenv(verbose=True)

class UserInfo(BaseModel):
    """用户信息结构体"""
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    work: str = Field(description="工作")

class MethodInfo(TypedDict):
    """事件信息结构"""
    name: Annotated[str, ..., "事件名"]
    time: Annotated[str, ..., "发生时间"]

agent = create_agent(
    model=init_chat_model(
        model="deepseek:deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    response_format=ToolStrategy(schema=Union[UserInfo, MethodInfo]),
)

res = agent.invoke({
    "messages": [
        HumanMessage("请在下面文本中提取用户信息:小明是一个30岁的程序员")
    ]
})

for msg in res["messages"]:
    msg.pretty_print()

""" 输出结果:
================================ Human Message =================================

请在下面文本中提取用户信息:小明是一个30岁的程序员
================================== Ai Message ==================================
Tool Calls:
  UserInfo (call_00_TJyg2MmgV2uHcDTmtOVi5258)
 Call ID: call_00_TJyg2MmgV2uHcDTmtOVi5258
  Args:
    name: 小明
    age: 30
    work: 程序员
================================= Tool Message =================================
Name: UserInfo

Returning structured response: name='小明' age=30 work='程序员'
"""
python
import os
from typing import Union
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field
from typing_extensions import TypedDict, Annotated

load_dotenv(verbose=True)

class UserInfo(BaseModel):
    """用户信息结构体"""
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    work: str = Field(description="工作")

class MethodInfo(TypedDict):
    """事件信息结构"""
    name: Annotated[str, ..., "事件名"]
    time: Annotated[str, ..., "发生时间"]

agent = create_agent(
    model=init_chat_model(
        model="deepseek:deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    response_format=ToolStrategy(schema=Union[UserInfo, MethodInfo]),
)

res = agent.invoke({
    "messages": [
        HumanMessage("请在下面文本中提取信息:昨天下午捡到了一个手机")
    ]
})

for msg in res["messages"]:
    msg.pretty_print()

""" 输出结果:
================================ Human Message =================================

请在下面文本中提取信息:昨天下午捡到了一个手机
================================== Ai Message ==================================
Tool Calls:
  MethodInfo (call_00_492b6atc9bw7g9IPrAqo6300)
 Call ID: call_00_492b6atc9bw7g9IPrAqo6300
  Args:
    name: 捡到手机
    time: 昨天下午
================================= Tool Message =================================
Name: MethodInfo

Returning structured response: {'name': '捡到手机', 'time': '昨天下午'}
"""

tool_message_content 参数 ​

tool_message_content 参数用来设置伪工具类返回值的 content 字段

python
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field
from rich import print as rprint

load_dotenv(verbose=True)

class UserInfo(BaseModel):
    """用户信息结构体"""
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    work: str = Field(description="工作")

agent = create_agent(
    model=init_chat_model(
        model="deepseek:deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    response_format=ToolStrategy(UserInfo, tool_message_content="格式化输出成功"),
)

res = agent.invoke({
    "messages": [
        HumanMessage("请在下面文本中提取信息:小明是一个30岁的程序员")
    ]
})

rprint(res)

""" 输出结果:
{
    'messages': [
        ...
        ToolMessage(
            content='格式化输出成功',
            name='UserInfo',
            id='8e3a88fa-d7a0-4ce3-bab2-69dbf6142b1e',
            tool_call_id='call_00_wL4BxYVpt9dNFG3Ov7Hp5622'
        )
    ],
    'structured_response': UserInfo(name='小明', age=30, work='程序员')
}
"""

handle_errors 参数 ​

受限于模型能力,大模型输出的内容可能并 不符合格式要求 ,ToolStrategy 通过其 handle_errors 参数提供了结构化过程错误处理策略,以下是主要的几种方式及其用途:

  • handle_errors=True:LangChain 默认方式,捕获所有异常,并使用 LangChain 内置的、信息明确的 错误消息模板 提示模型重试,确保最终能得到符合预定格式的有效数据。适用于大多数希望自动处理错误的通用场景。
  • handle_errors=False:关闭自动重试机制,任何异常都会直接抛出,会中断程序运行。
  • handle_errors="自定义字符串":捕获所有异常,使用开发者预设的固定字符串作为错误消息。适用于需要统一、友好的用户提示,或进行特定业务引导的场景。
  • handle_errors=ExceptionType:仅捕获指定类型或元组中的异常类型并进行重试,其他异常直接抛出。适用于需要精准控制,只对特定错误进行重试的场景。
  • handle_errors=callable:灵活性最高的方式,使用开发者自定义的函数来处理异常,可根据不同的异常类型返回差异化的提示信息。适用于需要复杂、精细化错误处理的场景。
True / Flase / Str ​
python
import os
from typing import Union

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field
from rich import print as rprint

load_dotenv(verbose=True)

class UserInfo(BaseModel):
    """用户信息结构体"""
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    work: str = Field(description="工作")

class MethodInfo(BaseModel):
    """事件信息结构体"""
    name: str = Field(description="事件名")
    time: str = Field(description="发生的时间")

agent = create_agent(
    model=init_chat_model(
        model="deepseek:deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    response_format=ToolStrategy(
        schema=Union[UserInfo, MethodInfo],
        tool_message_content="格式化输出成功",
        handle_errors=True,
    ),
)

res = agent.invoke({
    "messages": [
        HumanMessage("请在下面文本中提取信息:小明是一个30岁的程序员,昨天下午在图书馆捡了一部手机")
    ]
})

rprint(res)

""" 输出结果:
{
    'messages': [
        HumanMessage(
            content='请在下面文本中提取信息:小明是一个30岁的程序员,昨天下午在图书馆捡了一部手机',
            ...
        ),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'UserInfo',
                    'args': {'name': '小明', 'age': 30, 'work': '程序员'},
                    'id': 'call_00_YBE0hSd1vhIciAEwzRGw1901',
                    'type': 'tool_call'
                },
                {
                    'name': 'MethodInfo',
                    'args': {'name': '捡手机', 'time': '昨天下午'},
                    'id': 'call_01_yKU5TF6m21GH2X9d4cZL8219',
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(
            content='Error: Model incorrectly returned multiple structured responses (UserInfo, MethodInfo) when only one is expected.\n Please fix your mistakes.',
            name='UserInfo',
            id='7f6dd59a-1b13-4bb6-8cfa-68079ac90dbd',
            tool_call_id='call_00_YBE0hSd1vhIciAEwzRGw1901'
        ),
        ToolMessage(
            content='Error: Model incorrectly returned multiple structured responses (UserInfo, MethodInfo) when only one is expected.\n Please fix your mistakes.',
            name='MethodInfo',
            id='e0ff2301-7d90-4381-b0fa-026880a345e8',
            tool_call_id='call_01_yKU5TF6m21GH2X9d4cZL8219'
        ),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'UserInfo',
                    'args': {'name': '小明', 'age': 30, 'work': '程序员'},
                    'id': 'call_00_NqDeDlyjeWEgvD8hL18m7305',
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(
            content='格式化输出成功',
            name='UserInfo',
            id='3a5bbfe1-8efb-4466-a964-ec91d38ab4a8',
            tool_call_id='call_00_NqDeDlyjeWEgvD8hL18m7305'
        )
    ],
    'structured_response': UserInfo(name='小明', age=30, work='程序员')
}
"""
python
import os
from typing import Union

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field
from rich import print as rprint

load_dotenv(verbose=True)

class UserInfo(BaseModel):
    """用户信息结构体"""
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    work: str = Field(description="工作")

class MethodInfo(BaseModel):
    """事件信息结构体"""
    name: str = Field(description="事件名")
    time: str = Field(description="发生的时间")

agent = create_agent(
    model=init_chat_model(
        model="deepseek:deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    response_format=ToolStrategy(
        schema=Union[UserInfo, MethodInfo],
        tool_message_content="格式化输出成功",
        handle_errors=False,
    ),
)

res = agent.invoke({
    "messages": [
        HumanMessage("请在下面文本中提取信息:小明是一个30岁的程序员,昨天下午在图书馆捡了一部手机")
    ]
})

rprint(res)

""" 输出结果:
...
langchain.agents.structured_output.MultipleStructuredOutputsError: Model incorrectly returned multiple structured responses (UserInfo, MethodInfo) when only one is expected.
During task with name 'model' and id '4fb0a338-3171-081d-cd74-042008520fd4'
"""
python
import os
from typing import Union

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field
from rich import print as rprint

load_dotenv(verbose=True)

class UserInfo(BaseModel):
    """用户信息结构体"""
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    work: str = Field(description="工作")

class MethodInfo(BaseModel):
    """事件信息结构体"""
    name: str = Field(description="事件名")
    time: str = Field(description="发生的时间")

agent = create_agent(
    model=init_chat_model(
        model="deepseek:deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    response_format=ToolStrategy(
        schema=Union[UserInfo, MethodInfo],
        tool_message_content="格式化输出成功",
        handle_errors="Error: 请检查输入数据",
    ),
)

res = agent.invoke({
    "messages": [
        HumanMessage("请在下面文本中提取信息:小明是一个30岁的程序员,昨天下午在图书馆捡了一部手机")
    ]
})

rprint(res)

""" 输出结果:
{
    'messages': [
        HumanMessage(
            content='请在下面文本中提取信息:小明是一个30岁的程序员,昨天下午在图书馆捡了一部手机',
            ...
        ),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'UserInfo',
                    'args': {'name': '小明', 'age': 30, 'work': '程序员'},
                    'id': 'call_00_dpVVrRdiFb4HHpa2vLcp2538',
                    'type': 'tool_call'
                },
                {
                    'name': 'MethodInfo',
                    'args': {'name': '捡手机', 'time': '昨天下午'},
                    'id': 'call_01_VahPEmOCRDx0F8Esg67B7086',
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(
            content='Error: 请检查输入数据',
            name='UserInfo',
            id='42682896-33af-4c14-a24a-7b75c04c58b8',
            tool_call_id='call_00_dpVVrRdiFb4HHpa2vLcp2538'
        ),
        ToolMessage(
            content='Error: 请检查输入数据',
            name='MethodInfo',
            id='7ecd20a8-9eb5-4133-a071-886d5fd23a32',
            tool_call_id='call_01_VahPEmOCRDx0F8Esg67B7086'
        ),
        ...
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'UserInfo',
                    'args': {'name': '小明', 'age': 30, 'work': '程序员'},
                    'id': 'call_00_YtwvNn5ZnOqpNhmUlpsu7080',
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(
            content='格式化输出成功',
            name='UserInfo',
            id='fb5d90c2-0cf8-4c2f-bffd-5635c65ef05e',
            tool_call_id='call_00_YtwvNn5ZnOqpNhmUlpsu7080'
        )
    ],
    'structured_response': UserInfo(name='小明', age=30, work='程序员')
}
"""
ExceptionType ​

默认情况下,LangChain 会处理结构化输出处理时抛出的两类异常:

  1. MultipleStructuredOutputsError:多结构化输出错误,当返回的工具调用请求数量大于1时,抛出该异常,默认情况下 LangChain 会拦截异常并提醒模型重试。
  2. StructuredOutputValidationError:输出结构化验证错误,当输出格式不符合结构化要求时,抛出上述异常。
python
import os
from typing import Union

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy, MultipleStructuredOutputsError, StructuredOutputValidationError
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field
from rich import print as rprint

load_dotenv(verbose=True)

class UserInfo(BaseModel):
    """用户信息结构体"""
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    work: str = Field(description="工作")

class MethodInfo(BaseModel):
    """事件信息结构体"""
    name: str = Field(description="事件名")
    time: str = Field(description="发生的时间")

agent = create_agent(
    model=init_chat_model(
        model="deepseek:deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    response_format=ToolStrategy(
        schema=Union[UserInfo, MethodInfo],
        tool_message_content="格式化输出成功",
        handle_errors=(MultipleStructuredOutputsError, StructuredOutputValidationError),
    ),
)

res = agent.invoke({
    "messages": [
        HumanMessage("请在下面文本中提取信息:小明是一个30岁的程序员,昨天下午在图书馆捡了一部手机")
    ]
})

rprint(res)

""" 输出结果:
输出同 handle_errors=True,因为当前在 handle_errors 加入了 MultipleStructuredOutputsError 用于捕获工具调用请求数量大于1的情况

如果 handle_errors 中只有 StructuredOutputValidationError,效果等同于 handle_errors=False 效果
"""
callable ​
python
import os
from typing import Union

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy, MultipleStructuredOutputsError, StructuredOutputValidationError
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field
from rich import print as rprint

load_dotenv(verbose=True)

def handle_error_callable(error: Exception):
    """自定义错误处理器"""
    error_str = str(error)
    print(f"捕获到错误类型:{type(error).__name__}")
    print(f"错误详情:{error_str}")

    if isinstance(error, StructuredOutputValidationError):
        return "数据格式有误,请检查字段是否符合要求。"
    elif isinstance(error, MultipleStructuredOutputsError):
        return "检测到多个响应,请选择最相关的一个进行返回。"
    else:
        return error_str

class UserInfo(BaseModel):
    """用户信息结构体"""
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    work: str = Field(description="工作")

class MethodInfo(BaseModel):
    """事件信息结构体"""
    name: str = Field(description="事件名")
    time: str = Field(description="发生的时间")

agent = create_agent(
    model=init_chat_model(
        model="deepseek:deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    response_format=ToolStrategy(
        schema=Union[UserInfo, MethodInfo],
        tool_message_content="格式化输出成功",
        handle_errors=handle_error_callable
    ),
)

res = agent.invoke({
    "messages": [
        HumanMessage("请在下面文本中提取信息:小明是一个30岁的程序员,昨天下午在图书馆捡了一部手机")
    ]
})

rprint(res)

""" 输出结果:
捕获到错误类型:MultipleStructuredOutputsError
错误详情:Model incorrectly returned multiple structured responses (UserInfo, MethodInfo) when only one is expected.
捕获到错误类型:MultipleStructuredOutputsError
错误详情:Model incorrectly returned multiple structured responses (UserInfo, MethodInfo) when only one is expected.
{
    'messages': [
        HumanMessage(
            content='请在下面文本中提取信息:小明是一个30岁的程序员,昨天下午在图书馆捡了一部手机',
            additional_kwargs={},
            response_metadata={},
            id='d3c75502-2544-47b8-b624-796d794ed58f'
        ),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'UserInfo',
                    'args': {'name': '小明', 'age': 30, 'work': '程序员'},
                    'id': 'call_00_2AcocIbQs9Y4n75xobZ97155',
                    'type': 'tool_call'
                },
                {
                    'name': 'MethodInfo',
                    'args': {'name': '捡了一部手机', 'time': '昨天下午'},
                    'id': 'call_01_ucgctThywf1lo899976h7092',
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(
            content='检测到多个响应,请选择最相关的一个进行返回。',
            name='UserInfo',
            id='75500aed-cd08-435b-bbf7-0c46e0c2a30e',
            tool_call_id='call_00_2AcocIbQs9Y4n75xobZ97155'
        ),
        ToolMessage(
            content='检测到多个响应,请选择最相关的一个进行返回。',
            name='MethodInfo',
            id='aeb82ae9-c530-4304-8795-24564a0bda16',
            tool_call_id='call_01_ucgctThywf1lo899976h7092'
        ),
        ...
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'UserInfo',
                    'args': {'name': '小明', 'age': 30, 'work': '程序员'},
                    'id': 'call_00_A1gpQ6lqPDBtvkJaXeF90766',
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(
            content='格式化输出成功',
            name='UserInfo',
            id='38e328c1-d6ff-4aa7-98fe-dfd6917d9dee',
            tool_call_id='call_00_A1gpQ6lqPDBtvkJaXeF90766'
        )
    ],
    'structured_response': UserInfo(name='小明', age=30, work='程序员')
}
"""

流式输出 ​

流式输出说明 ​

流式输出好处:

  • 大型语言模型生成完整响应通常需要几秒钟时间,对于长输出可能达到 10-20 秒,用户期望即时反馈,流式传输让等待过程更加可控。
  • 相比非流式传输需要用户长时间等待完整响应,流式传输可以立即显示文字逐渐出现的效果,大幅降低用户的等待焦虑。

设置方式: 通过“ agent.stream(stream_mode=指定模式) ”来指定。具体模式有:values、updates(默认)、messages、custom、checkpoints、tasks、debug。

输出模式 ​

  1. values: 每个步骤执行后,都会输出完整的状态信息,适用于每一步都要获取完整状态、状态持久化场景。
  2. updates:默认模式。每个步骤执行后,只增量更新状态中发生变化的内容,用于监控 Agent 执行进度,例如观察 Agent 决定调用工具、工具执行结果等步骤。
  3. messages:打字机效果,该模式中会输出流式返回的Token以及相关的元数据。
  4. tasks:该模式会输出当前 task 任务开始和结束的时间,包含任务的结果和错误信息,该模式用于监控任务的生命周期。
  5. debug:该模式与tasks模式类似,比task模式多输出任务步骤、时间戳、task类型(task/task_result),该模式用于调试、监控task任务的生命周期。
  6. checkpoints:该模式中,每当检查点(checkpoint)被创建时会触发输出,输出包含检查点中的状态,用于需要状态持久化、工作流恢复或分布式执行跟踪的高级场景。
  7. custom:开发者通过 get_stream_writer 在工具或节点内部自定义发送的数据,用于输出业务逻辑相关的进度信息(如“已处理10/100条记录”)、自定义日志或指标。
python
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool

load_dotenv(verbose=True)

@tool
def getUserInfo(name: str):
    """查询学生信息

    Args:
        name: 学生姓名
    """
    return {"name": name, "性别": "男", "学号": 260000, "生日": "2000-07-13"}

@tool
def getMark(number: int):
    """查询成绩信息

    Args:
        number: 学号
    """
    return {"number": number, "语文": "40", "数学": "100", "英语": "5"}

agent = create_agent(
    model=init_chat_model(
        model="deepseek:deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    tools=[getUserInfo, getMark],
)

for msg in agent.stream({
    "messages": [
        HumanMessage("查询小明同学的基本信息和成绩信息")
    ]
}, stream_mode="messages"):
    print(msg[0].content, end="", flush=True)

""" 输出结果:
我来帮你查询小明同学的信息。首先需要查询他的基本信息以获取学号。{"name": "小明", "性别": "男", "学号": 260000, "生日": "2000-07-13"}已获取到小明的基本信息,现在根据学号查询成绩信息。{"number": 260000, "语文": "40", "数学": "100", "英语": "5"}小明同学的信息查询完成,结果如下:

**基本信息**
- 姓名:小明
- 性别:男
- 学号:260000
- 生日:2000-07-13

**成绩信息**
| 科目 | 成绩 |
|------|------|
| 语文 | 40 |
| 数学 | 100 |
| 英语 | 5 |

小明同学数学成绩优异(满分100分),但语文和英语成绩偏低,尤其是英语只有5分,建议加强这两科的学习。
"""

总结 ​

模式输出内容使用场景
values每个步骤执行后,都会输出完整的状态信息适用于每一步都要获取完整状态、状态持久化场景
updates每个步骤执行后,只增量更新状态中发生变化的内容用于监控Agent 执行进度,例如观察Agent决定调用工具、工具执行结果等步骤
messages输出流式返回的Token以及相关的元数据(如:来自哪个节点model/tool)实现类似ChatGPT 的打字机效果,为聊天机器人等交互式应用提供最佳的实时体验
tasks输出当前task任务开始和结束的时间,包含任务的结果和错误信息该模式用于监控任务的生命周期
debug与tasks模式类似,比task模式多输出任务步骤、时间戳、task类型(task/task_result)该模式用于调试、监控task任务的生命周期
checkpoints当检查点(checkpoint)被创建时会触发输出,输出包含检查点中的状态用于需要状态持久化、工作流恢复或分布式执行跟踪的高级场景
custom通过get_stream_writer在工具或节点内部自定义发送的数据用于输出业务逻辑相关的进度信息(如“已处理10/100条记录”)、自定义日志或指标
  • 实现 实时对话交互 ,优先选择messages模式。
  • 观察Agent的 思考与执行步骤,优先选择updates模式。
  • 需要查看 每一步状态 优先选择values/tasks/debug模式。
  • 在工具执行时 输出自定义业务 日志优先选择custom模式。

实战:多功能机器人 ​

  1. 天气查询:查询城市天气
  2. 时间查询:获取当前时间、日期计算
  3. 货币转换:多种货币之间转换
  4. 信息搜索:搜索产品、新闻等信息
python
import os
from datetime import datetime
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

load_dotenv(verbose=True)

@tool(parse_docstring=True)
def get_weather(city: str):
    """天气查询工具

    Args:
        city: 城市
    """
    cw = {
        "北京": "多云,15-22℃,空气质量良,湿度 45%",
        "上海": "晴天,18-25℃,空气质量优,湿度 60%",
        "深圳": "小雨,22-28℃,空气质量优,湿度 75%",
        "成都": "阴天,16-23℃,空气质量良,湿度 70%",
        "杭州": "晴天,17-24℃,空气质量优,湿度 55%",
        "广州": "多云,21-29℃,空气质量良,湿度 72%"
    }
    res = cw.get(city)
    if res:
        return f"{city}:{res}"
    else:
        return f"抱歉,暂不支持查询{city}的天气信息。当前支持:北京、上海、深圳、成都、杭州、广州"

@tool(parse_docstring=True)
def get_date(query_type: str):
    """查询日期

    Args:
        query_type: 查询类型,可选值:current(当前时间)date(今日日期)
    """
    now = datetime.now()
    if query_type == "current":
        return now.strftime("当前时间:%Y年%m月%d日 %H:%M:%S")
    elif query_type == "date":
        return now.strftime("今天是:%Y年%m月%d日")
    else:
        return f"不支持的查询类型:{query_type}。支持:current, date"

@tool(parse_docstring=True)
def convert_currency(money: float, source: str, target: str):
    """兑换货币

    Args:
        money: 兑换金额
        source: 来源国家货币
        target: 目标国家货币
    """
    rate = {
        "人民币": 1.0,
        "美元": 0.14,
        "欧元": 0.13,
        "英镑": 0.11,
        "日元": 20.8,
        "港币": 1.09,
    }
    rate_keys = rate.keys()
    if source not in rate_keys or target not in rate_keys:
        return "当前工具不支持该货币"
    return f"{money}{source} = {money / rate[source] * rate[target]}{target}"

@tool(parse_docstring=True)
def search_info(keyword: str, search_type: str):
    """搜索信息

    Args:
        keyword: 搜索的关键字
        search_type: 搜索类型。可选值:product-商品 news-新闻 all-全部
    """
    info = {
        "product": {
            "手机": "iPhone 15 (¥5999), 小米14 (¥3999), 华为Mate60 (¥6999)",
            "笔记本": "MacBook Pro (¥12999), ThinkPad X1 (¥9999), 华为MateBook(¥7999)",
            "耳机": "AirPods Pro (¥1999), Sony WH-1000XM5 (¥2499)"
        },
        "news": {
            "AI": "1. GPT-5 即将发布 2. AI 芯片市场增长 30% 3. 新AI法规出台",
            "科技": "1. 量子计算新突破 2. 6G 技术测试 3. 新能源汽车销量创新高"
        }
    }
    source = (info["product"] | info["news"]) if search_type == "all" else info[search_type]
    res = []

    for key, value in source.items():
        if keyword in key:
            res.append(f"{key}: {value}")

    if res:
        return "\n".join(res)
    else:
        return f"未找到{keyword}相关信息"

agent = create_agent(
    model = init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    tools=[get_weather, get_date, convert_currency, search_info],
    system_prompt="""你是一个多功能智能助手,可以帮助用户:
        🌤 查询天气:使用 get_weather 工具
        ⏰ 时间查询:使用 get_date 工具
        💱 货币转换:使用 convert_currency 工具
        🔍 信息搜索:使用 search_info 工具
        
        重要提示:
        1. 仔细阅读用户问题,确定需要使用哪个工具
        2. 如果需要多个工具,按顺序调用
        3. 总是用友好、专业的语气回答
        4. 如果工具返回了数据,要用通俗易懂的语言解释给用户
        5. 如果无法完成任务,诚实地告诉用户原因
        请始终使用中文回答。
    """
)

messages = []
count = 0

while True:
    user_input = input("请输入问题:")
    if user_input == "end":
        print("程序运行结束")
        break
    elif user_input == "clear":
        messages = []
    else:
        messages.append(user_input)
        count += 1
        print("=" * 10, f"第{count}轮", "=" * 10)
        for chunk in agent.stream({"messages": messages}, stream_mode="messages"):
            token, metadata = chunk

            if metadata.get("langgraph_node") == "tools" or getattr(token, "tool_calls", []):
                continue

            if token.content:
                print(token.content, end="")
        print("\n\n")

基于 MIT 许可发布