多Agent协调基础设施三栈：事件网格·能力注册·链路追踪实测

graph TB
    subgraph "协调基础设施层"
        EM[事件网格<br/>Event Mesh]
        CR[能力注册中心<br/>Capability Registry]
        DT[分布式链路追踪<br/>Distributed Tracing]
    end

    subgraph "Agent层"
        A1[研究Agent]
        A2[分析Agent]
        A3[写作Agent]
        A4[审核Agent]
    end

    A1 -->|发布事件| EM
    EM -->|路由到| A2
    A2 -->|发布事件| EM
    EM -->|路由到| A3
    A3 -->|发布事件| EM
    EM -->|路由到| A4

    CR -->|能力发现| A1
    CR -->|能力发现| A2
    CR -->|能力发现| A3

    DT -.->|span采集| EM
    DT -.->|span采集| A1
    DT -.->|span采集| A2

import asyncio
from nats.aio.client import Client as NATS

class AgentEventBus:
    def __init__(self, nats_url="nats://localhost:4222"):
        self.nc = NATS()
        self.nats_url = nats_url

    async def connect(self):
        await self.nc.connect(self.nats_url)

    async def publish_capability(self, agent_id: str, capabilities: list):
        """Agent上线时发布自己的能力声明"""
        msg = {
            "agent_id": agent_id,
            "capabilities": capabilities,
            "timestamp": time.time(),
            "ttl": 300  # 5分钟心跳
        }
        await self.nc.publish(
            f"agent.registry.{agent_id}", 
            json.dumps(msg).encode()
        )

    async def request_task_delegation(self, task_type: str, payload: dict):
        """请求具有特定能力的Agent处理任务"""
        response = await self.nc.request(
            f"agent.capability.{task_type}",
            json.dumps(payload).encode(),
            timeout=5.0
        )
        return json.loads(response.data.decode())

    async def subscribe_capability(self, task_type: str, handler):
        """订阅特定类型的任务请求"""
        async def callback(msg):
            result = await handler(json.loads(msg.data.decode()))
            await msg.respond(json.dumps(result).encode())

        await self.nc.subscribe(f"agent.capability.{task_type}", cb=callback)

# solace-agent-mesh 工作流定义示例
workflow:
  name: "research-pipeline"
  trigger:
    event: "user.request.research"

  steps:
    - agent: "web-search-agent"
      input: "${trigger.query}"
      output: "search_results"
      on_failure: "retry(3, backoff=exponential)"

    - agent: "analysis-agent"
      input: "${search_results}"
      output: "analysis"
      depends_on: ["web-search-agent"]

    - agent: "fact-check-agent"
      input: "${analysis}"
      output: "fact_check"
      depends_on: ["analysis-agent"]
      # Devil's Advocate模式：独立验证

    - agent: "synthesis-agent"
      input: 
        analysis: "${analysis}"
        fact_check: "${fact_check}"
      output: "final_report"
      depends_on: ["fact-check-agent"]

graph TD
    START[选择事件网格] --> Q1{Agent数量?}
    Q1 -->|<10| Q2{需要事件溯源?}
    Q1 -->|10-100| Q3{云厂商绑定?}
    Q1 -->|>100| Q4{延迟要求?}

    Q2 -->|是| KAFKA1[Kafka + Schema Registry]
    Q2 -->|否| NATS1[NATS JetStream]

    Q3 -->|AWS| EB[AWS EventBridge]
    Q3 -->|否| SOLACE[Solace Agent Mesh]

    Q4 -->|<5ms| NATS2[NATS Cluster]
    Q4 -->|5-50ms| SOLACE2[Solace Event Mesh]

    style START fill:#4a90d9,color:#fff
    style KAFKA1 fill:#d94a4a,color:#fff
    style NATS1 fill:#4ad97a,color:#fff
    style EB fill:#d9a84a,color:#fff
    style SOLACE fill:#9b4ad9,color:#fff
    style NATS2 fill:#4ad97a,color:#fff
    style SOLACE2 fill:#9b4ad9,color:#fff

