Langfuse : Plateforme open source d'observabilité et monitoring pour applications LLM

pip install langfuse

import os
from dotenv import load_dotenv

load_dotenv()

# Configuration Langfuse Cloud
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."  # Clé publique
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."  # Clé secrète
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"  # Optionnel (cloud par défaut)

# docker-compose.yml
version: '3.8'

services:
  langfuse-server:
    image: langfuse/langfuse:latest
    ports:
      - "3000:3000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/langfuse
      - NEXTAUTH_URL=http://localhost:3000
      - NEXTAUTH_SECRET=your-secret-key
      - SALT=your-salt-key
    depends_on:
      - db

  db:
    image: postgres:15
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
      - POSTGRES_DB=langfuse
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

docker-compose up -d

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langfuse.callback import CallbackHandler

# Initialiser le callback Langfuse
langfuse_handler = CallbackHandler()

# Créer votre chain
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
prompt = ChatPromptTemplate.from_template("Explique {concept} de manière simple")
chain = prompt | llm | StrOutputParser()

# Invoquer avec tracing Langfuse
result = chain.invoke(
    {"concept": "la blockchain"},
    config={"callbacks": [langfuse_handler]}
)

print(result)

from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_community.tools import DuckDuckGoSearchRun
from langchain.tools import Tool
from langfuse.callback import CallbackHandler

# Outils
search = DuckDuckGoSearchRun()
tools = [
    Tool(
        name="Search",
        func=search.run,
        description="Recherche sur le web"
    )
]

# Prompt
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
prompt = ChatPromptTemplate.from_messages([
    ("system", "Tu es un assistant de recherche expert."),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad")
])

# Agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Tracing avec Langfuse
langfuse_handler = CallbackHandler()

result = agent_executor.invoke(
    {"input": "Quels sont les derniers développements en IA générative ?"},
    config={"callbacks": [langfuse_handler]}
)

from langfuse.callback import CallbackHandler

# Créer un handler global
langfuse_handler = CallbackHandler()

# Configurer au niveau session
from langchain.globals import set_llm_cache
from langchain.callbacks import get_callback_manager

# Toutes les exécutions suivantes seront tracées
callback_manager = get_callback_manager()
callback_manager.add_handler(langfuse_handler)

Tu es un agent de support client pour {{company}}.

Question client : {{question}}

Réponds de manière professionnelle et empathique.

from langfuse import Langfuse

langfuse = Langfuse()

# Récupérer le prompt (avec cache)
prompt = langfuse.get_prompt("support-email-response")

# Compiler avec variables
compiled_prompt = prompt.compile(
    company="TechCorp",
    question="Comment annuler mon abonnement ?"
)

# Utiliser avec LLM
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke(compiled_prompt)

from langfuse import Langfuse

langfuse = Langfuse()

# Récupérer une trace
trace = langfuse.get_trace("trace-id")

# Évaluer la qualité avec GPT-4
from langfuse.model import CreateScore

langfuse.score(
    trace_id=trace.id,
    name="helpfulness",
    value=0.9,  # Score automatique
    comment="Réponse claire et complète",
    data_type="NUMERIC"
)

# Évaluer toutes les traces d'un dataset
dataset = langfuse.get_dataset("customer-support-qa")

for item in dataset.items:
    # Exécuter votre chain
    result = chain.invoke(item.input)

    # Comparer avec expected output
    is_correct = evaluate_similarity(result, item.expected_output)

    # Logger l'évaluation
    langfuse.score(
        trace_id=item.trace_id,
        name="correctness",
        value=1.0 if is_correct else 0.0
    )

# Créer une annotation queue dans l'interface
# Nom : "quality-review"
# Filter : traces avec score < 0.7

# Les annotateurs peuvent :
# - Réviser les traces
# - Ajouter des labels
# - Corriger les réponses
# - Créer des datasets pour fine-tuning

from langfuse import Langfuse

langfuse = Langfuse()

# Créer un dataset
dataset = langfuse.create_dataset(
    name="rag-qa-eval",
    description="Questions-réponses pour évaluation RAG"
)

# Ajouter des items
dataset.create_item(
    input={"query": "Qu'est-ce que le RAG ?"},
    expected_output="RAG (Retrieval-Augmented Generation) combine..."
)

# Tester votre chain sur le dataset
for item in dataset.items:
    result = rag_chain.invoke(item.input)

    # Comparer avec expected
    similarity_score = calculate_similarity(result, item.expected_output)

    langfuse.score(
        trace_id=item.trace_id,
        name="similarity",
        value=similarity_score
    )

# Configurer des alertes via API
langfuse.create_alert(
    name="high-latency",
    condition="avg_latency > 3000",  # Plus de 3s
    notification_channel="email",
    recipients=["[email protected]"]
)

langfuse.create_alert(
    name="low-quality",
    condition="score_helpfulness < 0.5",
    notification_channel="slack",
    webhook_url="https://hooks.slack.com/..."
)

# Exporter données vers votre data warehouse
traces = langfuse.get_traces(
    from_timestamp="2025-01-01",
    to_timestamp="2025-01-31",
    limit=10000
)

