六大Agent框架人机审批门控实战拆解

from langgraph.types import interrupt, Command
from langgraph.graph import StateGraph, START, END

def approve_purchase(state):
    """在执行购买前请求人类审批"""
    # 将审批信息呈现给人类
    decision = interrupt({
        "question": f"是否批准购买 {state['item']}，金额 ${state['amount']}?",
        "options": ["approve", "reject", "modify"]
    })
    # decision 来自人类的恢复输入
    if decision == "reject":
        return {"status": "cancelled", "reason": "human_rejected"}
    return {"approved": True, "modifier": decision}

def execute_purchase(state):
    if not state.get("approved"):
        return {"status": "skipped"}
    # 执行实际购买逻辑
    return {"status": "completed", "order_id": "ORD-12345"}

graph = StateGraph(PurchaseState)
graph.add_node("approve", approve_purchase)
graph.add_node("execute", execute_purchase)
graph.add_edge(START, "approve")
graph.add_edge("approve", "execute")
graph.add_edge("execute", END)

# 使用持久化后端
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
app = graph.compile(checkpointer=checkpointer)

# 第一次运行 —— 在审批节点中断
result = app.invoke(
    {"item": "GPU集群", "amount": 50000},
    config={"configurable": {"thread_id": "tx-001"}}
)
# result 中包含 interrupt 信息

# 人类审批后恢复
for event in app.stream(
    Command(resume="approve"),
    config={"configurable": {"thread_id": "tx-001"}}
):
    print(event)

from crewai import Agent, Task, Crew, Process

reviewer = Agent(
    role="合同审核员",
    goal="审核合同条款并标识风险",
    backstory="你是一名资深法律顾问",
    llm="gpt-4o"
)

# 关键：human_input=True 触发人类审批
review_task = Task(
    description="审核以下合同并给出修改建议：{contract_text}",
    expected_output="修改建议清单",
    agent=reviewer,
    human_input=True  # 执行到此任务时暂停，等待人类输入
)

finalizer = Agent(
    role="合同定稿员",
    goal="根据审核意见生成最终合同",
    backstory="你负责将审核意见落地到合同文本",
    llm="gpt-4o"
)

finalize_task = Task(
    description="根据审核意见修改合同",
    expected_output="最终合同文本",
    agent=finalizer
)

crew = Crew(
    agents=[reviewer, finalizer],
    tasks=[review_task, finalize_task],
    process=Process.sequential
)

result = crew.kickoff(inputs={"contract_text": "..."})

from agents import Agent, Runner, RunConfig, ToolCallApproval
from agents.items import ToolCallItem

async def approval_callback(tool_call: ToolCallItem, context):
    """在工具执行前请求人类审批"""
    tool_name = tool_call.raw_item.function.name
    args = tool_call.raw_item.function.arguments

    # 高风险工具需要人类审批
    HIGH_RISK_TOOLS = {"execute_trade", "send_email", "delete_record"}

    if tool_name in HIGH_RISK_TOOLS:
        print(f"[审批请求] 工具: {tool_name}")
        print(f"  参数: {args}")
        decision = input("批准？(y/n/m=修改): ").strip().lower()

        if decision == "y":
            return ToolCallApproval.APPROVE
        elif decision == "m":
            new_args = input("输入修改后的参数: ")
            return ToolCallApproval.MODIFY(args=new_args)
        else:
            return ToolCallApproval.REJECT

    return ToolCallApproval.APPROVE  # 低风险工具自动放行

trade_agent = Agent(
    name="交易助手",
    instructions="帮助用户执行股票交易",
    tools=[execute_trade, get_portfolio, analyze_risk]
)

config = RunConfig(
    tool_call_approval_callback=approval_callback,
    max_turns=10
)

result = await Runner.run(
    trade_agent,
    messages=[{"role": "user", "content": "买入100股AAPL"}],
    run_config=config
)

