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_agentAgentExecutor 等多种方式。

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 不会直接把不完整的数据丢给用户,而是会:

  1. 分析工具返回结果
  2. 判断是否满足需求
  3. 如果不满足,重新组织参数再次调用工具
  4. 直到获得满意的结果或达到最大重试次数

这个机制让 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(...) 来区分不同模式的输出。