Langfuse 2026 : observabilité LLM complète — traces, évaluations et coûts en Python

Compatibilité SDK : les exemples de cet article s'appuient sur les APIs langfuse v3/v4 (Python 3.10+). Les noms de méthodes, imports et comportements peuvent différer selon la version installée. Vérifiez votre version avec pip show langfuse, ne copiez-collez pas sans tester, et consultez le changelog officiel avant toute mise en production.

Ce que vous allez apprendre : pourquoi l'observabilité est indispensable en production LLM, comment instrumenter votre code Python avec le décorateur @observe, intégrer Langfuse à LangChain et LangGraph, versionner vos prompts, évaluer automatiquement vos réponses avec LLM-as-judge, suivre les coûts token par token, et déployer Langfuse en self-hosted avec Docker Compose.

TL;DR — L'essentiel en 30 secondes

• Installer : pip install langfuse + 3 variables d'env (PUBLIC_KEY, SECRET_KEY, BASE_URL)
• Tracer : décorateur @observe() sur vos fonctions — zéro configuration supplémentaire
• Intégrer : CallbackHandler pour LangChain/LangGraph en une ligne
• Évaluer : LLM-as-judge automatique, feedback utilisateur, scoring numérique
• Self-hosted : docker compose up — gratuit, données chez vous

28K+ stars GitHub · MIT (juin 2026) Largement adopté dans l'écosystème LLMOps OpenTelemetry natif

Objet	Rôle	Champs clés	Exemple
Trace	Container racine d'une exécution	user_id, session_id, input, output, metadata, tags	Une requête utilisateur de bout en bout
Span	Unité de travail générique	name, start_time, end_time, input, output	Retrieval, reranking, parsing
Generation	Appel LLM avec métadonnées complètes	model, usage_details, cost_details, prompt_version	Appel GPT-4o, Claude, Mistral
Event	Action ponctuelle sans durée	name, input, metadata	Tool call, erreur capturée
Score	Évaluation qualité	name, value, data_type, comment	hallucination: 0.1, correctness: "correct"

# Installation du SDK Python
pip install langfuse

# Variables d'environnement (dans ~/.bashrc, ~/.zshrc ou .env)
export LANGFUSE_PUBLIC_KEY="pk-lf-..."      # Clé publique du projet
export LANGFUSE_SECRET_KEY="sk-lf-..."      # Clé secrète du projet
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"  # EU (défaut)

# Alternatives selon la région
# export LANGFUSE_BASE_URL="https://us.cloud.langfuse.com"   # US
# export LANGFUSE_BASE_URL="http://localhost:3000"           # Self-hosted

from langfuse import get_client

langfuse = get_client()

# Teste la connexion et affiche les infos du projet
langfuse.auth_check()
# Output: True si connexion OK

Pattern singleton : get_client() retourne toujours la même instance dans un processus — pas besoin de passer le client en paramètre partout. Le SDK lit les variables d'env automatiquement.

from langfuse.decorators import observe, langfuse_context
from langfuse import get_client
from openai import OpenAI

langfuse = get_client()
openai_client = OpenAI()

# Chaque fonction décorée devient un span dans Langfuse
@observe()
def retrieve_context(query: str) -> str:
    """Simule une étape de retrieval documentaire."""
    # Votre logique de retrieval (BM25, ChromaDB, etc.)
    return f"Contexte pertinent pour : {query}"

@observe()
def generate_answer(query: str, context: str) -> str:
    """Appel LLM avec le contexte récupéré."""
    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Tu es un assistant expert."},
            {"role": "user",   "content": f"{query}\n\nContexte:\n{context}"}
        ],
    )
    return response.choices[0].message.content

# Fonction principale — crée la trace racine
@observe()
def rag_pipeline(query: str) -> str:
    # Associer un utilisateur à cette trace
    langfuse_context.update_current_trace(
        user_id="user_123",
        session_id="session_456",
        tags=["production", "rag"]
    )
    context = retrieve_context(query)
    answer  = generate_answer(query, context)
    return answer

# Exécution — Langfuse trace automatiquement les 3 spans
result = rag_pipeline("Qu'est-ce que le fine-tuning LoRA ?")
print(result)

