add_node / add_edge / add_conditional_edges: Die Blaupause mit Inhalt füllen

源码版本1.2.9

Verantwortung

StateGraph stellt drei zentrale Registriermethoden bereit: add_node wickelt eine Funktion / ein Runnable als Knoten ein (add_node:662), add_edge fügt eine feste Kante hinzu (add_edge:915), und add_conditional_edges fügt eine Verzweigungsbedingung hinzu (add_conditional_edges:969). Alle drei Methoden geben Self zurück, sind also verkettbar: builder.add_node("a", a).add_node("b", b).add_edge("a", "b").add_conditional_edges("b", route). Sie alle appenden nur in die Container builder.nodes / builder.edges / builder.branches; die eigentliche Übersetzung in PregelNode + Channel-Abonnements erfolgt erst beim compile() durch attach_node / attach_edge / attach_branch.

Sie stehen ganz außen in der Nutzer-API-Schicht: Beim Schreiben eines Graphen besteht fast jede Zeile aus Aufrufen dieser drei Methoden. add_node ist am komplexesten – es leitet aus der Funktionssignatur das input_schema ab, aus dem Rückgabetyp Literal[...] die destinations und wickelt den error_handler als separaten Knoten __error_handler__{node} ein (error handler Injektion:856-870). add_edge ist relativ direkt: entweder in das Set edges oder, bei mehreren Startpunkten, in das Set waiting_edges (Mehr-Eingangs-Kante, die wartet, bis alle Startpunkte fertig sind). add_conditional_edges ist am dünnsten: Es wickelt lediglich path in ein Runnable und überlässt BranchSpec.from_path (from_path:89) die Berechnung des ends-Dicts, das dann in branches[source] abgelegt wird.

Entwurfsmotivation

Warum leitet add_node so vieles aus der Funktionssignatur ab?

• Weniger Boilerplate: Ohne explizites input_schema wird aus der Typannotation des ersten Parameters abgeleitet (inferred input schema:815-825) – def my_node(state: MyState): ... verwendet automatisch MyState; ist der Rückgabetyp Literal["a", "b", "__end__"], gilt er automatisch als destinations (Literal destinations:840-846), ohne dass destinations=... angegeben werden muss.
• Knotenname optional: Der Parameter node kann entweder ein String oder die Funktion selbst sein; bei einem Funktionsobjekt wird __name__ als Knotenname verwendet (node Namensableitung:768-773) – bei einem Runnable ist es get_name().
• error_handler ist ein eigenständiger Knoten (error_handler Knoten:857-870) – er ist kein Feld des Specs, sondern ein regulärer Knoten namens __error_handler__{node}, auf den im Feld error_handler_node ein Zeiger gesetzt wird. Damit ist der Handler selbst ein Pregel-Knoten und profitiert von retry / metadata / tracing etc.
• Einheitliche Mehrfach-Eingangs-Semantik: add_edge(["a", "b"], "c") bedeutet nicht «nach a kommt c, nach b kommt c», sondern «erst wenn sowohl a als auch b fertig sind, wird c ausgeführt» (multi-start Semantik:917-921); gespeichert im Set waiting_edges und beim compile zu einem NamedBarrierValue-Channel übersetzt.
• Bedingte Kanten kommen ohne path_map aus: Ohne path_map wird aus dem Rückgabetyp Literal["a", "b"] der path-Funktion abgeleitet (Literal leitet path_map ab:103-115), sodass def route(state) -> Literal["a", "b"]: ... automatisch funktioniert; ist nichts angegeben, nimmt die Visualisierung an, dass ein Sprung zu jedem beliebigen Knoten möglich ist (warning:994-997).
• Branch mit gleichem Namen am selben Knoten wirft (branch 重名:1009-1012) – ein Knoten darf mehrere conditional edges haben, aber der Condition-Name jedes einzelnen muss eindeutig sein, da branches[source] ein Dict nach Name ist.

Schlüsseldateien

