跳到主要内容

Agent-Loop执行循环

1. 定义与边界

Agent Loop 是 Agent 的运行时核心:系统不断构造上下文,让模型决定下一步,执行工具或输出,接收观察结果,更新状态,直到完成、失败、超时或升级。

它不是简单的 while 循环。生产级 Agent Loop 需要同时处理状态、权限、工具错误、最大步数、成本、人工审批、trace 和恢复。

2. 为什么重要

Agent 的可靠性主要由执行循环决定:

  • 没有最大步数会导致死循环和成本失控。
  • 没有状态持久化会导致失败不可恢复。
  • 没有工具结果校验会导致错误扩散。
  • 没有 trace 会导致无法复盘。
  • 没有人类在环会导致高风险动作失控。

3. 核心机制

Agent Loop 的最小状态:

{
"run_id": "run_123",
"task": "分析订单退款资格",
"step": 3,
"max_steps": 8,
"plan": ["读取订单", "读取政策", "判断资格"],
"observations": [
{"tool": "get_order", "status": "ok", "summary": "订单已签收"}
],
"risk_state": {
"allowed_tools": ["get_order", "read_policy", "create_ticket"],
"pending_approval": null
}
}

状态字段分层

字段是否必须作用常见错误
run_id关联 trace、审批、日志和回放每次重试生成新 id,导致链路断裂
goal约束模型推进方向每轮只传用户原话,目标漂移
step / max_steps控制循环成本和死循环只在 prompt 里说“不要太多步”
observations记录工具证据和错误只保存自然语言摘要,丢失结构化错误码
pending_action写操作需要支持审批、中断和恢复审批回来后找不到原参数
risk_state生产必需记录权限、审批和策略结果工具调用前不做系统级校验
checkpoint长任务必需支持失败恢复进程崩溃后只能从头重跑
budget建议控制 token、工具成本和时间单次任务成本不可解释

4. 架构模式

直接循环

最简单,适合学习和小工具。

状态机循环

每个阶段有明确状态和允许转移,适合业务流程。

图结构循环

节点代表模型、工具、审批、校验;边决定转移。LangGraph 这类框架适合持久化、中断和恢复。

队列驱动循环

长任务拆成 job,工具调用和人工审批通过队列异步处理,适合企业生产。

5. 工程实现

class AgentLoop:
def run(self, task, user):
run = self.store.start_run(task, user)

while run.step < run.policy.max_steps:
context = self.context_builder.build(run)
decision = self.model.decide(context=context, tools=self.tool_schemas(run))
self.trace.model_decision(run.id, decision)

if decision.is_final:
output = self.output_guardrail.validate(decision.output)
return self.store.finish(run.id, output)

tool = self.tool_registry.resolve(decision.tool_name)
args = self.arg_validator.validate(tool, decision.arguments)
risk = self.policy.evaluate(user=user, tool=tool, args=args)

if risk.requires_approval:
approval = self.human_gate.request(run.id, tool.name, args, risk)
self.trace.approval(run.id, approval)
if not approval.approved:
return self.store.escalate(run.id, approval.reason)

try:
result = tool.invoke(args)
except ToolError as exc:
result = {"ok": False, "error": exc.code, "retriable": exc.retriable}

self.trace.tool_result(run.id, tool.name, args, result)
run = self.state_updater.apply(run, decision, result)

if self.stopper.should_stop(run):
return self.store.finish_or_fail(run)

return self.store.fail(run.id, "max_steps_exceeded")

6. 生产实践

问题实践
循环失控max_steps、timeout、预算、重复动作检测
工具失败错误码、可重试标记、指数退避、降级路径
状态丢失checkpoint、幂等工具、恢复点
输出不可控结构化输出、schema 校验、后处理校验
审批阻塞interrupt/resume、审批超时、升级队列
调试困难trace、span、工具参数脱敏记录

终止条件设计

Agent Loop 至少要有四类停止条件:

条件示例处理
成功完成创建工单、测试通过、报告生成校验输出并完成 run
明确失败必要数据不存在、权限不足返回失败原因或升级
风险升级高风险动作、政策冲突、低置信度HITL 或人工接管
资源耗尽max steps、timeout、预算超限停止并保存 trace

不应把停止条件完全交给模型自然语言判断。模型可以建议“我认为完成了”,但系统要用状态、工具结果和输出 schema 做最后判断。

7. 常见反模式

反模式触发信号后果修正
每轮塞完整历史prompt 越来越长,关键约束被淹没成本高、决策漂移用状态摘要、最近观察、按需检索
工具失败只给异常文本模型把 timeout 当业务失败错误恢复不可控结构化错误码、retriable 标记
不区分错误类型所有失败都重试或都终止重复提交或过早失败错误分类:可重试、需澄清、需升级
无幂等键重试后重复建单/扣费外部副作用重复写工具必须支持 idempotency key
最终输出不校验模型输出敏感信息或未完成状态合规和业务事故输出 schema、安全检查、业务状态校验
循环里混杂所有逻辑prompt、工具、审批、日志写在一起难测试、难回放拆分 context_builder、policy、tool_runtime、state_updater

8. 评测方法

Agent Loop 需要轨迹级评测:

指标检测方式
平均步数trace 统计
最大步数触发率超过阈值的 run 比例
重复工具调用率相同工具相同参数重复出现
工具错误恢复率retriable 错误后是否成功
终止正确率是否在完成/失败/升级时正确停止
审批覆盖率高风险工具是否都经过 HITL

Trace 事件建议统一成可回放结构:

{"type": "model_decision", "step": 3, "selected_tool": "read_policy", "args_hash": "h1"}
{"type": "tool_result", "step": 3, "tool": "read_policy", "status": "ok", "summary": "需物流争议"}
{"type": "state_update", "step": 3, "added_facts": ["policy_requires_dispute"]}
{"type": "stop_check", "step": 3, "matched": false, "reason": "need_create_ticket"}

9. 安全与治理

Agent Loop 应在每轮执行前后做安全控制:

  • 决策前:过滤可用工具,加入任务约束和权限上下文。
  • 参数前:校验 schema、资源归属、敏感字段。
  • 调用前:检查风险等级、审批状态、速率限制。
  • 调用后:脱敏工具结果,标注外部内容不可信。
  • 输出前:校验敏感信息、政策合规和业务状态。

10. 权威资料