AI Engineering Academy · Leçon

Langfuse pour une observabilité indépendante des modèles

Langfuse pour une observabilité indépendante des modèles est une leçon AI Engineering Academy gratuite sur CoddyKit. Ceci est la leçon 3 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Engineering Academy, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Engineering Academy comprend 4 leçons au total.

Langfuse : observabilité open source des LLM

Langfuse est une plateforme d’observabilité open source pour les applications LLM qui fonctionne avec n’importe quel fournisseur de modèles : OpenAI, Anthropic, Mistral, des modèles locaux via Ollama ou votre propre modèle affiné. Contrairement à LangSmith, qui vous lie à LangChain, Langfuse s’intègre à n’importe quel code Python grâce à un SDK simple. Vous pouvez héberger Langfuse vous-même gratuitement ou utiliser le service cloud géré sur cloud.langfuse.com.

# pip install langfuse
from langfuse import Langfuse

langfuse = Langfuse(
    public_key='pk-lf-...',
    secret_key='sk-lf-...',
    host='https://cloud.langfuse.com'  # or your self-hosted URL
)

print('Langfuse connected:', langfuse.auth_check())

Traces, étendues et générations

Langfuse utilise un modèle de données hiérarchique à trois niveaux. Une trace représente une requête utilisateur complète, de bout en bout. Au sein d’une trace, les étendues représentent les différentes étapes de traitement (récupération, prétraitement, appels d’outils). Les générations sont un type particulier d’étendue, spécialement conçu pour les appels aux LLM : elles enregistrent le modèle, les jetons du prompt, les jetons de complétion et le coût de manière structurée, ce qui permet de créer des tableaux de bord des coûts et des indicateurs de qualité.

from langfuse import Langfuse

langfuse = Langfuse()

# Create a trace for one user request
trace = langfuse.trace(
    name='rag-query',
    user_id='user_123',
    session_id='session_abc',
    tags=['production', 'rag']
)

# Add a retrieval span
retrieval_span = trace.span(
    name='vector-retrieval',
    input={'query': 'What is RAG?'}
)
chunks = vector_db.search('What is RAG?')
retrieval_span.end(output={'chunks': [c['text'][:100] for c in chunks]})

# Add an LLM generation
generation = trace.generation(
    name='answer-generation',
    model='gpt-4o',
    model_parameters={'temperature': 0.0},
    input=[{'role': 'user', 'content': 'Context: ...\nQuestion: What is RAG?'}]
)
response = openai_client.chat.completions.create(model='gpt-4o', messages=[...])
generation.end(
    output=response.choices[0].message.content,
    usage={'input': response.usage.prompt_tokens, 'output': response.usage.completion_tokens}
)

Le modèle d’intégration par décorateur

Langfuse fournit des décorateurs de fonctions qui enveloppent automatiquement vos fonctions avec des étendues de trace. Le décorateur @observe() capture les entrées et les sorties, la durée d’exécution et les éventuelles exceptions. C’est la manière la plus propre d’instrumenter du code existant sans le restructurer.

from langfuse.decorators import observe, langfuse_context

# @observe wraps the function as a span automatically
@observe()
def retrieve_chunks(query: str) -> list[dict]:
    return vector_db.search(query, top_k=5)

@observe()
def generate_answer(query: str, context: str) -> str:
    response = openai_client.chat.completions.create(
        model='gpt-4o',
        messages=[
            {'role': 'system', 'content': 'Answer using the context.'},
            {'role': 'user', 'content': f'Context: {context}\nQuestion: {query}'}
        ]
    )
    # Attach LLM usage data to the current span
    langfuse_context.update_current_observation(
        usage={'input': response.usage.prompt_tokens, 'output': response.usage.completion_tokens},
        model='gpt-4o'
    )
    return response.choices[0].message.content

@observe(name='rag-pipeline')  # top-level trace
def rag_pipeline(query: str) -> str:
    chunks = retrieve_chunks(query)  # becomes a nested span
    context = '\n'.join([c['text'] for c in chunks])
    return generate_answer(query, context)  # becomes another nested span

