title: "Тестирование агентов: Mocked LLM, поведенческие утверждения, тестовый фреймворк" slug: agent-testing-mocked-llm-2026-ru date: 2026-02-24 lang: ru

Тестирование агентов: Mocked LLM, поведенческие утверждения, тестовый фреймворк

Ключевые факты

• pytest 9.0.2 — новейший фреймворк тестирования для Python (март 2026)
• pytest-asyncio 1.3.0 — поддержка асинхронных тестов для агентов LangChain/LangGraph
• langchain 1.2.10 — основная библиотека LangChain для оркестрации LLM
• langgraph 1.0.10 — stateful multi-agent workflows с StateGraph
• pytest-recording 0.13.4 — интеграция VCR.py для записи/воспроизведения HTTP-взаимодействий
• LangGraph MemorySaver — checkpointer в памяти для тестирования stateful агентов без persistence
• LangGraph InMemoryStore — ephemeral key-value хранилище для тестовых fixtures
• LangChain FakeListChatModel — встроенный mock LLM, возвращающий ответы из предопределённого списка
• unittest.mock — стандартная библиотека Python для создания mock-объектов и patching
• BaseChatModel interface — mock на методе _generate() для замены поведения LLM
• Ключевая проблема — недетерминизм LLM делает традиционные unit-тесты ненадёжными и flaky
• Проблема стоимости — реальные API-вызовы в тестах быстро накапливаются с каждым CI-запуском
• Проблема латентности — реальные вызовы LLM вносят задержку 2-10 секунд на тест
• Тестовая пирамида для агентов — unit (mocked LLM, free, fast) → integration (recorded LLM, realistic) → eval (real LLM, expensive)
• Cassette recording — захватить реальные ответы LLM один раз, воспроизводить детерминировано навсегда
• pytest-recording modes — record (захватить новое), replay (использовать существующее), none (fail если отсутствует)
• Tool call assertions — наиболее ценный тест поведения: проверить последовательность tool-вызовов, аргументы и результаты
• Thread isolation — каждый тест нуждается в уникальном thread_id для предотвращения загрязнения состояния
• Exhaust assertion — проверить, что mock потребил все scripted ответы для обнаружения неожиданных путей агента
• Property-based testing с hypothesis — генерировать разнообразные входные данные автоматически для перехвата граничных случаев

Проблема тестирования агентов

Традиционное тестирование программного обеспечения предполагает детерминизм: при идентичных входных данных функция возвращает идентичные выходные данные. Это предположение полностью нарушается для LLM-агентов.

Недетерминизм: Даже при temperature=0 выходные данные LLM варьируются из-за квантизации, обновлений модели и изменений на стороне API. Тест, который проходит сегодня, может не пройти завтра при идентичном коде.

Накопление стоимости: Тестовый набор из 200 тестов, вызывающих GPT-4 по цене $0.03 за 1K токенов, обойдётся в $6+ за запуск CI. При 50 ежедневных коммитов это $300/день или $9,000/месяц только на тесты.

Латентность: Вызовы API LLM занимают 2-10 секунд. Тестовый набор из 200 тестов, требующий реальных API-вызовов, выполнялся бы 6-30 минут, что делает TDD невозможным.

Flakiness cascade: Когда тесты периодически не проходят из-за дисперсии LLM, инженеры теряют доверие к тестовому набору. Команды начинают игнорировать отказы CI, что подрывает цель автоматизированного тестирования.

Решение — многоуровневая стратегия, которая изолирует вызовы LLM от бизнес-логики, использует детерминированные mocks для unit-тестов и оставляет реальные вызовы LLM для высокоуровневых smoke-тестов.

Фреймворк решений

Unit-тесты (Mocked LLM)

Когда использовать: Тестирование логики агента, потока управления, выбора tool, обработки ошибок, управления состоянием.

Характеристики:

• Mock LLM возвращает scripted ответы через ScriptedLLM или FakeListChatModel
• Тесты выполняются в миллисекундах
• Нулевая стоимость API
• Детерминировано: одинаковый входной результат всегда даёт одинаковый выход
• Возможна высокая покрытие (сотни тестов)

Что тестировать:

• Агент вызывает правильный tool при конкретном вводе пользователя
• Аргументы tool соответствуют ожидаемому формату
• Агент корректно восстанавливается из ошибок tool
• Состояние сохраняется правильно по поворотам разговора
• Budget enforcement останавливает исполнение на лимитах
• История разговора включает ожидаемые сообщения

Ограничения: Не может проверить качество prompt engineering, рассуждения LLM или реальные режимы отказов LLM.

