企业官网建设流程全解析

【Agent工程】（7）—— 单 Agent 任务闭环

摘要：前几篇分别解决了任务卡片、验收、工具门禁与高危写确认。若运行时仍是一问一答、模型说完就算结束，这些能力无法串成可判定流程。本篇描述单 Agent 任务闭环：校验卡片、注入上下文、推理执行、收集终态、跑验收、结案或降级。贯穿示例继续用工单助手。适合读完 【Agent工程】（6）—— 危险写操作与二次确认、准备把分散模块收成一条运行链路的工程师。读完可以独立完成：实现一版最小编排器，驱动一张卡片从启动到验收。

1. 单轮对话不等于任务闭环

常见 Agent 服务形态是：用户发一句 → 模型多轮 tool call → 返回一段总结。对话结束，但系统不知道任务是否可验收地完成。

本专栏前几篇已经具备闭环所需的零件：

• 【Agent工程】（2）—— 模糊目标到任务卡片：结构化输入
• 【Agent工程】（3）—— 验收命令与完成定义：完成判定
• 【Agent工程】（4）—— 工具注册表与读写分离：工具目录与门禁
• 【Agent工程】（5）—— 工具作用域与对象级授权：对象级授权
• 【Agent工程】（6）—— 危险写操作与二次确认：高危写 pending

本篇回答：这些零件在运行时按什么顺序衔接。编排单元从单 Agent 开始；多 Agent 委托放在后续篇目。

2. 闭环的六个阶段

单 Agent 处理一张任务卡片，建议固定六阶段，顺序不要随模型「灵感」调整：

阶段 3内部仍走第 4～6 篇的门禁：注册表 → scope → 高危 pending。阶段 5区分自动项与人工项，口径与第 3 篇一致。

编排器可以是独立进程，也可以嵌在主服务里；关键是六阶段的顺序与输入输出固定，而不是某一家框架的类名。团队先用脚本把六阶段跑通，再考虑接入 LangGraph、Temporal 等外部编排，避免反过来被框架绑架阶段语义。

2.1 启动前校验不可跳过

卡片校验至少检查：

• 目标、允许工具、验收非空
• write_ticket_ids或等价 scope 存在（写任务）
• 预算字段有上限（步数或写调用次数）

校验失败应直接返回，不启动模型。否则会出现「跑到一半才发现验收没写」的半成品任务。

2.2 阶段边界不要混用

六阶段里，推理执行与跑验收的职责必须分开。常见错误是把验收逻辑写进系统提示，让模型自己判断「做完了没有」。模型在阶段 3 只负责提议工具；是否 done 由阶段 5 的验收运行器根据final_state决定。混用后，轨迹里缺少结构化终态，评测集也无法稳定回归。

3. 运行时上下文与状态机

每张卡片实例化一份runtime context（运行时上下文），生命周期与卡片绑定，不跨卡片复用 scope 或轨迹。

状态机建议至少包含：

从await_confirm超时应转入degraded，而不是无限等待。降级结果也要写入终态，供验收解释「为何无 notified」。

3.1 预算在循环内递减

预算来自卡片，典型字段：

{"budget":{"max_steps":12,"max_write_calls":5,"max_pending":2}}

每发生一次模型步进、写工具执行或 pending 创建，对应计数减一。归零时编排器应停止继续调模型，进入「收集终态 → 验收」，而不是悄悄放宽上限。

3.2 收集终态从哪来

阶段 4 的final_state应优先来自只读查询，而不是模型口述：

查询键（如 ticket_id）来自卡片 scope。终态收集失败时，不应标记为 done，而应进入 failed 或 degraded，并在报告里写明「终态不可达」。

4. 贯穿示例：工单升级闭环

目标：将T-1024升为紧急并通知值班。卡片含 scope、允许工具、验收与auto_confirm: false。

闭环里模型不负责宣告完成。编排器在阶段 5 根据验收报告设置status，对话摘要仅作展示。

4.1 失败与降级路径

重试策略应有限次、可配置，且不能绕过门禁。常见做法是：同一张卡片最多重跑 1 次推理，仍失败则输出验收报告给值班。

