테스트

테스트 (Test)

LangGraph 에이전트를 프로토타이핑했다면, 그다음 자연스러운 단계는 테스트를 추가하는 거예요. 이 가이드는 단위 테스트를 작성할 때 유용한 몇 가지 패턴을 다뤄요.

참고로 이 가이드는 LangGraph에 특화되어 있어서 커스텀 구조를 가진 그래프를 다루는 시나리오를 중심으로 해요. 이제 막 시작하는 분이라면 LangChain 내장 create_agent를 쓰는 Test를 먼저 확인해 보세요.

출처: 문서

본문

사전 준비 (Prerequisites)

먼저 pytest를 설치해요.

$ pip install -U pytest

시작하기 (Getting started)

많은 LangGraph 에이전트가 상태(state)에 의존하므로, 그래프를 쓰기 전에 매 테스트마다 그래프를 만들고, 새 checkpointer 인스턴스와 함께 테스트 안에서 컴파일하는 패턴이 유용해요.

아래 예시는 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"

개별 노드·엣지 테스트 (Testing individual nodes and edges)

컴파일된 LangGraph 에이전트는 각 개별 노드를 graph.nodes로 노출해요. 이 점을 활용해 에이전트 안의 개별 노드를 테스트할 수 있어요. 다만 이 방법은 그래프를 컴파일할 때 전달한 checkpointer를 우회한다는 점을 기억하세요.

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"

부분 실행 (Partial execution)

더 큰 그래프로 이루어진 에이전트라면, 전체 흐름을 end-to-end로 테스트하기보다 에이전트 안의 특정 실행 경로만 부분적으로 테스트하고 싶을 수 있어요. 어떤 경우엔 이런 섹션을 서브그래프로 재구성하는 게 의미상 맞을 수도 있는데, 그렇게 하면 평소처럼 격리해 호출할 수 있어요.

하지만 그래프의 전체 구조를 바꾸고 싶지 않다면, LangGraph의 영속성 메커니즘을 이용해 에이전트가 원하는 섹션 직전에서 일시정지된 것 같은 상태를 시뮬레이션하고, 원하는 섹션 끝에서 다시 일시정지되게 할 수 있어요. 절차는 다음과 같아요.

  1. 에이전트를 checkpointer와 함께 컴파일해요. (테스트 목적이라면 인메모리 checkpointer인 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"

더 알아보기 (Learn more)