# IMPORTANT : flush avant la fin du processus (Lambda, scripts batch)
langfuse.flush()

N'oubliez pas langfuse.flush() dans les scripts Python classiques, les fonctions Lambda, et les workers batch. Sans flush, les données en buffer peuvent ne pas être envoyées si le processus se termine trop vite. Dans une API FastAPI/Django, ce n'est pas nécessaire car le serveur tourne en continu.

from langfuse import get_client
import time

langfuse = get_client()

# 1. Créer la trace racine
trace = langfuse.trace(
    name="support-ticket-resolution",
    user_id="user_789",
    session_id="ticket_001",
    metadata={"ticket_type": "billing", "priority": "high"},
    tags=["support", "billing"]
)

# 2. Span de retrieval
retrieval_span = trace.span(
    name="retrieve-knowledge-base",
    input={"query": "remboursement facturation"}
)
# ... votre logique de retrieval ...
retrieved_docs = ["doc1", "doc2"]
retrieval_span.end(output={"docs_count": len(retrieved_docs)})

# 3. Generation LLM avec métadonnées complètes
generation = trace.generation(
    name="draft-response",
    model="gpt-4o",
    input=[
        {"role": "system", "content": "Tu es un agent support."},
        {"role": "user",   "content": "Je veux un remboursement."}
    ],
    model_parameters={"temperature": 0.3, "max_tokens": 500}
)

# ... appel LLM réel ici ...
llm_response = "Voici les étapes pour votre remboursement..."
generation.end(
    output=llm_response,
    usage={"prompt_tokens": 120, "completion_tokens": 85}
)

# 4. Scorer la trace (évaluation de qualité)
trace.score(
    name="response_quality",
    value=0.9,
    data_type="numeric",
    comment="Réponse complète et précise"
)

langfuse.flush()

# AVANT (sans Langfuse)
from openai import OpenAI

# APRÈS (avec Langfuse — un seul changement)
from langfuse.openai import OpenAI

# Le reste du code est identique
client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Explique le RAG en 3 points."}],
    # Métadonnées Langfuse en paramètres supplémentaires
    name="rag-explanation",          # Nom de la génération
    user_id="user_123",              # ID utilisateur
    session_id="session_456",        # ID session
    tags=["demo", "production"],      # Tags pour filtrer dans le dashboard
)

print(response.choices[0].message.content)
# Langfuse trace automatiquement : modèle, tokens, coût, latence

Compatibilité : ce wrapper fonctionne avec chat.completions.create(), embeddings.create(), et les appels en streaming. Il supporte aussi Azure OpenAI via AzureOpenAI du même module.

from langfuse.langchain import CallbackHandler
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Initialisation du handler avec contexte
langfuse_handler = CallbackHandler(
    session_id="session_001",
    user_id="user_123",
    tags=["langchain", "production"],
    metadata={"environment": "prod"}
)

# Votre chaîne LangChain habituelle
prompt = ChatPromptTemplate.from_messages([
    ("system", "Tu es un expert en IA générative."),
    ("user",   "{question}")
])
llm   = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm

# Passer le handler dans config — un seul endroit
response = chain.invoke(
    {"question": "Explique le fine-tuning LoRA."},
    config={"callbacks": [langfuse_handler]}
)

# Récupérer l'ID de trace pour du scoring post-hoc
trace_id = langfuse_handler.get_trace_id()
print(f"Trace ID : {trace_id}")

from langfuse.langchain import CallbackHandler
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# Handler Langfuse — même pattern que LangChain
langfuse_handler = CallbackHandler(
    user_id="user_456",
    tags=["agent", "langgraph"]
)

# Votre graph LangGraph habituel
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

# ... définition du graph ...
graph = StateGraph(AgentState)
# ... ajout des noeuds et edges ...
agent = graph.compile()

# Exécution avec callback — Langfuse trace chaque noeud du graph
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Analyse ce document."}]},
    config={"callbacks": [langfuse_handler]}
)

# Dans le dashboard, vous verrez chaque noeud du graph comme un span distinct
# avec ses inputs/outputs, latences et tokens