Integration-тесты (Recorded LLM)

Когда использовать: Тестирование реалистичного поведения LLM без текущих расходов API.

Характеристики:

• Первый запуск использует реальный LLM, ответ сохраняется в файл cassette
• Последующие запуски воспроизводят из cassette (детерминировано, бесплатно, быстро)
• Cassettes коммитятся в git
• Используйте pytest-recording или custom RecordingLLM

Что тестировать:

• Полные workflows агента с реалистичными ответами LLM
• Эффективность шаблонов prompt
• Multi-turn conversations
• Сложные цепи рассуждений

Ограничения: Cassettes устаревают по мере эволюции prompt. Требует периодической переподготовки.

Evaluation-тесты (Real LLM)

Когда использовать: Финальная валидация перед production, регрессионное тестирование на наборах данных LangSmith.

Характеристики:

• Делает реальные API-вызовы
• Медленное (минуты-часы)
• Дорого
• Запускать редко (nightly builds, только перед релизом)

Что тестировать:

• End-to-end smoke-тесты на production-подобных сценариях
• LangSmith evaluations с человеческими labeled наборами данных
• Бенчмарки производительности (latency, token usage)
• A/B тестирование вариаций prompt

Best practice: Запускать только при слияниях в main ветку, не на каждый коммит.

Таблица ссылок параметров

Распространённые ловушки

Ловушка 1: Общее состояние между тестами

Влияние: Тесты проходят индивидуально, но не проходят при совместном запуске. Сложно отлаживать.

❌ Неправильно: Переиспользование того же thread_id по тестам

@pytest.fixture
def agent_config():
    return {"configurable": {"thread_id": "test-thread"}}  # Одинаково для всех тестов!

def test_greeting(agent, agent_config):
    result = agent.invoke({"messages": [{"role": "user", "content": "Hello"}]}, agent_config)
    assert "Hello" in result["messages"][-1].content

def test_question(agent, agent_config):
    # FAILS: видит "Hello" из предыдущего теста в истории разговора
    result = agent.invoke({"messages": [{"role": "user", "content": "What's 2+2?"}]}, agent_config)

✅ Правильно: Уникальный thread_id на тест

import uuid

@pytest.fixture
def agent_config():
    return {"configurable": {"thread_id": f"test-{uuid.uuid4().hex[:8]}"}}

Ловушка 2: Не исчерпываются Mock-ответы

Влияние: Агент берёт неожиданный путь, тест проходит молча, баги проскальзывают.

❌ Неправильно: Игнорирование неиспользованных ответов

def test_search(agent):
    llm = ScriptedLLM(responses=[
        make_tool_call_response("web_search", {"query": "test"}),
        make_text_response("Result 1"),
        make_text_response("Result 2"),  # Никогда не потреблён — агент остановился рано!
    ])
    agent = build_agent(llm=llm)
    result = agent.invoke({"messages": [{"role": "user", "content": "Search for test"}]})
    # Тест проходит, но мы никогда не проверили поведение агента после первого результата

✅ Правильно: Утверждать все ответы потреблены

def test_search(agent):
    llm = ScriptedLLM(responses=[
        make_tool_call_response("web_search", {"query": "test"}),
        make_text_response("Result 1"),
    ])
    agent = build_agent(llm=llm)
    result = agent.invoke({"messages": [{"role": "user", "content": "Search for test"}]})
    llm.assert_exhausted()  # FAIL если агент не потребил все ответы

Ловушка 3: Тестирование реализации вместо поведения

Влияние: Тесты ломаются при рефакторинге, не защищают от багов.

❌ Неправильно: Утверждение переменных внутреннего состояния

def test_research_agent(agent):
    result = agent.invoke({"messages": [{"role": "user", "content": "Research AI"}]})
    assert agent._internal_step_count == 3  # Brittle!
    assert agent._search_cache["AI"] is not None  # Деталь реализации!

✅ Правильно: Утверждать наблюдаемое поведение (tool calls, outputs)

def test_research_agent(agent):
    result = agent.invoke({"messages": [{"role": "user", "content": "Research AI"}]})
    assertions = AgentRunAssertions(result["messages"])
    assertions.assert_tool_called_once("web_search")
    assertions.assert_final_message_contains("AI")

Ловушка 4: Запись Cassettes с ненулевой температурой

Влияние: Cassettes содержат варьирующиеся ответы, воспроизведение детерминировано, но ожидания теста не совпадают.

❌ Неправильно: Запись с creative LLM

