LangGraph源码学习路线：第 18 篇 Functional API、entrypoint 与 task源码解剖#

1
StateGraph / Checkpoint / Multi-Agent
2
  ↓
3
Functional API / task runtime
4
  ↓
5
Time Travel / Debug Replay
6
  ↓
7
工程化 Agent 能力

1
from langgraph.func import entrypoint, task
2
from langgraph.checkpoint.memory import MemorySaver
3

4
checkpointer = MemorySaver()
5

6
@task
7
def search_city(query: str) -> list[dict]:
8
    return [{"title": "Osaka food guide", "city": "Osaka"}]
9

10
@task
11
def write_plan(docs: list[dict]) -> str:
12
    return f"Plan with {len(docs)} verified docs"
13

14
@entrypoint(checkpointer=checkpointer)
15
def travel_workflow(user_request: str) -> str:
16
    docs_future = search_city(f"travel evidence: {user_request}")
17
    docs = docs_future.result()
18
    answer_future = write_plan(docs)
19
    return answer_future.result()
20

21
config = {"configurable": {"thread_id": "trip-001"}}
22
result = travel_workflow.invoke("大阪 4 天美食行程", config=config)

1
def invoke_entrypoint(wrapper, input_value, config):
2
    runtime_config = ensure_config(config)
3
    thread_id = runtime_config["configurable"].get("thread_id")
4
    checkpoint_id = runtime_config["configurable"].get("checkpoint_id")
5

6
    previous = None
7
    if wrapper.checkpointer is not None:
8
        previous = wrapper.checkpointer.get_tuple(runtime_config)
9

10
    context = FunctionalRuntimeContext(
11
        entrypoint_name=wrapper.name,
12
        thread_id=thread_id,
13
        checkpoint_id=checkpoint_id,
14
        checkpointer=wrapper.checkpointer,
15
        previous_checkpoint=previous,
16
    )
17

18
    with context.as_current():
19
        try:
20
            output = wrapper.func(input_value)
21
        except BaseException as error:
22
            context.record_error(error)
23
            raise
24

25
    if wrapper.checkpointer is not None:
26
        checkpoint = context.to_checkpoint(final_output=output)
27
        wrapper.checkpointer.put(runtime_config, checkpoint, context.metadata)
28

29
    return output

1
def call_task(task_wrapper, *args, **kwargs):
2
    context = FunctionalRuntimeContext.current()
3
    if context is None:
4
        return task_wrapper.func(*args, **kwargs)
5

6
    task_key = context.make_task_key(
7
        name=task_wrapper.name,
8
        args=args,
9
        kwargs=kwargs,
10
    )
11

12
    cached = context.lookup_task_result(task_key)
13
    if cached is not MISSING:
14
        return TaskFuture(task_key=task_key, status="cached", value=cached)
15

16
    call = TaskCall(task_key, task_wrapper.func, args, kwargs)
17
    context.schedule_task(call)
18
    return TaskFuture(task_key=task_key, status="pending")
19

20

21
def resolve_task_result(future):
22
    context = FunctionalRuntimeContext.current()
23
    if future.status == "cached":
24
        return future.value
25

26
    call = context.get_scheduled_task(future.task_key)
27
    try:
28
        value = call.func(*call.args, **call.kwargs)
29
    except BaseException as error:
30
        context.record_task_error(future.task_key, error)
31
        raise
32

33
    context.record_task_result(future.task_key, value)
34
    return value

1
langgraph/
2
├── graph/
3
├── pregel/
4
├── types.py
5
└── func/
6

7
langchain_core/
8
├── runnables/
9
├── tools/
10
├── retrievers.py
11
└── messages/

1
1. 先读官方概念文档，确认公共契约。
2
2. 再读公开 API 或装饰器入口。
3
3. 顺着构建期对象进入 runtime。
4
4. 追踪一次 invoke / stream / tool call / task call。
5
5. 单独检查异常、重试、恢复和扩展点。
6
6. 最后回到旅行规划助手做工程判断。

1
Public API / Decorator / Tool Protocol
2
  ↓
3
Runtime wrapper
4
  ↓
5
State / Config / Context
6
  ↓
7
Storage / Model / Tool / Graph Runtime

1
公共 API
2
  负责：给业务代码稳定入口。
3
  不负责：暴露所有内部调度细节。
4

5
运行时协议
6
  负责：让状态、配置、工具、模型和持久化协作。
7
  不负责：替业务判断所有策略。