Agents à plusieurs niveaux : sur un agent LangGraph de 8 noeuds, Langfuse crée automatiquement une trace avec 8 spans enfants. Vous voyez exactement où le temps est passé, quels noeuds appellent des LLMs (et combien de tokens), et à quelle étape une erreur se produit.

from langfuse import get_client
from openai import OpenAI

langfuse      = get_client()
openai_client = OpenAI()

# Récupérer le prompt depuis Langfuse (label "production" par défaut)
prompt_obj = langfuse.get_prompt("rag-system-prompt")

# Compiler avec les variables template (doubles accolades dans l'interface)
compiled_prompt = prompt_obj.compile(
    expertise_level="expert",
    language="français"
)

# Utiliser le prompt — Langfuse trace automatiquement la version utilisée
response = openai_client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": compiled_prompt},
        {"role": "user",   "content": "Qu'est-ce que le RAG ?"}
    ],
    langfuse_prompt=prompt_obj,  # Lie la trace au prompt (version, label)
)

# Version spécifique pour tests A/B
prompt_v2 = langfuse.get_prompt("rag-system-prompt", version=2)
prompt_staging = langfuse.get_prompt("rag-system-prompt", label="staging")

Avantages du Prompt Management

from langfuse import get_client

langfuse = get_client()

# Score numérique (hallucination : 0 = aucune, 1 = totale)
langfuse.score(
    trace_id="trace_abc123",
    name="hallucination",
    value=0.05,
    data_type="numeric",
    comment="Réponse factuelle vérifiée"
)

# Score catégoriel (correctness)
langfuse.score(
    trace_id="trace_abc123",
    name="correctness",
    value="correct",
    data_type="categorical"
)

# Score sur une generation spécifique (pas la trace entière)
langfuse.score(
    trace_id="trace_abc123",
    observation_id="gen_xyz789",
    name="relevance",
    value=0.92,
    data_type="numeric"
)

from langfuse.decorators import observe, langfuse_context
from langfuse import get_client
from openai import OpenAI
import json

langfuse      = get_client()
openai_client = OpenAI()

JUDGE_PROMPT = """Tu es un évaluateur objectif de réponses IA.

Question posée : {question}
Réponse générée : {answer}
Contexte fourni : {context}

Évalue sur trois critères (score de 0 à 1) :
- faithfulness : la réponse est-elle fidèle au contexte ?
- relevance    : la réponse répond-elle à la question ?
- conciseness  : la réponse est-elle concise et claire ?

Réponds uniquement en JSON : {{"faithfulness": 0.9, "relevance": 0.8, "conciseness": 0.7}}"""

@observe()
def evaluate_with_llm_judge(question: str, answer: str, context: str, trace_id: str):
    """Évalue une réponse RAG avec un LLM juge."""
    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",  # Modèle juge (plus rapide = moins cher)
        messages=[{
            "role": "user",
            "content": JUDGE_PROMPT.format(
                question=question, answer=answer, context=context
            )
        }],
        response_format={"type": "json_object"}
    )

    scores = json.loads(response.choices[0].message.content)

    # Envoyer les scores à la trace principale (pas celle du juge)
    for metric_name, score_value in scores.items():
        langfuse.score(
            trace_id=trace_id,
            name=metric_name,
            value=score_value,
            data_type="numeric",
            comment="LLM-as-judge : gpt-4o-mini"
        )

    return scores

# Usage dans un pipeline RAG
@observe()
def rag_with_evaluation(question: str) -> str:
    context = "Langfuse est une plateforme open-source LLMOps..."
    answer  = "Langfuse permet de monitorer vos LLMs."  # Réponse de votre LLM

    trace_id = langfuse_context.get_current_trace_id()

    # Évaluation asynchrone après génération
    scores = evaluate_with_llm_judge(question, answer, context, trace_id)
    print(f"Scores : {scores}")

    return answer

Méthode	Tokens auto	Coût auto	Configuration
Wrapper langfuse.openai	Automatique	Automatique	Aucune
CallbackHandler LangChain	Automatique	Automatique	Aucune
Traces manuelles	Manuel	Auto si tokens fournis	Passer usage_details
Ollama / LLM local	Via LiteLLM	0$ (local)	LiteLLM Proxy configuré

