Глава 19. Команды AI-агентов

"Один агент хорош. Команда агентов — мощнее в разы."

19.1. Зачем нужны команды агентов

Ограничения одного агента

Проблема: Один агент пытается делать всё.

Пример:

Задача: Создать маркетинговую кампанию

Один агент должен:
- Исследовать рынок
- Создать контент
- Спланировать размещение
- Проанализировать конкурентов
- Написать тексты
- Создать визуалы

Результат: Агент делает всё "средне", ни в чём не преуспевая.

Решение: Команда специализированных агентов

Команда:
├─ Market Researcher (исследует рынок)
├─ Content Writer (пишет тексты)
├─ Designer (создаёт визуалы)
├─ Strategist (планирует кампанию)
└─ Analyst (анализирует результаты)

Каждый агент — эксперт в своей области.

Аналогия с командой разработки

Традиционная команда:

• Frontend разработчик
• Backend разработчик
• DevOps инженер
• QA тестировщик

Команда AI-агентов:

• Research Agent
• Coding Agent
• Testing Agent
• Review Agent

Принцип тот же: Специализация → Лучшие результаты.

19.2. Паттерны взаимодействия агентов

Паттерн 1: Sequential (последовательный)

Описание: Агенты работают по очереди, как конвейер.

Input → Agent A → Output A → Agent B → Output B → Agent C → Final Output

Пример: Создание статьи

# Агент 1: Researcher
research_result = researcher_agent.run("Research quantum computing")

# Агент 2: Writer (использует результат исследователя)
draft = writer_agent.run(f"Write article using: {research_result}")

# Агент 3: Editor
final_article = editor_agent.run(f"Edit and improve: {draft}")

Когда использовать:

• Чёткая последовательность шагов
• Каждый шаг зависит от предыдущего

Паттерн 2: Parallel (параллельный)

Описание: Агенты работают одновременно на одной задаче.

                ┌─ Agent A ─┐
Input ──────────┼─ Agent B ─┼─────> Combine → Output
                └─ Agent C ─┘

Пример: Исследование с разных точек зрения

# Все агенты работают параллельно
results = await asyncio.gather(
    technical_agent.run("Analyze from technical perspective"),
    business_agent.run("Analyze from business perspective"),
    user_agent.run("Analyze from user perspective")
)

# Комбинируем результаты
final_analysis = combiner_agent.run(f"Combine insights: {results}")

Когда использовать:

• Нужны разные точки зрения
• Задачи независимы друг от друга

Паттерн 3: Hierarchical (иерархический)

Описание: Есть главный агент (manager), который управляет подчинёнными.

        Manager Agent
           │
   ┌───────┼───────┐
   │       │       │
Agent A  Agent B  Agent C

Пример: Проект-менеджер и команда

class ManagerAgent:
    def __init__(self):
        self.team = {
            "researcher": ResearchAgent(),
            "coder": CodingAgent(),
            "tester": TestingAgent()
        }

    def delegate_task(self, task):
        # Manager разбивает задачу
        plan = self.plan(task)

        # Делегирует подзадачи
        research = self.team["researcher"].run(plan["research"])
        code = self.team["coder"].run(plan["code"], context=research)
        tests = self.team["tester"].run(plan["tests"], code=code)

        # Собирает результаты
        return self.finalize(research, code, tests)

Когда использовать:

• Сложные задачи, требующие координации
• Нужен контроль и планирование

Паттерн 4: Debate (дебаты)

Описание: Агенты спорят, улучшая решение через критику.

Agent A: "Solution X is best because..."
Agent B: "I disagree, Solution Y is better because..."
Agent A: "Good point, but consider..."
...
Final: Consensus or Vote

Пример: Код-ревью

# Agent A предлагает решение
solution = coder_agent.run("Implement user authentication")

# Agent B критикует
critique = reviewer_agent.run(f"Review this code: {solution}")

# Agent A улучшает
improved = coder_agent.run(f"Improve based on: {critique}")

# Agent B одобряет
final_review = reviewer_agent.run(f"Final review: {improved}")

Когда использовать:

• Нужна проверка решения
• Важна надёжность

19.3. Создание команды без фреймворка

Простая команда: Sequential

from openai import OpenAI
import json

client = OpenAI()

class Agent:
    def __init__(self, role, instructions, tools=None):
        self.role = role
        self.instructions = instructions
        self.tools = tools or []

    def run(self, task, context=None):
        system_message = f"""
        Role: {self.role}
        Instructions: {self.instructions}
        """

        if context:
            system_message += f"\nContext from previous steps: {context}"

        messages = [
            {"role": "system", "content": system_message},
            {"role": "user", "content": task}
        ]

        response = client.chat.completions.create(
            model="gpt-4",
            messages=messages
        )

        return response.choices[0].message.content