Intégration avec n’importe quel fournisseur de LLM

Contrairement à l’intégration étroite de LangSmith avec LangChain, Langfuse fonctionne avec n’importe quel fournisseur de LLM en utilisant la même approche fondée sur les décorateurs. Que vous appeliez l’API d’Anthropic, un modèle Ollama local, un point de terminaison d’inférence Hugging Face ou un modèle personnalisé que vous avez affiné, Langfuse trace l’appel de la même manière. Cette neutralité vis-à-vis des fournisseurs est essentielle lorsque votre application utilise plusieurs modèles.

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

anthropic_client = anthropic.Anthropic()
openai_client = OpenAI()

@observe()
def call_claude(prompt: str) -> str:
    response = anthropic_client.messages.create(
        model='claude-3-5-sonnet-20241022',
        max_tokens=1024,
        messages=[{'role': 'user', 'content': prompt}]
    )
    langfuse_context.update_current_observation(
        model='claude-3-5-sonnet-20241022',
        usage={'input': response.usage.input_tokens, 'output': response.usage.output_tokens}
    )
    return response.content[0].text

@observe()
def call_gpt4(prompt: str) -> str:
    response = openai_client.chat.completions.create(
        model='gpt-4o',
        messages=[{'role': 'user', 'content': prompt}]
    )
    langfuse_context.update_current_observation(model='gpt-4o',
        usage={'input': response.usage.prompt_tokens, 'output': response.usage.completion_tokens})
    return response.choices[0].message.content

Tableaux de bord de suivi des coûts

Langfuse calcule automatiquement le coût à partir du nom du modèle et du nombre de jetons, en utilisant une table tarifaire intégrée qui couvre OpenAI, Anthropic, Mistral et des dizaines d’autres fournisseurs. Le tableau de bord des coûts affiche : les dépenses totales par période, le coût ventilé par modèle, le coût ventilé par fonctionnalité ou par utilisateur (à l’aide d’étiquettes et de métadonnées), ainsi que l’évolution des dépenses quotidiennes et hebdomadaires. Cette visibilité évite les mauvaises surprises sur la facture et aide à repérer les requêtes exceptionnellement coûteuses.

# Cost data is automatically computed - no manual config
# Langfuse knows: gpt-4o input = $0.005/1K tokens, output = $0.015/1K tokens

# Add metadata to enable cost breakdown by feature
@observe(name='rag-query')
def handle_rag_query(query: str, feature: str, user_id: str) -> str:
    langfuse_context.update_current_trace(
        user_id=user_id,
        tags=[feature, 'rag'],
        metadata={'feature': feature, 'query_length': len(query)}
    )
    return rag_pipeline(query)

# In Langfuse dashboard you can now filter costs by:
# - feature: 'document_qa', 'chat', 'summarization'
# - user_id: to see which users are your most expensive
# - model: to compare gpt-4o vs gpt-4o-mini costs
# - date range: to see daily/weekly/monthly trends

Ajout de scores issus des retours utilisateurs

Langfuse vous permet d’associer des retours utilisateurs aux traces a posteriori. Lorsqu’un utilisateur clique sur un pouce vers le haut ou vers le bas pour évaluer une réponse, vous pouvez enregistrer cette action sous forme de score sur la trace correspondante. Vous reliez ainsi les signaux réels de satisfaction des utilisateurs à l’ensemble du contexte de la trace, ce qui vous permet d’analyser ce qui distingue les réponses très bien notées des réponses mal notées.

from langfuse.decorators import observe, langfuse_context

@observe()
def generate_response(query: str) -> dict:
    answer = rag_pipeline(query)
    # Get the current trace ID to link feedback later
    trace_id = langfuse_context.get_current_trace_id()
    return {'answer': answer, 'trace_id': trace_id}

