引言 2026年，Agent 不再是"全自动"或"全人工"的二元选择——而是光谱上的渐进式协作。从"全自动执行"到"每一步都确认"，中间有无数种可能。好的 Agent 设计让人类知道何时需要介入，何时可以放手。本文探讨人机协作的设计模式。
一、人机协作光谱 全自主 Agent 全人工控制 ◀─────────┬─────────┬─────────┬─────────▶ │ │ │ 全自动执行 需确认 需监督 需审批 (Auto) (Confirm) (Supervise) (Approve) 设计原则：让 Agent 做它能做好的，让人做只有人能做的 模式 自动化程度 人类参与点 适用场景 全自动 100% 事后审计 低风险、高频、标准化任务 需确认 80% 关键决策点 中风险、需合规 需监督 50% 持续监控 高风险、创造性任务 需审批 20% 每步审批 高风险、不可撤销 二、全自动模式设计 2.1 适用条件 全自动 Agent 只有在满足以下条件时才安全：
class AutoModeSafetyCheck: """全自动模式安全检查""" SAFETY_RULES = [ # 1. 任务风险评级 ("task_risk_level", "low"), # 必须是低风险任务 ("output_reversibility", True), # 输出必须可撤销 ("cost_per_execution", 1.0), # 单次成本 < $1 ("data_access_level", "public"), # 仅访问公开数据 ("external_api_calls", "read_only"), # 外部调用只读 # 2. 质量保障 ("quality_score_threshold", 0.95), # 质量评分 > 95% ("error_rate_threshold", 0.01), # 错误率 < 1% ("test_coverage", 0.90), # 测试覆盖率 > 90% # 3. 监控 ("real_time_monitoring", True), # 实时监控 ("alert_on_anomaly", True), # 异常告警 ("automatic_rollback", True), # 自动回滚 ] def can_auto_mode( self, task: Task, agent: Agent ) -> SafetyAssessment: violations = [] # 检查任务风险 risk = self._assess_risk(task) if risk.level != "low": violations.append(f"Task risk level: {risk.level}") # 检查 Agent 历史表现 metrics = agent.get_performance_metrics(days=30) if metrics.error_rate > 0.01: violations.append(f"Error rate: {metrics.error_rate:.1%}") if metrics.avg_quality < 0.95: violations.append(f"Quality score: {metrics.avg_quality:.1%}") # 检查成本 estimated_cost = agent.estimate_cost(task) if estimated_cost > 1.0: violations.append(f"Estimated cost: ${estimated_cost:.2f}") return SafetyAssessment( can_auto=len(violations) == 0, violations=violations, recommended_mode="auto" if not violations else "confirm", confidence=1.0 - len(violations) * 0.2 ) 2.2 全自动 Agent 的护栏 class AutoModeGuardrails: """全自动模式的护栏""" def __init__(self): self.rules = self._load_guardrails() async def pre_check(self, task: Task) -> GuardrailResult: """执行前检查""" violations = [] # 1. 输入安全检查 if self._contains_sensitive_data(task.input): violations.append("Input contains sensitive data") # 2. 成本预估 estimated_cost = self._estimate_cost(task) if estimated_cost > self.rules.max_cost_per_run: violations.append(f"Estimated cost ${estimated_cost:.2f} > limit") # 3. 工具权限检查 required_tools = self._get_required_tools(task) unauthorized = [ t for t in required_tools if not self.rules.is_authorized(t, "auto") ] if unauthorized: violations.append(f"Unauthorized tools: {unauthorized}") return GuardrailResult( passed=len(violations) == 0, violations=violations ) async def post_check(self, output: str, actions: list) -> GuardrailResult: """执行后检查""" violations = [] # 1. 输出安全检查 safety = await self._check_output_safety(output) if not safety.safe: violations.append(f"Output safety: {safety.reason}") # 2. 行动审计 for action in actions: if action.type == "external_api" and action.method != "GET": violations.append(f"Non-read action executed: {action}") # 3. 成本检查 actual_cost = self._calculate_cost(actions) if actual_cost > self.rules.max_cost_per_run * 2: violations.append(f"Cost overrun: ${actual_cost:.2f}") return GuardrailResult( passed=len(violations) == 0, violations=violations, requires_escalation=len(violations) > 0 ) 三、需确认模式（Confirm） 3.1 确认点设计 class ConfirmationPoint: """确认点设计""" # 需要确认的情景 TRIGGERS = { "high_cost": lambda ctx: ctx.estimated_cost > 5.0, "external_api_write": lambda ctx: any( t.method != "GET" for t in ctx.tool_calls ), "irreversible_action": lambda ctx: any( t.tool in ["delete_file", "send_email", "publish_post"] for t in ctx.tool_calls ), "low_confidence": lambda ctx: ctx.confidence < 0.7, "new_tool": lambda ctx: any( t.tool not in ctx.agent.verified_tools for t in ctx.tool_calls ), } async def should_confirm(self, context: ExecutionContext) -> list[str]: """判断是否需要确认""" reasons = [] for trigger_name, trigger_fn in self.TRIGGERS.items(): if trigger_fn(context): reasons.append(trigger_name) return reasons def build_confirmation_ui( self, context: ExecutionContext, reasons: list[str] ) -> ConfirmationRequest: """构建确认 UI""" return ConfirmationRequest( title="Agent 需要确认", message=self._generate_message(reasons), plan=context.execution_plan, # Agent 的执行计划 estimated_cost=context.estimated_cost, estimated_time=context.estimated_time, risks=self._identify_risks(context), actions_preview=self._preview_actions(context.tool_calls), confirm_text="确认执行", cancel_text="取消", modify_text="修改计划", ) 3.2 确认 UI 实现 interface ConfirmationRequest { title: string; message: string; plan: ExecutionStep[]; estimated_cost: number; estimated_time: string; risks: string[]; actions_preview: ActionPreview[]; confirm_text: string; cancel_text: string; modify_text: string; } class AgentConfirmationDialog { render(request: ConfirmationRequest): JSX.Element { return ( <Card> <CardHeader> <Icon name="help-circle" /> <Title>{request.title}</Title> </CardHeader> <CardBody> {/* 原因说明 */} <Alert type="warning"> {request.message} </Alert> {/* 执行计划 */} <Section title="执行计划"> {request.plan.map((step, i) => ( <StepCard key={i} step={step} index={i} /> ))} </Section> {/* 风险评估 */} {request.risks.length > 0 && ( <Section title="⚠️ 风险提示"> <List items={request.risks} /> </Section> )} {/* 成本预估 */} <CostEstimate cost={request.estimated_cost} time={request.estimated_time} /> {/* 操作预览 */} <Section title="操作预览"> {request.actions_preview.map((action, i) => ( <ActionPreview key={i} action={action} /> ))} </Section> </CardBody> <CardFooter> <Button variant="outline" onClick={this.onCancel}> {request.cancel_text} </Button> <Button variant="outline" onClick={this.onModify}> {request.modify_text} </Button> <Button variant="primary" onClick={this.onConfirm}> {request.confirm_text} </Button> </CardFooter> </Card> ); } onConfirm = () => { this.props.onResponse({action: "confirm"}); }; onCancel = () => { this.props.onResponse({action: "cancel"}); }; onModify = () => { // 打开修改对话框 this.props.onResponse({ action: "modify", modifications: this.getModifications() }); }; } 四、需监督模式（Supervise） 4.1 实时监督界面 class SupervisedAgentUI: """需监督 Agent 的实时界面""" def render_live_view(self, session_id: str) -> str: """渲染实时监督界面""" # 获取 Agent 当前状态 state = self.agent.get_state(session_id) return f""" <div class="supervised-agent-ui"> <div class="agent-status"> <StatusBadge status="{state.status}" /> <span>当前步骤: {state.current_step}</span> <ProgressBar progress="{state.progress}" /> </div> <div class="reasoning-view"> <h4>🤔 Agent 正在思考</h4> <pre>{state.current_reasoning}</pre> </div> <div class="tool-execution-view"> <h4>🔧 工具执行</h4> {self._render_tool_execution(state.tool_history)} </div> <div class="controls"> <button onclick="pauseAgent()">⏸️ 暂停</button> <button onclick="resumeAgent()">▶️ 继续</button> <button onclick="stopAgent()">⏹️ 停止</button> <button onclick="provideFeedback()">💬 提供反馈</button> <button onclick="takeOver()">🤝 接管</button> </div> </div> """ def _render_tool_execution(self, history: list) -> str: html = "<ul class='tool-history'>" for item in history: status_icon = "✅" if item.success else "❌" html += f""" <li class='tool-item'> <span class='tool-name'>{item.tool_name}</span> <span class='tool-status'>{status_icon}</span> <pre class='tool-result'>{item.result_preview}</pre> </li> """ html += "</ul>" return html 4.2 监督模式的控制权转移 class ControlTransfer: """控制权转移管理""" async def request_control( self, from_entity: str, # "agent" or "human" to_entity: str, reason: str, context: dict ) -> ControlTransferResult: """请求控制权转移""" # 1. 检查是否可以转移 if not self._can_transfer(from_entity, to_entity): return ControlTransferResult( success=False, reason="Transfer not allowed in current state" ) # 2. 保存当前状态 snapshot = await self._take_snapshot(context["session_id"]) # 3. 转移控制权 self.current_controller = to_entity self.control_history.append({ "from": from_entity, "to": to_entity, "reason": reason, "timestamp": time.time(), "snapshot_id": snapshot.id }) # 4. 通知各方 await self._notify_control_change( from_entity, to_entity, reason ) return ControlTransferResult( success=True, snapshot=snapshot, instructions=self._get_instructions_for(to_entity) ) async def human_takeover( self, session_id: str, human_instructions: str ) -> str: """人类接管 Agent 执行""" # 请求控制权 result = await self.request_control( from_entity="agent", to_entity="human", reason="Human takeover", context={"session_id": session_id} ) if not result.success: raise ControlTransferError(result.reason) # 执行人类指令 response = await self.human_executor.execute( session_id=session_id, instructions=human_instructions, starting_from=result.snapshot ) # 可选：交还控制权给 Agent if response.hand_back_to_agent: await self.request_control( from_entity="human", to_entity="agent", reason="Task completed by human", context={"session_id": session_id} ) return response.output 五、需审批模式（Approve） 5.1 审批工作流 class ApprovalWorkflow: """审批工作流""" async def submit_for_approval( self, session_id: str, action: dict, priority: str = "normal" ) -> ApprovalRequest: """提交审批""" request = ApprovalRequest( id=str(uuid.uuid4()), session_id=session_id, action=action, priority=priority, submitted_by="agent", submitted_at=datetime.now(), status="pending", approvers=self._get_approvers(action), deadline=self._calculate_deadline(priority), ) # 保存审批请求 await self.approval_db.save(request) # 通知审批人 await self.notification_service.notify_approvers( request.approvers, f"Agent 请求审批: {request.id}", self._generate_approval_ui(request) ) # 设置超时处理 asyncio.create_task(self._handle_timeout(request)) return request async def handle_approval( self, request_id: str, approver: str, decision: str, # "approve" or "reject" comment: str = "" ) -> ApprovalResult: """处理审批""" request = await self.approval_db.get(request_id) # 检查审批人权限 if approver not in request.approvers: raise UnauthorizedApproverError(approver) # 记录审批 approval = Approval( request_id=request_id, approver=approver, decision=decision, comment=comment, timestamp=datetime.now() ) await self.approval_db.save_approval(approval) # 执行相应操作 if decision == "approve": return await self._execute_approved_action(request) else: return await self._handle_rejection(request, approval) async def _handle_timeout(self, request: ApprovalRequest): """处理审批超时""" await asyncio.sleep(request.deadline - time.time()) # 检查是否已审批 if request.status == "pending": # 根据策略处理 if request.priority == "urgent": # 紧急请求：自动升级 await self._escalate(request) else: # 普通请求：自动拒绝 await self.handle_approval( request.id, approver="system", decision="reject", comment="Approval timeout" ) 5.2 审批 UI class ApprovalUI { render(request: ApprovalRequest): JSX.Element { return ( <Card className="approval-request"> <CardHeader> <Icon name="clipboard-check" /> <Title>审批请求</Title> <Badge variant={request.priority}>{request.priority}</Badge> </CardHeader> <CardBody> <Section title="请求详情"> <Descriptions> <Description label="请求ID">{request.id}</Description> <Description label="提交时间"> {formatDateTime(request.submitted_at)} </Description> <Description label="截止时间"> {formatDateTime(request.deadline)} </Description> </Descriptions> </Section> <Section title="待审批操作"> <ActionDetail action={request.action} /> </Section> <Section title="风险分析"> <RiskAnalysis risk={request.risk_assessment} /> </Section> <Section title="审批意见"> <TextArea placeholder="请输入审批意见（可选）" value={this.state.comment} onChange={this.onCommentChange} /> </Section> </CardBody> <CardFooter> <Button variant="danger" onClick={this.onReject}> 拒绝 </Button> <Button variant="primary" onClick={this.onApprove}> 批准 </Button> </CardFooter> </Card> ); } onApprove = async () => { const result = await api.approveRequest( this.props.request.id, this.state.comment ); if (result.success) { toast.success("已批准"); this.props.onApproved(); } }; onReject = async () => { if (!this.state.comment) { toast.error("拒绝时必须填写意见"); return; } const result = await api.rejectRequest( this.props.request.id, this.state.comment ); if (result.success) { toast.success("已拒绝"); this.props.onRejected(); } }; } 六、渐进式自动化 6.1 信任度模型 class TrustAccumulationModel: """信任度积累模型——支持渐进式自动化""" def __init__(self): self.trust_factors = { "success_rate": 0.30, # 成功率权重 "quality_score": 0.25, # 质量评分权重 "consistency": 0.20, # 一致性权重 "user_feedback": 0.15, # 用户反馈权重 "expert_validation": 0.10, # 专家验证权重 } async def calculate_trust_score( self, agent_id: str, user_id: str, task_type: str ) -> TrustScore: """计算信任度评分""" # 获取历史表现 history = await self._get_agent_history( agent_id, user_id, task_type, days=30 ) if len(history) < 10: return TrustScore( score=0.0, level="new", automation_allowed=False, reason=" insufficient history" ) # 计算各因子得分 factors = {} factors["success_rate"] = history.success_rate factors["quality_score"] = history.avg_quality factors["consistency"] = 1.0 - history.variance factors["user_feedback"] = history.avg_user_rating / 5.0 factors["expert_validation"] = history.expert_approval_rate # 加权计算 score = sum( factors[factor] * weight for factor, weight in self.trust_factors.items() ) # 映射到自动化级别 if score >= 0.90: level, allowed = "full_auto", True elif score >= 0.75: level, allowed = "confirm", True elif score >= 0.50: level, allowed = "supervise", True else: level, allowed = "approve", False return TrustScore( score=score, level=level, automation_allowed=allowed, factors=factors, sample_size=len(history) ) 6.2 自动化级别升级 class AutomationLevelManager: """自动化级别管理""" LEVELS = ["approve", "supervise", "confirm", "auto"] async def try_upgrade_level( self, agent_id: str, user_id: str, task_type: str ) -> UpgradeResult: """尝试升级自动化级别""" current = await self._get_current_level(agent_id, user_id, task_type) trust = await self.trust_model.calculate_trust_score( agent_id, user_id, task_type ) if not trust.automation_allowed: return UpgradeResult( success=False, reason=trust.reason, current_level=current, suggested_level=current ) target_level = self._level_from_trust(trust.score) if self.LEVELS.index(target_level) <= self.LEVELS.index(current): return UpgradeResult( success=False, reason="Already at or above target level", current_level=current, suggested_level=current ) # 请求用户确认升级 confirmation = await self._request_upgrade_confirmation( user_id, current, target_level, trust ) if confirmation.approved: await self._set_level( agent_id, user_id, task_type, target_level ) return UpgradeResult( success=True, reason="User approved upgrade", current_level=current, suggested_level=target_level ) else: return UpgradeResult( success=False, reason="User declined upgrade", current_level=current, suggested_level=current ) 七、人机协作设计 Checklist □ 自动化级别与任务风险匹配 □ 全自动模式有完整护栏 □ 确认点在关键决策前触发 □ 监督模式提供实时状态视图 □ 审批工作流支持多级审批 □ 控制权可在人和 Agent 间转移 □ 信任度模型支持渐进式自动化 □ 所有人工介入点有完整上下文 □ 用户可随时暂停/继续/接管 □ 操作日志完整可追溯 结语 人机协作不是"人 vs 机器"的零和游戏，而是发挥各自优势的组合。Agent 擅长执行、搜索、计算；人类擅长判断、创意、伦理决策。好的协作设计让 Agent 知道何时该求助，让人类知道何时该放手。在 Agent 能力越来越强的2026年，最强大的不是完全自主的 Agent，而是最懂得与人类协作的 Agent。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
...
