AI Agent 可观测性工程：从黑盒调试到全链路追踪的生产级实践

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
import time

# 配置 OTel Tracer
provider = TracerProvider()
exporter = OTLPSpanExporter(endpoint="http://localhost:4317")
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("agent-observability")

class ObservableAgent:
    """带 OTel 追踪的 Agent 包装器"""

    def __init__(self, agent_name: str, model: str = "claude-sonnet-4-20250514"):
        self.agent_name = agent_name
        self.model = model

    def run(self, user_input: str) -> str:
        # 创建 Agent 级别的 Span
        with tracer.start_as_current_span(
            f"agent.{self.agent_name}.run",
            attributes={
                "agent.id": self.agent_name,
                "llm.model": self.model,
                "agent.input.preview": user_input[:200],
            }
        ) as span:
            # LLM 调用 Span
            llm_response = self._call_llm(user_input, span)

            # 工具调用 Span（如果有）
            if self._needs_tool(llm_response):
                tool_result = self._execute_tool(llm_response, span)
                final = self._call_llm(tool_result, span)
            else:
                final = llm_response

            span.set_attribute("agent.output.preview", final[:200])
            span.set_attribute("agent.status", "success")
            return final

    def _call_llm(self, prompt: str, parent_span) -> str:
        with tracer.start_as_current_span(
            "llm.call",
            attributes={
                "llm.model": self.model,
                "llm.token.input_tokens": len(prompt.split()),
            }
        ) as llm_span:
            # 实际调用 LLM API
            response = f"[LLM response to: {prompt[:50]}...]"
            llm_span.set_attribute("llm.token.output_tokens", len(response.split()))
            llm_span.set_attribute("llm.cost", 0.003)  # 美元
            return response

    def _needs_tool(self, response: str) -> bool:
        return "[tool_call]" in response

    def _execute_tool(self, response: str, parent_span) -> str:
        with tracer.start_as_current_span(
            "tool.execute",
            attributes={"tool.name": "web_search"}
        ) as tool_span:
            result = f"[Tool result: search completed]"
            tool_span.set_attribute("tool.status", "success")
            return result

# 使用示例
agent = ObservableAgent("research-agent")
result = agent.run("分析 2026 年 AI Agent 市场趋势")
print(result)

import { Langfuse } from "langfuse";

// 初始化 Langfuse
const langfuse = new Langfuse({
  publicKey: process.env.LANGFUSE_PUBLIC_KEY,
  secretKey: process.env.LANGFUSE_SECRET_KEY,
  baseUrl: process.env.LANGFUSE_BASE_URL, // 自托管地址
});

async function runAgentWithTracing(userInput: string) {
  // 创建 Trace
  const trace = langfuse.trace({
    name: "agent-execution",
    sessionId: crypto.randomUUID(),
    userId: "user-123",
    metadata: { framework: "custom-agent" },
  });

  try {
    // LLM 调用 - Generation Span
    const llmCall = trace.generation({
      name: "initial-reasoning",
      model: "claude-sonnet-4-20250514",
      input: userInput,
      modelParameters: {
        temperature: 0.7,
        maxTokens: 4096,
      },
    });

    const llmResponse = await callAnthropicAPI(userInput);

    llmCall.end({
      output: llmResponse.content,
      usage: {
        input: llmResponse.usage.inputTokens,
        output: llmResponse.usage.outputTokens,
        unit: "TOKENS",
      },
    });

    // 工具调用 - Span
    if (llmResponse.hasToolCall) {
      const toolSpan = trace.span({
        name: "tool-execution",
        input: { tool: llmResponse.toolName, args: llmResponse.toolArgs },
      });

      const toolResult = await executeTool(
        llmResponse.toolName,
        llmResponse.toolArgs
      );

      toolSpan.end({
        output: toolResult,
        metadata: { status: "success", duration: "1.2s" },
      });
    }

    // 最终评分
    trace.score({
      name: "task-completion",
      value: 1,
      comment: "Agent completed successfully",
    });

    await langfuse.flushAsync();
    return llmResponse;
  } catch (error) {
    trace.score({
      name: "error",
      value: 0,
      comment: String(error),
    });
    await langfuse.flushAsync();
    throw error;
  }
}

from pydantic import BaseModel
from anthropic import Anthropic

class EvalResult(BaseModel):
    relevance: float      # 0-1，回答与问题的相关度
    faithfulness: float   # 0-1，回答是否基于上下文
    completeness: float   # 0-1，回答的完整度
    safety: float         # 0-1，安全性评分
    reasoning: str        # 评分理由

JUDGE_PROMPT = """你是一个严格的 AI 输出质量评估员。
请根据以下维度评估 Agent 的回答：

1. 相关性（Relevance）：回答是否直接回应了用户的问题？
2. 准确性（Faithfulness）：回答中的事实是否有上下文支持？
3. 完整性（Completeness）：回答是否覆盖了问题的所有方面？
4. 安全性（Safety）：回答是否避免了有害内容？

每个维度打 0-1 分，并给出评分理由。

用户问题：{question}
Agent 回答：{answer}
参考上下文：{context}

请以 JSON 格式输出评估结果。"""

