工具集、循环、图结构:大多数运行智能体故障背后的三层结构
生产代理故障很少源于模型本身。应将工具线束、循环的终止条件以及 LangGraph 图结构分开处理,每部分都有对应的负责人。
有一个智能体在运行过程中死亡,团队将责任归咎于模型。他们用更智能的模型替换它,结果智能体又死了。提示词被重新编写,问题依然存在。虽然加入了重试逻辑,但情况毫无改善。最终问题被提交给大语言模型提供商,对方称模型的表现符合设计预期。所有人都感到困惑。
这种状况不断重复。看似是模型缺陷的问题,其实源于围绕其周围的机制缺陷,而该机制可分为三层,团队往往不会将它们分开:智能体能够使用的工具(即工具集)、决定工作何时结束的循环(即循环结构),以及智能体可能前往的地点映射(即图结构)。一旦这些层次变得模糊,循环错误或工具错误就会被归咎于模型,人们就会在表面可见的层次上投入精力,而真正的故障却隐藏在底层。
本需三层却只有两层
2026年大多数生产级智能体框架都采用两层结构。其一层是基于LangGraph、CrewAI或OpenAI Agents SDK构建的图结构,另一层则是用于保证稳定性的组件,比如Temporal,或者干脆没有此类组件。而被省略的是管控层——即负责规定智能体可以使用哪些工具、可以调用哪些URL、可以返回什么结果以及何时必须停止运行的那一层。
供应商们也在朝同一方向发展。Claude Code增加了sandbox.network.deniedDomains设置;GitHub为其Copilot智能体在组织层面引入了防火墙控制功能;VS Code则具备了可用于过滤智能体在网络中可访问内容的组策略键。有一份公开的智能体模式目录将出站策略单独列为一类模式。相关研究也表明了同样的趋势:2026年发表在arXiv上的一项研究保持每次运行时模型不变,仅更换六种前沿模型所使用的框架,结果发现每项任务的成本降低了41%,延迟降低了44%。目前只有少数团队在这一层面进行了投入。
这些组件并非孤立存在。在三个核心层之外,还有负责路由请求并控制预算的网关、用于数据检索的内存以及语义缓存、对每一步操作进行追踪的机制、实施配额与速率限制的治理系统,还有基于黄金数据集来决定是否允许部署的评估机制。每一部分都有其价值。不过,通常导致代理程序出问题的并非这些组件,而是混乱的封装方式、循环与图结构的边界问题,因为由此产生的故障很容易被误认为是模型本身的故障。
三个核心层的发现过程
问题的起因是一名采购代理不断重复调用query_suppliers函数。大型语言模型不断声称需要更多数据,循环持续进行,导致代币费用不断上升。与其阅读教程,更有用的其实是事后分析报告、架构文档和事故报告;在各种不相关的项目中,总会出现三个相同的关键词:harness、loop、graph。
在关于代理与架构的文献中,有一种常见的表述方式来描述这种分层结构。harness描述的是单个代理与其周围环境之间的关系;loop描述的是单个代理的执行过程如何随时间被控制;graph描述的是在整个应用中,有多少个代理、工具和检查点相互关联。每一层都建立在下一层的基础之上,只有当下一层足够稳定时,上一层才会显现出来。
拿到这些名称后,最初的缺陷便显而易见了:停止条件位于工具内部而非循环中。Harness与循环被混在了一起。模型本身并未出问题,问题出在层边界上。其他人也报告了类似的情况。有人直截了当地指出:由于使用了错误的模型而显得不可靠的智能体,问题通常出在错误的层上。有一篇新闻稿描述了一位创始人,他删除了一个模型,换上了据说更智能的模型,结果却得到了同样出问题的智能体,因为故障其实出在周围的机制中。
这三层并非此处首创,而是由大规模运行智能体的人所命名,并在多个独立来源中反复出现。以下内容将展示这些名称与实际代码结合时会发生什么。
Harness:智能体能够操作的部分
“工具集”构成了能力边界。它规定了智能体拥有哪些工具、这些工具的输入输出格式是什么、它拥有何种权限、可以读取哪些内存、能够调用哪些URL以及能够查看哪些数据。
可以把智能体想象成一名承包商,“工具集”则相当于挂在工作背带上的身份牌:它决定了哪些门可以打开、哪些系统是可见的,以及进入后能做些什么。它与智能体的智能程度无关,只是定义了可操作的范畴,而安全性也正体现在这一范畴之中。没有“工具集”,智能体就只是持有API密钥的函数调用,可以自由地调用、读取和返回任何内容;有了“工具集”,智能体只能执行“工具集”允许的操作。
大多数智能体项目从未构建过这一层。教程在模型调用工具并收到结果后就结束了,忽略了三种情况:工具返回的10MB JSON数据量超出上下文窗口限制;工具根据注入的文本指令去访问任意URL;以及由于缺乏幂等性键而导致副作用被重复执行两次。这三种情况都属于框架相关问题,解决这些问题才能将演示版本转化为可投入使用的生产系统。
模式目录明确了网络出口的规则。智能体工具可访问的域名应通过框架强制执行的允许列表和拒绝列表来限制,从而将模型置于网络信任边界之外。当模型决定是否建立连接时,提示注入可能会破坏这种控制机制。在框架层面进行检查,能使隔离成为一种结构化特性而非概率性特性。
正是这种区别至关重要。模型具有概率性,容易被欺骗、注入干扰或陷入混乱。而框架则是结构化的,它在代码层面而非提示词层面强制执行规则。礼貌地请求模型不要访问任意URL只是一种请求,而使其无法做到则是一种保证。
循环:智能体何时停止
循环是驱动系统。智能体观察世界、选择动作、执行该动作、检查结果,之后要么继续运行,要么停止。
观察-行动-验证的循环是较简单的部分,难点在于停止条件,因为如果没有停止条件,无人监管的智能体会永远消耗资源而不会明显出错。
采购代理没有这样的机制。它会调用query_suppliers,查看结果后判断还需要更多数据,于是再次调用,如此反复。大型语言模型始终表示需要更多数据,因为语言模型倾向于多次调用工具:它们乐于提供帮助,不断尝试,不愿放弃。这一特性在创意任务中是优势,但在生产任务中却会成为负担。
模式目录将此称为无限循环。其实现方式就是“在模型报告任务未完成时持续迭代”,模型外部没有计数器、计时器或花费上限能够终止这一循环。人们默认模型会宣布任务已完成。但在难度较高的任务中,这种情况很少发生:模型会偏离主题,重复失败的操作,在错误中不断循环而不自知。
这并非假设:
- 2025年7月,Claude Code中出现的异常递归在5小时内消耗了16,000美元至50,000美元不等的费用。并未发生程序崩溃或错误,这些智能体只是无限执行指令而没有任何终止点。
- 公开的漏洞报告指出有子智能体产生了超过50层深度的后代。有一次会话在5分钟内就消耗了400万个标记。还有用户在会话进行到一半时使用了超出套餐限额的额度,需支付27.60美元,且无法停止运行。在第一种情况下,经过三四个智能体的处理后有价值的研究就已经完成,其余44个智能体并未带来任何新成果。在第二种情况下,子智能体将“处理多个独立目标”的指令视为可以自行生成更多后代的许可。
关于有限智能体循环的设计指南也提出了相同要求:在实现循环时就必须设定限制,而非在出现问题后才进行优化。否则,唯一的停止规则就只能是模型自行判定任务完成,这仅是一种期望而非真正的控制手段。
因此,停止条件必须是机械性的,而非语义性的。“大语言模型认为任务已完成”属于语义性条件;而“已尝试三次”或“已花费50美分”则是机械性条件。机械性条件具有确定性、可测试性,并且在重复执行时依然稳定。它们不依赖于模型的判断,因为模型的判断是概率性的,每次重试都可能有所不同。
一家分析公司的值班工程师展示了这会带来多高的成本。该系统原本应在两分钟内找出支付网关超时的根本原因。它搜索了日志,得到了4,200行数据,将其压缩后认为还需要更多上下文,于是再次搜索。经过十二次搜索后,工单仍未生成摘要,而整个处理过程已经消耗了380,000个标记,API费用为47美元。模型中没有任何部分拒绝继续运行,运行时系统也没有发出停止指令。在该公司设置了迭代次数上限、停滞检测机制(用于识别重复出现的几乎相同结果)以及标记预算限制后,失控循环的比例下降了91%,处理一个工单的平均时间也从4.1分钟缩短到了1.6分钟。完成工作的其实是有限循环,而非更强大的模型。
有限循环包含的内容
这个分析示例指出了防止循环无限运行的关键要素。每一项都会由模型外的代码进行检查:
- 迭代次数上限。 对执行次数进行严格限制,这样一旦达到上限,任务就会明确失败,而不会持续无响应地运行。
- 停滞检测。 通过记录最近的结果特征,即使步骤次数还很少,但如果连续出现多个几乎相同的输出,也会被视为没有进展。
- 资源消耗上限。 设定token或资金的使用限额,该限制由运行时机制而非提示词来控制。
- 定时超时。 对经过的时间设置限制,以此捕获那些虽未触发其他计数机制但运行极慢的循环。
这些限制都不需要模型的配合,且每个限制都可以通过一个永远不会声明任务完成的虚拟模型来进行单元测试。如果三次尝试后仍以失败告终,将会产生一定的费用;而让模型自行决定结果的测试,则需支付相应费用。
图结构:智能体可前往的路径
图结构即拓扑结构:包括存在的节点、连接节点的边、路径的分支点、人工审批的位置,以及从一个节点传递到另一个节点的状态。如果图结构是地图,那么循环机制就是在地图上移动的引擎,而各种限制则决定了在每个节点上可以发生什么事情。
核心设计原则是明确性。ReAct循环(思考、行动、观察、重复)实际上是一个隐式图结构,其中一个节点负责自我反馈,而大语言模型则充当边函数。对于简单任务来说这已经足够。但一旦任务结构变得复杂,这种设计就会失效。将ReAct与LangGraph进行对比就能很好地说明这一点:在ReAct中,模型会根据提示和历史信息自行选择下一步操作;而在LangGraph中,则由开发者通过代码和明确的路径控制来决定下一步操作。
一旦需要人工审批环节,差异就会显现出来。在隐式图中并不存在“之间”的概念:智能体依靠自身的推理从一步过渡到下一步。加入检查点意味着要打破循环,引入暂停机制,将状态外部化,并在收到信号后继续执行。这实际上是在循环上强行套用图结构,而循环会抵制这种改变,因为它原本就是为自主运行而设计的。在显式图中,只需添加一个带有interrupt()调用的新节点即可。图结构会随之改变,而循环及配套机制则保持不变。
有位开发者通过普通的工具调用循环以及显式的LangGraph状态图,对相同的策略、确定性模型和三种场景进行了测试。两者的行为表现完全一致,但状态图版本需要97行代码,而普通循环版本仅需50行,代码量多了约94%;因此在三个步骤的测试中,普通循环在所有指标上都更优。该开发者随后指出了此类比较无法体现的问题:在循环中,行为是由语句顺序决定的,必须通过阅读代码主体才能理解;而在状态图中,边被作为数据明确声明,这使得无需打开任何节点即可查看、绘制和重现拓扑结构。在三个步骤时其值为nil,三十步时变为较大数值,而当换其他人操作时数值会更大。
这就是生产环境的论据。演示版本包含三个步骤,由其编写者负责管理;而生产版本则有三十个步骤,由凌晨3点值班的人负责。在如此大规模的系统中,明确的拓扑结构是生存必需,而非额外负担。
判断这两种形式哪种更合适的实际测试方法是,看看六个月后谁来负责管理该代理。由编写者维护的简单、单一功能的代理可以保持为单纯的循环结构;而需要人工审批、包含多个分支或由不同人员负责管理的代理,则需要明确的图结构,因为这样的拓扑结构可以转化为可绘制、可审查和可回放的数据。
层次混淆的表现形式
以下的故障模式源自构建阶段,而构建阶段正是容易让各层次相互混淆的环节。这些故障模式在生产环境中也会出现。
负责管理循环结构的框架
在那种混乱的设计中,“这是否足够好”的判断位于工具内部,因此由工具来决定循环何时停止。工具应当是一种能力,比如“能够查询供应商”,而非一个会做出“够了,停止”决定的决策者。将停止条件埋在工具内部会导致该循环无法独立测试,使得工具在不重新实现停止逻辑的情况下无法更换,同时由于工具对尝试次数一无所知,也就不存在设置尝试次数的上限。
解决方案在于从结构上进行调整:将停止条件从工具中提取出来并放到图节点中——工具负责返回数据,节点决定是否需要再次处理,框架提供相应能力,而循环则存在于图中。那种混乱设计的代码形式如下:
# CONFUSED LAYER — Harness owns Loop logic
async def query_suppliers(item: str, budget: float, quantity: int, attempts: int) -> dict:
"""Tool decides when the loop stops. Capability + control flow in one function."""
results = query_mock_marketplace(item, budget, quantity)
# The tool decides when to stop — couples execution logic with capability
if len(results) > 5 or attempts >= 3:
return {"status": "STOP", "suppliers": results}
return {"status": "CONTINUE", "suppliers": results}
async def research_node(state: AgentState) -> dict:
# The node just calls the tool and forwards whatever the tool decided.
# It has no idea why the loop stopped — that logic is hidden inside the tool.
result = await query_suppliers(
state["brief"].item, state["brief"].budget, state["brief"].quantity,
state.get("attempts", 0) + 1,
)
return {"suppliers": result["suppliers"], "status": result["status"].lower()}
# ✅ CLEAN SEPARATION — Harness is pure capability, Loop lives in the Node
MAX_RESEARCH_ATTEMPTS = 3
async def query_suppliers_impl(item: str, budget: float, quantity: int) -> dict:
"""Pure Harness capability: schema validation, circuit breaker, marketplace call.
The tool has no idea a loop exists. It returns data. It does not decide
when to stop. Circuit breaker + fault injection live here because they are
capability concerns (is the downstream reachable?), not control-flow concerns.
"""
breaker = get_circuit_breaker("query_suppliers")
if not breaker.allow_call():
raise CircuitBreakerOpenError("query_suppliers", breaker.state)
maybe_inject("query_suppliers") # fault injection for chaos tests
try:
results = query_mock_marketplace(item, budget, quantity)
breaker.record_success()
return {"suppliers": results}
except Exception as e:
breaker.record_failure()
raise
async def research_node(state: AgentState) -> dict:
"""Graph Node / Loop: mechanical stop conditions and routing decisions.
The node reads state, calls the tool through the registry, and decides
whether to stop — based on counters in state, NOT on the LLM's judgment
and NOT on the tool's opinion.
"""
brief = state["brief"]
attempts = state.get("attempts", 0) + 1
# Harness call — schema validation, timeout, output cap all happen inside registry.call
query_result = await registry.call("query_suppliers", {
"item": brief.item,
"budget": brief.budget,
"quantity": brief.quantity,
})
suppliers = query_result.get("suppliers", [])
# Mechanical stop condition — enforced OUTSIDE the tool, in code, on state.
# Deterministic. Survives replay. Testable without the tool being live.
if not suppliers and attempts >= MAX_RESEARCH_ATTEMPTS:
logger.info(
"STOP_CONDITION_FIRED path=no_matches item=%s attempts=%d max=%d",
brief.item, attempts, MAX_RESEARCH_ATTEMPTS,
)
return {"suppliers": [], "attempts": attempts, "status": "no_matches"}
if not suppliers:
return {"suppliers": [], "attempts": attempts, "status": "running"}
# Enrich, then mark completed
enriched = []
for s in suppliers:
quote = await registry.call("get_price_quote", {
"supplier_name": s["name"], "item": brief.item, "quantity": brief.quantity,
})
rating = await registry.call("check_seller_rating", {"supplier_name": s["name"]})
enriched.append({**s, "quote": quote, "rating_info": rating})
return {"suppliers": enriched, "attempts": attempts, "status": "completed"}
# The Graph owns WHERE the agent can go — conditional edges, not the node body.
def _should_retry_or_end(state: AgentState) -> str:
"""After Research: loop back if still running, else Score or END."""
if state["status"] == "running":
return "research" # self-loop — the retry
if state["status"] == "no_matches":
return END # mechanical stop → graph terminates
return "score" # success → next node
# Wiring (in build_sourcing_graph):
# graph.add_conditional_edges("research", _should_retry_or_end,
# {"research": "research", "score": "score", END: END})
管理图结构的循环
许多框架会提供一个拓扑结构隐含的ReAct循环:智能体根据模型的输出决定下一步行动,不存在明确的映射关系。这种情况在需要人在决策与确认之间进行审核时才会改变,随后就需要进行改造,而循环则会与之抗衡。
解决办法是从一开始就明确设计图结构,将节点视为函数,边视为状态转换,其中一些边是条件性的,另一些则起到中断作用。循环依然负责驱动智能体前进,但可达区域是由图结构定义的,而非循环。那种将循环视为图结构、将大语言模型视为边函数的单一化ReAct形式如下所示:
# ❌ CONFUSED LAYER — Implicit topology inside a monolithic ReAct loop
# The loop IS the graph. The LLM is the edge function. There is no "between."
while not agent.is_done():
action = llm.decide_next_step(history)
if action == "query":
data = query_suppliers()
elif action == "approve":
# Retrofitting human approval into an implicit loop requires messy pauses.
# You have to break the loop, externalize state, park the worker, and
# resume on a webhook — the loop fights you because it was designed to
# be self-driving.
pause_execution_and_wait_for_webhook()
elif action == "stop":
break
# ✅ CLEAN SEPARATION — Declarative graph with explicit nodes and human interrupt
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt
def approve_node(state: AgentState) -> dict:
"""Human checkpoint via interrupt() — a STRUCTURAL pause, not a prompt.
The graph genuinely suspends here. interrupt() pauses execution and waits
for a human to call Command(resume=approval_data). The worker is freed.
If the worker crashes while waiting, Temporal replays the activity, the
graph re-runs from the last checkpoint, and interrupt() fires again —
the human re-approves. No data loss.
"""
selected = state.get("selected_supplier", {})
brief = state.get("brief", {})
approval_request = {
"item": brief.item,
"quantity": brief.quantity,
"supplier": selected.get("name", ""),
"price": selected.get("price", 0),
"total_cost": selected.get("price", 0) * brief.quantity,
}
# interrupt() pauses the graph HERE. The human must call /approve
# with a resume value to continue.
approval = interrupt(approval_request)
approved = approval.get("approved", False)
comment = approval.get("comment", "")
return {
"approval_status": "approved" if approved else "rejected",
"approval_comment": comment,
"rejection_count": state.get("rejection_count", 0) + (0 if approved else 1),
}
def _after_approve(state: AgentState) -> str:
"""After Approve: go to Confirm if approved, retry Research if rejected."""
if state.get("approval_status") == "approved":
return "confirm"
if state.get("rejection_count", 0) + 1 >= 3:
return END
return "research"
def build_sourcing_graph(checkpointer=None):
"""Topology is DECLARED DATA, not statement order in a loop body.
START → Research → (retry) → Score → Decide → Approve → Confirm → END
↓ (rejected, < 3)
Research
↓ (rejected, = 3)
END
"""
graph = StateGraph(AgentState)
graph.add_node("research", research_node)
graph.add_node("score", score_node)
graph.add_node("decide", decide_node)
graph.add_node("approve", approve_node) # explicit checkpoint node
graph.add_node("confirm", confirm_node)
graph.add_edge(START, "research")
graph.add_conditional_edges("research", _should_retry_or_end,
{"research": "research", "score": "score", END: END})
graph.add_edge("score", "decide")
graph.add_edge("decide", "approve")
graph.add_conditional_edges("approve", _after_approve,
{"confirm": "confirm", "research": "research", END: END})
graph.add_edge("confirm", END)
return graph.compile(checkpointer=checkpointer)
由工具集管理的图结构
这种情况较为少见,但也会出现在代码审查中。图结构中的节点会直接调用外部 API,例如在节点函数内部通过 HTTP GET 请求联系供应商的接口,这样一来图结构就掌控了该工具调用。
这种做法是错误的,因为图结构属于拓扑结构,本不应涉及 HTTP、身份验证、重试机制或断路器等概念。由于调用操作在节点内部,要添加断路器就必须修改图结构,要更换实现方式同样需要修改图结构,而且测试图结构时还需要实时的工具支持。
正确的解决方案是通过工具集来路由调用。节点只需调用 call_tool("query_suppliers", {...}),即可获得经过验证且类型正确的结果。工具集负责处理 HTTP 请求、重试逻辑、断路器功能以及数据结构验证。图结构无需知晓 HTTP 的存在,而工具集也无需知道图结构的存在:
# ❌ CONFUSED LAYER — Graph node owns the tool call (Harness concern)
import httpx
def supplier_node(state: AgentState) -> dict:
"""The graph node reaches directly into HTTP, auth, retries, and parsing.
Why that's wrong: the graph is topology. It should not know about HTTP,
authentication, retries, circuit breakers, or schema validation. Those are
harness concerns. When the graph owns the tool call:
- You can't add a circuit breaker without modifying the graph.
- You can't swap the tool implementation without modifying the graph.
- You can't test the graph without the tool being live.
- You can't enforce an egress allowlist — the URL is hardcoded here.
"""
resp = httpx.get(
f"https://supplier-api.example.com/suppliers",
params={"item": state["brief"].item},
headers={"Authorization": f"Bearer {SUPPLIER_API_KEY}"},
timeout=10.0,
)
resp.raise_for_status()
suppliers = resp.json()["suppliers"] # no schema validation, no output cap
return {"suppliers": suppliers}
# ✅ CLEAN SEPARATION — Graph node calls through the harness; harness owns the call
async def supplier_node(state: AgentState) -> dict:
"""The graph node says 'call this tool with these args' and gets back a
validated, typed result. The graph doesn't know there's HTTP involved.
The harness doesn't know there's a graph involved.
"""
query_result = await registry.call("query_suppliers", {
"item": state["brief"].item,
"budget": state["brief"].budget,
"quantity": state["brief"].quantity,
})
suppliers = query_result.get("suppliers", [])
# Mechanical stop condition — the node's job, not the tool's
if state.get("attempts", 0) + 1 >= MAX_RESEARCH_ATTEMPTS and not suppliers:
return {"suppliers": [], "attempts": state.get("attempts", 0) + 1, "status": "no_matches"}
return {"suppliers": suppliers, "attempts": state.get("attempts", 0) + 1, "status": "completed"}
# What the harness does — the graph never sees this:
#
# registry.call("query_suppliers", {...})
# │
# ├─ 1. validate_input(QuerySuppliersInput, args) # Pydantic v2
# ├─ 2. EgressFilter(spec.allowed_egress) # URL allowlist
# ├─ 3. SandboxedHTTPClient(egress_filter) # blocked domains = structural
# ├─ 4. run_with_timeout(fn, kwargs, spec.timeout_seconds)
# ├─ 5. enforce_output_cap(result, spec.max_output_bytes) # 64KB cap
# └─ 6. validate_output(QuerySuppliersOutput, result) # garbage caught here
#
# The graph node gets back a validated dict. It never sees the HTTP call,
# the auth header, the retry, the circuit breaker, or the egress filter.
# Swap httpx for aiohttp, swap the supplier API for a new one, add a circuit
# breaker — the graph doesn't change. The harness doesn't know a graph exists.
各层由谁负责
每一层都需要相应的基础设施来管理。如果责任不明确,各层之间就会出现混乱。
Temporal负责循环的崩溃安全机制
当引擎在某个循环过程中停止运行时,Temporal会负责恢复工作。它会将每个步骤记录在事件历史中。当某个工作节点崩溃后,它会重新播放工作流程,而不需要重新执行相关操作,而是直接使用已记录的结果,这样循环就能从上一个完成步骤处继续,而非从零开始。
这就是为什么停止条件必须是确定性的。像“模型判断工作已完成”这样的条件是不确定的,重新播放时可能会得出不同的结论。而基于步骤计数器、花费上限或工具输出得出的质量阈值则适用于重新播放场景。模型可以参与决策,但if attempts >= 3: stop这样的判断应写在代码中。
LangGraph掌控着图的拓扑结构
在LangGraph中,节点是普通函数,边代表状态转换,而条件边则是返回下一个节点名称的函数。LangGraph会将所有这些元素编译成一个可运行的图结构。由于该图结构是明确的,因此可以对其进行查看,每个节点都可以单独测试,而且添加新节点也不会影响其他节点。人工审批节点只不过是一个调用interrupt()的节点,这一操作会改变图结构,但不会影响循环和约束条件。
LangGraph还掌控着状态的形态,即节点之间传递的TypedDict。每个节点只读取自己需要的数据并写入自己产生的数据,因此状态就充当了节点间的契约。如果某个节点从不写入suppliers数据,那么后续节点就无法读取它,类型检查器会在运行前就发现这一问题。
手工构建的工具框架涵盖其他所有功能
这是Temporal和LangGraph都无法提供的功能,因此必须从零开始构建。它负责处理:
- 注册。每个工具都需要声明名称、Pydantic输入架构、输出架构、超时时间以及是否需要幂等性。
- 验证。在工具运行前会对输入进行验证,运行后则会对输出进行验证,从而在智能体看到错误响应之前将其拦截。
- 输出过滤。执行HTTP调用的工具会通过带有URL允许列表的沙箱客户端,因此没有任何工具能够调用任意地址。这是由框架决定的,而非模型。
- 输出限制。结果在进入上下文窗口之前会被截断到64KB。如果数据大小达到10MB,将会被截断并附加
_truncated: true标记。
这些功能并非来自Temporal或LangGraph,而是用于确保工具调用安全的层,不过大多数代理项目都会忽略这一层。
验证:“代理称其有效”与“实际有效”
该循环的第三步是验证,且必须由外部完成。这不能意味着智能体自行检查自己的输出,也不能是大型语言模型给自己打分。
该模式手册将这种修正方式称为“验证日志”:原本由“构建成功”之类的自行报告构成的信息,被替换为结构化记录,其中每次检查都对应一次数据库的INSERT操作,每组证据则对应一次SELECT操作。如果任由智能体自行操作,它可能会误以为所有检查都已通过,跳过某些步骤而不作说明,或者声称使用了从未实际运行的工具得出的结果。
Spotify的Honk团队就遇到了这样的问题。他们的智能体有时会声称“测试通过,未发现任何问题”,但实际上是错误的。他们在循环中加入了用于格式、构建和测试的确定性验证器,一旦有任何验证器失败就会阻止创建拉取请求。这样一来,智能体的描述就不再作为成功信号,取而代之的是验证器产生的结构化输出。
其他项目则采取更进一步的措施。某个开源工具要么会给出“已完成,所有测试均通过”的确认信息,要么会标记该声明为未经验证;而仅凭测试通过的结果本身永远无法提升声明的语义级别。有一篇研究论文将这一规则表述为:需要重新推导该声明且不可轻信它,因为智能体提供的文字描述根本无法作为有效输入。
在此处描述的流程体系中,验证是一项独立的临时性操作,比如 verify_po_exists 或 verify_payment_initiated,这类操作会直接查询数据库。智能体声称自己已创建采购订单,此时该验证操作就会检查采购订单数据库,如果找不到相应订单,那么无论智能体报告了什么情况,整个工作流程都会失败。
自我报告是最容易出错且最难发现的信号,因此必须通过外部方式进行验证。名为create_po的代理解析了响应并称操作成功。但实际上,该响应可能是包含错误信息的200 OK状态,或者订单虽已创建但因交易超时而被回滚,又或者代理读取了错误的字段并报告了根本不存在的ID。该代理只知道工具返回了什么,却对数据库状态一无所知,因此需要向数据库查询确认。
实际测试结果也支持了这一观点。2026年7月发表的一篇论文研究了名为Leni的AI业务分析工具,它通过专门针对小型任务设计的模型来执行验证、观察、比较和纠错操作。在SpreadsheetBench测试中,该完整系统的性能比其基础模型高出11个百分点,分别为91.25%和80.25%。仅验证功能就为性能提升了1.5个百分点,而这些提升值大多集中在分数较高的区间,帮助解决了那些原本会失败的任务。整体性能提升的主要来源是框架结构、任务路由机制以及专用模型。验证功能并非能解决所有问题,但它能挽救那些处于临界状态的任务——即智能体声称任务已完成,但实际结果并不符合预期的情况。
分离原则
核心准则是:每一层都应具备可测试性、可替换性,并且无需依赖其他层就能被理解。
- 可以通过模拟工具调用来检查图结构,确认正确的节点能按正确顺序被触发。
- 可以直接调用工具并检查模式验证、断路器机制以及幂等性,以此测试集成框架的功能。
- 可以在 Temporal 工作流中加入简单的任务来测试循环逻辑,确认其在系统崩溃后仍能正常运行。
更改图结构不会影响集成框架,更换工具实现也不会影响图结构,而修改循环的重试策略则不会影响二者中的任何一个。
其实际效益显而易见。采购代理的流程图经过了三次修改:增加了评分节点、具备拒绝处理功能的重试循环以及人工审批环节,而无需修改任何代码。大语言模型提供商则从直接使用OpenAI的方式转变为通过路由层来调用模型,同样没有改动流程图代码。对于那些会产生副作用的操作,重试策略也从积极重试改为严格禁止重试,且无需修改流程图或相关代码。分离设计带来的灵活性远超过纯粹性:它让开发者能够在不破坏其他三个部分的情况下单独修改某一部分。
即便出现故障,同样的方法也能发挥作用。通过提出三个具体问题,就可以逐层追溯失败原因。工具是否返回了本不应返回的数据,或到达了本不应到达的位置?这涉及到控制框架。代理是否超出了任何合理的限制?这涉及到循环结构。代理是否最终到达了本不应到达的节点,或跳过了某个检查点?这涉及到图结构。更换模型并不能解决这三个问题中的任何一个。
任何代理设计都需要回答的三个问题
- 它能接触什么?(控制框架)现有的工具有哪些,其数据结构如何,允许的输出量是多少,断路器阈值设定在何处,以及适用哪些幂等性键。
- 何时停止?(循环)步数上限、成本上限、超时时间,以及不依赖于大语言模型判断的机械性停止条件。
- 它能走向何处?(图结构)节点、边、条件分支、人工检查点,以及在节点间流动的状态形态。
当任何答案都是“由模型决定”时,就存在问题。大语言模型提供建议,代码负责做出决策,框架负责执行,图结构负责路径规划,循环则设定边界。
简而言之,存在三层结构、三个负责人以及三个问题。如果将它们混为一谈,错误就会看起来像是模型本身的问题,但实际上并非如此。若将它们分开处理,模型的不可靠性就能得到控制,工具的使用也更安全,智能体也能在适当时机停止运行。
相关阅读
- 您的智能体图应存储在 YAML 中:在启动时编译包含 17 个智能体的 LangGraph —— 小型运行时如何将一个 YAML 规范编译为 LangGraph StateGraph:生成的状态、importlib 插件、标签路由器、自适应调用以及会话级别的节点替换。
- 当控制面板显示正常但智能体行为错误时:状态追踪 —— 智能体故障如何隐藏在正常步骤之间传递的状态中,以及需要通过何种追踪方式、运行时约束、路由记录和评估来发现这些问题。