@entrypoint / @task: Funktionale API

源码版本1.2.9

Verantwortung

@entrypoint und @task sind die beiden Grundpfeiler der funktionalen API von LangGraph, definiert in libs/langgraph/langgraph/func/__init__.py. @entrypoint verpackt eine gewöhnliche Python-Funktion (synchron oder asynchron) in eine Pregel-Instanz — der Rückgabewert ist ein kompilierter Graph, den man direkt per .invoke / .astream aufrufen kann (entrypoint.__call__:516). @task verpackt eine Funktion in ein _TaskFunction (_TaskFunction:59); beim Aufruf wird ein SyncAsyncFuture zurückgegeben, und dieser Aufruf ist nur innerhalb einer @entrypoint-Funktion oder eines StateGraph-Knotens erlaubt.

Die funktionale API liegt oberhalb von StateGraph und ist ein weiterer, höherer Einstieg — ohne erst ein StateGraph zu bauen und add_node / add_edge / compile aufzurufen, schreibst du einfach eine Funktion. Darunter läuft weiterhin Pregel: entrypoint.__call__ gibt am Ende return Pregel(...) zurück, wobei die entrypoint-Funktion in einen Ein-Knoten-Graphen verpackt wird (func.__name__ ist der einzige Knotenname, der Trigger ist START). Im Knoten wird der Rumpf der entrypoint-Funktion ausgeführt, und beim Rückgeben der Funktion wird der Rückgabewert in den END-Kanal geschrieben (Pregel construction:572).

Entwurfsmotivation

• Eine Ebene Boilerplate weniger: StateGraph verlangt explizite Deklaration des State-Schemas, der Knotenfunktionen, der Kanten und der bedingten Kanten; @entrypoint reicht eine einzige Funktion, deren Task-Aufrufe im Rumpf die Graphstruktur bilden. Für einfache Workflows (ein paar Aufrufe, ein paar Tools aneinandergereiht) ist die funktionale API leichter.
• Volle Pregel-Fähigkeit bleibt erhalten: Die funktionale API ist keine abgespeckte Variante — darunter läuft Pregel, deshalb funktionieren checkpointer, store, cache, retry_policy, context_schema, interrupt() und Command(resume=...) genauso. Ein interrupt() in der entrypoint-Funktion geht denselben Mechanismus wie in einem StateGraph-Knoten.
• entrypoint.final entkoppelt Rückgabewert und Persistenzwert: Viele konversationelle Workflows brauchen «diesen Aufruf dem Aufrufer X zurückgeben, aber im Checkpoint für den nächsten Aufruf Y speichern» (der previous-Parameter liest den letzten save). entrypoint.final(value=X, save=Y) (entrypoint.final:477) trennt diese beiden Dinge explizit, statt sich auf eine dict-Konvention zu verlassen.
• Modell des previous-Parameters: Die Signatur der entrypoint-Funktion kann *, previous: Any = None enthalten; beim nächsten Aufruf mit demselben thread_id ist previous der beim letzten save gesetzte Wert — eine Art «aufrufübergreifende Gedächtnisvariable», ohne dass man selbst ein State-Schema mit reducer deklarieren muss.
• Task-Aufruf ist Graphkante: Im Rumpf der entrypoint-Funktion gibt task_a(...) ein future zurück; .result() bzw. await holt das Ergebnis — dieser Aufruf ist ganz natürlich eine Kante. Pregel schickt das als PUSH-Task in den TASKS-Kanal, und prepare_next_tasks wird ihn im nächsten Schritt schedulen. Ein add_edge ist nicht nötig.

Schlüsseldateien

• _TaskFunction:59 — Produkt des @task-Decorators; hält func / retry_policy / cache_policy / timeout.
• _TaskFunction.__call__:86 — Beim Task-Aufruf wird _call_with_options durchlaufen, aus dem config wird der CONFIG_KEY_CALL-Callback geholt und der Task dem aktuellen runner übergeben.
• task decorator:110 — Einstieg des @task-Decorators, unterstützt die Formen @task und @task(retry_policy=...).
• entrypoint class:262 — Decorator-Klasse für @entrypoint; final ist als eingebettete dataclass an die Klasse gehängt.
• entrypoint.__init__:437 — Nimmt checkpointer / store / cache / retry_policy / timeout / context_schema und speichert sie auf self.
• entrypoint.final:477 — final(value=R, save=S) Datenklasse; value geht an den Aufrufer, save in den Checkpoint.
• entrypoint.__call__:516 — Verwandelt die Funktion in eine Pregel-Instanz; die eigentliche Umwandlungslogik liegt hier.
• Pregel construction:572 — Konstruiert Pregel(nodes={func.__name__: PregelNode(bound=bound, triggers=[START], ...)}); die Kanäle sind START / END / PREVIOUS, jeweils LastValue.
• get_runnable_for_entrypoint:175 — Verpackt die entrypoint-Funktion in ein RunnableCallable; die synchrone Variante läuft über run_in_executor.
• get_runnable_for_task:200 — Verpackt die Task-Funktion in ein RunnableSeq(run, ChannelWrite([RETURN])); der Task-Rückgabewert wird in den RETURN-Kanal geschrieben.