# Later, when user submits feedback:
def record_user_feedback(trace_id: str, score: int, comment: str):
    langfuse.score(
        trace_id=trace_id,
        name='user_satisfaction',  # score name
        value=score,               # 1 (thumbs up) or 0 (thumbs down)
        comment=comment,
        data_type='BOOLEAN'
    )

# Now in Langfuse: filter traces where user_satisfaction = 0
# to find the exact prompts and contexts that users rated negatively

Scores automatisés avec un LLM comme évaluateur

Au-delà des retours utilisateurs, Langfuse prend en charge la notation automatisée à l’aide d’évaluateurs qui utilisent un LLM comme juge. Vous pouvez définir des évaluateurs qui s’exécutent de manière asynchrone sur des traces échantillonnées et les notent selon des critères tels que la pertinence, la fidélité, la toxicité ou le respect du format. Ces scores automatisés alimentent le même tableau de bord que les retours humains, ce qui permet une surveillance continue de la qualité sans annotation humaine à grande échelle.

from langfuse import Langfuse

langfuse = Langfuse()

def auto_score_traces():
    # Get recent unscored traces
    traces = langfuse.fetch_traces(tags=['production'], limit=50)
    
    for trace in traces.data:
        question = trace.input.get('query', '')
        answer = trace.output.get('answer', '') if trace.output else ''
        
        if not question or not answer:
            continue
        
        # LLM-as-judge scoring
        score = evaluate_relevance(question, answer)  # returns 0.0-1.0
        
        langfuse.score(
            trace_id=trace.id,
            name='auto_relevance',
            value=score,
            data_type='NUMERIC',
            comment='Automated relevance score from LLM judge'
        )

# Run this as a scheduled job every hour

Gestion des prompts dans Langfuse

Langfuse inclut une fonctionnalité de gestion des prompts qui stocke vos prompts dans le cloud Langfuse et vous permet de les récupérer à l’exécution. Les versions des prompts sont ainsi découplées des déploiements du code : vous pouvez mettre à jour un prompt dans l’interface Langfuse, et la modification prend effet immédiatement, sans déployer de nouveau le code. Langfuse enregistre également la version du prompt utilisée par chaque trace, afin que vous puissiez comparer les performances entre les versions.

from langfuse import Langfuse

langfuse = Langfuse()

# Fetch the current production prompt by name
# The prompt lives in Langfuse UI, not in your code
prompt = langfuse.get_prompt('rag-system-prompt', version='production')

# Use it in your pipeline
messages = [
    {'role': 'system', 'content': prompt.compile(context_limit=4000)},
    {'role': 'user', 'content': query}
]

response = openai_client.chat.completions.create(model='gpt-4o', messages=messages)

# The trace is automatically linked to the prompt version
# In Langfuse you can filter: show me traces using prompt v3 vs v4
# and compare their quality scores

Héberger Langfuse soi-même

Langfuse peut être hébergé sur votre propre infrastructure au moyen d’une seule commande Docker Compose, en utilisant PostgreSQL pour le stockage. L’auto-hébergement signifie que vos données de traces ne quittent jamais votre infrastructure, ce qui est essentiel pour les applications traitant des PII, des données médicales ou du contenu propriétaire. La version auto-hébergée offre les mêmes fonctionnalités que le cloud géré, mais vous devez gérer l’infrastructure (sauvegardes, mise à l’échelle et mises à niveau).

# Self-host Langfuse with Docker Compose
# docker-compose.yml (simplified)
# version: '3'
# services:
#   langfuse:
#     image: langfuse/langfuse:2
#     ports:
#       - '3000:3000'
#     environment:
#       - DATABASE_URL=postgresql://langfuse:password@postgres/langfuse
#       - NEXTAUTH_SECRET=your-random-secret
#       - SALT=your-random-salt
#   postgres:
#     image: postgres:15
#     environment:
#       - POSTGRES_DB=langfuse
#       - POSTGRES_PASSWORD=password