• add_node Signatur:662-676 — Empfängt node / action / defer / metadata / input_schema / retry_policy / cache_policy / error_handler / destinations / timeout.
• node Namensableitung:768-790 — String wird direkt verwendet, Funktion über __name__, Runnable über get_name(); reservierte Wörter START / END / NS_SEP / NS_END führen zu Fehlern.
• Signatur-Typableitung:803-848 — Leitet aus den Type Hints von __call__ das inferred_input_schema ab und aus dem Rückgabetyp die Command[Literal[...]]-destinations.
• error_handler Injektion:856-870 — Erzeugt den eigenständigen Knoten __error_handler__{node}, der in self.nodes abgelegt wird, mit dem Feld is_error_handler=True.
• StateNodeSpec speichern:872-907 — Speichert den Spec in drei Stufen je nach input_schema / inferred_input_schema / self.state_schema in self.nodes.
• add_edge:915-967 — Einzelner Startpunkt geht ins Set edges, mehrere Startpunkte gehen ins Set waiting_edges; Startpunkt darf nicht END sein, Endpunkt nicht START.
• add_conditional_edges:969-1017 — Wickelt path in ein Runnable, BranchSpec.from_path berechnet ends, speichert in self.branches[source][name].
• BranchSpec.from_path:83-120 — Behandelt drei Formen von path_map: dict / list / Ableitung aus dem Rückgabe-Typ Literal; leitet außerdem input_schema ab.
• add_sequence:1019-1044 — Registriert eine Sequenz von Knoten auf einmal (Syntactic Sugar, intern Schleife aus add_node).
• attach_edge:1537-1561 — Compile-Zeit: Bei Einzel-Eingang wird den writers des Startknotens ein ChannelWrite auf den Endpunkt-Channel hinzugefügt; bei Mehrfacheingang wird ein NamedBarrierValue-Channel als Join registriert.
• attach_branch:1563-1596 — Compile-Zeit: Übersetzt den BranchSpec in ChannelWrite.register_writer und entscheidet anhand von ends, in welchen Ziel-Channel geschrieben wird.

Datenfluss

add_node erledigt die eigentliche Arbeit in zwei Abschnitten – es leitet input schema und destinations aus dem Rückgabetyp ab und speichert den Spec in 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

(Typableitung:806-846)

Anschließend wird in drei Stufen in das Dict self.nodes gespeichert – input_schema explizit / inferred_input_schema abgeleitet / Fallback auf 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 ist hingegen sehr dünn – der Kern besteht darin, path in ein Runnable zu wickeln und BranchSpec.from_path das ends-Dict berechnen zu lassen:

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

(add_conditional_edges Implementierung:1005-1017)

Grenzen und Fehler

• Doppelte Knotennamen werfen direkt (重名:792-793) – Node \` already present`, Überschreiben ist nicht erlaubt.
• Reservierte Knotennamen (reserved:794-801) – START / END / NS_SEP (|) / NS_END (:) sind unzulässig; die letzten beiden würden das Zusammenbauen von checkpoint_ns zerstören.
• add_edge mit END als Startpunkt wirft (END nicht als Start:939-940) und entsprechend START als Endpunkt (START nicht als Endpunkt:941-942) – diese Konstanten sind Ein- und Ausgänge des Graphen und dürfen nicht invertiert werden.
• Mehrfacheingang setzt voraus, dass alle Startpunkte vorher add_node waren (Mehrfacheingang-Validierung:956-964) – sonst findet attach_edge beim Compile die writers der Startpunkte nicht.
• Branch mit gleichem Namen am selben Knoten wirft (branch 重名:1009-1012), aber mehrere Branches mit unterschiedlichen Namen koexistieren – die Grundlage für gemischtes conditional + deterministisches Routing.
• add_node auf einem bereits kompilierten Graphen liefert nur eine Warnung, keinen Fehler (compiled warning:778-782); Änderungen werden nicht in die bereits kompilierte Instanz übernommen – ein typischer Reihenfolgen-Fallen.

Zusammenfassung

add_node / add_edge / add_conditional_edges sind die Schreib-Schnittstellen des StateGraph und selbst nicht kompliziert; komplex ist nur die «Magie», schema und destinations aus der Funktionssignatur abzuleiten. Wer diese drei Methoden verstanden hat, hat im Wesentlichen die gesamte mutable API von StateGraph gelesen; als Nächstes folgt compile, wie diese Specs in PregelNode übersetzt werden, und das Ausführungsmodell steht in Pregel 引擎.

Siehe offizielle Dokumentation: LangGraph 文档 · README.