多 Agent 协作系统架构设计：从理论到实战

# 概念示例：客服系统中的 Agent 分工
class Agent:
    def __init__(self, role, capabilities):
        self.role = role
        self.capabilities = capabilities

# 意图识别 Agent - 专注理解用户需求
intent_agent = Agent(
    role="intent_detector",
    capabilities=["text_classification", "intent_matching"]
)

# 知识检索 Agent - 专注搜索和提取信息
knowledge_agent = Agent(
    role="knowledge_retriever",
    capabilities=["vector_search", "knowledge_extraction"]
)

# 工具调用 Agent - 专注执行操作
tool_agent = Agent(
    role="tool_executor",
    capabilities=["api_call", "data_processing"]
)

┌─────────────────────────────────────────┐
│          用户请求                        │
└────────────────┬────────────────────────┘
                 ↓
┌─────────────────────────────────────────┐
│      Coordinator (协调者)               │
│  - 任务分解                              │
│  - Agent 分配                            │
│  - 结果整合                              │
└────────────────┬────────────────────────┘
                 ↓
    ┌────────────┼────────────┐
    ↓            ↓            ↓
┌──────┐    ┌──────┐    ┌──────┐
│Agent A│    │Agent B│    │Agent C│
│执行1 │    │执行2 │    │执行3 │
└──────┘    └──────┘    └──────┘
    ↓            ↓            ↓
    └────────────┼────────────┘
                 ↓
┌─────────────────────────────────────────┐
│      Coordinator (结果整合)              │
└────────────────┬────────────────────────┘
                 ↓
          最终响应

from typing import List, Dict, Any
from dataclasses import dataclass

@dataclass
class Task:
    id: str
    description: str
    required_capabilities: List[str]

@dataclass
class Agent:
    id: str
    role: str
    capabilities: List[str]

class Coordinator:
    def __init__(self):
        self.agents: Dict[str, Agent] = {}
        self.task_results: Dict[str, Any] = {}

    def register_agent(self, agent: Agent):
        """注册 Agent 及其能力"""
        self.agents[agent.id] = agent
        print(f"✅ 注册 Agent: {agent.role} ({agent.id})")

    def decompose_task(self, user_request: str) -> List[Task]:
        """分解用户请求为子任务"""
        # 实际中这里用 LLM 来智能分解
        if "搜索" in user_request and "总结" in user_request:
            return [
                Task("search", "搜索相关信息", ["search", "retrieval"]),
                Task("analyze", "分析搜索结果", ["analysis", "summarization"]),
            ]
        return [Task("default", user_request, ["general"])]

    def assign_task(self, task: Task) -> str:
        """根据任务需求分配最合适的 Agent"""
        for agent_id, agent in self.agents.items():
            if any(cap in agent.capabilities for cap in task.required_capabilities):
                print(f"📋 任务 {task.id} 分配给 {agent.role}")
                return agent_id
        return None

    def execute_task(self, agent_id: str, task: Task) -> Any:
        """调用 Agent 执行任务"""
        agent = self.agents[agent_id]
        print(f"⚙️ {agent.role} 执行任务: {task.description}")
        # 模拟执行
        return f"由 {agent.role} 处理的结果"

    def aggregate_results(self, results: Dict[str, Any]) -> str:
        """整合所有 Agent 的结果"""
        print("🔗 整合结果...")
        return "\n\n".join(f"[{k}] {v}" for k, v in results.items())

    def process(self, user_request: str) -> str:
        """完整处理流程"""
        print(f"\n📨 收到请求: {user_request}")

        # 1. 分解任务
        tasks = self.decompose_task(user_request)
        print(f"🔨 分解为 {len(tasks)} 个子任务")

        # 2. 分配和执行
        assignments = {}
        results = {}
        for task in tasks:
            agent_id = self.assign_task(task)
            if agent_id:
                assignments[task.id] = agent_id
                results[task.id] = self.execute_task(agent_id, task)

        # 3. 整合结果
        return self.aggregate_results(results)


# 实战：构建客服系统协调者
coordinator = Coordinator()

# 注册 Agent
coordinator.register_agent(Agent(
    "agent1", "意图识别", ["intent_detection", "nlp"]
))
coordinator.register_agent(Agent(
    "agent2", "知识检索", ["search", "retrieval", "vector_search"]
))
coordinator.register_agent(Agent(
    "agent3", "答案生成", ["generation", "summarization"]
))

# 处理用户请求
result = coordinator.process("搜索产品使用手册并总结关键步骤")
print("\n" + "="*50)
print(result)

用户请求 → Agent A (随机或负载均衡)
              ↓
         广播任务
              ↓
    ┌─────────┼─────────┐
    ↓         ↓         ↓
Agent B   Agent C   Agent D
(自主决策) (自主决策) (自主决策)
    ↓         ↓         ↓
    └─────────┼─────────┘
              ↓
         结果聚合
              ↓
          最终响应

import asyncio
from typing import List, Optional
from enum import Enum

class Message:
    def __init__(self, sender: str, content: str, task_id: str = None):
        self.sender = sender
        self.content = content
        self.task_id = task_id
        self.timestamp = asyncio.get_event_loop().time()

