add_node / add_edge / add_conditional_edges : remplir le blueprint

源码版本1.2.9

Responsabilités

StateGraph expose trois méthodes centrales d'enregistrement : add_node enveloppe une fonction / Runnable en nœud (add_node:662), add_edge ajoute une arête déterministe (add_edge:915), add_conditional_edges ajoute un branchement conditionnel (add_conditional_edges:969). Toutes trois renvoient Self, ce qui permet le chaînage : builder.add_node("a", a).add_node("b", b).add_edge("a", "b").add_conditional_edges("b", route). Elles ne font qu'ajouter dans les conteneurs builder.nodes / builder.edges / builder.branches ; la traduction réelle en PregelNode + abonnements de canaux n'a lieu qu'à compile(), via attach_node / attach_edge / attach_branch.

Elles se situent tout au bord de l'API utilisateur : presque chaque ligne que vous écrivez en construisant un graphe appelle l'une de ces trois méthodes. add_node est la plus complexe — elle déduit input_schema depuis la signature de la fonction, déduit les destinations depuis un type de retour Literal[...], et enveloppe error_handler dans un nœud séparé __error_handler__{node} (injection error handler:856-870). add_edge est plus direct : soit elle va dans le set edges, soit dans waiting_edges (arêtes multi-sources, qui attendent que toutes les sources soient terminées). add_conditional_edges est très mince : elle enveloppe path dans un Runnable, puis délègue à BranchSpec.from_path (from_path:89) pour calculer le dict ends stocké dans branches[source].

Motivation de conception

Pourquoi add_node déduit-il autant de choses depuis la signature de la fonction ?

• Moins de boilerplate : quand input_schema n'est pas passé explicitement, il est déduit de l'annotation du premier paramètre (inferred input schema:815-825) — ainsi def my_node(state: MyState): ... utilise automatiquement MyState ; quand le type de retour est Literal["a", "b", "__end__"], il est automatiquement pris comme destinations (Literal destinations:840-846), sans avoir à écrire destinations=....
• Nom de nœud optionnel : le paramètre node peut être soit une chaîne, soit la fonction elle-même ; dans ce cas __name__ est utilisé comme nom de nœud (inférence du nom de nœud:768-773) — pour un Runnable, c'est get_name().
• error_handler est un nœud à part (nœud error_handler:857-870) — ce n'est pas un champ du spec, mais un nœud ordinaire généré sous le nom __error_handler__{node}, avec un pointeur stocké dans le champ error_handler_node. Ainsi le handler est lui-même un nœud Pregel et bénéficie de retry / metadata / tracing comme n'importe quel nœud.
• Sémantique unifiée des arêtes multi-sources : add_edge(["a", "b"], "c") ne signifie pas « quand a finit, lance c ; quand b finit, lance c », mais « c n'est lancé qu'après que a et b ont tous les deux terminé » (sémantique multi-start:917-921), stocké dans le set waiting_edges, puis traduit à la compilation en canal NamedBarrierValue.
• La valeur de retour d'une arête conditionnelle peut omettre path_map : sans path_map, il est déduit du type de retour Literal["a", "b"] de path (inférence Literal path_map:103-115), ce qui fait fonctionner automatiquement def route(state) -> Literal["a", "b"]: ... ; en l'absence des deux, la visualisation du graphe supposera qu'elle peut sauter vers n'importe quel nœud (warning:994-997).
• Branches de même nœud et de même nom = erreur (branche en double:1009-1012) — un nœud peut avoir plusieurs arêtes conditionnelles, mais chacune doit avoir un nom de condition unique, car branches[source] est un dict indexé par nom.

Fichiers clés

• signature add_node:662-676 — reçoit node / action / defer / metadata / input_schema / retry_policy / cache_policy / error_handler / destinations / timeout.
• inférence du nom de nœud:768-790 — chaîne utilisée telle quelle, fonction via __name__, Runnable via get_name() ; les mots réservés START / END / NS_SEP / NS_END déclenchent une erreur.
• inférence de types depuis la signature:803-848 — déduit inferred_input_schema et les destinations Command[Literal[...]] du type de retour à partir des type hints de __call__.
• injection error_handler:856-870 — génère un nœud séparé __error_handler__{node} stocké dans self.nodes, avec le champ is_error_handler=True.
• stockage StateNodeSpec:872-907 — stocke le spec dans self.nodes selon trois cas : input_schema / inferred_input_schema / self.state_schema.
• add_edge:915-967 — source unique dans le set edges, sources multiples dans waiting_edges ; la source ne peut pas être END, la cible ne peut pas être START.
• add_conditional_edges:969-1017 — enveloppe path en Runnable, BranchSpec.from_path calcule ends, stocké dans self.branches[source][name].
• BranchSpec.from_path:83-120 — gère les trois formes de path_map : dict / list / inférence depuis le type de retour Literal ; déduit aussi input_schema.
• add_sequence:1019-1044 — enregistre en une fois une série de nœuds (sucre syntaxique, qui boucle en interne sur add_node).
• attach_edge:1537-1561 — à la compilation : pour une arête simple, ajoute un ChannelWrite aux writers du nœud source pointant vers le canal cible ; pour une arête multi-sources, enregistre un canal NamedBarrierValue comme jointure.
• attach_branch:1563-1596 — à la compilation : traduit un BranchSpec en ChannelWrite.register_writer, et selon ends décide vers quel canal cible écrire.