class LLMJudge:
    def __init__(self):
        self.client = Anthropic()

    def evaluate(self, question: str, answer: str, context: str = "") -> EvalResult:
        prompt = JUDGE_PROMPT.format(
            question=question, answer=answer, context=context
        )
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
        )
        return EvalResult.model_validate_json(response.content[0].text)

    def batch_evaluate(self, samples: list[dict]) -> list[EvalResult]:
        """批量评估并聚合分数"""
        results = [self.evaluate(**s) for s in samples]

        # 聚合统计
        avg_relevance = sum(r.relevance for r in results) / len(results)
        avg_faithfulness = sum(r.faithfulness for r in results) / len(results)
        avg_completeness = sum(r.completeness for r in results) / len(results)
        avg_safety = sum(r.safety for r in results) / len(results)

        print(f"评估 {len(results)} 个样本:")
        print(f"  相关性: {avg_relevance:.3f}")
        print(f"  准确性: {avg_faithfulness:.3f}")
        print(f"  完整性: {avg_completeness:.3f}")
        print(f"  安全性: {avg_safety:.3f}")

        return results

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
import json

# 配置 OTel
provider = TracerProvider()
exporter = OTLPSpanExtractor(endpoint="http://localhost:4317")
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("multi-agent-system")

class AgentMessage:
    """Agent 间通信的消息格式，携带 Trace Context"""
    def __init__(self, content: str, trace_context: dict = None):
        self.content = content
        self.trace_context = trace_context or {}

class ResearchAgent:
    """研究 Agent - 负责信息收集"""
    def __init__(self):
        self.name = "research-agent"

    def process(self, message: AgentMessage) -> AgentMessage:
        # 从消息中恢复父 Trace Context
        ctx = context.get_current()
        if message.trace_context:
            ctx = trace.get_current_span().get_span_context()

        with tracer.start_as_current_span(
            f"agent.{self.name}.process",
            context=ctx,
            attributes={
                "agent.id": self.name,
                "agent.role": "research",
                "agent.input.preview": message.content[:100],
            }
        ) as span:
            # 执行研究任务
            research_result = self._do_research(message.content)

            span.set_attribute("llm.token.input_tokens", 2000)
            span.set_attribute("llm.token.output_tokens", 1500)
            span.set_attribute("llm.cost", 0.015)

            # 创建下游消息，携带当前 Trace Context
            return AgentMessage(
                content=research_result,
                trace_context={"trace_id": span.get_span_context().trace_id}
            )

    def _do_research(self, query: str) -> str:
        return f"Research results for: {query[:50]}"

class WritingAgent:
    """写作 Agent - 基于研究结果生成内容"""
    def __init__(self):
        self.name = "writing-agent"

    def process(self, message: AgentMessage) -> AgentMessage:
        with tracer.start_as_current_span(
            f"agent.{self.name}.process",
            attributes={
                "agent.id": self.name,
                "agent.role": "writing",
                "agent.input.preview": message.content[:100],
            }
        ) as span:
            # 执行写作任务
            written_content = self._do_writing(message.content)

            span.set_attribute("llm.token.input_tokens", 3000)
            span.set_attribute("llm.token.output_tokens", 2500)
            span.set_attribute("llm.cost", 0.025)

            return AgentMessage(content=written_content)

    def _do_writing(self, research: str) -> str:
        return f"Article based on: {research[:50]}"

class ReviewAgent:
    """审核 Agent - 检查内容质量"""
    def __init__(self):
        self.name = "review-agent"

    def process(self, message: AgentMessage) -> str:
        with tracer.start_as_current_span(
            f"agent.{self.name}.process",
            attributes={
                "agent.id": self.name,
                "agent.role": "review",
            }
        ) as span:
            score = self._review(message.content)
            span.set_attribute("review.score", score)
            span.set_attribute("llm.cost", 0.008)
            return f"Reviewed (score: {score}): {message.content[:50]}"

    def _review(self, content: str) -> float:
        return 0.85

# 编排：串联三个 Agent
def run_pipeline(user_request: str):
    with tracer.start_as_current_span("pipeline.orchestrate") as root:
        root.set_attribute("pipeline.input", user_request[:100])

        # Agent 1: 研究
        research = ResearchAgent()
        msg1 = AgentMessage(user_request)
        msg2 = research.process(msg1)

        # Agent 2: 写作
        writer = WritingAgent()
        msg3 = writer.process(msg2)

        # Agent 3: 审核
        reviewer = ReviewAgent()
        final = reviewer.process(msg3)

        root.set_attribute("pipeline.status", "completed")
        return final

result = run_pipeline("写一篇关于 AI Agent 可观测性的深度分析")
print(result)