自主Agent审批门六栈：从全自动到人机协同实测

# OpenAI Agents SDK - 无门控全自主
from agents import Agent, Runner

agent = Agent(
    name="autonomous-worker",
    instructions="完成用户指定的数据分析任务",
    tools=[search_tool, analyze_tool, report_tool],
)

# 全自主运行，无任何中断点
result = await Runner.run(agent, input="分析Q1销售数据并生成报告")
# result.final_output 直接返回，中间无人工介入

# LangGraph 静态审批门实现
from langgraph.types import interrupt
from langgraph.graph import StateGraph

def execute_action(state):
    """执行动作前暂停，等待人类批准"""
    action = state["pending_action"]

    # 中断并等待人类审批
    approval = interrupt({
        "question": f"Agent请求执行: {action['tool']}({action['args']})",
        "action_summary": action["description"],
    })

    if approval["approved"]:
        result = tool_executor.run(action)
        return {"results": [result]}
    else:
        return {"results": [{"status": "rejected", "reason": approval.get("reason")}]}

# 构建工作流
graph = StateGraph(AgentState)
graph.add_node("plan", plan_step)
graph.add_node("execute", execute_action)  # 内含interrupt
graph.add_node("validate", validate_step)

graph.add_edge("plan", "execute")
graph.add_edge("execute", "validate")

# Claude Agent SDK - 动态风险分级实现
from claude_agent_sdk import Agent, Guardrail, ToolCall

# 定义动作风险等级
RISK_TIERS = {
    "read_file": "low",       # 读取 - 无需审批
    "search_web": "low",      # 搜索 - 无需审批  
    "write_file": "medium",   # 写入 - 需要摘要审批
    "delete_file": "high",    # 删除 - 需要详细审批
    "execute_shell": "high",  # Shell执行 - 需要详细审批+确认
    "send_email": "critical", # 发送邮件 - 需要逐字段审批
    "database_write": "critical", # 数据库写入 - 需要逐字段审批
}

def risk_classifier(tool_call: ToolCall) -> str:
    """对工具调用进行风险分级"""
    base_risk = RISK_TIERS.get(tool_call.tool_name, "medium")

    # 语义增强：检测高风险参数
    if "DROP" in str(tool_call.args).upper():
        return "critical"
    if "production" in str(tool_call.args).lower():
        return max(base_risk, "high", key=["low","medium","high","critical"].index)

    return base_risk

# 根据风险等级配置不同门控
agent = Agent(
    name="risk-aware-worker",
    guardrails=[
        Guardrail.on_tool_call(
            condition=lambda tc: risk_classifier(tc) == "critical",
            action="block",  # 直接阻止，要求人工介入
        ),
        Guardrail.on_tool_call(
            condition=lambda tc: risk_classifier(tc) == "high",
            action="confirm",  # 弹出详细审批
        ),
    ]
)

# OpenAI Agents SDK - 升级链实现
from agents import Agent, Guardrail

class EscalationChain:
    def __init__(self):
        self.chain = [
            {"level": "team_lead", "max_risk": "medium", "scope": "team"},
            {"level": "manager", "max_risk": "high", "scope": "department"},
            {"level": "director", "max_risk": "critical", "scope": "organization"},
            {"level": "ciso", "max_risk": "systemic", "scope": "enterprise"},
        ]

    def route(self, action, risk_level, scope):
        """根据风险和影响范围路由到正确的审批层级"""
        for level in self.chain:
            risk_order = ["low","medium","high","critical","systemic"]
            if risk_order.index(risk_level) <= risk_order.index(level["max_risk"]):
                if scope == level["scope"] or self._scope_covers(scope, level["scope"]):
                    return level["level"]
        return self.chain[-1]["level"]  # 兜底：最高层审批

    def _scope_covers(self, action_scope, approver_scope):
        scope_hierarchy = {"team": 1, "department": 2, "organization": 3, "enterprise": 4}
        return scope_hierarchy.get(approver_scope, 0) >= scope_hierarchy.get(action_scope, 0)

# 在Agent中集成升级链
escalation = EscalationChain()

agent = Agent(
    name="enterprise-worker",
    instructions="执行企业级数据分析任务",
)
# Agent执行高危动作时，自动路由到对应审批层级

