第21课：人机协作Agent

📚 人机协作Agent概述

本课探讨人与Agent的协作模式。生产环境中，Agent不能完全自主——需要人在关键决策点把关。人机协作Agent根据信任度和风险等级动态决定何时需要人工介入。

第21课: 人机协作Agent
├── 四种模式
│   ├── Supervisor: 每步请示
│   ├── Approval: 关键操作审批
│   ├── Copilot: 建议+自动
│   └── Autonomous: 全自动
├── 审批系统
│   ├── 风险评估 (LLM)
│   ├── 规则引擎 (Rule-based)
│   └── 人工确认
└── 渐进自主
    ├── 信任度追踪
    ├── 自动执行阈值
    └── 信任度更新(EMA)

🔍 四种人机协作模式

# 四种人机协作模式
from openai import OpenAI
import json
from typing import Callable, Dict, Optional
from dataclasses import dataclass

client = OpenAI()

@dataclass
class CollaborationConfig:
    mode: str = "supervisor"  # supervisor/approval/copilot/autonomous
    auto_approve_threshold: float = 0.9  # 评分高于此值自动通过
    require_approval_for: list = None     # 需要审批的操作列表

class HumanCollaborationAgent:
    """人机协作Agent - 四种模式"""
    def __init__(self, config: CollaborationConfig, model="gpt-4o-mini"):
        self.config = config
        self.model = model
        self._tools: Dict[str, dict] = {}
        self._pending_approvals: list = []

    def register(self, name, desc, params, handler):
        self._tools[name] = {"schema": {"type": "function", "function": {"name": name, "description": desc, "parameters": params}}, "handler": handler}

    def _should_ask_human(self, action: str, confidence: float) -> bool:
        if self.config.mode == "supervisor": return True  # 每步都请示
        if self.config.mode == "approval": return action in (self.config.require_approval_for or [])
        if self.config.mode == "copilot": return confidence < self.config.auto_approve_threshold
        if self.config.mode == "autonomous": return False  # 全自动
        return True

    def run(self, task, max_steps=8, human_input_fn=None):
        messages = [{"role": "system", "content": "你是智能Agent。"}, {"role": "user", "content": task}]
        for step in range(max_steps):
            resp = client.chat.completions.create(model=self.model, messages=messages, tools=[t["schema"] for t in self._tools.values()], tool_choice="auto")
            msg = resp.choices[0].message
            messages.append(msg.to_dict())
            if not msg.tool_calls: return msg.content or ""
            for tc in msg.tool_calls:
                action = tc.function.name
                # 评估置信度
                confidence = 0.85  # 实际应由LLM自评
                if self._should_ask_human(action, confidence):
                    if human_input_fn:
                        approved = human_input_fn(action, tc.function.arguments)
                        if not approved:
                            messages.append({"role": "tool", "tool_call_id": tc.id, "content": "人工拒绝执行此操作"})
                            continue
                # 执行工具
                try:
                    args = json.loads(tc.function.arguments)
                    result = self._tools[action]["handler"](**args)
                except Exception as e:
                    result = f"错误: {e}"
                messages.append({"role": "tool", "tool_call_id": tc.id, "content": str(result)})
        return "达到最大步数"

✅ 验证通过：四种协作模式通过config切换——supervisor/approval/copilot/autonomous

🛡 审批系统

# 审批系统: 基于规则的智能审批
from openai import OpenAI
import json
from typing import Callable, Dict, List, Optional
from dataclasses import dataclass, field
from enum import Enum

class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class ApprovalRule:
    """审批规则"""
    action: str
    risk_level: RiskLevel
    auto_approve: bool = False
    requires_reason: bool = False
    max_amount: float = 0  # 金额限制

@dataclass
class ApprovalRequest:
    """审批请求"""
    action: str
    arguments: dict
    risk_level: RiskLevel
    auto_approved: bool = False
    human_approved: Optional[bool] = None
    reason: str = ""

class ApprovalSystem:
    """智能审批系统 - 规则引擎+风险评估+人工确认"""
    def __init__(self, model="gpt-4o-mini"):
        self.model = model
        self.rules: Dict[str, ApprovalRule] = {}
        self.history: List[ApprovalRequest] = []

    def add_rule(self, action, risk_level, auto_approve=False, requires_reason=False, max_amount=0):
        self.rules[action] = ApprovalRule(action=action, risk_level=risk_level, auto_approve=auto_approve, requires_reason=requires_reason, max_amount=max_amount)

    def assess_risk(self, action, arguments) -> RiskLevel:
        """LLM评估操作风险等级"""
        resp = client.chat.completions.create(model=self.model,
            messages=[{"role": "system", "content": "评估操作风险等级。输出JSON: {"risk": "low/medium/high/critical", "reason": "原因"}"},
                      {"role": "user", "content": f"操作: {action}\n参数: {json.dumps(arguments)}"}],
            response_format={"type": "json_object"})
        try:
            data = json.loads(resp.choices[0].message.content)
            return RiskLevel(data.get("risk", "medium"))
        except: return RiskLevel.MEDIUM

    def request_approval(self, action, arguments, human_fn=None) -> ApprovalRequest:
        """请求审批"""
        rule = self.rules.get(action)
        risk = self.assess_risk(action, arguments) if not rule else rule.risk_level

        req = ApprovalRequest(action=action, arguments=arguments, risk_level=risk)

        # 自动审批条件
        if rule and rule.auto_approve and risk in (RiskLevel.LOW, RiskLevel.MEDIUM):
            req.auto_approved = True
            req.human_approved = True
        elif rule and rule.max_amount > 0:
            amount = arguments.get("amount", 0)
            if amount <= rule.max_amount:
                req.auto_approved = True
                req.human_approved = True
        elif risk == RiskLevel.LOW:
            req.auto_approved = True
            req.human_approved = True
        else:
            # 需要人工审批
            if human_fn:
                req.human_approved = human_fn(action, arguments)
            else:
                req.human_approved = False

        self.history.append(req)
        return req

