Agent 人机协作回路：审阅-修正-重试的工程化闭环

2026/7/8 15:06:40

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: 更新 PR（force-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_factory=list) created_at: datetime = field(default_factory=datetime.now) updated_at: datetime = field(default_factory=datetime.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_id=task_id, description=description) self._tasks[task_id] = task # Agent 执行任务 try: output = await asyncio.wait_for( self._agent.execute(description), timeout=300.0, # 5 分钟超时 ) except asyncio.TimeoutError: task.status = TaskStatus.FAILED return task task.agent_output = output branch_name = f"agent/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(hours=2) 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(f"Task {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, f"Updated 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 f"task-{ts}-{desc_hash}"