class SwarmAgent:
    def __init__(self, agent_id: str, role: str, capabilities: List[str]):
        self.agent_id = agent_id
        self.role = role
        self.capabilities = capabilities
        self.message_queue = asyncio.Queue()
        self.knowledge_base = {}

    async def receive(self, message: Message):
        """接收消息"""
        await self.message_queue.put(message)

    async def process_messages(self):
        """处理消息队列"""
        while True:
            message = await self.message_queue.get()
            print(f"[{self.role}] 收到消息: {message.content[:50]}...")

            # 判断是否需要处理
            if self.can_handle(message):
                result = await self.handle(message)
                return result

            # 如果自己处理不了，转发给其他 Agent
            # 实际中这里需要更智能的转发逻辑
            print(f"[{self.role}] 无法处理，转发给其他 Agent")

    def can_handle(self, message: Message) -> bool:
        """判断是否能处理该消息"""
        # 简化版：根据关键词判断
        if "搜索" in message.content and "search" in self.capabilities:
            return True
        if "生成" in message.content and "generation" in self.capabilities:
            return True
        return False

    async def handle(self, message: Message) -> str:
        """处理消息并返回结果"""
        print(f"[{self.role}] 开始处理...")
        # 模拟处理时间
        await asyncio.sleep(0.5)
        return f"[{self.role}] 处理结果"

class SwarmNetwork:
    def __init__(self):
        self.agents: List[SwarmAgent] = []

    def add_agent(self, agent: SwarmAgent):
        self.agents.append(agent)

    async def broadcast(self, message: Message):
        """广播消息给所有 Agent"""
        tasks = []
        for agent in self.agents:
            tasks.append(agent.receive(message))
        await asyncio.gather(*tasks)

    async def process_request(self, user_request: str) -> str:
        """处理用户请求"""
        print(f"\n📨 Swarm 收到请求: {user_request}")

        # 广播任务
        message = Message("user", user_request)
        await self.broadcast(message)

        # 启动所有 Agent 处理（实际中需要更智能的协调）
        tasks = []
        for agent in self.agents:
            tasks.append(agent.process_messages())

        # 等待第一个完成的结果
        done, pending = await asyncio.wait(
            tasks,
            return_when=asyncio.FIRST_COMPLETED
        )

        # 取消其他任务
        for task in pending:
            task.cancel()

        # 获取结果
        for task in done:
            result = task.result()
            if result:
                return result

        return "无法处理该请求"


# 实战：构建 Swarm 系统
network = SwarmNetwork()

# 添加 Agent
network.add_agent(SwarmAgent("swarm1", "搜索专家", ["search", "retrieval"]))
network.add_agent(SwarmAgent("swarm2", "生成专家", ["generation", "summarization"]))
network.add_agent(SwarmAgent("swarm3", "分析专家", ["analysis", "reasoning"]))

# 处理请求
result = asyncio.run(network.process_request("搜索最新 AI 技术趋势并生成报告"))
print("\n" + "="*50)
print(result)

┌─────────────────────────────────────────┐
│      Orchestrator (编排器)              │
│  - 全局规划                              │
│  - 资源分配                              │
│  - 质量监控                              │
└────────────────┬────────────────────────┘
                 ↓
┌─────────────────────────────────────────┐
│      Manager Pool (管理层)               │
│  - 任务分解                              │
│  - 局部协调                              │
└────────────────┬────────────────────────┘
                 ↓
    ┌────────────┼────────────┐
    ↓            ↓            ↓
┌──────┐    ┌──────┐    ┌──────┐
│Team A│    │Team B│    │Team C│
│协作组│    │协作组│    │协作组│
│(内部  │    │(内部  │    │(内部  │
│ Swarm)│    │ Swarm)│    │ Swarm)│
└──────┘    └──────┘    └──────┘

from abc import ABC, abstractmethod
from typing import List, Optional

class Orchestrator:
    """全局编排器"""
    def __init__(self):
        self.managers = []

    def register_manager(self, manager):
        self.managers.append(manager)

    def plan(self, request: str) -> dict:
        """全局规划：决定用哪些 Manager"""
        # 实际中这里用 LLM 分析请求
        if "复杂" in request:
            return {"managers": ["manager1", "manager2"], "strategy": "parallel"}
        return {"managers": ["manager1"], "strategy": "sequential"}

    def execute(self, request: str) -> str:
        """执行全局计划"""
        plan = self.plan(request)
        print(f"📋 全局计划: {plan}")

        if plan["strategy"] == "parallel":
            # 并行执行多个 Manager
            results = []
            for manager_id in plan["managers"]:
                manager = self.get_manager(manager_id)
                if manager:
                    results.append(manager.execute(request))
            return "\n".join(results)
        else:
            # 串行执行
            manager = self.get_manager(plan["managers"][0])
            if manager:
                return manager.execute(request)

    def get_manager(self, manager_id: str):
        for manager in self.managers:
            if manager.id == manager_id:
                return manager
        return None


