Architecture LangChain & LangGraph

Maîtrisez LangChain 1.x et LangGraph modernes pour construire des applications LLM sophistiquées avec agents, gestion d'état, mémoire et intégration d'outils.

Quand utiliser cette compétence

Construire des agents IA autonomes avec accès aux outils
Implémenter des workflows LLM complexes multi-étapes
Gérer la mémoire de conversation et l'état
Intégrer les LLM avec des sources de données externes et des API
Créer des composants d'application LLM modulaires et réutilisables
Implémenter des pipelines de traitement de documents
Construire des applications LLM de qualité production

Structure des paquets (LangChain 1.x)

langchain (1.2.x)         # Orchestration haut niveau
langchain-core (1.2.x)    # Abstractions centrales (messages, prompts, outils)
langchain-community       # Intégrations tierces
langgraph                 # Orchestration d'agents et gestion d'état
langchain-openai          # Intégrations OpenAI
langchain-anthropic       # Intégrations Anthropic/Claude
langchain-voyageai        # Embeddings Voyage AI
langchain-pinecone        # Magasin vectoriel Pinecone

Concepts clés

1. Agents LangGraph

LangGraph est le standard pour construire des agents en 2026. Il fournit :

Fonctionnalités clés :

StateGraph : Gestion d'état explicite avec état typé
Durable Execution : Les agents persistent à travers les défaillances
Human-in-the-Loop : Inspectez et modifiez l'état à tout moment
Memory : Mémoire court et long terme entre les sessions
Checkpointing : Sauvegardez et reprenez l'état de l'agent

Patterns d'agents :

ReAct : Reasoning + Acting avec create_react_agent
Plan-and-Execute : Nœuds de planification et exécution séparés
Multi-Agent : Routage superviseur entre agents spécialisés
Tool-Calling : Invocation d'outils structurée avec schémas Pydantic

2. Gestion d'état

LangGraph utilise TypedDict pour l'état explicite :

from typing import Annotated, TypedDict
from langgraph.graph import MessagesState

# État basé sur les messages simples
class AgentState(MessagesState):
    """Étend MessagesState avec des champs personnalisés."""
    context: Annotated[list, "retrieved documents"]

# État personnalisé pour les agents complexes
class CustomState(TypedDict):
    messages: Annotated[list, "conversation history"]
    context: Annotated[dict, "retrieved context"]
    current_step: str
    results: list

3. Systèmes de mémoire

Implémentations modernes de mémoire :

ConversationBufferMemory : Stocke tous les messages (conversations courtes)
ConversationSummaryMemory : Résume les anciens messages (conversations longues)
ConversationTokenBufferMemory : Fenêtrage basé sur les tokens
VectorStoreRetrieverMemory : Récupération par similarité sémantique
LangGraph Checkpointers : État persistant entre les sessions

4. Traitement de documents

Chargement, transformation et stockage de documents :

Composants :

Document Loaders : Chargez à partir de diverses sources
Text Splitters : Découpez les documents intelligemment
Vector Stores : Stockez et récupérez les embeddings
Retrievers : Récupérez les documents pertinents

5. Callbacks & Tracing

LangSmith est le standard pour l'observabilité :

Enregistrement des requêtes/réponses
Suivi de l'utilisation des tokens
Surveillance de la latence
Suivi des erreurs
Visualisation des traces

Démarrage rapide

Agent ReAct moderne avec LangGraph

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
import ast
import operator

# Initialisez le LLM (Claude Sonnet 4.6 recommandé)
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0)

# Définissez les outils avec des schémas Pydantic
@tool
def search_database(query: str) -> str:
    """Recherchez dans la base de données interne."""
    # Votre logique de recherche de base de données
    return f"Results for: {query}"

