Работаю над внедрением 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
