Работаю над внедрением LLM-агентов в бизнес-процессы. За последний год видел десятки проектов — своих и чужих.
Паттерн один и тот же: красивый PoC → попытка масштабирования → провал.
MIT Media Lab в 2025 году подтвердил это цифрами: 95% пилотов по GenAI не дают измеримого влияния на P&L. Gartner добавляет: 40%+ проектов по агентному ИИ отменят к 2027.
Причина не в моделях. GPT-4, Claude, GigaChat — все работают нормально.
Проблема на стыке технологий и инфраструктуры. И о ней интеграторы предпочитают молчать, т.к. её решение — это 70-80% бюджета проекта.
Проблема 1: Human-in-the-loop без нормального интерфейса
Агент не работает в изоляции. Ему нужно взаимодействовать с людьми — запрашивать данные, уточнять контекст, эскалировать решения.
Человек может написать коллеге: «Слушай, а что там с договором по Газпрому?»
Агент так не умеет. Ему нужен формализованный канал, структурированные запросы, чёткие протоколы.
Типичный антипаттерн в PoC:
def process_request(request):
response = llm.complete(request)
if response.confidence < 0.8:
# "Потом добавим эскалацию"
return response # Агент просто возвращает неуверенный ответ
И это «потом» никогда не наступает...
Что нужно на самом деле:
class HumanInTheLoop:
def __init__(self, notification_service, timeout_hours=4):
self.notifications = notification_service
self.timeout = timeout_hours
self.escalation_queue = asyncio.Queue()
async def request_human_input(
self,
context: dict,
question: str,
assignee: str,
priority: Priority = Priority.NORMAL
) -> HumanResponse:
ticket_id = await self.create_ticket(context, question, assignee)
# Отправляем в канал (Slack/TG/email)
await self.notifications.send(
to=assignee,
message=self.format_request(context, question),
ticket_id=ticket_id,
reply_options=["approve", "reject", "need_info"]
)
# Ждём ответа с таймаутом
try:
response = await asyncio.wait_for(
self.wait_for_response(ticket_id),
timeout=self.timeout * 3600
)
except asyncio.TimeoutError:
# Эскалация на следующий уровень
await self.escalate(ticket_id, priority.next_level())
response = await self.wait_for_response(ticket_id)
return response
Это не «добавим потом». Это core-функциональность.
Без неё агент зависнет на первом же нестандартном случае.
Проблема 2: Tacit knowledge нигде не записано
В компании есть неявные правила.
«К Петрову лучше идти после обеда». «Если сумма больше миллиона — сначала к финдиру». «Договоры с ООО на УСН проверяем тщательнее».
Люди это знают. Агент — нет.
И оказывается, это знание нигде не записано. Его нужно извлечь, структурировать, загрузить в систему.
Минимальная структура:
# rules/contract_approval.yaml
domain: contract_approval
version: 2.1
routing_rules:
- condition:
contract_type: "аренда"
amount_rub: ">500000"
route_to: "legal_senior"
sla_hours: 48
- condition:
contractor_type: "ИП"
tax_system: "УСН"
additional_checks:
- "tax_risk_assessment"
- "bankruptcy_check"
- condition:
time: "friday_after_15"
urgency: "not_critical"
action: "postpone_to_monday"
reason: "Петров не отвечает в пятницу вечером :)"
implicit_rules:
- name: "million_threshold"
description: "Суммы >1M требуют визы финдира"
extracted_from: "interview_cfo_2025-01-10"
confidence: 0.95
Процесс извлечения знаний — это отдельная работа на 3-6 недель:
Интервью с ключевыми сотрудниками
Анализ истории решений в CRM/почте
Разбор инцидентов («почему этот договор согласовывали 2 недели?»)
Валидация с экспертами
Версионирование
И это не разовая история. Это процесс, который должен работать постоянно.
Проблема 3: Интеграция без разрушения
У компании есть работающие процессы. Кривые, с костылями, но работающие.
Агент должен встроиться, не сломав. Или процессы нужно перестроить под агента.
Первое — сложно. Второе — дорого и рискованно.
Обычно об этом узнают уже после старта проекта...
Паттерн «Shadow Mode» — агент работает параллельно с человеком, но решений не принимает:
class ShadowModeAgent:
"""
Агент работает параллельно с человеком.
Не принимает решений, только предлагает и учится.
"""
def __init__(self, production_system, learning_period_days=30):
self.prod = production_system
self.learning_period = learning_period_days
self.decisions_log = []
async def process(self, task):
# Агент делает своё предсказание
agent_decision = await self.make_decision(task)
# Но решение принимает человек (как раньше)
human_decision = await self.prod.get_human_decision(task)
# Логируем расхождения
if agent_decision != human_decision:
await self.log_discrepancy(
task=task,
agent=agent_decision,
human=human_decision
)
self.update_accuracy_metrics(agent_decision, human_decision)
return human_decision # В прод идёт человеческое решение
def is_ready_for_production(self) -> bool:
return (
self.accuracy > 0.95 and
self.days_in_shadow >= self.learning_period and
self.critical_errors == 0
)
Shadow mode на 30 дней — это минимум. Иначе первый же косяк агента в проде парализует процесс.
Проблема 4: Observability — не «логи», а полная картина
«У нас есть логи» — это не observability.
Нужна полная трассировка: что агент сделал, почему, на основании каких данных, с какой уверенностью.
Без этого отладка превращается в гадание на кофейной гуще.
from opentelemetry import trace
tracer = trace.get_tracer("ai-agent")
class ObservableAgent:
async def process_request(self, request: Request) -> Response:
with tracer.start_as_current_span("agent_process") as span:
span.set_attribute("request.id", request.id)
# Шаг 1: Классификация
with tracer.start_as_current_span("classify"):
classification = await self.classify(request)
span.set_attribute("classification.confidence", classification.score)
# Шаг 2: Retrieval
with tracer.start_as_current_span("retrieve_context"):
context = await self.retrieve(request, classification)
span.set_attribute("context.documents_count", len(context.docs))
# Шаг 3: Generation
with tracer.start_as_current_span("generate"):
response = await self.generate(request, context)
span.set_attribute("generation.tokens_used", response.tokens)
span.set_attribute("generation.model", response.model)
return response
Что должно быть:
Traces — полный путь запроса через все компоненты
Metrics — latency, throughput, error rate, confidence distribution
Logs — структурированные, с correlation_id
Alerts — на аномалии в confidence, на рост ошибок
Проблема 5: Кто отвечает за ошибки
Агент согласовал договор с ошибкой. Компания потеряла деньги.
Кто виноват?
Это не философский вопрос. Это юридический и организационный.
Модель «Supervised Autonomy» — агент автономен в рамках, но есть supervisor:
class SupervisedAgent:
def __init__(self, autonomy_config: AutonomyConfig):
self.config = autonomy_config
async def process(self, task: Task) -> Decision:
decision = await self.make_decision(task)
# Проверяем границы автономии
if self.is_within_autonomy(task, decision):
await self.log_autonomous_decision(task, decision)
return decision
else:
# Требуется подтверждение
return await self.request_approval(
task=task,
proposed_decision=decision,
reason=self.get_escalation_reason(task, decision)
)
def is_within_autonomy(self, task: Task, decision: Decision) -> bool:
return (
task.amount <= self.config.max_autonomous_amount and
task.risk_score <= self.config.max_risk_score and
decision.confidence >= self.config.min_confidence and
task.type in self.config.autonomous_task_types
)
Без чёткой модели ответственности первый же провал парализует использование агента.
Проблема 6: Graceful degradation
API OpenAI лёг. Или лимиты кончились. Или сеть упала.
Что делает агент? Правильно — ничего.
А процесс, который от него зависит, встаёт. И люди сидят.
class ResilientAgent:
def __init__(self):
self.primary_llm = OpenAIClient()
self.fallback_llm = GigaChatClient() # Локальный fallback
self.circuit_breaker = CircuitBreaker(
failure_threshold=5,
recovery_timeout=60
)
self.queue = PersistentQueue()
async def process(self, task: Task) -> Response:
try:
if self.circuit_breaker.is_open("primary"):
raise CircuitOpen()
response = await asyncio.wait_for(
self.primary_llm.complete(task),
timeout=30
)
self.circuit_breaker.record_success("primary")
return response
except (Timeout, APIError, CircuitOpen) as e:
self.circuit_breaker.record_failure("primary")
# Пробуем fallback
try:
return await self.fallback_llm.complete(task)
except Exception:
# Последний resort
if task.can_be_delayed:
await self.queue.put(task)
return Response(status="queued")
else:
return await self.escalate_to_human(task)
Без fallback и circuit breaker один сбой API → весь отдел не работает.
Итого
Сама интеграция LLM — это 20-30% проекта. Остальное:
Human-in-the-loop система — 15-20%, 2-4 недели
Извлечение знаний — 20-25%, 3-6 недель
Интеграция и миграция — 10-15%, 2-4 недели
Observability — 10-15%, 1-2 недели
Governance — 5-10%, 1-2 недели
Отказоустойчивость — 10-15%, 2-3 недели
Если интегратор обещает «внедрить за 2 недели» — он либо не понимает задачу, либо делает PoC, который развалится в проде.
В следующей статье — подробнее про Human-in-the-loop: архитектура, паттерны, типичные ошибки.
Анатолий Лапков, внедряю LLM-агентов в бизнес. Telegram: @futex_ai
