目录1. 指定模型1.1 原理ModelRequestoverrideModelResponse2. 指定工具2.1 原理3.指定提示词3.1 原理4. 结构化输出4.1 原理5. 定义状态5.1 原理6. 人机协作6.1 HumanInTheLoopMiddleware6.2 原理Interrupt6.3 调用版本v1 vs v27. 流式传输7.1 原理同时输出推理过程、工具调用增量和自定义进度流式传输与人机协作流式传输子代理1. 指定模型模型是 Agent 的推理引擎。LangChain 支持两种模式静态模型Agent 生命周期内模型固定适合大多数简单场景动态模型运行时根据消息长度、任务复杂度、用户等级等上下文动态切换模型例如简单问答用 gpt-4o-mini复杂推理用 gpt-4o兼顾成本与效果1.1 原理静态模型在 create_agent 时传入Agent 内部直接持有该模型实例动态模型通过中间件的 wrap_model_call 钩子在每次模型调用前获取 ModelRequest读取状态或运行时上下文调用 request.override(model...) 替换模型然后继续执行 handler根据对话轮数切换模型from langchain.agents import create_agent from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse from langchain.tools import tool from langchain_openai import ChatOpenAI tool def get_weather(city: str) - str: return f{city} 阳光明媚 model_mini ChatOpenAI(modelgpt-4o-mini) model_pro ChatOpenAI(modelgpt-4o) wrap_model_call def dynamic_model(request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]): if len(request.state.get(messages, [])) 3: request request.override(modelmodel_pro) return handler(request) agent create_agent( modelmodel_mini, tools[get_weather], middleware[dynamic_model] )Callable[[ModelRequest], ModelResponse]组成部分解释Callable这是 Pythontyping模块中的泛型类型表示“一个可被调用的对象”[ModelRequest]这是参数列表。方括号内只有一个类型意味着这个可调用对象只接收 1 个位置参数且该参数必须是ModelRequest类型ModelResponse这是返回值类型。意味着这个可调用对象执行完毕后必须返回一个ModelResponse类型的对象ModelRequestModelRequest 是 LangChain Agent 中间件Middleware体系中的核心数据模型它封装了一次模型调用LLM Call 所需的所有信息属性名类型必需描述modelBaseChatModel是当前步骤将要使用的聊天模型实例messageslist[AnyMessage]是发送给模型的消息列表不包含系统提示词system_promptstr否系统提示词。已弃用建议使用system_messagesystem_messageSystemMessage否系统消息对象是设置系统提示词的首选方式toolslist[BaseTool 或 dict]否当前步骤可供模型调用的工具列表tool_choiceAny否工具调用策略例如auto、none或强制调用某个特定工具response_formatResponseFormat否结构化输出格式的配置让模型以指定格式返回结果model_settingsdict[str, Any]否额外的模型调用参数如temperature、headers等stateAgentState否Agent 的当前状态包含了所有对话历史等上下文信息runtimeRuntime[ContextT]否运行时上下文包含store长期记忆等对象overrideModelRequest 最重要的方法是 override()它不会修改原有的 ModelRequest 对象而是创建一个全新的、带有指定属性修改的副本# 在中间件中修改模型和系统提示词 new_request request.override( modeldifferent_model, # 替换为另一个模型 system_messageSystemMessage(content新的系统提示词), # 修改系统提示 tool_choicenone, # 禁用工具调用 ) # 然后将 new_request 传递给 handlerModelResponsedataclass class ModelResponse: result: list[BaseMessage] # 模型返回的消息列表 structured_response: ResponseT | None None # 可选的结构化输出ModelResponse 是 LangChain Agent 中间件体系中模型调用环节LLM Call的输出结果对象。它封装了 LLM 返回的原始响应消息及其元数据ModelResponse 本身是一个轻量级的容器它的核心属性非常精简属性名是否必需描述message是LLM 返回的完整响应消息对象所有实际数据都封装在这里面structured_response否在 create_agent 中配置了 response_format 参数后这里会存放解析好的结构化数据对象2. 指定工具工具赋予 Agent 执行具体操作的能力。除了静态工具列表LangChain 支持两种动态扩展方式运行时过滤根据用户认证状态、功能开关等从预注册工具中筛选可见工具运行时注入动态添加全新工具如临时计算器无需重启 Agent同时Agent 天然支持顺序/并行调用、错误重试和状态持久化2.1 原理静态工具在 create_agent 时注册到 Agent 的图结构中动态过滤在中间件的 wrap_model_call 中修改 request.tools模型只能看到过滤后的工具动态注入在中间件的 wrap_tool_call 中当检测到某个工具名时用新工具实例覆盖原工具错误处理wrap_tool_call 包裹工具执行异常时返回友好的 ToolMessage让模型有机会重试或改述运行时根据认证状态过滤工具from langchain.agents import create_agent, AgentState from langchain.agents.middleware import wrap_model_call, ModelRequest from langchain.tools import tool tool def public_search(q: str) - str: return 公开结果 tool def private_search(q: str) - str: return 私密结果 class State(AgentState): authenticated: bool wrap_model_call(state_schemaState) def filter_tools(request: ModelRequest, handler): if not request.state.get(authenticated, False): request request.override(tools[t for t in request.tools if t.name.startswith(public_)]) return handler(request) agent create_agent( modelgpt-4o-mini, tools[public_search, private_search], middleware[filter_tools] ) # 调用时传入 authenticatedFalse 则只能使用 public_search3.指定提示词系统提示词System Prompt定义了 Agent 的角色和回答风格。LangChain 支持静态提示固定文本适用于通用场景动态提示根据用户角色、对话阶段、上下文动态生成实现个性化接口/参数类型说明dynamic_prompt装饰器函数接收ModelRequest返回提示字符串3.1 原理dynamic_prompt 中间件在每次模型调用前执行返回的字符串会作为系统消息插入到消息列表最前面或覆盖默认的 system prompt根据用户角色切换提示from langchain.agents import create_agent from langchain.agents.middleware import dynamic_prompt, ModelRequest from typing import TypedDict class Context(TypedDict): user_role: str dynamic_prompt def role_prompt(request: ModelRequest) - str: role request.runtime.context.get(user_role, 初学者) base 你是乐于助人的助手。 if role 专家: return base 提供深入技术解答。 return base 用简单语言解释。 agent create_agent( modelgpt-4o-mini, middleware[role_prompt], context_schemaContext ) agent.invoke({messages: [...]}, context{user_role: 初学者})只要使用了 dynamic_prompt原始的静态系统提示词就会被“接管”并替换这种覆盖仅对当前的这一次模型调用生效修改不会持久化到 Agent 的配置中下一次模型调用时如果没有触发动态提示词系统会恢复使用默认的静态提示词4. 结构化输出让 Agent 的输出符合 Pydantic 模型的结构化数据如 JSON。LangChain 提供两种策略ToolStrategy利用工具调用将输出伪装成工具参数兼容所有支持工具调用的模型ProviderStrategy使用模型原生结构化输出更可靠高效接口/参数类型说明示例response_formatBaseModel/ToolStrategy/ProviderStrategy直接传模型自动选择或显式指定策略response_formatContactInforesult[structured_response]BaseModel输出解析后的结构化对象print(result[structured_response].name)4.1 原理自动选择若传 Pydantic 模型LangChain 先检测模型是否支持原生结构化输出是则用 ProviderStrategy否则回退 ToolStrategyToolStrategy创建一个虚拟工具其 args_schema 为目标 Pydantic 模型强制模型以工具调用形式输出参数ProviderStrategy直接使用 model.with_structured_output() 等方法提取联系人信息from pydantic import BaseModel from langchain.agents import create_agent class ContactInfo(BaseModel): name: str email: str phone: str agent create_agent( modelgpt-4o-mini, response_formatContactInfo # 自动策略 ) result agent.invoke({ messages: [{role: user, content: 张三zhangsanexample.com13800138000}] }) print(result[structured_response]) # ContactInfo(...)5. 定义状态Agent 默认只维护消息列表对话历史。但复杂场景需要额外状态如用户偏好、临时标志、中间计算结果。LangChain 通过 自定义状态继承 AgentState 的 TypedDict实现短期记忆5.1 原理状态在 Agent 执行期间持续存在同一线程 ID 下中间件可通过 request.state 读取或修改通过 before_model 等钩子工具可通过 ToolCallRequest.state 访问状态中间件绑定状态from langchain.agents import AgentState, create_agent from langchain.agents.middleware import AgentMiddleware class CustomState(AgentState): user_preferences: dict class PreferencesMiddleware(AgentMiddleware): state_schema CustomState def before_model(self, state: CustomState, runtime): pref state.get(user_preferences, {}) # 可在此修改消息或做日志 print(f风格: {pref.get(style)}) agent create_agent( modelgpt-4o-mini, middleware[PreferencesMiddleware()] ) agent.invoke({ messages: [{role: user, content: 解释AI}], user_preferences: {style: 简明} })全局状态agent create_agent( modelopenai:gpt-4o-mini, state_schemaCustomState # 快捷定义 )6. 人机协作在敏感操作如写文件、执行 SQL前暂停 Agent等待人工审批、编辑或拒绝通过 HumanInTheLoopMiddleware 配合 LangGraph 的检查点机制实现可中断/恢复的持久化执行核心依赖基于 LangGraph 的持久化层Checkpointer实现执行状态的保存与恢复6.1 HumanInTheLoopMiddleware参数类型必填说明interrupt_ondict[str, bool | InterruptOnConfig]是核心配置定义哪些工具需要中断以及允许的决策类型description_prefixstr否中断消息的全局前缀用于向审核者说明情况人工决策类型说明示例用例approve完全批准工具按原参数执行发送邮件草稿、提交订单edit允许修改工具参数后再执行修改邮件收件人、调整 SQL 条件reject拒绝执行将反馈添加到对话中拒绝 SQL 删除操作并说明原因interrupt_on 的键是工具名称值有三种配置方式值类型说明True启用中断并允许所有三种决策类型approve、edit、rejectFalse禁用中断工具调用将被自动批准直接执行InterruptOnConfig精细控制中断行为可自定义允许的决策列表、审核描述动态控制是否触发中断InterruptOnConfig 对象属性属性类型必填说明allowed_decisionslist[DecisionType]是明确允许的决策类型列表。例如[approve, reject]表示只允许批准或拒绝descriptionstr | Callable否覆盖全局的description_prefix为当前工具提供自定义的中断描述interrupt_on{ write_file: True, # 全决策可用 execute_sql: {allowed_decisions: [approve, reject]}, # 禁止编辑 read_data: False, # 自动通过 }6.2 原理中断触发中间件在工具调用前检查 interrupt_on 配置若需中断则调用 LangGraph 的 interrupt() 函数保存当前状态快照并暂停执行等待审批调用方从 result.interrupts 获取待审核的 action_requests工具名、参数、描述和 review_configs允许的决策类型( Interrupt( value{ action_requests: [ { name: execute_sql_tool, args: {sql: DELETE FROM data WHERE id 1;}, description: 工具执行尚待批准\n\nTool: execute_sql_tool\nArgs: {sql: DELETE FROM data WHERE id 1;} } ], review_configs: [ { action_name: execute_sql_tool, allowed_decisions: [approve, reject] } ] }, ide678d1947c5028c88b98e961062d7b4d ), )Interrupt一个 Interrupt 对象主要包含以下两个核心属性属性名类型描述valueAny(JSON-serializable)核心数据。包含了中断的具体信息其结构由调用interrupt()时传入的值决定。在使用HumanInTheLoopMiddleware时它是一个包含action_requests和review_configs的字典idstr中断的唯一标识符。用于在恢复执行时通过Command精确地定位和响应这个特定的中断恢复执行调用方通过 Command(resume{decisions: [...]}) 传入决策列表Agent 从检查点恢复状态并继续执行决策顺序约束decisions 列表的顺序必须与 action_requests 的顺序严格一致from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command # 1. 创建 Agent配置 HITL 中间件和检查点 agent create_agent( modelgpt-4.1, tools[execute_sql, write_file, read_data], middleware[ HumanInTheLoopMiddleware( interrupt_on{ write_file: True, # 全部三种决策 execute_sql: {allowed_decisions: [approve, reject]}, # 仅批准/拒绝 read_data: False, # 自动放行 }, description_prefix工具执行尚待批准, ) ], checkpointerInMemorySaver(), # 生产环境使用 PostgresSaver ) config {configurable: {thread_id: 123}} # 2. 首次调用可能触发中断 result agent.invoke( {messages: [{role: user, content: 删除 users 表中的过期记录}]}, configconfig, versionv2, # 必须指定 v2 以获取中断信息 ) # 3. 检查是否被中断 if result.interrupts: print(result.interrupts) # 查看待审核的动作和配置 # 4. 提交决策恢复执行三种决策示例 # 批准 agent.invoke( Command(resume{decisions: [{type: approve}]}), configconfig, versionv2, ) # 拒绝附带说明 agent.invoke( Command(resume{ decisions: [{ type: reject, message: 生产环境不允许删除操作请改用软删除。 }] }), configconfig, versionv2, ) # 编辑修改参数后执行 agent.invoke( Command(resume{ decisions: [{ type: edit, edited_action: { name: execute_sql, args: {sql: UPDATE users SET statusinactive WHERE id 100} } }] }), configconfig, versionv2, )6.3 调用版本v1 vs v2特性versionv1(默认)versionv2(推荐/必须)invoke返回值类型普通字典 (dict)结构化对象 (GraphOutput)获取中断信息从字典的__interrupt__键读取通过返回对象的.interrupts属性直接获取类型安全无完全类型安全Type-safeHITL 支持不推荐获取中断信息繁琐必须是获取和处理中断信息的官方方式GraphOutput属性类型说明valueOutputT图执行完成的最终状态或输出。根据你的状态模式State Schema它可能是一个字典、Pydantic 模型或数据类dataclassinterruptstuple[Interrupt, ...]一个元组包含执行过程中产生的所有中断Interrupt对象。如果未发生中断则为空元组将 versionv2需要 LangGraph 1.1.传递给 stream() 或 astream() 以获取统一的输出格式。每个数据块都是一个具有 type、ns 和 data 键的 StreamPart 字典StreamPart 是一个可区分的联合类型可以通过其 type 字段来判断当前数据块属于哪种类型并访问对应的 data 字段其统一的字典结构包含以下核心字段type: 字符串标识数据块类型是进行逻辑判断的关键。ns: 元组标识数据来源的节点命名空间根图为空。data: 实际的数据负载其类型取决于 type 的值。interrupts: 仅当 typevalues 时存在一个元组包含本步骤产生的所有中断信息for chunk in agent.stream(..., stream_mode[values], versionv2): # 直接获取当前状态的快照 state chunk[data] # 直接检查 state 里有没有 interrupts if state.get(interrupts): interrupts state[interrupts] # 这就是 list[Interrupt] # 处理中断...type值data字段的类型与内容主要用途values当前步骤之后的完整状态对象如果状态模式是 Pydantic 模型会自动转换获取执行到当前节点的完整状态快照用于检查中断updates一个字典键为节点名值为该节点产生的状态更新关注具体是哪个节点产生了什么变化进行细粒度监控messages一个元组包含(消息块, 元数据)处理流式输出的消息块custom用户通过get_stream_writer()发送的自定义数据在流中传输自定义的事件或数据tasks关于正在执行或已完成的任务的信息监控后台任务的执行状态debug调试信息进行问题排查和性能分析7. 流式传输7.1 原理提供四种流模式满足不同实时反馈需求LangGraph 底层支持多种流模式Agent 封装后统一接口messages实时展示 LLM 逐字输出、工具调用参数生成过程updates实时展示当前执行到哪一步如“思考中”、“调用工具中”且只关心增量values需要获取截止当前的全部对话历史做快照保存或上下文分析custom 允许开发者在任意节点通过 writer 发送任意 JSON 可序列化数据同时输出推理过程、工具调用增量和自定义进度from langchain.agents import create_agent from langchain.messages import AIMessageChunk, ToolMessage from langgraph.config import get_stream_writer from langchain_openai import ChatOpenAI def get_weather(city: str) - str: writer get_stream_writer() writer(f正在查询 {city} 天气...) return f{city} 晴朗 model ChatOpenAI( modelgpt-5-mini, reasoning{ # 关键配置 effort: medium, # 推理程度low, medium, or high summary: auto, # 推理摘要detailed, auto, or None } ) agent create_agent(modelmodel, tools[get_weather]) for chunk in agent.stream( {messages: [{role: user, content: 上海天气}]}, stream_mode[messages, updates, custom], versionv2 ): if chunk[type] custom: print([进度], chunk[data]) elif chunk[type] messages: token, meta chunk[data] if isinstance(token, AIMessageChunk): # 推理块 for b in token.content_blocks: if b.get(type) reasoning: print([推理], b.get(reasoning, ), end) # 工具调用增量 if token.tool_call_chunks: print([工具参数], token.tool_call_chunks) # 文本增量 if token.text: print([文本], token.text, end) elif chunk[type] updates: for node, update in chunk[data].items(): if node tools: last update[messages][-1] if isinstance(last, ToolMessage): print([工具结果], last.content)必须确保模型支持并开启了推理输出如 reasoning 参数。注意模型不同参数不同无论使用哪个提供商都可通过 content_blocks 中的 type: reasoning统一访问推理内容通常配合 stream_modemessages 使用逐块输出推理文本实时显示工具调用的参数JSON片段stream_modemessages提供 AIMessageChunk其中包含 tool_call_chunks增量参数最终获取完整解析后的工具调用消息stream_modeupdates在步骤如 model 节点完成后提供完整的 AIMessage其中包含已解析的 tool_calls流式传输与人机协作使用 HumanInTheLoopMiddleware 中间件指定需要中断的工具流式传输时捕获 updates 模式中的 __interrupt__ 节点构造 Command(resume...) 来恢复执行from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware from langchain_core.messages import AIMessageChunk, AnyMessage, AIMessage, ToolMessage from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command, Interrupt def get_weather(city: str) - str: 获取天气 return f{city} 天气晴朗 agent create_agent( gpt-5-mini, tools[get_weather], # 允许全部三种决策approve、edit、reject middleware[HumanInTheLoopMiddleware(interrupt_on{get_weather: True})], checkpointerInMemorySaver(), ) def render_interrupt(interrupt: Interrupt) - None: interrupts interrupt.value for request in interrupts[action_requests]: print(request[description]) def render_chunk(token: AIMessageChunk): if token.text: print(token.text, end|) if token.tool_call_chunks: print(token.tool_call_chunks) # 增量块 def render_completed(msg: AnyMessage): if isinstance(msg, AIMessage) and msg.tool_calls: print(f完整工具调用: {msg.tool_calls}) if isinstance(msg, ToolMessage): print(f工具响应: {msg.content_blocks}) config {configurable: {thread_id: 1}} interrupts [] # 第一次流式遇到中断 for chunk in agent.stream( {messages: [{role: user, content: 查询北京和上海的天气}]}, configconfig, stream_mode[updates], versionv2, ): if chunk[type] updates: for source, update in chunk[data].items(): if source __interrupt__: interrupts.extend(update) # 解析中断展示给用户收集决策... render_interrupt(update[0]) 输出 Tool execution requires approval # 获取北京天气被拦截 Tool: get_weather Args: {city: 北京} Tool execution requires approval # 获取上海天气被拦截 Tool: get_weather Args: {city: 上海} # 假设用户决定批准第一个调用编辑第二个调用城市改为西安 # 必须分别用各自的 ID 作为键 decisions { interrupts[0].id: {decisions: [{type: approve}]}, # 北京用北京ID interrupts[1].id: {decisions: [{type: edit, edited_action: {name: get_weather, args: {city: 西安}}}]} # 上海用上海ID } # 第二次流式恢复执行 for chunk in agent.stream( Command(resumedecisions), configconfig, stream_mode[messages, updates], versionv2, ): if chunk[type] messages: token, metadata chunk[data] if isinstance(token, AIMessageChunk): render_chunk(token) elif chunk[type] updates: for source, update in chunk[data].items(): if source in (model, tools): render_completed(update[messages][-1]) if source __interrupt__: interrupts.extend(update) render_interrupt(update[0])流式传输子代理当一个代理调用另一个代理时流式输出需要能区分消息的来源以便正确渲染实现原理在创建子代理时使用 name 参数为其命名在主代理流式调用时设置 subgraphsTrue 确保子代理的内部流式事件也能被捕获在 messages 模式的元数据中通过 lc_agent_name 键获取当前产生消息的代理名称可在渲染前切换显示标签使前端能明确区分不同代理的输出from langchain.agents import create_agent from langchain.chat_models import init_chat_model # 1. 创建子代理 weather_agent create_agent( modelopenai:gpt-5.2, tools[get_weather], # get_weather 定义同前 nameweather_agent, ) # 2. 创建主代理包含调用子代理的工具 def call_weather_agent(query: str) - str: result weather_agent.invoke({messages: [{role: user, content: query}]}) return result[messages][-1].text supervisor create_agent( modelopenai:gpt-5.2, tools[call_weather_agent], namesupervisor, ) # 3. 流式传输并识别当前活跃的代理 current_agent None for chunk in supervisor.stream( {messages: [{role: user, content: 波士顿天气如何}]}, stream_mode[messages], subgraphsTrue, # 重要允许子图流式输出 versionv2, ): if chunk[type] messages: token, meta chunk[data] agent_name meta.get(lc_agent_name) if agent_name and agent_name ! current_agent: print(f\n {agent_name} 正在输出) current_agent agent_name # 渲染token if isinstance(token, AIMessageChunk) and token.text: print(token.text, end)