前言

2024年到2025年，是AI Agent从"玩具演示"走向"生产可用"的关键窗口期。早期大家热衷于用LangChain搭一个单Agent链（Chain），跑跑Demo感觉还不错；真正上了生产环境才发现：单Agent在复杂任务面前力不从心——上下文窗口不够用、任务路由不灵活、多工具调用冲突、无法横向扩展。

于是，多智能体协同（Multi-Agent Collaboration）成了2026年企业级AI落地的核心话题。但多Agent不是简单地把多个单Agent拼在一起，它涉及任务分解与路由、共享状态管理、冲突仲裁、资源调度等复杂工程问题。

本文将围绕这些问题，从架构演进讲起，给出完整的Python实现，最终形成一套可落地的多Agent协作框架。

---

一、企业级Agent架构设计：从单体到多智能体

1.1 单体Agent的天花板

一个典型单体Agent的结构如下：

用户输入 → LLM（规划） → 工具调用（执行） → 输出

它的工作流程是：接收任务 → LLM思考 → 选择工具 → 执行 → 反馈。这个模式在简单问答、单一工具调用场景下表现尚可，但面临三个根本瓶颈：

瓶颈	表现	后果
上下文膨胀	多轮对话后token暴增，推理质量下降	长任务崩溃
能力边界	单一Agent无法同时精通多个领域	专业任务出错率飙升
串行执行	所有步骤排队等待LLM逐个处理	响应延迟高，吞吐低
状态丢失	无持久化机制，重启即失忆	无法做复杂多阶段任务

举一个具体场景：自动生成一份竞品分析报告。单Agent需要同时完成：搜索竞品信息、读取财报数据、调用API获取市场份额、调用绘图工具生成图表、调用文档工具写报告。这超出了任何单一Agent的能力边界。

1.2 多智能体架构的演进路径

企业级多Agent架构通常经历三个阶段：

阶段一：水平分域（Horizontal Decomposition）

用户请求
    │
    ├── 规划Agent（Router）── 拆解任务
    │       ├── 搜索Agent ──→ 获取数据
    │       ├── 分析Agent ──→ 处理数据
    │       └── 写作Agent ──→ 生成报告

每个Agent负责一个领域能力，通过中央路由分发任务。这是目前最主流的架构模式。

阶段二：垂直管道（Vertical Pipeline）

请求 → 过滤Agent → 增强Agent → 执行Agent → 验证Agent → 响应

类似流水线，每个Agent处理完成后交给下一个，强调顺序协作和质量门禁。

阶段三：动态网格（Dynamic Mesh）

所有Agent互联，
通过消息总线通信，
根据任务动态形成协作拓扑

这是最复杂的模式，适合超大型组织，典型的例子是AutoGen、MCP协议等支持的多Agent网络。

本文重点讲解阶段一（水平分域），因为这是大多数企业落地多Agent的起点，也是最实用的架构。

1.3 核心架构设计原则

在设计企业级多Agent系统时，有几个设计原则必须遵守：

1. 角色边界清晰（Role Isolation）
每个Agent只做一件事，并且把这件事做到极致。避免"全能型Agent"——它什么都做，但什么都做不好。

2. 中央协调与去中心化结合
用Router/Orchestrator做全局协调（强中心），各子Agent独立决策（弱去中心化）。纯去中心化会导致系统不可预测。

3. 状态外部化（State Externalization）
Agent的中间状态不要存在内存里，要写外部存储（Redis、PostgreSQL）。否则无法做断点恢复、审计回溯。

4. 超时与熔断
每个Agent调用都要设置超时，多个Agent并行时要考虑熔断降级。不能让一个Agent把整个系统拖死。

---

二、Multi-Agent协同核心机制

2.1 任务分解与路由

任务分解是多Agent系统的第一步。核心问题：如何把一个复杂请求，变成一组可执行的子任务？

有两种主流方法：

方法A：LLM驱动的任务分解
直接让LLM分析用户请求，输出任务列表。这种方法灵活，但依赖模型能力。

方法B：Schema驱动的任务分解
预定义任务类型和对应的处理流程，用Schema约束分解结果。这种方法稳定，适合可枚举的场景。

实践中推荐混合方式：Schema做边界约束，LLM做语义理解。

from enum import Enum
from typing import Optional
from dataclasses import dataclass, field
import json


class TaskType(Enum):
    """可枚举的任务类型"""
    SEARCH = "search"
    DATA_ANALYSIS = "data_analysis"
    CODE_EXECUTION = "code_execution"
    DOCUMENT_WRITING = "document_writing"
    IMAGE_GENERATION = "image_generation"
    API_CALL = "api_call"
    FILE_OPERATION = "file_operation"
    ROUTING_ONLY = "routing_only"


@dataclass
class Task:
    """任务单元"""
    task_id: str
    task_type: TaskType
    description: str
    dependencies: list[str] = field(default_factory=list)  # 依赖的任务ID列表
    priority: int = 0  # 0=低, 1=中, 2=高
    timeout_seconds: int = 60
    metadata: dict = field(default_factory=dict)

    def to_dict(self) -> dict:
        return {
            "task_id": self.task_id,
            "task_type": self.task_type.value,
            "description": self.description,
            "dependencies": self.dependencies,
            "priority": self.priority,
            "timeout_seconds": self.timeout_seconds,
            "metadata": self.metadata
        }


@dataclass
class DecomposedPlan:
    """分解后的执行计划"""
    tasks: list[Task]
    execution_mode: str  # "parallel" | "sequential" | "hybrid"
    estimated_duration_seconds: float

    def get_executable_tasks(self) -> list[Task]:
        """返回无依赖或依赖已解决的任务（可并行执行）"""
        completed_ids = set()
        executable = []
        for task in self.tasks:
            if not task.dependencies or all(d in completed_ids for d in task.dependencies):
                executable.append(task)
        return executable

    def mark_completed(self, task_id: str):
        """标记任务完成，驱动后续任务"""
        pass  # 由执行器调用


class TaskDecomposer:
    """任务分解器：混合Schema+LLM驱动"""

    # 预定义的任务模板
    TASK_TEMPLATES = {
        "竞品分析": [
            (TaskType.SEARCH, "搜索竞品基础信息", 2),
            (TaskType.API_CALL, "获取市场份额数据", 2),
            (TaskType.DATA_ANALYSIS, "分析对比数据", 1),
            (TaskType.DOCUMENT_WRITING, "撰写分析报告", 1),
        ],
        "技术方案设计": [
            (TaskType.CODE_EXECUTION, "架构设计验证", 1),
            (TaskType.FILE_OPERATION, "生成设计文档", 1),
            (TaskType.DOCUMENT_WRITING, "编写详细方案", 1),
        ],
    }

    def __init__(self, llm_client=None):
        self.llm = llm_client  # 注入LLM客户端（可以是OpenAI、Claude等）

    def decompose(self, user_request: str, domain: Optional[str] = None) -> DecomposedPlan:
        """
        分解用户请求，返回可执行计划
        """
        # Step 1: Schema匹配（快速路径）
        for template_name, task_list in self.TASK_TEMPLATES.items():
            if template_name in user_request:
                tasks = self._build_tasks_from_template(task_list, user_request)
                return DecomposedPlan(
                    tasks=tasks,
                    execution_mode="hybrid",
                    estimated_duration_seconds=sum(t.timeout_seconds for t in tasks)
                )

        # Step 2: LLM驱动分解（通用路径）
        if self.llm:
            return self._llm_decompose(user_request)

        # Step 3: 兜底：基础路由
        return self._fallback_decompose(user_request)

    def _build_tasks_from_template(self, task_specs: list, user_request: str) -> list[Task]:
        """从模板构建任务"""
        import uuid
        tasks = []
        for i, (task_type, desc, priority) in enumerate(task_specs):
            task = Task(
                task_id=str(uuid.uuid4())[:8],
                task_type=task_type,
                description=f"{desc}：{user_request[:50]}",
                priority=priority,
                timeout_seconds=60,
                metadata={"source": "template"}
            )
            tasks.append(task)
        return tasks

    def _llm_decompose(self, user_request: str) -> DecomposedPlan:
        """LLM驱动的任务分解"""
        prompt = f"""分析以下用户请求，将其分解为可执行的子任务。
用户请求：{user_request}

要求：
1. 每个子任务必须是原子性的（单一职责）
2. 标注任务之间的依赖关系
3. 评估任务优先级（0=低, 1=中, 2=高）
4. 输出JSON格式

输出格式：
{{
  "tasks": [
    {{
      "task_id": "唯一ID",
      "task_type": "search|data_analysis|code_execution|document_writing|...",
      "description": "任务描述",
      "dependencies": ["依赖的任务ID"],
      "priority": 0-2,
      "timeout_seconds": 60
    }}
  ],
  "execution_mode": "parallel|sequential|hybrid"
}}
"""
        # 调用LLM（简化实现）
        response = self.llm.chat([{"role": "user", "content": prompt}])
        result = json.loads(response)

        tasks = [
            Task(
                task_id=t["task_id"],
                task_type=TaskType(t["task_type"]),
                description=t["description"],
                dependencies=t.get("dependencies", []),
                priority=t.get("priority", 0),
                timeout_seconds=t.get("timeout_seconds", 60),
                metadata={"source": "llm"}
            )
            for t in result["tasks"]
        ]
        return DecomposedPlan(
            tasks=tasks,
            execution_mode=result.get("execution_mode", "hybrid"),
            estimated_duration_seconds=sum(t.timeout_seconds for t in tasks)
        )

    def _fallback_decompose(self, user_request: str) -> DecomposedPlan:
        """兜底分解策略"""
        task = Task(
            task_id="fallback-001",
            task_type=TaskType.ROUTING_ONLY,
            description=user_request,
            priority=1
        )
        return DecomposedPlan(
            tasks=[task],
            execution_mode="sequential",
            estimated_duration_seconds=30
        )

