Субагенты и Handoffs | Курс LangChain Agents урок 11

Цель урока: реализовать два паттерна из урока 10, Subagents и Handoffs, и собрать финального агента-аналитика из трёх специализированных субагентов.

Теория

Как технически работают субагенты

Субагент - это обычный агент, обёрнутый в инструмент с помощью @tool. Главный агент вызывает его как любой другой инструмент: решает когда вызвать, передаёт запрос, получает строку с результатом.

Субагенты stateless: каждый вызов создаёт чистый контекст. Главный агент накапливает историю, субагенты работают в изоляции.

Как технически работают Handoffs

Инструмент handoff возвращает Command с обновлением состояния графа. Ключ состояния меняется (например current_step), middleware или граф реагирует на изменение и переключает конфигурацию агента, системный промпт и доступные инструменты.

Примеры кода

Subagents: базовая реализация

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI

load_dotenv()

model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

# Субагент 1: поиск
search_subagent = create_agent(
    model=model,
    tools=[],
    system_prompt="Ты специалист по поиску информации. Отвечай структурировано.",
)

# Субагент 2: анализ
analyst_subagent = create_agent(
    model=model,
    tools=[],
    system_prompt="Ты аналитик данных. Выявляй ключевые факты и тренды.",
)


@tool("search_info", description="Найти информацию по теме")
def call_search_agent(query: str) -> str:
    """Делегировать поиск специализированному агенту."""
    result = search_subagent.invoke({
        "messages": [{"role": "user", "content": query}]
    })
    return result["messages"][-1].content


@tool("analyze_data", description="Проанализировать данные или текст")
def call_analyst_agent(data: str) -> str:
    """Делегировать анализ специализированному агенту."""
    result = analyst_subagent.invoke({
        "messages": [{"role": "user", "content": data}]
    })
    return result["messages"][-1].content


# Главный агент координирует субагентов
main_agent = create_agent(
    model=model,
    tools=[call_search_agent, call_analyst_agent],
    system_prompt=(
        "Ты координатор. Для поиска информации используй search_info. "
        "Для анализа используй analyze_data. "
        "Отвечай кратко."
    ),
)

response = main_agent.invoke({
    "messages": [{"role": "user", "content": "Найди информацию об Anthropic и проанализируй."}]
})
print(response["messages"][-1].content[:200])

Subagents: параллельный вызов

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

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI

load_dotenv()

model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

python_agent = create_agent(
    model=model,
    tools=[],
    system_prompt="Ты эксперт по Python.",
)

rust_agent = create_agent(
    model=model,
    tools=[],
    system_prompt="Ты эксперт по Rust.",
)

go_agent = create_agent(
    model=model,
    tools=[],
    system_prompt="Ты эксперт по Go.",
)


@tool("ask_python", description="Задать вопрос эксперту по Python")
def call_python(question: str) -> str:
    result = python_agent.invoke({"messages": [{"role": "user", "content": question}]})
    return result["messages"][-1].content


@tool("ask_rust", description="Задать вопрос эксперту по Rust")
def call_rust(question: str) -> str:
    result = rust_agent.invoke({"messages": [{"role": "user", "content": question}]})
    return result["messages"][-1].content


@tool("ask_go", description="Задать вопрос эксперту по Go")
def call_go(question: str) -> str:
    result = go_agent.invoke({"messages": [{"role": "user", "content": question}]})
    return result["messages"][-1].content


coordinator = create_agent(
    model=model,
    tools=[call_python, call_rust, call_go],
    system_prompt=(
        "Сравни языки программирования, опрашивая экспертов параллельно. "
        "Задай один и тот же вопрос всем трём экспертам одновременно, "
        "затем объедини ответы."
    ),
)

response = coordinator.invoke({
    "messages": [{"role": "user", "content": "Сравни производительность Python, Rust и Go для веб-сервисов."}]
})
print(response["messages"][-1].content[:300])

Handoffs: переключение конфигурации

Handoffs меняет поведение агента через состояние. Инструмент возвращает Command с обновлением current_step, middleware читает состояние и применяет нужный системный промпт.

import os
from typing import Any, NotRequired
from dotenv import load_dotenv
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model, dynamic_prompt, ModelRequest
from langchain.tools import tool, ToolRuntime
from langchain.messages import ToolMessage
from langchain_openai import ChatOpenAI
from langgraph.types import Command
from langgraph.runtime import Runtime
from langgraph.checkpoint.memory import InMemorySaver

load_dotenv()


class SupportState(AgentState):
    current_step: NotRequired[str]


PROMPTS = {
    "intake": (
        "Ты оператор поддержки. Выясни суть проблемы клиента. "
        "Когда поймёшь проблему, переключись в режим решения через transfer_to_resolver."
    ),
    "resolver": (
        "Ты специалист по решению проблем. "
        "Предложи конкретное решение для проблемы клиента."
    ),
}


@dynamic_prompt
def step_based_prompt(request: ModelRequest) -> str:
    step = request.state.get("current_step", "intake")
    return PROMPTS.get(step, PROMPTS["intake"])