# Anthropic式反馈纠偏环实现
class FeedbackLoop:
    def __init__(self, agent):
        self.agent = agent
        self.feedback_history = []

    async def run_with_feedback(self, task, max_corrections=3):
        context = {"task": task, "previous_feedback": []}

        for iteration in range(max_corrections):
            # Agent自主执行
            result = await self.agent.run(context)

            # 人类审核
            human_review = await self.request_review(result)

            if human_review["approved"]:
                return result  # 审核通过，完成

            # 将反馈注入上下文，影响后续执行
            correction = {
                "iteration": iteration,
                "issues": human_review["issues"],
                "corrections": human_review["corrections"],
                "original_output": result.output,
            }
            context["previous_feedback"].append(correction)
            self.feedback_history.append(correction)

            # Agent基于反馈重新执行
            context["instruction"] = (
                f"你之前的输出存在以下问题：{human_review['issues']}。"
                f"请按照以下纠正方向修改：{human_review['corrections']}"
            )

        return None  # 超过最大纠正次数，任务失败

    async def request_review(self, result):
        """向人类审核者请求审批"""
        return await human_review_queue.submit(result)

# LangGraph - 人类副驾驶模式
from langgraph.graph import StateGraph
from langgraph.types import interrupt

def copilot_step(state):
    """Agent生成建议，人类做决策"""
    # Agent分析当前状态，生成候选方案
    analysis = agent.analyze(state)
    proposals = agent.generate_proposals(analysis, top_k=3)

    # 将方案呈现给人类，等待选择
    choice = interrupt({
        "type": "copilot_proposal",
        "current_state": state["summary"],
        "analysis": analysis,
        "proposals": [
            {
                "id": i,
                "action": p.action,
                "reasoning": p.reasoning,
                "risk": p.risk_assessment,
                "expected_outcome": p.expected_outcome,
            }
            for i, p in enumerate(proposals)
        ],
        "options": ["accept_proposal", "modify_proposal", "reject_all", "custom_action"],
    })

    if choice["action"] == "accept_proposal":
        return agent.execute(choice["proposal_id"])
    elif choice["action"] == "modify_proposal":
        return agent.execute_with_modification(choice["modifications"])
    elif choice["action"] == "custom_action":
        return agent.execute_human_instruction(choice["instruction"])
    else:
        return state  # 人类选择跳过此步

graph TB
    subgraph "自主Agent审批门六栈架构"
        direction TB

        subgraph S1["栈1：无门控全自主"]
            A1[接收任务] --> A2[规划]
            A2 --> A3[直接执行]
            A3 --> A4[返回结果]
        end

        subgraph S2["栈2：静态审批门"]
            B1[接收任务] --> B2[规划]
            B2 --> B3[暂停⏸️]
            B3 -->|人类批准| B4[执行]
            B3 -->|人类拒绝| B5[修改方案]
            B5 --> B3
            B4 --> B6[暂停⏸️]
            B6 -->|人类批准| B7[返回结果]
        end

        subgraph S3["栈3：动态风险分级"]
            C1[接收任务] --> C2[规划]
            C2 --> C3{风险分类器}
            C3 -->|低风险| C4[直接执行]
            C3 -->|中风险| C5[摘要审批]
            C3 -->|高风险| C6[详细审批]
            C3 -->|关键| C7[阻止+人工]
        end

        subgraph S4["栈4：升级链"]
            D1[高风险动作] --> D2{范围判定}
            D2 -->|团队级| D3[Team Lead审批]
            D2 -->|部门级| D4[Manager审批]
            D2 -->|组织级| D5[Director审批]
            D2 -->|全局| D6[CISO审批]
        end

        subgraph S5["栈5：反馈纠偏环"]
            E1[自主执行] --> E2[人类审核]
            E2 -->|通过| E3[完成✅]
            E2 -->|纠正| E4[反馈注入]
            E4 --> E1
        end

        subgraph S6["栈6：人类副驾驶"]
            F1[Agent分析] --> F2[生成候选方案]
            F2 --> F3[人类选择]
            F3 -->|接受| F4[Agent执行]
            F3 -->|修改| F5[Agent调整执行]
            F3 -->|自定义| F6[Agent按指令执行]
        end
    end

    style S1 fill:#e8f5e9
    style S2 fill:#fff3e0
    style S3 fill:#e3f2fd
    style S4 fill:#fce4ec
    style S5 fill:#f3e5f5
    style S6 fill:#fff9c4

graph LR
    A[Agent发起动作] --> B{风险分类}
    B -->|"低风险<br>读操作/搜索"| C[栈1：直接执行]
    B -->|"中风险<br>写操作"| D[栈3：摘要审批]
    B -->|"高风险<br>删除/Shell"| E[栈4：升级链]
    B -->|"关键<br>外部发送/支付"| F[栈6：人类副驾驶]

    C --> G{执行结果}
    D --> G
    E --> G
    F --> G

    G --> H[栈5：反馈纠偏环]
    H -->|反馈注入| A

    style C fill:#c8e6c9
    style D fill:#bbdefb
    style E fill:#ffcdd2
    style F fill:#fff9c4
    style H fill:#e1bee7