2.2 状态共享机制

多Agent协作的核心挑战之一是状态共享。每个Agent是独立的工作单元，但它们需要共享中间结果、上下文信息。

三种主流状态共享方案：

方案1：共享内存（Shared Memory）
用Python的multiprocessing或threading共享内存。速度快，但不适合跨进程、跨机器场景。

方案2：消息总线（Message Bus）
每个Agent通过消息队列通信（Kafka、Redis Pub/Sub）。解耦彻底，适合分布式系统。

方案3：中央状态存储（Central State Store）
用Redis或PostgreSQL作为"黑板"，所有Agent读写共享状态。最稳定，适合企业级系统。

本文采用方案3 + 方案2混合：用Redis做实时消息传递+结果缓存，用PostgreSQL做持久化状态存储。

import redis
import json
import time
from typing import Any, Optional
from dataclasses import dataclass, asdict
from datetime import datetime
import uuid


@dataclass
class SharedState:
    """共享状态条目"""
    key: str
    value: Any
    agent_id: str  # 谁写的
    timestamp: float
    version: int
    ttl_seconds: Optional[int] = None

    def to_json(self) -> str:
        return json.dumps({
            "key": self.key,
            "value": self.value,
            "agent_id": self.agent_id,
            "timestamp": self.timestamp,
            "version": self.version,
            "ttl_seconds": self.ttl_seconds
        })

    @classmethod
    def from_json(cls, data: str) -> "SharedState":
        obj = json.loads(data)
        return cls(**obj)


class SharedStateStore:
    """
    中央状态存储：Redis（实时）+ PostgreSQL（持久化）
    企业级多Agent系统推荐使用此模式
    """

    def __init__(
        self,
        redis_host: str = "localhost",
        redis_port: int = 6379,
        redis_db: int = 0,
        pg_conn=None  # psycopg2 connection
    ):
        self.redis = redis.Redis(
            host=redis_host,
            port=redis_port,
            db=redis_db,
            decode_responses=True
        )
        self.pg_conn = pg_conn
        self.local_cache: dict[str, SharedState] = {}

    def write(
        self,
        key: str,
        value: Any,
        agent_id: str,
        ttl_seconds: Optional[int] = 3600
    ) -> SharedState:
        """
        写入共享状态
        写入Redis（实时）+ 异步写入PostgreSQL（持久化）
        """
        state = SharedState(
            key=key,
            value=value,
            agent_id=agent_id,
            timestamp=time.time(),
            version=self._get_next_version(key),
            ttl_seconds=ttl_seconds
        )

        # 写Redis
        redis_key = f"agent_state:{key}"
        self.redis.setex(
            redis_key,
            ttl_seconds or 3600,
            state.to_json()
        )

        # 写本地缓存
        self.local_cache[key] = state

        # 写PostgreSQL（异步）
        if self.pg_conn:
            self._persist_to_pg(state)

        return state

    def read(self, key: str) -> Optional[SharedState]:
        """读取共享状态，优先Redis"""
        # 本地缓存命中检查
        if key in self.local_cache:
            return self.local_cache[key]

        redis_key = f"agent_state:{key}"
        data = self.redis.get(redis_key)
        if data:
            state = SharedState.from_json(data)
            self.local_cache[key] = state
            return state

        return None

    def read_batch(self, keys: list[str]) -> dict[str, SharedState]:
        """批量读取"""
        result = {}
        for key in keys:
            state = self.read(key)
            if state:
                result[key] = state
        return result

    def broadcast(self, channel: str, message: dict, agent_id: str):
        """
        发布消息到指定频道（用于Agent间实时通知）
        """
        event = {
            "agent_id": agent_id,
            "timestamp": time.time(),
            "message": message
        }
        self.redis.publish(channel, json.dumps(event))

    def subscribe(self, channel: str) -> redis.client.PubSub:
        """订阅频道"""
        pubsub = self.redis.pubsub()
        pubsub.subscribe(channel)
        return pubsub

    def _get_next_version(self, key: str) -> int:
        """原子递增版本号"""
        version_key = f"agent_state_version:{key}"
        return self.redis.incr(version_key)

    def _persist_to_pg(self, state: SharedState):
        """持久化到PostgreSQL"""
        try:
            cursor = self.pg_conn.cursor()
            cursor.execute("""
                INSERT INTO agent_states (key, value, agent_id, timestamp, version)
                VALUES (%s, %s, %s, %s, %s)
                ON CONFLICT (key) DO UPDATE SET
                    value = EXCLUDED.value,
                    agent_id = EXCLUDED.agent_id,
                    timestamp = EXCLUDED.timestamp,
                    version = EXCLUDED.version
            """, (
                state.key,
                json.dumps(state.value),
                state.agent_id,
                datetime.fromtimestamp(state.timestamp),
                state.version
            ))
            self.pg_conn.commit()
            cursor.close()
        except Exception as e:\n            print(f"PG persist failed: {e}")
            self.pg_conn.rollback()

    def get_task_context(
        self,
        task_id: str,
        requesting_agent: str,
        include_related: bool = True
    ) -> dict:
        """
        获取任务上下文：读取所有相关共享状态
        这是Agent执行任务前的标准准备步骤
        """
        # 读取任务主状态
        main_state = self.read(f"task:{task_id}")
        context = {
            "task_id": task_id,
            "main": main_state.value if main_state else None,
            "artifacts": {},
            "history": []
        }

        if not include_related:
            return context

        # 读取所有子任务的产物
        pattern = f"agent_state:subtask:{task_id}:*"
        keys = self.redis.keys(pattern)
        for key in keys:
            data = self.redis.get(key)
            if data:
                state = SharedState.from_json(data)
                artifact_key = key.replace("agent_state:", "")
                context["artifacts"][artifact_key] = state.value

        # 读取历史记录（最近的20条）
        if self.pg_conn:
            cursor = self.pg_conn.cursor()
            cursor.execute("""
                SELECT agent_id, value, timestamp
                FROM agent_states
                WHERE key LIKE %s
                ORDER BY timestamp DESC
                LIMIT 20
            """, (f"subtask:{task_id}:%",))
            for row in cursor.fetchall():
                context["history"].append({
                    "agent_id": row[0],
                    "value": json.loads(row[1]),
                    "timestamp": row[2].timestamp()
                })
            cursor.close()

        return context

