Как тестировать LangGraph-приложение — какие стратегии тестирования доступны?

Уровень 1: юнит-тесты узлов

Узлы — обычные Python-функции; тестируются без компиляции графа.

import pytest
from unittest.mock import MagicMock, patch
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage


class State(TypedDict):
    messages: Annotated[list, add_messages]


def call_model(state: State) -> dict:
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(model="gpt-4o-mini")
    response = llm.invoke(state["messages"])
    return {"messages": [response]}


def test_call_model_unit():
    """Тест узла с мок-LLM."""
    mock_response = AIMessage(content="Ответ модели")

    with patch("langchain_openai.ChatOpenAI.invoke", return_value=mock_response):
        state = {"messages": [HumanMessage(content="Привет")]}
        result = call_model(state)

    assert result["messages"][0].content == "Ответ модели"

Стратегии тестирования LangGraph

Граф LangGraph состоит из независимых узлов, редукторов и маршрутизаторов — каждый элемент тестируется отдельно, что упрощает изоляцию проблем.

Уровень 2: интеграционные тесты графа

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END


def build_test_graph(llm_mock):
    """Строим граф с подменённой моделью."""
    def node(state: State) -> dict:
        response = llm_mock.invoke(state["messages"])
        return {"messages": [response]}

    builder = StateGraph(State)
    builder.add_node("model", node)
    builder.set_entry_point("model")
    builder.add_edge("model", END)
    return builder.compile(checkpointer=MemorySaver())


def test_graph_full_run():
    llm_mock = MagicMock()
    llm_mock.invoke.return_value = AIMessage(content="OK")

    graph = build_test_graph(llm_mock)
    config = {"configurable": {"thread_id": "test-1"}}

    result = graph.invoke(
        {"messages": [HumanMessage(content="test")]},
        config=config,
    )

    assert len(result["messages"]) == 2
    assert result["messages"][-1].content == "OK"

    # Проверяем сохранённое состояние
    saved = graph.get_state(config)
    assert saved is not None

Тестирование маршрутизаторов

from langgraph.graph import END


def should_continue(state: State) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return END


@pytest.mark.parametrize("has_tool_calls,expected", [
    (True, "tools"),
    (False, END),
])
def test_router(has_tool_calls, expected):
    msg = AIMessage(content="test")
    if has_tool_calls:
        msg.tool_calls = [{"name": "search", "args": {}, "id": "1"}]
    else:
        msg.tool_calls = []

    state = {"messages": [msg]}
    assert should_continue(state) == expected

Тестирование с FakeListChatModel

from langchain_core.language_models.fake import FakeListChatModel
from langchain_core.messages import AIMessage

# FakeListChatModel возвращает заранее заданные ответы по очереди
fake_llm = FakeListChatModel(
    responses=["Первый ответ", "Второй ответ", "Финальный ответ"]
)

# Используйте его вместо ChatOpenAI в тестах — не требует API ключа

E2E тесты с LangSmith

# Установите LANGCHAIN_TRACING_V2=true, LANGCHAIN_API_KEY=...
# Тесты автоматически создают трейсы в LangSmith
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "test-suite"

# При запросе pytest трейс появится в LangSmith UI

Подводные камни

• Реальные API-вызовы в тестах — без моков тесты медленные и платные. Всегда мокируйте LLM в юнит/интеграционных тестах.
• thread_id в тестах — используйте уникальные thread_id (например, через uuid.uuid4()), иначе состояние одного теста влияет на другой при shared MemorySaver.
• Недетерминизм LLM — E2E тесты с реальной моделью флакуют. Используйте temperature=0 и seed где возможно.
• Тестирование параллельных ветвей Send — порядок результатов недетерминирован; проверяйте множества, а не списки: assert set(result) == {"a", "b"}.
• Async тесты требуют pytest-asyncio — добавьте @pytest.mark.asyncio или настройте asyncio_mode = "auto" в pytest.ini.
• FakeListChatModel исчерпывается — если граф делает больше вызовов, чем ответов в списке, бросается исключение; планируйте размер списка заранее.
• Проверяйте редукторы отдельно — кастомные функции-редукторы легко сломать; тестируйте их как чистые функции независимо от графа.

• Объяснять testing strategies только синтаксисом без shape, dtype, состояния или режима выполнения.
• Игнорировать leakage, воспроизводимость, пустые входы и скрытые копии данных.
• Не проверять production-симптомы: latency, память, ретраи, дрейф качества и несовпадение версий.

• Может ли связать testing strategies с реальным контрактом входов и выходов.
• Упоминает ли тесты, метрики, reproducibility и диагностику ошибок.
• Видит ли различие между demo-кодом в ноутбуке и production-пайплайном.