# Convertir en DataFrame pour analyse
import pandas as pd

df = pd.DataFrame([
    {
        "id": t.id,
        "timestamp": t.timestamp,
        "model": t.model,
        "tokens": t.usage.total_tokens,
        "cost": t.calculated_total_cost,
        "latency": t.latency_ms,
        "user_id": t.user_id
    }
    for t in traces
])

# Analyses custom
df.groupby("model").agg({
    "cost": "sum",
    "latency": "mean",
    "tokens": "sum"
})

from langfuse.openai import openai

# Drop-in replacement pour OpenAI SDK
client = openai.OpenAI()

completion = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Explique la photosynthèse"}],
    # Métadonnées Langfuse
    langfuse_user_id="user-123",
    langfuse_session_id="session-456",
    langfuse_tags=["biology", "education"]
)

# Automatiquement tracé dans Langfuse !

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.callbacks import CallbackManager
from langfuse.llama_index import LlamaIndexCallbackHandler

# Configurer callback
langfuse_handler = LlamaIndexCallbackHandler()
callback_manager = CallbackManager([langfuse_handler])

# Créer index
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(
    documents,
    callback_manager=callback_manager
)

# Interroger
query_engine = index.as_query_engine()
response = query_engine.query("Résume les documents")

from langfuse import Langfuse

langfuse = Langfuse()

# Tracer manuellement
trace = langfuse.trace(
    name="custom-rag-pipeline",
    user_id="user-123",
    metadata={"env": "production"}
)

# Span 1 : Retrieval
with trace.span(name="retrieval") as span:
    docs = vector_store.search(query, k=5)
    span.update(
        output={"num_docs": len(docs)},
        metadata={"similarity_threshold": 0.8}
    )

# Span 2 : Reranking
with trace.span(name="reranking") as span:
    reranked_docs = reranker.rerank(docs, query)
    span.update(output={"num_docs": len(reranked_docs)})

# Span 3 : Generation
with trace.span(name="generation") as span:
    response = llm.generate(
        prompt=f"Context: {reranked_docs}\n\nQuestion: {query}"
    )
    span.update(
        output={"response": response},
        model="gpt-4",
        usage={"total_tokens": 1500},
        calculated_cost_usd=0.03
    )

# Tracker coûts par utilisateur
trace = langfuse.trace(
    name="user-query",
    user_id="user-123",
    tags=["tier-premium"],  # Segmenter par tier
)

# Analyser dans dashboard :
# - Coût moyen par utilisateur
# - Utilisateurs les plus coûteux
# - Coûts par feature

import random

# Tracer seulement 10% en production (réduire coûts stockage)
if random.random() < 0.1 or user_is_internal:
    config = {"callbacks": [langfuse_handler]}
else:
    config = {}

result = chain.invoke(input, config=config)

# Tracker version de votre app
trace = langfuse.trace(
    name="query",
    release="v2.3.1",  # Version de votre app
    metadata={
        "commit_sha": "abc123",
        "deployed_at": "2025-01-15T10:00:00Z"
    }
)

# Comparer performances entre versions

# Anonymiser données sensibles
def anonymize_pii(text: str) -> str:
    # Remplacer emails, téléphones, etc.
    return re.sub(r'\S+@\S+', '[EMAIL]', text)

# Tracer sans données sensibles
trace = langfuse.trace(
    name="support-query",
    input=anonymize_pii(user_query),
    output=anonymize_pii(bot_response)
)

# 1. Collecter feedback utilisateur
@app.post("/feedback")
def submit_feedback(trace_id: str, rating: int):
    langfuse.score(
        trace_id=trace_id,
        name="user-rating",
        value=rating / 5.0,  # Normaliser 1-5 → 0-1
        data_type="NUMERIC"
    )

# 2. Identifier prompts problématiques
# Dans dashboard : Filter traces where user-rating < 0.4

# 3. Créer dataset depuis traces mauvaises
dataset = langfuse.create_dataset("low-rated-queries")
for trace in low_rated_traces:
    dataset.create_item(
        input=trace.input,
        expected_output=corrected_response  # Corrigé manuellement
    )

# 4. Tester nouveau prompt
for item in dataset.items:
    result = new_chain.invoke(item.input)
    similarity = calculate_similarity(result, item.expected_output)
    langfuse.score(trace_id=item.trace_id, name="improvement", value=similarity)

# 5. Déployer si amélioration > 20%

import random

# 2 versions de prompt dans Langfuse
# - "email-response-v1" (actuel)
# - "email-response-v2" (nouveau)

langfuse = Langfuse()

# Sélectionner version aléatoirement
prompt_version = "v2" if random.random() < 0.5 else "v1"
prompt = langfuse.get_prompt(f"email-response-{prompt_version}")

# Tracer avec tag
trace = langfuse.trace(
    name="email-generation",
    tags=[f"prompt-{prompt_version}"]
)

# Après 1 semaine, comparer dans dashboard :
# - Avg user-rating par version
# - Coûts moyens
# - Latence
# - Taux de succès