8

9
扩展接口
10
  负责：允许业务插入可维护的定制逻辑。
11
  不负责：保证错误扩展仍然安全。

1
先找公开入口
2
  ↓
3
确认输入输出类型
4
  ↓
5
沿调用方追到核心实现
6
  ↓
7
记录状态与对象变化
8
  ↓
9
检查分支、异常和结束条件
10
  ↓
11
回到设计目的

1
def build_entrypoint(func=None, *, checkpointer=None, name=None):
2
    def decorate(user_func):
3
        workflow_name = name or user_func.__name__
4
        wrapper = EntrypointWrapper(
5
            name=workflow_name,
6
            func=user_func,
7
            checkpointer=checkpointer,
8
        )
9
        wrapper.invoke = lambda input_value, config=None: invoke_entrypoint(
10
            wrapper, input_value, config
11
        )
12
        wrapper.stream = lambda input_value, config=None, **kwargs: stream_entrypoint(
13
            wrapper, input_value, config, **kwargs
14
        )
15
        return wrapper
16

17
    if func is None:
18
        return decorate
19
    return decorate(func)

1
def build_task(func=None, *, name=None):
2
    def decorate(user_func):
3
        task_name = name or user_func.__name__
4
        wrapper = TaskWrapper(name=task_name, func=user_func)
5

6
        def task_call(*args, **kwargs):
7
            context = FunctionalRuntimeContext.current()
8
            if context is None:
9
                return user_func(*args, **kwargs)
10

11
            task_key = context.make_task_key(task_name, args, kwargs)
12
            return context.create_future(
13
                task_key=task_key,
14
                task_name=task_name,
15
                func=user_func,
16
                args=args,
17
                kwargs=kwargs,
18
            )
19

20
        wrapper.__call__ = task_call
21
        return wrapper
22

23
    if func is None:
24
        return decorate
25
    return decorate(func)

1
def prepare_checkpoint_context(wrapper, config):
2
    runtime_config = ensure_config(config)
3
    configurable = runtime_config.setdefault("configurable", {})
4

5
    thread_id = configurable.get("thread_id")
6
    if wrapper.checkpointer is not None and not thread_id:
7
        raise ValueError("checkpointed entrypoint requires configurable.thread_id")
8

9
    checkpoint_tuple = None
10
    if wrapper.checkpointer is not None:
11
        checkpoint_tuple = wrapper.checkpointer.get_tuple(runtime_config)
12

13
    return FunctionalRuntimeContext(
14
        entrypoint_name=wrapper.name,
15
        config=runtime_config,
16
        checkpointer=wrapper.checkpointer,
17
        previous_checkpoint=checkpoint_tuple,
18
    )

1
def run_functional_workflow(wrapper, input_value, config):
2
    context = prepare_checkpoint_context(wrapper, config)
3

4
    with context.as_current():
5
        context.emit_event("entrypoint_start", input=input_value)
6
        output = wrapper.func(input_value)
7
        context.emit_event("entrypoint_end", output=output)
8

9
    checkpoint = context.build_checkpoint(
10
        values={"__entrypoint_output__": output},
11
        task_results=context.task_results,
12
        pending_tasks=context.pending_tasks,
13
    )
14

15
    if context.checkpointer is not None:
16
        context.checkpointer.put(context.config, checkpoint, metadata=context.metadata)
17

18
    return output

1
def task_future_result(future):
2
    context = FunctionalRuntimeContext.current()
3
    task_key = future.task_key
4

5
    if context.has_checkpointed_result(task_key):
6
        value = context.get_checkpointed_result(task_key)
7
        context.emit_event("task_result_reused", task_key=task_key)
8
        return value
9

10
    if context.has_in_memory_result(task_key):
11
        return context.task_results[task_key]
12

13
    call = context.scheduled_tasks[task_key]
14
    context.emit_event("task_start", task_key=task_key)
15
    value = call.func(*call.args, **call.kwargs)
16
    context.task_results[task_key] = value
17
    context.emit_event("task_end", task_key=task_key, result=value)
18
    return value

1
def resume_from_checkpoint(wrapper, input_value, config):
2
    context = prepare_checkpoint_context(wrapper, config)
3

4
    for saved_task in context.previous_task_results:
5
        context.task_results[saved_task.key] = saved_task.value
6

7
    with context.as_current():
8
        output = wrapper.func(input_value)
9

10
    context.write_new_checkpoint(output)
11
    return output

