测试

编写测试是保证 Agent 质量的关键步骤。本文介绍 LangGraph 特有的测试模式。

安装测试框架

pip install -U pytest

基础测试

测试完整执行

import pytest
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

def create_graph() -> StateGraph:
    class State(TypedDict):
        value: str

    graph = StateGraph(State)
    graph.add_node("step1", lambda s: {"value": "hello from step1"})
    graph.add_node("step2", lambda s: {"value": "hello from step2"})
    graph.add_edge(START, "step1")
    graph.add_edge("step1", "step2")
    graph.add_edge("step2", END)
    return graph

def test_full_execution():
    checkpointer = MemorySaver()
    graph = create_graph().compile(checkpointer=checkpointer)
    result = graph.invoke(
        {"value": ""},
        config={"configurable": {"thread_id": "1"}}
    )
    assert result["value"] == "hello from step2"

测试单个节点

def test_individual_node():
    graph = create_graph().compile(checkpointer=MemorySaver())
    # 直接调用某个节点，绕过 checkpointer
    result = graph.nodes["step1"].invoke({"value": ""})
    assert result["value"] == "hello from step1"

部分执行测试

测试图中某一段执行路径，而不是全流程：

def create_large_graph() -> StateGraph:
    class State(TypedDict):
        value: str

    graph = StateGraph(State)
    graph.add_node("node1", lambda s: {"value": "n1"})
    graph.add_node("node2", lambda s: {"value": "n2"})
    graph.add_node("node3", lambda s: {"value": "n3"})
    graph.add_node("node4", lambda s: {"value": "n4"})
    graph.add_edge(START, "node1")
    graph.add_edge("node1", "node2")
    graph.add_edge("node2", "node3")
    graph.add_edge("node3", "node4")
    graph.add_edge("node4", END)
    return graph

def test_partial_execution():
    graph = create_large_graph().compile(checkpointer=MemorySaver())

    # 模拟 node1 执行后的状态，从 node2 开始
    graph.update_state(
        config={"configurable": {"thread_id": "1"}},
        values={"value": "initial_value"},
        as_node="node1",  # 伪装成 node1 已执行
    )

    # 只执行 node2 → node3
    result = graph.invoke(
        None,
        config={"configurable": {"thread_id": "1"}},
        interrupt_after="node3",  # node3 执行完就停
    )
    assert result["value"] == "n3"

测试条件路由

def test_conditional_routing():
    """测试路由逻辑是否按预期工作"""
    from typing import Literal

    def route(state) -> Literal["a", "b"]:
        return "a" if state["x"] > 0 else "b"

    assert route({"x": 5}) == "a"
    assert route({"x": 0}) == "b"
    assert route({"x": -1}) == "b"

测试带 Interrupt 的图

from langgraph.types import Command, interrupt

def test_human_in_the_loop():
    from langgraph.graph import StateGraph, START, END
    from langgraph.checkpoint.memory import MemorySaver

    def ask(state):
        answer = interrupt("确认?")
        return {"confirmed": answer}

    graph = (
        StateGraph(State)
        .add_node("ask", ask)
        .add_edge(START, "ask")
        .add_edge("ask", END)
        .compile(checkpointer=MemorySaver())
    )

    config = {"configurable": {"thread_id": "test"}}

    # 第一次调用会停下
    graph.invoke({"confirmed": False}, config)

    # 恢复执行
    result = graph.invoke(Command(resume=True), config)
    assert result["confirmed"] is True

Mock LLM 调用

def test_with_mock_llm(monkeypatch):
    """使用 mock 避免真实 LLM 调用"""

    def mock_llm_invoke(messages):
        return AIMessage(content="mock response")

    monkeypatch.setattr("module.llm.invoke", mock_llm_invoke)
    result = agent.invoke({"messages": [HumanMessage(content="测试")]})
    assert "mock" in result["messages"][-1].content

最佳实践

1. 每个测试用新的 checkpointer，避免状态污染
2. 测试正常路径和异常路径（API 失败、超时）
3. Mock 外部依赖（LLM、API），保证测试速度和确定性
4. 测试条件边的路由函数，它们是 Agent 行为的核心
5. 使用 fixture 复用图构建代码

参考

• 官方文档：Test
• pytest 文档