LangGraph 19：功能式 API

发表于 更新于 分类于
用 @task 与 @entrypoint 以函数式写法组织图，覆盖并行任务、yield 等待、流式输出、interrupt 与 Store。

前言

前面各篇都在手写 StateGraph：显式声明节点、边、条件路由。
对线性、并行这类形状，要敲的样板有点多。
LangGraph 1.x 提供了第二种编程模型——功能式 API：用 @task 标后台步骤、用 @entrypoint 写主流程，yield 一行就是在「等某个任务跑完」。
本文用短示例讲清 @task 与 @entrypoint，再补流式、HITL 与 Store。
HITL 的 interrupt 语义、Checkpoint 与 Store 的机制仍以 《LangGraph 11：HITL 人机协同》《LangGraph 08：Checkpoint 持久化》《LangGraph 10：Store 与 Context》 为准，本篇只演示功能式写法。
示例统一对接 火山方舟 Coding Plan，模型用 ark-code-latest。
下文需要 Python 3.12+，依赖用 uv 管理。

概要

功能式 API 用普通函数描述图：

  1. @task：把函数标记为任务，先启动后等待，天然并行。
  2. @entrypoint：把函数标记为入口，yield 某个任务() 表示等它返回。
  3. 状态仍用 TypedDict + Annotated[..., add_messages] 归并；返回 dict 即更新状态。
  4. checkpointer= / store= 直接写在装饰器上，thread_id 在 invoke 时传。
  5. interrupt / Command(resume=...) 在入口函数里照常能用。

对比 StateGraph：显式节点边适合复杂条件路由与循环；功能式更短，适合线性/并行/简单路由。

依赖

建议使用 Python 3.12 及以上。

1
2
3
4
uv init langgraph-functional
cd langgraph-functional
uv venv --python 3.12
uv add "langgraph>=1.0,<2.0" "langchain>=1.0,<2.0" langchain-openai python-dotenv rich

在项目根目录创建 .env ，写入 Coding Plan 的 Key 与专用 Base URL。

1
2
OPENAI_API_KEY=你的火山方舟 API Key
OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

请勿把 Base URL 写成普通方舟 .../api/v3 ，以免无法抵扣 Coding Plan 额度。

任务

@task 装饰一个普通函数，把它变成可并行调度的任务。
任务里的代码在独立步骤里执行，失败可单独重试，不会拖垮整张图。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langgraph.func import task

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0.7,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)


@task
def call_joke(topic: str) -> str:
    return model.invoke(f"写一个关于「{topic}」的短笑话").content


@task
def call_poem(topic: str) -> str:
    return model.invoke(f"写一首关于「{topic}」的短诗").content

保存为 src/tasks.py。
@task 只负责「这段逻辑可单独调度」，真正跑起来要靠入口函数里的 yield。

入口

@entrypoint 装饰主流程函数，函数签名第一参数是状态。
函数体里 yield 任务调用() 表示等这个任务跑完，yield 的返回值就是任务结果。
返回的 dict 会像节点返回一样按 reducer 归并进状态。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from typing import Annotated, TypedDict

from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint
from langgraph.graph import add_messages
from rich import print as rprint

from src.tasks import call_joke, call_poem


class OverallState(TypedDict):
    topic: str
    messages: Annotated[list, add_messages]


@entrypoint(checkpointer=InMemorySaver())
def creative_workflow(state: OverallState):
    joke = yield call_joke(state["topic"])
    poem = yield call_poem(state["topic"])
    return {"messages": [HumanMessage(content=f"笑话：{joke}\n诗歌：{poem}")]}

保存为 src/workflow.py。
运行方式与 StateGraph 一样：invoke 传初始状态，config 里带 thread_id。

1
2
3
config = {"configurable": {"thread_id": "f-1"}}
result = creative_workflow.invoke({"topic": "橘猫"}, config=config)
rprint(result["messages"][-1].content)

checkpointer=InMemorySaver() 直接写在装饰器上，thread_id 决定会话。
想两个任务同时跑，就先都启动再统一等待：