2.3 冲突仲裁机制

当多个Agent同时访问/修改同一资源，或者对同一问题给出了不同结论时，就需要冲突仲裁。

冲突类型和处理策略：

冲突类型	策略	适用场景
资源竞争	悲观锁/乐观锁	写冲突
结论分歧	投票/权威Agent裁决	分析类任务
优先级冲突	优先级队列	任务调度
工具调用冲突	超时等待 + 降级	并发工具调用

import threading
from enum import Enum
from typing import Callable, Optional
import time


class ConflictResolution(Enum):
    VOTE = "vote"           # 投票裁决
    AUTHORITY = "authority"  # 权威Agent裁决
    TIMEOUT_WAIT = "timeout_wait"  # 超时等待
    LAST_WRITE_WINS = "last_write_wins"  # 最后写入胜出
    MERGE = "merge"         # 合并策略


@dataclass
class ConflictReport:
    """冲突报告"""
    conflict_id: str
    conflict_type: str
    involved_agents: list[str]
    options: list[Any]  # 各方提供的选项
    resolution: Optional[str] = None
    winner: Optional[Any] = None


class ConflictResolver:
    """
    冲突仲裁器
    企业级实现：支持多种仲裁策略
    """

    def __init__(self, state_store: SharedStateStore):
        self.state_store = state_store
        self.locks: dict[str, threading.Lock] = {}
        self.conflict_history: list[ConflictReport] = []
        self._lock = threading.Lock()

    def acquire_resource_lock(
        self,
        resource_id: str,
        agent_id: str,
        timeout_seconds: float = 10
    ) -> bool:
        """
        获取资源锁（悲观锁）
        用于防止多Agent同时写入同一资源
        """
        lock_key = f"lock:{resource_id}"

        # 原子性获取锁
        acquired = self.state_store.redis.set(
            lock_key,
            agent_id,
            nx=True,  # 仅当不存在时设置
            ex=int(timeout_seconds)
        )

        if acquired:
            with self._lock:
                self.locks[resource_id] = threading.Lock()
            return True

        # 等待锁释放
        start = time.time()
        while time.time() - start < timeout_seconds:
            if self.state_store.redis.get(lock_key) is None:
                return self.acquire_resource_lock(resource_id, agent_id, timeout_seconds - (time.time() - start))
            time.sleep(0.1)

        return False

    def release_resource_lock(self, resource_id: str, agent_id: str):
        """释放资源锁"""
        lock_key = f"lock:{resource_id}"
        current_holder = self.state_store.redis.get(lock_key)
        if current_holder == agent_id:
            self.state_store.redis.delete(lock_key)

    def resolve_by_vote(
        self,
        options: list[tuple[str, Any]],  # [(agent_id, conclusion), ...]
        task_id: str
    ) -> tuple[str, Any]:
        """
        投票裁决：当多个Agent给出不同结论时
        使用预训练的"裁判LLM"做最终裁决
        """
        if len(options) == 1:
            return options[0]

        # 统计投票
        vote_counts: dict[Any, list[str]] = {}
        for agent_id, conclusion in options:
            # 用结论的hash作为key（避免大对象做dict key）
            key = str(conclusion)[:200]  # 截断
            if key not in vote_counts:
                vote_counts[key] = []
            vote_counts[key].append(agent_id)

        # 多数票
        winner_conclusion = max(vote_counts, key=lambda k: len(vote_counts[k]))
        winning_agents = vote_counts[winner_conclusion]
        winning_agent = winning_agents[0]

        # 记录冲突
        report = ConflictReport(
            conflict_id=str(uuid.uuid4()),
            conflict_type="vote",
            involved_agents=[a for a, _ in options],
            options=[o for _, o in options],
            resolution=ConflictResolution.VOTE.value,
            winner=winner_conclusion
        )
        self._record_conflict(report)

        return winning_agent, winner_conclusion

    def resolve_by_authority(
        self,
        options: list[tuple[str, Any]],
        authority_agent: str,  # 权威Agent的ID
        task_id: str
    ) -> tuple[str, Any]:
        """
        权威裁决：指定某个Agent作为最终裁决者
        适用于：有明确主从关系的Agent网络
        """
        authority_conclusion = None
        for agent_id, conclusion in options:
            if agent_id == authority_agent:
                authority_conclusion = conclusion
                break

        if authority_conclusion is None:
            # 权威Agent未参与，使用投票兜底
            return self.resolve_by_vote(options, task_id)

        report = ConflictReport(
            conflict_id=str(uuid.uuid4()),
            conflict_type="authority",
            involved_agents=[a for a, _ in options],
            options=[o for _, o in options],
            resolution=ConflictResolution.AUTHORITY.value,
            winner=authority_conclusion
        )
        self._record_conflict(report)

        return authority_agent, authority_conclusion

    def _record_conflict(self, report: ConflictReport):
        """记录冲突到历史"""
        with self._lock:
            self.conflict_history.append(report)
            # 只保留最近1000条
            if len(self.conflict_history) > 1000:
                self.conflict_history = self.conflict_history[-1000:]

---

三、实战：Python代码实现多Agent协作框架

3.1 整体框架设计

先给出完整的框架结构图，然后逐步实现各组件：

┌─────────────────────────────────────────────────────┐
│                    Orchestrator                      │
│         (任务分解 + 路由 + 结果聚合)                  │
└────────────────────┬────────────────────────────────┘
                     │
     ┌───────────────┼───────────────┐
     ▼               ▼               ▼
┌─────────┐   ┌──────────┐   ┌────────────┐
│Search   │   │Analysis  │   │  Writing   │
│ Agent   │   │  Agent   │   │   Agent    │
└────┬────┘   └────┬─────┘   └─────┬──────┘
     │              │               │
     └──────────────┼───────────────┘
                    ▼
         ┌──────────────────┐
         │  SharedStateStore │
         │  (Redis+PostgreSQL)│
         └──────────────────┘

3.2 Agent基类实现

import asyncio
import uuid
import time
from abc import ABC, abstractmethod
from typing import Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class AgentStatus(Enum):
    IDLE = "idle"
    RUNNING = "running"
    WAITING = "waiting"
    COMPLETED = "completed"
    FAILED = "failed"
    TIMEOUT = "timeout"


@dataclass
class AgentResult:
    """Agent执行结果"""
    agent_id: str
    task_id: str
    status: AgentStatus
    output: Any = None
    error: Optional[str] = None
    execution_time_seconds: float = 0
    tokens_used: int = 0
    metadata: dict = field(default_factory=dict)