Flux de données

add_node fait réellement son travail dans ces deux passages : inférer l'input schema et les destinations du type de retour, puis stocker le spec dans self.nodes :

if (
    isfunction(action)
    or ismethod(action)
    or ismethod(getattr(action, "__call__", None))
) and (
    hints := get_type_hints(getattr(action, "__call__"))
    or get_type_hints(action)
):
    if input_schema is None:
        first_parameter_name = next(
            iter(inspect.signature(cast(FunctionType, action)).parameters.keys())
        )
        if input_hint := hints.get(first_parameter_name):
            if isinstance(input_hint, type) and get_type_hints(input_hint):
                inferred_input_schema = input_hint
    if rtn := hints.get("return"):
        rtn_origin = get_origin(rtn)
        if rtn_origin is Union:
            rtn_args = get_args(rtn)
            for arg in rtn_args:
                arg_origin = get_origin(arg)
                if arg_origin is Command:
                    rtn = arg
                    rtn_origin = arg_origin
                    break
        if (
            rtn_origin is Command
            and (rargs := get_args(rtn))
            and get_origin(rargs[0]) is Literal
            and (vals := get_args(rargs[0]))
        ):
            ends = vals

(inférence de types:806-846)

Le spec est ensuite stocké dans le dict self.nodes selon trois cas — input_schema explicite / inferred_input_schema déduit / repli sur self.state_schema :

if input_schema is not None:
    self.nodes[node] = StateNodeSpec[NodeInputT, ContextT](
        coerce_to_runnable(action, name=node, trace=False),
        metadata,
        input_schema=input_schema,
        retry_policy=retry_policy,
        cache_policy=cache_policy,
        error_handler_node=handler_node_name,
        ends=ends,
        defer=defer,
        timeout=timeout,
    )

(StateNodeSpec:872-883)

add_conditional_edges en revanche est très mince — le cœur est d'envelopper path dans un Runnable puis de laisser BranchSpec.from_path calculer le dict ends :

path = coerce_to_runnable(path, name=None, trace=True)
name = path.name or "condition"
if name in self.branches[source]:
    raise ValueError(
        f"Branch with name `{path.name}` already exists for node `{source}`"
    )
self.branches[source][name] = BranchSpec.from_path(path, path_map, True)
if schema := self.branches[source][name].input_schema:
    self._add_schema(schema)
return self

(implémentation add_conditional_edges:1005-1017)

Limites et échecs

• Nom de nœud en double = erreur directe (renom:792-793) — Node already present, pas de surcharge autorisée.
• Mots réservés pour les noms de nœuds (reserved:794-801) — START / END / NS_SEP (|) / NS_END (:) sont interdits, les deux derniers casseraient le assemblage de checkpoint_ns.
• add_edge avec source END = erreur (END interdit en source:939-940), et symétriquement START ne peut pas être la cible (START interdit en cible:941-942) — ces deux constantes sont l'entrée et la sortie du graphe, elles ne peuvent pas être inversées.
• Les arêtes multi-sources exigent que toutes les sources aient été add_node au préalable (validation multi-sources:956-964) — sinon attach_edge ne trouvera pas les writers de la source à la compilation.
• Deux branches de même nom sur un même nœud = erreur (branche en double:1009-1012), mais plusieurs branches de noms distincts peuvent coexister — c'est la base du routage mixte conditionnel + déterministe.
• add_node sur un graphe déjà compilé ne fait qu'un warning (compiled warning:778-782), les modifications ne sont pas répercutées sur l'instance compilée — piège classique d'ordre.

Résumé

add_node / add_edge / add_conditional_edges sont l'interface d'écriture de StateGraph, pas complexe en soi ; ce qui l'est, c'est la « magie » d'inférence du schema et des destinations depuis la signature de la fonction. Une fois ces trois méthodes comprises, vous avez essentiellement lu toute l'API mutable de StateGraph ; la suite est compile qui traduit ces specs en PregelNode, et le modèle d'exécution dans moteur Pregel.

Voir la documentation officielle : LangGraph docs · README.