4.2 对话层与编排层的分工

用户看到的聊天摘要，应由编排器在结案后生成，输入是验收报告与final_state，而不是模型在阶段 3 的即兴总结。这样即使模型说「已完成」，只要验收未通过，对外状态仍是 failed 或 await_manual，避免感觉完成与断言完成再次分叉（见第 3 篇）。

5. 可运行示意：最小编排器

下面代码把校验、模拟执行、验收串成一条链。工具实现仍用占位函数，重点看阶段顺序与状态变迁。

5.1 闭环主流程

from__future__importannotationsfromtypingimportAnydefvalidate_card(card:dict[str,Any])->dict[str,Any]:missing=[]ifnotcard.get("goal"):missing.append("goal")ifnotcard.get("allowed_tools"):missing.append("allowed_tools")ifnotcard.get("acceptance"):missing.append("acceptance")ifnot(card.get("scope")or{}).get("write_ticket_ids"):missing.append("scope.write_ticket_ids")budget=card.get("budget")or{}ifbudget.get("max_steps")isNone:missing.append("budget.max_steps")ifmissing:return{"ok":False,"missing":missing}return{"ok":True}defrun_acceptance(card:dict[str,Any],final_state:dict[str,Any])->dict[str,Any]:acc=card["acceptance"]failed=[]iffinal_state.get("priority")!=acc.get("priority_equals"):failed.append("priority_ok")ifacc.get("notify_required")andnot(final_state.get("notified")orfinal_state.get("notify_degraded")):failed.append("notify_ok")used=set(final_state.get("used_tools")or[])allowed=set(card["allowed_tools"])ifnotused.issubset(allowed):failed.append("tools_ok")manual=[]ifacc.get("manual_tone_review"):manual.append("tone_review")auto_passed=len(failed)==0passed=auto_passedandnotmanualreturn{"auto_passed":auto_passed,"passed":passed,"failed":failed,"manual":manual,}defrun_task_loop(card:dict[str,Any],executor_result:dict[str,Any])->dict[str,Any]:check=validate_card(card)ifnotcheck["ok"]:return{"status":"rejected","reason":"invalid_card",**check}ctx:dict[str,Any]={"card_id":card.get("card_id","local"),"status":"running","trace":executor_result.get("trace",{}),}ifexecutor_result.get("await_confirm"):ctx["status"]="await_confirm"returnctx ctx["status"]="accepting"final_state=executor_result["final_state"]report=run_acceptance(card,final_state)ctx["acceptance"]=reportifreport["passed"]:ctx["status"]="done"elifreport["auto_passed"]andreport["manual"]:ctx["status"]="await_manual"eliffinal_state.get("notify_degraded"):ctx["status"]="degraded"else:ctx["status"]="failed"ctx["final_state"]=final_statereturnctxif__name__=="__main__":card={"card_id":"card-100","goal":"将 T-1024 升级为紧急并通知值班","allowed_tools":["get_ticket","update_ticket_priority","notify_oncall",],"scope":{"write_ticket_ids":["T-1024"]},"acceptance":{"priority_equals":"urgent","notify_required":True,"manual_tone_review":False,},"budget":{"max_steps":12,"max_write_calls":5,"max_pending":2},}executor_result={"await_confirm":False,"trace":{"used_tools":["update_ticket_priority","notify_oncall"]},"final_state":{"ticket_id":"T-1024","priority":"urgent","notified":True,"used_tools":["update_ticket_priority","notify_oncall"],},}print(run_task_loop(card,executor_result))

5.2 待确认与降级路径

if__name__=="__main__":card={"card_id":"card-101","goal":"将 T-1024 升级为紧急并通知值班","allowed_tools":["update_ticket_priority","notify_oncall"],"scope":{"write_ticket_ids":["T-1024"]},"acceptance":{"priority_equals":"urgent","notify_required":True,},"budget":{"max_steps":12},}blocked=run_task_loop(card,{"await_confirm":True,"trace":{"pending_ids":["p-abc"]},"final_state":{},},)assertblocked["status"]=="await_confirm"degraded=run_task_loop(card,{"await_confirm":False,"trace":{"used_tools":["update_ticket_priority"]},"final_state":{"priority":"urgent","notified":False,"notify_degraded":True,"used_tools":["update_ticket_priority"],},},)assertdegraded["status"]=="degraded"print("待确认暂停与 notify 降级路径正常")

