LangSmith : Monitoring, debugging et évaluation de vos applications LangChain en production

# Votre code
result = rag_chain.invoke({"query": "Comment fonctionne le fine-tuning ?"})

# Que s'est-il passé ? 🤷
# - Quels documents ont été récupérés ?
# - Quel prompt exact a été envoyé au LLM ?
# - Combien de tokens ont été consommés ?
# - Pourquoi la réponse est-elle incorrecte ?
# - Combien de temps chaque étape a pris ?

import os
from dotenv import load_dotenv

load_dotenv()

# Configuration LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-..."  # Votre clé API
os.environ["LANGCHAIN_PROJECT"] = "mon-projet"  # Nom du projet

# C'est tout ! Vos chains sont maintenant tracées automatiquement

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template("Explique {concept} en une phrase")

chain = prompt | llm

# Cette exécution sera automatiquement tracée dans LangSmith
result = chain.invoke({"concept": "la photosynthèse"})
print(result.content)

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.chains import RetrievalQA

# Setup (une seule fois)
vectorstore = FAISS.from_documents(documents, OpenAIEmbeddings())
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-4"),
    retriever=vectorstore.as_retriever()
)

# Exécution (tracée automatiquement)
result = qa_chain.invoke({"query": "Qu'est-ce que LoRA ?"})

📊 Run: RetrievalQA (2.3s, $0.004)
  ├─ 🔍 Retriever (0.8s)
  │   ├─ Embedding query (0.2s, 50 tokens)
  │   └─ FAISS search (0.6s, 4 documents)
  ├─ 💬 ChatOpenAI (1.5s, $0.004)
  │   ├─ Input: 1200 tokens
  │   ├─ Output: 150 tokens
  │   └─ Total cost: $0.0041
  └─ ✅ Output: "LoRA est une technique..."

# Exécution 1
result1 = chain.invoke({"query": "Explique l'IA"})

# Exécution 2 (avec paramètres différents)
result2 = chain.invoke({"query": "Explique l'IA", "temperature": 0.9})

# Ajout de métadonnées pour filtrage
from langchain.callbacks import LangChainTracer

tracer = LangChainTracer(
    project_name="mon-projet",
    tags=["production", "user-123", "v2.0"]
)

result = chain.invoke(
    {"query": "..."},
    config={"callbacks": [tracer]}
)

try:
    result = problematic_chain.invoke({"query": "..."})
except Exception as e:
    # L'erreur est automatiquement capturée dans LangSmith
    print(f"Erreur: {e}")

from langsmith import Client

client = Client()

# Créer le dataset
dataset_name = "evaluation-rag"
dataset = client.create_dataset(dataset_name)

# Ajouter des exemples
examples = [
    {
        "inputs": {"query": "Qu'est-ce que LoRA ?"},
        "outputs": {"answer": "LoRA est une technique de fine-tuning..."}
    },
    {
        "inputs": {"query": "Différence entre LoRA et QLoRA ?"},
        "outputs": {"answer": "QLoRA est une version quantifiée..."}
    },
    {
        "inputs": {"query": "Avantages du fine-tuning ?"},
        "outputs": {"answer": "Le fine-tuning permet de personnaliser..."}
    }
]

for example in examples:
    client.create_example(
        inputs=example["inputs"],
        outputs=example["outputs"],
        dataset_id=dataset.id
    )

from langsmith.evaluation import evaluate

# Votre chain à évaluer
def my_chain(inputs: dict) -> dict:
    result = qa_chain.invoke(inputs["query"])
    return {"answer": result["result"]}

# Évaluateurs personnalisés
def correctness_evaluator(run, example) -> dict:
    """Évalue si la réponse est correcte"""
    predicted = run.outputs["answer"]
    expected = example.outputs["answer"]

    # Simple comparaison (à améliorer avec un LLM)
    is_correct = expected.lower() in predicted.lower()

    return {
        "key": "correctness",
        "score": 1.0 if is_correct else 0.0
    }

def length_evaluator(run, example) -> dict:
    """Évalue la longueur de la réponse"""
    answer_length = len(run.outputs["answer"].split())

    return {
        "key": "length",
        "score": 1.0 if 20 <= answer_length <= 200 else 0.5
    }

# Lancer l'évaluation
results = evaluate(
    my_chain,
    data=dataset_name,
    evaluators=[correctness_evaluator, length_evaluator],
    experiment_prefix="rag-v1"
)

# Afficher les résultats
print(f"Précision : {results['results']['correctness']['mean']}")
print(f"Qualité longueur : {results['results']['length']['mean']}")

from langsmith.evaluation import LangChainStringEvaluator