1
def call_task_or_plain_function(task_wrapper, *args, **kwargs):
2
    context = FunctionalRuntimeContext.current()
3
    if context is None:
4
        return task_wrapper.func(*args, **kwargs)
5

6
    return context.create_future(
7
        task_name=task_wrapper.name,
8
        func=task_wrapper.func,
9
        args=args,
10
        kwargs=kwargs,
11
    )

1
def validate_checkpoint_config(checkpointer, config):
2
    if checkpointer is None:
3
        return "stateless"
4

5
    configurable = ensure_config(config).get("configurable", {})
6
    thread_id = configurable.get("thread_id")
7
    if not thread_id:
8
        raise ValueError("checkpointed workflow requires thread_id")
9

10
    return "durable"

1
def execute_task_call(context, call):
2
    try:
3
        value = call.func(*call.args, **call.kwargs)
4
    except BaseException as error:
5
        context.record_task_error(call.key, error)
6
        context.emit_event("task_failed", task_key=call.key, error=repr(error))
7
        raise
8

9
    context.record_task_result(call.key, value)
10
    return value

1
def should_wrap_as_task(operation):
2
    if operation.has_external_side_effect and not operation.is_idempotent:
3
        return "wrap_and_add_business_idempotency_key"
4
    if operation.is_expensive_or_slow:
5
        return "wrap_as_task"
6
    if operation.is_pure_local_calculation:
7
        return "plain_python_is_ok"

1
@entrypoint(checkpointer=checkpointer)
2
def answer_trip(request: str) -> str:
3
    parsed = parse_request(request).result()
4

5
    if parsed["needs_search"]:
6
        docs = search_docs(parsed["query"]).result()
7
    else:
8
        docs = []
9

10
    draft = write_draft(parsed, docs).result()
11
    reviewed = review_plan(draft).result()
12
    return reviewed

1
@task
2
def run_planner_graph(planner_input: dict) -> dict:
3
    return planner_graph.invoke(planner_input)
4

5
@entrypoint(checkpointer=checkpointer)
6
def travel_agent(request: str) -> str:
7
    intent = parse_request(request).result()
8
    plan_state = run_planner_graph(intent).result()
9
    return render_answer(plan_state).result()

1
def debug_functional_run(workflow, thread_id):
2
    config = {"configurable": {"thread_id": thread_id}}
3
    history = list(workflow.get_state_history(config))
4

5
    bad_snapshot = choose_snapshot(history, where="after_search")
6
    fork_config = workflow.update_state(
7
        bad_snapshot.config,
8
        {"search_result_override": [{"title": "corrected evidence"}]},
9
    )
10

11
    return workflow.invoke(None, config=fork_config)

1
性能：额外抽象会增加序列化、检索、trace 和存储成本。
2
可靠性：可恢复机制要求状态 schema、幂等工具和错误分类清楚。
3
安全：metadata、state、tool args、retrieved docs 都可能包含敏感信息。
4
可观测性：至少记录 thread_id、node/tool、策略命中、错误和耗时。

1
公开 API 通常只负责归一化和装配；真正的运行语义在 runtime、协议对象和扩展点之间流转。

1
state、store、Document、metadata、trace、prompt messages 是不同协议；只有经过明确注入的内容才进入模型上下文。

1
可恢复异常可以转成 retry、fallback、interrupt；不可恢复异常必须上抛并进入 trace。

1
Python callable
2
  ↓
3
@entrypoint / @task wrapper
4
  ↓
5
entrypoint context + task key
6
  ↓
7
checkpointer / stream / invoke 能力

1
workflow.invoke(input, config)
2
  ↓
3
读取 thread checkpoint
4
  ↓
5
执行 entrypoint Python 控制流
6
  ↓
7
task(...) 返回 future
8
  ↓
9
future.result() 执行或复用 task result
10
  ↓
11
写入 checkpoint 并返回 final output

1
task 在 entrypoint 外 → 普通函数语义
2
task 在 entrypoint 内 → future/result + checkpoint 语义
3
已有 task result → 复用 checkpoint，避免重复副作用
4
task 执行失败 → 记录 task error 并继续抛出
5
外部副作用 → 需要业务幂等键，checkpoint 不负责撤销

1
第 19 篇：Time Travel、update_state 与 Debug Replay 源码解剖

1
这项能力在旅行规划助手中应该放在哪一层？
2
它和前一篇能力如何协作？
3
哪些边界需要通过源码证据确认？