Как правильно использовать суб-агентов в AI-разработке: 3 реальных сценария

Проблема: почему ваш LLM "тупит" и дорого стоит

Если вы читали нашу предыдущую статью "Production-ready AI-агент с нуля", то наверняка сталкивались с классической проблемой: ваш основной агент становится медленным, дорогим и непредсказуемым, когда ему приходится обрабатывать слишком много контекста. Это особенно актуально в сложных системах, где агент должен анализировать десятки файлов, принимать решения на основе множества факторов и поддерживать длинные диалоги.

Решение: архитектура с суб-агентами

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

Сценарий 1: Контекст-менеджер для фильтрации файлов

Представьте: пользователь загружает 10 документов и задает вопрос, относящийся только к 2 из них. Наивный подход — скормить все 10 файлов основному агенту. Умный подход — использовать суб-агент для предварительной фильтрации.

1 Создаем базовый класс суб-агента

from typing import List, Dict, Any
import json

class SubAgent:
    """Базовый класс для всех суб-агентов"""
    
    def __init__(self, model_client, system_prompt: str):
        self.client = model_client
        self.system_prompt = system_prompt
    
    def process(self, input_data: Any) -> Any:
        """Основной метод обработки"""
        raise NotImplementedError("Должен быть реализован в дочерних классах")
    
    def _call_llm(self, prompt: str) -> str:
        """Вспомогательный метод для вызова LLM"""
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": prompt}
        ]
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",  # Используем легкую модель для суб-агентов
            messages=messages,
            temperature=0.1,
            max_tokens=1000
        )
        return response.choices[0].message.content

2 Реализуем контекст-менеджера

class ContextManagerAgent(SubAgent):
    """Суб-агент для фильтрации и ранжирования документов"""
    
    def __init__(self, model_client):
        system_prompt = """Ты — контекст-менеджер. Твоя задача анализировать вопрос пользователя 
        и определять, какие документы из предоставленного списка релевантны для ответа.
        Возвращай JSON с двумя полями: relevant_docs (индексы релевантных документов) 
        и relevance_score (оценка релевантности от 0 до 1 для каждого документа)."""
        super().__init__(model_client, system_prompt)
    
    def process(self, question: str, documents: List[str]) -> Dict[str, Any]:
        """Фильтрует документы на основе вопроса"""
        prompt = f"""Вопрос пользователя: {question}
        
        Документы для анализа (пронумерованы от 0 до {len(documents)-1}):
        {self._format_documents(documents)}
        
        Верни JSON в указанном формате."""
        
        response = self._call_llm(prompt)
        
        try:
            result = json.loads(response)
            # Фильтруем только документы с высокой релевантностью
            filtered_docs = []
            for idx, score in enumerate(result.get('relevance_score', [])):
                if score > 0.3:  # Порог релевантности
                    filtered_docs.append({
                        'index': idx,
                        'content': documents[idx],
                        'score': score
                    })
            return {
                'filtered_documents': filtered_docs,
                'total_docs': len(documents),
                'filtered_count': len(filtered_docs)
            }
        except json.JSONDecodeError:
            # Fallback: возвращаем все документы, если парсинг не удался
            return {
                'filtered_documents': [
                    {'index': i, 'content': doc, 'score': 1.0}
                    for i, doc in enumerate(documents)
                ],
                'total_docs': len(documents),
                'filtered_count': len(documents)
            }
    
    def _format_documents(self, documents: List[str]) -> str:
        """Форматирует документы для промпта"""
        formatted = []
        for i, doc in enumerate(documents):
            # Берем только первые 500 символов для экономии токенов
            preview = doc[:500] + ("..." if len(doc) > 500 else "")
            formatted.append(f"[Документ {i}] {preview}")
        return "\n\n".join(formatted)

3 Используем в основном агенте

class MainAgent:
    def __init__(self, model_client):
        self.client = model_client
        self.context_manager = ContextManagerAgent(model_client)
    
    def answer_question(self, question: str, documents: List[str]) -> str:
        """Основной метод ответа на вопрос"""
        
        # Шаг 1: Фильтрация документов через суб-агент
        context_result = self.context_manager.process(question, documents)
        filtered_docs = [doc['content'] for doc in context_result['filtered_documents']]
        
        print(f"Суб-агент отфильтровал {context_result['filtered_count']} из {context_result['total_docs']} документов")
        
        # Шаг 2: Основной агент работает только с релевантным контекстом
        context_text = "\n\n".join(filtered_docs)
        
        prompt = f"""На основе следующих документов ответь на вопрос:
        
        Контекст:
        {context_text}
        
        Вопрос: {question}
        
        Ответ должен быть точным и основанным только на предоставленном контексте."""
        
        messages = [
            {"role": "system", "content": "Ты — helpful assistant."},
            {"role": "user", "content": prompt}
        ]
        
        response = self.client.chat.completions.create(
            model="gpt-4",  # Основная мощная модель
            messages=messages,
            temperature=0.7
        )
        
        return response.choices[0].message.content