# Évaluateur de similarité sémantique
embedding_evaluator = LangChainStringEvaluator(
    "embedding_distance",
    config={"embeddings": OpenAIEmbeddings()}
)

# Évaluateur basé sur un LLM
llm_evaluator = LangChainStringEvaluator(
    "qa",  # Question-Answering correctness
    config={"llm": ChatOpenAI(model="gpt-4")}
)

# Utilisation
results = evaluate(
    my_chain,
    data=dataset_name,
    evaluators=[embedding_evaluator, llm_evaluator]
)

# Ajout de métadonnées pour le monitoring
from langchain.callbacks import LangChainTracer

def run_with_metadata(query: str, user_id: str, version: str):
    """Exécute avec métadonnées de production"""

    metadata = {
        "user_id": user_id,
        "version": version,
        "environment": "production"
    }

    result = chain.invoke(
        {"query": query},
        config={
            "metadata": metadata,
            "tags": ["production", f"v{version}"]
        }
    )

    return result

# Utilisation
result = run_with_metadata(
    query="Qu'est-ce que le fine-tuning ?",
    user_id="user-12345",
    version="2.1.0"
)

from langsmith import Client

client = Client()

# Après avoir montré la réponse à l'utilisateur
run_id = "..."  # ID de la trace (disponible dans les callbacks)
user_feedback = "positive"  # ou "negative"

client.create_feedback(
    run_id=run_id,
    key="user_rating",
    score=1.0 if user_feedback == "positive" else 0.0,
    comment="L'utilisateur a aimé la réponse"
)

# L'utilisateur a cliqué sur "Voir plus"
client.create_feedback(
    run_id=run_id,
    key="engagement",
    score=1.0,
    comment="L'utilisateur a interagi avec la réponse"
)

# L'utilisateur a reformulé sa question
client.create_feedback(
    run_id=run_id,
    key="reformulation",
    score=0.5,
    comment="Réponse possiblement incomplète"
)

# Version A : GPT-4o-mini
chain_a = create_chain(model="gpt-4o-mini", temperature=0.3)

# Version B : GPT-4
chain_b = create_chain(model="gpt-4", temperature=0.7)

# Évaluation comparative
results_a = evaluate(
    chain_a,
    data=dataset_name,
    experiment_prefix="variant-a-gpt4mini"
)

results_b = evaluate(
    chain_b,
    data=dataset_name,
    experiment_prefix="variant-b-gpt4"
)

# Comparer dans l'interface LangSmith
print(f"Variant A - Score: {results_a['results']['correctness']['mean']}")
print(f"Variant B - Score: {results_b['results']['correctness']['mean']}")

# Configuration une fois
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-..."

# Tout le reste est automatique !
chain = prompt | llm | output_parser
result = chain.invoke({"input": "..."})
# ✅ Tracé dans LangSmith

from langgraph.graph import StateGraph
from langsmith import traceable

# Les graphes sont automatiquement tracés
graph = StateGraph(State)
graph.add_node("research", research_node)
graph.add_node("writing", writing_node)
app = graph.compile()

# Exécution tracée
result = app.invoke({"query": "..."})
# ✅ Chaque nœud apparaît dans la trace

📊 Run: StateGraph (4.5s)
  ├─ 🔬 research_node (2.1s)
  │   ├─ Search tool (1.8s)
  │   └─ Processing (0.3s)
  ├─ ✍️ writing_node (2.0s)
  │   └─ GPT-4 call (1.9s)
  └─ 📋 review_node (0.4s)

from langchain.callbacks.base import BaseCallbackHandler

class CustomLangSmithCallback(BaseCallbackHandler):
    """Callback pour ajouter des métriques custom"""

    def on_chain_start(self, serialized, inputs, **kwargs):
        run_id = kwargs.get("run_id")
        # Logique personnalisée
        print(f"Chain started: {run_id}")

    def on_chain_end(self, outputs, **kwargs):
        run_id = kwargs.get("run_id")
        # Enregistrer métrique custom dans LangSmith
        from langsmith import Client
        client = Client()
        client.create_feedback(
            run_id=run_id,
            key="custom_metric",
            score=calculate_custom_score(outputs)
        )

# Utilisation
result = chain.invoke(
    {"input": "..."},
    config={"callbacks": [CustomLangSmithCallback()]}
)

# Dans l'interface LangSmith :
# 1. Sélectionnez une trace
# 2. Ajoutez un commentaire
# 3. Taguez un collègue
# 4. Créez une issue si nécessaire

# Version 1
prompt_v1 = ChatPromptTemplate.from_template(
    "Réponds en une phrase : {question}"
)

