多 Agent 协作与生产部署

┌─────────────┐
│  主管 Agent  │ ← 任务分解、结果汇总
├─────────────┤
│  ┌───┐ ┌───┐ │
│  │ A │ │ B │ │ ← 各下属 Agent 专注执行
│  └───┘ └───┘ │
└─────────────┘

class ManagerAgent:
    """主管 Agent：负责任务分解和结果整合"""

    def __init__(self, llm, workers: Dict[str, 'WorkerAgent']):
        self.llm = llm
        self.workers = workers

    def execute(self, task: str) -> str:
        # 1. 分解任务
        subtasks = self._decompose(task)

        # 2. 分配子任务
        results = {}
        for worker_name, subtask in subtasks.items():
            worker = self.workers.get(worker_name)
            if worker:
                results[worker_name] = worker.execute(subtask)

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

    def _decompose(self, task: str) -> Dict[str, str]:
        """将任务分解为子任务分配给各 Worker"""
        response = self.llm.chat(
            f"将以下任务分解，返回 JSON 格式：\n"
            f"{{\"worker_A\": \"子任务描述\", \"worker_B\": ...}}\n"
            f"可用 Worker: {list(self.workers.keys())}\n"
            f"任务：{task}"
        )
        return json.loads(response)

class DebateTeam:
    """辩论模式：多角度分析"""

    def __init__(self, agents: List[Agent], moderator: Agent):
        self.agents = agents
        self.moderator = moderator

    def debate(self, question: str, rounds: int = 3) -> str:
        opinions = []

        # 第一轮：各自发表观点
        for agent in self.agents:
            opinions.append({
                "agent": agent.name,
                "view": agent.analyze(question)
            })

        # 后续轮次：互相反驳和完善
        for round_num in range(1, rounds):
            for i, agent in enumerate(self.agents):
                others_views = [
                    o["view"] for j, o in enumerate(opinions)
                    if j != i and len(o["view"]) > 0
                ]
                rebuttal = agent.rebuttal(question, others_views)
                opinions[i]["view"] += f"\n[反驳] {rebuttal}"

        # 主持人总结
        return self.moderator.summarize(question, opinions)

Agent A（分析） → Agent B（设计） → Agent C（实现） → Agent D（测试）

class PipelineOrchestrator:
    """流水线编排"""

    def __init__(self, stages: List[Tuple[str, Agent]]):
        self.stages = stages

    def execute(self, input_data: Any) -> Any:
        current = input_data
        for stage_name, agent in self.stages:
            print(f"▶️ 阶段: {stage_name}")
            current = agent.process(current)
            if self._validate_output(stage_name, current):
                continue
            else:
                print(f"⚠️ {stage_name} 输出验证失败，重新执行")
                current = agent.process(current)
        return current

@dataclass
class AgentMessage:
    """Agent 间通信消息"""
    sender: str
    receiver: str
    msg_type: str  # "request" | "response" | "broadcast"
    content: Any
    metadata: Dict = field(default_factory=dict)
    timestamp: float = field(default_factory=time.time)
    msg_id: str = field(default_factory=lambda: str(uuid.uuid4()))

class MessageBus:
    """消息总线：解耦 Agent 间通信"""

    def __init__(self):
        self.subscribers: Dict[str, List[Callable]] = {}
        self.message_history: List[AgentMessage] = []

    def subscribe(self, agent_name: str, callback: Callable):
        if agent_name not in self.subscribers:
            self.subscribers[agent_name] = []
        self.subscribers[agent_name].append(callback)

    def publish(self, message: AgentMessage):
        self.message_history.append(message)

        # 投递到指定接收者
        if message.receiver in self.subscribers:
            for cb in self.subscribers[message.receiver]:
                cb(message)

        # 广播
        if message.msg_type == "broadcast":
            for agent_name, callbacks in self.subscribers.items():
                if agent_name != message.sender:
                    for cb in callbacks:
                        cb(message)

class HumanApproval:
    """高危操作的人工审批"""

    RISKY_OPERATIONS = [
        "delete_resource", "modify_permissions",
        "execute_shell", "transfer_funds"
    ]

    def execute_with_approval(self, tool_name: str, tool_args: Dict):
        if tool_name in self.RISKY_OPERATIONS:
            # 暂停等待人工审批
            approval = self._request_approval(tool_name, tool_args)
            if not approval:
                return {"error": "操作被人工拒绝"}

        return self.tools[tool_name](**tool_args)

┌─────────┐    ┌──────────┐    ┌──────────┐
│ 客户端   │───▶│ API 网关 │───▶│ 任务队列  │
└─────────┘    └──────────┘    └──────────┘
                                    │
                          ┌─────────▼─────────┐
                          │   Agent 工作节点池  │
                          │  ┌───┐ ┌───┐ ┌───┐ │
                          │  │ A │ │ B │ │ C │ │
                          │  └───┘ └───┘ └───┘ │
                          └─────────┬─────────┘
                                    │
                    ┌───────────────┼───────────────┐
                    │               │               │
              ┌─────▼───┐   ┌──────▼────┐   ┌─────▼───┐
              │ 向量 DB  │   │   工具 API │   │   监控   │
              └─────────┘   └───────────┘   └─────────┘

# agent_config.yaml
agent:
  max_concurrent_tasks: 10
  task_timeout: 300  # 秒
  retry_policy:
    max_retries: 3
    backoff_base: 2
  rate_limits:
    llm_calls: 100  # 每分钟
    tool_calls: 50   # 每分钟

monitoring:
  metrics:
    - response_time
    - token_usage
    - tool_success_rate
    - agent_error_rate
  logging:
    level: INFO
    retention_days: 30
    sensitive_fields:
      - api_key
      - password
      - token

import prometheus_client as prom

# 定义指标
agent_requests = prom.Counter(
    'agent_requests_total',
    'Agent 请求总数',
    ['agent_name', 'status']
)

agent_latency = prom.Histogram(
    'agent_task_duration_seconds',
    '任务执行耗时分布',
    ['agent_name'],
    buckets=[1, 5, 10, 30, 60, 120, 300]
)

token_usage = prom.Counter(
    'agent_token_usage_total',
    'Token 消耗总数',
    ['model', 'type']  # type: input/output
)

tool_calls = prom.Counter(
    'agent_tool_calls_total',
    '工具调用次数',
    ['tool_name', 'status']
)