PAYATHON 2026

如何测试 LangGraph Functional API 的 entrypoint 并注入 checkpointer？

技术老齐

建议采用第一种方案：通过工厂函数接收 checkpointer，每个测试分别创建新的 InMemorySaver 和 entrypoint。这个建议基于官方的测试隔离思路和 Functional API 的构造接口；已核对的官方文档并未将三种写法中的任何一种指定为 Functional API 测试的唯一标准。

官方资料支持什么

官方测试指南建议在每个测试中创建 graph，并使用新的 checkpointer 实例编译，以隔离测试状态。示例使用的是 StateGraph。

Functional API 文档说明，@entrypoint(checkpointer=...) 会把函数构造成可执行的 workflow 对象，通过 invoke 或 ainvoke 调用。使用 checkpointing 时，需要在调用配置中提供 thread_id。

结合这两点，可以用工厂函数在构造 entrypoint 时显式注入测试依赖。这是基于官方接口对 Functional API 的工程建议。

推荐实现：工厂函数配合 pytest fixture

下面用一个累加 workflow 演示依赖注入和状态测试。previous 用于读取同一线程上一次保存的返回值，具体行为见官方短期记忆说明。

# workflows.py
from langgraph.func import entrypoint


def wf1_builder(checkpointer):
    @entrypoint(checkpointer=checkpointer)
    async def wf1(value: int, *, previous: int | None = None) -> int:
        return (previous or 0) + value

    return wf1

下面的测试使用 Python 3.10+ 的类型注解语法，假设已安装 pytest 和 pytest-asyncio：

# test_workflows.py
import pytest
from langgraph.checkpoint.memory import InMemorySaver

from workflows import wf1_builder


@pytest.fixture
def workflow():
    return wf1_builder(InMemorySaver())


@pytest.mark.asyncio
async def test_wf1_returns_result(workflow):
    config = {"configurable": {"thread_id": "test-thread"}}

    result = await workflow.ainvoke(2, config=config)

    assert result == 2


@pytest.mark.asyncio
async def test_wf1_persists_state_per_thread(workflow):
    config = {"configurable": {"thread_id": "test-thread"}}

    first = await workflow.ainvoke(2, config=config)
    second = await workflow.ainvoke(3, config=config)
    other = await workflow.ainvoke(
        3,
        config={"configurable": {"thread_id": "other-thread"}},
    )

    assert first == 2
    assert second == 5
    assert other == 3

fixture 默认按测试函数创建实例，因此两个测试各自拥有独立的 workflow 和 saver。第二个测试验证了在同一个 saver 中，相同 thread_id 的状态可以延续，不同 thread_id 的状态相互隔离。

原问题的调用片段省略了 config。测试启用了 checkpointer 的 entrypoint 时，应补上 config={"configurable": {"thread_id": "..."}}。

copy 能否用来替换 checkpointer

可以考虑，不过要核对项目锁定版本的实现。官方仓库中的 Pregel.copy 源码会收集当前对象的属性，用 update 中的值覆盖，再构造新对象。因此，在对应实现下，原问题中的写法有源码依据：

wf1_test = wf1.copy({"checkpointer": InMemorySaver()})

这种方式不会递归复制所有依赖。替换 checkpointer 后，模型客户端、闭包或其他可变对象仍可能被共享，不能据此认定整个 workflow 已完全隔离。

如果模块导入时已经执行 build_postgres_saver()，后续调用 copy 无法撤销这次初始化。该函数是否会立即连接数据库，取决于它的具体实现。

对于已有的顶层 entrypoint，可以用 copy 替换 checkpointer；对于新代码，工厂函数更容易看清依赖从哪里传入，也便于同时替换其他测试依赖。

根据环境选择全局 checkpointer 有什么问题

这种方式可以运行，但模块顶层的环境判断通常在导入时执行。测试中再修改 ENV，不会自动重建已经装饰好的 entrypoint。如果多个测试共用一个全局 InMemorySaver，还需要处理状态残留和线程标识冲突。

更容易管理的方式是在应用启动时根据环境创建 saver，再传给同一个工厂函数；测试则直接传入自己的 saver。环境配置负责选择后端，workflow 的构造过程保持显式。

InMemorySaver 测试能覆盖到哪里

InMemorySaver 适合验证 workflow 的执行结果、线程状态和恢复流程。如果业务依赖中断恢复，还应在同一个测试中保留 saver 和 thread_id，覆盖暂停与恢复的完整过程。

PostgreSQL 后端的连接、初始化、权限和实际存储行为需要单独进行集成测试。生产环境创建的 saver 必须在 workflow 使用期间保持有效；使用 ainvoke 时，应选择支持相应异步操作的 saver。

工厂函数不必负责打开或关闭数据库连接。由应用启动逻辑或测试 fixture 管理资源生命周期，再把可用的 saver 传入即可。

备注：内容仅供参考。