真实系统里，阶段 3 的 executor 应调用模型与工具网关；本篇编排器只规定何时停、何时验、如何结案。

联调时建议用同一张工单卡片跑三条路径：全成功 done、notify pending 阻塞、notify 降级 degraded。三条路径的 status 与验收报告应可重复生成，便于接入 CI 回归。

6. 与前几篇的衔接关系

闭环本身不替代任何一层。少一层就会在对应阶段失焦：无 scope 则越权在轨迹里爆炸；无 pending 则 notify 误发；无验收则done只是主观判断。

阶段 3 与阶段 4 之间建议留一个显式「执行结束」信号：可以是模型输出结构化finish、也可以是编排器发现连续两轮无 tool call。没有该信号时，不要提前拉终态，否则容易在 notify 仍 pending 时误跑验收。

6.1 轨迹字段约定

建议在trace中统一记录：

• used_tools：注册表认可的 tool_id 列表
• rejected_calls：门禁拒绝明细（含 reason）
• pending_ids及最终 approved / rejected 状态
• step_count：实际消耗步数

验收与评测集应直接消费这些字段，而不是解析聊天文本。

6.2 结案类型与对外展示

UI 与 webhook 应订阅 status，而不是订阅模型最后一句话。

7. 四个常见误区

还有一种隐蔽做法：把验收写在提示词里让模型「自检」。模型自检不能替代第 3 篇验收运行器；闭环的阶段 5 必须读final_state与trace，不能读模型总结。

7.1 编排器与模型 API 的边界

编排器负责：阶段顺序、预算、状态机、调用门禁与 pending 回调。模型 API 负责：在允许工具范围内提议下一步。不要把门禁逻辑写进 prompt 指望模型遵守；也不要让模型直接修改status字段。两者接口清晰，后续换模型或换工具实现时，编排层可以保持稳定。

8. 适用边界

本篇方法适合：

• 单 Agent、单卡片、串行执行
• 工具面已具备门禁与 pending
• 需要可判定 done / failed / degraded

本篇不覆盖：

• 多 Agent 分工与委托——下一篇编排单元展开
• 并行子任务、队列调度——后续运行时篇目
• 长时运行（跨小时 human-in-the-loop）——需持久化 context

若任务极短、只有只读查询，闭环可简化：跳过 pending 相关状态，但仍建议保留校验与验收骨架，便于日后加写工具时不改架构。

8.1 从对话式原型迁移

存量对话式 Agent 迁移可以分三步：

1. 补卡片与验收：先把现有 prompt 里的目标、工具、完成条件抽成字段
2. 外挂编排器：模型 API 不变，工具调用改经门禁与 trace
3. 改结束条件：由编排器根据验收设 status，UI 不再以「模型停止」为准

每一步都可以单独上线，但只有当第三步完成，系统才称得上任务闭环。

8.2 最小可观测性

闭环上线时至少暴露以下指标，便于值班与回归：

指标应从 runtime context 汇总，不要从聊天日志反推。

9. 术语速查

10. 小结与下一篇

单 Agent 任务闭环把前几篇能力串成固定顺序：

1. 校验卡片再启动，拒绝半成品输入
2. runtime context绑定卡片，轨迹与 scope 不跨任务漂移
3. 推理执行走工具门禁与 pending，预算在环内递减
4. 收集终态后跑验收，用报告驱动 done / degraded

下一篇继续编排单元：多个子 Agent 如何分工、委托范围与成本如何控制，避免「单 Agent 闭环」复制多份却无人协调。

系列导航：

• 上一篇：【Agent工程】（6）—— 危险写操作与二次确认
• 下一篇：【Agent工程】（8）—— 子 Agent 分工与委托边界（撰写中）