多Agent运维五栈实测：从CI/CD到Saga补偿的全闭环突围

Prompt变更 → LangSmith Experiment触发 → 离线Eval跑分 → 
阈值门控(Pass/Fail) → Canary部署(5%流量) → 
在线Eval监控 → 全量发布或回滚

# LangSmith CI/CD GitHub Actions 配置
name: Agent Eval Gate
on:
  pull_request:
    paths:
      - 'agents/**'
      - 'prompts/**'
jobs:
  eval:
    runs-on: ubuntu-latest
    steps:
      - uses: langchain-samples/cicd-pipeline-example@v2
        with:
          experiment-name: ${{ github.event.pull_request.title }}
          dataset: production-eval-v3
          eval-threshold: 0.85  # 低于85%自动阻止合并
          langsmith-api-key: ${{ secrets.LANGSMITH_API_KEY }}

# TrueFoundry Agent Canary 配置
from truefoundry.deploy import AgentDeployment

deployment = AgentDeployment(
    name="research-agent",
    variants=[
        {"name": "stable-v2", "weight": 90, "revision": "agent:v2.3.1"},
        {"name": "canary-v3", "weight": 10, "revision": "agent:v3.0.0-beta"},
    ],
    session_sticky=True,  # 同一对话粘性路由
    rollback_on={
        "tool_error_rate > 0.05": "stable-v2",
        "latency_p99 > 8s": "stable-v2",
    }
)

用户层 → SSE/Streamable HTTP（Agent → 用户流式输出）
Agent层 → WebSocket（Agent ↔ Agent双向通信）  
工具层 → Streamable HTTP（Agent → MCP Server工具调用）

# 多Agent系统分层协议配置（概念模型）
from multi_agent import AgentBus, StreamingConfig

# Agent间通信：WebSocket
bus = AgentBus(
    protocol="websocket",
    broker="redis-pubsub",  # 支持横向扩展
    message_schema={
        "sender": "agent-id",
        "recipient": "agent-id | broadcast",
        "type": "partial_result | final_result | error",
        "payload": "...",
        "span_id": "trace-span-xxx",  # 关联追踪
    }
)

# 用户层输出：SSE
streaming = StreamingConfig(
    user_protocol="sse",
    chunk_size="token",  # 逐Token流式
    include_thinking=True,  # 流式输出思考过程
)

graph TD
    A[Agent请求] --> B{Capability级熔断}
    B -->|OPEN| C[返回降级响应]
    B -->|CLOSED| D{Provider级熔断}
    D -->|OPEN| E[切换备用Provider]
    D -->|CLOSED| F{Tool级熔断}
    F -->|OPEN| G[跳过或Mock该工具]
    F -->|CLOSED| H[正常执行工具调用]

    I[错误计数器] --> J{错误率>阈值?}
    J -->|是| K[熔断器→OPEN]
    K --> L[等待冷却时间]
    L --> M[半开状态→试探请求]
    M -->|成功| N[熔断器→CLOSED]
    M -->|失败| K

// reaatech/circuit-breaker-agents 三层熔断配置
import { CircuitBreaker } from "@reaatech/circuit-breaker-agents";

const toolBreaker = new CircuitBreaker({
  name: "web-search-tool",
  threshold: 3,           // 连续失败3次触发
  cooldown: 30_000,       // 冷却30秒
  fallback: mockSearch,   // 降级到Mock搜索
});

const providerBreaker = new CircuitBreaker({
  name: "openai-provider",
  threshold: 5,
  cooldown: 60_000,
  fallback: anthropicProvider,  // 降级到备用Provider
});

const capabilityBreaker = new CircuitBreaker({
  name: "web-browsing-capability",
  threshold: 3,  // 3个搜索工具全部熔断后触发
  cooldown: 120_000,
  fallback: () => "I cannot browse the web right now. Using cached knowledge.",
});