# Создание команды
researcher = Agent(
    role="Researcher",
    instructions="Research topics thoroughly. Provide facts and sources."
)

writer = Agent(
    role="Writer",
    instructions="Write engaging, clear articles. Use storytelling."
)

editor = Agent(
    role="Editor",
    instructions="Edit for clarity, grammar, and flow. Be concise."
)

# Sequential workflow
def create_article(topic):
    print(f"📊 Researcher: Researching {topic}...")
    research = researcher.run(f"Research: {topic}")

    print(f"✍️  Writer: Writing article...")
    draft = writer.run(f"Write article about {topic}", context=research)

    print(f"📝 Editor: Editing...")
    final = editor.run(f"Edit this article", context=draft)

    return final

# Использование
article = create_article("The future of AI agents")
print(article)

Parallel команда с async

import asyncio

async def parallel_research(topic):
    # Создаём агентов с разными перспективами
    technical_agent = Agent("Technical Expert", "Analyze technical aspects")
    business_agent = Agent("Business Analyst", "Analyze business implications")
    ethical_agent = Agent("Ethics Advisor", "Consider ethical implications")

    # Запускаем параллельно
    results = await asyncio.gather(
        asyncio.to_thread(technical_agent.run, f"Analyze {topic} technically"),
        asyncio.to_thread(business_agent.run, f"Analyze {topic} for business"),
        asyncio.to_thread(ethical_agent.run, f"Analyze {topic} ethically")
    )

    # Синтезируем результаты
    synthesizer = Agent("Synthesizer", "Combine multiple perspectives")
    final = synthesizer.run(
        f"Combine these perspectives: {json.dumps(results)}"
    )

    return final

# Использование
result = asyncio.run(parallel_research("AI in healthcare"))

19.4. Команды с LangChain

Sequential Chain

from langchain.chains import LLMChain, SequentialChain
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate

llm = ChatOpenAI(model="gpt-4")

# Chain 1: Research
research_template = """
Research the following topic and provide key facts:
Topic: {topic}

Research:"""

research_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate(input_variables=["topic"], template=research_template),
    output_key="research"
)

# Chain 2: Write
write_template = """
Write a blog post about {topic} using this research:
{research}

Article:"""

write_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate(
        input_variables=["topic", "research"],
        template=write_template
    ),
    output_key="draft"
)

# Chain 3: Edit
edit_template = """
Edit and improve this article:
{draft}

Final version:"""

edit_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate(input_variables=["draft"], template=edit_template),
    output_key="final"
)

# Combine chains
overall_chain = SequentialChain(
    chains=[research_chain, write_chain, edit_chain],
    input_variables=["topic"],
    output_variables=["research", "draft", "final"],
    verbose=True
)

# Run
result = overall_chain({"topic": "Quantum computing"})
print(result["final"])

Multi-Agent с LangChain Agents

from langchain.agents import AgentExecutor, create_react_agent
from langchain import hub

# Получить промпт для ReAct агента
prompt = hub.pull("hwchase17/react")

# Создание нескольких агентов
def create_specialized_agent(role, tools):
    agent = create_react_agent(llm, tools, prompt)
    return AgentExecutor(agent=agent, tools=tools, verbose=True)

# Агенты
research_agent = create_specialized_agent("Researcher", [web_search_tool])
code_agent = create_specialized_agent("Coder", [python_repl_tool])
test_agent = create_specialized_agent("Tester", [pytest_tool])

# Координатор
def develop_feature(feature_description):
    # 1. Research
    research = research_agent.invoke(
        {"input": f"Research best practices for: {feature_description}"}
    )

    # 2. Code
    code = code_agent.invoke({
        "input": f"Implement {feature_description} using: {research['output']}"
    })

    # 3. Test
    tests = test_agent.invoke({
        "input": f"Write tests for: {code['output']}"
    })

    return {
        "research": research["output"],
        "code": code["output"],
        "tests": tests["output"]
    }

# Использование
result = develop_feature("user authentication with JWT")

19.5. CrewAI — фреймворк для команд

Что такое CrewAI

CrewAI — специализированный фреймворк для создания команд AI-агентов.

Особенности:

• Роли и цели для каждого агента
• Task management
• Delegation между агентами
• Последовательные и параллельные задачи

Установка

pip install crewai crewai-tools

Создание команды

from crewai import Agent, Task, Crew
from crewai_tools import SerperDevTool, FileReadTool