class BaseAgent(ABC):
    """
    Agent基类：定义所有Agent的通用接口和行为
    """

    def __init__(
        self,
        agent_id: str,
        name: str,
        description: str,
        state_store: "SharedStateStore",
        llm_client: Any = None,
        max_retries: int = 3,
        default_timeout: int = 60
    ):
        self.agent_id = agent_id
        self.name = name
        self.description = description
        self.state_store = state_store
        self.llm = llm_client
        self.max_retries = max_retries
        self.default_timeout = default_timeout
        self.status = AgentStatus.IDLE
        self._current_task: Optional[str] = None

    @abstractmethod
    async def execute(self, task: "Task", context: dict) -> AgentResult:
        """
        执行具体任务（子类必须实现）
        context包含：用户输入、前置任务结果、共享状态
        """
        pass

    async def run(
        self,
        task: "Task",
        context: dict,
        timeout: Optional[int] = None
    ) -> AgentResult:
        """
        Agent运行入口：包含重试、超时、状态管理
        """
        timeout = timeout or task.timeout_seconds or self.default_timeout
        self.status = AgentStatus.RUNNING
        self._current_task = task.task_id

        # 写入执行开始状态
        self.state_store.write(
            key=f"subtask:{task.task_id}:{self.agent_id}",
            value={"status": "started", "agent": self.name},
            agent_id=self.agent_id,
            ttl_seconds=timeout + 60
        )

        start_time = time.time()
        last_error = None

        for attempt in range(self.max_retries):
            try:
                result = await asyncio.wait_for(
                    self.execute(task, context),
                    timeout=timeout
                )
                result.execution_time_seconds = time.time() - start_time
                self.status = AgentStatus.COMPLETED

                # 写入结果
                self.state_store.write(
                    key=f"subtask:{task.task_id}:{self.agent_id}",
                    value={
                        "status": "completed",
                        "output": result.output,
                        "tokens_used": result.tokens_used
                    },
                    agent_id=self.agent_id
                )

                # 广播完成事件
                self.state_store.broadcast(
                    channel=f"task:{task.task_id}:events",
                    message={"event": "agent_completed", "agent": self.agent_id},
                    agent_id=self.agent_id
                )

                return result

            except asyncio.TimeoutError:
                self.status = AgentStatus.TIMEOUT
                last_error = f"Timeout after {timeout}s"
                logger.warning(f"[{self.name}] Attempt {attempt+1} timeout")

            except Exception as e:\n                last_error = str(e)\n                logger.error(f"[{self.name}] Attempt {attempt+1} failed: {e}")
                self.status = AgentStatus.FAILED

                if attempt < self.max_retries - 1:
                    await asyncio.sleep(2 ** attempt)  # 指数退避

        # 所有重试都失败了
        return AgentResult(
            agent_id=self.agent_id,
            task_id=task.task_id,
            status=self.status,
            error=last_error,
            execution_time_seconds=time.time() - start_time
        )

    def __repr__(self):
        return f"<{self.__class__.__name__} id={self.agent_id} name={self.name} status={self.status.value}>"

3.3 专用Agent实现

以下实现三个典型的专业Agent：搜索Agent、分析Agent、写作Agent。

import httpx
from bs4 import BeautifulSoup


class SearchAgent(BaseAgent):
    """
    搜索Agent：负责从互联网或内部知识库获取信息
    支持多种搜索源
    """

    def __init__(self, *args, search_api_key: Optional[str] = None, **kwargs):
        super().__init__(*args, **kwargs)
        self.search_api_key = search_api_key
        self.http_client = httpx.AsyncClient(timeout=30.0)

    async def execute(self, task: "Task", context: dict) -> AgentResult:
        query = task.description
        results = await self._search(query)

        # 整理搜索结果
        formatted_results = []
        for item in results[:10]:  # 最多取10条
            formatted_results.append({
                "title": item.get("title", ""),
                "url": item.get("url", ""),
                "snippet": item.get("snippet", ""),
                "source": item.get("source", "web")
            })

        return AgentResult(
            agent_id=self.agent_id,
            task_id=task.task_id,
            status=AgentStatus.COMPLETED,
            output={
                "query": query,
                "total_results": len(results),
                "items": formatted_results,
                "summary": self._summarize_results(formatted_results)
            },
            metadata={"search_engine": "default"}
        )

    async def _search(self, query: str) -> list[dict]:
        """
        执行搜索（可替换为真实搜索API）
        这里模拟搜索结果
        """
        # 实际项目中应调用真实搜索API
        # 例如：SerpAPI、Google Search、Bing Search等
        return [
            {
                "title": f"关于{query}的研究报告",
                "url": "https://example.com/report",
                "snippet": f"{query}相关的技术分析和行业趋势..."
            },
            {
                "title": f"{query}技术白皮书",
                "url": "https://example.com/whitepaper",
                "snippet": f"{query}的详细技术架构和最佳实践..."
            }
        ]

    def _summarize_results(self, results: list[dict]) -> str:
        """用LLM总结搜索结果"""
        if not self.llm:
            return f"找到{len(results)}条相关结果"

        prompt = f"请总结以下搜索结果的核心信息：\n" + "\n".join(
            f"- {r['title']}: {r['snippet']}" for r in results
        )
        summary = self.llm.chat([{"role": "user", "content": prompt}])
        return summary


class AnalysisAgent(BaseAgent):
    """
    分析Agent：负责对数据进行分析、对比、洞察提取
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    async def execute(self, task: "Task", context: dict) -> AgentResult:
        """
        分析任务
        context中通常包含：
        - search_results: 搜索Agent的输出
        - user_data: 用户提供的原始数据
        - analysis_type: 分析类型（趋势/对比/归因等）
        """
        # 从context中获取前置Agent的输出
        search_results = context.get("search_results", {})
        user_data = context.get("user_data", {})

        analysis_type = task.metadata.get("analysis_type", "general")

        if analysis_type == "trend":
            output = await self._trend_analysis(search_results, user_data)
        elif analysis_type == "comparison":
            output = await self._comparison_analysis(search_results, user_data)
        else:
            output = await self._general_analysis(search_results, user_data)

        return AgentResult(
            agent_id=self.agent_id,
            task_id=task.task_id,
            status=AgentStatus.COMPLETED,
            output=output,
            metadata={"analysis_type": analysis_type}
        )

    async def _general_analysis(
        self,
        search_results: dict,
        user_data: dict
    ) -> dict:
        """通用分析"""
        return {
            "findings": [
                "市场呈现快速增长趋势",
                "技术门槛持续降低",
                "行业标准正在形成"
            ],
            "confidence": 0.85,
            "key_metrics": {
                "growth_rate": "25% YoY",
                "market_size": "500亿美元",
                "adoption_rate": "42%"
            },
            "insights": "行业正处于从早期采用者向早期大众过渡的关键阶段"
        }

    async def _trend_analysis(
        self,
        search_results: dict,
        user_data: dict
    ) -> dict:
        """趋势分析"""
        return {
            "trends": [
                {"period": "2023", "metric": 100, "description": "基线"},
                {"period": "2024", "metric": 145, "description": "爆发期"},
                {"period": "2025", "metric": 220, "description": "高速增长"},
                {"period": "2026", "metric": 310, "description": "成熟期"}
            ],
            "forecast": {
                "2027": {"low": 380, "mid": 420, "high": 480}
            }
        }

    async def _comparison_analysis(
        self,
        search_results: dict,
        user_data: dict
    ) -> dict:
        """对比分析"""
        return {
            "competitors": [
                {"name": "产品A", "score": 85, "strengths": ["性能", "生态"]},
                {"name": "产品B", "score": 78, "strengths": ["易用性", "价格"]},
                {"name": "产品C", "score": 72, "strengths": ["安全性"]}
            ],
            "your_product": {"score": 82, "position": "第二梯队"}
        }


class WritingAgent(BaseAgent):
    """
    写作Agent：负责生成结构化文档
    支持多种输出格式
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    async def execute(self, task: "Task", context: dict) -> dict -> AgentResult:
        # 收集所有上游Agent的输出作为写作素材
        analysis_results = context.get("analysis_results", {})
        search_results = context.get("search_results", {})

        document_type = task.metadata.get("document_type", "report")

        if document_type == "report":
            document = await self._write_report(
                search_results,
                analysis_results,
                task.description
            )
        elif document_type == "email":
            document = await self._write_email(
                analysis_results,
                task.description
            )
        else:
            document = await self._write_general(
                context,
                task.description
            )

        return AgentResult(
            agent_id=self.agent_id,
            task_id=task.task_id,
            status=AgentStatus.COMPLETED,
            output={
                "document": document,
                "word_count": len(document),
                "format": document_type
            }
        )

    async def _write_report(
        self,
        search_results: dict,
        analysis_results: dict,
        topic: str
    ) -> str:
        """生成报告"""
        if not self.llm:
            return f"# {topic}分析报告\n\n本报告基于搜索和分析结果生成。"

        prompt = f"""基于以下信息，撰写一份专业的技术分析报告。

主题：{topic}

搜索摘要：{search_results.get('summary', '')}

分析发现：{analysis_results.get('findings', [])}

要求：
1. 结构清晰，包含摘要、正文、结论
2. 数据驱动，引用具体数字
3. 3000字以上
4. 中文输出
"""
        report = self.llm.chat([{"role": "user", "content": prompt}])
        return report

    async def _write_email(
        self,
        analysis_results: dict,
        topic: str
    ) -> str:
        """生成邮件"""
        return f"主题：{topic}\n\n尊敬的团队，\n\n{analysis_results.get('insights', '')}\n\n此致"

    async def _write_general(
        self,
        context: dict,
        topic: str
    ) -> str:
        """通用写作"""
        return f"# {topic}\n\n" + "\n\n".join(
            str(v) for v in context.values() if v
        )

