21a - AgentOps 概念：AI Agent 的可观测性与运维

┌─────────────────────────────────────────────────────────┐
│                    AgentOps 核心能力                      │
├─────────────┬─────────────┬──────────────┬──────────────┤
│   可观测性   │   成本管理   │   质量保证    │   合规审计   │
├─────────────┼─────────────┼──────────────┼──────────────┤
│ • 分布式追踪 │ • Token 计量 │ • 幻觉检测   │ • 决策日志   │
│ • 推理链可视 │ • 成本归因   │ • 输出评估   │ • 操作审计   │
│ • 工具调用   │ • 预算告警   │ • 漂移检测   │ • 数据溯源   │
│   成功率监控 │ • 模型路由   │ • A/B 测试   │ • 合规报告   │
│ • 会话回放   │ • 缓存优化   │ • 用户反馈   │ • 权限追踪   │
└─────────────┴─────────────┴──────────────┴──────────────┘

传统软件：input → deterministic function → predictable output
AI Agent：input → LLM reasoning (temperature, context) → variable output
                    ↓
              需要统计级监控，而非单次断言

# AgentOps 成本追踪示例（概念代码）
@track_cost
def agent_workflow(user_query: str):
    # 每次 LLM 调用自动记录 Token 消耗
    plan = llm.plan(user_query)          # 输入: 500 tokens, 输出: 200 tokens
    result = llm.execute(plan)           # 输入: 800 tokens, 输出: 1500 tokens
    verified = llm.verify(result)        # 输入: 2000 tokens, 输出: 100 tokens
    
    # AgentOps 自动汇总：
    # 总 Token: 5,100 | 总成本: $0.015 | 归因: user_123 / workflow_research

用户查询: "帮我分析上周的销售数据并生成报告"
    │
    ├── Step 1: 规划 (Planning)
    │   └── 决策: 需要查询数据库 → 分析数据 → 生成图表 → 撰写报告
    │
    ├── Step 2: 工具调用 - SQL 查询
    │   ├── 工具: database_query
    │   ├── 参数: SELECT * FROM sales WHERE date >= '2025-07-01'
    │   ├── 结果: 1,247 行数据
    │   └── 耗时: 1.2s | Token: 350
    │
    ├── Step 3: 数据分析
    │   ├── LLM 调用: 分析销售趋势
    │   ├── 结果: 识别出 3 个关键趋势
    │   └── 耗时: 3.5s | Token: 2,100
    │
    ├── Step 4: 自我修正 ⚠️
    │   ├── 发现: 图表数据与原始数据不一致
    │   ├── 动作: 重新计算
    │   └── 耗时: 2.1s | Token: 800
    │
    └── Step 5: 生成报告
        ├── 输出: Markdown 格式报告
        └── 耗时: 4.2s | Token: 3,500
    
    总计: 11.0s | 6,750 tokens | $0.020 | 成功

┌──────────────────────────────────────────────────────────────┐
│                        应用层                                 │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐    │
│  │ Agent A  │  │ Agent B  │  │ Agent C  │  │ Agent D  │    │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘    │
│       │              │              │              │          │
│  ┌────▼──────────────▼──────────────▼──────────────▼────┐    │
│  │              AgentOps SDK / 插桩层                     │    │
│  │  • 自动捕获 LLM 调用  • 记录工具调用  • 追踪推理链    │    │
│  └────────────────────────┬─────────────────────────────┘    │
└───────────────────────────┼──────────────────────────────────┘
                            │ traces, metrics, logs
                            ▼
┌───────────────────────────────────────────────────────────────┐
│                    AgentOps 平台层                              │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐   │
│  │  数据收集    │  │  数据处理    │  │  存储               │   │
│  │  • 追踪收集  │→│  • 聚合计算  │→│  • 时序数据库       │   │
│  │  • 指标采集  │  │  • 评估打分  │  │  • 向量存储         │   │
│  │  • 日志接收  │  │  • 异常检测  │  │  • Prompt/Response  │   │
│  └─────────────┘  └─────────────┘  └─────────────────────┘   │
│                                                               │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐   │
│  │  可视化      │  │  告警        │  │  分析               │   │
│  │  • 仪表板    │  │  • 成本告警  │  │  • 趋势分析         │   │
│  │  • 追踪视图  │  │  • 质量告警  │  │  • 根因分析         │   │
│  │  • 会话回放  │  │  • 漂移告警  │  │  • 优化建议         │   │
│  └─────────────┘  └─────────────┘  └─────────────────────┘   │
└───────────────────────────────────────────────────────────────┘

# 只需修改 API base URL，一行代码接入
import openai
 
client = openai.OpenAI(
    api_key="sk-xxx",
    base_url="https://oai.helicone.ai/v1",  # 原来是 https://api.openai.com/v1
    default_headers={
        "Helicone-Auth": "Bearer hlc-xxx"
    }
)

from langfuse.decorators import observe, langfuse_context
 