# Инструменты
search_tool = SerperDevTool()
file_tool = FileReadTool()

# Агент 1: Researcher
researcher = Agent(
    role="Senior Researcher",
    goal="Research and analyze information thoroughly",
    backstory="""You are an experienced researcher with a keen eye for detail.
    You excel at finding relevant information and synthesizing insights.""",
    tools=[search_tool],
    verbose=True
)

# Агент 2: Writer
writer = Agent(
    role="Content Writer",
    goal="Create engaging, informative content",
    backstory="""You are a skilled writer who can take complex information
    and make it accessible and engaging.""",
    verbose=True
)

# Агент 3: Editor
editor = Agent(
    role="Editor",
    goal="Ensure content quality and accuracy",
    backstory="""You are a meticulous editor with high standards.
    You improve clarity, fix errors, and ensure consistency.""",
    verbose=True
)

# Задачи
research_task = Task(
    description="Research the latest trends in AI agent technology",
    agent=researcher,
    expected_output="Comprehensive research report with key findings"
)

write_task = Task(
    description="Write a blog post based on the research",
    agent=writer,
    expected_output="Engaging 1000-word blog post",
    context=[research_task]  # Зависит от research_task
)

edit_task = Task(
    description="Edit and polish the blog post",
    agent=editor,
    expected_output="Final edited blog post",
    context=[write_task]
)

# Создание команды
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, edit_task],
    verbose=True
)

# Запуск
result = crew.kickoff()
print(result)

Delegation в CrewAI

Агент-менеджер может делегировать задачи:

manager = Agent(
    role="Project Manager",
    goal="Coordinate the team and ensure tasks are completed",
    backstory="You are an experienced PM who knows how to delegate effectively",
    allow_delegation=True,  # Может делегировать
    verbose=True
)

# Manager получает задачу и делегирует подзадачи команде

Parallel Execution

# Параллельное выполнение задач
crew = Crew(
    agents=[agent1, agent2, agent3],
    tasks=[task1, task2, task3],
    process=Process.parallel,  # Параллельно
    verbose=True
)

19.6. AutoGen — фреймворк от Microsoft

Что такое AutoGen

AutoGen — фреймворк для создания multi-agent диалогов.

Особенности:

• Агенты общаются друг с другом
• Human-in-the-loop (человек в процессе)
• Code execution

Установка

pip install pyautogen

Пример: Два агента решают задачу

import autogen

config_list = [{
    "model": "gpt-4",
    "api_key": "YOUR_API_KEY"
}]

# Агент-ассистент
assistant = autogen.AssistantAgent(
    name="Assistant",
    llm_config={"config_list": config_list}
)

# Агент-пользователь (может выполнять код)
user_proxy = autogen.UserProxyAgent(
    name="User",
    human_input_mode="NEVER",  # Автоматический режим
    code_execution_config={"work_dir": "coding"}
)

# Запуск диалога
user_proxy.initiate_chat(
    assistant,
    message="Plot a chart of NVIDIA and TESLA stock prices over the last year"
)

Что происходит:

1. Assistant пишет Python код для получения данных
2. User_proxy выполняет код
3. Assistant анализирует результат
4. User_proxy создаёт график
5. Цикл продолжается до завершения задачи

Group Chat (несколько агентов)

# Создаём нескольких агентов
coder = autogen.AssistantAgent(
    name="Coder",
    system_message="You are a Python expert. Write clean, efficient code.",
    llm_config={"config_list": config_list}
)

reviewer = autogen.AssistantAgent(
    name="Reviewer",
    system_message="You review code for bugs and improvements.",
    llm_config={"config_list": config_list}
)

tester = autogen.AssistantAgent(
    name="Tester",
    system_message="You write comprehensive tests.",
    llm_config={"config_list": config_list}
)

user = autogen.UserProxyAgent(
    name="User",
    human_input_mode="TERMINATE",
    code_execution_config={"work_dir": "projects"}
)

# Group chat
groupchat = autogen.GroupChat(
    agents=[user, coder, reviewer, tester],
    messages=[],
    max_round=12
)

manager = autogen.GroupChatManager(groupchat=groupchat)

# Запуск
user.initiate_chat(
    manager,
    message="Create a function to calculate fibonacci, review it, and write tests"
)

Диалог:

User: Create a function to calculate fibonacci...

Coder: Here's the implementation:
[код]

Reviewer: Good, but you should handle negative numbers.

Coder: Updated version:
[улучшенный код]

Tester: Here are comprehensive tests:
[тесты]

User: [выполняет тесты]
TERMINATE

