Цепочки инструментов в продакшене: когда агенты съедают все токены и становятся непредсказуемыми

Почему ваши агенты начинают врать на 15-м шаге

Вы построили красивую цепочку инструментов. Агент А делает запрос к API, агент Б анализирует ответ, агент В форматирует результат. В демо работает идеально. В продакшене - катастрофа.

На 15-м вызове цепочка внезапно возвращает ерунду. Токены улетают в космос. Латентность растет. А самое страшное - вы не можете воспроизвести ошибку. Потому что агенты недетерминированы.

Три главные ошибки, которые вы делаете прямо сейчас

Посмотрите на свой код. Уверен, вы найдете хотя бы одну из этих проблем.

1. Слепое наращивание контекста

Каждый шаг цепочки добавляет в контекст свои рассуждения. К 10-му шагу у вас уже 5000 токенов reasoning. Модель начинает терять нить.

# КАК НЕ НАДО ДЕЛАТЬ
context = ""
for step in chain:
    reasoning = llm.generate(f"Подумай о {step}...")
    context += reasoning  # Остановись, убийца токенов!
    result = llm.generate(f"Сделай {step} с учетом: {context}")

Этот код съедает бюджет быстрее, чем команда маркетинга на корпоративе.

2. Полная свобода выбора инструментов

Даете агенту 15 инструментов и говорите "выбирай сам". Он выбирает. Случайно. Иногда берет search_api, иногда google_search. Результаты разные. Отладка - ад.

3. Отсутствие четких конечных состояний

Цепочка выполняется "пока не надоест". Агент решает, когда остановиться. Иногда останавливается рано. Иногда никогда. Классическая проблема когда промпт длиннее мозга - только на уровне цепочки.

Детерминизм: как заставить агентов работать предсказуемо

Детерминизм в цепочках инструментов - это не про одинаковые ответы. Это про предсказуемое поведение. Если на входе Х, цепочка всегда делает шаги А→Б→В. Никаких сюрпризов.

Шаг 1: Жесткий роутинг вместо свободного выбора

Забудьте про "агент сам решит, какой инструмент использовать". В продакшене вы решаете за него.

# Вместо этого:
# tool = llm.choose_tool(task, available_tools)  # Случайный выбор

# Делайте так:
def route_task(task):
    if "search" in task and "realtime" in task:
        return "search_api"
    elif "search" in task and "historical" in task:
        return "database_lookup"
    else:
        return "fallback_tool"

Да, это менее "умно". Зато предсказуемо. Для сложных решений используйте семантический роутинг, но с четкими правилами.

Шаг 2: Фиксированная длина цепочки

Никаких "продолжай, пока не решишь задачу". Задайте максимальное количество шагов. Когда лимит исчерпан - цепочка падает с четкой ошибкой.

MAX_STEPS = 5
def execute_chain(input_data):
    result = input_data
    for step in range(MAX_STEPS):
        result = process_step(result)
        if is_final_result(result):
            return result
    raise ChainTimeoutError(f"Цепочка превысила {MAX_STEPS} шагов")

Шаг 3: Валидация выходов каждого шага

Каждый шаг должен возвращать данные в строго определенном формате. Если формат нарушен - цепочка останавливается сразу, а не на 10 шагов позже с криптической ошибкой.

from pydantic import BaseModel

class StepOutput(BaseModel):
    data: dict
    next_step: str
    metadata: dict

def validated_step(input_data):
    raw_output = llm.process(input_data)
    try:
        return StepOutput.model_validate_json(raw_output)
    except ValidationError as e:
        log_error(f"Шаг вернул невалидные данные: {e}")
        raise StepValidationError(f"Ошибка валидации: {e}")

Оптимизация токенов: перестаньте платить за reasoning

Reasoning токены - это роскошь для демо. В продакшене они слишком дороги. Особенно в длинных цепочках.

1Отделите reasoning от передачи данных

Агент думает внутри шага. На выходе - только результат. Не тащите мыслительный процесс в следующий шаг.

# Старый подход (плохо):
thought = "Я думаю, что нужно сделать X, потому что Y и Z..."
output = f"{thought}\nРезультат: {data}"

# Новый подход (хорошо):
# Внутри шага:
thought = llm.reason(input)  # Эти токены никуда не уходят
result = extract_result(thought)  # Только суть
return result  # 50 токенов вместо 500

2Используйте семантическое кеширование

Если шаг А уже обрабатывал похожий запрос - не вызывайте LLM снова. Используйте кеш.

from sentence_transformers import SentenceTransformer
import redis

