Observability & Tracing

# implement rag_pipeline, retrieve_documents, generate_answer
# Key operations: retrieval pipeline, vector search, RAG pipeline
from langfuse.decorators import observe, langfuse_context
from langfuse.openai import openai # drop-in replacement with tracing

# The @observe decorator automatically creates traces and spans
@observe()
def rag_pipeline(query: str) -> str:
 """RAG pipeline with automatic Langfuse tracing."""
 # Each decorated function becomes a span in the trace
 context_docs = retrieve_documents(query)
 answer = generate_answer(query, context_docs)
 return answer

@observe()
def retrieve_documents(query: str) -> list[str]:
 """Retrieve relevant documents from vector store."""
 # Add custom metadata to the current span
 langfuse_context.update_current_observation(
 metadata={"collection": "knowledge_base_v2", "top_k": 5}
 )
 # Retrieval logic here...
 docs = ["Document 1 content...", "Document 2 content..."]
 return docs

@observe()
def generate_answer(query: str, docs: list[str]) -> str:
 """Generate answer using retrieved context."""
 context = "\n".join(docs)
 # The Langfuse OpenAI wrapper auto-traces this call
 response = openai.chat.completions.create(
 model="gpt-4o",
 messages=[
 {"role": "system", "content": f"Answer based on context:\n{context}"},
 {"role": "user", "content": query},
 ],
 temperature=0,
 )
 return response.choices[0].message.content

# Traces appear automatically in the Langfuse dashboard
result = rag_pipeline("What are the benefits of RAG?")

# implement answer_question, search_knowledge_base, call_llm
# Key operations: retrieval pipeline, vector search, RAG pipeline
from langsmith import traceable, Client
from openai import OpenAI
import os

# Enable LangSmith tracing via environment variable
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "my-rag-app"

client = OpenAI()

@traceable(run_type="chain")
def answer_question(question: str) -> dict:
 """Answer a question with LangSmith tracing."""
 docs = search_knowledge_base(question)
 response = call_llm(question, docs)
 return {"answer": response, "sources": docs}

@traceable(run_type="retriever")
def search_knowledge_base(query: str) -> list[str]:
 """Search vector store for relevant documents."""
 # Retrieval logic here...
 return ["Relevant document content..."]

@traceable(run_type="llm")
def call_llm(question: str, context_docs: list[str]) -> str:
 """Call LLM with retrieved context."""
 context = "\n".join(context_docs)
 response = client.chat.completions.create(
 model="gpt-4o",
 messages=[
 {"role": "system", "content": f"Use this context:\n{context}"},
 {"role": "user", "content": question},
 ],
 )
 return response.choices[0].message.content

# Define JSONFormatter; implement format, log_llm_call, wrapper
# See inline comments for step-by-step details.
import logging
import json
import time
from datetime import datetime, timezone
from functools import wraps

# Configure JSON structured logging
class JSONFormatter(logging.Formatter):
 def format(self, record):
 log_data = {
 "timestamp": datetime.now(timezone.utc).isoformat(),
 "level": record.levelname,
 "message": record.getMessage(),
 }
 if hasattr(record, "llm_data"):
 log_data.update(record.llm_data)
 return json.dumps(log_data)

logger = logging.getLogger("llm_app")
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)

def log_llm_call(func):
 """Decorator to log LLM calls with structured metadata."""
 @wraps(func)
 def wrapper(*args, **kwargs):
 start = time.time()
 try:
 result = func(*args, **kwargs)
 latency_ms = (time.time() - start) * 1000

 record = logger.makeRecord(
 "llm_app", logging.INFO, "", 0,
 f"LLM call: {func.__name__}", (), None
 )
 record.llm_data = {
 "event": "llm_call",
 "function": func.__name__,
 "latency_ms": round(latency_ms, 1),
 "status": "success",
 }
 logger.handle(record)
 return result

 except Exception as e:
 latency_ms = (time.time() - start) * 1000
 record = logger.makeRecord(
 "llm_app", logging.ERROR, "", 0,
 f"LLM call failed: {func.__name__}", (), None
 )
 record.llm_data = {
 "event": "llm_call_error",
 "function": func.__name__,
 "latency_ms": round(latency_ms, 1),
 "error": str(e),
 }
 logger.handle(record)
 raise
 return wrapper

# Define AlertRule, LLMAlertManager; implement __init__, check_metrics
# Key operations: evaluation logic
from dataclasses import dataclass
from typing import Callable, Optional

@dataclass
class AlertRule:
 """Definition of an alerting rule for LLM monitoring."""
 name: str
 metric: str
 threshold: float
 comparison: str # "gt" (greater than) or "lt" (less than)
 window_minutes: int = 15
 severity: str = "warning" # warning, critical
 notify_channel: str = "slack"

class LLMAlertManager:
 """Manages alerting rules for LLM applications."""

 DEFAULT_RULES = [
 AlertRule("High Latency", "latency_p95_ms", 5000, "gt", severity="warning"),
 AlertRule("Error Rate", "error_rate", 0.01, "gt", severity="critical"),
 AlertRule("Cost Spike", "cost_per_request_usd", 0.10, "gt", severity="warning"),
 AlertRule("Low Faithfulness", "faithfulness_score", 0.7, "lt", severity="critical"),
 AlertRule("Hallucination Spike", "hallucination_rate", 0.05, "gt", severity="critical"),
 ]

 def __init__(self, rules: Optional[list[AlertRule]] = None):
 self.rules = rules or self.DEFAULT_RULES

 def check_metrics(self, current_metrics: dict) -> list[dict]:
 """Evaluate all rules against current metrics."""
 fired = []
 for rule in self.rules:
 if rule.metric not in current_metrics:
 continue
 value = current_metrics[rule.metric]
 triggered = (
 (rule.comparison == "gt" and value > rule.threshold)
 or (rule.comparison == "lt" and value < rule.threshold)
 )
 if triggered:
 fired.append({
 "alert": rule.name,
 "severity": rule.severity,
 "metric": rule.metric,
 "value": value,
 "threshold": rule.threshold,
 })
 return fired