What is LLM Observability?
The Engineer's Practical Guide

# Clone and start Langfuse locally
git clone https://github.com/langfuse/langfuse.git
cd langfuse
docker compose up -d

# Langfuse UI will be at http://localhost:3000
# Create a project and grab your API keys

# Install the Python SDK
pip install langfuse openai

from langfuse import Langfuse
from langfuse.openai import openai  # drop-in replacement
import os

# Init — reads LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_HOST from env
langfuse = Langfuse(
    public_key="pk-lf-...",
    secret_key="sk-lf-...",
    host="http://localhost:3000"  # or https://cloud.langfuse.com
)

# This single import swap gives you automatic tracing
# of every OpenAI call: prompt, response, tokens, latency, cost
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful MLOps assistant."},
        {"role": "user", "content": "What is MLflow used for?"}
    ],
    # Optional: tag this trace for filtering in the UI
    name="mlops-qa",
    metadata={"feature": "chat", "user_id": "u_123"}
)

print(response.choices[0].message.content)
# All trace data is now visible in Langfuse UI — zero extra code needed

from langfuse import Langfuse
from langfuse.openai import openai
from langfuse.decorators import langfuse_context, observe

langfuse = Langfuse()

# @observe creates a span for this function automatically
@observe()
def retrieve_chunks(query: str, top_k: int = 5) -> list:
    """Simulated vector store retrieval"""
    # In production: call your Chroma / Pinecone / Weaviate here
    chunks = [
        {"text": "MLflow is an open source platform for ML lifecycle management...", "score": 0.92},
        {"text": "MLflow Tracking logs parameters, metrics, and artifacts...", "score": 0.87},
    ]
    # Log retrieval metadata to the span
    langfuse_context.update_current_observation(
        input=query,
        output=chunks,
        metadata={"top_k": top_k, "chunk_count": len(chunks)}
    )
    return chunks

@observe()
def build_prompt(query: str, chunks: list) -> str:
    """Assemble the final prompt from query + retrieved context"""
    context = "\n\n".join([c["text"] for c in chunks])
    return f"""Answer using only the context below.

Context:
{context}

Question: {query}
Answer:"""

@observe()  # The root trace — wraps the whole pipeline
def rag_answer(query: str) -> str:
    # Step 1: retrieve — traced as a child span
    chunks = retrieve_chunks(query)
    
    # Step 2: build prompt — traced as a child span
    prompt = build_prompt(query, chunks)
    
    # Step 3: generate — traced via patched OpenAI client
    response = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}]
    )
    answer = response.choices[0].message.content
    
    # Step 4: score the output quality (0-1 scale)
    langfuse_context.score_current_trace(
        name="answer_quality",
        value=1.0,  # replace with your eval logic
        comment="Auto-scored: retrieval found relevant chunks"
    )
    
    return answer

# Run it
result = rag_answer("What is MLflow used for?")
print(result)

# Flush traces before script exits
langfuse.flush()

from openai import OpenAI

raw_client = OpenAI()  # unpatched — don't trace the judge calls

def evaluate_faithfulness(question: str, context: str, answer: str) -> float:
    """
    LLM-as-judge: score whether the answer is faithful to the retrieved context.
    Returns a score from 0.0 (hallucination) to 1.0 (fully grounded).
    """
    judge_prompt = f"""You are evaluating an AI assistant's answer for faithfulness.

RETRIEVED CONTEXT:
{context}

QUESTION: {question}

ANSWER: {answer}

Task: Score whether the answer is ONLY based on the retrieved context (not hallucinated).
Respond with JSON only: {{"score": 0.0-1.0, "reason": "brief explanation"}}
0.0 = completely hallucinated | 1.0 = fully grounded in context"""

    response = raw_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": judge_prompt}],
        response_format={"type": "json_object"}
    )
    
    import json
    result = json.loads(response.choices[0].message.content)
    return result["score"], result["reason"]

# Then post the score back to Langfuse for any trace
score, reason = evaluate_faithfulness(question, context, answer)

langfuse.score(
    trace_id="your-trace-id",  # from langfuse_context.get_current_trace_id()
    name="faithfulness",
    value=score,
    comment=reason
)

# When a user clicks 👍 or 👎, post feedback to Langfuse
# Store the trace_id alongside your response when you serve it

from langfuse import Langfuse
from langfuse.decorators import langfuse_context

langfuse = Langfuse()

def handle_user_feedback(trace_id: str, thumbs_up: bool, comment: str = None):
    """Record user feedback against the trace that generated the response"""
    langfuse.score(
        trace_id=trace_id,
        name="user_feedback",
        value=1 if thumbs_up else 0,
        comment=comment
    )

# Example: in your FastAPI endpoint
# @app.post("/feedback")
# async def feedback(trace_id: str, positive: bool, comment: str = None):
#     handle_user_feedback(trace_id, positive, comment)
#     return {"status": "recorded"}

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)
from datasets import Dataset

# Collect your RAG pipeline outputs
data = {
    "question":  ["What is MLflow used for?"],
    "answer":    ["MLflow is used for experiment tracking..."],
    "contexts":  [["MLflow is an open source platform...", "MLflow Tracking logs..."]],
    "ground_truth": ["MLflow manages the ML lifecycle including tracking..."]
}

dataset = Dataset.from_dict(data)

# Run RAGAS evaluation — gives you all 4 RAG metrics at once
result = evaluate(
    dataset=dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall]
)

print(result)
# {'faithfulness': 0.92, 'answer_relevancy': 0.88,
#  'context_precision': 0.94, 'context_recall': 0.81}

# Then post these scores to Langfuse for the corresponding trace