@observe()
def research_agent(query: str):
    # Langfuse 自动追踪整个函数的执行链路
    plan = create_plan(query)
    
    for step in plan.steps:
        result = execute_step(step)
        langfuse_context.update_current_observation(
            metadata={"step": step.name, "tokens": result.token_count}
        )
    
    return compile_results()

你的 Agent 框架是什么？
├── LangChain → LangSmith（原生集成，30 分钟接入）
├── CrewAI → AgentOps.ai（官方推荐集成）
├── LlamaIndex → Arize Phoenix（深度集成）
├── 自研框架 → 继续判断 ↓
│
你最关心什么？
├── 快速启动 + 成本监控 → Helicone（15 分钟接入）
├── 数据主权 + 自托管 → Langfuse（开源可自托管）
├── 系统化评估 + A/B 测试 → Braintrust
├── 全栈可观测性 → Datadog LLM Observability
├── 安全合规 → Lunary
└── ML 训练 + Agent 监控 → W&B Weave

你是一位 AgentOps 架构师。请根据以下信息，为我的 AI Agent 系统设计监控方案：

## 系统信息
- Agent 类型：[客服 Agent / 编码助手 / 研究 Agent / 工作流 Agent]
- 使用的 LLM：[GPT-4o / Claude 3.5 / Gemini 2.5 Pro / 混合]
- Agent 框架：[LangChain / CrewAI / LlamaIndex / 自研]
- 日均请求量：[100 / 1,000 / 10,000 / 100,000+]
- 团队规模：[1-3 / 3-10 / 10+]
- 预算范围：[免费 / <$100/月 / <$500/月 / 不限]
- 数据主权要求：[无 / 需要自托管 / 需要合规认证]

## 请输出
1. 推荐的 AgentOps 平台（含理由）
2. 需要监控的核心指标（按优先级排序）
3. 告警规则建议
4. 成本优化策略
5. 分阶段实施路线图（MVP → 生产 → 规模化）

请为我的 [Agent 类型] 设计一个 AgentOps 监控仪表板，包含以下面板：

1. **概览面板**：关键指标卡片（成功率、平均延迟、日成本、活跃用户）
2. **成本面板**：Token 消耗趋势、按 Agent/用户/模型的成本分布
3. **质量面板**：幻觉率趋势、用户满意度、输出评分分布
4. **性能面板**：延迟分布（P50/P95/P99）、工具调用成功率
5. **告警面板**：活跃告警、历史告警、告警趋势

请为每个面板提供：
- 具体的指标定义和计算公式
- 推荐的可视化类型（折线图/柱状图/热力图/数字卡片）
- 告警阈值建议

我的 AI Agent 出现了以下问题：[描述问题]

请帮我按照以下步骤排查：

1. **症状分析**：这个问题属于哪类故障？（幻觉/延迟/成本/工具失败/循环）
2. **数据收集**：我需要查看哪些 AgentOps 指标和追踪数据？
3. **根因假设**：列出 3-5 个可能的根因
4. **验证步骤**：如何验证每个假设？
5. **修复方案**：针对最可能的根因，提供修复建议
6. **预防措施**：如何防止类似问题再次发生？

# pip install langfuse
from langfuse.decorators import observe
from langfuse import Langfuse
 
langfuse = Langfuse(
    public_key="pk-xxx",
    secret_key="sk-xxx",
    host="https://cloud.langfuse.com"  # 或自托管地址
)
 
@observe()
def customer_support_agent(user_message: str, session_id: str):
    # 1. 检索知识库
    context = retrieve_knowledge(user_message)
    
    # 2. 生成回复
    response = llm.chat(
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Context: {context}\n\nUser: {user_message}"}
        ]
    )
    
    # 3. 记录用户反馈
    langfuse.score(
        trace_id=langfuse_context.get_current_trace_id(),
        name="user_feedback",
        value=1  # 后续由用户评分更新
    )
    
    return response

# 在 Langfuse 仪表板中配置：
# 1. 成本追踪：自动计算每次交互的 Token 成本
# 2. 日预算告警：日成本 > $50 时发送 Slack 通知
# 3. 单次交互告警：单次交互 Token > 10,000 时标记异常

# 使用 LLM-as-Judge 自动评估输出质量
@observe()
def evaluate_response(user_query: str, agent_response: str, context: str):
    evaluation = llm.chat(
        model="gpt-4o-mini",  # 用便宜模型做评估
        messages=[{
            "role": "system",
            "content": """评估以下 AI 客服回复的质量，打分 1-5：
            - 准确性：回复是否基于提供的上下文？
            - 完整性：是否回答了用户的问题？
            - 安全性：是否包含不当内容？
            返回 JSON: {"accuracy": N, "completeness": N, "safety": N}"""
        }, {
            "role": "user",
            "content": f"用户问题: {user_query}\n上下文: {context}\nAgent回复: {agent_response}"
        }]
    )
    return evaluation