Сценарий 2: Пре-процессор для сложных промптов

В статье "Agent Skills" мы обсуждали, как важно структурировать инструкции. Суб-агент может превращать расплывчатые запросы пользователя в структурированные задания для основного агента.

class PromptPreprocessorAgent(SubAgent):
    """Суб-агент для структурирования и уточнения запросов"""
    
    def __init__(self, model_client):
        system_prompt = """Ты — пре-процессор промптов. Твоя задача:
        1. Анализировать нечеткие запросы пользователя
        2. Определять истинный intent (намерение)
        3. Разбивать сложные задачи на подзадачи
        4. Формулировать четкие инструкции для основного агента
        
        Возвращай JSON с полями: intent, subtasks, clear_instruction."""
        super().__init__(model_client, system_prompt)
    
    def process(self, user_query: str) -> Dict[str, Any]:
        prompt = f"""Обработай следующий запрос пользователя:
        
        Запрос: {user_query}
        
        Верни структурированный JSON."""
        
        response = self._call_llm(prompt)
        
        try:
            return json.loads(response)
        except:
            # Fallback
            return {
                'intent': 'general_query',
                'subtasks': [user_query],
                'clear_instruction': user_query
            }


# Пример использования
preprocessor = PromptPreprocessorAgent(client)

# Расплывчатый запрос пользователя
vague_query = "Сделай что-то с этими данными, чтобы было красиво и информативно"

# Суб-агент структурирует запрос
structured = preprocessor.process(vague_query)
print(f"Intent: {structured['intent']}")  # Например: "data_visualization"
print(f"Subtasks: {structured['subtasks']}")  # Например: ["analyze_data", "create_charts", "write_summary"]

Сценарий 3: Валидатор и санитайзер ответов

После того как основной агент сгенерировал ответ, но перед тем как отдать его пользователю, суб-агент может проверить качество, безопасность и соответствие требованиям.

class ValidationAgent(SubAgent):
    """Суб-агент для валидации ответов основного агента"""
    
    def __init__(self, model_client, validation_rules: List[str]):
        system_prompt = f"""Ты — валидатор ответов. Проверяй ответы по следующим критериям:
        {chr(10).join(validation_rules)}
        
        Возвращай JSON с полями: is_valid (bool), issues (список проблем), 
        corrected_answer (исправленная версия, если есть проблемы)."""
        super().__init__(model_client, system_prompt)
        self.rules = validation_rules
    
    def process(self, original_answer: str, context: str = None) -> Dict[str, Any]:
        prompt = f"""Проверь следующий ответ:
        
        Ответ для проверки:
        {original_answer}
        """
        
        if context:
            prompt += f"\n\nКонтекст (для проверки соответствия):\n{context}"
        
        response = self._call_llm(prompt)
        
        try:
            result = json.loads(response)
            return result
        except:
            return {
                'is_valid': True,
                'issues': [],
                'corrected_answer': original_answer
            }


# Пример правил валидации для финансового ассистента
financial_rules = [
    "1. Ответ должен содержать disclaimer о нефинансовых рекомендациях",
    "2. Не должно быть гарантий будущих доходов",
    "3. Все числа должны быть проверены на арифметическую корректность",
    "4. Ответ не должен содержать персональных финансовых советов"
]

validator = ValidationAgent(client, financial_rules)

# Допустим, основной агент сгенерировал такой ответ:
agent_response = "Инвестируйте в акции компании XYZ, они гарантированно вырастут на 50% в следующем году!"

# Валидатор найдет проблемы
validation_result = validator.process(agent_response)
print(f"Ответ валиден: {validation_result['is_valid']}")  # False
print(f"Проблемы: {validation_result['issues']}")  # ['нет disclaimer', 'есть гарантии доходов']

Архитектурные паттерны для работы с суб-агентами

Паттерн 1: Pipeline (Конвейер)

Суб-агенты выстраиваются в цепочку, где выход одного становится входом для другого. Идеально для сложных multi-step задач.