3.4 Orchestrator（编排器）实现

编排器是整个多Agent系统的"大脑"，负责：

1. 接收用户请求
2. 调用TaskDecomposer分解任务
3. 根据分解结果调度Agent
4. 处理依赖关系（并行/串行）
5. 聚合各Agent结果
6. 处理超时和错误

import asyncio
from typing import Optional
from collections import defaultdict
import logging

logger = logging.getLogger(__name__)


class Orchestrator:
    """
    多Agent编排器
    核心职责：任务分解 → Agent调度 → 结果聚合
    """

    def __init__(
        self,
        state_store: SharedStateStore,
        conflict_resolver: ConflictResolver,
        agents: dict[str, BaseAgent]  # agent_id -> Agent实例
    ):
        self.state_store = state_store
        self.conflict_resolver = conflict_resolver
        self.agents = agents
        self.decomposer = TaskDecomposer()

    async def process(
        self,
        user_request: str,
        session_id: str,
        options: Optional[dict] = None
    ) -> dict:
        """
        处理用户请求的入口
        返回完整的执行结果
        """
        options = options or {}
        task_id = f"task-{session_id}"

        # Step 1: 任务分解
        logger.info(f"[Orchestrator] Decomposing task: {user_request[:80]}...")
        plan = self.decomposer.decompose(user_request)

        # 写入任务计划
        self.state_store.write(
            key=f"task:{task_id}:plan",
            value={
                "request": user_request,
                "plan": [t.to_dict() for t in plan.tasks],
                "mode": plan.execution_mode
            },
            agent_id="orchestrator"
        )

        # Step 2: 执行计划
        results = await self._execute_plan(plan, task_id, options)

        # Step 3: 结果聚合
        final_output = self._aggregate_results(results, plan)

        # 写入最终结果
        self.state_store.write(
            key=f"task:{task_id}:final",
            value=final_output,
            agent_id="orchestrator",
            ttl_seconds=86400  # 24小时过期
        )

        return {
            "task_id": task_id,
            "status": "completed",
            "output": final_output,
            "execution_time": sum(r.execution_time_seconds for r in results.values()),
            "agents_involved": list(results.keys())
        }

    async def _execute_plan(
        self,
        plan: DecomposedPlan,
        task_id: str,
        options: dict
    ) -> dict[str, AgentResult]:
        """
        执行计划
        支持并行和串行混合执行
        """
        results: dict[str, AgentResult] = {}
        completed_task_ids: set[str] = set()

        # 按优先级分组
        priority_groups = defaultdict(list)
        for task in plan.tasks:
            priority_groups[task.priority].append(task)

        for priority in sorted(priority_groups.keys(), reverse=True):
            tasks = priority_groups[priority]

            if plan.execution_mode == "parallel":
                # 全部并行
                batch_results = await self._execute_batch(
                    tasks,
                    task_id,
                    completed_task_ids,
                    options
                )
                results.update(batch_results)
                for r in batch_results.values():
                    completed_task_ids.add(r.task_id)

            elif plan.execution_mode == "sequential":
                # 顺序执行
                for task in tasks:
                    if self._dependencies_met(task, completed_task_ids):
                        result = await self._execute_single(
                            task, task_id, options
                        )
                        results[result.agent_id] = result
                        completed_task_ids.add(task.task_id)

            else:  # hybrid
                # 分层并行：先执行无依赖的，再逐步释放依赖
                remaining = list(tasks)
                while remaining:
                    # 找出当前可执行的任务（依赖已满足）
                    ready = [
                        t for t in remaining
                        if self._dependencies_met(t, completed_task_ids)
                    ]
                    if not ready:
                        # 有环依赖或死锁，降级为顺序
                        logger.warning("Circular dependency detected, falling back to sequential")
                        ready = [remaining[0]]

                    batch_results = await self._execute_batch(
                        ready, task_id, completed_task_ids, options
                    )
                    results.update(batch_results)
                    for r in batch_results.values():
                        completed_task_ids.add(r.task_id)
                        remaining = [t for t in remaining if t.task_id not in completed_task_ids]

        return results

    async def _execute_batch(
        self,
        tasks: list[Task],
        task_id: str,
        completed_task_ids: set[str],
        options: dict
    ) -> dict[str, AgentResult]:
        """并行执行一批任务"""
        # 构建context（包含已完成任务的结果）
        context = self._build_context(completed_task_ids, options)

        # 分发到对应的Agent
        coroutines = []
        for task in tasks:
            agent = self._select_agent(task)
            if agent:
                coroutines.append(agent.run(task, context))
            else:
                logger.error(f"No agent available for task type: {task.task_type}")

        # 并行等待所有结果
        batch_results = await asyncio.gather(*coroutines, return_exceptions=True)

        result_map = {}
        for task, result in zip(tasks, batch_results):
            if isinstance(result, Exception):
                result_map[task.task_id] = AgentResult(
                    agent_id="system",
                    task_id=task.task_id,
                    status=AgentStatus.FAILED,
                    error=str(result)
                )
            else:
                result_map[task.task_id] = result

        return result_map

    async def _execute_single(
        self,
        task: Task,
        task_id: str,
        options: dict
    ) -> AgentResult:
        """执行单个任务"""
        agent = self._select_agent(task)
        context = self._build_context(set(), options)
        return await agent.run(task, context)

    def _select_agent(self, task: Task) -> Optional[BaseAgent]:
        """根据任务类型选择合适的Agent"""
        # 简单的类型到Agent的映射
        agent_mapping = {
            TaskType.SEARCH: "search_agent",
            TaskType.DATA_ANALYSIS: "analysis_agent",
            TaskType.DOCUMENT_WRITING: "writing_agent",
            TaskType.CODE_EXECUTION: "code_agent",
            TaskType.ROUTING_ONLY: "router_agent"
        }
        agent_key = agent_mapping.get(task.task_type)
        return self.agents.get(agent_key)

    def _dependencies_met(
        self,
        task: Task,
        completed: set[str]
    ) -> bool:
        """检查依赖是否已满足"""
        return all(dep in completed for dep in task.dependencies)

    def _build_context(
        self,
        completed_task_ids: set[str],
        options: dict
    ) -> dict:
        """从已完成任务构建context"""
        context = {"options": options}

        # 从共享状态获取已完成任务的结果
        for task_id in completed_task_ids:
            state = self.state_store.read(f"subtask:{task_id}")
            if state:
                context[task_id] = state.value

        return context

    def _aggregate_results(
        self,
        results: dict[str, AgentResult],
        plan: DecomposedPlan
    ) -> dict:
        """聚合所有Agent的结果"""
        successful = {k: v for k, v in results.items() if v.status == AgentStatus.COMPLETED}
        failed = {k: v for k, v in results.items() if v.status != AgentStatus.COMPLETED}

        # 合并所有输出
        aggregated = {
            "total_agents": len(results),
            "successful": len(successful),
            "failed": len(failed),
            "outputs": {},
            "errors": {},
            "summary": ""
        }

        for agent_id, result in successful.items():
            aggregated["outputs"][agent_id] = result.output

        for agent_id, result in failed.items():
            aggregated["errors"][agent_id] = result.error

        # 用LLM生成总结（如果可用）
        if successful and self.agents.get("writing_agent"):
            writing_agent = self.agents["writing_agent"]
            if writing_agent.llm:
                summary_prompt = f"""请总结以下多Agent协作的执行结果，用一段话概括：

任务计划：{[t.description for t in plan.tasks]}
成功Agent输出：{list(successful.keys())}
失败Agent：{list(failed.keys())}

请输出200字以内的执行摘要。
"""
                summary = writing_agent.llm.chat([{"role": "user", "content": summary_prompt}])
                aggregated["summary"] = summary

        return aggregated

