Agent 人机协作设计：从全自动到 Human-in-the-loop

       全自主 Agent                    全人工控制
◀─────────┬─────────┬─────────┬─────────▶
          │         │         │
    全自动执行   需确认   需监督   需审批
     (Auto)    (Confirm) (Supervise) (Approve)

设计原则：让 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
        )

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
        )

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="修改计划",
        )

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()
    });
  };
}

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

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

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"
                )

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();
    }
  };
}

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)
        )

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
            )

□ 自动化级别与任务风险匹配
□ 全自动模式有完整护栏
□ 确认点在关键决策前触发
□ 监督模式提供实时状态视图
□ 审批工作流支持多级审批
□ 控制权可在人和 Agent 间转移
□ 信任度模型支持渐进式自动化
□ 所有人工介入点有完整上下文
□ 用户可随时暂停/继续/接管
□ 操作日志完整可追溯