Петля вызовов кастомного AI-агента после завершения демо и способы контроля затрат на токены
makedream٢٨ يوليو ٢٠٢٦
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
Взять фреймворк и создать 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% затрат, распределяя запросы между дорогостоящими и бюджетными моделями в зависимости от сложности вопроса.
Семантическое кэширование строится в следующем порядке:
Чтобы кардинально заблокировать бесконечные циклы, паттерн 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, физически предотвращая вход в цикл из-за галлюцинаций.
Оставлять промпты внутри кода нельзя: вы не сможете отловить регрессионные сбои (Regression Failure), когда после изменения модели перестает работать даже то, что раньше функционировало исправно. Промпты следует вынести из кода Python и управлять ими в отдельных файлах YAML.
`yaml
name: "agent_reasoning"
version: "1.2.0"
model: "gpt-4o"
temperature: 0.1
messages:
Вынесенные таким образом промпты автоматически тестируются в 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
correctness_metric = GEval(
name="Точность и соответствие схеме",
criteria="Отвечает ли LLM точно на вопрос и полностью ли соответствует запрошенному формату JSON?",
evaluation_params=[SingleTurnParams.ACTUAL_OUTPUT, SingleTurnParams.EXPECTED_OUTPUT],
threshold=0.7
)
@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)])
`
Система оценки связывается в три этапа:
deepeval test run.`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
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)
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-модули и продолжать эксплуатировать систему внутри компании, требуются регулярные процедуры проверки.
| Периодичность | Элемент операционной проверки | Подробное содержание работ |
|---|---|---|
| Ежедневно | Уровень ошибок и расход токенов | Проверка уровня ошибок HTTP 5xx и расхода токенов по отделам на дашборде Langfuse |
| Ежедневно | История срабатываний Circuit Breaker | Сбор и исправление паттернов имен инструментов и аргументов сессий, прерванных из-за обнаружения циклов |
| Еженедельно | Извлечение запросов с неудачным RAG-поиском | Отбор запросов с баллом сходства ниже 0.6 и добавление их в тестовый набор правильных ответов |
| Еженедельно | Проверка ложных срабатываний маскирования PII | Выборочная проверка логов Presidio на предмет пропуска маскирования |
| Ежемесячно | Обновление критериев оценки CI/CD | Редактирование автоматических тест-кейсов с учетом изменений в бизнесе |
Если вы выбираете между собственным сервингом моделей (на базе 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-дневные этапы.
| Период | Цель внедрения | Конкретное содержание работ |
|---|---|---|
| Дни 1~30 | Прозрачность и защита данных | Установка движка маскирования Presidio и внедрение SDK логирования Langfuse |
| Дни 31~60 | Предотвращение циклов и сокращение затрат | Внедрение семантического кэширования Redis и подключение Circuit Breaker на Python |
| Дни 61~90 | Автоматизация проверки и контроль доступа | Управление промптами через Git, интеграция DeepEval с CI/CD, применение метаданных RBAC в Qdrant |
В первый месяц устанавливаются решения для наблюдения, чтобы предотвратить утечку персональных данных и сохранять все запросы в логах. Во второй месяц подключаются кэширование повторяющихся промптов и роутер предотвращения циклов, что отсекает непредвиденные расходы. В последний месяц завершается создание системы версионирования промптов и автоматического тестирования, что позволяет стабильно эксплуатировать систему без опасений за сбои агента при каждом деплое.