recorder = RecordingLLM(
    inner_llm=ChatOpenAI(model="gpt-4o", temperature=0.9),  # Высокая дисперсия!
    record_mode=True
)

✅ Правильно: Минимизировать дисперсию во время записи

recorder = RecordingLLM(
    inner_llm=ChatOpenAI(model="gpt-4o-mini", temperature=0, seed=42),
    record_mode=True
)

Ловушка 5: Не изолируются побочные эффекты Tool

Влияние: Тесты изменяют файловую систему, базы данных или внешние сервисы. Очистка ручная и подвержена ошибкам.

❌ Неправильно: Реальные tools в тестах

@tool
def write_file(path: str, content: str) -> str:
    """Запишите содержимое в файл."""
    with open(path, "w") as f:  # Пишет в реальную файловую систему!
        f.write(content)
    return "Written"

✅ Правильно: Mock tools с контролируемым поведением

@pytest.fixture
def mock_write_tool():
    written_files = {}

    @tool
    def write_file(path: str, content: str) -> str:
        """Запишите содержимое в файл."""
        written_files[path] = content  # Сохраняет в памяти
        return "Written"

    write_file.written_files = written_files
    return write_file

Mocking LLM-ответов

Scripted Response Mock

Простейший mock возвращает ответы по очереди из предопределённого списка. Выбрасывает AssertionError, если агент вызывает LLM больше раз, чем ожидается.

from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage, BaseMessage
from langchain_core.outputs import ChatResult, ChatGeneration
from typing import Optional, Any

class ScriptedLLM(BaseChatModel):
    """Возвращает ответы из предопределённой последовательности."""
    responses: list[dict]
    _call_index: int = 0

    class Config:
        arbitrary_types_allowed = True

    def _generate(
        self,
        messages: list[BaseMessage],
        stop: Optional[list[str]] = None,
        **kwargs: Any,
    ) -> ChatResult:
        if self._call_index >= len(self.responses):
            raise AssertionError(
                f"ScriptedLLM exhausted: {len(self.responses)} responses "
                f"configured, call {self._call_index + 1} received.\n"
                f"Last messages: {[m.content for m in messages[-3:]]}"
            )

        response = self.responses[self._call_index]
        self._call_index += 1

        return ChatResult(
            generations=[
                ChatGeneration(
                    message=AIMessage(
                        content=response.get("content", ""),
                        tool_calls=response.get("tool_calls", []),
                    )
                )
            ]
        )

    @property
    def _llm_type(self) -> str:
        return "scripted_mock"

    def assert_exhausted(self) -> None:
        """Утверждать, что все scripted ответы были потреблены."""
        assert self._call_index == len(self.responses), (
            f"Not all responses consumed: used {self._call_index} "
            f"of {len(self.responses)}"
        )

# Вспомогательные функции
def make_tool_call_response(tool_name: str, tool_args: dict, call_id: str = "call_abc123") -> dict:
    return {
        "content": "",
        "tool_calls": [{
            "id": call_id,
            "name": tool_name,
            "args": tool_args,
            "type": "tool_call",
        }]
    }

def make_text_response(content: str) -> dict:
    return {"content": content, "tool_calls": []}

Conditional Mock для логики ветвления

При тестировании агентов с условным поведением (например, «если пользователь просит поиск, вызвать search tool; иначе ответить напрямую»), используйте rule-based mock.

from typing import Callable

class ConditionalLLM(BaseChatModel):
    """Возвращает разные ответы в зависимости от условий содержимого сообщения."""
    rules: list[tuple[Callable[[list[BaseMessage]], bool], dict]]
    default_response: dict = {"content": "Default response", "tool_calls": []}
    call_log: list[dict] = []

    def _generate(
        self,
        messages: list[BaseMessage],
        stop: Optional[list[str]] = None,
        **kwargs: Any,
    ) -> ChatResult:
        matched_response = self.default_response
        for condition, response in self.rules:
            if condition(messages):
                matched_response = response
                break

        self.call_log.append({
            "messages": [{"role": m.type, "content": str(m.content)} for m in messages],
            "response": matched_response,
        })

        return ChatResult(
            generations=[
                ChatGeneration(
                    message=AIMessage(
                        content=matched_response.get("content", ""),
                        tool_calls=matched_response.get("tool_calls", []),
                    )
                )
            ]
        )

    @property
    def _llm_type(self) -> str:
        return "conditional_mock"

# Вспомогательный инструмент для построения rules
def contains_text(text: str) -> Callable[[list[BaseMessage]], bool]:
    def check(messages: list[BaseMessage]) -> bool:
        return any(text.lower() in str(m.content).lower() for m in messages)
    return check