from google.adk.agents import LlmAgent
from google.adk.models.types import Content, Part

async def approve_financial_tools(callback_context, tool_call):
    """在金融工具执行前拦截并审批"""
    tool_name = tool_call.function.name
    args = tool_call.function.args

    FINANCIAL_TOOLS = {"transfer_money", "approve_loan", "execute_trade"}

    if tool_name not in FINANCIAL_TOOLS:
        return None  # 非金融工具自动放行

    # 构建审批请求并发送给审批系统
    approval_request = {
        "tool": tool_name,
        "args": args,
        "requester": callback_context.agent_name,
        "timestamp": callback_context.invocation_id
    }

    # 调用企业审批系统API
    decision = await call_approval_system(approval_request)

    if decision["status"] == "approved":
        callback_context.state["approval_id"] = decision["approval_id"]
        return None  # 放行
    else:
        # 返回拒绝信息作为工具输出
        return Content(
            parts=[Part(text=f"审批被拒绝：{decision['reason']}")],
            role="model"
        )

banking_agent = LlmAgent(
    name="银行助手",
    model="gemini-2.5-pro",
    instruction="帮助客户处理银行业务",
    tools=[transfer_money, approve_loan, get_balance],
    before_tool_callback=approve_financial_tools
)

// C# 示例 —— MAF + AG-UI 审批模式
using Microsoft.AgentFramework;
using Microsoft.AgentFramework.Activities;

public class InvoiceApprovalAgent : Agent
{
    [AgentTool("提交发票审批")]
    public async Task<ApprovalResult> SubmitForApproval(
        Invoice invoice,
        AgentContext context)
    {
        // 创建审批Activity
        var approval = new ApprovalActivity
        {
            Title = $"发票审批：{invoice.Vendor} - ¥{invoice.Amount}",
            Description = invoice.Description,
            Approvers = DetermineApprovers(invoice.Amount),
            // 金额 > ¥50000 需要两级审批
            ApprovalLevels = invoice.Amount > 50000 ? 2 : 1,
            Timeout = TimeSpan.FromHours(24),
            Data = new Dictionary<string, object>
            {
                ["invoice_id"] = invoice.Id,
                ["amount"] = invoice.Amount,
                ["vendor"] = invoice.Vendor
            }
        };

        // 发出审批请求，阻塞等待结果
        var result = await context.WaitForApprovalAsync(approval);

        return result.Status switch
        {
            ApprovalStatus.Approved => new ApprovalResult { 
                Approved = true, 
                ApprovalId = result.ApprovalId 
            },
            ApprovalStatus.Rejected => new ApprovalResult { 
                Approved = false, 
                Reason = result.RejectionReason 
            },
            ApprovalStatus.Timeout => new ApprovalResult { 
                Approved = false, 
                Reason = "审批超时，已自动拒绝" 
            },
            _ => throw new InvalidOperationException()
        };
    }

    private List<string> DetermineApprovers(decimal amount) =>
        amount switch
        {
            > 100000 => new() { "CFO", "CEO" },     // 超大额：CFO + CEO
            > 50000  => new() { "CFO", "总监" },     // 大额：CFO + 总监
            _        => new() { "直属经理" }          // 普通额：直属经理
        };
}

from strands import Agent, tool
from strands_tools import approval

@tool
def process_refund(order_id: str, amount: float, reason: str):
    """处理退款请求"""
    # 实际退款逻辑
    return {"status": "processed", "refund_id": f"REF-{order_id}"}

@tool  
def check_risk_level(amount: float) -> str:
    """评估退款风险等级"""
    if amount > 10000:
        return "HIGH"
    elif amount > 1000:
        return "MEDIUM"
    return "LOW"

