1.理解 Agent

1.1 什么是 Agent？

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

Agent 的关键能力：

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

1.2 Agent 的核心组件

实际开发中几个要素并不需要同时出现，一句话总结：

必须的：行动（Action）

几乎总是存在的：工具（Tool）

有条件存在的：规划决策（Planning）

最容易被省略的：记忆（Memory）

1.3 Agent 的创建和调用

当下所有 Agent 的创建方式统一为一个入口：create_agent( )

同时在底层通过“中间件机制（Middleware）”和“标准模型接口（invoke / stream）”实现全局统一。这让框架更轻、更稳，也更易于被集成到其他 Agent 平台中

用法如下：

2. Agent 基本用法1：模型的传入方式

在 LangChain 1.2 中， create_agent 是构建智能体的核心方式，底层基于 LangGraph 实现

更多参考：create_agent | langchain | LangChain Reference

模型是 Agent 的“大脑”，负责决策和推理。根据模型传入 agent 方式的不同，分为两种方式：

• 传入模型字符串
• 传入模型对象

2.1 传入模型字符串

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

语法如下：

agent 本质：LangGraph 的CompiledStateGraph 实例，底层实现是一个图结构

通过上述代码最后一行代码可以显示 agent 的图结构

2.2 传入模型对象

用法如下：

from langchain.chat_models import init_chat_model from dotenv import load_dotenv import os from rich import print as rprint from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langchain_core.tools import tool from langchain.agents import create_agent # 1.加载配置信息 load_dotenv(override=True) api_key = os.getenv("BAILIAN_API_KEY") base_url = os.getenv("BAILIAN_API_BASE") # 2.初始化模型 llm = init_chat_model( model_provider="openai", # 模型提供商 model="deepseek-v4.1-flash", # 模型名称 api_key=api_key, # API Key base_url=base_url, # API Base URL ) # 3.创建智能体 agent = create_agent(llm) # 4.查看智能体 print(type(agent)) from IPython.display import Image, display display(Image(agent.get_graph().draw_mermaid_png()))

输出与上文一致

3. Agent的基本用法2：如何调用Agent

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

• 输入：传入的参数为字典类型，字典内通过 messages 字段传递消息列表。即：“ {"messages": [{"role": "...", "content": "..."}]} ”
• 输出：通过 invoke 调用Agent，底层可能会经历多轮交互，返回的是完整的消息列表，被封装在字典中，是 messages 字段的值

案例：

from langchain.chat_models import init_chat_model from dotenv import load_dotenv import os from rich import print as rprint from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langchain_core.tools import tool from langchain.agents import create_agent @tool(parse_docstring=True) def get_weather(city: str) -> str: """获取当地天气信息 Args: city (str): 城市名称 Returns: str: 天气信息 """ return "晴天, 20度℃" @tool(parse_docstring=True) def get_history(city: str) -> str: """获取当地历史信息 Args: city (str): 城市名称 Returns: str: 历史信息 """ return "历史悠久" # 1.加载配置信息 load_dotenv(override=True) api_key = os.getenv("BAILIAN_API_KEY") base_url = os.getenv("BAILIAN_API_BASE") # 2.初始化模型 llm = init_chat_model( model_provider="openai", # 模型提供商 model="deepseek-v4.1-flash", # 模型名称 api_key=api_key, # API Key base_url=base_url, # API Base URL ) # 3.创建智能体 agent = create_agent( model=llm, tools=[get_weather, get_history], system_prompt="Agent 的行为指令" ) # 4.调用智能体 response = agent.invoke({ "messages":[{"role": "user", "content": "今天北京的天气如何？北京这座城市历史如何?"}], }) rprint(response)

输出如下：

