StateGraph : le workflow comme graphe d'états

L'idée : une machine à états explicite

LangGraph modélise un workflow comme un graphe : des nodes (fonctions) qui transforment un state partagé, reliés par des edges (dont certains conditionnels). Là où une fonction géante enchaînerait des if/else opaques, le graphe rend la topologie déclarée, visualisable et streamable.

Le state : un TypedDict, contrat de tout le pipeline

from typing_extensions import TypedDict

class UnifiedState(TypedDict, total=False):
    # INPUT
    context: ContextMetadata
    raw_document: str
    # EXTRACTION
    structured_order: StructuredOrder | None
    # MATCHING
    match_results: list[MatchResult]
    # GLOBAL
    status: WorkflowStatus
    errors: list[str]

total=False : chaque node ne remplit que sa section. Le state est le contrat central — c'est lui qu'on lit pour comprendre le pipeline (le projet utilise un state « plat » unique partagé par tous les subgraphs, section par section).

Nodes et edges : la topologie du matcher

workflow = StateGraph(UnifiedState)
workflow.add_node("extractor", extract_node)      # un FICHIER par node (règle projet)
workflow.add_node("structurer", structure_node)
workflow.add_node("validator", validate_node)
workflow.add_node("matcher_fast", fast_match_node)
workflow.add_node("matcher_deep", deep_match_node)

workflow.add_edge(START, "extractor")
workflow.add_edge("extractor", "structurer")
workflow.add_conditional_edges(
    "matcher_fast",
    _should_deep_match,                    # fonction PURE : lit le state, renvoie une clé
    {"deep": "matcher_deep", "done": END},
)
graph = workflow.compile()

Un node = une fonction async qui reçoit le state et retourne un delta (return {"match_results": results}) — LangGraph fusionne. Les décisions de routage sont des fonctions pures (_should_deep_match lit le score et renvoie "deep" ou "done") : testables unitairement sans LLM ni DB.

Pourquoi un graphe et pas des appels de fonctions ?

1. Observabilité : chaque node est une frontière naturelle de trace/timing (Langfuse, module 7) et d'événement SSE (« matcher_fast terminé » → barre de progression) ;
2. Évolutivité de topologie : ajouter un node « OCR si scanné » = un edge conditionnel, pas une réécriture ;
3. Testabilité : chaque node se teste isolément avec un state d'entrée fabriqué ;
4. Persistance : le checkpointer (leçon 3) sauvegarde le state à chaque étape — impossible avec une pile d'appels.