19.7. Практика: Создание команды разработки

Задание: Team of Software Engineers

Команда:

• Product Manager (определяет требования)
• Software Architect (проектирует архитектуру)
• Backend Developer (пишет backend код)
• Frontend Developer (пишет frontend код)
• QA Engineer (пишет тесты)
• DevOps (настраивает CI/CD)

Задача: Создать TODO приложение

Workflow:

from crewai import Agent, Task, Crew

# 1. Product Manager
pm = Agent(
    role="Product Manager",
    goal="Define clear requirements",
    backstory="Experienced PM who understands user needs"
)

pm_task = Task(
    description="Define requirements for a TODO app",
    agent=pm,
    expected_output="Detailed requirements document"
)

# 2. Architect
architect = Agent(
    role="Software Architect",
    goal="Design system architecture",
    backstory="Senior architect with 10+ years experience"
)

arch_task = Task(
    description="Design architecture for TODO app",
    agent=architect,
    expected_output="Architecture diagram and tech stack",
    context=[pm_task]
)

# 3. Backend Dev
backend_dev = Agent(
    role="Backend Developer",
    goal="Implement robust backend",
    backstory="Python expert, loves FastAPI"
)

backend_task = Task(
    description="Implement backend API",
    agent=backend_dev,
    expected_output="Backend code with API endpoints",
    context=[arch_task]
)

# 4. Frontend Dev
frontend_dev = Agent(
    role="Frontend Developer",
    goal="Create beautiful UI",
    backstory="React specialist with eye for design"
)

frontend_task = Task(
    description="Implement frontend",
    agent=frontend_dev,
    expected_output="React frontend code",
    context=[arch_task, backend_task]
)

# 5. QA Engineer
qa = Agent(
    role="QA Engineer",
    goal="Ensure quality through testing",
    backstory="Detail-oriented tester"
)

qa_task = Task(
    description="Write comprehensive tests",
    agent=qa,
    expected_output="Test suite (unit + integration + e2e)",
    context=[backend_task, frontend_task]
)

# 6. DevOps
devops = Agent(
    role="DevOps Engineer",
    goal="Automate deployment",
    backstory="K8s and CI/CD expert"
)

devops_task = Task(
    description="Setup CI/CD pipeline and deployment",
    agent=devops,
    expected_output="Deployment scripts and CI/CD config",
    context=[backend_task, frontend_task, qa_task]
)

# Команда
crew = Crew(
    agents=[pm, architect, backend_dev, frontend_dev, qa, devops],
    tasks=[pm_task, arch_task, backend_task, frontend_task, qa_task, devops_task],
    verbose=True
)

# Запуск проекта!
result = crew.kickoff()

Результат: Полностью готовый проект за 1 запуск!

19.8. Best Practices для команд агентов

1. Чёткие роли и ответственности

Плохо:

agent = Agent(role="Helper", goal="Help with tasks")

Хорошо:

agent = Agent(
    role="Senior Backend Developer",
    goal="Implement scalable REST APIs using best practices",
    backstory="10 years of experience with Python, Django, FastAPI"
)

2. Контекст между агентами

Обеспечьте передачу информации:

task2 = Task(
    description="Write tests",
    context=[task1],  # task2 видит результат task1
    agent=tester
)

3. Человек в цикле (Human-in-the-loop)

Для критичных решений:

user_proxy = autogen.UserProxyAgent(
    human_input_mode="ALWAYS",  # Спрашивать пользователя
    code_execution_config={"work_dir": "code"}
)

4. Ограничение итераций

Избегайте бесконечных циклов:

crew = Crew(
    agents=[...],
    tasks=[...],
    max_iterations=10  # Максимум 10 итераций
)

5. Логирование и мониторинг

import logging

logging.basicConfig(level=logging.INFO)

# В процессе
def run_with_logging(crew):
    logging.info("Starting crew...")
    result = crew.kickoff()
    logging.info(f"Finished. Result: {result}")
    return result

Ключевые выводы главы

✅ Команда > Один агент: Специализация даёт лучшие результаты

✅ 4 паттерна: Sequential, Parallel, Hierarchical, Debate

✅ Фреймворки: LangChain (chains), CrewAI (roles/tasks), AutoGen (dialogs)

✅ CrewAI: Лучший для последовательных команд с ролями

✅ AutoGen: Лучший для диалогов между агентами и code execution

✅ Делегирование: Менеджер-агент координирует команду

✅ Контекст: Передавайте результаты между агентами

✅ Human-in-the-loop: Включайте человека в критические моменты

Следующая глава: Платформы для агентов — n8n, LangFlow, Flowise