{ 'messages': [ HumanMessage( content='今天北京的天气如何？北京这座城市历史如何?', additional_kwargs={}, response_metadata={}, id='6c52411e-4deb-490f-a8f3-2848ecd9ff2b' ), AIMessage( content='我来帮您查询北京的天气和历史信息。', additional_kwargs={'refusal': None}, response_metadata={ 'token_usage': { 'completion_tokens': 96, 'prompt_tokens': 361, 'total_tokens': 457, 'completion_tokens_details': { 'accepted_prediction_tokens': None, 'audio_tokens': None, 'reasoning_tokens': 21, 'rejected_prediction_tokens': None, 'text_tokens': 75 }, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0, 'text_tokens': 361} }, 'model_provider': 'openai', 'model_name': 'deepseek-v4.1-flash', 'system_fingerprint': None, 'id': 'chatcmpl-dede3d58-c465-9a17-a8ec-1d3ee26c24d5', 'finish_reason': 'tool_calls', 'logprobs': None }, id='lc_run--01a15860-34f0-7993-b616-8c9dd5e944e8-0', tool_calls=[ { 'name': 'get_weather', 'args': {'city': '北京'}, 'id': 'call_7da856d6f90446d7aab382d2', 'type': 'tool_call' }, { 'name': 'get_history', 'args': {'city': '北京'}, 'id': 'call_ad44233baf4644bd9ae887ae', 'type': 'tool_call' } ], invalid_tool_calls=[], usage_metadata={ 'input_tokens': 361, 'output_tokens': 96, 'total_tokens': 457, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 21} } ), ToolMessage( content='晴天, 20度℃', name='get_weather', id='e60e1afd-f2e2-4e01-8d4f-292cd78c1d3f', tool_call_id='call_7da856d6f90446d7aab382d2' ), ToolMessage( content='历史悠久', name='get_history', id='f24543e7-8933-4b86-bd29-93f7e05b054e', tool_call_id='call_ad44233baf4644bd9ae887ae' ), AIMessage( content='我为您查询到了以下信息：\n\n## 🌤️ 今日北京天气\n- **天气状况**：晴天 ☀️\n- **气温**：20℃\n\n今天北京天气晴朗，气温 20 度，比较舒适宜人，适合外出活动，建议根据早晚温差适当增减衣物。\n\n## 📜 北京城市历史\n- **历史概况**：历史悠久\n\n关于北京的历史，工具返回的信息较为简略。不过补充一些背景供您参考：北京是中国著名的古都， 拥有三千多年的建城史和八百多年的建都史，曾是元、明、清等多个朝代的都城，拥有故宫、天坛、长城等众多世界级历史文化遗 产。\n\n如果您希望了解更详细的历史内容（如具体朝代、名胜古迹等），欢迎告诉我，我可以再帮您进一步查询。', additional_kwargs={'refusal': None}, response_metadata={ 'token_usage': { 'completion_tokens': 187, 'prompt_tokens': 463, 'total_tokens': 650, 'completion_tokens_details': { 'accepted_prediction_tokens': None, 'audio_tokens': None, 'reasoning_tokens': 18, 'rejected_prediction_tokens': None, 'text_tokens': 168 }, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0, 'text_tokens': 463} }, 'model_provider': 'openai', 'model_name': 'deepseek-v4.1-flash', 'system_fingerprint': None, 'id': 'chatcmpl-9d8f113a-5e42-904b-a4a3-e744bc5b8b5f', 'finish_reason': 'stop', 'logprobs': None }, id='lc_run--01a15860-3de2-7282-9adf-1cde6f7de6e4-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={ 'input_tokens': 463, 'output_tokens': 187, 'total_tokens': 650, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 18} } ) ] }

4. Agent的基本用法3：绑定工具

Agent支持静态和动态绑定工具，动态绑定需要用到中间件

工具可以是 LangChain 内置的，也可以是自定义的

官方工具文档： guan'fgTool integrations - Docs by LangChain

4.1 基本用法

Agent 支持绑定一个或多个工具

语法如下：

agent = create_agent( model = **** tool = [工具1, 工具2, 工具3, ...] ... )

案例：使用内置工具

from langchain.chat_models import init_chat_model from dotenv import load_dotenv import os from rich import print as rprint from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langchain_core.tools import tool from langchain.agents import create_agent from langchain_tavily import TavilySearch # 1.加载配置信息 load_dotenv(override=True) api_key = os.getenv("BAILIAN_API_KEY") base_url = os.getenv("BAILIAN_API_BASE") # 定义内置工具 web_search = TavilySearch( max_results=2, tavily_api_key=os.getenv("TAVILY_API_KEY"), ) # 2.初始化模型 llm = init_chat_model( model_provider="openai", # 模型提供商 model="deepseek-v4.1-flash", # 模型名称 api_key=api_key, # API Key base_url=base_url, # API Base URL ) # 3.创建智能体 agent = create_agent( model=llm, tools=[web_search], name="搜索引擎工具", ) # 4.调用智能体 response = agent.invoke({ "messages":[{"role": "user", "content": "帮我查一下2026年无畏契约上海冠军赛的冠军是谁？"}] }) rprint(response["messages"][-1].content)

