TopolearnSome workflows read better as ordinary functions with loops and ifs than as explicit graphs.
The Functional API gives you LangGraph's runtime features without drawing a graph. Two decorators:
@taskmarks a unit of work (an API call, a model call). Calling a task returns a future, a placeholder for a result that may still be running; call.result()to wait for it. Several tasks started before you wait run in parallel.@entrypoint(checkpointer=...)marks the workflow function. Inside it you use normal Python:if,for,while, function calls.
You run an entrypoint like a graph: .invoke(input, config), .stream(...), with a thread_id when it has a checkpointer.
How persistence works here: task results are saved as they complete. If the workflow is resumed (after an interrupt() or a crash), finished tasks are not re-executed; their saved results are reused. So put side effects and expensive calls inside tasks, and keep the entrypoint body deterministic (same inputs, same path).
Other features:
interrupt()works inside an entrypoint, resumed withCommand(resume=...).- A parameter named
previousreceives the last return value on this thread, giving simple short-term memory. entrypoint.final(value=..., save=...)returns one value to the caller while saving a different one for next time.
Graph API or Functional API? Graphs make complex branching, parallel paths and team collaboration visible. Functions suit linear or loop-heavy logic, or adding durability to existing code. They share one runtime and can call each other.
Example
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import Command, interrupt
@task
def write_section(topic: str) -> str:
return f"A paragraph about {topic}." # real code: a model call
@entrypoint(checkpointer=InMemorySaver())
def write_report(topics: list[str]) -> str:
futures = [write_section(t) for t in topics] # run in parallel
draft = "\n".join(f.result() for f in futures)
verdict = interrupt({"review": draft}) # pause for a human
return draft if verdict == "ok" else "Rejected"
cfg = {"configurable": {"thread_id": "r1"}}
for part in write_report.stream(["bees", "wasps"], cfg):
print(part) # ... {'__interrupt__': (Interrupt(value={'review': ...}),)}
print(write_report.invoke(Command(resume="ok"), cfg))
# tasks are not re-run on resume; their saved results are reusedRemember
- @entrypoint = workflow in plain Python; @task = checkpointed unit of work
- Tasks return futures; .result() waits; start several to run in parallel
- Finished tasks are reused, not re-run, on resume
- Same runtime as the Graph API: interrupts, streaming, persistence
Watch out for
- Doing side effects directly in the entrypoint body instead of in a @task, so they repeat when the workflow resumes.