import etcd3
import json
import time
import threading

class CapabilityRegistry:
    def __init__(self, etcd_host="localhost", etcd_port=2379):
        self.etcd = etcd3.client(host=etcd_host, port=etcd_port)
        self.lease = None
        self._heartbeat_thread = None

    def register_agent(self, agent_id: str, capabilities: list, 
                       metadata: dict = None):
        """注册Agent能力，带TTL心跳"""
        self.lease = self.etcd.lease(ttl=30)  # 30秒TTL

        registration = {
            "agent_id": agent_id,
            "capabilities": capabilities,
            "metadata": metadata or {},
            "status": "ready",
            "current_load": 0,
            "max_load": 10,
            "registered_at": time.time()
        }

        key = f"agents/{agent_id}"
        self.etcd.put(
            key, 
            json.dumps(registration),
            lease=self.lease
        )

        # 为每个能力建立索引
        for cap in capabilities:
            idx_key = f"capabilities/{cap}/{agent_id}"
            self.etcd.put(idx_key, json.dumps({
                "load": 0, "latency_p50": 0
            }), lease=self.lease)

        # 启动心跳续约
        self._start_heartbeat()

    def discover_agents(self, capability: str, min_load=0, max_load=10):
        """发现具有特定能力且负载可用的Agent"""
        prefix = f"capabilities/{capability}/"
        candidates = []

        for value, metadata in self.etcd.get_prefix(prefix):
            agent_data = json.loads(value)
            if min_load <= agent_data["load"] <= max_load:
                agent_id = metadata.key.decode().split("/")[-1]
                candidates.append({
                    "agent_id": agent_id,
                    "load": agent_data["load"]
                })

        # 按负载排序，返回最优选择
        return sorted(candidates, key=lambda x: x["load"])

    def _start_heartbeat(self):
        """后台线程持续续约lease"""
        def heartbeat():
            while True:
                try:
                    self.lease.refresh()
                except Exception:
                    break
                time.sleep(10)

        self._heartbeat_thread = threading.Thread(
            target=heartbeat, daemon=True
        )
        self._heartbeat_thread.start()

from opentelemetry import trace, context
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
    OTLPSpanExporter
)

# 初始化追踪
provider = TracerProvider()
processor = BatchSpanProcessor(
    OTLPSpanExporter(endpoint="http://localhost:4317")
)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("agent-system")

class TracedAgent:
    def __init__(self, agent_id: str, capabilities: list):
        self.agent_id = agent_id
        self.capabilities = capabilities

    async def execute_task(self, task: dict, parent_span=None):
        """执行任务并记录完整追踪"""
        # 创建子span，继承父span的trace上下文
        ctx = context.get_current() if parent_span else None

        with tracer.start_as_current_span(
            f"agent.{self.agent_id}.execute",
            attributes={
                "agent.id": self.agent_id,
                "agent.capabilities": str(self.capabilities),
                "task.type": task.get("type", "unknown"),
                "task.id": task.get("id", ""),
            },
            context=ctx
        ) as span:
            try:
                # 阶段1：规划
                with tracer.start_span("plan") as plan_span:
                    plan = await self._plan(task)
                    plan_span.set_attribute(
                        "plan.steps", len(plan.get("steps", []))
                    )

                # 阶段2：执行
                with tracer.start_span("execute") as exec_span:
                    result = await self._execute(plan)
                    exec_span.set_attribute(
                        "result.status", result.get("status")
                    )

                # 阶段3：委派子任务（如果有）
                if result.get("needs_delegation"):
                    with tracer.start_span("delegate") as del_span:
                        delegated = await self._delegate_subtasks(
                            result["subtasks"]
                        )
                        del_span.set_attribute(
                            "delegated.count", len(delegated)
                        )
                        result["delegated_results"] = delegated

                span.set_attribute("execution.success", True)
                return result

            except Exception as e:
                span.set_attribute("execution.success", False)
                span.set_attribute("error.type", type(e).__name__)
                span.set_attribute("error.message", str(e))
                span.set_status(trace.StatusCode.ERROR, str(e))
                raise