输出结果如下：

一次完整的 Function calling 执行流程如下：

注意：：只给 Agent 需要的工具，工具太多会混淆。一般 2-5 个工具最佳

4.2 工具调用流程

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

本质上就是经典的ReAct 结构：一个具备“思考-行动-观察” 不断循环的自主工作者

用户输入 ——> 模型思考 ——> 调用工具 ——> 模型观察工具调用结果 ——> 模型思考 ——> ... ——> 结果输出

Agent 会在一个循环中反复调用模型和工具，直到某次模型输出中不再包含工具调用则结束，最后综合所有信息给出最终答案，执行流程图如下：

4.3 重试机制

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

案例：

from langchain.chat_models import init_chat_model from dotenv import load_dotenv import os from rich import print as rprint from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage from langchain_core.tools import tool from langchain.agents import create_agent number = 0 # 0.定义工具 @tool(parse_docstring=True) def get_current_weather(city: str) -> str: """ 获取指定城市的天气信息 Args: city (str): 城市名称 Returns: str: 天气信息 """ global number number += 1 if number < 3: return "工具调用失败,请重试!" else: return city + "晴天, 温度15度, 风力3级" # 1.加载配置信息 load_dotenv(override=True) api_key = os.getenv("BAILIAN_API_KEY") base_url = os.getenv("BAILIAN_API_BASE") # 2.初始化模型 llm = init_chat_model( model_provider="openai", # 模型提供商 model="deepseek-v4.1-flash", # 模型名称 api_key=api_key, # API Key base_url=base_url, # API Base URL ) # 3.创建智能体 agent = create_agent( model=llm, tools=[get_current_weather], name="天气助手1", ) # 4.调用智能体 response = agent.invoke({ "messages":[ {"role": "system", "content": "你是一个天气查询助手, 当天气工具的调用结果为'工具调用失败,请重试!'时，请不要放弃，至少重复调用3此"}, {"role": "user", "content": "杭州天气如何？"} ] }) for message in response["messages"]: if isinstance(message, ToolMessage): rprint(message.content)

输出结果：

4.4 常见问题

问题1：Agent 如何选择工具？

依据 ：工具的 docstring

AI 会根据：

1. 问题内容
2. 每个工具的描述
3. 自动选择最匹配的工具

问题2：Agent 为什么没有调用工具？

原因：

• 工具的 docstring 不清晰
• 问题表述不明确
• 模型认为不需要工具

问题3：Agent 选错工具？

原因：

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

解决方案：

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

问题4：如何知道 Agent 何时完成？

当 AIMessage 不包含 tool_calls 时：

问题5：Agent 可以调用多少次工具？

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

• 超时
• 达到 token 限制
• 模型决定停止

问题6：如何限制工具调用次数？

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

5. Agent的高级用法1：设置Agent名称

创建Agent时，LangChain允许用户指定其名称

这时输出的 AI Message带有 Name 信息

6. Agent的高级用法2：系统提示词

使用 create_agent 创建 Agent 时，需传入模型和工具、可选地传入系统提示词

系统提示词为 Agent 提供了任务背景、行为准则和操作指南

系统指令：即 SystemMessage，通过 system_prompt 设置，定义 Agent 行为。这个参数可以是str 或者 SystemMessage 类型

使用建议：

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

用法如下：

提示词设置有两种方式：静态设置和动态设置

• 静态设置：Agent 实例创建时通过 system_prompt 参数一次性写死，之后所有 invoke / stream 调用，自动带上这段系统指令；每次调用都相同，运行时不可变
• 动态设置：每一轮模型调用前实时生成，根据本次请求的上下文、用户身份、会话状态、数据库记忆，动态拼接不同的 system 指令，方法是：@dynamic_prompt 中间件 + runtime.context

7. Agent的高级用法3：结构化输出

agent 结构化输出：允许 Agent 以特定、可预测的格式返回数据，而不是传统的自然语言响应

通过结构化输出，开发者可以直接获得Pydantic模型、JSON对象或数据类等结构化数据，这些数据能够被应用程序直接使用，无需复杂的解析过程

