Documentation Index
Fetch the complete documentation index at: https://langchain.idochub.dev/llms.txt
Use this file to discover all available pages before exploring further.
在完成 LangGraph 代理的原型开发后,自然的下一步是为其添加测试。本指南介绍编写单元测试时可使用的一些实用模式。
请注意,本指南专为 LangGraph 设计,涵盖具有自定义结构图的测试场景。如果您刚刚入门,建议先查看这一章节,它使用的是 LangChain 内置的 create_agent() 方法。
前置条件
首先,请确保已安装 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
from langgraph.graph import StateGraph
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)
# 仅调用 node1
result = compiled_graph.nodes["node1"].invoke(
{"my_key": "initial_value"},
)
assert result["my_key"] == "hello from node1"
部分执行
对于由大型图结构组成的代理,您可能希望测试代理中的部分执行路径,而非完整的端到端流程。在某些情况下,从语义上将这些部分重构为子图可能更合理,这样您就可以像平常一样独立调用它们。
但如果您不希望更改代理图的整体结构,可以利用 LangGraph 的持久化机制,模拟代理在目标段落开始前暂停、并在目标段落结束后再次暂停的状态。步骤如下:
- 使用检查点器编译您的代理(测试时使用内存检查点器
InMemorySaver 即可)。
- 调用代理的
update_state 方法,并将 as_node 参数设置为您希望测试起始节点之前的那个节点名称。
- 使用与更新状态时相同的
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"
}
},
# 传入 node2 的状态 —— 模拟 node1 结束后的状态
values={"my_key": "initial_value"},
# 将保存状态标记为来自 node1
# 执行将从 node2 开始恢复
as_node="node1",
)
result = compiled_graph.invoke(
# 通过传入 None 恢复执行
None,
config={"configurable": {"thread_id": "1"}},
# 在 node3 后中断,避免执行 node4
interrupt_after="node3",
)
assert result["my_key"] == "hello from node3"