model = SentenceTransformer('all-MiniLM-L6-v2')
cache = RedisCache()

def cached_step(input_text):
    embedding = model.encode(input_text)
    cached = cache.get_similar(embedding, threshold=0.95)
    if cached:
        return cached
    
    result = llm.process(input_text)
    cache.store(embedding, result)
    return result

Это снижает вызовы на 40-60% в реальных продакшен-нагрузках. Особенно для повторяющихся операций.

3Сжимайте промежуточные результаты

Когда нужно передать большой объем данных между шагами, не кидайте сырой текст. Извлекайте суть.

def compress_for_next_step(data):
    """Сжимает данные для передачи следующему агенту"""
    if len(data) > 1000:  # Слишком много для передачи
        summary = llm.summarize(data, max_tokens=200)
        return {
            "compressed": True,
            "summary": summary,
            "full_data_ref": "s3://bucket/key"  # Ссылка на полные данные
        }
    return {"compressed": False, "data": data}

Архитектурные паттерны для продакшена

Оркестрация агентов - это не про гибкость. Это про контроль. Вот три паттерна, которые работают под нагрузкой.

Паттерн 1: Диспетчер с конечным автоматом

Каждый шаг - состояние. Переходы между состояниями жестко заданы. Никаких неожиданных прыжков.

class ChainStateMachine:
    states = {
        "start": ["parse", "validate"],
        "parse": ["enrich", "error"],
        "enrich": ["format", "error"],
        "format": ["end"],
        "error": ["end"],
        "end": []
    }
    
    def transition(self, current_state, action):
        allowed = self.states.get(current_state, [])
        if action not in allowed:
            raise InvalidTransition(f"Нельзя перейти из {current_state} в {action}")
        return action

Паттерн 2: Декларативные workflow

Опишите цепочку как YAML или JSON. Исполняющая система следует описанию, не оставляя места для импровизации.

# workflow.yaml
name: "document_processing"
steps:
  - id: "extract_text"
    tool: "pdf_extractor"
    timeout: 30s
    retries: 2
    
  - id: "analyze_content"
    tool: "llm_analyzer"
    depends_on: ["extract_text"]
    parameters:
      model: "gpt-4o-mini"
      max_tokens: 500
    
  - id: "format_output"
    tool: "json_formatter"
    depends_on: ["analyze_content"]
    output_schema: "output_schema.json"

Используйте готовые системы вроде Prefect или Dagster. Не изобретайте велосипед.

Паттерн 3: Двухфазное выполнение

Сначала планирование (какие шаги выполнить), потом исполнение (строго по плану). План можно кешировать для одинаковых запроссов.

class TwoPhaseChain:
    def plan(self, input_data):
        """Фаза 1: Создать план выполнения"""
        plan = llm.generate_plan(input_data)
        self.validate_plan(plan)  # Проверка на безопасность
        self.cache_plan(input_data.hash(), plan)
        return plan
    
    def execute(self, plan):
        """Фаза 2: Выполнить план без отклонений"""
        results = []
        for step in plan["steps"]:
            result = self.execute_step(step)
            results.append(result)
            if not step.get("continue_on_error", False) and result.error:
                break  # Жесткое прерывание при ошибке
        return results

Отладка цепочек: когда всё пошло не так

Цепочка упала в 3 утра. Логи есть. Что смотреть в первую очередь?

1. Токен-лимиты каждого шага - кто превысил квоту?
2. Валидацию выходов - на каком шаге данные стали некорректными?
3. Время выполнения - какой шаг тормозит всю цепочку?
4. Состояние кеша - почему не сработало кеширование?

Инструменты для отладки:

• LangSmith - трассировка вызовов, но дорого для больших объемов
• Custom dashboard - свой дашборд с метриками по каждому шагу
• Structural logging - логи в JSON с correlation_id для отслеживания цепочек

Чего ждать дальше: эволюция или революция?

Текущий подход к цепочкам инструментов - это костыль. Умные системы должны планировать сложные задачи, но мы не можем им доверить это из-за недетерминизма.

Будущее за гибридными системами:

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

Пока индустрия не решит фундаментальную проблему детерминизма, мы будем вынуждены ограничивать "интеллект" наших агентов. И это правильно. Надежность важнее умности.

Ваша цепочка инструментов не должна быть умнее вашей возможности её отладить. Сначала сделайте её предсказуемой. Потом - умной. В такой последовательности.

И если вы думаете, что можете обойтись без этих ограничений - вспомните последний инцидент в 3 утра. Вспомнили? Теперь идите настраивать валидацию выходов.