7.1 模型 vs Agent 的结构化输出对比

7.2 结构化输出的4种策略

LangChain 的 create_agent( ) 函数自动处理结构化输出的全过程；用户只需通过“response_format” 参数设置期望的输出模式（Schema）；当模型生成结构化数据时，系统会自动捕获、验证并将结果存储在 Agent 状态的structured_response键中

用法如下：

create_agent（）函数中的 response_format 参数支持四种不同的策略设置方式

① ProviderStrategy

使用模型提供商的原生结构化输出功能实现结构化输出

• 原生结构化输出：指的是大语言模型（LLM）提供商通过其 API 直接提供的、在模型响应阶段就强制保证输出格式符合预定规范的能力，这种能力能够在模型生成内容的源头确保结构化准确性
• 适用于支持原生结构化输出的模型，比如 OpenAI、Anthropic Claude 或 xAI Gro k等

案例如下：

② ToolStrategy

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

原理：动态创建一个" 虚拟工具 "，该工具的输入参数对应着期望的数据结构；当模型需要生成最终答案时，系统会引导模型 "调用" 这个虚拟工具 ，从而间接产生符合要求的结构化数据

案例：

③ type / AutoStrategy

官方没有在参数列表或官方文档列出这种策略，但阅读源码可以看到：

当我们直接传入一个定义类型时，LangChain 会自动包装为 AutoStrategy，触发自动选择策略：如果模型支持原生结构化输出（如OpenAI、Anthropic Claude或xAI Grok），则优先使用 ProviderStrategy；否则使用 ToolStrategy

④ None

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

总结：

在实际大模型 Agent 开发场景中，如果使用到了结构化输出，推荐使用 “ToolStrategy” 策略

7.3 ToolStrategy使用详解

ToolStrategy 通过工具调用 （Tool Calling/Function Calling）实现结构化输出，所以 LangChain会在消息列表末尾追加一条 ToolMessage，但实际上没有实际的工具执行，这是一条伪消息

ToolStrategy 适用于任何支持工具调用的现代模型，ToolStrategy 的配置包含三个主要参数：

• schema：必需参数，与提供商策略的 schema 参数功能一致，支持Pydantic 模型 、 TypedDict 、 JSON Schema 、 数据类(@dataclass)，同时还支持联合类型 Union [类型1, 类型 2] （允许模型根据输入内容选择最匹配的数据结构）
• tool_message_content：可选参数，用于自定义生成结构化输出时，会话历史中记录的提示信息。默认使用展示输出数据的标准响应语句
• handle_errors：可选参数，用于指定数据校验失败时的重试策略， 默认值为True

7.3.1 结构化输出：schema 参数

输出模式1：Pydantic类型

Pydantic类型的 Schema 支持数据验证，是优先推荐使用的方式

输出如下：

观察日志可知，这种方式将结构化信息作为伪工具传递，显然使用了Function Calling方法

输出模式2：TypedDict 类型

TypedDict 允许为字典对象定义固定的键名和对应的值类型 ，是带有类型提示的字典结构

1. TypedDict 字段定义采用Annotated [类型, 默认值, "描述"]格式
2. 可选字段使用 Optional 包装，默认值在 Annotated 中指定
3. TypedDict不支持运行时验证

案例如下：

输出模式3：JsonSchema 类型

JSON Schema 是提供一个标准的 JSON Schema 字典来定义结构，适合需要与多种编程语言交互或进行复杂数据约束定义的场景

案例如下：

以上代码中定义 json_schema 的时候指定的 title, description, type, properties, required 是遵循 JSON Schema 规范的标准关键字，是固定写法

输出模式4：@dataclass类型

@dataclass是Python 3.7引入的一个装饰器，用于简化数据存储类的定义

多 schema 联合模式

ToolStrategy 允许指定多个类型 “Union[类型1, 类型2]” 这种写法，LLM能够根据输入文本的内容，智能地选择最合适的一个 数据模型（Schema）来生成结构化输出，最终会只有一种类型输出

输出结果如下：

继续：

输出结果如下：

7.3.2 自定义工具消息：tool_message_content 参数

我们已经知道，采用 ToolStrategy 策略处理结构化输出时，LangChain 会在消息列表末尾追加一条 Tool_message，让整个链路完整

tool_message_content：定制 Tool_message 消息内容，将指定的内容写入对话历史的提示信息，这样做的好处如下:

1. 在最终用户可见的对话流中，使用 更自然的消息替代原始数据
2. 用简短的确认信息替代可能很长的数据块， 减少token消耗

注意：

• 当不设置 tool_message_content 时：模型收到的 ToolMessage 里就包含了像 {'name': '张三', 'email': 'zhangsan@email.com'... ...} 这样的具体数据
• 当设置了 tool_message_content 时：模型收到的 ToolMessage 只是一个预定义的确认信息，如“ 格式化输出成功！ ”这种方式节省了上下文窗口的令牌消耗，并且让对话流对最终用户更友好

说明：无论 tool_message_content 如何设置，成功提取的结构化数据最终都会正确存入 result["structured_response"]返回，自定义消息仅影响对话历史中的一条记录

7.3.3 错误处理：handle_errors 参数

ToolStrategy 通过其 handle_errors 参数提供了结构化过程错误处理策略，以下是主要的几种方式及其用途：

• handle_errors=True：默认方式 ， 捕获所有异常 ，并使用 LangChain 内置的错误消息模板提示模型重试，确保最终能得到符合预定格式的有效数据
• handle_errors=False：关闭自动重试机制，任何异常都会直接抛出 ，会中断程序运行
• handle_errors="自定义字符串"：捕获所有异常，但使用开发者预设的固定字符串作为错误消 息
• handle_errors=ExceptionType：仅捕获指定类型(如ValueError) 或元组中的异常类型并进行重 试， 其他异常直接抛出
• handle_errors=callable：灵活性最高的方式，使用开发者自定义的函数来处理异常 ，可根据不 同的异常类型返回差异化的提示信息

情况1：设置为True/False/固定字符串

设计思路：模型对于单条信息的格式化输出请求，输出了多个工具调用请求。也称为多结构化输出错误

1）handle_errors设置为True

运行后结果如下：

模型接收到错误反馈后，再次生成新的调用请求，直至成功或达到内部的最大重试次数

注意：ToolStrategy 中自定义的 tool_message_content控制的是成功后的消息内容，不影响错误消息内容

2）handle_errors 设置为 False

运行后会直接报错

3）handle_errors设置为“请检查输入数据”

以上代码中注意如下几点：

1. ToolStrategy 允许指定多个类型“Union[ContactInfo, EventDetails]” 这种写法，但是最终只会转换成一种结构化类型输出

2. 当 ToolStrategy 通过“Union[ContactInfo, EventDetails]”指定多个类型时，在内部调用生成结构化类型工具会报错；此时：

• handle_errors=True（默认值）开始发挥作用，系统会生成一个ToolMessage，明确告诉 LLM“Error: Model incorrectly returned multiple structured responses (ContactInfo, EventDetails) when only one is expected.”，大模型收到这个精准的反馈后，会重新进行推理，最终选择并输出一个最符合要求的Schema
• 如果handle_errors 设置为False，执行代码过程直接报错

3. 当格式化输出有错误时，Agent内部会进行工具调用重试，直到符合要求格式化输出前， 可能会进行多次重试

情况2：设置为指定异常类型

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

1. MultipleStructuredOutputsError

多结构化输出错误，当返回的工具调用请求数量大于1时，抛出该异常，默认情况下 LangChain 会拦截异常并提醒模型重试

2. StructuredOutputValidationError

输出结构化验证错误，当输出格式不符合结构化要求时，抛出上述异常。 默认情况下 LangChain 会拦截该异常并自动重试

案例如下：

输出结果如下：

情况3：设置为自定义错误处理函数

指定异常处理函数并返回字符串时，LangChain 会在遇到异常时自动重试并将异常处理函数的返回值作为ToolMessage的内容；我们也可以选择在异常处理函数中抛出异常

案例如下：

运行后结果如下：

8. Agent的高级用法4：流式输出及模式

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

8.1 具体的输出模式

8.1.1 values 输出模式

stream_mode = "values"：每个步骤执行后，都会输出完整的状态信息，适用于每一步都要获取完整状态、状态持久化场景

案例：