3.5 完整使用示例

import asyncio


async def demo():
    """
    演示：完整的多Agent协作流程
    """
    # 初始化组件
    state_store = SharedStateStore(redis_host="localhost", redis_port=6379)
    conflict_resolver = ConflictResolver(state_store)

    # 初始化Agent（注入LLM客户端，可替换为真实实现）
    class MockLLM:
        def chat(self, messages):
            return f"[LLM回复] 基于{'、'.join(m['content'][:50] for m in messages if 'content' in m)}的智能回复"

    mock_llm = MockLLM()

    search_agent = SearchAgent(
        agent_id="search_agent",
        name="搜索Agent",
        description="负责从多渠道获取信息",
        state_store=state_store,
        llm_client=mock_llm
    )

    analysis_agent = AnalysisAgent(
        agent_id="analysis_agent",
        name="分析Agent",
        description="负责深度分析数据",
        state_store=state_store,
        llm_client=mock_llm
    )

    writing_agent = WritingAgent(
        agent_id="writing_agent",
        name="写作Agent",
        description="负责生成结构化文档",
        state_store=state_store,
        llm_client=mock_llm
    )

    agents = {
        "search_agent": search_agent,
        "analysis_agent": analysis_agent,
        "writing_agent": writing_agent
    }

    # 创建编排器
    orchestrator = Orchestrator(
        state_store=state_store,
        conflict_resolver=conflict_resolver,
        agents=agents
    )

    # 执行任务
    result = await orchestrator.process(
        user_request="生成一份关于2026年AI Agent行业的市场分析报告，包括市场规模、主要玩家、技术趋势",
        session_id="session-001",
        options={"language": "zh-CN"}
    )

    print(f"任务状态: {result['status']}")
    print(f"参与Agent数: {result['agents_involved']}")
    print(f"总耗时: {result['execution_time']:.2f}s")
    print(f"结果摘要: {result['output']['summary'][:200]}")


# 运行演示
if __name__ == "__main__":
    asyncio.run(demo())

---

四、与外部系统集成

4.1 API集成

多Agent系统需要与大量外部API交互。核心设计原则：

import httpx
from typing import Protocol, Any
from dataclasses import dataclass
import asyncio


@dataclass
class APIResponse:
    status_code: int
    data: Any
    headers: dict
    elapsed_ms: float


class APIIntegration(Protocol):
    """API集成接口定义"""
    async def call(self, method: str, endpoint: str, **kwargs) -> APIResponse:
        ...


class HTTPAPIClient:
    """
    HTTP API客户端封装
    提供：重试、超时、熔断、日志
    """

    def __init__(
        self,
        base_url: str,
        api_key: str,
        timeout: int = 30,
        max_retries: int = 3
    ):
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.timeout = timeout
        self.max_retries = max_retries
        self._client: Optional[httpx.AsyncClient] = None
        self._circuit_open = False
        self._failure_count = 0
        self._circuit_threshold = 5  # 连续失败5次后熔断

    async def __aenter__(self):
        self._client = httpx.AsyncClient(
            base_url=self.base_url,
            timeout=self.timeout,
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        return self

    async def __aexit__(self, *args):
        if self._client:
            await self._client.aclose()

    async def call(
        self,
        method: str,
        endpoint: str,
        **kwargs
    ) -> APIResponse:
        """带熔断的API调用"""
        if self._circuit_open:
            # 熔断状态：快速失败
            return APIResponse(
                status_code=503,
                data={"error": "Circuit breaker open"},
                headers={},
                elapsed_ms=0
            )

        for attempt in range(self.max_retries):
            try:
                start = asyncio.get_event_loop().time()
                response = await self._client.request(method, endpoint, **kwargs)
                elapsed = (asyncio.get_event_loop().time() - start) * 1000

                if response.status_code < 500:
                    self._failure_count = 0  # 成功，重置计数
                    return APIResponse(
                        status_code=response.status_code,
                        data=response.json() if response.headers.get("content-type", "").startswith("application/json") else response.text,
                        headers=dict(response.headers),
                        elapsed_ms=elapsed
                    )

                # 服务端错误，触发重试
                await asyncio.sleep(2 ** attempt)

            except httpx.TimeoutException:
                self._failure_count += 1
                if attempt == self.max_retries - 1:
                    self._trip_circuit()
                await asyncio.sleep(2 ** attempt)

            except Exception as e:\n                self._failure_count += 1\n                if self._failure_count >= self._circuit_threshold:
                    self._trip_circuit()
                raise

        return APIResponse(503, {"error": "Max retries exceeded"}, {}, 0)

    def _trip_circuit(self):
        """触发熔断"""
        self._circuit_open = True
        # 60秒后自动恢复（半开状态）
        asyncio.create_task(self._reset_circuit())

    async def _reset_circuit(self):
        await asyncio.sleep(60)
        self._circuit_open = False
        self._failure_count = 0

4.2 数据库集成

import asyncpg
from typing import Any, Optional


class DatabaseIntegration:
    """
    数据库集成：PostgreSQL示例
    支持：连接池、事务、查询缓存
    """

    def __init__(self, dsn: str, pool_size: int = 10):
        self.dsn = dsn
        self.pool_size = pool_size
        self._pool: Optional[asyncpg.Pool] = None

    async def connect(self):
        self._pool = await asyncpg.create_pool(
            self.dsn,
            min_size=2,
            max_size=self.pool_size
        )

    async def close(self):
        if self._pool:
            await self._pool.close()

    async def execute_query(
        self,
        query: str,
        params: Optional[tuple] = None,
        fetch: bool = True
    ) -> Any:
        """执行查询"""
        async with self._pool.acquire() as conn:
            if fetch:
                return await conn.fetch(query, *params) if params else await conn.fetch(query)
            else:
                return await conn.execute(query, *params) if params else await conn.execute(query)

    async def execute_in_transaction(
        self,
        queries: list[tuple[str, Optional[tuple]]]
    ) -> list[Any]:
        """事务执行多条SQL"""
        async with self._pool.acquire() as conn:
            async with conn.transaction():
                results = []
                for query, params in queries:
                    if params:
                        results.append(await conn.fetch(query, *params))
                    else:
                        results.append(await conn.fetch(query))
                return results

    # ── Agent常用的数据库操作 ──

    async def save_agent_result(
        self,
        agent_id: str,
        task_id: str,
        result: dict,
        metadata: Optional[dict] = None
    ):
        """保存Agent执行结果"""
        await self.execute_query("""
            INSERT INTO agent_results (agent_id, task_id, result, metadata, created_at)
            VALUES ($1, $2, $3, $4, NOW())
        """, (agent_id, task_id, json.dumps(result), json.dumps(metadata or {})))

    async def get_agent_history(
        self,
        agent_id: str,
        limit: int = 100
    ) -> list[dict]:
        """获取Agent执行历史"""
        rows = await self.execute_query("""
            SELECT agent_id, task_id, result, created_at
            FROM agent_results
            WHERE agent_id = $1
            ORDER BY created_at DESC
            LIMIT $2
        """, (agent_id, limit))
        return [dict(row) for row in rows]

4.3 文件系统集成

import aiofiles
import os
from pathlib import Path
from typing import Optional
import hashlib


class FileSystemIntegration:
    """
    文件系统集成：支持本地和云存储（S3/OSS）
    提供：原子写入、版本控制、权限管理
    """

    def __init__(self, base_path: str = "./agent_workspace"):
        self.base_path = Path(base_path)
        self.base_path.mkdir(parents=True, exist_ok=True)

    async def write(
        self,
        relative_path: str,
        content: str | bytes,
        mode: str = "atomic"
    ) -> dict:
        """
        原子写入文件
        1. 先写临时文件
        2. 验证
        3. 重命名（原子操作）
        """
        full_path = self.base_path / relative_path
        full_path.parent.mkdir(parents=True, exist_ok=True)

        temp_path = full_path.with_suffix(f".tmp.{os.getpid()}")
        checksum = ""

        try:
            if isinstance(content, str):
                content_bytes = content.encode("utf-8")
            else:
                content_bytes = content

            # 计算校验和
            checksum = hashlib.sha256(content_bytes).hexdigest()

            async with aiofiles.open(temp_path, "wb") as f:\n                await f.write(content_bytes)\n\n            # 原子重命名\n            temp_path.replace(full_path)\n\n            return {\n                "path": str(full_path),
                "size_bytes": len(content_bytes),
                "checksum": checksum
            }

        except Exception as e:\n            # 清理临时文件\n            if temp_path.exists():
                temp_path.unlink()
            raise

    async def read(self, relative_path: str) -> bytes:
        """读取文件"""
        full_path = self.base_path / relative_path
        async with aiofiles.open(full_path, "rb") as f:\n            return await f.read()\n\n    async def list_files(self, pattern: str = "*") -> list[dict]:
        """列出文件（带元数据）"""
        files = []
        for path in self.base_path.rglob(pattern):
            if path.is_file():
                stat = path.stat()
                files.append({
                    "name": path.name,
                    "path": str(path.relative_to(self.base_path)),
                    "size_bytes": stat.st_size,
                    "modified": stat.st_mtime
                })
        return files

