Skip to main content

Documentation Index

Fetch the complete documentation index at: https://nvd-54.mintlify.app/llms.txt

Use this file to discover all available pages before exploring further.

在完成 LangGraph 智能体的原型开发后,自然的下一步是添加测试。本指南涵盖了编写单元测试时可以使用的一些有用模式。 请注意,本指南是 LangGraph 特定的,涵盖了具有自定义结构的图的场景——如果你刚开始入门,请查看使用 LangChain 内置 create_agent测试

前提条件

首先,确保你已安装 pytest
$ pip install -U pytest

快速上手

由于许多 LangGraph 智能体依赖于状态,一个有用的模式是在每个使用它的测试之前创建图,然后在测试中使用新的检查点器实例编译它。 以下示例展示了如何使用一个简单的线性图来实现这一点,该图依次经过 node1node2。每个节点更新单个状态键 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:
    # 在此示例中将被忽略
    checkpointer = MemorySaver()
    graph = create_graph()
    compiled_graph = graph.compile(checkpointer=checkpointer)
    # 仅调用节点 1
    result = compiled_graph.nodes["node1"].invoke(
        {"my_key": "initial_value"},
    )
    assert result["my_key"] == "hello from node1"

部分执行

对于由较大图组成的智能体,你可能希望测试智能体中的部分执行路径,而不是端到端的完整流程。在某些情况下,将这些部分重构为子图在语义上可能更有意义,这样你可以像平常一样独立调用它们。 然而,如果你不想更改智能体图的整体结构,可以使用 LangGraph 的持久化机制来模拟智能体在所需部分开始之前暂停的状态,并在所需部分结束时再次暂停。步骤如下:
  1. 使用检查点器编译你的智能体(内存检查点器 InMemorySaver 足以用于测试)。
  2. 调用智能体的 update_state 方法,将 as_node 参数设置为你想要开始测试的节点之前的节点名称。
  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"
          }
        },
        # 传入节点 2 的状态 - 模拟节点 1 结束时的状态
        values={"my_key": "initial_value"},
        # 更新保存的状态,就好像它来自节点 1
        # 执行将从节点 2 恢复
        as_node="node1",
    )
    result = compiled_graph.invoke(
        # 通过传入 None 恢复执行
        None,
        config={"configurable": {"thread_id": "1"}},
        # 在节点 3 之后停止,这样节点 4 不会运行
        interrupt_after="node3",
    )
    assert result["my_key"] == "hello from node3"