@tool
def calculate(expression: str) -> str:
    """Évaluez avec sécurité une expression mathématique.

    Supporte : +, -, *, /, **, %, parenthèses
    Exemple : '(2 + 3) * 4' retourne '20'
    """
    # Évaluation mathématique sécurisée avec ast
    allowed_operators = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
        ast.Pow: operator.pow,
        ast.Mod: operator.mod,
        ast.USub: operator.neg,
    }

    def _eval(node):
        if isinstance(node, ast.Constant):
            return node.value
        elif isinstance(node, ast.BinOp):
            left = _eval(node.left)
            right = _eval(node.right)
            return allowed_operators[type(node.op)](left, right)
        elif isinstance(node, ast.UnaryOp):
            operand = _eval(node.operand)
            return allowed_operators[type(node.op)](operand)
        else:
            raise ValueError(f"Unsupported operation: {type(node)}")

    try:
        tree = ast.parse(expression, mode='eval')
        return str(_eval(tree.body))
    except Exception as e:
        return f"Error: {e}"

tools = [search_database, calculate]

# Créez un checkpointer pour la persistance de la mémoire
checkpointer = MemorySaver()

# Créez l'agent ReAct
agent = create_react_agent(
    llm,
    tools,
    checkpointer=checkpointer
)

# Exécutez l'agent avec ID de thread pour la mémoire
config = {"configurable": {"thread_id": "user-123"}}
result = await agent.ainvoke(
    {"messages": [("user", "Search for Python tutorials and calculate 25 * 4")]},
    config=config
)

Patterns d'architecture

Pattern 1 : RAG avec LangGraph

from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic
from langchain_voyageai import VoyageAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from typing import TypedDict, Annotated

class RAGState(TypedDict):
    question: str
    context: Annotated[list[Document], "retrieved documents"]
    answer: str

# Initialisez les composants
llm = ChatAnthropic(model="claude-sonnet-4-6")
embeddings = VoyageAIEmbeddings(model="voyage-3-large")
vectorstore = PineconeVectorStore(index_name="docs", embedding=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# Définissez les nœuds
async def retrieve(state: RAGState) -> RAGState:
    """Récupérez les documents pertinents."""
    docs = await retriever.ainvoke(state["question"])
    return {"context": docs}

async def generate(state: RAGState) -> RAGState:
    """Générez la réponse à partir du contexte."""
    prompt = ChatPromptTemplate.from_template(
        """Answer based on the context below. If you cannot answer, say so.

        Context: {context}

        Question: {question}

        Answer:"""
    )
    context_text = "\n\n".join(doc.page_content for doc in state["context"])
    response = await llm.ainvoke(
        prompt.format(context=context_text, question=state["question"])
    )
    return {"answer": response.content}

# Construisez le graphe
builder = StateGraph(RAGState)
builder.add_node("retrieve", retrieve)
builder.add_node("generate", generate)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)

rag_chain = builder.compile()

# Utilisez la chaîne
result = await rag_chain.ainvoke({"question": "What is the main topic?"})

Pattern 2 : Agent personnalisé avec outils structurés

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    """Entrée pour la recherche en base de données."""
    query: str = Field(description="Search query")
    filters: dict = Field(default={}, description="Optional filters")

class EmailInput(BaseModel):
    """Entrée pour envoyer un email."""
    recipient: str = Field(description="Email recipient")
    subject: str = Field(description="Email subject")
    content: str = Field(description="Email body")

async def search_database(query: str, filters: dict = {}) -> str:
    """Recherchez dans la base de données interne."""
    # Votre logique de recherche de base de données
    return f"Results for '{query}' with filters {filters}"

async def send_email(recipient: str, subject: str, content: str) -> str:
    """Envoyez un email au destinataire spécifié."""
    # Logique d'envoi d'email
    return f"Email sent to {recipient}"

tools = [
    StructuredTool.from_function(
        coroutine=search_database,
        name="search_database",
        description="Search internal database",
        args_schema=SearchInput
    ),
    StructuredTool.from_function(
        coroutine=send_email,
        name="send_email",
        description="Send an email",
        args_schema=EmailInput
    )
]

agent = create_react_agent(llm, tools)

Pattern 3 : Workflow multi-étapes avec StateGraph

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Literal

class WorkflowState(TypedDict):
    text: str
    entities: list
    analysis: str
    summary: str
    current_step: str

async def extract_entities(state: WorkflowState) -> WorkflowState:
    """Extrayez les entités clés du texte."""
    prompt = f"Extract key entities from: {state['text']}\n\nReturn as JSON list."
    response = await llm.ainvoke(prompt)
    return {"entities": response.content, "current_step": "analyze"}