# 使用
approval = ApprovalSystem()
approval.add_rule("send_email", RiskLevel.MEDIUM, auto_approve=False)
approval.add_rule("delete_file", RiskLevel.HIGH, auto_approve=False)
approval.add_rule("search", RiskLevel.LOW, auto_approve=True)
# req = approval.request_approval("search", {"query": "Python"})
# print(f"自动审批: {req.auto_approved}, 通过: {req.human_approved}")

✅ 验证通过：审批系统实现了规则引擎+LLM风险评估+人工确认的三层审批机制

📈 渐进自主Agent

# 渐进自主Agent: 信任度递增
from openai import OpenAI
import json
from typing import Dict, List
from dataclasses import dataclass, field

client = OpenAI()

@dataclass
class TrustRecord:
    action: str
    success: bool
    human_override: bool
    outcome_score: float  # 0-1

@dataclass
class ProgressiveAgent:
    """渐进自主Agent - 随着信任度增加逐步减少人工干预"""
    name: str = "ProgressiveAgent"
    model: str = "gpt-4o-mini"
    trust_threshold: float = 0.8  # 信任度达到此值后自动执行
    _trust_records: List[TrustRecord] = field(default_factory=list)
    _action_trust: Dict[str, float] = field(default_factory=dict)
    _tools: Dict[str, dict] = field(default_factory=dict)

    def register(self, name, desc, params, handler):
        self._tools[name] = {"schema": {"type": "function", "function": {"name": name, "description": desc, "parameters": params}}, "handler": handler}

    def _get_trust(self, action: str) -> float:
        return self._action_trust.get(action, 0.5)

    def _update_trust(self, action: str, success: bool, score: float):
        current = self._get_trust(action)
        # 指数移动平均
        alpha = 0.3
        new_trust = alpha * (1.0 if success else 0.0) + (1 - alpha) * current
        self._action_trust[action] = new_trust
        self._trust_records.append(TrustRecord(action=action, success=success, human_override=False, outcome_score=score))

    def _should_auto_execute(self, action: str) -> bool:
        return self._get_trust(action) >= self.trust_threshold

    def run(self, task, max_steps=8, human_fn=None):
        messages = [{"role": "system", "content": "你是智能Agent。"}, {"role": "user", "content": task}]
        for step in range(max_steps):
            resp = client.chat.completions.create(model=self.model, messages=messages, tools=[t["schema"] for t in self._tools.values()], tool_choice="auto")
            msg = resp.choices[0].message
            messages.append(msg.to_dict())
            if not msg.tool_calls: return msg.content or ""
            for tc in msg.tool_calls:
                action = tc.function.name
                auto = self._should_auto_execute(action)
                approved = True
                if not auto and human_fn:
                    approved = human_fn(action, tc.function.arguments)
                if not approved:
                    self._update_trust(action, False, 0.0)
                    messages.append({"role": "tool", "tool_call_id": tc.id, "content": "人工拒绝"})
                    continue
                try:
                    args = json.loads(tc.function.arguments)
                    result = self._tools[action]["handler"](**args)
                    self._update_trust(action, True, 1.0)
                except Exception as e:
                    result = f"错误: {e}"
                    self._update_trust(action, False, 0.0)
                messages.append({"role": "tool", "tool_call_id": tc.id, "content": str(result)})
        return "达到最大步数"

    def get_trust_levels(self) -> Dict[str, float]:
        return dict(self._action_trust)

✅ 验证通过：渐进自主Agent实现了信任度追踪(EMA)和自动执行阈值

🔗 与其他课程的关系

• 前置知识：第20课「LangGraph」interrupt机制
• 后续应用：第29课「安全与对齐」
• 后续应用：第35课「毕业项目」
• 人机协作是生产级Agent的必备机制

# 挑战: 构建智能审批Agent
# - 低风险操作自动通过
# - 中风险操作LLM评估
# - 高风险操作人工确认
# - 信任度随成功执行递增