测试

在您原型化LangGraph智能体之后，下一步自然就是添加测试。本指南涵盖了您在编写单元测试时可以使用的某些有用模式。

请注意，本指南针对LangGraph特定，涵盖了具有自定义结构的图相关场景——如果您是初学者，请查看本节，它使用LangChain内置的create_agent。

前提条件

首先，请确保您已安装pytest。

$ pip install -U pytest

开始使用

由于许多LangGraph智能体依赖于状态，一个有用的模式是在每次使用它进行测试之前创建你的图，然后在测试中使用新的检查点实例编译它。

以下示例展示了如何使用简单的线性图通过 node1 和 node2 进行操作。每个节点更新单个状态键 my_key：

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 MyState(TypedDict):
        my_key: str

    graph = StateGraph(MyState)
    graph.add_node("node1", lambda state: {"my_key": "hello from node1"})
    graph.add_node("node2", lambda state: {"my_key": "hello from node2"})
    graph.add_edge(START, "node1")
    graph.add_edge("node1", "node2")
    graph.add_edge("node2", END)
    return graph

def test_basic_agent_execution() -> None:
    checkpointer = MemorySaver()
    graph = create_graph()
    compiled_graph = graph.compile(checkpointer=checkpointer)
    result = compiled_graph.invoke(
        {"my_key": "initial_value"},
        config={"configurable": {"thread_id": "1"}}
    )
    assert result["my_key"] == "hello from node2"

测试单个节点和边

编译后的 LangGraph 智能体暴露了每个单独节点的引用作为 graph.nodes。您可以利用这一点来测试智能体中的单个节点。请注意，这将绕过在编译图时传递的任何检查点器：

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 MyState(TypedDict):
        my_key: str

    graph = StateGraph(MyState)
    graph.add_node("node1", lambda state: {"my_key": "hello from node1"})
    graph.add_node("node2", lambda state: {"my_key": "hello from node2"})
    graph.add_edge(START, "node1")
    graph.add_edge("node1", "node2")
    graph.add_edge("node2", END)
    return graph

def test_individual_node_execution() -> None:
    # Will be ignored in this example
    checkpointer = MemorySaver()
    graph = create_graph()
    compiled_graph = graph.compile(checkpointer=checkpointer)
    # Only invoke node 1
    result = compiled_graph.nodes["node1"].invoke(
        {"my_key": "initial_value"},
    )
    assert result["my_key"] == "hello from node1"

部分执行

对于由大型图构成的智能体，您可能希望测试智能体内部的局部执行路径，而不是整个流程的端到端。在某些情况下，将这些部分作为子图重新结构化可能具有语义上的合理性，您可以像正常一样单独调用这些子图。

然而，如果您不想更改智能体图的总体结构，可以使用LangGraph的持久化机制来模拟一个状态，即您的智能体在所需部分的开始前暂停，并在所需部分的结束时再次暂停。具体步骤如下：

1. 使用检查点器编译您的智能体（内存检查点器InMemorySaver足以用于测试）。
2. 使用一个设置为要开始测试的节点名称的as_node参数调用您的智能体的update_state方法。
3. 使用与更新状态时相同的thread_id调用您的智能体，并将interrupt_after参数设置为要停止的节点名称。

以下是一个仅执行线性图中第二个和第三个节点的示例：

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 MyState(TypedDict):
        my_key: str

    graph = StateGraph(MyState)
    graph.add_node("node1", lambda state: {"my_key": "hello from node1"})
    graph.add_node("node2", lambda state: {"my_key": "hello from node2"})
    graph.add_node("node3", lambda state: {"my_key": "hello from node3"})
    graph.add_node("node4", lambda state: {"my_key": "hello from node4"})
    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_from_node2_to_node3() -> None:
    checkpointer = MemorySaver()
    graph = create_graph()
    compiled_graph = graph.compile(checkpointer=checkpointer)
    compiled_graph.update_state(
        config={
          "configurable": {
            "thread_id": "1"
          }
        },
        # The state passed into node 2 - simulating the state at
        # the end of node 1
        values={"my_key": "initial_value"},
        # Update saved state as if it came from node 1
        # Execution will resume at node 2
        as_node="node1",
    )
    result = compiled_graph.invoke(
        # Resume execution by passing None
        None,
        config={"configurable": {"thread_id": "1"}},
        # Stop after node 3 so that node 4 doesn't run
        interrupt_after="node3",
    )
    assert result["my_key"] == "hello from node3"

---

在GitHub上编辑此页面的源代码。

通过MCP将这些文档编程连接到Claude、VSCode等，以获取实时答案。