有条件停止工具:在LangGraph中,伪影效果优于return_direct
当静态的return_direct无法做出决策时,可通过针对每次调用的工具组件来停止/继续执行——手动构建ReAct及create_agent中间件。
何时静态 return_direct 不是合适的选择
凡是发布过 LangGraph 工具调用代理的人都会遇到 return_direct=True 的情况:跳过通过模型返回工具结果的操作并结束循环。在需要根据本次调用的结果而非注册了哪种工具来决定停止条件时,这种做法看似完美。
本教程会遇到这一限制,首先手动构建一个最简的 ReAct 循环,然后再利用中间件在 create_agent 中实现相同的功能。简短的答案是:确实可行——但由于消息顺序的细微问题,最初的中间件方案可能会失败,而非因为框架隐式地丢弃了更新信息。
为便于理解,特此说明版本标识:“legacy”指langgraph==0.6.6(即在create_react_agent被替换为create_agent之前的最后版本);“current”指使用langchain==1.4.2并搭配langgraph==1.2.11的版本。每个ReAct智能体都由模型与工具构成循环结构,此处重点关注的是从工具返回到模型的路径——以及在特定调用场景下该路径应何时消失。
打破默认循环的需求
集成搜索工具看似很平常:模型调用search(query),读取结果后给出答案或继续处理。但有两个特性使得原有的循环结构并不适用。
当搜索成功时,工具会返回一个庞大的JSON数据页。若要将这些数据重新放入上下文中供模型再次处理,不仅成本高昂且往往毫无意义——既然搜索已经给出了答案,再次调用几乎只是以极高代价重新表述已有内容而已。
当操作失败时,错误会分为两种对立情况:一种是真正的死路(没有任何匹配项——重新尝试纯属浪费),另一种则是暂时的超时或503错误(此时重新尝试是合理的)。因此规则是根据每次调用来确定的:成功则停止,可重试的错误则继续,致命错误则停止——这一判断依据是请求数据中的信息,而非工具的静态类型。
为何 return_direct 无法做到这一点
在固定版本的 langgraph.prebuilt.chat_agent_executor 中,路由逻辑如下:
should_return_direct = {t.name for t in tool_classes if t.return_direct}
...
def route_tool_responses(state):
for m in reversed(_get_state_value(state, "messages")):
if not isinstance(m, ToolMessage):
break
if m.name in should_return_direct:
return END
...
return entrypoint
should_return_direct 是在构建图结构时根据工具的 .return_direct 属性一次性计算得出的。该标志表示“此工具始终会结束循环”,它没有针对每次调用的不同模式。这是类别不匹配的问题,而非该标志本身的缺陷。
论坛中的讨论都反映了同样的问题:庞大的工具结果会导致不必要的后续模型调用,维护者通常建议手动将 tool_node 连接到 END。关于 Command 更新以及 return_direct(包括 langgraph#5496)的单独讨论为这一话题提供了背景信息;核心观点是无需视为缺陷——静态标志根本无法承载动态结果。
控制流结构
简单来说:
Tool call
├── success or unfixable failure → stop, use the tool's result
└── fixable failure → let the model decide
有两个目标节点,每次调用都会重新选择。本文的其余部分两次实现了这种结构——一次通过手动方式,一次通过中间件。
分离的通道:content 和 artifact
LangChain 的 @tool 已通过 response_format="content_and_artifact" 将模型看到的内容与应用程序代码接收到的内容分开。该工具会返回 (content, artifact),其中 ToolMessage.content 会传递给模型,而 artifact 则保留在消息中用于流程协调,不会进入大语言模型的处理路径。
@tool(response_format="content_and_artifact")
def search(query: str):
return "the content the LLM sees", {"stop": True, "debug": "extra stuff"}
node = ToolNode([search])
result = node.invoke(state)
msg = result["messages"][0]
# msg.content -> "the content the LLM sees"
# msg.artifact -> {"stop": True, "debug": "extra stuff"}
期望的模式是:
tool result
│
┌──────────┴──────────┐
↓ ↓
content artifact
│ │
↓ ↓
model router
│
continue / stop
而 return_direct 则会将所有内容合并为单一的静态答案:
return_direct content_and_artifact
│ │
└── tool content → model
definition artifact → routing metadata
→ routing
要么使用一个固定标志来同时回答“用户会看到什么?”和“循环是否应该结束?”这两个问题,要么设置两个独立的通道分别回答每个问题。
传统的手动构建的 ReAct
不必刻意设计循环结构,只需将 create_react_agent 简化为核心框架:保留节点名称与循环逻辑,删除提示词钩子、结构化响应格式、动态模型解析功能、剩余步骤计数、检查点机制、中断处理以及并行 Send 发送功能。
最终仅保留三个节点:
agent— 调用模型;若存在tool_calls则继续执行,否则结束。tools— 普通的ToolNode,用于添加ToolMessage的处理结果。finalize— 不调用模型,直接将工具选定的最终文本原封不动地封装为AIMessage。
还有两个路由节点:
agent节点之后的should_continue:若有工具调用则进入tools节点,否则返回END。
route_after_tools位于tools之后:检查生成结果后,要么返回给agent,要么进入finalize阶段(取代了原有的静态return_direct判断逻辑)。finalize是一种有意为之的权衡:它跳过了大语言模型的调用,直接显示工具生成的输出内容,但要求工具必须输出可读的文本,并且模型无法将此结果与其他证据结合。对于“工具输出本身就是答案”的情况,这种权衡更为合适。
搜索结果作为元数据,而非图结构指令
搜索工具会根据本次调用产生的结果设置artifact["stop"]。stop属于应用层元数据,并非LangChain预留的字段。关键在于,工具负责报告结果,而调度系统则负责解读这些结果,这样就能让路由逻辑与工具无法看到的策略保持兼容。
@tool(response_format="content_and_artifact")
def search(query: str) -> tuple[str, dict]:
"""Search a knowledge base for information about the query."""
outcome = force_outcome or rng.choices(
list(resolved_weights), weights=list(resolved_weights.values())
)[0]
if outcome == "retryable":
return rng.choice(_RETRYABLE_MESSAGES), {"stop": False} if outcome == "fatal":
return rng.choice(_FATAL_MESSAGES), {"stop": True} query_lower = query.lower()
for topic, page in _INDEX.items():
if topic in query_lower or query_lower in topic:
return page, {"stop": True}
return "Nothing in the index overlaps with this query.", {"stop": True}
三种情况:
- 使用真实内容时成功 →
stop=True(再次调用模型只会重新表述内容)。 - 可重试的失败 →
stop=False(给模型另一次尝试机会)。 - 致命的失败 →
stop=True(不断循环只会浪费资源在同样的无结果上)。
stop=False 并不意味着“立即重试”——它只是暂不强制停止。模型仍可选择重新调用搜索功能、尝试其他方法或直接给出答案。路由器最终会简化为一行代码来检查结果:
def route_after_tools(self, state: AgentState) -> str:
last_message = state["messages"][-1]
if (
isinstance(last_message, ToolMessage)
and isinstance(last_message.artifact, dict)
and last_message.artifact.get("stop")
):
return "finalize"
return "agent"
若使用强制限定为 "success" | "retryable" | "fatal" 的结构,那么在真实的 Groq 模型中路径将是确定的:成功和致命错误会直接进入 tools → finalize → END 步骤,不会再调用模型;可重试的错误则会返回到 agent 步骤。
边界条件:当路由决策还取决于剩余步骤、之前的尝试次数或认证标志时,仅凭结果数据是不够的——路由器必须读取更完整的图结构状态。content_and_artifact在“该工具的检测结果”决定下一跳路径时尤为有用。
超越简单搜索
任何能提供比“成功/失败”更丰富信息的工具都适用:write_record工具可以设置already_applied状态;轮询工具则可为模型从未描述过的界面设置progress进度值。结果数据只是普通数据,可用于条件节点、中间件或从不直接操作图结构的界面。return_direct实际上是内置在定义中的路由决策方式,它没有“暂存信息、稍后决定”的模式。
在create_agent中的相同原理
依赖版本:Python 3.12,langchain==1.4.2 / langgraph==1.2.11,langchain-groq==1.1.3。create_agent用声明式连接方式及中间件替代了手工绘制的图结构。
初步思路是:使用wrap_tool_call,当设置stop时返回Command(goto=END)。
class StopOnArtifact(AgentMiddleware):
def wrap_tool_call(self, request, handler):
result = handler(request)
if isinstance(result, ToolMessage):
stop = isinstance(result.artifact, dict) and result.artifact.get("stop")
if stop:
relay = AIMessage(content=str(result.content))
return Command(goto=END, update={"messages": [result, relay]})
return Command(goto="model", update={"messages": [result]})
return result
在测试版本中,只有当通过return_direct的方式已经能够到达END时,该路径才会直接跳转。中间件可以在循环期间计算出stop=True,但仍然会让流程返回给模型,直到模型最终无需工具即可给出答案。在当前版本中这似乎与#5496的实现一致——直到两种脚本化变体证明了并非如此:
A: update={"messages": [result]} -> stops correctly
B: update={"messages": [result, relay]} -> loops back to the model
版本A可以正常工作。版本B在同一次更新中添加了一个中继AIMessage,但其中没有tool_calls。退出检查会回溯到最后一个AIMessage来评估return_direct;它找到了该中继,但未发现任何工具调用,因此会持续循环。虽然已经应用了Command,但消息的排序方式使得原始的工具调用消息被退出检查所忽略。这并非失败的更新,也不是#5496问题。
即便解决了那个问题,实际使用的方案仍然采用before_model:它根本不需要return_direct。
class StopOnArtifact(AgentMiddleware):
@hook_config(can_jump_to=["end"])
def before_model(self, state, runtime):
last = state["messages"][-1]
if isinstance(last, ToolMessage) and isinstance(last.artifact, dict) and last.artifact.get("stop"):
relay = AIMessage(content=str(last.content))
return {"jump_to": "end", "messages": [relay]}
return None
before_model会在每次调用模型之前执行——在后续迭代中,则是在工具函数执行之后。@hook_config(can_jump_to=["end"])允许忽略任何工具标志直接跳转到END。返回{"jump_to": "end", ...}仅表示状态更新,图中的边会读取该值。同一个钩子既负责检测相关对象,又负责构建中继AIMessage——这项任务被拆分到route_after_tools和finalize中处理。
强制输出结果与手动构建的图结构一致:成功或因工具内容问题导致的致命短路;可重试的情况则会重新进入模型处理流程。
要点总结
content_and_artifact并非为路由功能而设计。它将不同受众区分开来——模型可看到的内容与仅应用能访问的元数据——通过这种分离方式,无需让模型去处理控制流逻辑即可明确判断“是否应停止”。而return_direct则将呈现与终止功能合并为一个静态标志,当每次调用中这两者的答案需要不同时,它就会出错。
如果某个用例需要条件性停止操作,应将工具处理结果与路由决策分开:在答案旁边提供元数据,由调度系统来做出决定。content_and_artifact已经提供了这样的机制。
团队在首次测试通过后容易忽略的设计要点
一旦三种预设结果出现,条件停止机制似乎就已解决问题。生产环境则需要处理并发情况:在一个模型轮次中执行两次工具调用,或者进行多轮搜索却只需最终得到一个结果。需确定是否只要出现任意一种停止条件就会中断整个流程,还是所有条件都必须同时满足,又或者存在优先级顺序。应将此类规则编码到路由器中,而非依赖经验知识。
可观测性系统应能在显示 ToolMessage 的同时呈现相关结果,且不得记录 content 中的敏感信息。当触发停止条件时,需记录是哪条规则起了作用——成功、致命错误还是策略覆盖——以便技术支持人员能够解释为何助手没有“继续思考”。同时应结合令牌计数功能:成功即终止的核心目的就是减少模型调用次数,仪表板应能体现由此带来的节省效果。
在 LangGraph 的不同版本之间移植代码模式时需格外谨慎。中间件钩子名称、Command 的可达性以及返回后的直接退出检查在 0.6 版本到 1.x 版本之间发生了变化。应在每次升级时都运行相应的测试,以确定操作是成功、可重试还是会导致故障。如果某个钩子突然陷入无限循环,应在上报框架错误之前先检查消息列表的结构——消息转发往往是导致此类问题的常见原因。
最后,切勿“仅此一次”地将控制标志塞进 content 中。一旦模型在文本中看到 stop=true,它就可能会向用户描述控制流程或泄露内部代码。设置这样的机制是为了让协调过程更加高效,同时保持面向用户的接口整洁。
将该模式映射到相邻的框架上
在 LangGraph 之外也存在着内容与控制信息的区分现象。任何将工具的标准输出合并到唯一消息通道中的智能体运行时,最终都会发明临时的标记、JSON 封装或旁路元数据。当平台提供官方的旁路通道时,请优先使用;若没有,则需创建有文档记录的封装格式;绝不能依赖模型去忽略隐藏在正文中的控制令牌。
如果一个团队需要同时支持旧的 create_react_agent 图表和新的 create_agent 应用,应保持工具的接口规范不变,仅更换路由器的实现方式。这样就能将版本变动的影响限制在调度测试范围内。当中间件功能增多——如身份验证、支出上限设置、个人敏感信息脱敏等——应在解析 stop 指令之前执行这些钩子函数,以避免将策略拒绝误认为是成功的短路处理。尽管这些钩子的执行顺序看似属于内部实现细节,但它仍是代理程序公开行为的一部分。
为未来的读者说明为何存在finalize(或before_model跳转):这是一种明确的产品设计决策,即允许工具文本在未经润色处理的情况下直接呈现给用户。如果产品日后希望采用口语化总结风格,只需在停止路径上重新添加模型节点,而非让工具同时承担两种表达方式的功能。将“计算结果”与“叙述结果”分开,可使工具在语音、聊天及API客户端之间重复使用。
关于停止与继续的直观理解
想象这样一个结账工具:有时会返回已完成的收据,有时显示“支付处理超时”,还有时会提示“卡片被永久拒绝”。这三种情况分别对应成功终止、可重试的继续处理以及致命终止。收据的content可以是供客户查看的HTML格式;相关数据则包含{ "stop": true, "reason": "completed" }。超时情况会在content中为模型留下简短说明,同时在数据中标记为{ "stop": false, "reason": "transient" }。卡片被永久拒绝时,会通过一条对用户友好的消息终止循环,并标记为{ "stop": true, "reason": "fatal" },以避免代理程序不断向支付处理系统发起请求。同样的架构无需重写路由器,只需调整工具的映射规则,即可应用于搜索、工单创建或文档导出等功能。
配套验证习惯
在持续集成环境中,应使用会发出预定工具调用的虚拟聊天模型来维持强制输出机制。真正的Groq运行仅用于偶尔的端到端可靠性验证,而非每次提交时都进行。需确认具体的路径顺序:哪些节点被执行、是否发生了第二次模型调用,以及在终止路径上最终生成的内容是否与工具输出内容一致。当有人“简化”中间件并重新引入return_direct时,该机制应会发出明显错误提示。应在机制旁边保存标准参考转录内容,以便于对比故障差异。条件终止是一种行为规范;测试则是确保该规范在LangGraph不同版本更新以及面对仅粗略阅读原始设计文档的工程师时依然得以遵循的手段。
如果后续产品需要该模型即使在成功情况下也要将当前轮次的结果与之前的结果进行融合,应在“finalize”节点之后添加一个可选的优化节点,而非删除短路逻辑。特性标志比代码重写更有效:stop_mode=hard|polish|never 可让实验在不破坏结果规范的前提下继续进行。在选择默认值之前,应先在同一查询集上测试每种模式下的令牌消耗情况。
采用该模式的读者规范
在复制正文之前,先复制工件架构和路由器测试代码。这篇文章的价值在于职责分离,而非搜索工具的相关趣事。如果你的领域使用不同的故障标签,应将其映射到相同的三个类别中,并保持路由器的功能简洁。在真正出现故障需求之前,不要添加第四个类别。遇到疑问时,最好继续使用模型,而非因模糊的错误就强制停止——那些隐藏部分故障的静默短路,比多进行一次低成本模型调用并向用户说明不确定性更为糟糕。
将相关代码库与文章一起发布,这样读者就可以在自己的环境中验证边缘情况,再决定是否在生产环境中使用该模式。