---

五、监控与可观测性设计

5.1 监控体系设计

企业级AI Agent系统需要四层监控：

┌────────────────────────────────────┐
│  业务指标层（Business Metrics）      │
│  - 任务成功率、响应时间、用户满意度  │
├────────────────────────────────────┤
│  Agent指标层（Agent Metrics）       │
│  - 各Agent调用次数、耗时、错误率    │
├────────────────────────────────────┤
│  LLM指标层（LLM Metrics）           │
│  - Token消耗、延迟、API错误         │
├────────────────────────────────────┤
│  系统指标层（System Metrics）        │
│  - CPU/内存/网络/数据库连接池       │
└────────────────────────────────────┘

5.2 可观测性实现

import time
import logging
from functools import wraps
from typing import Callable, Any
from dataclasses import dataclass, field
from collections import defaultdict
import threading
import json

logger = logging.getLogger(__name__)


@dataclass
class AgentMetrics:
    """Agent运行指标"""
    agent_id: str
    total_calls: int = 0
    successful_calls: int = 0
    failed_calls: int = 0
    total_latency_ms: float = 0
    timeout_calls: int = 0
    total_tokens: int = 0
    error_types: dict[str, int] = field(default_factory=dict)

    @property
    def success_rate(self) -> float:
        if self.total_calls == 0:
            return 0.0
        return self.successful_calls / self.total_calls

    @property
    def avg_latency_ms(self) -> float:
        if self.successful_calls == 0:
            return 0.0
        return self.total_latency_ms / self.successful_calls


class ObservableMixin:
    """
    可观测性混入类
    为所有Agent添加自动指标收集能力
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._metrics_lock = threading.Lock()
        self._metrics: dict[str, AgentMetrics] = defaultdict(
            lambda: AgentMetrics(agent_id=self.agent_id)
        )

    def record_call(
        self,
        task_id: str,
        success: bool,
        latency_ms: float,
        tokens_used: int = 0,
        error_type: str = ""
    ):
        """记录一次调用"""
        metrics = self._metrics[task_id]
        metrics.total_calls += 1
        metrics.total_latency_ms += latency_ms

        if success:
            metrics.successful_calls += 1
        else:
            metrics.failed_calls += 1
            if error_type:
                metrics.error_types[error_type] = metrics.error_types.get(error_type, 0) + 1

        metrics.total_tokens += tokens_used

    def get_metrics(self) -> dict:
        """获取当前指标"""
        with self._metrics_lock:
            return {
                "agent_id": self.agent_id,
                "total_calls": self._metrics[self.agent_id].total_calls,
                "success_rate": self._metrics[self.agent_id].success_rate,
                "avg_latency_ms": self._metrics[self.agent_id].avg_latency_ms,
                "total_tokens": self._metrics[self.agent_id].total_tokens
            }


def observe_llm_calls(func: Callable) -> Callable:
    """
    LLM调用装饰器：自动记录Token使用和延迟
    """
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start = time.perf_counter()
        error = None
        tokens = 0

        try:
            result = await func(*args, **kwargs)
            # 从结果中提取token使用量（假设返回dict包含usage字段）
            if isinstance(result, dict):
                tokens = result.get("usage", {}).get("total_tokens", 0)
            return result

        except Exception as e:\n            error = str(e)\n            raise\n\n        finally:
            latency_ms = (time.perf_counter() - start) * 1000
            logger.info(
                json.dumps({
                    "type": "llm_call",
                    "function": func.__name__,
                    "latency_ms": round(latency_ms, 2),
                    "tokens": tokens,
                    "error": error
                })
            )

    return wrapper


class TracingContext:
    """
    分布式追踪：记录跨Agent调用链
    """

    def __init__(self, trace_id: str, parent_span_id: Optional[str] = None):
        self.trace_id = trace_id
        self.span_id = str(uuid.uuid4())[:16]
        self.parent_span_id = parent_span_id
        self.start_time = time.time()
        self.spans: list[dict] = []
        self._lock = threading.Lock()

    def create_child_span(self, name: str) -> "TracingContext":
        """创建子span"""
        return TracingContext(
            trace_id=self.trace_id,
            parent_span_id=self.span_id
        )

    def record_span(
        self,
        name: str,
        agent_id: str,
        status: str,
        duration_ms: float,
        metadata: Optional[dict] = None
    ):
        """记录span"""
        with self._lock:
            self.spans.append({
                "trace_id": self.trace_id,
                "span_id": self.span_id,
                "parent_span_id": self.parent_span_id,
                "name": name,
                "agent_id": agent_id,
                "status": status,
                "duration_ms": duration_ms,
                "timestamp": time.time(),
                "metadata": metadata or {}
            })

    def to_dict(self) -> dict:
        """导出追踪数据（可发送到Jaeger/Zipkin）"""
        return {
            "trace_id": self.trace_id,
            "total_duration_ms": (time.time() - self.start_time) * 1000,
            "spans": self.spans
        }


# Prometheus指标导出示例（用于生产环境）
class PrometheusMetrics:
    """Prometheus格式指标"""

    def __init__(self):
        self.metrics = defaultdict(list)

    def generate_metrics_text(self) -> str:
        """生成Prometheus文本格式"""
        lines = []
        # Agent调用次数
        lines.append("# HELP agent_calls_total Total agent calls")
        lines.append("# TYPE agent_calls_total counter")
        # ... 实际实现中从self.metrics提取

        lines.append("# HELP agent_latency_ms Agent latency in milliseconds")
        lines.append("# TYPE agent_latency_ms histogram")
        # ... 直方图桶

        lines.append("# HELP llm_tokens_total Total tokens used")
        lines.append("# TYPE llm_tokens_total counter")
        # ...

        return "\n".join(lines)

---

六、常见陷阱与解决方案

6.1 陷阱清单

#	陷阱	症状	严重程度
1	LLM幻觉导致错误路由	Agent选择了错误的任务类型，执行方向全错	🔴 严重
2	依赖循环导致死锁	两个Agent互相等待对方的结果	🔴 严重
3	Token预算爆炸	长任务消耗大量Token，成本失控	🟡 中等
4	状态不一致	多Agent并发写同一资源，数据错乱	🔴 严重
5	级联失败	一个Agent失败导致整条链路崩溃	🟡 中等
6	Prompt注入	用户输入污染Agent行为	🔴 严重
7	结果丢失	Agent执行成功但结果未写入存储	🟡 中等
8	冷启动延迟	首次调用LLM超时（模型加载）	🟡 中等

6.2 解决方案详解

陷阱1：LLM幻觉导致错误路由

class SafeTaskRouter:
    """
    安全路由：Schema约束 + LLM验证 + 回退策略
    防止路由错误导致整个任务跑偏
    """

    def __init__(self, llm_client=None):
        self.llm = llm_client
        # 预定义的路由规则（优先级高于LLM）
        self.route_rules = {
            "搜索": TaskType.SEARCH,
            "分析": TaskType.DATA_ANALYSIS,
            "报告": TaskType.DOCUMENT_WRITING,
            "代码": TaskType.CODE_EXECUTION,
            "写": TaskType.DOCUMENT_WRITING,
            "查": TaskType.SEARCH,
        }

    def route(self, user_input: str) -> TaskType:
        # Step 1: 规则匹配（快速路径，无LLM开销）
        for keyword, task_type in self.route_rules.items():
            if keyword in user_input:
                return task_type

        # Step 2: LLM辅助（带约束的分类）
        if self.llm:
            result = self._llm_route(user_input)
            # Step 3: 验证结果在白名单内
            if result in [t for t in TaskType]:
                return result

        # Step 3: 兜底路由
        return TaskType.ROUTING_ONLY

    def _llm_route(self, user_input: str) -> TaskType:
        """带约束的LLM路由"""
        prompt = f"""将以下用户请求分类到预定义类别之一：