@tool
def transfer_to_resolver(runtime: ToolRuntime[None, SupportState]) -> Command:
    """Переключиться в режим решения проблемы."""
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content="Переключаюсь в режим решения...",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
            "current_step": "resolver",
        }
    )


model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

agent = create_agent(
    model=model,
    tools=[transfer_to_resolver],
    state_schema=SupportState,
    middleware=[step_based_prompt],
)


agent = create_agent(
    model=model,
    tools=[transfer_to_resolver],
    state_schema=SupportState,
    middleware=[step_based_prompt],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "support-1"}}

# Первый шаг: intake
r1 = agent.invoke(
    {"messages": [{"role": "user", "content": "Приложение не запускается."}]},
    config,
)
print("Intake:", r1["messages"][-1].content[:100])

# Второй шаг: после handoff агент в режиме resolver
r2 = agent.invoke(
    {"messages": [{"role": "user", "content": "Ошибка: порт 8080 занят."}]},
    config,
)
print("Resolver:", r2["messages"][-1].content[:100])

ToolMessage с tool_call_id обязателен: без него история диалога станет некорректной, так как модель ждёт ответа на вызов инструмента.

Сквозной проект: финальный агент-аналитик

Финальный агент использует async checkpointer для которого нужен дополнительный пакет:

pip install aiosqlite

Собираем полноценного агента с возможностями из всех уроков курса:

Урок 3: субагент поиска использует реальные инструменты через MCP
Урок 6: краткосрочная память на AsyncSqliteSaver, история переживает перезапуск
Урок 7: долгосрочная память, сохраняем и переиспользуем результаты исследований
Урок 9: guardrail отсекает нерелевантные запросы

Так как субагент поиска работает через MCP (async), весь координатор тоже async.

import asyncio
import os
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from typing_extensions import TypedDict
from dotenv import load_dotenv
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model, PIIMiddleware, ModelCallLimitMiddleware
from langchain.tools import tool, ToolRuntime
from langchain.messages import AIMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from langgraph.store.memory import InMemoryStore
from langgraph.runtime import Runtime

load_dotenv()

SERVER_PATH = str(Path(__file__).parent.parent / "agents_course" / "tools_server.py")
model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))


# --- Долгосрочная память (урок 7) ---

store = InMemoryStore()


@dataclass
class AnalystContext:
    user_id: str


class ResearchNote(TypedDict):
    topic: str
    summary: str


@tool
def save_research_note(note: ResearchNote, runtime: ToolRuntime[AnalystContext]) -> str:
    """Сохранить результат исследования в долгосрочную память."""
    key = note["topic"].lower().replace(" ", "-")
    runtime.store.put((runtime.context.user_id, "research"), key, note)
    return f"Заметка по теме '{note['topic']}' сохранена."


@tool
def get_research_note(topic: str, runtime: ToolRuntime[AnalystContext]) -> str:
    """Проверить, есть ли уже сохранённое исследование по теме."""
    key = topic.lower().replace(" ", "-")
    item = runtime.store.get((runtime.context.user_id, "research"), key)
    if item:
        return f"Уже исследовано: {item.value['summary']}"
    return f"По теме '{topic}' исследований нет, нужно провести."


# --- Guardrail (урок 9) ---

OFF_TOPIC_KEYWORDS = ["политика", "религия", "криптовалюта", "ставки", "казино"]