from pydantic import BaseModel, Field from langchain.agents import create_agent from langchain.agents.structured_output import ProviderStrategy from langchain.messages import HumanMessage, SystemMessage from langchain_core.tools import tool from rich import print as rprint from langchain.chat_models import init_chat_model from dotenv import load_dotenv import os # 0.定义工具 @tool(parse_docstring=True) def get_weather(city: str) -> str: """ 获取指定城市的天气信息 Args: city (str): 城市名称 Returns: str: 天气信息 """ return f"{city}的天气是晴天，温度是25度" @tool(parse_docstring=True) def get_location(city: str) -> str: """ 获取指定城市的地理位置信息 Args: city (str): 城市名称 Returns: str: 地理位置信息 """ return f"{city}的经纬度是30.123, 120.123" # 1.加载环境变量 load_dotenv(override=True) api_key = os.getenv("BAILIAN_API_KEY") base_url = os.getenv("BAILIAN_API_BASE") # 2.初始化模型 llm = init_chat_model( model_provider="openai", model = "deepseek-v4.1-flash", api_key = api_key, base_url = base_url, ) # 3.agent 初始化 agent = create_agent( model = llm, tools = [get_weather, get_location], ) # 4.构建消息列表 messages = { "messages":[ SystemMessage(content="你是一个AI助手，你需要根据用户的输入，进行智能处理，并返回结果。"), HumanMessage(content="杭州天气如何？杭州在哪里？"), ] } # 5.调用agent response = agent.stream( messages, stream_mode="values" ) for chunk in response: rprint(chunk) print( "-" * 50)

运行后结果示意如下：

8.2.2 updates 输出模式

stream_mode = "updates"：默认模式；每个步骤执行后，只增量更新状态中发生变化的内容，用于监控 Agent 执行进度

用法如下：

输出结果示例如下：

8.2.3 messages 输出模式

stream_mode = "messages"：输出流式返回的 Token 以及相关的元数据，可以用在实现类似 ChatGPT 的打字机效果场景，为聊天机器人等交互式应用提供最佳的实时体验

用法如下：

输出结果如下：

8.2.4 tasks 输出模式

stream_mode = "tasks"：输出当前task任务开始和结束的时间，包含任务的结果和错误信息，该模式用于监控任务的生命周期

8.2.5 debug 输出模式

stream_mode = "debug"：与tasks模式类似，比 task 模式多输出任务步骤、时间戳、task 类型（task/task_result），该模式用于调试、监控 task 任务的生命周期

8.2.6 checkpoints 输出模式

stream_mode = "checkpoints"：每当检查点（checkpoint）被创建时会触发输出，输出包含检查点中的状态，用于需要状态持久化、工作流恢复或分布式执行跟踪的高级场景

用法：

from langgraph.checkpoint.memory import InMemorySaver #--其他代码同上-- # 1. 创建内存检查点存储 checkpointer = InMemorySaver() # 2. 创建Agent agent = create_agent( model=llm, tools=[get_weather, get_location], checkpointer=checkpointer # 启用检查点 ) # 3. 创建唯一的会话ID config = {"configurable": {"thread_id": "session01"}} # 4.构建消息列表 messages = { "messages":[ SystemMessage(content="你是一个AI助手，你需要根据用户的输入，进行智能处理，并返回结果。"), HumanMessage(content="杭州天气如何？杭州在哪里？"), ] } # 5.调用agent response = agent.stream( messages, config=config, stream_mode="checkpoints" ) checkpoint_count = 0 for chunk in response: checkpoint_count += 1 print(f"检查点 #{checkpoint_count}") print(chunk) print("-" * 50)

输出结果如下：每次输出都会将相关 MESSAGE 追加到 values.messages 中

8.2.7 custom 输出模式

stream_mode = "custom"：开发者通过get_stream_writer在工具或节点内部自定义发送的数据 ，用于输出业务逻辑相关的进度信息（如“已处理10/100条记录”）、自定义日志或指标

案例：

输出结果如下：

8.2 流式输出总结

我们可以根据不同的目标来选择不同的输出模式：

• 实现实时对话交互，优先选择 messages 模式
• 观察Agent的思考与执行步骤 ，优先选择 updates 模式
• 需要查看每一步状态，优先选择 values/tasks/debug 模式
• 在工具执行时输出自定义业务日志，优先选择 custom 模式

此外，以上这些模式还可以组合使用，例如，可以同时指定 stream_mode=[“tasks”，“updates”] ，这 样在同一个循环里既能查看Agent task任务执行内容， 又能显示Agent每步的更新