AI 可观测性实践：让你的 Agent 透明可见

from dataclasses import dataclass
from typing import Optional

@dataclass
class LLMCallMetrics:
    # 延迟指标
    time_to_first_token: float  # 首 token 延迟(ms)
    total_latency: float        # 总延迟(ms)
    # 成本指标
    prompt_tokens: int
    completion_tokens: int
    total_cost: float
    # 质量指标
    model_name: str
    temperature: float
    finish_reason: str
    # 检索指标(RAG场景)
    retrieval_count: Optional[int] = None
    retrieval_relevance: Optional[float] = None
    context_utilization: Optional[float] = None

@dataclass
class AgentTrace:
    trace_id: str
    spans: list[dict]  # 每个步骤一个 span
    # span 结构: name, start, end, input, output, metadata

from langfuse import Langfuse
from langfuse.decorators import langfuse_context, observe
import functools, time

langfuse = Langfuse(
    public_key="pk-lf-xxx",
    secret_key="sk-lf-xxx",
    host="https://your-langfuse.com"
)

def trace_llm(name="llm_call"):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            start = time.time()
            span = langfuse_context.start_span(name=name, input=kwargs)
            try:
                result = await func(*args, **kwargs)
                ttft = time.time() - start
                span.end(
                    output=result,
                    metadata={"ttft_ms": ttft * 1000, "model": kwargs.get("model", "unknown")}
                )
                langfuse_context.update_current_observation(
                    usage={
                        "prompt_tokens": result.get("usage", {}).get("prompt_tokens", 0),
                        "completion_tokens": result.get("usage", {}).get("completion_tokens", 0),
                        "total_tokens": result.get("usage", {}).get("total_tokens", 0),
                    },
                    metadata={"latency_ms": (time.time() - start) * 1000}
                )
                return result
            except Exception as e:
                span.end(level="ERROR", status_message=str(e))
                raise
        return wrapper
    return decorator

@observe(name="rag_pipeline")
async def rag_pipeline(query: str):
    docs = await retrieve(query)
    answer = await generate(query, docs)
    return answer

@observe(name="retrieve")
async def retrieve(query: str):
    results = await vector_search(query)
    return {"query": query, "docs": results}

@trace_llm(name="generate")
async def generate(query: str, docs: list):
    resp = await llm_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"Q: {query}\nContext: {docs}"}]
    )
    return resp