Datenfluss

Der Kern, wie @entrypoint eine Funktion in ein Pregel verwandelt, ist der Aufbau eines minimalen Graphen mit nur drei Kanälen:

graph: Pregel[Any, ContextT, Any, Any] = Pregel(
    nodes={
        func.__name__: PregelNode(
            bound=bound,
            triggers=[START],
            channels=START,
            timeout=self.timeout,
            writers=[
                ChannelWrite(
                    [
                        ChannelWriteEntry(END, mapper=_pluck_return_value),
                        ChannelWriteEntry(PREVIOUS, mapper=_pluck_save_value),
                    ]
                )
            ],
        )
    },
    channels={
        START: EphemeralValue(input_type),
        END: LastValue(output_type, END),
        PREVIOUS: LastValue(save_type, PREVIOUS),
    },
    input_channels=START,
    output_channels=END,
    stream_channels=END,
    stream_mode=stream_mode,
    stream_eager=True,
    checkpointer=self.checkpointer,
    store=self.store,
    cache=self.cache,
    # ...
)

START ist ein EphemeralValue (wird zu jedem Schrittanfang zurückgesetzt), END und PREVIOUS sind LastValue (behalten nur den neuesten Wert). Der Rückgabewert der entrypoint-Funktion wird von zwei Mappern aufgetrennt: _pluck_return_value schreibt entrypoint.final.value (oder den nackten Wert) in END, damit der Aufrufer ihn erhält; _pluck_save_value schreibt entrypoint.final.save (oder den nackten Wert) in den PREVIOUS-Kanal, der beim nächsten Aufruf als previous dient (pluck mappers:547). stream_mode="updates" ist der Default der funktionalen API (nicht values), weil eine entrypoint intern mehrere Tasks enthalten kann und updates Zwischenfortschritte besser wiedergibt.

Task-Aufrufkette: task_fn(arg) → _TaskFunction.__call__ → _call_with_options (_call_with_options:276) → aus config[CONF][CONFIG_KEY_CALL] den impl-Callback holen → impl(func, args, kwargs, retry_policy=..., callbacks=..., timeout=...) gibt ein SyncAsyncFuture zurück. Dieses impl wird von PregelRunner.tick / atick beim Schedulen über partial(...) injiziert (CONFIG_KEY_CALL inject:211) — der Task läuft also nicht selbst, sondern übergibt Funktion und Argumente an den aktuellen runner; der runner entscheidet, wann er gleichzeitig ausgeführt wird, und liefert das Ergebnis nach Abschluss über das future zurück.

Grenzen und Fehler

• Keine Generatoren unterstützt: Die entrypoint-Funktion darf nicht def gen / async def gen (mit yield) sein; __call__ wirft am Anfang explizit NotImplementedError (no generators:525). Für Streaming-Ausgabe verwende StreamWriter (stream_mode="custom") statt yield.
• Task kann nicht von außen aufgerufen werden: Ein Aufruf task_fn(...) außerhalb eines @entrypoint- / StateGraph-Knotens schlägt fehl, weil get_config()[CONF][CONFIG_KEY_CALL] nicht existiert. Ein Task lebt nur im Kontext des aktuellen runner.
• Sync-Task kann kein timeout: task(timeout=...) gilt nur für asynchrone Funktionen; synchrone Funktionen geben sync_timeout_unsupported (sync timeout unsupported:237) — ein synchroner Task läuft in einem Executor-Thread, und Python kann innerhalb eines Prozesses nicht sicher abbrechen.
• previous ist beim ersten Mal None: Beim ersten Aufruf ist previous None, die entrypoint-Funktion muss also den Zweig «kein vorheriger Wert» selbst behandeln. Wird entrypoint.final nicht verwendet, wird der Rückgabewert sowohl in END als auch in PREVIOUS geschrieben, sodass previous beim nächsten Mal der vorherige Rückgabewert ist.
• final muss korrekt parametrisiert werden: Die Rückgabetyp-Annotation -> entrypoint.final[int, str] benötigt beide Typparameter; nur einer oder keiner führt in __call__ zu einem TypeError (final param check:562).
• Abhängig vom checkpointer: Ohne entrypoint(checkpointer=...) sind weder interrupt() noch das aufrufübergreifende Gedächtnis von previous nutzbar — ersteres wirft eine RuntimeError, letzteres ist jedes Mal None.

Zusammenfassung

Die funktionale API ist syntaktischer Zucker über StateGraph; darunter steht ein Pregel-Ein-Knoten-Graph plus TASKS-Kanal plus ein vom runner injiziertes CONFIG_KEY_CALL. Die Folge der task(...)-Aufrufe im entrypoint-Rumpf ist die implizite Graphstruktur. Nebenläufigkeitsdetails stehen in /func/concurrency, Unterbrechung/Wiederherstellung in /interrupt/interrupt. Siehe offizielle Dokumentation: LangGraph 文档 · README。