sequenceDiagram
    participant User
    participant EM as 事件网格<br/>(NATS)
    participant CR as 能力注册<br/>(etcd)
    participant Orch as 编排Agent
    participant Res as 研究Agent
    participant Ana as 分析Agent
    participant FC as 事实核查Agent
    participant DT as 链路追踪<br/>(OpenTelemetry)

    User->>EM: 发布 research.request

    EM->>Orch: 路由到编排Agent
    DT->>DT: 记录 span: orchestrator.receive

    Orch->>CR: 查询 capability=financial_research
    CR-->>Orch: 返回 [research-agent, alt-research-agent]

    Orch->>EM: 委派任务到 research-agent
    DT->>DT: 记录 span: research.execute

    Res->>EM: 发布 research.complete
    EM->>Orch: 研究结果返回

    Orch->>CR: 查询 capability=data_analysis
    CR-->>Orch: 返回 [analysis-agent]

    Orch->>EM: 并行委派到 analysis + fact-check
    par 并行执行
        EM->>Ana: 分析任务
        DT->>DT: span: analysis.execute
    and
        EM->>FC: 事实核查任务
        DT->>DT: span: factcheck.execute
    end

    Ana->>EM: analysis.complete
    FC->>EM: factcheck.complete
    EM->>Orch: 两个结果聚合

    Orch->>EM: 合成最终报告
    EM->>User: research.complete

    DT->>DT: 记录完整trace: 8 spans, 3.2s

import asyncio
from agent_event_bus import AgentEventBus
from capability_registry import CapabilityRegistry
from traced_agent import TracedAgent
from opentelemetry import trace

tracer = trace.get_tracer("coordinator")

class MultiAgentCoordinator:
    def __init__(self):
        self.bus = AgentEventBus()
        self.registry = CapabilityRegistry()
        self.agents = {}

    async def initialize(self):
        await self.bus.connect()
        # 注册已知Agent
        for agent_config in self._load_agent_configs():
            agent = TracedAgent(
                agent_id=agent_config["id"],
                capabilities=agent_config["capabilities"]
            )
            self.agents[agent_config["id"]] = agent
            self.registry.register_agent(
                agent_config["id"],
                agent_config["capabilities"],
                metadata={"model": agent_config.get("model")}
            )

    async def handle_research_request(self, request: dict):
        """完整的协调流程"""
        with tracer.start_as_current_span(
            "coordinator.research_pipeline",
            attributes={"request.id": request["id"]}
        ) as root_span:

            # 1. 发现研究Agent
            candidates = self.registry.discover_agents(
                capability="financial_research"
            )
            if not candidates:
                raise RuntimeError("No research agent available")

            best_agent = candidates[0]["agent_id"]

            # 2. 委派研究任务
            with tracer.start_span("delegate.research"):
                research_result = await self.bus.request_task_delegation(
                    task_type="financial_research",
                    payload={
                        "query": request["query"],
                        "target_agent": best_agent
                    }
                )

            # 3. 并行执行分析和核查
            with tracer.start_span("parallel.analysis_and_check"):
                analysis_coro = self.bus.request_task_delegation(
                    "data_analysis", {"data": research_result}
                )
                factcheck_coro = self.bus.request_task_delegation(
                    "fact_checking", {"claims": research_result}
                )

                analysis, factcheck = await asyncio.gather(
                    analysis_coro, factcheck_coro
                )

            # 4. 质量门控：核查不通过则回退
            if factcheck.get("confidence", 0) < 0.7:
                root_span.set_attribute(
                    "quality_gate", "failed"
                )
                return {
                    "status": "revision_needed",
                    "issues": factcheck.get("issues", []),
                    "research": research_result
                }

            # 5. 合成最终报告
            return {
                "status": "complete",
                "report": analysis.get("report"),
                "confidence": factcheck.get("confidence"),
                "trace_id": root_span.context.trace_id
            }