class Manager:
    """局部管理层：负责一个协作组"""
    def __init__(self, manager_id: str, domain: str):
        self.id = manager_id
        self.domain = domain
        self.agents = []

    def add_agent(self, agent):
        self.agents.append(agent)

    def decompose(self, request: str) -> List[dict]:
        """分解任务给协作组内的 Agent"""
        # 根据领域特性分解
        tasks = []
        for agent in self.agents:
            if agent.can_handle(request):
                tasks.append({
                    "agent": agent,
                    "task": request
                })
        return tasks

    def coordinate(self, tasks: List[dict]) -> str:
        """协调协作组内的 Agent 执行"""
        results = []
        for task in tasks:
            result = task["agent"].execute(task["task"])
            results.append(result)
        return "\n".join(results)

    def execute(self, request: str) -> str:
        """执行请求"""
        print(f"🔧 [{self.id}] 处理: {request}")
        tasks = self.decompose(request)
        return self.coordinate(tasks)


class WorkerAgent:
    """执行 Agent"""
    def __init__(self, agent_id: str, role: str, capabilities: List[str]):
        self.id = agent_id
        self.role = role
        self.capabilities = capabilities

    def can_handle(self, request: str) -> bool:
        # 简化判断
        for cap in self.capabilities:
            if cap.lower() in request.lower():
                return True
        return False

    def execute(self, task: str) -> str:
        """执行任务"""
        print(f"  ⚡ [{self.role}] 执行: {task[:30]}...")
        # 模拟执行
        return f"[{self.role}] 完成: {task}"


# 实战：构建混合架构系统
orchestrator = Orchestrator()

# 创建 Manager 1：负责客服领域
manager1 = Manager("manager1", "customer_service")
manager1.add_agent(WorkerAgent("w1", "意图识别", ["intent", "understand"]))
manager1.add_agent(WorkerAgent("w2", "知识检索", ["search", "retrieve"]))
manager1.add_agent(WorkerAgent("w3", "答案生成", ["generate", "answer"]))

# 创建 Manager 2：负责数据分析领域
manager2 = Manager("manager2", "data_analysis")
manager2.add_agent(WorkerAgent("w4", "数据收集", ["collect", "fetch"]))
manager2.add_agent(WorkerAgent("w5", "数据处理", ["process", "clean"]))
manager2.add_agent(WorkerAgent("w6", "数据可视化", ["visualize", "chart"]))

# 注册 Manager
orchestrator.register_manager(manager1)
orchestrator.register_manager(manager2)

# 处理请求
result = orchestrator.execute("复杂的客户咨询，需要搜索和生成答案")
print("\n" + "="*50)
print(result)

class SyncCommunicator:
    def send_and_wait(self, receiver: str, message: str) -> str:
        """发送消息并等待响应"""
        print(f"📤 发送到 {receiver}: {message}")
        # 阻塞等待响应
        response = self._execute(receiver, message)
        print(f"📥 收到响应: {response}")
        return response

    def _execute(self, receiver: str, message: str) -> str:
        # 模拟执行
        return f"来自 {receiver} 的响应"

import asyncio

class AsyncCommunicator:
    def __init__(self):
        self.pending_requests = {}

    async def send_async(self, receiver: str, message: str) -> str:
        """异步发送消息"""
        task_id = str(id(message))

        # 创建 Future 等待响应
        future = asyncio.Future()
        self.pending_requests[task_id] = future

        # 发送消息
        print(f"📤 异步发送到 {receiver}: {message}")
        asyncio.create_task(self._execute_async(receiver, message, task_id))

        # 等待响应
        response = await future
        print(f"📥 收到响应: {response}")
        return response

    async def _execute_async(self, receiver: str, message: str, task_id: str):
        """异步执行"""
        await asyncio.sleep(1)  # 模拟耗时
        # 返回结果
        if task_id in self.pending_requests:
            self.pending_requests[task_id].set_result(
                f"来自 {receiver} 的响应"
            )

import json
from datetime import datetime
from typing import Optional

class AgentMessage:
    def __init__(
        self,
        sender: str,
        receiver: str,
        message_type: str,
        content: dict,
        task_id: Optional[str] = None,
        reply_to: Optional[str] = None
    ):
        self.sender = sender
        self.receiver = receiver
        self.message_type = message_type  # request, response, broadcast
        self.content = content
        self.task_id = task_id or str(datetime.now().timestamp())
        self.reply_to = reply_to
        self.timestamp = datetime.now().isoformat()

    def to_json(self) -> str:
        """序列化为 JSON"""
        return json.dumps(self.__dict__, ensure_ascii=False)

    @classmethod
    def from_json(cls, json_str: str) -> 'AgentMessage':
        """从 JSON 反序列化"""
        data = json.loads(json_str)
        return cls(**data)

# 使用示例
message = AgentMessage(
    sender="agent1",
    receiver="agent2",
    message_type="request",
    content={
        "action": "search",
        "query": "Python 多进程编程",
        "filters": {"language": "zh-CN"}
    }
)

print(message.to_json())

from typing import List, Tuple
import heapq