# Использование
llm = ConditionalLLM(
    rules=[
        (contains_text("search"), make_tool_call_response("web_search", {"query": "test"})),
        (contains_text("calculate"), make_tool_call_response("calculator", {"expression": "2+2"})),
    ],
    default_response=make_text_response("I can help with that.")
)

FakeListChatModel (встроенная)

LangChain предоставляет FakeListChatModel для простых последовательных ответов без написания custom mock-классов.

from langchain_core.language_models.fake_chat_models import FakeListChatModel

llm = FakeListChatModel(responses=["First response", "Second response"])

Ограничение: FakeListChatModel поддерживает только текстовые ответы, не tool calls. Используйте ScriptedLLM для tool-calling агентов.

Integration-тестирование с VCR/Recording

Cassette recording захватывает реальные ответы LLM один раз, затем воспроизводит их детерминировано. Это предоставляет реалистичное поведение LLM без текущих расходов API.

Использование pytest-recording

import pytest

@pytest.mark.vcr
def test_research_workflow(agent):
    """Первый запуск: вызывает реальный API, сохраняет в cassette.
    Последующие запуски: воспроизведение из cassette.
    """
    result = agent.invoke({
        "messages": [{"role": "user", "content": "What are the latest AI trends?"}]
    })
    assert "AI" in result["messages"][-1].content

Структура директорий:

tests/
  cassettes/
    test_research_workflow.yaml  # Автогенерируется

Переподготовка: Удалите файл cassette и запустите тест снова.

Custom RecordingLLM

Для большего контроля реализуйте custom recording mock, который хеширует содержимое сообщения для создания cache ключей.

import json
import hashlib
from pathlib import Path
from langchain_openai import ChatOpenAI