# Version 2
prompt_v2 = ChatPromptTemplate.from_template(
    "Réponds de manière concise et précise : {question}"
)

# LangSmith enregistre les deux versions
# Vous pouvez comparer leur performance

from langsmith import Client

client = Client()

# Exporter un dataset
dataset = client.read_dataset(dataset_name="my-dataset")
client.export_dataset(dataset.id, format="csv", filename="dataset.csv")

# Partager une trace
trace_url = f"https://smith.langchain.com/public/{run_id}"
# Envoyez ce lien à votre équipe

# Un projet par environnement
os.environ["LANGCHAIN_PROJECT"] = "my-app-dev"      # Développement
os.environ["LANGCHAIN_PROJECT"] = "my-app-staging"  # Staging
os.environ["LANGCHAIN_PROJECT"] = "my-app-prod"     # Production

# Stratégie de tagging
tags = [
    f"env:{environment}",           # dev, staging, prod
    f"version:{app_version}",       # v2.1.0
    f"model:{model_name}",          # gpt-4, claude-3
    f"feature:{feature_name}",      # rag, agent, chat
    f"user_tier:{user_tier}"        # free, pro, enterprise
]

result = chain.invoke(
    inputs,
    config={"tags": tags}
)

metadata = {
    "user_id": user_id,
    "session_id": session_id,
    "request_id": request_id,
    "ab_variant": "variant_a",
    "geographic_region": "eu-west",
    "client_version": "mobile-ios-2.1"
}

result = chain.invoke(
    inputs,
    config={"metadata": metadata}
)

datasets/
├── smoke-tests/          # Tests rapides (10 exemples)
├── regression-suite/     # Suite complète (100+ exemples)
├── edge-cases/           # Cas limites
├── production-samples/   # Échantillons réels
└── by-feature/
    ├── rag-tests/
    ├── agent-tests/
    └── chat-tests/

# .github/workflows/evaluate.yml
name: Evaluate on Dataset

on: [pull_request]

jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2

      - name: Setup Python
        uses: actions/setup-python@v2

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Run evaluation
        env:
          LANGCHAIN_API_KEY: ${{ secrets.LANGCHAIN_API_KEY }}
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          python scripts/evaluate.py --dataset regression-suite

      - name: Check threshold
        run: |
          # Échoue si score < 0.85
          python scripts/check_threshold.py --min-score 0.85

# Environnement de dev
os.environ["LANGCHAIN_PROJECT"] = "my-app-dev"

# Itération rapide
while not satisfied:
    # Modifiez le code
    result = chain.invoke(test_input)

    # Vérifiez dans LangSmith
    # Ajustez si nécessaire

# Changement de projet pour staging
os.environ["LANGCHAIN_PROJECT"] = "my-app-staging"

# Évaluation sur dataset
results = evaluate(
    my_chain,
    data="regression-suite",
    experiment_prefix=f"pre-deploy-{version}"
)

# Validation
assert results["results"]["correctness"]["mean"] >= 0.85

# Si OK → déploiement

# Production
os.environ["LANGCHAIN_PROJECT"] = "my-app-prod"

def production_handler(request):
    """Handler avec monitoring complet"""

    metadata = {
        "user_id": request.user_id,
        "endpoint": request.endpoint,
        "version": APP_VERSION
    }

    try:
        result = chain.invoke(
            request.input,
            config={
                "metadata": metadata,
                "tags": ["production"]
            }
        )

        # Feedback implicite basé sur le temps de lecture
        if request.reading_time > 30:  # secondes
            client.create_feedback(
                run_id=result.run_id,
                key="engagement",
                score=1.0
            )

        return result

    except Exception as e:
        # Erreur automatiquement capturée dans LangSmith
        logger.error(f"Chain failed: {e}")
        raise

# Chaque semaine
def weekly_improvement_cycle():
    """Cycle d'amélioration hebdomadaire"""

    # 1. Analyser les feedbacks négatifs
    negative_runs = client.list_runs(
        project_name="my-app-prod",
        filter='eq(feedback_key, "user_rating") and eq(feedback_score, 0.0)',
        start_time=one_week_ago
    )

    # 2. Créer dataset avec cas problématiques
    for run in negative_runs[:20]:
        client.create_example(
            inputs=run.inputs,
            outputs=run.outputs,
            dataset_name="issues-to-fix"
        )

    # 3. Tester solutions
    improved_chain = optimize_prompts()

    # 4. Évaluer
    results = evaluate(
        improved_chain,
        data="issues-to-fix"
    )

    # 5. Déployer si amélioration > 10%
    if results["improvement"] > 0.10:
        deploy_new_version(improved_chain)