class TaskScheduler:
    def __init__(self):
        self.agents = []  # (capability_score, agent)

    def register_agent(self, agent, capabilities: List[str]):
        """注册 Agent 及其能力"""
        # 计算能力得分（简化版）
        score = len(capabilities)
        heapq.heappush(self.agents, (-score, agent))  # 负数用于最大堆

    def assign_task(self, task_requirements: List[str]) -> Optional:
        """根据任务需求分配最合适的 Agent"""
        if not self.agents:
            return None

        # 找到能力匹配度最高的 Agent
        best_agent = None
        best_score = -1

        for neg_score, agent in self.agents:
            capabilities = agent.capabilities
            # 计算匹配得分
            matched = sum(1 for req in task_requirements if req in capabilities)
            if matched > best_score:
                best_score = matched
                best_agent = agent

        if best_score > 0:
            print(f"🎯 任务分配给 {best_agent.role} (匹配度: {best_score})")
            return best_agent
        else:
            print("⚠️ 没有合适的 Agent")
            return None


class CapableAgent:
    def __init__(self, agent_id, role, capabilities):
        self.id = agent_id
        self.role = role
        self.capabilities = capabilities


# 使用示例
scheduler = TaskScheduler()

# 注册 Agent
scheduler.register_agent(
    CapableAgent("a1", "全栈专家", ["search", "generate", "analyze", "code"])
)
scheduler.register_agent(
    CapableAgent("a2", "搜索专家", ["search", "retrieve", "index"])
)
scheduler.register_agent(
    CapableAgent("a3", "生成专家", ["generate", "write", "summarize"])
)

# 分配任务
agent = scheduler.assign_task(["search", "retrieve"])
agent = scheduler.assign_task(["generate", "write"])

import threading
import time
from collections import defaultdict

class LoadBalancedScheduler:
    def __init__(self):
        self.agents = []
        self.agent_load = defaultdict(int)  # Agent ID -> 当前任务数
        self.lock = threading.Lock()

    def register_agent(self, agent):
        self.agents.append(agent)

    def assign_task(self, task_requirements: List[str]) -> Optional:
        """基于负载均衡分配任务"""
        with self.lock:
            # 找到能处理且负载最低的 Agent
            best_agent = None
            min_load = float('inf')

            for agent in self.agents:
                # 检查能力匹配
                can_handle = any(
                    req in agent.capabilities
                    for req in task_requirements
                )

                if can_handle and self.agent_load[agent.id] < min_load:
                    min_load = self.agent_load[agent.id]
                    best_agent = agent

            if best_agent:
                self.agent_load[best_agent.id] += 1
                print(f"⚖️ 分配给 {best_agent.role} (当前负载: {min_load + 1})")
                return best_agent

        return None

    def complete_task(self, agent_id: str):
        """标记任务完成"""
        with self.lock:
            self.agent_load[agent_id] -= 1
            if self.agent_load[agent_id] < 0:
                self.agent_load[agent_id] = 0
            print(f"✅ Agent {agent_id} 任务完成，当前负载: {self.agent_load[agent_id]}")

    def get_load_status(self) -> dict:
        """获取负载状态"""
        return dict(self.agent_load)


# 模拟并发任务分配
scheduler = LoadBalancedScheduler()

# 注册 3 个相同能力的 Agent
for i in range(1, 4):
    scheduler.register_agent(
        CapableAgent(f"agent{i}", f"Worker {i}", ["search", "generate"])
    )

# 并发分配任务
def worker(task_id):
    agent = scheduler.assign_task(["search"])
    if agent:
        time.sleep(0.5)  # 模拟处理
        scheduler.complete_task(agent.id)
        return f"任务 {task_id} 完成"
    return f"任务 {task_id} 失败"

import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(worker, i) for i in range(10)]
    for future in concurrent.futures.as_completed(futures):
        print(future.result())

print("\n最终负载状态:", scheduler.get_load_status())

from enum import IntEnum

class TaskPriority(IntEnum):
    CRITICAL = 1  # 关键任务
    HIGH = 2      # 高优先级
    NORMAL = 3    # 正常优先级
    LOW = 4       # 低优先级

class ConflictResolver:
    def __init__(self):
        self.task_queue = []
        self.lock = threading.Lock()

    def submit_task(self, agent_id: str, task: str, priority: TaskPriority):
        """提交任务"""
        with self.lock:
            heapq.heappush(self.task_queue, (priority, task, agent_id))
            print(f"📝 提交任务: {task} (优先级: {priority.name})")

    def resolve_conflict(self) -> Optional[dict]:
        """解决冲突，返回优先级最高的任务"""
        with self.lock:
            if not self.task_queue:
                return None

            priority, task, agent_id = heapq.heappop(self.task_queue)
            print(f"🎯 执行任务: {task} (优先级: {priority.name})")
            return {"task": task, "agent_id": agent_id, "priority": priority}


# 使用示例
resolver = ConflictResolver()

# 提交多个任务（相同 Agent）
resolver.submit_task("agent1", "紧急修复 Bug", TaskPriority.CRITICAL)
resolver.submit_task("agent1", "生成代码", TaskPriority.NORMAL)
resolver.submit_task("agent1", "搜索资料", TaskPriority.HIGH)

# 按优先级处理
while True:
    task_info = resolver.resolve_conflict()
    if not task_info:
        break
    # 执行任务...

from datetime import datetime

