跳到主要内容

文档索引

在以下地址获取完整的文档索引:https://docs.langchain.org.cn/llms.txt

在进一步探索之前,请使用此文件发现所有可用页面。

在您完成 LangGraph Agent 的原型设计后,自然的下一步就是添加测试。本指南介绍了一些在编写单元测试时可以使用的实用模式。 请注意,本指南专用于 LangGraph,涵盖了具有自定义结构的图相关场景——如果您才刚刚开始,请查看使用 LangChain 内置 create_agent测试指南。

先决条件

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

入门

由于许多 LangGraph Agent 依赖于状态,一种有用的模式是在每次测试之前创建图,然后在测试中使用新的检查点实例对其进行编译。 下面的示例展示了如何在一个简单的线性图中实现这一点,该图按顺序经过 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 Agent 会将各个节点的引用公开为 graph.nodes。您可以利用这一点来测试 Agent 中的单个节点。注意,这将绕过在编译图时传递的任何检查点。
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"

部分执行

对于由大型图组成的 Agent,您可能希望测试 Agent 内的部分执行路径,而不是整个流程的端到端测试。在某些情况下,将这些部分重构为子图在语义上更有意义,您可以像正常情况一样单独调用它们。 但是,如果您不想更改 Agent 图的整体结构,可以使用 LangGraph 的持久化机制来模拟 Agent 在所需部分开始前暂停,并在所需部分结束后再次暂停的状态。步骤如下:
  1. 使用检查点编译您的 Agent(内存中检查点 InMemorySaver 即可满足测试需求)。
  2. 调用 Agent 的 update_state 方法,并将 as_node 参数设置为您想要开始测试的节点之前的节点名称。
  3. 使用您更新状态时所用的相同 thread_id 调用您的 Agent,并将 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"

© . This site is unofficial and not affiliated with LangChain, Inc.