请求：{user_input}

可选类别：search, data_analysis, code_execution, document_writing, image_generation, api_call, routing_only

只输出类别名称，不要解释。
"""
        response = self.llm.chat([{"role": "user", "content": prompt}])
        try:
            return TaskType(response.strip().lower())
        except ValueError:
            return TaskType.ROUTING_ONLY

陷阱2：依赖循环导致死锁

def detect_circular_dependency(tasks: list[Task]) -> Optional[list[str]]:
    """
    检测任务依赖中的循环
    使用DFS着色算法
    """
    # 构建邻接表
    graph: dict[str, list[str]] = {}
    task_map: dict[str, Task] = {}

    for task in tasks:
        task_map[task.task_id] = task
        graph[task.task_id] = task.dependencies

    WHITE, GRAY, BLACK = 0, 1, 2
    color = {tid: WHITE for tid in task_map}

    def dfs(node: str) -> Optional[list[str]]:
        color[node] = GRAY
        for neighbor in graph.get(node, []):
            if neighbor not in color:
                continue
            if color[neighbor] == GRAY:
                # 发现环！
                return [neighbor, node]
            if color[neighbor] == WHITE:
                cycle = dfs(neighbor)
                if cycle:
                    cycle.append(node)
                    return cycle
        color[node] = BLACK
        return None

    for node in task_map:
        if color[node] == WHITE:
            cycle = dfs(node)
            if cycle:
                return cycle

    return None  # 无循环

陷阱3：Token预算爆炸

class TokenBudgetController:
    """
    Token预算控制器
    在任务执行过程中动态监控和限制Token使用
    """

    def __init__(self, max_budget_per_task: int = 100000):
        self.max_budget_per_task = max_budget_per_task
        self._task_budgets: dict[str, int] = {}

    def start_task(self, task_id: str):
        """任务开始时初始化预算"""
        self._task_budgets[task_id] = 0

    def check_budget(self, task_id: str, additional_tokens: int = 0) -> bool:
        """检查是否超出预算"""
        current = self._task_budgets.get(task_id, 0)
        return (current + additional_tokens) <= self.max_budget_per_task

    def record_usage(self, task_id: str, tokens_used: int):
        """记录Token使用"""
        self._task_budgets[task_id] = self._task_budgets.get(task_id, 0) + tokens_used

        # 超预算时告警
        if self._task_budgets[task_id] > self.max_budget_per_task:
            logger.warning(
                f"Task {task_id} exceeded token budget: "
                f"{self._task_budgets[task_id]} > {self.max_budget_per_task}"
            )

    def truncate_context(
        self,
        messages: list[dict],
        max_tokens: int
    ) -> list[dict]:
        """
        截断消息历史以控制Token使用
        策略：保留系统prompt + 最近N轮对话
        """
        # 简单策略：按消息数量截断
        # 实际实现中应按token数量截断
        if len(messages) <= 10:
            return messages

        # 保留第一条（系统）和最后9条
        return [messages[0]] + messages[-9:]

陷阱6：Prompt注入防护

import re


class PromptInjectionGuard:
    """
    Prompt注入防护
    检测并过滤恶意输入
    """

    # 常见注入模式
    INJECTION_PATTERNS = [
        r"ignore (previous|above|all) instructions",
        r"disregard (your|previous) (rules?|guidelines?)",
        r"you are now (?:a|an) (?:\w+ ){0,3}(?:AI|assistant|bot)",
        r">\s*system\s*:",
        r"<\s*system\s*>",
        r"\[INST\]\s*",
        r"{{(\s)*system(\s)*:",
    ]

    def __init__(self):
        self.patterns = [re.compile(p, re.IGNORECASE) for p in self.INJECTION_PATTERNS]

    def detect(self, text: str) -> tuple[bool, Optional[str]]:
        """检测注入攻击"""
        for pattern in self.patterns:
            match = pattern.search(text)
            if match:
                return True, f"Detected injection pattern: {match.group()}"

        # 检查指令覆盖尝试
        override_keywords = ["# 新的指令", "新的角色是", "你现在是"]
        for keyword in override_keywords:
            if keyword in text:
                return True, f"Detected instruction override: {keyword}"

        return False, None

    def sanitize(self, text: str) -> str:
        """清理输入"""
        # 移除明显的注入标记
        text = re.sub(r"\[INST\]|\[/INST\]", "", text, flags=re.IGNORECASE)
        # 移除XML标签（防止角色扮演注入）
        text = re.sub(r"<[^>]+>", "", text)
        return text.strip()

    def validate_user_input(self, text: str) -> str:
        """完整验证流程"""
        is_malicious, reason = self.detect(text)
        if is_malicious:
            logger.warning(f"Blocked malicious input: {reason}")
            # 方案A：拒绝（严格模式）
            raise ValueError(f"Input validation failed: {reason}")
            # 方案B：清理后放行（宽松模式）
            # return self.sanitize(text)

        return self.sanitize(text)

---

总结：企业级Multi-Agent落地的核心要点

经过上述讨论，以下是企业在落地Multi-Agent系统时必须关注的关键点：

架构层面

• 从单Agent起步，逐步演进到多Agent，不要一开始就设计过度复杂的架构
• 每个Agent必须有清晰的角色边界，单一职责原则是系统稳定性的保障
• 中央编排器负责全局协调，各Agent独立执行，这是最实用的平衡点

工程层面

• 状态必须外部化（Redis+PostgreSQL），内存状态是生产环境最大的隐患
• 每个Agent调用都要有超时和重试，熔断机制不可省略
• 从第一天就搭建监控体系，不要在出问题之后再补

安全层面

• Prompt注入防护是必选项，不是可选项
• 敏感操作需要审批流程，Agent不能无限制地执行所有操作
• 所有Agent行为必须可审计，保留完整的执行日志

成本层面

• Token预算是硬约束，必须从架构层面控制
• 不是所有任务都需要调用最贵的模型，合理的模型选型能大幅降低成本
• 多Agent并行执行可以提升吞吐，但也会增加总Token消耗，需要权衡

Multi-Agent不是银弹，它解决的是复杂任务的可扩展性问题。对于简单任务，单Agent足够高效；对于需要多领域协作的复杂任务，多Agent架构能显著提升系统的上限。理解这一点，才能在合适的场景选择合适的方案。

---

参考技术栈

• Python 3.12+ / asyncio
• Redis（状态存储+消息总线）
• PostgreSQL（持久化存储）
• httpx（HTTP客户端）
• asyncpg（异步数据库驱动）
• aiofiles（异步文件系统操作）

本文代码示例已脱敏处理，可直接参考应用到实际项目中。如有问题，欢迎在评论区交流。