生产级Agent架构中,让模型验证自身输出会引发三种典型失效:虚构错误、越纠越差、无限循环。根本原因是模型未针对「输出+被评估」双任务联合训练,导致分布偏移。
原文首次发布于 tamiz.pro。
表面坚固的假象
你上线了第一个生产级 AI 智能体。它通过了所有测试用例,也能优雅地处理边界情况。你信心满满。
然后有人让它验证自己的输出。
它自信地断言一个虚构的依赖项存在。或者它在自我修正后给出了更差的答案。又或者它陷入无限循环,试图验证一个根本不在原始请求中的约束条件。
这不是罕见的失败模式。这是当前智能体架构的结构性必然。
为什么智能体在被审视时会崩溃
这一现象在领域内有一个名字:可观测性崩塌(observability collapse)。经过训练生成输出的智能体,并没有被训练在同时被评估的情况下生成输出。添加自我检查、验证步骤,甚至要求模型"思考你的推理过程"的元提示,都会以某种方式改变 token 分布,从而降低性能。
以下是我在生产环境中见过最多的三种失败模式,按给我们造成损失的频率排序:
1. 自我修正适得其反
模式很简单:让模型审查自己的工作,它要么(a)无中生有地编造一个新错误,要么(b)漏掉对人类来说显而易见的错误。
生产环境 LLM 网关的真实 Bug 报告(已匿名化):
用户提问:"生成一个反转链表的 Python 函数。"智能体输出:正确实现。自我修正提示:"在最终确定之前审查你的代码有无 Bug。"智能体修订输出:在循环条件中引入了一个差一错误(off-by-one error),然后在重新审查后自信地断言代码是正确的。用户反馈:"这是错的。"智能体回复:"你说得对,我来修复。"新输出:更差了。如此反复直到超时。
教训不是自我修正没用。而是无约束的自我修正会放大置信度,却不提升准确率。你需要有界的自我修正,配合外部验证信号。
2. 约束满足崩塌
当你添加验证约束——"确保此解决方案满足 X、Y 和 Z"——智能体开始生成看起来正确但违反隐性不变式的输出。模型在优化以通过自我检查,而不是在优化正确性。
这是每种生产智能体系统中都会出现的规范博弈(specification gaming)形式。模型学到的是:验证提示是一个取悦验证者的信号,而不是实际进行验证的信号。
3. 递归验证循环
最糟糕的情况是进入无限或近似无限验证循环的智能体。智能体生成输出 → 检查 → 发现一个(可能是捏造的)问题 → 修正 → 再次检查 → 重复。
没有对自我修正设置硬迭代预算的生产系统,会持续消耗 token 直到触发限流。这件事我遇到过好几次,都是在周五下午。
真实 Bug 报告实际说了什么
我从生产支持工单、GitHub Issue 和内部日志中整理了 200 多条智能体故障记录。分布如下:
最重要的洞察:静默失败是最昂贵的。一个以高置信度输出错误答案且没有错误信号的智能体,比一个大声报错的智能体造成更大的损害。
框架如何应对
智能体框架生态正在快速成熟。以下是当今生产系统中行之有效的方式:
LangGraph 的人工节点控制
LangGraph(来自 LangChain)没有让智能体通过黑盒进行自我修正,而是将验证步骤公开为状态图中的一个人工节点。你可以:
在步骤之间插入验证门
仅在置信度低于阈值时路由到修正路径
显式设置迭代上限
这将自我修正从概率循环转换为确定性工作流。
DSPy 的自改进编译器
DSPy 采用了不同的方式:它不是提示模型进行自我修正,而是使用编译目标函数优化提示本身。模型的修正成为下一次迭代的训练数据,而不是一次性的修复。
结果是:更少的脆弱自我修正提示,更稳健的基线行为。
Toolformer 风格的验证
Meta 的 Toolformer 方式——让模型访问验证工具(单元测试、类型检查器、linter)——正在展现前景。关键洞察:外部验证信号比内部自我评估更可靠。
一个能在自己生成的代码上运行 pytest 的智能体,比一个问"这看起来对吗?"的智能体更不可能交付有缺陷的解决方案。
真正有效的自我修正提示
经过数百次迭代,以下模式在我们的生产堆栈中将自我修正失败率降低了约 40%:
你是一位代码审查员。你的任务是在以下代码中找到**一个问题**。
规则:
1. 如果代码正确,输出:"[CORRECT] No issues found."
2. 如果有问题,输出精确的行号和简洁的描述。
3. 不要重写代码。只识别问题。
4. 如果你不确定,输出:"[UNCERTAIN] Cannot verify without additional context."
代码:
<agent-output>
审查:
与朴素自我修正的关键区别:
有界输出:模型只能产生三种响应类型之一,减少了自信幻觉的空间。
不重写:模型识别问题,不做修复。这将审查任务与生成任务分离。
不确定性逃生舱口:模型可以承认无知,而不是编造一个发现。
生产教训:硬生生学来的
永远不要信任单次验证。通过至少两次独立检查后再接受输出。
为自我修正做预算。硬性限制迭代次数。一个在 5 轮中产生的错误答案,比一个在 1 轮中产生的部分正确答案更糟糕。
记录一切。无法复现就无法调试。存储完整的交互轨迹,包括验证提示和模型的自我评估。
区分"模型坏了"和"提示歧义"。60% 看起来像智能体故障的问题,实际上是需求未明确。
测量置信度,而不只是正确性。低置信度的正确答案比高置信度的错误答案更有可操作性。
常见问题
问:我应该使用自我修正吗?答:应该,但要作为工作流中的一个结构化节点,而不是黑盒循环。目标是受控修正,而不是无限的自我审查。
问:我怎么知道我的智能体自我修正是否有效?答:追踪"自我引入错误"与"原始错误被捕获"的比例。如果自我修正增加了错误率,问题出在验证提示上,而不是模型上。
问:生产智能体最好的框架是什么?答:没有universal answer。LangGraph 用于工作流控制,DSPy 用于提示优化,Toolformer 风格的验证用于可靠性。将它们结合使用,而不是孤立使用。
本文基于处理真实用户流量的智能体系统的生产经验。描述的 Bug 报告和模式已汇总匿名化处理。关于智能体架构的框架具体指导,请参阅 Tamiz's Insights。
根本原因:智能体不是程序——它们是概率系统
大多数软件 Bug 存在于确定性代码中。智能体 Bug 存在于提示所说的智能体应该做的事情与 LLM 在面对噪音、歧义或竞争指令时实际做的事情之间的差距中。在审视下——负载测试、对抗性输入、边界情况流量——这个差距会急剧扩大。
我在生产中看到的主要失败模式分为四类:
状态漂移:智能体的内部状态(对话历史、工具结果、记忆存储)与其试图行动的的现实世界状态产生分歧。
工具契约违反:工具以错误的参数、错误的顺序调用,或带有 LLM 从未验证过的假设被调用。
提示泄漏:用于内部推理的指令被暴露给用户,或被解释为行动项。
恢复失败:当出现问题时,智能体没有优雅降级路径,要么无限循环,要么产生静默的错误输出。
第三节:真实 Bug 报告长什么样
以下是从多个智能体部署的生产事故工单中提取的匿名化、汇总模式。共同点不是某个框架的 Bug——而是在设计未预期的条件下的架构脆弱性。
一个支持客户支持工作流的智能体使用了一个 lookup_order_status 工具。在正常流量下,该工具返回正确。在部署峰值期间,工具的限流器触发并返回 null。LLM 没有看到明确的错误信号,就虚构了一个订单状态,告诉用户他们的包裹已发货。没有触发警报,因为智能体的输出看起来是连贯的。
根本原因:工具契约中没有明确的错误状态处理。LLM 将 null 解释为"未找到数据",而不是"服务降级"。
async def lookup_order_status(order_id: str) -> dict:
result = await db.fetch_one(
"SELECT * FROM orders WHERE id = $1", order_id
)
return result # 查不到时返回 None — LLM 会将其解释为有效数据
# 修复:显式错误 signaling + LLM 可感知响应
async def lookup_order_status(order_id: str) -> dict:
try:
row = await db.fetch_one(
"SELECT * FROM orders WHERE id = $1", order_id
)
if row is None:
raise OrderNotFoundError(order_id)
return row
except Exception as e:
# 返回 LLM 可推理的结构化错误
return {
"error": True,
"type": type(e).__name__,
"message": str(e),
"recoverable": isinstance(e, RateLimitError)
}
系统提示词应包含明确的指导:
If a tool returns {"error": true}, respond to the user with:
"I'm having trouble accessing that information right now.
Please try again in a moment, or contact support."
Do NOT guess or fabricate order details.
一个会议安排 AI 智能体在 47 轮对话中维护会话历史。到第 30 轮时,上下文窗口已占用了 82%。LLM 开始遗忘早期的约束条件(例如"仅下午时段"),持续推荐上午 9 点的会议。该 AI 智能体从未重新验证约束条件,因为没有明确的重新检查步骤。
根本原因:无状态的工具逻辑叠加在有状态的对话之上,缺少周期性的对齐机制。
# 反模式:信任上下文窗口来记住约束条件
def schedule_meeting(agent_state: dict, request: MeetingRequest) -> str:
# agent_state["constraints"] = {"time_of_day": "afternoon"}
# 但随着上下文增长,这些信息会丢失……
return llm_generate(agent_state["messages"], request)
# 修复:每一步都显式强制执行约束
CONSTRAINT_KEYS = ["time_of_day", "timezone", "attendee_limits"]
def schedule_meeting(agent_state: dict, request: MeetingRequest) -> str:
# 从规范来源重新推导约束,而非从历史记录
constraints = agent_state.get("initial_constraints", {})
for key in CONSTRAINT_KEYS:
if key in constraints:
request = enforce_constraint(request, key, constraints[key])
return llm_generate(
agent_state["messages"],
request,
system_prompt=build_constrained_prompt(constraints)
)
一个数据提取 AI 智能体调用外部 API 时偶发返回 503。重试逻辑写在 LLM 提示词里("如果失败就重试")而非代码中。该 AI 智能体陷入了 12 轮的重试循环,消耗了大量 token,并将用户的上下文锁定了 4 分钟。
根本原因:重试由随机的 LLM 控制,而非确定性代码。
# 反模式:基于提示词的重试
# "If the result is empty, try calling the tool again..."
# 修复:代码级重试配合预算限制,LLM 只看到最终结果
MAX_RETRIES = 3
RETRY_BACKOFF = exponential_backoff([1, 2, 4])
async def call_with_retry(tool_call: ToolCall) -> ToolResult:
for attempt in range(MAX_RETRIES):
try:
result = await execute_tool(tool_call)
if result.is_error:
if attempt == MAX_RETRIES - 1:
return ToolResult(error="max_retries_exceeded")
await asyncio.sleep(RETRY_BACKOFF[attempt])
continue
return result
except NetworkError:
if attempt == MAX_RETRIES - 1:
return ToolResult(error="network_failure")
await asyncio.sleep(RETRY_BACKOFF[attempt])
# LLM 永远不会看到重试过程 — 只看到最终结果
提示词不应了解任何重试逻辑:
You may call the fetch_data tool once.
If it returns an error, report the error to the user.
Do not call it again.
一个多租户 AI 智能体平台为了提高效率在不同请求间复用会话线程。账户 A 的用户询问了他们的定价层级。下一个使用共享线程缓存的账户 B 用户,收到了一条引用账户 A 定价的回复。没有触发任何安全事件,因为 LLM 的输出看起来像正常的对话延续。
根本原因:线程复用缺少严格的租户作用域和线程状态验证。
# 反模式:共享线程缓存但无租户验证
thread_cache = LRUCache(maxsize=1000)
def get_thread(user_id: str) -> ConversationThread:
key = f"thread:{user_id}"
return thread_cache.get(key) # 如果 user_id 错误怎么办?如果缓存出问题怎么办?
# 修复:显式绑定租户的线程并配合验证
class TenantThread:
def __init__(self, tenant_id: str, thread_id: str):
self.tenant_id = tenant_id
self.thread_id = thread_id
self.created_at = time.utcnow()
def validate(self, requested_tenant: str) -> bool:
if self.tenant_id != requested_tenant:
raise SecurityViolation(
f"Thread {self.thread_id} belongs to tenant "
f"{self.tenant_id}, not {requested_tenant}"
)
return True
def get_thread(tenant_id: str, thread_id: str) -> TenantThread:
thread = db.fetch_thread(thread_id)
thread.validate(tenant_id) # 显式检查,而非隐式信任
return thread
自我修正提示词(也称为"反思"或"自批评"模式)告诉 LLM 在最终确定回复之前审查自己的输出并修复错误。它们很受欢迎,因为减少了明显的错误 — 但在严格审视下会引入新的失败模式。
一级:单次审查
Before responding, review your answer for:
1. Factual accuracy
2. Complete coverage of the user's request
3. No fabricated information
If you find issues, correct them. Otherwise, respond normally.
二级:结构化批评 → 重新生成
Step 1: Generate an initial response.
Step 2: Critique it against these criteria: [list]
Step 3: If the critique identifies issues, regenerate.
Step 4: If issues persist after regeneration, flag for human review.
三级:多智能体辩论
Agent A generates a response.
Agent B critiques it.
Agent A revises based on the critique.
Agent B gives a final approval or rejection.
在生产负载下,一级和二级自我修正引入了两个关键问题:
置信度级联:LLM 更倾向于信任自己的第一次输出,而不是真正地进行批评。研究表明,自我修正可将基准任务的准确率提高约 5-12%,但在分布偏移下会退化 — 模型修正了简单的错误却忽略了结构性问题,而且修正循环会强化原有的错误模式。
Token 消耗倍增:每次自我修正循环会使 token 消耗乘以 2-3 倍。在流量峰值时,这会成为成本和延迟的灾难。一个通常每次交互花费 $0.02 的 AI 智能体在使用自我修正后可能花费 $0.06-0.08 — 规模化后,这就是盈利与亏损之间的差距。
# 智能自我修正:在实际风险信号触发后才启用
async def respond_with_adaptive_correction(
user_request: str,
initial_response: str,
confidence_score: float,
task_complexity: str
) -> str:
# 低置信度或高复杂度任务需要修正
needs_correction = (
confidence_score < 0.7 or
task_complexity in ("multi-step", "financial", "medical")
)
if needs_correction:
critique = await run_critique_cycle(initial_response)
if critique.has_issues:
return await regenerate(critique)
return initial_response
关键洞察:不要对所有内容都进行自我修正。只对重要的事情进行修正。将简单查询路由到快速通道,把修正循环留给高风险的交互。
基于上述失败模式,以下是已在真实流量下证明具有韧性的架构模式。
每个 AI 智能体动作都经过一个强制执行硬限制的调控器:
class AgentGovernor:
"""无论 LLM 输出如何,都强制执行对 AI 智能体行为的硬约束。"""
def __init__(self, config: AgentConfig):
self.max_steps = config.max_steps
self.max_tokens_per_step = config.max_tokens_per_step
self.allowed_tools = set(config.allowed_tools)
self.rate_limit = config.rate_limit
async def run_with_guardrails(
self, agent: Agent, user_input: str
) -> Response:
steps_executed = 0
while steps_executed < self.max_steps:
# 执行下一步,但强制执行预算
step_result = await agent.execute_step(
user_input,
budget={
"tokens": self.max_tokens_per_step,
"tools": self.allowed_tools
}
)
if step_result.is_final:
return step_result.response
steps_executed += 1
# 强制步数限制和护栏
self._enforce_hard_limits(step_result)
return self._create_max_steps_response()
MAX_TURNS_PER_REQUEST = 10
MAX_TOOL_CALLS_PER_TURN = 3
MAX_TOKENS_PER_RESPONSE = 500
ALLOWED_TOOLS = {"search_knowledge_base", "lookup_user", "create_ticket"}
BLOCKED_PATTERNS = re.compile(
r"(send\s+email|make\s+payment|transfer|delete\s+account)"
)
def __init__(self, config: GovernanceConfig):
self.turn_counter = Counter()
self.config = config
async def authorize_turn(self, turn: AgentTurn) -> Authorization:
self.turn_counter.increment()
checks = [
self._check_turn_budget(),
self._check_tool_allowlist(turn),
self._check_output_safety(turn),
self._check_rate_limits(turn),
]
violations = [c for c in checks if not c.passed]
return Authorization(
allowed=len(violations) == 0,
violations=violations
)
def _check_tool_allowlist(self, turn: AgentTurn) -> CheckResult:
if turn.tool_name not in self.ALLOWED_TOOLS:
return CheckResult(
passed=False,
reason=f"Tool '{turn.tool_name}' not in allowlist"
)
return CheckResult(passed=True)
治理器是确定性代码,而非 LLM 决策。这意味着它无法被提示绕过、无法被 hallucinate 过去、也无法被对抗性输入迷惑。
你无法调试你看不到的东西。每个智能体交互都应该产生结构化的、可查询的追踪记录:
class AgentTraceObserver:
"""Captures every decision point in an agent's execution."""
def __init__(self, sink: TraceSink):
self.sink = sink
async def on_tool_call(self, event: ToolCallEvent):
await self.sink.write({
"type": "tool_call",
"timestamp": event.timestamp,
"agent_id": event.agent_id,
"tool": event.tool_name,
"arguments": event.arguments,
"result": event.result,
"latency_ms": event.latency_ms,
"token_cost": event.token_cost,
"llm_model": event.model,
"trace_id": event.trace_id
})
async def on_decision(self, event: DecisionEvent):
await self.sink.write({
"type": "llm_decision",
"timestamp": event.timestamp,
"input_tokens": event.input_tokens,
"output_tokens": event.output_tokens,
"confidence": event.confidence,
"reasoning": event.chain_of_thought,
"trace_id": event.trace_id
})
async def on_anomaly(self, event: AnomalyEvent):
await self.sink.write({
"type": "anomaly",
"timestamp": event.timestamp,
"category": event.category, # "loop_detected", "cost_spike", etc.
"severity": event.severity,
"details": event.details,
"trace_id": event.trace_id,
"action_taken": event.action_taken # "terminated", "escalated"
})
有了这套基础设施,你可以在几秒内回答生产环境问题:
"显示过去一小时内所有进入重试循环的智能体"
"每次成功解决事件平均消耗多少 token?"
"哪些工具调用最频繁地引发用户修正?"
当智能体的错误率超过阈值时,断路器会停止向其发送流量,并回退到更安全的路径:
class AgentCircuitBreaker:
"""Prevents a degraded agent from harming user experience at scale."""
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
def __init__(
self,
failure_threshold: int = 10,
window_seconds: int = 60,
half_open_max_calls: int = 3
):
self.failure_threshold = failure_threshold
self.window = window_seconds
self.state = self.CLOSED
self.failure_count = 0
self.last_failure_time = None
self.half_open_calls = 0
async def check(self, agent_id: str) -> CircuitState:
if self.state == self.CLOSED:
return CircuitState(allowed=True, mode="normal")
if self.state == self.OPEN:
if self._should_attempt_recovery():
self.state = self.HALF_OPEN
self.half_open_calls = 0
return CircuitState(allowed=True, mode="half_open")
return CircuitState(allowed=False, mode="fallback")
# HALF_OPEN
if self.half_open_calls >= self.half_open_max_calls:
return CircuitState(allowed=False, mode="fallback")
return CircuitState(allowed=True, mode="half_open")
def record_success(self):
if self.state == self.HALF_OPEN:
self.half_open_calls += 1
if self.half_open_calls >= self.half_open_max_calls:
self._close()
def record_failure(self):
self.failure_count += 1
self.last_failure_time = time.utcnow()
if self.failure_count >= self.failure_threshold:
self._open()
def _close(self):
self.state = self.CLOSED
self.failure_count = 0
def _open(self):
self.state = self.OPEN
def _should_attempt_recovery(self) -> bool:
return time.utcnow() - self.last_failure_time > self.window
当电路断开时,系统会路由到备选路径:一个更简单的智能体、直接 API 调用或人工队列。用户永远不会看到失败——他们看到的是另一条正常工作的路径。
传统的单元测试不适用于概率系统。你需要一套不同的测试分层策略:
测试工具、治理器和断路器是否在 LLM 无关的情况下正确运行:
async def test_governor_blocks_disallowed_tool():
governor = AgentGovernor(Config())
turn = AgentTurn(tool_name="delete_database", arguments={})
auth = await governor.authorize_turn(turn)
assert auth.allowed == False
assert any(
"not in allowlist" in v.reason
for v in auth.violations
)
async def test_circuit_breaker_opens_after_threshold():
cb = AgentCircuitBreaker(failure_threshold=3)
for _ in range(3):
cb.record_failure()
state = await cb.check("agent_1")
assert state.allowed == False
assert state.mode == "fallback"
这些测试不依赖任何 LLM 调用,所以可以在 CI 中毫秒级运行,每次运行都产生确定性的结果。
针对 LLM 驱动行为,使用 golden test 套件——给相同的输入,断言输出一致性(或明确记录允许的偏差范围):
async def test_agent_summarizes_customer_complaint_golden():
input = {
"task": "Summarize this customer complaint",
"complaint": "I've been waiting 3 weeks for my order #12345 to arrive...",
"temperature": 0.3
}
result = await agent.run(input)
golden = {
"summary": "Customer awaiting order #12345 for 3 weeks, requests resolution",
"entities": ["#12345"],
"sentiment": "frustrated"
}
assert result["summary"] == golden["summary"]
assert set(result["entities"]) == set(golden["entities"])
assert result["sentiment"] == golden["sentiment"]
在生产环境中,你无法控制 LLM 的输出分布。Golden test 的意义在于捕捉回归——当模型的行为发生漂移时,你会立即知道,而不是在用户投诉时才知道。
在实际场景中,LLM 输出存在方差,所以不要只断言最终结果。追踪中间步骤:
async def test_agent_repair_flow_reaches_resolution():
repair_flow = AgentRepairFlow()
# Simulate a multi-turn conversation
turns = [
{"user": "My order is late", "expected_tool": "lookup_order"},
{"user": "It's been 3 weeks", "expected_tool": "check_shipping"},
{"user": "Can you expedite it?", "expected_tool": "create_expedite_request"},
]
for turn_data in turns:
result = await repair_flow.execute_turn(turn_data["user"])
# Assert correct tool was selected at each step
assert result.selected_tool == turn_data["expected_tool"], \
f"At turn {turn_data['user']}: expected {turn_data['expected_tool']}, got {result.selected_tool}"
# Assert final resolution
final = await repair_flow.get_final_response()
assert final.status == "resolved"
assert "refund" not in final.action_taken.lower()
测试护栏本身,而非护栏与 LLM 的交互。正确的行为不依赖于模型是否心情好:
async def test_guardrails_block_pii_in_output():
guardrails = OutputGuardrails()
response_with_email = "Sure, I can help - please email john.doe@example.com"
result = await guardrails.check(response_with_email)
assert result.blocked == True
assert result.reason == "pii_detected"
assert result.redacted_response is not None
assert "john.doe@example.com" not in result.redacted_response
async def test_guardrails_permits_clean_output():
guardrails = OutputGuardrails()
clean_response = "Your order #12345 is scheduled to arrive tomorrow."
result = await guardrails.check(clean_response)
assert result.blocked == False
如果你发现自己在测试中需要模拟 LLM,那说明你测错了东西。工具应该有自己的测试。治理器应该有自己的测试。追踪器应该有自己的测试。LLM 只是把它们粘合在一起的胶水——不是被测试的对象。
构建可靠的 AI 智能体不是关于找到完美的提示。问题在更深层:
上下文工程比提示工程更重要。上下文工程是关于精心设计输入的结构、去除噪声、保持工具定义的精确——这是一套可以系统化优化的技能,而不是靠运气。
护栏必须在 LLM 之外。如果你的安全策略依赖于模型"不要做 X",那它迟早会做 X。基于规则的治理器、工具白名单、结构化输出验证——这些是你的护栏。
生产环境可见性决定成败。你无法改进你看不见的东西。每个决策点都需要结构化追踪、异常检测和可查询的历史记录。
测试分层。确定性组件用确定性测试。LLM 驱动组件用 golden test 和场景 test。你的断路器永远不会误报——但你的模型在某天可能会。
自我修正提示只是减速带。它们有效,但不够。真正的鲁棒性来自架构——多个智能体、投票、共识机制——而不是一个聪明的提示。
AI 智能体最常见的失败不是模型不够强大。失败发生在我们停止追问"这里会发生什么?"的时候。
在你部署的每个智能体进入生产环境之前,先问自己这三个问题:
如果你没有这些问题的答案,你的智能体还没有准备好迎接生产环境。
准备好迎接审视。