add_node / add_edge / add_conditional_edges: rellenar el plano

源码版本1.2.9

Responsabilidades

StateGraph ofrece tres métodos de registro centrales: add_node envuelve una función / Runnable como nodo(add_node:662), add_edge añade un borde determinista(add_edge:915), y add_conditional_edges añade una rama condicional(add_conditional_edges:969). Los tres devuelven Self, así que se pueden encadenar: builder.add_node("a", a).add_node("b", b).add_edge("a", "b").add_conditional_edges("b", route). Los tres se limitan a añadir entradas a builder.nodes / builder.edges / builder.branches; la traducción real a PregelNode + suscripciones de canal se hace en compile() con attach_node / attach_edge / attach_branch.

Su posición es la más externa de la API de usuario: casi cada línea que escribes al construir un grafo es una llamada a uno de estos tres métodos. add_node es el más complejo — deduce input_schema de la signatura de la función, deduce destinations del tipo de retorno Literal[...], y envuelve error_handler como un nodo independiente __error_handler__{node}(error handler 注入:856-870). add_edge es directo: va al set edges, o a waiting_edges si tiene múltiples orígenes (espera a que terminen todos). add_conditional_edges es el más fino; solo envuelve path en un Runnable y lo pasa a BranchSpec.from_path(from_path:89) para calcular el dict ends y guardarlo en branches[source].

Motivación de diseño

¿Por qué add_node deduce tantas cosas a partir de la signatura de la función?

• Menos boilerplate: si no se pasa input_schema explícito, se deduce del tipo del primer parámetro(inferred input schema:815-825), de modo que def my_node(state: MyState): ... use automáticamente MyState; si el tipo de retorno es Literal["a", "b", "__end__"], se trata automáticamente como destinations(Literal destinations:840-846), sin necesidad de destinations=....
• El nombre del nodo se puede omitir: el parámetro node puede ser tanto un string como la propia función; si es función, se toma __name__ como nombre(node 名推断:768-773), y si es Runnable, se usa get_name().
• error_handler es un nodo independiente(error_handler 节点:857-870) — no es un campo del spec, sino que se genera como un nodo regular llamado __error_handler__{node}, y se anota el puntero en el campo error_handler_node. Así, el handler es a su vez un nodo Pregel y disfruta de retry / metadata / tracing como cualquier nodo.
• Semántica unificada de múltiples orígenes: add_edge(["a", "b"], "c") no es «a termina, corre c; b termina, corre c», sino «corre c solo cuando a y b han terminado»(multi-start 语义:917-921); se guarda en waiting_edges para que en compilación se traduzca a un canal NamedBarrierValue.
• El valor de retorno de un borde condicional puede omitir el path_map: si no se pasa path_map, se deduce del tipo de retorno Literal["a", "b"] de la función path(Literal 推 path_map:103-115), de modo que def route(state) -> Literal["a", "b"]: ... funciona solo; si tampoco se da, la visualización del grafo asumirá que puede saltar a cualquier nodo(warning:994-997).
• Misma condición de borde con mismo nombre da error en un mismo nodo(branch 重名:1009-1012) — un nodo puede tener varios conditional edge, pero el name de cada uno debe ser único, porque branches[source] es un dict por name.

Archivos clave

• add_node 签名:662-676 — recibe node / action / defer / metadata / input_schema / retry_policy / cache_policy / error_handler / destinations / timeout.
• node 名推断:768-790 — string se usa tal cual; función usa __name__; Runnable usa get_name(); las palabras reservadas START / END / NS_SEP / NS_END dan error.
• 签名类型推断:803-848 — deduce inferred_input_schema y los destinations Command[Literal[...]] del tipo de retorno a partir de los type hints de __call__.
• error_handler 注入:856-870 — genera el nodo independiente __error_handler__{node} y lo guarda en self.nodes con el flag is_error_handler=True.
• 存 StateNodeSpec:872-907 — guarda el spec en self.nodes según tres niveles: input_schema / inferred_input_schema / self.state_schema.
• add_edge:915-967 — origen único entra en el set edges; varios orígenes entran en waiting_edges; el origen no puede ser END ni el destino START.
• add_conditional_edges:969-1017 — envuelve path en un Runnable, BranchSpec.from_path calcula ends y lo guarda en self.branches[source][name].
• BranchSpec.from_path:83-120 — maneja tres formas de path_map: dict / list / deducción desde el tipo de retorno Literal si no se pasa; también deduce input_schema.
• add_sequence:1019-1044 — registra de una sola vez una secuencia de nodos (azúcar sintáctico; por dentro es un bucle de add_node).
• attach_edge:1537-1561 — en compilación: el borde de origen único añade un ChannelWrite al writer del nodo origen apuntando al canal de destino; el de múltiples orígenes registra un canal NamedBarrierValue como join.
• attach_branch:1563-1596 — en compilación: traduce BranchSpec a ChannelWrite.register_writer; según ends, decide a qué canal de destino escribir.

Flujo de datos

add_node realmente solo hace dos cosas en su trabajo útil — deducir el input schema y los destinations del tipo de retorno, y guardar el spec en 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

(类型推断:806-846)

Finalmente, según el resultado de la deducción, se guarda en el dict self.nodes en uno de tres niveles — input_schema explícito / inferred_input_schema deducido / fallback a 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 sí es muy fino — el núcleo es envolver path en un Runnable y que BranchSpec.from_path calcule el 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

(add_conditional_edges 实现:1005-1017)

Límites y fallos

• Nombre de nodo duplicado da error directo(重名:792-793) — Node \` already present`; no se permite sobreescribir.
• Palabras reservadas como nombre de nodo(reserved:794-801) — START / END / NS_SEP (|) / NS_END (:) no se pueden usar; los dos últimos romperían la concatenación de checkpoint_ns.
• add_edge con origen END da error(END 不能当起点:939-940); análogamente, START no puede ser destino(START 不能当终点:941-942) — estas dos constantes son la entrada y salida del grafo y no se pueden invertir.
• Los múltiples orígenes exigen que cada uno esté en add_node(多入边校验:956-964) — si no, en compilación attach_edge no encontraría el writers del origen.
• Misma rama con mismo nombre en un nodo da error(branch 重名:1009-1012); pero varias ramas con nombres distintos pueden coexistir — es la base del enrutamiento mixto conditional + deterministic.
• add_node sobre un grafo ya compilado solo hace warning, no error(compiled warning:778-782); los cambios no se reflejan en la instancia ya compilada — trampa típica de orden.

Resumen

add_node / add_edge / add_conditional_edges son la interfaz de escritura de StateGraph; no son complejos en sí, lo complejo es la «magia» de deducir schema y destinations a partir de la signatura de la función. Entendidos estos tres métodos, básicamente se ha leído toda la API mutable de StateGraph; el siguiente paso es compile y cómo traduce esos spec a PregelNode; el modelo de ejecución en Motor Pregel.

Véase la documentación oficial: documentación de LangGraph · README