@before_model(can_jump_to=["end"])
def topic_guardrail(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    last_message = state["messages"][-1]
    if not isinstance(last_message, HumanMessage):
        return None
    content = str(last_message.content).lower()
    for keyword in OFF_TOPIC_KEYWORDS:
        if keyword in content:
            return {
                "messages": [AIMessage("Я аналитик данных. Обращайтесь с аналитическими задачами.")],
                "jump_to": "end",
            }
    return None


async def main():
    # --- MCP инструменты для субагента поиска (урок 3) ---
    client = MultiServerMCPClient({
        "analyst_tools": {
            "transport": "stdio",
            "command": sys.executable,
            "args": [SERVER_PATH],
        }
    })
    mcp_tools = await client.get_tools()

    # --- Субагенты ---

    search_agent = create_agent(
        model=model,
        tools=mcp_tools,  # реальный поиск через MCP
        system_prompt=(
            "Ты специалист по поиску. Используй инструменты для сбора актуальных данных. "
            "Отвечай структурированными фактами: 3-5 ключевых пункта по теме."
        ),
    )

    analyst_agent = create_agent(
        model=model,
        tools=[],
        system_prompt="Ты аналитик данных. Получаешь факты, выявляешь тренды и выводы. Будь краток.",
    )

    reporter_agent = create_agent(
        model=model,
        tools=[],
        system_prompt="Ты редактор отчётов. Оформляй краткий структурированный отчёт в 3-4 предложения.",
    )

    # --- Инструменты обёртки над субагентами ---

    @tool("search_topic", description="Собрать актуальную информацию по теме через поиск")
    async def call_search(query: str) -> str:
        result = await search_agent.ainvoke({
            "messages": [{"role": "user", "content": query}]
        })
        return result["messages"][-1].content

    @tool("analyze_findings", description="Проанализировать собранные данные")
    async def call_analyst(findings: str) -> str:
        result = await analyst_agent.ainvoke({
            "messages": [{"role": "user", "content": findings}]
        })
        return result["messages"][-1].content

    @tool("write_report", description="Составить итоговый отчёт")
    async def call_reporter(analysis: str) -> str:
        result = await reporter_agent.ainvoke({
            "messages": [{"role": "user", "content": analysis}]
        })
        return result["messages"][-1].content

    # --- Главный координирующий агент ---

    async with AsyncSqliteSaver.from_conn_string("analyst.db") as checkpointer:  # краткосрочная память (урок 6)
        main_agent = create_agent(
            model=model,
            tools=[call_search, call_analyst, call_reporter, save_research_note, get_research_note],
            checkpointer=checkpointer,
            store=store,
            context_schema=AnalystContext,
            middleware=[
                topic_guardrail,
                PIIMiddleware("email", strategy="redact", apply_to_input=True),
                ModelCallLimitMiddleware(run_limit=15, exit_behavior="end"),
            ],
            system_prompt=(
                "Ты координатор исследований. Для каждого запроса:\n"
                "1. Проверь долгосрочную память через get_research_note\n"
                "2. Если данных нет, используй search_topic для сбора информации\n"
                "3. Передай результаты в analyze_findings\n"
                "4. Оформи итог через write_report\n"
                "5. Сохрани результат через save_research_note\n"
                "Работай последовательно."
            ),
        )

        context = AnalystContext(user_id="analyst-1")
        config = {"configurable": {"thread_id": "final-analyst"}}

        response = await main_agent.ainvoke(
            {"messages": [{"role": "user", "content": "Исследуй компанию Anthropic."}]},
            config,
            context=context,
        )
        print(response["messages"][-1].content[:500])


if __name__ == "__main__":
    asyncio.run(main())

Частые ошибки

Субагент возвращает слишком много данных

# Плохо: возвращаем весь список сообщений субагента главному агенту
@tool
def call_subagent(query: str) -> str:
    result = subagent.invoke(...)
    return str(result["messages"])  # тысячи токенов

# Хорошо: только последний ответ
@tool
def call_subagent(query: str) -> str:
    result = subagent.invoke(...)
    return result["messages"][-1].content

Handoff без ToolMessage

# Неправильно: нет ToolMessage, история диалога сломается
@tool
def transfer(runtime: ToolRuntime) -> Command:
    return Command(update={"current_step": "specialist"})

# Правильно: ToolMessage с tool_call_id закрывает цикл запрос-ответ
@tool
def transfer(runtime: ToolRuntime) -> Command:
    return Command(update={
        "messages": [ToolMessage(content="Переключился", tool_call_id=runtime.tool_call_id)],
        "current_step": "specialist",
    })

Задание

Доработайте финального агента-аналитика:

Добавьте четвёртого субагента fact_checker, он получает отчёт и проверяет нет ли противоречий или очевидных ошибок
Добавьте к главному агенту SummarizationMiddleware и ModelCallLimitMiddleware из урока 5
Убедитесь что цепочка search → analyze → report → fact_check выполняется в нужном порядке

<< урок 10

Итог курса

Вы прошли полный путь от первого агента до мультиагентной системы.

Что вы узнали:

В основе курса лежит агент-аналитик, который развивался от урока к уроку:

Урок 1: базовый агент на create_agent с инструментами
Урок 2: краткосрочная история через checkpointer и thread_id
Урок 3: внешние инструменты через MCP
Урок 4: контекст-инжиниринг, @dynamic_prompt и Runtime Context
Урок 5: встроенный middleware, лимиты и суммаризация
Урок 6: краткосрочная память, диалог между сессиями
Урок 7: долгосрочная память через InMemoryStore и эмбеддинги
Урок 8: Human-in-the-loop, подтверждение действий пользователем
Урок 9: guardrails, защита от нежелательных тем и утечки PII
Урок 10: паттерны мультиагентных систем и критерии выбора
Урок 11: субагенты как инструменты, Handoffs через Command

Ключевые принципы, которые прошли через весь курс:

Агент это цикл: получить состояние, вызвать модель, выполнить инструменты, повторить. Всё остальное, надстройки над этим циклом.

Middleware перехватывает вызовы до и после модели. Это единственная точка для сквозной логики: логирование, ограничения, трансформация данных.

Субагент - это инструмент. Главный агент не знает и не должен знать, что внутри инструмента находиться другой агент.

Сложность оправдана только когда один агент не справляется: переполняется контекст, нужна параллельность или разные команды развивают систему независимо.

Подписывайтесь на мой Telegram канал

Если вам нужен ментор и вы хотите научиться разрабатывать AI агентов, пишите, обсудим условия