大模型应用开发中的可观测性实战：从日志到调用链，构建 LLM 应用的可靠性工程

import json
import logging
from pythonjsonlogger import jsonlogger

class LLMJsonFormatter(jsonlogger.JsonFormatter):
    def add_fields(self, log_record, record, message_dict):
        super().add_fields(log_record, record, message_dict)
        if not log_record.get('timestamp'):
            log_record['timestamp'] = datetime.utcnow().isoformat()
        # 强制注入全局 trace_id（需从 context 中取）
        log_record['trace_id'] = get_current_trace_id() or 'unknown'

logger = logging.getLogger('llm_app')
handler = logging.StreamHandler()
handler.setFormatter(LLMJsonFormatter())
logger.addHandler(handler)

# 使用示例
logger.info(
    "retrieval_completed",
    extra={
        "step": "retrieve",
        "query": user_query,
        "top_doc_ids": [d.metadata['id'] for d in docs],
        "top_scores": [s for _, s in docs_with_scores],
        "cost_ms": elapsed
    }
)

# docker-compose 部分
services:
  app:
    image: my-llm-app
    environment:
      - OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318
      - OTEL_SERVICE_NAME=llm-qa
  otel-collector:
    image: otel/opentelemetry-collector-contrib:latest
    volumes:
      - ./otel-config.yaml:/etc/otel/config.yaml
    command: ["--config=/etc/otel/config.yaml"]
  jaeger:
    image: jaegertracing/all-in-one:latest
    ports:
      - "16686:16686"

from opentelemetry import trace
tracer = trace.get_tracer(__name__)

@app.post("/ask")
async def ask(query: str):
    with tracer.start_as_current_span("full_pipeline") as parent_span:
        parent_span.set_attribute("query", query)

        with tracer.start_as_current_span("retrieve") as span:
            docs = retriever.get_relevant_documents(query)
            span.set_attribute("doc_count", len(docs))

        with tracer.start_as_current_span("rerank") as span:
            top_docs = reranker.rerank(query, docs, top_n=3)
            span.set_attribute("top_scores", [d.score for d in top_docs])

        with tracer.start_as_current_span("llm_generate") as span:
            answer = llm.generate(query, top_docs)
            span.set_attribute("token_usage", answer.usage)

        return answer

# evaluator.py
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_relevancy

def online_evaluate(sample: dict):
    # sample 包含 question, answer, contexts, ground_truth (若有)
    score = evaluate(
        dataset=[sample],
        metrics=[faithfulness, answer_relevancy, context_relevancy]
    )
    # 推送至 Prometheus
    for metric_name, value in score.items():
        llm_metric.labels(metric=metric_name).set(value)
    return score

from prometheus_client import Gauge, generate_latest

llm_metric = Gauge('llm_qa_ragas_score', 'RAGAS evaluation score', ['metric'])

@app.get("/metrics")
def metrics():
    return Response(generate_latest(), media_type="text/plain")

import hashlib
import redis
from sentence_transformers import SentenceTransformer

cache = redis.Redis(host='localhost', decode_responses=True)
encoder = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

def get_cached_answer(query):
    # 对 query 做语义哈希（如用 embedding 的前 10 维作为 key）
    emb = encoder.encode(query)
    key = hashlib.md5(emb[:10].tobytes()).hexdigest()
    cached = cache.get(f"qa:{key}")
    if cached:
        logger.info("cache_hit", extra={"query": query})
        return json.loads(cached)
    return None