# Accéder aux métriques de coût dans le dashboard Langfuse :
# Dashboard → Cost → Filtrer par modèle, utilisateur, date, tag

# Dans les traces manuelles, passer les tokens explicitement
generation = trace.generation(
    name="llm-call",
    model="gpt-4o",
    usage={
        "input":  1250,   # tokens input
        "output": 420,    # tokens output
        "unit":   "TOKENS"
    }
    # Langfuse calcule le coût USD automatiquement via son catalogue de prix
)

from langfuse.decorators import observe, langfuse_context
from langfuse.openai import OpenAI
from langfuse import get_client
import chromadb
import json

langfuse = get_client()
client   = OpenAI()

# ChromaDB comme vector store
chroma = chromadb.Client()
collection = chroma.get_or_create_collection("knowledge-base")

@observe()
def embed_query(query: str) -> list[float]:
    """Génère l'embedding de la requête."""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=query
    )
    return response.data[0].embedding

@observe()
def retrieve_documents(query: str, n_results: int = 5) -> list[str]:
    """Retrieval vectoriel depuis ChromaDB."""
    embedding = embed_query(query)
    results   = collection.query(query_embeddings=[embedding], n_results=n_results)

    # Logger les métadonnées du retrieval dans le span
    langfuse_context.update_current_observation(
        metadata={
            "n_results":  n_results,
            "distances":  results["distances"][0][:3] if results["distances"] else []
        }
    )
    return results["documents"][0] if results["documents"] else []

@observe()
def generate_response(query: str, context_docs: list[str]) -> str:
    """Génère la réponse finale avec le contexte récupéré."""
    # Récupérer le prompt depuis Langfuse Prompt Management
    prompt_obj     = langfuse.get_prompt("rag-system")
    system_prompt  = prompt_obj.compile(
        context="\n\n".join(context_docs),
        language="français"
    )

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user",   "content": query}
        ],
        langfuse_prompt=prompt_obj,   # Lie la trace à la version du prompt
        temperature=0.2,
        max_tokens=800
    )
    return response.choices[0].message.content

@observe()
def rag_query(query: str, user_id: str, session_id: str) -> dict:
    """Pipeline RAG complet avec monitoring Langfuse."""
    # Associer utilisateur et session à la trace racine
    langfuse_context.update_current_trace(
        user_id=user_id,
        session_id=session_id,
        input={"query": query},
        tags=["rag", "production"]
    )

    # Pipeline RAG
    docs   = retrieve_documents(query, n_results=5)
    answer = generate_response(query, docs)

    # Évaluation LLM-as-judge
    trace_id = langfuse_context.get_current_trace_id()
    evaluate_faithfulness(query, answer, docs, trace_id)

    # Mettre à jour l'output de la trace racine
    langfuse_context.update_current_trace(output={"answer": answer})

    return {"answer": answer, "trace_id": trace_id, "sources": len(docs)}

@observe(as_type="generation")
def evaluate_faithfulness(query, answer, docs, trace_id):
    """LLM-as-judge pour évaluer la fidélité au contexte."""
    context_str = "\n".join(docs[:3])
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content":
            f"Contexte: {context_str}\nRéponse: {answer}\n"
            f"La réponse est-elle fidèle au contexte ? Score 0-1 en JSON: {{'score': 0.9}}"
        }],
        response_format={"type": "json_object"}
    )
    score = json.loads(response.choices[0].message.content)["score"]
    langfuse.score(trace_id=trace_id, name="faithfulness", value=score)

# 1. Cloner le dépôt officiel
git clone https://github.com/langfuse/langfuse.git
cd langfuse

# 2. Éditer docker-compose.yml — remplacer TOUS les secrets marqués # CHANGEME
#    Utiliser des mots de passe longs et aléatoires (openssl rand -base64 32)

# 3. Lancer la stack complète
docker compose up -d

# 4. Accéder à l'interface web
open http://localhost:3000

# 5. Configurer votre SDK Python pour pointer vers le self-hosted
export LANGFUSE_BASE_URL="http://localhost:3000"  # ou votre domaine