每个Agent步骤 = Forward Action + Compensating Action

任务执行：
  Agent A: Forward(创建订单) → Agent B: Forward(扣减库存) → 
  Agent C: Forward(发送确认邮件) ✓ 完成

失败补偿：
  Agent A: Forward(创建订单) → Agent B: Forward(扣减库存) → 
  Agent C: Forward(发送邮件) ✗ 失败 →
  Agent B: Compensate(恢复库存) → Agent A: Compensate(取消订单)

# A2A-SAGA补偿事务示例（概念模型）
from a2a_saga import SagaOrchestrator, SagaStep

saga = SagaOrchestrator(
    steps=[
        SagaStep(
            agent="order-agent",
            forward_action="create_order",
            compensate_action="cancel_order",
        ),
        SagaStep(
            agent="inventory-agent", 
            forward_action="deduct_inventory",
            compensate_action="restore_inventory",
        ),
        SagaStep(
            agent="notification-agent",
            forward_action="send_confirmation_email",
            compensate_action="send_cancellation_email",
        ),
    ],
    compensation_order="reverse",  # 反向补偿
    timeout=30,  # 单步超时
    on_compensation_failure="alert_human",  # 补偿失败的兜底
)

result = await saga.execute(customer_order)
if result.status == "compensated":
    alert_team(f"订单流程部分失败，已自动补偿。需人工检查：{result.details}")

Root Span（用户请求）
  └── Agent Span（路由Agent）
        ├── LLM Span（GPT-4o调用）
        │     ├── Input Tokens: 1,204
        │     ├── Output Tokens: 87
        │     └── Latency: 1.2s
        ├── Tool Span（Web搜索）
        │     ├── Query: "2026量子计算突破"
        │     ├── Results: 5
        │     └── Latency: 3.1s
        └── Agent Span（委托给研究Agent）
              ├── LLM Span（Claude调用）
              └── Tool Span（PDF解析）

# AG2 OpenTelemetry 自动追踪配置
from ag2 import AssistantAgent
from ag2.tracing import OpenTelemetryTracer

# 启用OTel追踪
tracer = OpenTelemetryTracer(
    exporter="otlp",  # 导出到Jaeger/Grafana Tempo
    service_name="multi-agent-pipeline",
)

researcher = AssistantAgent(
    name="researcher",
    system_message="You are a research specialist.",
    llm_config={"model": "gpt-4o"},
)

writer = AssistantAgent(
    name="writer", 
    system_message="You are a technical writer.",
    llm_config={"model": "gpt-4o"},
)

# 自动产生层级Span：
# Root → researcher Agent Span → researcher LLM Span
#      → writer Agent Span → writer LLM Span
result = researcher.initiate_chat(
    writer,
    message="Research and write about quantum computing in 2026",
)

graph TB
    subgraph "AgentOps CI/CD 层"
        A1[Prompt版本管控] --> A2[Eval门控]
        A2 --> A3[Canary灰度]
        A3 --> A4[全量发布/回滚]
    end

    subgraph "运行时通信层"
        B1[SSE: Agent→用户]
        B2[WebSocket: Agent↔Agent]
        B3[Streamable HTTP: Agent→MCP工具]
    end

    subgraph "弹性防护层"
        C1[Tool级熔断] --> C2[Provider级熔断] --> C3[Capability级熔断]
    end

    subgraph "事务补偿层"
        D1[Saga编排器] --> D2[Forward执行链]
        D2 -- 成功 --> D3[事务完成]
        D2 -- 失败 --> D4[Compensate补偿链]
        D4 --> D5[人工兜底]
    end

    subgraph "可观测性层"
        E1[Span层级追踪] --> E2[因果调试分析]
        E2 --> E3[实时告警]
        E3 --> E4[反馈CI/CD门控]
    end

    A4 --> B2
    B2 --> C1
    C3 --> D1
    D3 --> E1
    E4 -.-> A1