Взять фреймворк и создать AI-агента на внутренних данных компании — задача, которая на этапе прототипа проходит гладко. Настоящие проблемы начинаются в тот момент, когда этот код выводится в продакшен. Агент, который отлично работал на экране, из-за одного неожиданного пользовательского ввода может зациклиться и начать бесконечно повторять вызовы, в результате чего за выходные вам может прийти счет за API на миллионы вон.

Значительная часть сбоев, возникающих на практике, происходит не из-за ограничений самой модели LLM, а из-за ошибок управления состоянием и отсутствия внешнего контроля. Ситуацию, когда по указанию руководства был создан кастомный инструмент, но вместо основной работы приходится целый день отлаживать галлюцинации агента и разбирать поступающие чеки за токены, необходимо прекратить.

Три препятствия в продакшен-среде

Выйдя за пределы демо-среды, агент сталкивается с тремя преградами: работа по принципу «черного ящика», недетерминированные циклы и риск утечки данных.

Если входные и выходные данные на каждом этапе — процесса рассуждения (Chain-of-Thought), вызова внешних инструментов (Tool Call) и поиска в векторной базе данных — не логируются, при возникновении проблемы определить её причину невозможно. На один только поиск того, где именно сбился промпт, уходит целый день.

Ещё более опасен бесконечный цикл. Агент с архитектурой ReAct (Reasoning + Acting) при неоднозначном результате выполнения инструмента продолжает отправлять один и тот же запрос снова и снова.

┌─────────────────────────────────────────────────────────────────────────┐ │ ReAct Architecture Loop │ │ │ │ ┌────────────┐ User Query ┌────────────┐ Tool Call Request │ │ │ User │ ──────────────> │ Main LLM │ ────────────────────┐ │ │ └────────────┘ └────────────┘ │ │ │ ▲ ▼ │ │ │ Observation ┌──────┐ │ │ │ (Ambiguous/Failed) │ Tool │ │ │ └─────────────────────── │ A │ │ │ └──────┘ │ │ * Problem: When Observation fails, LLM retries Tool A endlessly. │ └─────────────────────────────────────────────────────────────────────────┘

Как только система не может выйти из условия завершения и продолжает непрерывно отправлять запросы к API с теми же аргументами, месячный бюджет на токены истощается за считанные минуты. Каскадные сбои, когда главный и дочерние агенты вызывали друг друга и повторяли попытки, составляют более 30% всех причин сбоев всей системы.

Также возникает проблема снижения производительности самой модели, когда история диалога постоянно накапливается и полностью заполняет контекстное окно. Вдобавок к этому промпты, захардкоженные в виде строк внутри кода Python, заставляют заново пересобирать и деплоить всю систему даже при изменении одной простой формулировки.

Система наблюдаемости, разрывающая бесконечные циклы и предотвращающая счета-бомбы

Необходимо разделить решения для локальной отладки и инструменты наблюдаемости в продакшене, правильно расставив точки сбора данных. На этапе разработки качество эмбеддингов RAG и вызовы инструментов проверяются с помощью Arize Phoenix на базе OpenTelemetry. В продакшен-среде подключается Langfuse с бэкендом на ClickHouse для мониторинга трейс-логов и расхода токенов в режиме реального времени.

┌─────────────────────────────────────────────────────────────────────────┐ │ Semantic Caching & Routing Flow │ │ │ │ Client Query │ │ │ │ │ ▼ │ │ ┌───────────┐ Similarity >= 0.92? ┌─────────────────────────────┐ │ │ │ Redis Vector │ ─────────────────────────> │ Return Cached Response │ │ │ │ Cache │ (Cache Hit) │ (Latency -88%, Cost -86%) │ │ │ └───────────┘ └─────────────────────────────┘ │ │ │ │ │ │ (Cache Miss) │ │ ▼ │ │ ┌───────────┐ Execution & Save Cache ┌─────────────────────────────┐ │ │ │ External │ ────────────────────────> │ Store Result as Vector in │ │ │ │ LLM API │ │ Backend Database │ │ │ └───────────┘ └─────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────┘

Расходы на API, возникающие из-за повторяющихся одинаковых запросов, блокируются с помощью слоя семантического кэширования (Semantic Caching).

Согласно данным анализа 60 000 запросов, опубликованным AWS, при внедрении семантического кэша на основе косинусного сходства затраты на инференс LLM сократились максимум на 86%, а задержка ответа уменьшилась на 88%. Фреймворк RouteLLM от LMSYS также сэкономил 85% затрат, распределяя запросы между дорогостоящими и бюджетными моделями в зависимости от сложности вопроса.

Семантическое кэширование строится в следующем порядке:

1. Подготовка бэкенда Redis Vector Search: Запускается инстанс Redis для преобразования и хранения входных промптов в виде векторов эмбеддингов.
2. Оценка сходства и возврат кэша: Если косинусное сходство между новым вектором промпта и ранее сохраненными векторами составляет 0.92 или выше, сохраненный ответ отдается мгновенно без вызова внешнего LLM.
3. Сохранение при промахе кэша (Cache Miss): Внешний API вызывается только тогда, когда сходство ниже 0.92, а полученный ответ обновляется и сохраняется в Redis.

Чтобы кардинально заблокировать бесконечные циклы, паттерн Circuit Breaker («автоматический выключатель»), проверяющий состояние агента, должен быть внедрен непосредственно в поток управления пайплайном. Полагаться на стандартные опции повтора (retry), предоставляемые фреймворком, опасно: при возникновении исключения упадет весь сервис. Безопаснее всего хэшировать имя инструмента и значения аргументов с помощью SHA-256, добавлять их в список и при накоплении 3 одинаковых хэшей подряд принудительно перенаправлять выполнение на другой узел.

`python
import hashlib
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
steps: int
tool_hashes: List[str]

def hash_tool_call(tool_name: str, tool_args: str) -> str:
raw_str = f"{tool_name}:{tool_args}"
return hashlib.sha256(raw_str.encode('utf-8')).hexdigest()

def agent_circuit_breaker_router(state: AgentState) -> str:
# 1. Если общее количество шагов превышает 5, переход на заданный запасной узел
if state["steps"] > 5:
return "fallback_graceful_node"

# 2. Если один и тот же Tool с теми же Argument вызывается 3 раза подряд, немедленная блокировка
hashes = state.get("tool_hashes", [])
if len(hashes) >= 3 and hashes[-1] == hashes[-2] == hashes[-3]:
    return "fallback_graceful_node"
    
# 3. Проверка ключевого слова нормального завершения
last_message = state["messages"][-1]
if "FINAL_ANSWER" in last_message.content:
    return "end"
    
return "continue_tools"

`

Независимо от недетерминированного состояния вывода LLM, этот роутер проверяет условные операторы в среде выполнения Python, физически предотвращая вход в цикл из-за галлюцинаций.

Разделение версий промптов и автоматизация CI/CD

Оставлять промпты внутри кода нельзя: вы не сможете отловить регрессионные сбои (Regression Failure), когда после изменения модели перестает работать даже то, что раньше функционировало исправно. Промпты следует вынести из кода Python и управлять ими в отдельных файлах YAML.

`yaml
name: "agent_reasoning"
version: "1.2.0"
model: "gpt-4o"
temperature: 0.1
messages:

• role: "system"
  content: |
  Вы — агент для анализа внутренней базы данных компании.
  Отвечайте строго в соответствии с заданной JSON-схемой. Если информации недостаточно, сгенерируйте уточняющие вопросы.
• role: "user"
  content: "{{ user_input }}"
  `

Вынесенные таким образом промпты автоматически тестируются в CI/CD-пайплайне с связке с OpenSource-фреймворком оценки DeepEval и Pytest. Метрика G-Eval фиксируется в качестве эталона, и перед деплоем показатели ответов модели верифицируются.

`python
import pytest
from deepeval import assert_test
from deepeval.metrics import GEval, TaskCompletionMetric
from deepeval.test_case import LLMTestCase, SingleTurnParams

1. Определение критериев оценки

correctness_metric = GEval(
name="Точность и соответствие схеме",
criteria="Отвечает ли LLM точно на вопрос и полностью ли соответствует запрошенному формату JSON?",
evaluation_params=[SingleTurnParams.ACTUAL_OUTPUT, SingleTurnParams.EXPECTED_OUTPUT],
threshold=0.7
)

2. Регрессионное тестирование на базе Pytest

@pytest.mark.parametrize(
"user_input, expected_output",
[
("2024년 1분기 매출 데이터를 요약해줘.", "1분기 총 매출은 50억 원입니다."),
("퇴직금 계산 규정을 알려줘.", "퇴직금은 근속연수 1년에 대해 30일분 이상의 평균임금입니다.")
]
)
def test_agent_regression(user_input, expected_output):
actual_output = run_in_house_agent(user_input)

test_case = LLMTestCase(
    input=user_input,
    actual_output=actual_output,
    expected_output=expected_output
)

# Остановка сборки при нехватке баллов до установленного порога
assert_test(test_case, [correctness_metric, TaskCompletionMetric(threshold=0.8)])

`

Система оценки связывается в три этапа:

1. Создание датасета для оценки: Формируется набор основных доменных вопросов и правильных ответов, задается допустимый пороговый балл (Threshold).
2. Интеграция с GitHub Actions: Настраивается экшен, чтобы при создании PR после редактирования кода автоматически запускалась команда deepeval test run.
3. Применение теневого тестирования (Shadow Testing): Перед слиянием с веткой main запросы из реального трафика асинхронно дублируются и отправляются для сравнения качества ответов и задержки с предыдущей версией, после чего выполняется финальный деплой.