# 定义审批策略
async def approval_gate(tool_name: str, tool_args: dict, context):
    """Strands的工具审批门控"""
    if tool_name == "process_refund":
        risk = await context.agent.call_tool(
            "check_risk_level", 
            {"amount": tool_args["amount"]}
        )

        if risk == "HIGH":
            # 高风险退款需要人类审批
            decision = await request_human_approval(
                title=f"高风险退款审批",
                details=f"订单: {tool_args['order_id']}, "
                       f"金额: ${tool_args['amount']}",
                approvers=["finance_lead"]
            )
            if not decision.approved:
                return {"status": "rejected", "reason": decision.reason}

    return None  # 放行

refund_agent = Agent(
    model="anthropic.claude-sonnet-4-20250514",
    tools=[process_refund, check_risk_level],
    system_prompt="你是退款处理助手，帮助客户处理退款请求",
    tool_approval_callback=approval_gate
)

result = refund_agent("客户要求退还订单 #12345 的 ¥15000")

graph TB
    subgraph "LangGraph: 状态图中断模型"
        LG1[Agent节点] -->|interrupt| LG2[状态快照持久化]
        LG2 -->|等待| LG3[人类审批]
        LG3 -->|Command resume| LG4[恢复执行]
    end

    subgraph "CrewAI: 任务级标记模型"
        CA1[Task执行] -->|human_input=True| CA2[CLI暂停]
        CA2 -->|键盘输入| CA3[继续下一Task]
    end

    subgraph "OpenAI SDK: 回调审批模型"
        OA1[Runner循环] -->|Tool Call| OA2[回调检查]
        OA2 -->|Approve/Reject| OA3[执行/跳过]
    end

    subgraph "Google ADK: 六类回调模型"
        GA1[Agent生命周期] --> GA2[before_agent]
        GA2 --> GA3[before_model]
        GA3 --> GA4[before_tool]
        GA4 -->|拦截决策| GA5[放行/替换/拒绝]
    end

    subgraph "MAF: AG-UI协议模型"
        MF1[Agent Activity] -->|ApprovalActivity| MF2[AG-UI路由]
        MF2 -->|多审批人| MF3[审批界面]
        MF3 -->|超时策略| MF4[结果回传]
    end

    subgraph "AWS Strands: 模型驱动模型"
        AS1[模型推理] -->|工具选择| AS2[回调门控]
        AS2 -->|放行| AS3[Bedrock执行]
        AS2 -->|拦截| AS4[人类审批]
    end

    style LG2 fill:#e1f5fe
    style CA2 fill:#fff3e0
    style OA2 fill:#e8f5e9
    style GA4 fill:#fce4ec
    style MF2 fill:#f3e5f5
    style AS2 fill:#fff8e1

graph LR
    START[需要HITL审批] --> Q1{审批复杂度?}
    Q1 -->|"简单:执行前确认"| CREW[CrewAI]
    Q1 -->|"中等:多条件门控"| Q2{云厂商绑定?}
    Q1 -->|"复杂:多级多审批人"| Q3{技术栈偏好?}

    Q2 -->|"OpenAI生态"| OPEN[OpenAI Agents SDK]
    Q2 -->|"Google Cloud"| ADK[Google ADK]
    Q2 -->|"AWS"| STRANDS[AWS Strands]
    Q2 -->|"云中立"| LANG[LangGraph]

    Q3 -->|".NET/Azure"| MAF[Microsoft Agent Framework]
    Q3 -->|"Python优先"| LANG2[LangGraph]

    CREW --> END1[快速原型/内部工具]
    OPEN --> END2[SaaS产品]
    ADK --> END3[GCP企业应用]
    STRANDS --> END4[AWS企业应用]
    MAF --> END5[大型企业平台]
    LANG --> END6[通用生产系统]
    LANG2 --> END6

    style CREW fill:#fff3e0
    style OPEN fill:#e8f5e9
    style ADK fill:#fce4ec
    style STRANDS fill:#fff8e1
    style MAF fill:#f3e5f5
    style LANG fill:#e1f5fe
    style LANG2 fill:#e1f5fe