class RecordingLLM(BaseChatModel):
    """Записывает реальные вызовы LLM на диск и воспроизводит при последующих запусках."""
    inner_llm: Optional[BaseChatModel] = None
    cassette_dir: str = "tests/cassettes"
    record_mode: bool = False
    _cassette: dict = {}

    def model_post_init(self, __context: Any) -> None:
        cassette_path = Path(self.cassette_dir)
        cassette_path.mkdir(parents=True, exist_ok=True)

        cassette_file = cassette_path / "recordings.json"
        if cassette_file.exists():
            with open(cassette_file) as f:
                self._cassette = json.load(f)

    def _make_key(self, messages: list[BaseMessage]) -> str:
        content = json.dumps([
            {"role": m.type, "content": str(m.content)}
            for m in messages
        ], sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()[:16]

    def _generate(
        self,
        messages: list[BaseMessage],
        stop: Optional[list[str]] = None,
        **kwargs: Any,
    ) -> ChatResult:
        key = self._make_key(messages)

        if key in self._cassette and not self.record_mode:
            recorded = self._cassette[key]
            return ChatResult(
                generations=[
                    ChatGeneration(
                        message=AIMessage(
                            content=recorded.get("content", ""),
                            tool_calls=recorded.get("tool_calls", []),
                        )
                    )
                ]
            )

        if self.record_mode and self.inner_llm:
            result = self.inner_llm._generate(messages, stop=stop, **kwargs)
            ai_message = result.generations[0].message

            self._cassette[key] = {
                "content": ai_message.content,
                "tool_calls": getattr(ai_message, "tool_calls", []),
            }

            cassette_file = Path(self.cassette_dir) / "recordings.json"
            with open(cassette_file, "w") as f:
                json.dump(self._cassette, f, indent=2)

            return result

        raise ValueError(f"No recording for key {key} and record_mode=False.")

    @property
    def _llm_type(self) -> str:
        return "recording_mock"

Использование:

# Режим записи (запустить один раз)
# LLM_TEST_MODE=record pytest tests/integration/

@pytest.fixture(scope="session")
def llm():
    mode = os.environ.get("LLM_TEST_MODE", "replay")
    if mode == "record":
        return RecordingLLM(
            inner_llm=ChatOpenAI(model="gpt-4o-mini", temperature=0),
            record_mode=True
        )
    return RecordingLLM(record_mode=False)

LangGraph-специфичное тестирование

Тестирование структуры StateGraph

Проверьте топологию графика перед тестированием исполнения.

from langgraph.graph import StateGraph

def test_graph_has_required_nodes(agent):
    """Проверить, что граф содержит ожидаемые узлы."""
    graph = agent.get_graph()
    node_names = [node.id for node in graph.nodes]

    assert "agent" in node_names
    assert "tools" in node_names
    assert "__end__" in node_names

def test_all_nodes_reach_end(agent):
    """Проверить отсутствие dead-end узлов."""
    graph = agent.get_graph()

    def can_reach_end(node_id: str, visited: set) -> bool:
        if node_id == "__end__":
            return True
        if node_id in visited:
            return False
        visited.add(node_id)

        edges = [e for e in graph.edges if e.source == node_id]
        return any(can_reach_end(e.target, visited.copy()) for e in edges)

    for node in graph.nodes:
        if node.id != "__end__":
            assert can_reach_end(node.id, set()), f"Node '{node.id}' cannot reach END"

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

Используйте MemorySaver для test checkpoints вместо реальной persistence.

from langgraph.checkpoint.memory import MemorySaver
import uuid

@pytest.fixture
def memory_checkpointer():
    return MemorySaver()

@pytest.fixture
def agent_config():
    return {
        "configurable": {"thread_id": f"test-{uuid.uuid4().hex[:8]}"},
        "recursion_limit": 10,
    }

def test_multi_turn_conversation(agent, memory_checkpointer, agent_config):
    """Состояние сохраняется по поворотам."""
    llm = ConditionalLLM(
        rules=[
            (contains_text("my name is"), make_text_response("Nice to meet you!")),
            (contains_text("what is my name"), make_text_response("You said your name is Alice.")),
        ]
    )
    app = build_agent(llm=llm, checkpointer=memory_checkpointer)

    # Поворот 1
    app.invoke({"messages": [{"role": "user", "content": "My name is Alice"}]}, agent_config)

    # Поворот 2
    result = app.invoke({"messages": [{"role": "user", "content": "What is my name?"}]}, agent_config)

    assert "Alice" in result["messages"][-1].content

Tool Mocking в LangGraph

Замените реальные tools на mocks, которые отслеживают вызовы.

from langchain_core.tools import tool

@pytest.fixture
def mock_search_tool():
    call_history = []

    @tool
    def web_search(query: str) -> str:
        """Поиск в интернете."""
        call_history.append({"query": query})
        return f"Mock result for: {query}"

    web_search.call_history = call_history
    return web_search

def test_agent_calls_search(mock_search_tool, agent_config):
    llm = ScriptedLLM(responses=[
        make_tool_call_response("web_search", {"query": "AI news"}),
        make_text_response("Here are the latest AI developments..."),
    ])

    app = build_agent(llm=llm, tools=[mock_search_tool])
    app.invoke({"messages": [{"role": "user", "content": "Search for AI news"}]}, agent_config)

    assert len(mock_search_tool.call_history) == 1
    assert mock_search_tool.call_history[0]["query"] == "AI news"

Производительность и бенчмарки

Примечание: Приведённые ниже цифры являются иллюстративными оценками на основе типичных production-конфигураций, а не измерениями конкретной системы.

Скорость исполнения тестов

Unit-тесты с mocked LLM: Тесты завершаются в 10-50ms каждый. Набор из 200 unit-тестов выполняется менее чем за 10 секунд.

Integration-тесты с cassettes: Первый запуск (запись) занимает 2-10 секунд на тест из-за реальных API-вызовов. Последующие запуски (воспроизведение) завершаются в 50-200ms на тест, аналогично unit-тестам, но с реалистичными ответами LLM.

Evaluation-тесты с реальным LLM: Тесты занимают 2-10 секунд каждый в зависимости от сложности агента и latency LLM. Набор из 20 smoke-тестов может занять 1-3 минуты.

Сравнение стоимости

Реальный API в каждом тесте: Набор из 200 тестов, вызывающих GPT-4 с 1K токенами prompt + 500 токенов completion на тест, стоит примерно $3-6 за запуск. При 50 CI-запусках в день, ежемесячная стоимость превышает $4,500.

Cassette воспроизведение: После начальной записи, стоимость падает до нуля при последующих запусках. Ежемесячная экономия $4,000+ по сравнению с реальными API-вызовами.

Mocked unit-тесты: Нулевая стоимость, нулевая латентность, неограниченные запуски тестов.

Распределение покрытия тестами

Для production system агента сбалансированное распределение тестов может быть:

• Unit-тесты (mocked): 70-80% всех тестов, охватывая потоки управления, обработку ошибок, управление состоянием
• Integration-тесты (cassettes): 15-25% тестов, охватывая реалистичные workflows и эффективность prompt
• Evaluation-тесты (real LLM): 5-10% тестов, охватывая end-to-end smoke-тесты и обнаружение регрессий

Это распределение балансирует скорость, стоимость и уверенность.