class TimestampResolver:
    def __init__(self):
        self.pending_tasks = []

    def submit_task(self, task: str, agent_id: str):
        """提交任务（带时间戳）"""
        timestamp = datetime.now()
        self.pending_tasks.append({
            "task": task,
            "agent_id": agent_id,
            "timestamp": timestamp
        })
        print(f"📝 [{timestamp.strftime('%H:%M:%S')}] 提交任务: {task}")

    def resolve_by_timestamp(self) -> Optional[dict]:
        """按时间戳解决冲突（先进先出）"""
        if not self.pending_tasks:
            return None

        # 按时间戳排序
        self.pending_tasks.sort(key=lambda x: x["timestamp"])

        # 取最早的任务
        task = self.pending_tasks.pop(0)
        print(f"⏱️ 处理最早任务: {task['task']}")
        return task


# 使用示例
resolver = TimestampResolver()
resolver.submit_task("任务 A", "agent1")
import time
time.sleep(0.2)
resolver.submit_task("任务 B", "agent1")
resolver.submit_task("任务 C", "agent1")

# 按时间戳处理
while True:
    task = resolver.resolve_by_timestamp()
    if not task:
        break

from threading import Lock

class SharedState:
    """共享状态存储（线程安全）"""
    def __init__(self):
        self.data = {}
        self.lock = Lock()

    def set(self, key: str, value):
        """设置状态"""
        with self.lock:
            self.data[key] = value
            print(f"📝 状态更新: {key} = {value}")

    def get(self, key: str):
        """获取状态"""
        with self.lock:
            return self.data.get(key)

    def update(self, updates: dict):
        """批量更新"""
        with self.lock:
            self.data.update(updates)
            print(f"📝 批量更新: {updates}")


class StateAwareAgent:
    """感知状态的 Agent"""
    def __init__(self, agent_id, role, shared_state: SharedState):
        self.id = agent_id
        self.role = role
        self.state = shared_state

    def execute(self, task: str):
        """执行任务并更新状态"""
        print(f"⚡ [{self.role}] 执行: {task}")

        # 读取当前状态
        context = self.state.get("context")
        print(f"📖 当前上下文: {context}")

        # 更新状态
        self.state.set(f"{self.role}_progress", "processing")
        self.state.set("last_updated_by", self.id)

        # 模拟处理
        # ...

        # 完成后更新
        self.state.set(f"{self.role}_progress", "completed")


# 使用示例
shared_state = SharedState()
shared_state.set("context", "用户查询 Python 性能优化")

agent1 = StateAwareAgent("a1", "搜索 Agent", shared_state)
agent2 = StateAwareAgent("a2", "分析 Agent", shared_state)

agent1.execute("搜索性能优化资料")
agent2.execute("分析搜索结果")

from datetime import datetime
from typing import List, Dict

class Event:
    def __init__(self, event_type: str, data: dict, agent_id: str):
        self.event_type = event_type
        self.data = data
        self.agent_id = agent_id
        self.timestamp = datetime.now()

class EventStore:
    """事件存储"""
    def __init__(self):
        self.events: List[Event] = []

    def append(self, event: Event):
        """追加事件"""
        self.events.append(event)
        print(f"📌 事件: {event.event_type} by {event.agent_id}")

    def get_events_for_agent(self, agent_id: str) -> List[Event]:
        """获取 Agent 的所有事件"""
        return [e for e in self.events if e.agent_id == agent_id]

    def replay_state(self) -> dict:
        """重放事件，重建当前状态"""
        state = {}
        for event in self.events:
            # 根据事件类型更新状态
            if event.event_type == "task_started":
                state[f"{event.data['task_id']}_status"] = "in_progress"
            elif event.event_type == "task_completed":
                state[f"{event.data['task_id']}_status"] = "completed"
        return state


class EventDrivenAgent:
    """事件驱动的 Agent"""
    def __init__(self, agent_id: str, role: str, event_store: EventStore):
        self.id = agent_id
        self.role = role
        self.event_store = event_store

    def emit_event(self, event_type: str, data: dict):
        """发出事件"""
        event = Event(event_type, data, self.id)
        self.event_store.append(event)

    def execute_task(self, task_id: str, task: str):
        """执行任务并记录事件"""
        self.emit_event("task_started", {"task_id": task_id, "task": task})

        # 执行任务
        print(f"⚡ [{self.role}] 执行: {task}")

        # 完成
        self.emit_event("task_completed", {"task_id": task_id, "result": "success"})


# 使用示例
event_store = EventStore()
agent = EventDrivenAgent("agent1", "Worker", event_store)

agent.execute_task("task1", "处理用户请求")

# 查看状态
current_state = event_store.replay_state()
print(f"\n📊 当前状态: {current_state}")

# 架构图（文字描述）
"""
┌─────────────────────────────────────────┐
│           用户请求                        │
└────────────────┬────────────────────────┘
                 ↓
┌─────────────────────────────────────────┐
│      Orchestrator (编排器)              │
│  - 流程编排                              │
│  - 路由决策                              │
│  - 超时控制                              │
└────────────────┬────────────────────────┘
                 ↓
       ┌─────────┼─────────┐
       ↓         ↓         ↓
  ┌────────┐ ┌────────┐ ┌────────┐
  │ 意图   │ │ 知识   │ │ 答案   │
  │ 识别组 │ │ 检索组 │ │ 生成组 │
  └────────┘ └────────┘ └────────┘
     Swarm      Swarm      Swarm
"""

