LangChain deep dive 2026 — LCEL, LangGraph, LangSmith

TL;DR — En 2026, LangChain n'est plus le couteau suisse anarchique de 2023. La vraie valeur s'est concentrée sur trois piliers : LCEL (Runnable + | pipe) pour composer des chaînes typées, LangGraph pour les agents et workflows stateful (le seul truc qu'on ne peut pas refaire en 200 lignes), et LangSmith pour le tracing/eval. Le reste (loaders, retrievers, output parsers) reste utile pour un POC en 1 jour mais s'évapore en prod où on préfère le raw SDK Anthropic/OpenAI. Règle pragmatique freelance FR : POC en LangChain → réécriture raw SDK quand le client signe.

🧠 Mental model

                  ┌─────────────────────────────────────────────┐
                  │            LANGCHAIN 2026 STACK              │
                  └─────────────────────────────────────────────┘

   ┌──────────────────┐    ┌──────────────────┐    ┌────────────────┐
   │  langchain-core  │ ←  │  langchain-XYZ   │ ←  │  langchain     │
   │  (Runnable, LCEL)│    │  (anthropic,     │    │  (legacy chains)│
   │  STABLE          │    │   openai, etc.)  │    │  ⚠ deprecated  │
   └────────┬─────────┘    └──────────────────┘    └────────────────┘
            │
            │ compose
            ▼
   ┌─────────────────────────────────────────────────┐
   │              LangGraph (stateful)                │
   │   nodes + edges + state + checkpoints + HITL    │ ← vraie valeur ajoutée
   └─────────────────────────────────────────────────┘
            │
            │ traces
            ▼
   ┌─────────────────────────────────────────────────┐
   │              LangSmith (observability)          │
   │   tracing, datasets, evals, prompt hub, A/B     │ ← SaaS payant utile
   └─────────────────────────────────────────────────┘

Analogie : LangChain = React. LCEL = JSX (déclaratif, composable).
LangGraph = Redux/XState (state machines pour agents).
LangSmith = Sentry + DataDog spécialisé LLM.

LCEL pipe analogy :
   prompt | model | parser   ≈   stdin | grep | awk   (Unix pipes typés)

LCEL impose une API unique sur tout objet : .invoke(), .batch(), .stream(), .astream(), .with_retry(), .with_fallbacks(), .bind(). Tout Runnable est composable avec |. C'est ce qui sauve LangChain en 2026 — l'ancien LLMChain, ConversationChain, RetrievalQA sont marqués deprecated depuis v0.3.

🛠️ Code minimal

# pip install langchain-core langchain-anthropic langchain-community

from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# ⚠️ Opus 4.8 / 4.7 rejettent temperature/top_p/top_k (HTTP 400). On NE passe PLUS
# temperature=0 sur ces modèles. Pour piloter le déterminisme/effort → output_config
# (cf. section "Thinking & effort" plus bas). Sur Sonnet 4.6, temperature reste accepté.
model = ChatAnthropic(model="claude-opus-4-8")
prompt = ChatPromptTemplate.from_messages([
    ("system", "Tu es un avocat français spécialisé droit du travail."),
    ("human", "{question}")
])
chain = prompt | model | StrOutputParser()

# invoke / stream / batch — même API
result = chain.invoke({"question": "Délai de prescription prud'homal ?"})
for chunk in chain.stream({"question": "Idem ?"}):
    print(chunk, end="", flush=True)
results = chain.batch([{"question": q} for q in ["Q1", "Q2", "Q3"]],
                     config={"max_concurrency": 5})

// npm i @langchain/core @langchain/anthropic
import { ChatAnthropic } from "@langchain/anthropic";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { StringOutputParser } from "@langchain/core/output_parsers";

// ⚠️ Opus 4.8/4.7 : pas de temperature/top_p/top_k (400). On retire temperature.
const model = new ChatAnthropic({ model: "claude-opus-4-8" });
const prompt = ChatPromptTemplate.fromMessages([
  ["system", "Tu es un avocat français..."],
  ["human", "{question}"],
]);
const chain = prompt.pipe(model).pipe(new StringOutputParser());
const result = await chain.invoke({ question: "..." });

Parité Python/JS est excellente en 2026 — l'écart de 2024 a été comblé. Choisir le langage selon l'écosystème client (Next.js → JS, Django/FastAPI → Python).

🧵 Thinking, effort & structured outputs à travers ChatAnthropic

C'est ici que 80 % des bugs Claude+LangChain en 2026 se nichent. Le wrapper LangChain est une couche par-dessus le SDK Anthropic : il ne te protège pas des changements de l'API sous-jacente. Les trois pièges suivants envoient des HTTP 400 en prod si tu copies-colles du code de 2024-2025.

1. budget_tokens est MORT sur Opus 4.8 / 4.7. L'ancien extended thinking (thinking={"type": "enabled", "budget_tokens": N}) renvoie un 400 sur les modèles 4.7/4.8. Le remplaçant est l'adaptive thinking + le paramètre effort (low | medium | high | xhigh | max). À travers ChatAnthropic tu passes ça par thinking et output_config (champs natifs API que le wrapper relaie) :

from langchain_anthropic import ChatAnthropic

# ✅ 2026 — adaptive thinking + effort (Opus 4.8/4.7)
model = ChatAnthropic(
    model="claude-opus-4-8",
    max_tokens=8000,
    thinking={"type": "adaptive", "display": "summarized"},  # display: défaut "omitted" → texte vide
    model_kwargs={"output_config": {"effort": "high"}},      # low|medium|high|xhigh|max
)

# ❌ 2024 — 400 sur Opus 4.8 : budget_tokens supprimé, temperature supprimée
# model = ChatAnthropic(model="claude-opus-4-7",
#                       thinking={"type": "enabled", "budget_tokens": 8000},
#                       temperature=0)

Pourquoi display="summarized" ? Par défaut Opus 4.8 renvoie des blocs thinking au texte vide ("omitted"). Si ton UI streame le raisonnement, l'utilisateur voit une longue pause muette avant la réponse — exactement le genre de régression silencieuse qui passe les tests unitaires et casse l'UX en démo client.

2. temperature / top_p / top_k → 400 sur Opus 4.8 / 4.7. Supprime-les des constructeurs ChatAnthropic. Sur Sonnet 4.6 et Haiku 4.5 ils restent acceptés. Une codebase multi-modèle doit donc poser temperature conditionnellement selon le modèle (cf. exercice 4) — sinon un fallback Opus→Sonnet qui marche masque un Sonnet→Opus qui 400.

3. Streaming cassé par les parsers structurés. StrOutputParser streame token-par-token sans problème. Mais PydanticOutputParser / JsonOutputParser attendent souvent la fin du message → plus de stream réel. Pire, l'ancienne façon de "forcer du JSON" — un prefill assistant ({"role": "assistant", "content": "{"}) — renvoie un 400 sur Opus 4.8/4.7/4.6 et Sonnet 4.6. La bonne réponse 2026 :

from pydantic import BaseModel

class Reponse(BaseModel):
    montant: float
    article: str

# ✅ structured output natif (mappe sur output_config.format côté API)
structured = model.with_structured_output(Reponse, method="json_schema", strict=True)
res = structured.invoke("Indemnité pour 8 ans d'ancienneté ?")  # → Reponse(montant=..., article=...)

method="json_schema" (et non function_calling) délègue au mode structured outputs natif d'Anthropic, plus fiable que le tool-calling détourné. Mental model du staff engineer : « le wrapper te donne with_structured_output, mais c'est l'API Anthropic qui contraint la sortie — donc lis les release notes Anthropic, pas seulement le changelog LangChain. »

🎬 Cas d'usage concrets

Cas 1 — POC RAG juridique en 1 jour (cabinet d'avocats Paris 8e)

Contexte — Cabinet de 15 avocats spécialisés droit social, 12K notes internes en PDF + Word sur 8 ans. Veulent prototyper "ChatGPT interne sur nos archives" avant de débloquer budget. Mission : 5 jours, livrable démo.

Approche LangChain — Jour 1 ingestion (UnstructuredFileLoader + RecursiveCharacterTextSplitter + Chroma), jour 2 chain RAG (LCEL), jour 3 chat UI (Streamlit), jour 4 polish + dataset eval LangSmith, jour 5 démo. Sans framework, j'aurais codé les loaders moi-même pour PDF/DOCX/EML = 1,5j perdu.

Ce qui a marché — MultiQueryRetriever qui reformule la question en 3 variantes avant retrieval. Sur "indemnité licenciement" il génère "rupture conventionnelle indemnités", "calcul indemnité légale", "barème Macron" → recall +25% sur l'eval set de 50 questions.

Ce qui a coincé — Chunking par défaut massacre les articles de loi (coupe au milieu d'un alinéa). J'ai dû passer en chunking custom avec regex sur Art. \d+ — du coup l'avantage framework s'effrite.

ROI — TJM 1200 € × 5 j = 6000 € HT. Le cabinet a signé un contrat de build prod à 35 K€. POC LangChain = lead magnet.

Cas 2 — Abstraction multi-provider pour SaaS B2B (legaltech Lyon, 8 dev)

Contexte — Startup legaltech, produit SaaS de génération de contrats. Clients enterprise demandent : "On veut héberger sur Azure / on veut Mistral pour souveraineté FR / on veut Anthropic". Doivent supporter 4 providers sans dupliquer la codebase.

Approche LangChain — BaseChatModel interface comme point d'entrée unique. Factory pattern : get_model(provider, model_name) → BaseChatModel. Le reste de l'app (prompts, chains, retrieval) ne connaît que BaseChatModel. Bascule provider = 1 ligne de config.

Pitfall réel — Les capabilities divergent : tool calling Anthropic ≠ OpenAI ≠ Mistral. Le bind_tools() unifié de LangChain marche pour 80 % des cas, mais structured output strict (JSON mode) varie. Solution : couche métier qui détecte le provider et adapte. LangChain ne fait pas tout, mais évite 70 % du boilerplate.

ROI client — Sans LangChain : 3 semaines pour ajouter un provider. Avec LangChain : 2 jours. Sur 4 providers, économie ~10 semaines-dev = 80 K€.

Cas 3 — Réécriture en raw SDK après POC validé (assurance Niort, 2 M docs)

Contexte — Mutuelle française, POC RAG sur sinistres validé en LangChain. Passage en prod : 2M docs, 500 req/min en pic, SLA 99.9 %. Architecture cible : FastAPI + Anthropic SDK direct + Qdrant.

Pourquoi réécrire — En prod on a besoin de : streaming SSE custom (LangChain l'expose mal sur FastAPI), prompt caching Anthropic explicite (les wrappers LangChain ne posent pas toujours cache_control au bon endroit), retries custom avec backoff non-standard, métriques Prometheus précises par étape. Le coût d'abstraction devient un fardeau.

Stratégie hybride retenue — On garde LangGraph pour orchestrer les étapes (extraction → retrieval → reranking → génération) parce que ça vaut le coup. Mais chaque nœud appelle le SDK Anthropic en raw. Best of both.

ROI — Latence p95 -40 % (de 4.2s à 2.5s). Coût tokens -22 % (cache control posé correctement, économie ~3 K€/mois sur le compte client).

🛠️ Exemple end-to-end

Comparaison directe : même RAG juridique en LangChain (POC) puis raw Anthropic SDK (prod).

Version A — LangChain LCEL (100 lignes, POC en 1 journée)

# pyproject : langchain-core langchain-anthropic langchain-community
#             langchain-chroma chromadb pypdf langsmith
import os, glob
from langchain_anthropic import ChatAnthropic
from langchain_chroma import Chroma
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_anthropic import AnthropicEmbeddings  # via Voyage

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "rag-juridique-poc"

# --- 1. Ingestion ----------------------------------------------------------
def ingest(pdf_dir: str, persist_dir: str = "./chroma_db"):
    docs = []
    for path in glob.glob(f"{pdf_dir}/**/*.pdf", recursive=True):
        docs.extend(PyPDFLoader(path).load())

    splitter = RecursiveCharacterTextSplitter(
        chunk_size=800,
        chunk_overlap=120,
        separators=["\n\nArt. ", "\n\n", "\n", ". "],  # respect articles
    )
    chunks = splitter.split_documents(docs)

    vectorstore = Chroma.from_documents(
        chunks,
        embedding=AnthropicEmbeddings(model="voyage-3-large"),
        persist_directory=persist_dir,
    )
    return vectorstore

# --- 2. RAG chain (LCEL) ---------------------------------------------------
def build_chain(vectorstore):
    retriever = vectorstore.as_retriever(
        search_type="mmr",
        search_kwargs={"k": 6, "fetch_k": 20, "lambda_mult": 0.5},
    )

    prompt = ChatPromptTemplate.from_messages([
        ("system",
         "Tu es un assistant juridique pour avocats français.\n"
         "Réponds UNIQUEMENT à partir du CONTEXTE fourni.\n"
         "Cite les sources (nom de fichier + page).\n"
         "Si l'information n'est pas dans le contexte, dis-le explicitement."),
        ("human",
         "CONTEXTE :\n{context}\n\n"
         "QUESTION : {question}\n\n"
         "Réponse structurée avec citations :"),
    ])

    model = ChatAnthropic(
        model="claude-opus-4-8",
        max_tokens=1500,   # pas de temperature sur Opus 4.8 (400)
    )

    def format_docs(docs):
        return "\n\n---\n\n".join(
            f"[{d.metadata.get('source','?')} p.{d.metadata.get('page','?')}]\n{d.page_content}"
            for d in docs
        )

    chain = (
        {"context": retriever | format_docs, "question": RunnablePassthrough()}
        | prompt
        | model
        | StrOutputParser()
    )
    return chain

# --- 3. Utilisation --------------------------------------------------------
if __name__ == "__main__":
    vs = ingest("./pdfs_droit_travail")
    chain = build_chain(vs)

    questions = [
        "Quel est le délai de prescription pour une action prud'homale ?",
        "Comment calculer l'indemnité légale de licenciement ?",
        "Une rupture conventionnelle peut-elle être contestée ?",
    ]
    for q in questions:
        print(f"\n## {q}\n")
        for chunk in chain.stream(q):
            print(chunk, end="", flush=True)

Bilan version A : 100 lignes, démarrable en 1 journée, tracing LangSmith gratuit, démo client convaincante.

Version B — Raw Anthropic SDK (200 lignes, prod)

# pyproject : anthropic qdrant-client voyageai fastapi prometheus-client
import os, asyncio, hashlib, time
from typing import AsyncIterator
import anthropic
import voyageai
from qdrant_client import AsyncQdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from prometheus_client import Counter, Histogram

# --- métriques --------------------------------------------------------------
RAG_LATENCY = Histogram("rag_latency_seconds", "RAG end-to-end", ["step"])
RAG_TOKENS = Counter("rag_tokens_total", "Tokens used", ["kind"])
RAG_CACHE_HITS = Counter("rag_cache_hits_total", "Anthropic cache hits")

claude = anthropic.AsyncAnthropic()
vo = voyageai.AsyncClient()
qdrant = AsyncQdrantClient(url=os.environ["QDRANT_URL"])

SYSTEM = (
    "Tu es un assistant juridique pour avocats français.\n"
    "Réponds UNIQUEMENT à partir du CONTEXTE fourni.\n"
    "Cite les sources (nom de fichier + page).\n"
    "Si l'information n'est pas dans le contexte, dis-le explicitement."
)

# --- 1. Embedding + retrieval ----------------------------------------------
async def embed(text: str) -> list[float]:
    with RAG_LATENCY.labels("embed").time():
        r = await vo.embed([text], model="voyage-3-large", input_type="query")
        return r.embeddings[0]

async def retrieve(query: str, k: int = 6) -> list[dict]:
    qvec = await embed(query)
    with RAG_LATENCY.labels("search").time():
        hits = await qdrant.search(
            collection_name="droit_travail",
            query_vector=qvec,
            limit=k,
            with_payload=True,
        )
    return [{"text": h.payload["text"], "src": h.payload["src"],
             "page": h.payload["page"], "score": h.score} for h in hits]

# --- 2. MMR re-ranking maison (Maximal Marginal Relevance) -----------------
async def mmr(query: str, candidates: list[dict], k: int = 6, lam: float = 0.5):
    qvec = await embed(query)
    cand_vecs = (await vo.embed(
        [c["text"] for c in candidates],
        model="voyage-3-large", input_type="document"
    )).embeddings
    selected, selected_vecs = [], []
    pool = list(zip(candidates, cand_vecs))
    while pool and len(selected) < k:
        best, best_score = None, -1
        for c, v in pool:
            sim_q = sum(a*b for a, b in zip(qvec, v))
            sim_s = max((sum(a*b for a, b in zip(sv, v))
                         for sv in selected_vecs), default=0)
            score = lam * sim_q - (1 - lam) * sim_s
            if score > best_score:
                best_score, best = score, (c, v)
        selected.append(best[0]); selected_vecs.append(best[1])
        pool.remove(best)
    return selected

# --- 3. Stream answer avec prompt caching ---------------------------------
def format_context(docs: list[dict]) -> str:
    return "\n\n---\n\n".join(
        f"[{d['src']} p.{d['page']}]\n{d['text']}" for d in docs
    )

async def answer(question: str) -> AsyncIterator[str]:
    t0 = time.time()
    candidates = await retrieve(question, k=20)
    top = await mmr(question, candidates, k=6)
    context = format_context(top)

    # Cache control SUR LE SYSTEM PROMPT (stable, ré-utilisé toutes les requêtes).
    # Rappel prefix-match : tools → system → messages. Le breakpoint sur le DERNIER
    # bloc system cache aussi les tools. Le contexte RAG change à chaque requête → il
    # vit APRÈS le breakpoint (dans le user message), donc il n'invalide jamais le cache.
    # ⚠️ Pas de temperature sur Opus 4.8 (400). Minimum cacheable Opus 4.8 = 4096 tokens :
    # un SYSTEM plus court que ça ne sera PAS caché (cache_creation_input_tokens=0, sans erreur).
    async with claude.messages.stream(
        model="claude-opus-4-8",
        max_tokens=1500,
        system=[
            {"type": "text", "text": SYSTEM, "cache_control": {"type": "ephemeral"}},
        ],
        messages=[{
            "role": "user",
            "content": f"CONTEXTE :\n{context}\n\nQUESTION : {question}",
        }],
    ) as stream:
        async for event in stream:
            if event.type == "content_block_delta":
                yield event.delta.text
        final = await stream.get_final_message()
        RAG_TOKENS.labels("input").inc(final.usage.input_tokens)
        RAG_TOKENS.labels("output").inc(final.usage.output_tokens)
        cached = getattr(final.usage, "cache_read_input_tokens", 0) or 0
        if cached > 0:
            RAG_CACHE_HITS.inc()
        RAG_LATENCY.labels("generate").observe(time.time() - t0)

# --- 4. FastAPI SSE endpoint ----------------------------------------------
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/ask")
async def ask(q: str):
    async def sse():
        async for chunk in answer(q):
            yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(sse(), media_type="text/event-stream")

Bilan version B : 200 lignes, contrôle total, latence p95 -40 %, économie tokens -22 %, métriques Prometheus précises, prompt caching maîtrisé. Maintenance plus lourde mais alignée sur les besoins prod.

Comparatif chiffré (eval set 200 questions, mutuelle Niort) :

Métrique	LangChain v0.3	Raw Anthropic SDK
Latence p50	1.8 s	1.1 s
Latence p95	4.2 s	2.5 s
Coût / 1K req	4.10 €	3.20 €
Lignes code	~100	~200
Time-to-POC	1 j	3 j
Time-to-prod-ready	4 j (fragile)	8 j (solide)

🎯 Patterns courants

1. Fallback chain pour robustesse

primary = ChatAnthropic(model="claude-opus-4-8")
fallback = ChatAnthropic(model="claude-sonnet-4-6")
chain = (prompt | primary.with_fallbacks([fallback]) | StrOutputParser())
# Note prod : with_fallbacks bascule sur TOUTE exception (y compris une 400 de schéma
# qui se reproduira sur les deux modèles → double facturation pour rien). Restreindre
# avec exceptions_to_handle=(anthropic.OverloadedError, anthropic.APITimeoutError) pour
# ne basculer que sur les pannes transitoires, pas sur les bugs déterministes.

2. with_retry avec backoff exponentiel intégré

robust = model.with_retry(
    stop_after_attempt=4,
    wait_exponential_jitter=True,
    retry_if_exception_type=(anthropic.APITimeoutError, anthropic.RateLimitError),
)

3. RunnableParallel pour fan-out

from langchain_core.runnables import RunnableParallel
multi = RunnableParallel(
    summary=summarize_chain,
    classification=classify_chain,
    sentiment=sentiment_chain,
)
result = multi.invoke({"text": doc})   # 3 appels en parallèle

4. chain.with_config(callbacks=[...]) pour observability custom

5. LangGraph state machine pour agents

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, list
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]
    docs: list
    classification: str

g = StateGraph(State)
g.add_node("retrieve", retrieve_node)
g.add_node("classify", classify_node)
g.add_node("answer", answer_node)
g.add_edge("retrieve", "classify")
g.add_conditional_edges("classify", route_by_intent, {
    "legal": "answer", "out_of_scope": END
})
app = g.compile(checkpointer=PostgresSaver(...))

6. with_structured_output(SchemaPydantic, method="json_schema", strict=True) — mappe sur le mode structured outputs natif d'Anthropic (cf. section Thinking/effort). Préférer json_schema à function_calling : plus fiable, et streamable contrairement aux output parsers.

7. Streaming token-by-token avec astream_events — granularité fine pour UI réactive.

🔄 Versions & écosystème 2026

• langchain-core 0.4.x — Runnable v2, breaking changes mineurs, stable depuis Q1 2026.
• langchain 0.3.x — chains legacy déprecated (LLMChain, RetrievalQA), supprimés en 0.4.
• langchain-anthropic 0.5.x — support Claude Opus 4.8 (flagship 2026, 5 $/25 $ par M tok input/output à 1M de contexte), prompt caching (cache_control), computer use, adaptive thinking (le budget_tokens est mort → cf. encadré ci-dessous).
• langchain-openai 0.4.x — support o3, GPT-5, structured outputs natifs.
• LangGraph 0.3.x — l'extraction officielle des agents hors LangChain, devient produit phare. Checkpointing Postgres/Redis natif, HITL (human-in-the-loop), time travel debugging.
• LangSmith — SaaS payant ($39/dev/mois plan team en 2026), self-hosted enterprise. Datasets, evaluators custom, A/B testing prompts, observability prod.
• LangChain JS 0.3.x — parité ≈ 95 % avec Python. Edge runtime (Cloudflare Workers, Vercel Edge) supporté.
• Churn d'API — toujours présent mais ralenti. La 0.3 a stabilisé les imports (from langchain_core.X au lieu de from langchain.X). Migration 0.2 → 0.3 prend ~1 j sur un projet moyen.

Écosystème concurrent : Haystack (production-focused, plus stable), LlamaIndex (data-layer), DSPy (declarative + optim), Vercel AI SDK (JS/TS only), Pydantic AI (typed, mince), Burr (state machine légère).

⚠️ Pitfalls

1. Imports legacy — from langchain.chat_models import ChatAnthropic est mort. Toujours from langchain_anthropic import ChatAnthropic. Sinon DeprecationWarning + bugs subtils.
2. Verbosity du tracing — par défaut LangSmith log TOUT, y compris des docs sensibles RGPD. Configurer LANGCHAIN_TRACING_V2=false ou LANGSMITH_HIDE_INPUTS=true selon contexte client.
3. RunnablePassthrough mal compris — sert à passer la donnée brute à travers la chaîne. Beaucoup de devs bricolent des lambdas à la place.
4. Streaming cassé par les parsers — StrOutputParser stream OK, mais PydanticOutputParser ou JsonOutputParser attendent souvent la fin → pas de stream réel. Utiliser with_structured_output(method="json_schema", strict=True) à la place.
5. Token usage caché — par défaut model.invoke() ne retourne pas l'usage. Activer model.with_config({"callbacks": [UsageMetadataCallbackHandler()]}) ou utiliser model.bind(stream_usage=True).
6. Prompt caching non-automatique — LangChain ne pose pas cache_control Anthropic tout seul. Soit on passe par model.bind(extra_body={...}), soit on raw SDK pour cette partie. Et un invalidateur silencieux classique : un datetime.now() ou un ID de session interpolé dans le system prompt change le préfixe à chaque requête → cache_read_input_tokens reste à 0 sans aucune erreur. Vérifier en lisant usage.cache_read_input_tokens, pas en croyant le code.
7. MultiQueryRetriever lent — génère N reformulations en série par défaut. Forcer parallélisme via .batch().
8. Tools cross-provider divergents — bind_tools() lisse 80 % mais parallel_tool_calls, tool_choice strict, JSON mode diffèrent. Tester par provider.
9. LangGraph state mutability — modifier state["messages"].append(...) en Python casse le checkpointing. Toujours retourner un nouveau dict.
10. Coût LangSmith en prod — facturé au trace + au token. Sur 500 req/min avec 30 spans/trace → facture rapide. Sampler (LANGCHAIN_SAMPLING_RATE=0.05) en prod.
11. budget_tokens / temperature → 400 sur Opus 4.8 & 4.7 — le wrapper relaie tel quel ce que tu mets dans le constructeur. Du vieux code ChatAnthropic(thinking={"type":"enabled","budget_tokens":N}) ou temperature=0 part en HTTP 400. Adaptive thinking + output_config.effort, et plus de paramètres de sampling sur la famille Opus 4.7+. (Cf. section Thinking/effort.)
12. Prefill assistant → 400 — l'astuce historique « forcer du JSON » en pré-remplissant un tour assistant ({"role":"assistant","content":"{"}) renvoie un 400 sur Opus 4.6/4.7/4.8 et Sonnet 4.6. Utiliser with_structured_output(..., method="json_schema", strict=True).
13. with_fallbacks trop large — par défaut il bascule sur toute exception. Une 400 de schéma se reproduira sur le modèle de secours → tu paies deux appels pour deux échecs. Restreindre aux pannes transitoires (OverloadedError, APITimeoutError).

💰 Pricing / ROI client

LangChain (open source) — gratuit. Pas de coût licence.

LangSmith (SaaS) :

• Developer : $0 jusqu'à 5K traces/mois.
• Plus : $39/dev/mois, 10K traces inclus, $0.50/1K traces additionnels.
• Enterprise : custom, self-hosted possible.

LangGraph Cloud (Platform) — managed deployment des agents, $50/mois/agent minimum, scaling auto. Alternative : self-host (LangGraph Server open source).

ROI typique freelance FR :

• POC RAG 5 jours : TJM 1200 € × 5 = 6 000 € HT. Si client signe la prod ensuite, build 25-45 K€.
• Mission "abstraction multi-provider" SaaS : 15 j × 1300 € = 19 500 € HT. Économise au client ~6 semaines de dev maison sur 12 mois (~50 K€).
• Refactor LangChain → raw SDK prod : 10-20 j × 1400 € = 14-28 K€ HT. ROI client : latence -30-40 %, coût tokens -20 %, soit ~3-8 K€/mois économisés sur volumes 500K-2M req/mois.

Argumentaire commercial — "Je livre un POC LangChain fonctionnel en 1 semaine. Si tu valides, on garde LangGraph pour l'orchestration et on raw-SDK le hot path. Best of both, sans dette."

🧪 Testing / Eval

# pytest + LangSmith datasets
from langsmith import Client, evaluate
from langsmith.evaluation import LangChainStringEvaluator

client = Client()

# 1. Créer un dataset à partir d'un CSV de questions / réponses attendues
dataset = client.create_dataset("rag-juridique-v1")
client.create_examples(
    inputs=[{"question": q} for q in questions],
    outputs=[{"answer": a} for a in gold_answers],
    dataset_id=dataset.id,
)

# 2. Evaluators : exactitude, citation présente, longueur raisonnable
def cites_source(run, example) -> dict:
    out = run.outputs["output"]
    has_cite = any(tag in out for tag in ["[", "Art.", "p."])
    return {"key": "cites_source", "score": int(has_cite)}

correctness = LangChainStringEvaluator(
    "labeled_score_string",
    config={"llm": ChatAnthropic(model="claude-opus-4-8")},  # juge = modèle fort, pas de temperature
)

# 3. Run eval
results = evaluate(
    lambda inp: chain.invoke(inp["question"]),
    data=dataset.name,
    evaluators=[cites_source, correctness],
    experiment_prefix="rag-poc-v1",
)
print(f"Mean correctness: {results.aggregate_scores['correctness']:.2f}")

Bonnes pratiques eval freelance FR :

• Dataset gold ≥ 100 paires Q/A validées par expert métier (avocat, RH, etc.).
• Sliding window — re-run eval à chaque PR sur main, breaker si correctness < 0.85.
• Mix automatique + humain — 80 % LLM-as-judge, 20 % spot-check humain manuel.

🔁 Quand utiliser / éviter

Utiliser LangChain quand :

• POC en < 5 jours pour valider un use case
• Besoin d'abstraction multi-provider stable (LangChain.js / Python)
• Workflow stateful agents → LangGraph (vraie valeur)
• Equipe junior LLM qui doit livrer vite
• Observability intégrée souhaitée (LangSmith)

Éviter LangChain quand :

• Prod haute fréquence (> 1000 req/min) avec SLA serré → raw SDK
• Besoins streaming très spécifiques (SSE custom, chunks intermédiaires)
• Maîtrise fine du prompt caching, des tools Anthropic strict mode
• Application mono-provider (juste OpenAI ou juste Anthropic) sans intention de bouger
• Equipe expérimentée qui veut < 500 lignes de dette de dépendance

Hybride recommandé : LangGraph pour orchestrer + raw SDK dans les nœuds.

🎓 Approfondissement — LangGraph en détail

LangGraph mérite sa propre section car c'est la vraie raison de rester sur LangChain en 2026. C'est un produit séparé maintenant (pip install langgraph), avec son propre cycle de release.

Concepts clés

• State : dict typé Python (TypedDict ou Pydantic) qui circule entre nodes. Mutation explicite via reducers (add_messages, custom).
• Nodes : fonctions qui reçoivent le state et retournent un partiel state update.
• Edges : transitions. Soit fixes (g.add_edge("A", "B")), soit conditionnelles (g.add_conditional_edges("A", router_fn, {"yes": "B", "no": END})).
• Checkpointers : persistance du state (MemorySaver pour tests, PostgresSaver / RedisSaver / SqliteSaver pour prod). Permet de reprendre une conversation à mi-chemin.
• Interrupts : pause l'exécution pour HITL (human-in-the-loop). L'humain peut éditer le state avant reprise.
• Time travel : checkpointers gardent un historique, on peut revenir à un état précédent et explorer un fork.

Pattern réel : agent juridique avec HITL

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import interrupt, Command
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class LegalAgentState(TypedDict):
    messages: Annotated[list, add_messages]
    documents: list
    draft_answer: str | None
    human_approved: bool

def retrieve_docs(state: LegalAgentState):
    query = state["messages"][-1].content
    docs = vector_store.similarity_search(query, k=6)
    return {"documents": docs}

def draft_answer(state: LegalAgentState):
    prompt = build_legal_prompt(state["messages"], state["documents"])
    draft = llm.invoke(prompt).content
    return {"draft_answer": draft}

def human_review(state: LegalAgentState):
    decision = interrupt({
        "draft": state["draft_answer"],
        "documents": [d.metadata for d in state["documents"]],
    })
    return {"human_approved": decision["approved"], "draft_answer": decision.get("edited", state["draft_answer"])}

def finalize(state: LegalAgentState):
    if state["human_approved"]:
        return {"messages": [AIMessage(content=state["draft_answer"])]}
    return {"messages": [AIMessage(content="Réponse refusée par l'avocat superviseur.")]}

g = StateGraph(LegalAgentState)
g.add_node("retrieve", retrieve_docs)
g.add_node("draft", draft_answer)
g.add_node("review", human_review)
g.add_node("finalize", finalize)
g.add_edge(START, "retrieve")
g.add_edge("retrieve", "draft")
g.add_edge("draft", "review")
g.add_edge("review", "finalize")
g.add_edge("finalize", END)

checkpointer = PostgresSaver.from_conn_string(os.environ["POSTGRES_URL"])
app = g.compile(checkpointer=checkpointer)

# Premier appel — s'arrête sur le interrupt
config = {"configurable": {"thread_id": "legal-session-42"}}
app.invoke({"messages": [HumanMessage("Question juridique...")]}, config=config)

# Plus tard, après revue humaine :
app.invoke(Command(resume={"approved": True, "edited": "Texte révisé..."}), config=config)

Ce pattern est ce qui rend LangGraph incontournable pour les workflows réglementés (legaltech, healthtech, finance) où un humain doit valider avant action critique.

LangGraph Studio (visualisation)

Outil desktop (Mac/Linux) qui visualise les graphes en temps réel, permet d'inspecter le state à chaque step, de modifier et rejouer. Mode dev critique pour debug. Gratuit pour les projets open source, intégré au tier LangSmith payant pour usage commercial.

LangGraph Platform (deployment)

Mode managed : déployer ses graphes comme services REST avec auto-scaling, HITL via webhooks, state store managé. ~$50/mois/agent. Alternative self-host : LangGraph Server (open source), à déployer sur K8s.

Comparaison rapide avec alternatives

Framework	Forces	Faiblesses
LangGraph	Stateful complet, HITL, checkpointing	Verbose, courbe d'apprentissage
LlamaIndex Workflows	Event-driven, Python pur	Communauté plus petite, moins d'outils
Pydantic AI	Type safety extrême, mince	Pas de checkpointing natif
Burr	State machine ultra simple	Moins de features
CrewAI	Multi-agent collaboratif	Opinionated, moins flexible
AutoGen	Microsoft, multi-agent conversational	Stateful limité

🏋️ Exercices

Progression : on part d'un RAG LangChain qui tourne, et on l'amène en état prod défendable face à un architecte. Chaque exercice escalade. Fais-les dans l'ordre.

Exercice 1 — LCEL streaming sans casse (échauffement)

Objectif — Construire prompt | ChatAnthropic(model="claude-opus-4-8") | StrOutputParser() qui streame vraiment token-par-token, puis prouver via usage que tu lis bien input/output tokens.

Indice/Solution — for chunk in chain.stream(...). Pas de temperature sur Opus 4.8 (400). Pour l'usage : model.bind(stream_usage=True) ou un UsageMetadataCallbackHandler. Piège : remplacer StrOutputParser par PydanticOutputParser et observer que le stream s'arrête de couler → comprendre pourquoi (le parser attend le message complet).

Exercice 2 — Casser puis réparer le thinking

Objectif — Écrire délibérément le code qui renvoie un HTTP 400 (thinking={"type":"enabled","budget_tokens":4000} + temperature=0 sur Opus 4.8), lire le message d'erreur, puis migrer vers adaptive thinking + effort en faisant apparaître le résumé de raisonnement dans le stream.

Indice/Solution — La 400 vient de budget_tokens ET de temperature (deux causes distinctes — corrige-les séparément pour bien voir laquelle déclenche quoi). Solution : thinking={"type":"adaptive","display":"summarized"} + model_kwargs={"output_config":{"effort":"high"}}, et temperature supprimé. Vérifie qu'avec display="omitted" (le défaut) tu reçois des blocs thinking au texte vide.

Exercice 3 — Structured output strict + streamable

Objectif — Extraire {montant: float, article: str, source: str} d'une question juridique, garantie schéma-valide, SANS prefill assistant (qui 400) et sans casser un éventuel stream amont.

Indice/Solution — with_structured_output(Schema, method="json_schema", strict=True). Compare avec method="function_calling" : mesure le taux de sorties invalides sur 50 questions. Bonus dur : que se passe-t-il si Claude refuse (stop_reason="refusal") ? La sortie ne respecte alors pas le schéma — gère ce cas explicitement au lieu de planter sur un parse.

Exercice 4 — Abstraction multi-provider qui ne 400 pas en silence

Objectif — Une factory get_model(provider, name) qui construit ChatAnthropic correctement selon le modèle : temperature autorisé sur Sonnet 4.6 / Haiku 4.5 mais INTERDIT sur Opus 4.8/4.7. Écrire le test qui aurait attrapé le bug « le fallback Opus→Sonnet marche, mais Sonnet→Opus 400 ».

Indice/Solution — Pose les params conditionnellement : kwargs = {}; if not name.startswith("claude-opus-4-7") and not name.startswith("claude-opus-4-8"): kwargs["temperature"]=0. Le test : paramétrer chaque modèle de la matrice et asserter qu'un appel minimal renvoie 200, pas 400. Sans ce test, le bug est invisible jusqu'à ce qu'un client bascule sur Opus en prod.

Exercice 5 — Prouver le prompt caching (défends le chiffre)

Objectif — Sur le RAG raw-SDK de la section end-to-end, poser cache_control sur le system prompt, envoyer 5 requêtes au contexte différent mais system identique, et démontrer chiffres à l'appui que le system est bien caché. Puis introduire un invalidateur silencieux et observer la régression.

Indice/Solution — Lis final.usage.cache_creation_input_tokens (1ʳᵉ requête) puis cache_read_input_tokens (requêtes 2-5). L'invalidateur : interpole datetime.now() dans le SYSTEM → cache_read retombe à 0 sans erreur. Subtilité Opus 4.8 : si le system fait < 4096 tokens, rien n'est caché (minimum cacheable), cache_creation_input_tokens=0 silencieusement — un système court ne profite jamais du cache. Défends le ROI : économie ≈ (tokens system cachés) × 0,9 × prix input × (req − 1).

Exercice 6 — LangGraph HITL production-grade (final boss)

Objectif — Reprendre l'agent juridique avec interrupt() + PostgresSaver, et le rendre robuste : reprise après crash du process (le checkpoint doit survivre), idempotence sur thread_id, et un evaluator LangSmith qui breake la CI si la correctness tombe sous 0,85.

Indice/Solution — Le state DOIT être immutable (retourner un nouveau dict, jamais state["x"].append). Tuer le process entre l'interrupt et le Command(resume=...), relancer, vérifier que le state reprend depuis Postgres. Idempotence : un même thread_id rejoué ne doit pas dupliquer le travail. Pour la CI : evaluate(...) sur un dataset gold ≥ 100 paires, assert results.aggregate_scores["correctness"] >= 0.85. Question qui sépare le mid du senior : pourquoi le HITL ici justifie LangGraph plutôt que 200 lignes maison ? → checkpointing + time-travel + reprise transactionnelle qu'on ne réécrit pas à la main pour un domaine réglementé.

🎤 En entretien

« Quand garderais-tu LangChain en prod, et quand le jetterais-tu ? » Garder LangGraph pour l'orchestration stateful (HITL, checkpointing, reprise) — c'est la seule brique qu'on ne refait pas en 200 lignes ; raw-SDK le hot path (streaming SSE custom, prompt caching précis, métriques par étape). Hybride : nœuds LangGraph qui appellent le SDK Anthropic en direct.

« Pourquoi temperature=0 plante sur Opus 4.8 et comment tu rends une factory multi-provider safe ? » Sur la famille Opus 4.7+ les paramètres de sampling (temperature/top_p/top_k) ET budget_tokens sont supprimés → 400 ; on pilote via adaptive thinking + output_config.effort. La factory pose ces params conditionnellement selon le modèle, avec un test qui couvre toute la matrice (sinon un fallback unidirectionnel masque le bug).

« Comment tu garantis et débogues le prompt caching à travers LangChain ? » LangChain ne pose pas cache_control automatiquement → model.bind(extra_body=...) ou raw SDK. On débogue en lisant usage.cache_read_input_tokens (≠ 0 = hit) ; un 0 persistant signale un invalidateur silencieux (timestamp/UUID dans le préfixe, JSON non trié, set de tools variable). Et sur Opus 4.8 le préfixe doit dépasser 4096 tokens sinon rien ne cache.

« LCEL with_structured_output te garantit-il un JSON valide ? Et si le modèle refuse ? » La garantie schéma vient de l'API Anthropic (mode json_schema/strict), pas du wrapper — donc on suit les release notes Anthropic, pas seulement LangChain. Le prefill assistant pour forcer du JSON est mort (400). Et un stop_reason="refusal" produit une sortie hors-schéma : il faut un branchement explicite avant le parse, sinon ça crash en prod sur un faux positif des classifieurs de sécurité.

🔗 Liens

• LangChain docs Python : https://python.langchain.com
• LangChain docs JS : https://js.langchain.com
• LangGraph : https://langchain-ai.github.io/langgraph/
• LangGraph Studio : https://studio.langchain.com
• LangSmith : https://docs.smith.langchain.com
• LCEL spec : https://python.langchain.com/docs/concepts/lcel/
• Critique honnête (Hamel Husain 2024, encore pertinente) : https://hamel.dev/blog/posts/evals/
• Migration 0.2 → 0.3 : https://python.langchain.com/docs/versions/v0_3/
• Comparaison frameworks (Pydantic AI team) : https://ai.pydantic.dev/comparisons/
• LangGraph templates : https://github.com/langchain-ai/langgraph/tree/main/examples
• LangChain Academy (cours gratuit) : https://academy.langchain.com