`yaml
name: AI Agent Evaluation Gate

on:
pull_request:
branches: [ main ]

jobs:
eval-gate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install Dependencies
run: |
pip install poetry
poetry install
- name: Run DeepEval Suite
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
poetry run deepeval test run tests/test_evals.py
`

Маскирование и контроль доступа

Чтобы предотвратить утечку внутренних данных компании, необходимы гардрейлы (Guardrails), обезличивающие персональные данные (PII) до того, как запрос к API уйдет вовне. Подключение движка Microsoft Presidio на шлюзе позволяет автоматически маскировать регистрационные номера граждан, email, номера телефонов и табельные номера сотрудников.

`python
from presidio_analyzer import AnalyzerEngine, PatternRecognizer
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

1. Регистрация Analyzer Engine и правил для табельных номеров

analyzer = AnalyzerEngine()
employee_id_recognizer = PatternRecognizer(
supported_entity="EMPLOYEE_ID",
regex="EMP-[0-9]{6}",
score=0.95
)
analyzer.registry.add_recognizer(employee_id_recognizer)

2. Инициализация Anonymizer Engine

anonymizer = AnonymizerEngine()

def sanitize_user_prompt(raw_prompt: str) -> str:
results = analyzer.analyze(
text=raw_prompt,
entities=["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "EMPLOYEE_ID"],
language="en"
)

anonymized_result = anonymizer.anonymize(
    text=raw_prompt,
    analyzer_results=results,
    operators={
        "DEFAULT": OperatorConfig("replace", {"new_value": "<REDACTED>"}),
        "EMPLOYEE_ID": OperatorConfig("mask", {"chars_to_mask": 6, "masking_char": "*", "from_end": True})
    }
)
return anonymized_result.text

`

При подключении поиска RAG также необходимо принять меры, чтобы извлекались только документы, соответствующие уровню доступа пользователя. Векторные базы данных, такие как Qdrant или Pinecone, обрабатывают разграничение прав с помощью фильтрации метаданных (Metadata Filtering) при поиске по сходству.

`python
from qdrant_client import QdrantClient
from qdrant_client.http import models

client = QdrantClient(host="localhost", port=6333)

def search_documents_with_rbac(query_vector: list, user_department: str, user_clearance_level: int):
search_result = client.search(
collection_name="enterprise_knowledge_base",
query_vector=query_vector,
query_filter=models.Filter(
must=[
models.FieldCondition(
key="department",
match=models.MatchValue(value=user_department)
),
models.FieldCondition(
key="security_level",
range=models.Range(lte=user_clearance_level)
)
]
),
limit=5
)
return search_result
`

Базовая защита от промпт-инъекций заключается в полном разделении каналов системного промпта и пользовательского ввода. Опасные операции, редактирующие базу данных или вызывающие внешние API, должны либо проходить процедуру ручного подтверждения человеком (Human-in-the-Loop), либо ограничиваться в правах выполнения внутри изолированного песочницы (sandbox).

Жизненный цикл инфраструктуры и поэтапное обеспечение целостности

Чтобы заменить SaaS-модули и продолжать эксплуатировать систему внутри компании, требуются регулярные процедуры проверки.

Если вы выбираете между собственным сервингом моделей (на базе vLLM) и подпиской на внешние API, сделайте расчет на основе ежедневного трафика.

Ежемесячная аренда инстанса AWS EC2 g5.2xlarge с 1 GPU NVIDIA A10G составляет около $880. В то же время стоимость API GPT-4o находится на уровне $2.50 за 1 млн входных токенов и $10.00 за 1 млн выходных токенов. Если ежедневный трафик составляет менее 50 млн токенов, с учетом трудозатрат DevOps-инженера и фиксированных расходов на GPU выгоднее использовать подписку на API. Переходить на собственный сервинг на базе vLLM имеет смысл только тогда, когда суточный трафик превышает 100 млн токенов или когда критически важна полная изоляция внутри корпоративной сети.

Для подготовки к сбоям основной модели создается 4-уровневая структура резервного роутинга.

┌─────────────────────────────────────────────────────────────────────────┐ │ 4-Tier Graceful Degradation Architecture │ │ │ │ [Tier 1] Primary High-Performance Model (e.g., GPT-4o) │ │ │ │ │ ▼ (API Failure / Timeout / Circuit Breaker) │ │ [Tier 2] Lightweight Routing Model (e.g., GPT-4o-mini / On-Prem vLLM) │ │ │ │ │ ▼ (Continuous Outage) │ │ [Tier 3] Deterministic Regex & SQL Rule Engine │ │ │ │ │ ▼ (Unrecoverable Error) │ │ [Tier 4] Static Error Message & Async Admin Ticket Generation │ └─────────────────────────────────────────────────────────────────────────┘

Порядок стабилизации эксплуатации

Работу по отладке системы кастомных агентов рекомендуется разбивать на 90-дневные этапы.

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