from typing import Dict, Optional
import asyncio

class CustomerServiceOrchestrator:
    def __init__(self):
        self.intent_group = None
        self.knowledge_group = None
        self.answer_group = None

        # 配置超时时间（秒）
        self.timeouts = {
            "intent_detection": 2,
            "knowledge_retrieval": 3,
            "answer_generation": 3
        }

    def setup_groups(self, intent_group, knowledge_group, answer_group):
        """设置协作组"""
        self.intent_group = intent_group
        self.knowledge_group = knowledge_group
        self.answer_group = answer_group

    async def process(self, user_query: str, user_id: str) -> dict:
        """完整处理流程"""
        start_time = asyncio.get_event_loop().time()

        try:
            # Step 1: 意图识别
            print(f"\n📨 用户 [{user_id}] 询问: {user_query}")

            intent = await self._detect_intent_with_timeout(user_query)
            if not intent:
                return {"status": "error", "message": "意图识别失败"}

            print(f"🎯 意图: {intent['type']} (置信度: {intent['confidence']:.2%})")

            # Step 2: 根据意图处理
            if intent['type'] == "complaint":
                # 投诉直接转人工
                return {"status": "escalate", "message": "转人工客服"}
            elif intent['type'] == "inquiry":
                # 询问：检索知识 + 生成答案
                knowledge = await self._retrieve_knowledge(user_query)
                if knowledge:
                    answer = await self._generate_answer(user_query, knowledge)
                    return {"status": "success", "answer": answer}
                else:
                    return {"status": "escalate", "message": "转人工客服"}
            else:
                # 其他：简单回复
                return {"status": "success", "answer": "我已收到您的消息"}

        except asyncio.TimeoutError:
            elapsed = asyncio.get_event_loop().time() - start_time
            print(f"⏰ 超时: {elapsed:.2f} 秒")
            return {"status": "timeout", "message": "处理超时，请稍后重试"}
        except Exception as e:
            print(f"❌ 错误: {e}")
            return {"status": "error", "message": "系统错误"}

    async def _detect_intent_with_timeout(self, query: str) -> Optional[dict]:
        """带超时的意图识别"""
        try:
            return await asyncio.wait_for(
                self.intent_group.detect_intent(query),
                timeout=self.timeouts["intent_detection"]
            )
        except asyncio.TimeoutError:
            print("⏰ 意图识别超时")
            return None

    async def _retrieve_knowledge(self, query: str) -> Optional[dict]:
        """检索知识库"""
        try:
            return await asyncio.wait_for(
                self.knowledge_group.search(query),
                timeout=self.timeouts["knowledge_retrieval"]
            )
        except asyncio.TimeoutError:
            print("⏰ 知识检索超时")
            return None

    async def _generate_answer(self, query: str, knowledge: dict) -> Optional[str]:
        """生成答案"""
        try:
            return await asyncio.wait_for(
                self.answer_group.generate(query, knowledge),
                timeout=self.timeouts["answer_generation"]
            )
        except asyncio.TimeoutError:
            print("⏰ 答案生成超时")
            return None

class IntentAgent:
    """意图识别 Agent"""
    def __init__(self, agent_id: str, model: str):
        self.id = agent_id
        self.model = model

    async def detect(self, query: str) -> dict:
        """检测意图"""
        # 模拟 LLM 调用
        print(f"  🔍 [{self.id}] 分析: {query[:20]}...")

        # 简化版意图识别规则
        if "投诉" in query or "不爽" in query or "差" in query:
            return {"type": "complaint", "confidence": 0.9}
        elif "?" in query or "如何" in query or "怎么" in query:
            return {"type": "inquiry", "confidence": 0.85}
        else:
            return {"type": "other", "confidence": 0.7}

class IntentSwarmGroup:
    """意图识别协作组"""
    def __init__(self):
        self.agents = [
            IntentAgent("intent1", "gpt-4"),
            IntentAgent("intent2", "claude-3"),
            IntentAgent("intent3", "glm-4")
        ]

    async def detect_intent(self, query: str) -> dict:
        """协作检测意图（投票机制）"""
        print("\n🤖 意图识别协作组启动...")

        # 并行调用所有 Agent
        tasks = [agent.detect(query) for agent in self.agents]
        results = await asyncio.gather(*tasks)

        # 投票：选择置信度最高的结果
        best_result = max(results, key=lambda x: x["confidence"])

        # 计算平均置信度
        avg_confidence = sum(r["confidence"] for r in results) / len(results)
        print(f"  📊 平均置信度: {avg_confidence:.2%}")

        # 如果平均置信度太低，返回不确定
        if avg_confidence < 0.7:
            return {"type": "unknown", "confidence": avg_confidence}

        return best_result