1
2
3
4
5
6
7
@entrypoint(checkpointer=InMemorySaver())
def parallel_workflow(state: OverallState):
    joke_future = call_joke(state["topic"])
    poem_future = call_poem(state["topic"])
    joke = yield joke_future
    poem = yield poem_future
    return {"messages": [HumanMessage(content=f"笑话：{joke}\n诗歌：{poem}")]}

call_joke 与 call_poem 都提交后才 yield，两者并行执行，总耗时约等于较慢那一个。

流式输出

stream 与 stream_mode 在功能式 API 同样可用。
yield 一个普通值（非任务调用）会作为 custom 流式事件吐给调用方。

1
2
3
4
5
6
7
8
9
10
11
from langgraph.func import entrypoint

from src.tasks import call_poem


@entrypoint(checkpointer=InMemorySaver())
def poem_workflow(state: OverallState):
    yield "开始写诗……"
    poem = yield call_poem(state["topic"])
    yield "诗句完成。"
    return {"messages": [HumanMessage(content=poem)]}

跑法：

1
2
3
config = {"configurable": {"thread_id": "f-2"}}
for chunk in poem_workflow.stream({"topic": "夜雨"}, config=config, stream_mode="custom"):
    print(chunk)

stream_mode="custom" 只收到 yield 出的普通值；
要看每步状态变化用 stream_mode="updates"，与 《LangGraph 13：流式输出》 一致。

HITL 与中断

interrupt 在入口函数里照常使用：图跑到这里暂停，等 Command(resume=...) 续跑。
审批、编辑、重试的语义与 StateGraph 版本完全一致。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from langgraph.types import Command, interrupt

from src.tasks import call_poem


@entrypoint(checkpointer=InMemorySaver())
def approval_workflow(state: OverallState):
    poem = yield call_poem(state["topic"])
    approved = interrupt(
        {
            "instruction": "是否通过这首短诗？",
            "poem": poem,
            "choices": ["是", "否"],
        }
    )
    if approved:
        return {"messages": [HumanMessage(content=f"定稿：{poem}"]}
    return {"messages": [HumanMessage(content="未通过。")]}

先跑到中断，再带着 resume 续跑：

1
2
3
4
5
6
config = {"configurable": {"thread_id": "f-3"}}
first = approval_workflow.invoke({"topic": "晨光"}, config=config)
rprint(first["__interrupt__"])

final = approval_workflow.invoke(Command(resume=True), config=config)
rprint(final["messages"][-1].content)

第一次 invoke 返回带 __interrupt__ 的中间状态，续跑用 Command(resume=True)。
自定义审批面板的接入方式见 《LangGraph 17：自建中断面板》，协议在这里同样适用。

Store 与记忆

store= 写在装饰器上即可获得跨线程长期记忆，用法与 StateGraph 版本一致。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()


@entrypoint(checkpointer=InMemorySaver(), store=store)
def memory_workflow(state: OverallState):
    last = store.get(("memories", "default"), "last_topic")
    previous = last.value["topic"] if last else "（暂无）"
    poem = yield call_poem(state["topic"])
    store.put(("memories", "default"), "last_topic", {"topic": state["topic"]})
    return {
        "messages": [
            HumanMessage(content=f"上一主题：{previous}\n本次诗歌：{poem}")
        ]
    }

store 跨线程、不依赖 thread_id；会话内记忆仍走 checkpointer。
两者分工见 《LangGraph 10：Store 与 Context》。

总结

  1. @task 标记后台步骤，先启动再 yield 等待，天然并行。
  2. @entrypoint 写主流程，yield 任务() 等待，返回 dict 即更新状态。
  3. 状态归并、checkpointer=、store=、thread_id 语义与 StateGraph 一致。
  4. yield 普通值走 stream_mode="custom"，interrupt / Command(resume=...) 原样可用。
  5. 功能式适合线性/并行；复杂条件路由与循环还是 StateGraph 更直观。