skb3002/agenthub/llm-observability

Trace          — one end-to-end user interaction (may span minutes)
  └── Span     — one logical step (retrieval, LLM call, tool call)
        └── Event — one observation (token count, score)

Self-hosted + full control?        → Langfuse
LangChain stack, managed?          → LangSmith
Eval-heavy research loop?          → Arize Phoenix
Polyglot / infra team owns APM?    → OpenTelemetry GenAI
Zero integration lift?             → Helicone (proxy)

from langfuse.decorators import observe, langfuse_context
@observe()
async def answer_question(question: str) -> str:
    retrieved = await retrieve(question)
    response = await call_llm(question, retrieved)
    return response
@observe(as_type="generation")
async def call_llm(question: str, context: list[str]) -> str:
    langfuse_context.update_current_observation(
        model="gemini-2.0-flash",
        input={"question": question, "context": context},
    )
    result = await llm.generate(...)
    langfuse_context.update_current_observation(
        usage={"input": result.prompt_tokens, "output": result.completion_tokens},
        output=result.text,
    )
    return result.text

trace = langfuse.trace(name="answer", user_id=user_id, session_id=session_id)
generation = trace.generation(
    name="gemini-call",
    model="gemini-2.0-flash",
    input=prompt,
    metadata={"temperature": 0.2},
)
# ... make the call ...
generation.end(
    output=response.text,
    usage={"input": n_in, "output": n_out},
)

from opentelemetry import trace
tracer = trace.get_tracer("llm")
with tracer.start_as_current_span("llm.call") as span:
    span.set_attribute("gen_ai.system", "google")
    span.set_attribute("gen_ai.request.model", "gemini-2.0-flash")
    span.set_attribute("gen_ai.request.temperature", 0.2)
    response = await client.generate(prompt)
    span.set_attribute("gen_ai.response.finish_reasons", ["stop"])
    span.set_attribute("gen_ai.usage.input_tokens", response.prompt_tokens)
    span.set_attribute("gen_ai.usage.output_tokens", response.completion_tokens)

COST_PER_1K = {
    "gemini-2.0-flash": {"input": 0.000075, "output": 0.0003},
    "gpt-4o":           {"input": 0.0025,   "output": 0.01},
    "claude-opus-4-7":  {"input": 0.015,    "output": 0.075},
}
def cost(model: str, in_tokens: int, out_tokens: int) -> float:
    rates = COST_PER_1K[model]
    return (in_tokens / 1000) * rates["input"] + (out_tokens / 1000) * rates["output"]

Pattern:
├── Sample 1-5% of production traces
├── Run LLM-as-judge evaluator
├── Store score as trace score
└── Alert on score regression (e.g. factuality drops from 0.9 → 0.7)

def mask_pii(text: str) -> str:
    text = re.sub(r"[\w.-]+@[\w.-]+", "<EMAIL>", text)
    text = re.sub(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b", "<PHONE>", text)
    return text
langfuse_context.update_current_observation(
    input=mask_pii(user_message),
)

trace: "answer_question"
├── span: "retrieve"
│     ├── event: "embed_query" (tokens, latency)
│     ├── event: "vector_search" (k, latency, hits)
│     └── event: "rerank" (model, scores)
├── span: "generate"
│     ├── attr: model, temperature
│     └── attr: tokens_in/out, cost
└── span: "evaluate" (optional, async)

1. Write an observability ABC (abstract base class) in your app.
2. Each platform = one concrete implementation.
3. Swap via config, run both in parallel for 1-2 weeks.
4. Compare traces, confirm parity, cut over.