优化人机协同以提升 AI 智能体吞吐量

import asyncio
import json
import random
from typing import Dict, Any, Tuple

# 模拟对 n1n.ai 等聚合器的 API 调用
async def call_llm(prompt: str, model: str = "deepseek-v3") -> str:
    # 模拟网络延迟
    await asyncio.sleep(0.5)
    
    # 用于演示的模拟响应
    if "refund" in prompt.lower() and "unopened" in prompt.lower():
        return json.dumps({
            "decision": "approve",
            "reasoning": "Customer returned the item unopened within 30 days.",
            "confidence_score": 0.95
        })
    else:
        return json.dumps({
            "decision": "escalate",
            "reasoning": "Customer claims item was damaged, but no photo was provided.",
            "confidence_score": 0.62
        })

# 异步人类审查队列模拟器
class HumanReviewQueue:
    def __init__(self):
        self.queue = asyncio.Queue()

    async def add_to_queue(self, task_id: str, data: Dict[str, Any]):
        print(f"[人类队列] 任务 {task_id} 已路由至人类审查。原因：置信度低。")
        await self.queue.put((task_id, data))

    async def process_next(self) -> Tuple[str, str]:
        task_id, data = await self.queue.get()
        # 模拟人类需要 2 秒钟来审查和批准
        await asyncio.sleep(2.0)
        decision = "approved_by_human"
        print(f"[人类队列] 任务 {task_id} 已由人类处理完成：{decision}")
        self.queue.task_done()
        return task_id, decision

human_queue = HumanReviewQueue()

async def process_refund_request(request_id: str, customer_email: str, details: str):
    print(f"[智能体] 正在处理用户 {customer_email} 的退款请求 {request_id}...")
    
    prompt = f"Analyze this refund request: {details}. Return JSON with decision, reasoning, and confidence_score."
    
    # 调用大模型（在生产环境中推荐使用 n1n.ai 聚合模式以提高稳定性）
    response_raw = await call_llm(prompt)
    response = json.loads(response_raw)
    
    confidence = response.get("confidence_score", 0.0)
    decision = response.get("decision")
    
    print(f"[智能体] 初步决策: {decision} (置信度: {confidence})")
    
    # 动态路由决策
    CONFIDENCE_THRESHOLD = 0.80
    
    if confidence >= CONFIDENCE_THRESHOLD:
        # 高置信度通道：立即执行
        await execute_refund(request_id, customer_email, decision)
    else:
        # 低置信度通道：异步路由至人类审查
        # 我们不阻塞主线程；直接让出执行权并将任务入队
        task_data = {
            "customer_email": customer_email,
            "details": details,
            "agent_decision": decision,
            "confidence": confidence
        }
        asyncio.create_task(route_to_human_workflow(request_id, task_data))
        print(f"[智能体] 请求 {request_id} 已分流至后台人类审查线程。")

async def route_to_human_workflow(task_id: str, task_data: Dict[str, Any]):
    await human_queue.add_to_queue(task_id, task_data)
    # 在生产系统中，这里通常会写入数据库并触发 Webhook。
    # 在此模拟中，我们在后台处理它。
    resolved_id, human_decision = await human_queue.process_next()
    await execute_refund(resolved_id, task_data["customer_email"], human_decision)

async def execute_refund(request_id: str, email: str, action: str):
    print(f"[执行系统] 退款 {request_id} 执行成功。操作: {action} (用户: {email})。")

# 主执行循环
async def main():
    # 请求 1：高置信度（未拆封商品）
    await process_refund_request("REQ-001", "[email protected]", "I returned my unopened shoes within 10 days.")
    
    # 请求 2：低置信度（商品损坏，需要核实）
    await process_refund_request("REQ-002", "[email protected]", "The box arrived crushed and the screen is cracked.")
    
    # 保持脚本运行以允许后台任务完成
    await asyncio.sleep(3.0)

if __name__ == "__main__":
    asyncio.run(main())