Configuration minimale requise

Sécurité : ne jamais déployer avec les mots de passe par défaut du docker-compose. Générez des secrets forts avec openssl rand -base64 32 pour chaque service (PostgreSQL, ClickHouse, Redis, MinIO, NextAuth). Placez Langfuse derrière un reverse proxy HTTPS (Nginx, Traefik) avant de l'exposer en production.

Critère	Langfuse	LangSmith	Arize Phoenix	Helicone
Licence	MIT Open Source	Propriétaire	ELv2	Propriétaire
Self-hosted	Oui, 1ère classe	Enterprise requis	Local seulement	Via proxy
Framework-agnostic	Oui (OTel)	LangChain-first	Oui	Via proxy
Prompt Management	Complet	Complet	Non	Non
Évaluations	LLM-judge, code, human	LLM-judge, rules	Riche (templates)	Basique
Alerting production	Non	Oui	Partiel	Via gateway
Coût cloud relatif (voir tarifs officiels)	Économique	Significativement plus élevé	Gratuit (local)	Intermédiaire
Idéal pour	La plupart des équipes	Équipes 100% LangChain	Expérimentation offline	Proxy simple et rapide

Verdict 2026 : Langfuse est généralement le choix le plus économique et le plus flexible pour la plupart des équipes — open-source, auto-hébergeable, framework-agnostic. Choisissez LangSmith si vous êtes 100% dans l'écosystème LangChain et avez besoin d'alertes de production. Choisissez Arize Phoenix si vous faites beaucoup d'expérimentation offline. Les tarifs évoluant régulièrement, consultez directement langfuse.com/pricing et langsmith avant de décider.

Qu'est-ce que Langfuse ?

Langfuse est une plateforme open-source (MIT) d'observabilité pour applications LLM. Elle permet de tracer chaque appel LLM, monitorer les coûts et latences, versionner les prompts, évaluer automatiquement les réponses et déboguer les pipelines d'agents en production. largement adopté dans l'écosystème LLMOps mondial.

Comment installer Langfuse en Python ?

pip install langfuse, puis configurez trois variables d'environnement : LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY (obtenus dans Settings → API Keys sur cloud.langfuse.com) et LANGFUSE_BASE_URL (URL de votre instance cloud ou self-hosted). Le SDK les détecte automatiquement.

Langfuse est-il gratuit ?

Oui, Langfuse est open-source (MIT) et se déploie gratuitement en self-hosted. Le plan cloud Hobby est gratuit avec 50 000 unités/mois, 30 jours de rétention et 2 utilisateurs — suffisant pour développer et tester. Le plan Core à 29$/mois monte à 100 000 unités/mois avec utilisateurs illimités.

Quelle est la différence entre Langfuse et LangSmith ?

Langfuse est 100% open-source (MIT), framework-agnostic via OpenTelemetry, self-hostable gratuitement, et significativement moins cher en cloud. LangSmith est propriétaire, optimisé pour LangChain, et propose de meilleures alertes de production. Pour la plupart des équipes en 2026, Langfuse est le meilleur choix sauf si vous êtes 100% LangChain et avez besoin d'alertes avancées.

Langfuse fonctionne-t-il avec Ollama et les LLMs locaux ?

Oui. Via LiteLLM Proxy (qui expose Ollama comme une API OpenAI-compatible), via le décorateur @observe pour une instrumentation manuelle, ou via OpenTelemetry. Le coût est 0$ (LLM local), mais Langfuse trace latences, inputs/outputs et évaluations. Idéal pour monitorer des pipelines RAG locaux sans coût API.

Pour démarrer : créez un compte sur cloud.langfuse.com (gratuit), installez pip install langfuse, ajoutez @observe() sur votre fonction principale et exécutez-la. Votre première trace apparaîtra dans le dashboard en moins de 30 secondes.

Envie d'aller plus loin ? Consultez notre guide sur LangGraph pour construire des agents IA stateful et notre tutoriel MCP en Python pour connecter vos agents à des outils externes — puis instrumentez tout avec Langfuse.

Articles liés