async def analyze_entities(state: WorkflowState) -> WorkflowState:
    """Analysez les entités extraites."""
    prompt = f"Analyze these entities: {state['entities']}\n\nProvide insights."
    response = await llm.ainvoke(prompt)
    return {"analysis": response.content, "current_step": "summarize"}

async def generate_summary(state: WorkflowState) -> WorkflowState:
    """Générez le résumé final."""
    prompt = f"""Summarize:
    Entities: {state['entities']}
    Analysis: {state['analysis']}

    Provide a concise summary."""
    response = await llm.ainvoke(prompt)
    return {"summary": response.content, "current_step": "complete"}

def route_step(state: WorkflowState) -> Literal["analyze", "summarize", "end"]:
    """Routez vers l'étape suivante selon l'état actuel."""
    step = state.get("current_step", "extract")
    if step == "analyze":
        return "analyze"
    elif step == "summarize":
        return "summarize"
    return "end"

# Construisez le workflow
builder = StateGraph(WorkflowState)
builder.add_node("extract", extract_entities)
builder.add_node("analyze", analyze_entities)
builder.add_node("summarize", generate_summary)

builder.add_edge(START, "extract")
builder.add_conditional_edges("extract", route_step, {
    "analyze": "analyze",
    "summarize": "summarize",
    "end": END
})
builder.add_conditional_edges("analyze", route_step, {
    "summarize": "summarize",
    "end": END
})
builder.add_edge("summarize", END)

workflow = builder.compile()

Pattern 4 : Orchestration multi-agents

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent
from langchain_core.messages import HumanMessage
from typing import Literal

class MultiAgentState(TypedDict):
    messages: list
    next_agent: str

# Créez des agents spécialisés
researcher = create_react_agent(llm, research_tools)
writer = create_react_agent(llm, writing_tools)
reviewer = create_react_agent(llm, review_tools)

async def supervisor(state: MultiAgentState) -> MultiAgentState:
    """Routez vers l'agent approprié selon la tâche."""
    prompt = f"""Based on the conversation, which agent should handle this?

    Options:
    - researcher: For finding information
    - writer: For creating content
    - reviewer: For reviewing and editing
    - FINISH: Task is complete

    Messages: {state['messages']}

    Respond with just the agent name."""

    response = await llm.ainvoke(prompt)
    return {"next_agent": response.content.strip().lower()}

def route_to_agent(state: MultiAgentState) -> Literal["researcher", "writer", "reviewer", "end"]:
    """Routez selon la décision du superviseur."""
    next_agent = state.get("next_agent", "").lower()
    if next_agent == "finish":
        return "end"
    return next_agent if next_agent in ["researcher", "writer", "reviewer"] else "end"

# Construisez le graphe multi-agents
builder = StateGraph(MultiAgentState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("writer", writer)
builder.add_node("reviewer", reviewer)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route_to_agent, {
    "researcher": "researcher",
    "writer": "writer",
    "reviewer": "reviewer",
    "end": END
})

# Chaque agent revient au superviseur
for agent in ["researcher", "writer", "reviewer"]:
    builder.add_edge(agent, "supervisor")

multi_agent = builder.compile()

Gestion de la mémoire

Mémoire basée sur les tokens avec LangGraph

from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent

# Checkpointer en mémoire (développement)
checkpointer = MemorySaver()

# Créez l'agent avec mémoire persistante
agent = create_react_agent(llm, tools, checkpointer=checkpointer)

# Chaque thread_id maintient une conversation séparée
config = {"configurable": {"thread_id": "session-abc123"}}

# Les messages persistent entre les appels avec le même thread_id
result1 = await agent.ainvoke({"messages": [("user", "My name is Alice")]}, config)
result2 = await agent.ainvoke({"messages": [("user", "What's my name?")]}, config)
# L'agent se souvient : "Your name is Alice"

Mémoire production avec PostgreSQL

from langgraph.checkpoint.postgres import PostgresSaver

# Checkpointer production
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/langgraph"
)

agent = create_react_agent(llm, tools, checkpointer=checkpointer)

Mémoire vectorielle pour le contexte long terme

from langchain_community.vectorstores import Chroma
from langchain_voyageai import VoyageAIEmbeddings