class AgentPipeline:
    """Оркестратор конвейера суб-агентов"""
    
    def __init__(self, agents: List[SubAgent]):
        self.agents = agents
    
    def execute(self, initial_input: Any) -> Any:
        current_result = initial_input
        
        for i, agent in enumerate(self.agents):
            print(f"Выполняется агент {i+1}/{len(self.agents)}")
            current_result = agent.process(current_result)
            
            # Можно добавить проверки после каждого шага
            if isinstance(current_result, dict) and 'error' in current_result:
                print(f"Ошибка на шаге {i+1}: {current_result['error']}")
                break
        
        return current_result


# Пример конвейера для обработки документа
pipeline = AgentPipeline([
    ContextManagerAgent(client),      # 1. Фильтрация документов
    PromptPreprocessorAgent(client),  # 2. Структурирование запроса
    # Основной агент выполнялся бы здесь
    ValidationAgent(client, rules)    # 3. Валидация ответа
])

Паттерн 2: Router (Маршрутизатор)

Специальный суб-агент анализирует входной запрос и решает, какой другой суб-агент (или комбинацию) использовать.

class RouterAgent(SubAgent):
    """Суб-агент для маршрутизации запросов"""
    
    def __init__(self, model_client, available_agents: Dict[str, SubAgent]):
        system_prompt = """Ты — интеллектуальный роутер. Анализируй запрос пользователя 
        и определяй, какие специализированные агенты нужны для его обработки.
        Возвращай JSON с полем 'required_agents' — список названий агентов."""
        super().__init__(model_client, system_prompt)
        self.agents = available_agents
    
    def process(self, query: str) -> Dict[str, Any]:
        prompt = f"Запрос: {query}"
        response = self._call_llm(prompt)
        
        try:
            result = json.loads(response)
            agent_names = result.get('required_agents', [])
            
            # Выбираем только доступных агентов
            selected_agents = []
            for name in agent_names:
                if name in self.agents:
                    selected_agents.append(self.agents[name])
                else:
                    print(f"Предупреждение: агент '{name}' не найден")
            
            return {
                'selected_agents': selected_agents,
                'routing_logic': result
            }
        except:
            # Fallback: используем все доступные агенты
            return {
                'selected_agents': list(self.agents.values()),
                'routing_logic': {'fallback': True}
            }

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

Частые ошибки и как их избежать

1. Бесконечная рекурсия суб-агентов

   Ошибка: Суб-агент вызывает другой суб-агент, который вызывает третий, и так до бесконечности.

   Решение: Установить максимальную глубину вложенности и добавить circuit breaker.

2. Потеря контекста между агентами

   Ошибка: Каждый суб-агент работает изолированно, не передавая важный контекст.

   Решение: Использовать shared context object, который передается через всю цепочку.

3. Слишком много маленьких агентов

   Ошибка: Создание микро-агента для каждой мелочи усложняет систему.

   Решение: Объединять связанные функции в одного агента. Правило: один агент = одна ответственность, но не слишком узкая.

4. Отсутствие fallback-механизмов

   Ошибка: Если суб-агент падает, вся система останавливается.

   Решение: Добавлять try-catch и default поведения для каждого суб-агента.

FAQ

Как выбрать, какие задачи выносить в суб-агенты?

Используйте правило "стоимость vs сложность". Если задача:

• Требует много контекста (фильтрация документов) — выносите
• Может быть выполнена маленькой моделью (классификация, структурирование) — выносите
• Требует специализированных знаний (валидация по правилам) — выносите
• Является критически важной для качества (финальная проверка) — выносите

Суб-агенты vs инструменты (tools) — в чем разница?

Инструменты — это детерминированные функции (поиск в базе, вычисления). Суб-агенты — это LLM-based помощники, которые могут принимать решения, анализировать и генерировать контент. В статье "Агентные workflow на практике" мы подробно разбираем эту разницу.

Как тестировать системы с суб-агентами?

Тестируйте каждого суб-агента изолированно с unit-тестами, затем интеграционные тесты для цепочек. Используйте моки для LLM-вызовов в тестах. Мониторьте стоимость и latency в production.

Заключение

Суб-агенты — это не просто "еще один слой абстракции". Это архитектурный паттерн, который позволяет:

• Сократить стоимость эксплуатации AI-систем на 30-70%
• Увеличить качество ответов за счет специализации
• Упростить отладку и мониторинг
• Создать более устойчивые к ошибкам системы

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