테스트

Alpha Notice: These docs cover the v1-alpha release. Content is incomplete and subject to change.

For the latest stable version, see the v0 LangChain Python or LangChain JavaScript docs.

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:
    # 이 예제에서는 무시됩니다
    checkpointer = MemorySaver()
    graph = create_graph()
    compiled_graph = graph.compile(checkpointer=checkpointer)
    # node 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"
          }
        },
        # node 2로 전달될 상태 - node 1 종료 시점의 상태를 시뮬레이션합니다
        values={"my_key": "initial_value"},
        # node 1에서 온 것처럼 저장된 상태를 업데이트합니다
        # 실행은 node 2에서 재개됩니다
        as_node="node1",
    )
    result = compiled_graph.invoke(
        # None을 전달하여 실행을 재개합니다
        None,
        config={"configurable": {"thread_id": "1"}},
        # node 3 이후에 중지하여 node 4가 실행되지 않도록 합니다
        interrupt_after="node3",
    )
    assert result["my_key"] == "hello from node3"

---

Edit the source of this page on GitHub