Agent 人机协作回路审阅-修正-重试的工程化闭环一、Agent 的输出不能直接上线让 Agent 生成一段代码就自动提交到生产环境这是灾难的配方。Agent 的幻觉率大约在 3-15% 之间取决于任务复杂度和模型这意味着每 10 次输出就有 1 次可能有问题。解决思路不是让 Agent 做得更完美——这在可预见的未来不现实。而是构建一个人机协作闭环Agent 生成 → 人类审阅 → Agent 修正 → 重新提交循环往复直到质量过关。这个闭环的设计难点在于如何最小化人类的参与时间如果每轮审阅都要 5 分钟那还不如人直接写。目标是把人类从执行者变成审批者——只看差异、只点通过/驳回一轮审阅不超过 30 秒。sequenceDiagram participant H as 人类审阅者 participant C as 协作控制器 participant A as Agent participant G as Git 仓库 H-C: 提交任务指令 C-A: 分配任务 上下文 A-A: 规划 执行工具链 A-C: 提交产出 (PR/MR) C-H: 通知审阅仅展示 diff alt 通过 H-C: 批准 C-G: Merge 标记完成 else 驳回附修改意见 H-C: 驳回 修正指令 C-A: 传递修正指令 diff 上下文 A-A: 针对性修正 A-C: 更新 PRforce-push C-H: 通知重新审阅 end Note over A,C: 最多重试 3 轮超过则转人工处理二、闭环的四个核心组件1. 任务分配与上下文注入不能直接把用户的自然语言丢给 Agent。需要做任务拆解Task Decomposition和上下文注入Context Injection。拆解的目的是把大任务分解为 Agent 能稳定执行的小步骤注入的目的是让 Agent 知道当前项目的技术栈、代码规范、已有代码库。2. 差异审查界面人类的审阅成本集中在理解 Agent 做了什么。一个 good diff 应该只展示变更部分不展示上下文无关代码高亮可能有问题的地方由静态分析工具标记附带 Agent 的变更理由说明3. 修正闭环驳回时修正指令必须精确 可执行。不能写这里写得不对要写第三个参数应该传入 timeout 而不是 retries参考 utils/timeout.go:42。4. 重试上限设置 3 轮重试上限。超过 3 轮修正仍未通过自动转入人工处理队列。防止 Agent 和人类在低质量任务上互相消耗。三、协作控制器实现import asyncio import hashlib import json from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import Enum from typing import Optional class ReviewStatus(Enum): PENDING pending APPROVED approved CHANGES_REQUESTED changes_requested class TaskStatus(Enum): IN_PROGRESS in_progress AWAITING_REVIEW awaiting_review REVISION_NEEDED revision_needed COMPLETED completed FAILED failed # 超过重试上限 dataclass class ReviewFeedback: 审阅反馈。 设计决策 - 包含 diff 上下文便于 Agent 定位 - 修正指令必须是可操作的 - 严重性标记用于优先级排序 file_path: str line_range: tuple[int, int] instruction: str # 可执行的修正指令 severity: str normal # critical / normal / suggestion diff_snippet: str # 原始 diff 片段供 Agent 上下文感知 dataclass class CollaborationTask: task_id: str description: str status: TaskStatus TaskStatus.IN_PROGRESS retry_count: int 0 max_retries: int 3 agent_output: Optional[str] None review_feedback: list[ReviewFeedback] field(default_factorylist) created_at: datetime field(default_factorydatetime.now) updated_at: datetime field(default_factorydatetime.now) class CollaborationController: 人机协作闭环控制器。 架构决策 - 状态机驱动IN_PROGRESS → AWAITING_REVIEW → (APPROVED|REVISION_NEEDED) - 超时保护审阅超过 2 小时自动提醒超过 24 小时自动转人工 - 重试上限3 轮第四轮自动 FAILED def __init__(self, agent, git_repo, notifier): self._agent agent self._git git_repo self._notifier notifier self._tasks: dict[str, CollaborationTask] {} # 审阅超时锁——防止任务长时间无人审阅 self._review_deadlines: dict[str, datetime] {} async def submit_task(self, description: str) - CollaborationTask: 提交任务Agent 生成初稿 → 创建 PR → 通知审阅 task_id self._generate_task_id(description) task CollaborationTask(task_idtask_id, descriptiondescription) self._tasks[task_id] task # Agent 执行任务 try: output await asyncio.wait_for( self._agent.execute(description), timeout300.0, # 5 分钟超时 ) except asyncio.TimeoutError: task.status TaskStatus.FAILED return task task.agent_output output branch_name fagent/task-{task_id[:8]} # 创建分支 提交 PR await self._git.create_branch(branch_name) await self._git.commit_and_push(branch_name, output) pr_url await self._git.create_pr(branch_name, description) task.status TaskStatus.AWAITING_REVIEW task.updated_at datetime.now() self._review_deadlines[task_id] datetime.now() timedelta(hours2) await self._notifier.send_review_request(task_id, pr_url) return task async def submit_review( self, task_id: str, status: ReviewStatus, feedback: Optional[list[ReviewFeedback]] None ) - CollaborationTask: 提交审阅意见 task self._tasks.get(task_id) if not task: raise ValueError(fTask {task_id} not found) if status ReviewStatus.APPROVED: task.status TaskStatus.COMPLETED await self._git.merge_pr(task_id) return task if status ReviewStatus.CHANGES_REQUESTED: task.retry_count 1 if task.retry_count task.max_retries: task.status TaskStatus.FAILED await self._notifier.send_failure_alert(task_id, exceeded max retries) return task task.review_feedback feedback or [] task.status TaskStatus.REVISION_NEEDED task.updated_at datetime.now() # 异步触发 Agent 修正不阻塞审阅者 asyncio.create_task(self._agent_revise(task)) return task return task async def _agent_revise(self, task: CollaborationTask) - None: Agent 根据审阅反馈修正代码。 关键让 Agent 在一轮内处理所有反馈而非逐条逐轮 ——减少审阅轮次是关键 revision_prompt self._build_revision_prompt(task) try: revised_output await self._agent.execute(revision_prompt) except Exception as e: task.status TaskStatus.FAILED return task.agent_output revised_output task.status TaskStatus.AWAITING_REVIEW task.updated_at datetime.now() await self._notifier.send_review_request( task.task_id, fUpdated PR for task {task.task_id[:8]} ) def _build_revision_prompt(self, task: CollaborationTask) - str: 构建修正提示。 把多条反馈合并成一个 prompt减少 Agent 推理轮次。 按严重性排序——critical 的改对了再说 suggestion。 critical_feedbacks [f for f in task.review_feedback if f.severity critical] normal_feedbacks [f for f in task.review_feedback if f.severity ! critical] prompt_parts [请根据以下审阅意见修正代码\n] for fb in critical_feedbacks normal_feedbacks: prompt_parts.append( f- {fb.file_path}:{fb.line_range[0]}-{fb.line_range[1]}\n f 修正方向: {fb.instruction} ) return \n.join(prompt_parts) def _generate_task_id(self, description: str) - str: ts datetime.now().strftime(%Y%m%d%H%M%S) desc_hash hashlib.sha256(description.encode()).hexdigest()[:8] return ftask-{ts}-{desc_hash}四、边界与工程代价审阅疲劳的问题人类审阅者一天最多审 20-30 个 Agent PR。超过这个数注意力衰减审批质量下降。需要引入自动化前置检查lint、test、type-check先过滤掉明显有问题的产出减少人类审阅量。修正收敛性Agent 不一定能根据反馈正确修正。最坏的情况是每轮修正引入新的问题导致重试循环。这就是为什么需要 3 轮上限——保护人类时间。不适合的场景需要深度架构判断的任务Agent 无法理解系统级的耦合关系安全关键代码认证、加密、支付一次性脚本/临时任务人机协作回路的设置成本 直接手写五、总结人机协作回路的本质是把人类从写代码角色升级为审代码角色。这要求闭环设计上做到Agent 的产出可审查干净 diff、人类的反馈可执行精确指令、修正的循环可收敛重试上限。三个条件缺了任何一个闭环都会变成负担而非增效。