搜索文档
理解 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()。
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": "问题"}]
})模型的传入方式
传入字符串
from langchain.agents import create_agent
agent = create_agent(
model="ollama:qwen3.5"
)
print(type(agent))
""" 输出结果:
<class 'langgraph.graph.state.CompiledStateGraph'>
"""传入模型对象
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字段的值。
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 会在需要调用工具时自动调用。

调用自定义工具
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': {}
}
)
]
}
"""接入内置工具
绑定内置的 TavilySearch 搜索工具,可以借助 Tavily 进行网络搜索和信息爬取。
这里我们需要先在 tavily 官网注册并获得API-KEY(每月有免费额度):https://www.tavily.com/
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可以在工具调用结果不满足要求时,自主重试。
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 为什么没有调用工具?
- 工具的 docstring 不清晰
- 问题表述不明确
- 模型认为不需要工具
面试题:Agent 选错工具?
原因:
- 多个工具的功能描述相似
- 工具太多导致混淆
解决:
- 只给必要的工具
- 工具描述要有明确区分
- 在 system_prompt 中说明工具使用场景
面试题:如何知道 Agent 何时完成?
当 AIMessage 不包含 tool_calls 时视为完成。
面试题:Agent 可以调用多少次工具?
默认没有限制,直到得到最终答案。但可能会:
- 超时
- 达到 token 限制
- 模型决定停止
面试题:如何限制工具调用次数?
LangChain 1.0 的 create_agent 默认使用 LangGraph,可以通过配置限制:
config = {
"recursion_limit": 5 # 最多 5 步
}
res = agent.invoke(input, config=config)设置 Agent 名称
创建 Agent 时可以通过 name 属性配置 Agent 名称。
使用场景:
- 多 Agent 交互:name 会作为 Agent 的唯一标识。
- 流式输出归因;在启用流式输出时,name 可用于标识当前输出内容来自哪个Agent。
- 消息身份标记:设置 name 后,Agent 产生的 AIMessage 会携带对应的name信息。
- 调试与trace可读性。
- 组件化封装。
- 前端展示与运行态可观测性。
- 作为稳定的运行时身份标识。
设置系统提示词
创建 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”工具调用的方式实现结构化输出。
此策略兼容绝大多数 支持工具调用 的现代模型,其核心原理是动态创建一个 "虚拟工具",该工具的输入参数对应着期望的数据结构。
当模型需要生成最终答案时,系统会引导模型 "调用"这个虚拟工具 ,从而间接产生符合要求的结构化数据。
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 联合类型
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='程序员'
"""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 字段
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
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='程序员')
}
"""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'
"""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 会处理结构化输出处理时抛出的两类异常:
- MultipleStructuredOutputsError:多结构化输出错误,当返回的工具调用请求数量大于1时,抛出该异常,默认情况下 LangChain 会拦截异常并提醒模型重试。
- StructuredOutputValidationError:输出结构化验证错误,当输出格式不符合结构化要求时,抛出上述异常。
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
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。
输出模式
- values: 每个步骤执行后,都会输出完整的状态信息,适用于每一步都要获取完整状态、状态持久化场景。
- updates:默认模式。每个步骤执行后,只增量更新状态中发生变化的内容,用于监控 Agent 执行进度,例如观察 Agent 决定调用工具、工具执行结果等步骤。
- messages:打字机效果,该模式中会输出流式返回的Token以及相关的元数据。
- tasks:该模式会输出当前 task 任务开始和结束的时间,包含任务的结果和错误信息,该模式用于监控任务的生命周期。
- debug:该模式与tasks模式类似,比task模式多输出任务步骤、时间戳、task类型(task/task_result),该模式用于调试、监控task任务的生命周期。
- checkpoints:该模式中,每当检查点(checkpoint)被创建时会触发输出,输出包含检查点中的状态,用于需要状态持久化、工作流恢复或分布式执行跟踪的高级场景。
- custom:开发者通过
get_stream_writer在工具或节点内部自定义发送的数据,用于输出业务逻辑相关的进度信息(如“已处理10/100条记录”)、自定义日志或指标。
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模式。
实战:多功能机器人
- 天气查询:查询城市天气
- 时间查询:获取当前时间、日期计算
- 货币转换:多种货币之间转换
- 信息搜索:搜索产品、新闻等信息
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")