LangChain 07 - 智能体学习导航与实战
LangChain 07 - 智能体学习导航与实战
模块 1:你只需要记住这些
1. 什么是 Agent(智能体)
Agent 是一个能够自主完成多步任务的 AI 系统。它不是一个简单的"输入→输出"函数,而是一个能够:
- 感知:接收用户输入
- 决策:判断需要调用什么工具
- 行动:执行工具调用
- 观察:获取工具返回结果
- 循环:根据结果继续推理,直到完成任务
的循环系统。
和普通模型调用的核心区别:
| 对比维度 | 普通模型调用 | Agent |
|---|---|---|
| 调用次数 | 一次 | 多次(循环) |
| 是否使用工具 | 否 | 是 |
| 是否自主决策 | 否,你写好流程 | 是,Agent 自己决定 |
| 输出结构 | 直接返回文本 | 返回完整消息列表 |
2. Agent 的四个核心组件
Agent 不是凭空运作的,它依赖四个核心组件:
① Action(行动)— 必需
Action 是 Agent 执行的具体操作,最核心的 Action 就是调用 LLM 进行推理。没有 Action,Agent 就无法"思考"。
② Tool(工具)— 几乎总有
Tool 是 Agent 与外部世界交互的手段。没有工具的 Agent 只能靠模型自身的知识回答问题,能力受限。工具可以是:
- 自定义函数(用
@tool装饰器定义) - 内置工具(如 TavilySearch 搜索工具)
- 第三方工具
③ Planning(规划)— 有条件
Planning 是 Agent 对任务的分解和规划能力。当任务简单时,Agent 可能不需要规划就直接行动;当任务复杂时,Agent 会先拆解任务再逐步执行。
⚠️ 注意:Planning 是否生效取决于模型本身的能力,强模型规划能力更强。
④ Memory(记忆)— 最易省略
Memory 是 Agent 对上下文的记忆。默认情况下,Agent 在单次调用中有短期记忆(对话历史),但跨会话的长期记忆需要额外配置(如 checkpointer)。
四个组件的必要性排序:
Action(必需)> Tool(几乎总有)> Planning(有条件)> Memory(最易省略)
3. create_agent() — 统一创建入口
LangChain 1.x 使用 create_agent() 作为统一的 Agent 创建函数,取代了旧版的 initialize_agent、AgentExecutor 等多种方式。
from langchain.agents import create_agent
agent = create_agent(
model=model, # 模型(必需)
tools=[tool1, tool2], # 工具列表(几乎必需)
)
💡 核心参数速查:
| 参数 | 类型 | 说明 |
|---|---|---|
model |
str 或 Model 对象 | 使用的 LLM(必需) |
tools |
list | 绑定的工具列表 |
name |
str | Agent 名称 |
system_prompt |
str 或 SystemMessage | 系统提示词 |
response_format |
策略对象 | 结构化输出策略 |
checkpointer |
Checkpointer | 检查点存储(用于记忆/恢复) |
模型传入的两种方式:
方式 1 — 传字符串(内部自动初始化):
agent = create_agent(
model="openai:gpt-4o-mini",
tools=[my_tool],
)
方式 2 — 传模型对象(推荐,更灵活):
from langchain.chat_models import init_chat_model
model = init_chat_model(
model="deepseek-chat",
model_provider="openai",
api_key=xxx,
base_url=xxx,
)
agent = create_agent(model=model, tools=[my_tool])
推荐用方式 2,因为可以精细控制 temperature、api_key 等参数。
4. Agent 的调用方式
Agent 的调用和普通模型不同,输入和输出都是特定格式:
输入格式:
result = agent.invoke({
"messages": [{"role": "user", "content": "你的问题"}]
})
也可以使用消息对象:
from langchain_core.messages import HumanMessage
result = agent.invoke({
"messages": [HumanMessage(content="你的问题")]
})
输出格式:
返回一个字典,核心是 messages 列表,包含了完整的对话历史:
{
"messages": [
HumanMessage(content="..."), # 用户消息
AIMessage(content="", tool_calls=[...]), # AI 决定调用工具
ToolMessage(content="..."), # 工具返回结果
AIMessage(content="最终答案") # AI 给出最终回答
]
}
获取最终答案的方法:
# 方法 1:取最后一条消息
final_answer = result["messages"][-1].content
# 方法 2:反向查找最后一条有内容的 AI 消息
for msg in reversed(result["messages"]):
if msg.type == "ai" and msg.content:
final_answer = msg.content
break
5. Agent 的本质 — LangGraph 图
这是理解 Agent 的关键:
Agent 的本质是 LangGraph 的 CompiledStateGraph 实例。
这意味着 Agent 不是一段简单的函数调用,而是一张"状态图":
[开始] → [模型推理节点] → [需要调用工具?]
↓ 是
[工具执行节点] → [模型推理节点] → ...
↓ 否
[结束,返回结果]
你可以用 type(agent) 验证:
print(type(agent))
#
这个知识点很重要,因为它解释了:
- 为什么 Agent 能循环推理(图可以有环)
- 为什么 Agent 有流式输出(图可以逐步输出节点结果)
- 为什么 Agent 有 checkpointer(图可以保存状态)
6. 绑定工具的三种方式
方式 1 — @tool 自定义工具:
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""获取指定城市的天气信息"""
# 函数体
return f"{city}:晴天,25℃"
agent = create_agent(model=model, tools=[get_weather])
关键点:
@tool装饰器把普通函数变成 LangChain 工具- 函数的 docstring 就是工具描述,LLM 靠它判断"什么时候该用这个工具"
- 参数的类型注解会自动变成工具的参数 Schema
方式 2 — TavilySearch 内置搜索工具:
from langchain_tavily import TavilySearch
search_tool = TilySearch(
max_results=3,
topic="general"
)
agent = create_agent(model=model, tools=[search_tool])
TavilySearch 是一个专门为 LLM 设计的搜索 API,需要配置 TAVILY_API_KEY。
方式 3 — 多工具组合:
agent = create_agent(
model=model,
tools=[get_weather, calculator, search_tool, convert_currency],
)
Agent 会根据用户问题自主选择合适的工具,甚至可以连续调用多个工具。
7. ReAct 循环 — Agent 的核心工作流
ReAct = Reasoning + Acting(推理 + 行动)
完整流程:
用户提问
↓
[思考] AI 分析问题,决定调用什么工具
↓
[行动] AI 发起 tool_calls
↓
[观察] 工具执行,返回 ToolMessage
↓
[思考] AI 根据工具结果,决定是否继续调用工具
↓
...(循环直到 AI 认为可以回答)
↓
[最终答案] AI 输出 content,不再调用工具
对应的消息流转:
HumanMessage("北京天气怎么样")
→ AIMessage(tool_calls=[{name: "get_weather", args: {city: "北京"}}])
→ ToolMessage(content="北京:晴天,25℃")
→ AIMessage(content="北京今天晴天,气温25度")
关键理解:
- AIMessage 的
tool_calls字段:记录 AI 决定调用哪个工具、传什么参数 - ToolMessage 的
tool_call_id:和 AIMessage 的 tool_calls 一一对应 - 最后一条 AIMessage 的
content就是最终答案
8. 重试机制
Agent 有一个重要的能力:当工具返回的结果不满足要求时,Agent 会自主重试。
比如工具返回了不完整的数据,Agent 不会直接把不完整的数据丢给用户,而是会:
- 分析工具返回结果
- 判断是否满足需求
- 如果不满足,重新组织参数再次调用工具
- 直到获得满意的结果或达到最大重试次数
这个机制让 Agent 比简单的"调一次工具就完事"更健壮。
9. 常见问题
问题 1:API 密钥配置
Agent 调用模型需要 API Key,通常通过 .env 文件配置:
DEEPSEEK_API_KEY=sk-xxxx
DEEPSEEK_BASE_URL=https://api.deepseek.com
TAVILY_API_KEY=tvly-xxxx
用 load_dotenv() 加载。
问题 2:工具描述(docstring)的重要性
LLM 选择工具时,完全依赖工具的描述(docstring)。如果描述写得不好:
- 描述太模糊 → LLM 不知道什么时候该用
- 描述太相似 → LLM 在多个工具间犹豫
- 没有描述 → LLM 可能完全忽略这个工具
好的工具描述应该包含:
- 这个工具做什么
- 什么时候应该用
- 参数含义
- 返回值格式
- 示例(可选但推荐)
10. 高级用法 — 设置 Agent 名称
agent = create_agent(
model=model,
tools=[my_tool],
name="my_custom_agent",
)
设置名称后,在 LangSmith 追踪中可以更容易地识别这个 Agent。
11. 高级用法 — 系统提示词
系统提示词用于定义 Agent 的"人设"和行为规范。
方式 1 — 传字符串:
agent = create_agent(
model=model,
tools=[my_tool],
system_prompt="你是一个专业的客服助手,请用中文回答。",
)
方式 2 — 传 SystemMessage 对象:
from langchain_core.messages import SystemMessage
agent = create_agent(
model=model,
tools=[my_tool],
system_prompt=SystemMessage(content="你是一个专业的客服助手,请用中文回答。"),
)
两种方式效果相同,字符串方式更简洁,SystemMessage 方式可以在 content 中使用更复杂的结构(如多模态内容)。
系统提示词的价值:
- 定义 Agent 角色(客服、分析师、翻译…)
- 规定回答语言、风格、格式
- 指导工具使用策略
- 约束行为边界
12. 结构化输出 — response_format
Agent 支持通过 response_format 参数让最终输出变为结构化数据。
四种策略:
| 策略 | 说明 | 适用场景 |
|---|---|---|
ProviderStrategy |
依赖模型提供商原生结构化输出能力 | 模型支持时优先使用 |
ToolStrategy |
通过工具调用实现结构化输出 | 需要更多控制时使用 |
AutoStrategy |
自动选择最佳策略 | 想省事时用 |
None(默认) |
不做结构化输出 | 普通对话场景 |
使用示例:
from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.agents.output_policies import ProviderStrategy, ToolStrategy
class WeatherSummary(BaseModel):
"""天气摘要"""
city: str = Field(description="城市")
temperature: float = Field(description="温度")
description: str = Field(description="天气描述")
# 策略 1: ProviderStrategy
agent = create_agent(
model=model,
tools=[get_weather],
response_format=ProviderStrategy(schema=WeatherSummary),
)
# 策略 2: ToolStrategy
agent = create_agent(
model=model,
tools=[get_weather],
response_format=ToolStrategy(schema=WeatherSummary),
)
获取结构化输出:
result = agent.invoke({"messages": [...]})
# result["structured_response"] 就是 WeatherSummary 实例
structured = result["structured_response"]
print(structured.city, structured.temperature)
13. ToolStrategy 详解
ToolStrategy 是最灵活的结构化输出策略,有三个关键参数:
参数 1 — schema:定义输出结构
支持五种 Schema 类型:
# 1. Pydantic BaseModel
class Info(BaseModel):
name: str = Field(description="名称")
# 2. TypedDict
class InfoTD(TypedDict):
name: Annotated[str, "名称"]
# 3. JSON Schema(字典)
info_schema = {"type": "object", "properties": {"name": {"type": "string"}}}
# 4. @dataclass
@dataclass
class InfoDC:
name: str = Field(description="名称")
# 5. Union 多 Schema(让 Agent 选择输出哪种结构)
from typing import Union
agent = create_agent(
model=model,
tools=[...],
response_format=ToolStrategy(schema=Union[InfoA, InfoB]),
)
Union 模式的特殊之处:Agent 会根据用户输入自主判断应该用哪种结构输出。
参数 2 — tool_message_content:工具消息内容
控制结构化输出工具被调用后,返回的 ToolMessage 包含什么内容:
ToolStrategy(
schema=WeatherSummary,
tool_message_content="结构化输出已生成", # 自定义消息内容
)
参数 3 — handle_errors:错误处理
控制结构化输出解析失败时的行为,支持五种配置:
# 1. True(默认):捕获异常,返回错误信息作为工具结果
ToolStrategy(schema=Info, handle_errors=True)
# 2. False:不处理异常,直接抛出
ToolStrategy(schema=Info, handle_errors=False)
# 3. 固定字符串:用固定消息替代错误信息
ToolStrategy(schema=Info, handle_errors="输出格式有误,请重试")
# 4. 异常类型元组:只捕获指定异常
ToolStrategy(schema=Info, handle_errors=(ValueError, TypeError))
# 5. 自定义函数:完全控制错误处理逻辑
def my_error_handler(error: Exception) -> str:
return f"解析失败:{error},请调整格式后重试"
ToolStrategy(schema=Info, handle_errors=my_error_handler)
14. 流式输出 — agent.stream()
Agent 支持流式输出,通过 stream_mode 参数控制输出模式。
七种输出模式总览:
| 模式 | 输出内容 | 使用场景 |
|---|---|---|
values |
每步执行后的完整状态 | 需要每步完整状态、状态持久化 |
updates(默认) |
每步增量更新(只含变化部分) | 监控 Agent 执行进度 |
messages |
流式 Token + 元数据 | ChatGPT 打字机效果 |
tasks |
任务开始/结束时间 + 结果 | 监控任务生命周期 |
debug |
比 tasks 多步骤、时间戳、task类型 | 调试任务生命周期 |
checkpoints |
检查点被创建时的状态 | 状态持久化、工作流恢复 |
custom |
工具内部自定义发送的数据 | 业务进度、自定义日志 |
各模式代码示例:
# --- values 模式:每步输出完整状态 ---
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "你好"}]},
stream_mode="values"
):
print(chunk)
print("-" * 50)
# --- updates 模式(默认):每步输出增量更新 ---
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "你好"}]},
stream_mode="updates" # 可省略,默认就是 updates
):
print(chunk)
print("-" * 50)
# --- messages 模式:流式 Token ---
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "你好"}]},
stream_mode="messages"
):
print(chunk)
print("-" * 50)
# --- tasks 模式:任务生命周期 ---
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "你好"}]},
stream_mode="tasks"
):
print(chunk)
print("-" * 50)
# --- debug 模式:调试任务 ---
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "你好"}]},
stream_mode="debug"
):
print(chunk)
print("-" * 50)
checkpoints 模式 — 需要配置 checkpointer:
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
agent = create_agent(
model=model,
tools=[my_tool],
checkpointer=checkpointer, # 启用检查点
)
config = {"configurable": {"thread_id": "session01"}}
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "你好"}]},
config=config,
stream_mode="checkpoints"
):
print(chunk)
print("-" * 50)
custom 模式 — 工具内部自定义输出:
from langgraph.config import get_stream_writer
@tool
def generate_report() -> str:
"""生成报告"""
writer = get_stream_writer()
writer({"type": "报告", "message": "开始生成"})
for i in range(1, 4):
time.sleep(0.5)
writer({"type": "报告", "message": f"进度:{i * 25}%"})
writer({"type": "报告", "message": "完成"})
return "报告已生成"
agent = create_agent(model=model, tools=[generate_report])
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "生成报告"}]},
stream_mode="custom"
):
print(chunk)
print("-" * 50)
多模式组合 — 同时输出多种模式:
for stream_mode, chunk in agent.stream(
{"messages": [{"role": "user", "content": "你好"}]},
stream_mode=["tasks", "updates"] # 传入列表
):
print(f"当前流模式: {stream_mode}, 当前数据: {chunk}")
print("-" * 50)
多模式组合时,stream() 返回的是 (stream_mode, chunk) 元组,你可以通过 stream_mode 判断当前数据来自哪种模式。
模式选择指南:
实时对话交互 → messages
观察思考与执行 → updates
查看每步完整状态 → values / tasks / debug
状态持久化/恢复 → checkpoints
工具内业务日志 → custom
想同时看多种 → 多模式组合
模块 2:带教式理解
1. 为什么说 Agent 的本质是"图"
普通模型调用是线性的:输入 → 模型 → 输出。
Agent 不是线性的,它是循环的:输入 → 模型推理 → 需要工具?→ 是 → 执行工具 → 模型推理 → ... → 输出。
这种"有条件跳转 + 循环"的结构,在计算机科学里就叫"状态图"(State Graph)。LangChain 的 Agent 底层就是用 LangGraph 实现的。
理解这一点的好处:
- 你可以用 LangGraph 的 API 进一步定制 Agent(添加节点、修改边)
- 你能理解为什么 Agent 有 checkpointer(图可以保存状态)
- 你能理解为什么 Agent 有流式输出(图可以逐步输出节点结果)
2. 为什么工具的 docstring 这么重要
Agent 选择工具的逻辑是:把所有工具的描述(docstring)和参数 Schema 放进 prompt,让 LLM 决定调用哪个。
所以 docstring 不是给开发者看的注释,而是给 LLM 看的说明书。
对比:
# 差的描述 — LLM 不知道什么时候用
@tool
def search(query: str) -> str:
"""搜索"""
...
# 好的描述 — LLM 知道什么时候用、怎么用
@tool
def search(query: str) -> str:
"""搜索互联网获取最新信息。
当用户询问实时信息、新闻、或你不确定的事实 时使用此工具。
Args:
query: 搜索关键词
Returns:
搜索结果摘要
"""
...
3. 为什么用 messages 列表而不是简单的字符串
Agent 的输入是 {"messages": [...]},输出也包含 messages 列表。为什么不用简单的字符串?
因为 Agent 的执行过程中会产生多种类型的消息:
HumanMessage— 用户输入AIMessage— AI 的回复(可能含 tool_calls)ToolMessage— 工具执行结果
这些消息需要按顺序保存,因为下一轮推理需要完整的历史上下文。如果只存字符串,就会丢失"谁说了什么"和"工具返回了什么"的信息。
这也是为什么封装 Agent 类时要维护 self.messages——它就是对话的完整记忆。
4. 反向查找最终答案的技巧
PDF 中获取最终答案的写法是:
for msg in reversed(self.messages):
if msg.type == "ai" and msg.content:
return msg.content
为什么要 reversed + msg.content 判断?
因为消息列表中可能有多个 AIMessage:
- 第一个 AIMessage 可能只有 tool_calls,content 为空
- 中间的 AIMessage 也可能有 tool_calls
- 最后一个 AIMessage 才是最终答案,content 有值
所以反向遍历,找到第一个 type == "ai" 且 content 非空的消息,就是最终答案。
5. response_format 的 ProviderStrategy vs ToolStrategy 区别
两种策略实现结构化输出的底层机制不同:
ProviderStrategy:
- 依赖模型提供商的原生结构化输出能力
- 相当于直接告诉模型"你必须按这个格式输出"
- 类似第 6 章学的
with_structured_output() - 前提:模型必须支持
ToolStrategy:
- 通过"添加一个特殊工具"来实现
- Agent 调用这个工具时,参数就是结构化数据
- 更灵活:支持 Union 多 Schema、错误处理等
- 兼容性更好:不依赖模型原生能力
选择直觉:
- 模型支持且需求简单 → ProviderStrategy
- 需要多 Schema 选择、错误处理 → ToolStrategy
- 不确定 → AutoStrategy(自动选)
6. handle_errors 为什么有这么多配置选项
因为不同的应用场景对错误处理的要求不同:
- True(默认):适合大多数场景,Agent 自己看到错误信息后可以重试
- False:适合调试阶段,想立刻发现格式问题
- 固定字符串:适合不想暴露底层错误细节给 Agent 的场景
- 异常类型元组:适合只想捕获特定异常的精细控制
- 自定义函数:适合需要根据错误类型返回不同处理逻辑的复杂场景
最常用的是 True(让 Agent 自己处理错误并重试)和自定义函数(完全控制)。
7. 流式输出七种模式怎么记
按"输出什么"分三组:
第一组 — 输出状态:
values:完整状态(每次都是全量)updates:增量更新(只含变化部分)checkpoints:检查点状态(需要配置 checkpointer)
第二组 — 输出 Token/任务:
messages:Token 流(打字机效果)tasks:任务生命周期debug:任务详情(比 tasks 多时间戳等)
第三组 — 输出自定义:
custom:工具内部自定义数据
记忆口诀:
状态看 values/updates,实时看 messages,调试看 tasks/debug,持久化看 checkpoints,自定义看 custom。
8. custom 模式的 get_stream_writer 是什么
get_stream_writer() 是 LangGraph 提供的一个函数,它返回一个"写入器"。
在工具函数内部调用 writer(data),数据就会通过 agent.stream(stream_mode="custom") 输出给调用方。
这就像是工具函数内部开了一个"实时通信通道":
@tool
def long_running_task() -> str:
"""耗时任务"""
writer = get_stream_writer()
writer("开始处理...") # 调用方能立刻收到
for i in range(100):
# ... 做一些工作 ...
writer(f"已处理 {i+1}/100") # 调方能实时看到进度
writer("处理完成")
return "结果"
适用场景:
- 耗时工具的进度报告
- 工具内部的关键事件通知
- 自定义业务指标输出
⚠️ 注意:writer 的参数可以是任意可序列化的 Python 对象(字典、字符串等),但建议用字典以便包含多个字段。
9. 多模式组合时为什么返回元组
当 stream_mode=["tasks", "updates"] 时,stream() 返回 (stream_mode, chunk) 元组。
因为不同模式的输出内容不同,你需要知道当前 chunk 来自哪种模式才能正确解析:
for stream_mode, chunk in agent.stream(..., stream_mode=["tasks", "updates"]):
if stream_mode == "tasks":
# 处理任务信息
print(f"任务: {chunk}")
elif stream_mode == "updates":
# 处理更新信息
print(f"更新: {chunk}")
这种设计让你在一个循环里同时获取多种维度的信息,而不需要调用多次 stream()。
这一章结束后,你应该能自己回答
1、Agent 和普通模型调用的核心区别是什么?
Agent 能自主决策、调用工具、循环推理,直到完成任务。普通模型调用是"一次输入一次输出",不会使用工具,也不会自主决定下一步做什么。Agent 的输入是 {"messages": [...]},输出包含完整的消息列表(含 HumanMessage、AIMessage、ToolMessage)。
2、Agent 的四个核心组件是什么?哪些是必需的?
四个核心组件是 Action(行动)、Tool(工具)、Planning(规划)、Memory(记忆)。Action 是必需的(没有 Action 就无法推理),Tool 几乎总有(没有工具 Agent 能力受限),Planning 有条件(取决于模型能力和任务复杂度),Memory 最易省略(默认只有短期记忆)。
3、Agent 的本质是什么数据结构?
Agent 的本质是 LangGraph 的 CompiledStateGraph 实例——一张状态图。图中有节点(模型推理节点、工具执行节点)和边(条件跳转),支持循环(工具调用后回到模型推理)。这也解释了 Agent 为什么能循环推理、有流式输出、有 checkpointer。
4、ReAct 循环的完整消息流程是什么?
HumanMessage(用户提问)→ AIMessage(tool_calls=[…])(AI 决定调用工具)→ ToolMessage(content=“…”)(工具返回结果)→ AIMessage(tool_calls=[…])(AI 可能继续调用工具)→ … → AIMessage(content=“最终答案”)(AI 给出回答,不再调用工具)。
5、ToolStrategy 的 handle_errors 有哪些配置方式?
五种:① True(默认,捕获异常返回错误信息)② False(不处理,直接抛出)③ 固定字符串(用固定消息替代错误)④ 异常类型元组(只捕获指定异常)⑤ 自定义函数(完全控制错误处理逻辑)。最常用的是 True(让 Agent 自动重试)和自定义函数。
6、流式输出的七种模式分别适合什么场景?
values — 需要每步完整状态;updates(默认)— 监控执行进度;messages — 打字机效果;tasks — 监控任务生命周期;debug — 调试任务详情;checkpoints — 状态持久化/恢复;custom — 工具内业务日志。可以组合使用,如 stream_mode=["tasks", "updates"]。
7、custom 模式的 get_stream_writer 怎么用?
在工具函数内部调用 get_stream_writer() 获取写入器,然后调用 writer(data) 发送数据。调用方通过 agent.stream(stream_mode="custom") 接收。适合耗时工具的进度报告、关键事件通知、自定义指标输出。writer 参数可以是字典或字符串等可序列化对象。
小测试
1. create_agent() 的 model 参数可以传什么类型?
可以传字符串(如 "openai:gpt-4o-mini",内部自动初始化)或模型对象(通过 init_chat_model() 创建)。推荐传模型对象,可以精细控制 temperature、api_key 等参数。
2. Agent 的输出中,如何获取最终答案?
输出是一个字典,result["messages"] 包含完整消息列表。取最后一条消息的 content 即可,或反向查找最后一条 type == "ai" 且 content 非空的消息(因为中间的 AIMessage 可能只有 tool_calls 而 content 为空)。
3. ToolStrategy 的 schema 参数支持哪几种类型?
五种:① Pydantic BaseModel ② TypedDict ③ JSON Schema(字典)④ @dataclass ⑤ Union 多 Schema(让 Agent 自主选择输出哪种结构)。Union 模式是 ToolStrategy 独有的高级特性。
4. checkpoints 模式需要什么前置条件?
需要配置 checkpointer,例如 InMemorySaver()。创建 Agent 时传入 checkpointer=checkpointer,调用 stream() 时传入 config={"configurable": {"thread_id": "xxx"}}。没有 checkpointer 就无法使用 checkpoints 模式。
5. 多模式组合时,stream() 的返回值是什么格式?
返回 (stream_mode, chunk) 元组。stream_mode 是字符串(如 "tasks"、"updates"),chunk 是该模式的数据。通过遍历 for stream_mode, chunk in agent.stream(...) 来区分不同模式的输出。









