테스트

테스트

LangGraph 에이전트 프로토타입을 만들었다면, 다음 단계로 테스트를 추가하는 게 자연스러워요. 이 가이드는 단위 테스트를 작성할 때 쓸 수 있는 몇 가지 유용한 패턴을 다뤄요.

이 가이드는 LangGraph 특화로, 커스텀 구조를 가진 그래프를 중심으로 한 시나리오를 다뤄요. 막 시작했다면 LangChain의 내장 create_agent를 쓰는 Test부터 보는 게 좋아요.

사전 요구사항

먼저 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"
          }
        },
        # 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 4가 실행되지 않도록 node 3 이후에 멈춤
        interrupt_after="node3",
    )
    assert result["my_key"] == "hello from node3"

더 알아보기 (Learn more)