class KnowledgeAgent:
    """知识检索 Agent"""
    def __init__(self, agent_id: str, data_source: str):
        self.id = agent_id
        self.data_source = data_source

    async def search(self, query: str) -> dict:
        """搜索知识"""
        print(f"  🔍 [{self.id}] 在 {self.data_source} 搜索...")

        # 模拟检索结果
        # 实际中这里会调用向量数据库、搜索引擎等
        if "退款" in query or "退货" in query:
            return {
                "source": self.data_source,
                "content": "退款政策：商品支持 7 天无理由退货",
                "relevance": 0.9
            }
        elif "配送" in query or "快递" in query:
            return {
                "source": self.data_source,
                "content": "配送时间：市区 24 小时，外省 48-72 小时",
                "relevance": 0.85
            }
        else:
            return {
                "source": self.data_source,
                "content": "未找到相关信息",
                "relevance": 0.3
            }

class KnowledgeSwarmGroup:
    """知识检索协作组"""
    def __init__(self):
        self.agents = [
            KnowledgeAgent("kb1", "知识库"),
            KnowledgeAgent("kb2", "FAQ"),
            KnowledgeAgent("kb3", "历史对话")
        ]

    async def search(self, query: str) -> Optional[dict]:
        """协作搜索知识"""
        print("\n📚 知识检索协作组启动...")

        # 并行搜索
        tasks = [agent.search(query) for agent in self.agents]
        results = await asyncio.gather(*tasks)

        # 选择相关性最高的结果
        best_result = max(results, key=lambda x: x["relevance"])

        if best_result["relevance"] > 0.7:
            print(f"  ✅ 找到知识: {best_result['source']}")
            return best_result
        else:
            print(f"  ⚠️ 未找到相关知识")
            return None

class AnswerAgent:
    """答案生成 Agent"""
    def __init__(self, agent_id: str, style: str):
        self.id = agent_id
        self.style = style  # formal, casual, concise

    async def generate(self, query: str, knowledge: dict) -> str:
        """生成答案"""
        print(f"  ✍️ [{self.id}] 生成答案（{self.style}风格）...")

        # 模拟 LLM 生成
        knowledge_text = knowledge["content"]

        if self.style == "formal":
            answer = f"尊敬的用户，关于您的问题"{query}"，根据我们的{knowledge['source']}，{knowledge_text}。"
        elif self.style == "casual":
            answer = f"您好！关于"{query}"的问题，我们查了一下，{knowledge_text}~"
        else:  # concise
            answer = f"{query}：{knowledge_text}。"

        return answer

class AnswerSwarmGroup:
    """答案生成协作组"""
    def __init__(self):
        self.agents = [
            AnswerAgent("ans1", "formal"),
            AnswerAgent("ans2", "casual"),
            AnswerAgent("ans3", "concise")
        ]

    async def generate(self, query: str, knowledge: dict) -> Optional[str]:
        """协作生成答案"""
        print("\n💬 答案生成协作组启动...")

        # 选择生成风格（这里简化为默认选 formal）
        # 实际中可以根据用户画像、上下文等动态选择
        agent = self.agents[0]  # formal

        answer = await agent.generate(query, knowledge)
        print(f"  ✅ 生成答案: {answer[:50]}...")

        return answer

# 完整运行示例
async def main():
    # 创建协作组
    intent_group = IntentSwarmGroup()
    knowledge_group = KnowledgeSwarmGroup()
    answer_group = AnswerSwarmGroup()

    # 创建编排器
    orchestrator = CustomerServiceOrchestrator()
    orchestrator.setup_groups(intent_group, knowledge_group, answer_group)

    # 模拟用户请求
    queries = [
        ("用户001", "我想退款，请问怎么操作？"),
        ("用户002", "你们这个产品太差了！"),
        ("用户003", "多久能收到货？"),
    ]

    # 处理请求
    for user_id, query in queries:
        print("\n" + "="*60)
        result = await orchestrator.process(query, user_id)

        if result["status"] == "success":
            print(f"\n💬 回复: {result['answer']}")
        elif result["status"] == "escalate":
            print(f"\n👨‍💼 {result['message']}")
        else:
            print(f"\n⚠️ {result['message']}")

# 运行
asyncio.run(main())

============================================================
📨 用户 [用户001] 询问: 我想退款，请问怎么操作？

🤖 意图识别协作组启动...
  🔍 [intent1] 分析: 我想退款，请问怎么操作？...
  🔍 [intent2] 分析: 我想退款，请问怎么操作？...
  🔍 [intent3] 分析: 我想退款，请问怎么操作？...
📊 平均置信度: 85.00%
🎯 意图: inquiry (置信度: 85%)

📚 知识检索协作组启动...
  🔍 [kb1] 在 知识库 搜索...
  🔍 [kb2] 在 FAQ 搜索...
  🔍 [kb3] 在 历史对话 搜索...
  ✅ 找到知识: 知识库

💬 答案生成协作组启动...
  ✍️ [ans1] 生成答案（formal风格）...
  ✅ 生成答案: 尊敬的用户，关于您的问题"我想退款，请问怎么操作？...

💬 回复: 尊敬的用户，关于您的问题"我想退款，请问怎么操作？"，根据我们的知识库，退款政策：商品支持 7 天无理由退货。

import logging
import json
from datetime import datetime