# After docker-compose up, point your SDK to:
langfuse = Langfuse(
    public_key='pk-lf-your-key',
    secret_key='sk-lf-your-key',
    host='http://localhost:3000'  # your self-hosted instance
)

Langfuse ou LangSmith : lequel choisir ?

Choisissez LangSmith si vous utilisez intensivement LangChain et souhaitez un traçage automatique sans configuration, une intégration approfondie avec les évaluations LangChain et que la dépendance à un fournisseur ne vous pose pas de problème. Choisissez Langfuse si vous utilisez plusieurs fournisseurs de LLM, devez effectuer un auto-hébergement pour respecter les exigences de confidentialité des données, souhaitez la transparence de l’open source ou développez avec des infrastructures autres que LangChain. Les deux solutions sont prêtes pour la production et proposent toutes deux des offres gratuites généreuses.

Intégration d’OpenTelemetry pour les LLM

Pour les équipes qui utilisent déjà OpenTelemetry pour le traçage distribué, Langfuse prend en charge l’ingestion via OTLP (protocole OpenTelemetry). Vous pouvez envoyer directement à Langfuse les données de traces de LLM provenant de vos exportateurs OTel existants, sans modifier votre instrumentation. Vous disposez ainsi d’une pile d’observabilité unifiée où les traces de LLM, les étendues des requêtes de base de données et les traces des requêtes HTTP résident dans le même système, avec des identifiants de corrélation cohérents.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# Configure OTel to send to Langfuse OTLP endpoint
exporter = OTLPSpanExporter(
    endpoint='https://cloud.langfuse.com/api/public/otel/v1/traces',
    headers={
        'Authorization': 'Basic ' + base64.b64encode(b'pk-lf-xxx:sk-lf-xxx').decode()
    }
)

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)

# Now create spans as usual - they appear in Langfuse automatically
tracer = trace.get_tracer('my-llm-app')
with tracer.start_as_current_span('rag-query') as span:
    span.set_attribute('llm.model', 'gpt-4o')
    span.set_attribute('llm.prompt_tokens', 500)
    result = rag_pipeline(query)

Vérification rapide

Vérifiez votre compréhension de l’observabilité indépendante du modèle avec Langfuse à partir de cette leçon.

Récapitulatif de la leçon

Dans cette leçon, vous avez appris que Langfuse fournit une observabilité open source des LLM, indépendante du modèle, grâce à un modèle de données hiérarchique composé de traces, d’étendues et de générations ; que le décorateur @observe() permet d’instrumenter du code existant avec un minimum de modifications ; et que le suivi des coûts, les scores issus des retours utilisateurs et la notation automatisée par un LLM comme juge font de Langfuse une plateforme complète de surveillance de la qualité. Nous allons maintenant mettre en place des alertes sur la latence, les coûts et la dégradation de la qualité.

Questions Fréquemment Posées

La leçon « Langfuse pour une observabilité indépendante des modèles » est-elle gratuite ?

Oui — le texte complet de « Langfuse pour une observabilité indépendante des modèles » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Engineering Academy, passe à CoddyKit PRO. Le cours AI Engineering Academy comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Langfuse pour une observabilité indépendante des modèles » ?

Intégrez Langfuse comme solution open source de remplacement compatible avec tout fournisseur de LLM, capturez des étendues personnalisées pour la recherche et les appels d’outils, puis configurez de… Tu pratiques AI Engineering Academy avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer AI Engineering Academy ?

Aucune expérience préalable n'est requise. AI Engineering Academy sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 3 sur 4.

Combien de temps prend la leçon « Langfuse pour une observabilité indépendante des modèles » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon AI Engineering Academy ?

Oui. Chaque leçon AI Engineering Academy inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

Pourquoi les applications LLM sont difficiles à déboguer
Tracer avec LangSmith
Langfuse pour une observabilité indépendante des modèles
Alerter en cas de latence, de coût ou de dégradation de la qualité

← Retour à AI Engineering Academy