为LangGraph@entrypoint函数切换测试Checkpointer的正确方法

针对你使用LangGraph函数式API（@entrypoint）时无法在运行时切换生产/测试Checkpointer的问题，结合你提到的环境（langgraph==0.2.0），给出以下规范解决方案：

一、推荐方案：依赖注入式构建器函数 #

这是最可靠且符合测试隔离原则的方式，核心是将Checkpointer作为参数传入构建逻辑，避免装饰阶段绑定固定实例：

实现步骤： #

1. 定义工作流核心逻辑：先写不带@entrypoint装饰的基础函数，专注业务逻辑：

from langgraph.graph import StateGraph
from langgraph.checkpoint import BaseCheckpointSaver
from typing import TypedDict

def build_workflow(checkpointer: BaseCheckpointSaver | None = None):
    # 定义状态结构
    class State(TypedDict):
        input: str
        output: str

    # 定义节点函数
    def process_node(state: State) -> State:
        return {"output": f"Processed: {state['input']}"}

    # 构建图并设置entrypoint
    graph = StateGraph(State)
    graph.add_node("process", process_node)
    graph.set_entry_point("process")
    graph.set_finish_point("process")

    # 返回编译后的图，传入指定checkpointer
    return graph.compile(checkpointer=checkpointer)

2. 生产环境初始化：

from langgraph_checkpoint_postgres import PostgresSaver

# 初始化Postgres Checkpointer
postgres_checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@host:port/db")
production_workflow = build_workflow(checkpointer=postgres_checkpointer)

3. 单元测试初始化：

from langgraph.checkpoint.memory import InMemorySaver

def test_workflow():
    # 每个测试用例创建独立的InMemorySaver实例，避免并行测试冲突
    test_checkpointer = InMemorySaver()
    test_workflow = build_workflow(checkpointer=test_checkpointer)
    
    # 执行测试逻辑
    result = test_workflow.invoke({"input": "test data"})
    assert result["output"] == "Processed: test data"

之前构建器方案出错的原因： #

大概率是你在构建时没有正确传递checkpointer，或者复用了全局的checkpointer实例导致测试隔离失效。确保每个测试用例都创建独立的InMemorySaver实例，不要共享全局对象。

二、替代方案：动态修改已编译图的Checkpointer #

如果你不想重构现有代码，可以通过访问已编译图的内部属性来替换Checkpointer（注意：这依赖langgraph内部实现，版本升级时需验证）：

def switch_checkpointer(workflow, new_checkpointer):
    # 针对langgraph==0.2.0，已编译的entrypoint图内部存储在.wrapped属性中
    if hasattr(workflow, "wrapped"):
        workflow.wrapped.checkpointer = new_checkpointer
    return workflow

# 测试时使用
test_checkpointer = InMemorySaver()
test_workflow = switch_checkpointer(production_workflow, test_checkpointer)

注意事项： #

• 这种方式避免了copy()方法的问题（因为RunnableCallable确实没有copy属性）
• 仅在无法重构现有代码时使用，长期来看推荐用构建器方案

三、避坑指南 #

• 禁止全局Checkpointer实例：全局变量会导致并行测试时状态污染，必须每个测试用例创建独立实例
• 验证Checkpointer生效：测试时可以通过workflow.wrapped.checkpointer属性确认是否正确设置
• 版本兼容：langgraph 0.2.x的函数式API编译后结构稳定，但升级到0.3+时需重新验证内部属性路径

内容的提问来源于stack exchange，提问作者Джон Сноу