class AgentLogger:
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.logger = logging.getLogger(f"Agent.{agent_id}")

    def log_decision(self, task: str, decision: dict, reasoning: str):
        """记录决策过程"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "agent_id": self.agent_id,
            "task": task,
            "decision": decision,
            "reasoning": reasoning
        }
        self.logger.info(json.dumps(log_entry, ensure_ascii=False))

    def log_performance(self, task: str, duration_ms: int, token_count: int):
        """记录性能"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "agent_id": self.agent_id,
            "task": task,
            "duration_ms": duration_ms,
            "token_count": token_count
        }
        self.logger.info(json.dumps(log_entry, ensure_ascii=False))


# 使用示例
logger = AgentLogger("agent1")
logger.log_decision(
    task="意图识别",
    decision={"type": "inquiry", "confidence": 0.85},
    reasoning="用户问题包含 '?' 和 '如何'，符合询问模式"
)
logger.log_performance(
    task="意图识别",
    duration_ms=1200,
    token_count=450
)

class BadAgent:
    """什么都做 - 违反单一职责"""
    def process(self, request):
        intent = self.detect_intent(request)
        knowledge = self.search_knowledge(request)
        answer = self.generate_answer(request, knowledge)
        return answer

# 拆分成 3 个 Agent
class IntentAgent:
    def detect(self, request): ...

class KnowledgeAgent:
    def search(self, request): ...

class AnswerAgent:
    def generate(self, request, knowledge): ...

result = agent1.process(request)
result = agent2.process(result)  # 等待 agent1
result = agent3.process(result)  # 等待 agent2

tasks = [agent1.process, agent2.process, agent3.process]
results = await asyncio.gather(*[t(request) for t in tasks])

async def with_fallback(primary, fallback, task):
    """带降级的执行"""
    try:
        return await asyncio.wait_for(primary(task), timeout=3)
    except (asyncio.TimeoutError, Exception) as e:
        print(f"⚠️ 主方法失败，使用降级方案: {e}")
        return await fallback(task)

@log_performance
@log_decision
async def process(self, request):
    """自动记录性能和决策"""
    result = await self._do_process(request)
    return result

V1: 单 Agent（验证可行性）
V2: 2-3 个 Agent（简单协作）
V3: 根据实际需求扩展

# 所有 Agent 调用都必须带超时
try:
    result = await asyncio.wait_for(agent.process(task), timeout=3)
except asyncio.TimeoutError:
    return fallback_result

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.last_failure_time = None

    async def call(self, func, *args):
        """带熔断保护的调用"""
        if self.is_open():
            raise Exception("熔断器已打开，拒绝调用")

        try:
            result = await func(*args)
            self.reset()
            return result
        except Exception as e:
            self.record_failure()
            raise e

    def is_open(self) -> bool:
        """检查熔断器是否打开"""
        if self.failure_count >= self.failure_threshold:
            if time.time() - self.last_failure_time > self.timeout:
                self.reset()  # 超时后尝试恢复
                return False
            return True
        return False

    def record_failure(self):
        """记录失败"""
        self.failure_count += 1
        self.last_failure_time = time.time()

    def reset(self):
        """重置熔断器"""
        self.failure_count = 0
        self.last_failure_time = None

import uuid

def with_trace_id(func):
    """自动添加 trace_id"""
    async def wrapper(*args, **kwargs):
        trace_id = kwargs.get("trace_id") or str(uuid.uuid4())
        kwargs["trace_id"] = trace_id
        print(f"🔍 [{trace_id}] 调用: {func.__name__}")
        return await func(*args, **kwargs)
    return wrapper

@with_trace_id
async def agent1_process(request, trace_id=None):
    # ...
    pass

class DecisionVisualizer:
    def __init__(self):
        self.decisions = []

    def record(self, agent_id: str, task: str, options: list, selected: str, reasoning: str):
        """记录决策"""
        self.decisions.append({
            "agent_id": agent_id,
            "task": task,
            "options": options,
            "selected": selected,
            "reasoning": reasoning
        })

    def visualize(self):
        """可视化决策树"""
        for i, decision in enumerate(self.decisions, 1):
            print(f"\n[{i}] Agent: {decision['agent_id']}")
            print(f"    任务: {decision['task']}")
            print(f"    选项: {', '.join(decision['options'])}")
            print(f"    选择: {decision['selected']} ⬅️")
            print(f"    原因: {decision['reasoning']}")


# 使用示例
visualizer = DecisionVisualizer()
visualizer.record(
    agent_id="agent1",
    task="选择下一个步骤",
    options=["搜索", "生成", "分析"],
    selected="搜索",
    reasoning="用户问题包含具体关键词，需要先获取信息"
)
visualizer.visualize()

class ABTestAgent:
    def __init__(self, version_a, version_b, split_ratio=0.5):
        self.version_a = version_a
        self.version_b = version_b
        self.split_ratio = split_ratio

    async def process(self, task):
        """A/B 测试"""
        import random
        if random.random() < self.split_ratio:
            print("🅰️ 使用版本 A")
            result = await self.version_a.process(task)
            result["version"] = "A"
        else:
            print("🅱️ 使用版本 B")
            result = await self.version_b.process(task)
            result["version"] = "B"

        return result