embeddings = VoyageAIEmbeddings(model="voyage-3-large")
memory_store = Chroma(
    collection_name="conversation_memory",
    embedding_function=embeddings,
    persist_directory="./memory_db"
)

async def retrieve_relevant_memory(query: str, k: int = 5) -> list:
    """Récupérez les conversations passées pertinentes."""
    docs = await memory_store.asimilarity_search(query, k=k)
    return [doc.page_content for doc in docs]

async def store_memory(content: str, metadata: dict = {}):
    """Stockez la conversation dans la mémoire long terme."""
    await memory_store.aadd_texts([content], metadatas=[metadata])

Système de Callbacks & LangSmith

Tracing LangSmith

import os
from langchain_anthropic import ChatAnthropic

# Activez le tracing LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-project"

# Toutes les opérations LangChain/LangGraph sont automatiquement tracées
llm = ChatAnthropic(model="claude-sonnet-4-6")

Gestionnaire de Callback personnalisé

from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict, List

class CustomCallbackHandler(BaseCallbackHandler):
    def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs
    ) -> None:
        print(f"LLM started with {len(prompts)} prompts")

    def on_llm_end(self, response, **kwargs) -> None:
        print(f"LLM completed: {len(response.generations)} generations")

    def on_llm_error(self, error: Exception, **kwargs) -> None:
        print(f"LLM error: {error}")

    def on_tool_start(
        self, serialized: Dict[str, Any], input_str: str, **kwargs
    ) -> None:
        print(f"Tool started: {serialized.get('name')}")

    def on_tool_end(self, output: str, **kwargs) -> None:
        print(f"Tool completed: {output[:100]}...")

# Utilisez les callbacks
result = await agent.ainvoke(
    {"messages": [("user", "query")]},
    config={"callbacks": [CustomCallbackHandler()]}
)

Streaming de réponses

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-6", streaming=True)

# Stream les tokens
async for chunk in llm.astream("Tell me a story"):
    print(chunk.content, end="", flush=True)

# Stream les événements de l'agent
async for event in agent.astream_events(
    {"messages": [("user", "Search and summarize")]},
    version="v2"
):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")
    elif event["event"] == "on_tool_start":
        print(f"\n[Using tool: {event['name']}]")

Stratégies de test

import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_agent_tool_selection():
    """Testez que l'agent sélectionne le bon outil."""
    with patch.object(llm, 'ainvoke') as mock_llm:
        mock_llm.return_value = AsyncMock(content="Using search_database")

        result = await agent.ainvoke({
            "messages": [("user", "search for documents")]
        })

        # Vérifiez que l'outil a été appelé
        assert "search_database" in str(result)

@pytest.mark.asyncio
async def test_memory_persistence():
    """Testez que la mémoire persiste entre les appels."""
    config = {"configurable": {"thread_id": "test-thread"}}

    # Premier message
    await agent.ainvoke(
        {"messages": [("user", "Remember: the code is 12345")]},
        config
    )

    # Deuxième message devrait se souvenir
    result = await agent.ainvoke(
        {"messages": [("user", "What was the code?")]},
        config
    )

    assert "12345" in result["messages"][-1].content

Optimisation des performances

1. Cache avec Redis

from langchain_community.cache import RedisCache
from langchain_core.globals import set_llm_cache
import redis

redis_client = redis.Redis.from_url("redis://localhost:6379")
set_llm_cache(RedisCache(redis_client))

2. Traitement par batch asynchrone

import asyncio
from langchain_core.documents import Document

async def process_documents(documents: list[Document]) -> list:
    """Traitez les documents en parallèle."""
    tasks = [process_single(doc) for doc in documents]
    return await asyncio.gather(*tasks)

async def process_single(doc: Document) -> dict:
    """Traitez un document unique."""
    chunks = text_splitter.split_documents([doc])
    embeddings = await embeddings_model.aembed_documents(
        [c.page_content for c in chunks]
    )
    return {"doc_id": doc.metadata.get("id"), "embeddings": embeddings}

3. Connection pooling

from langchain_pinecone import PineconeVectorStore
from pinecone import Pinecone

# Réutilisez le client Pinecone
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
index = pc.Index("my-index")

# Créez le magasin vectoriel avec l'index existant
vectorstore = PineconeVectorStore(index=index, embedding=embeddings)

langchain-architecture