Агенты: автоматизация цикла tool use | Курс по LangChain урок 7

Цель урока

Вы поймёте разницу между цепочкой и агентом. Научитесь создавать агента с помощью create_agent, управлять его поведением, добавлять память по сессиям и наблюдать за шагами рассуждений в реальном времени.

Необходимые знания:

Уроки 1–6 (LCEL, промпты, память, нелинейные пайплайны, инструменты)

Ключевые концепции:

Разница между цепочкой и агентом
Паттерн ReAct
create_agent из langchain.agents
Системный промпт агента
Память агента и checkpointer
Стриминг шагов агента
Когда использовать агента, а когда цепочку

Цепочка vs Агент

В предыдущих уроках цепочки были детерминированными, разработчик заранее определяет каждый шаг. Агент работает иначе, он сам решает, какие шаги предпринять.

	Цепочка	Агент
Шаги	Фиксированы разработчиком	Определяются моделью в рантайме
Инструменты	Вызываются явно	Выбираются и вызываются моделью
Гибкость	Предсказуема	Адаптируется к задаче
Контроль	Полный	Частичный
Применение	Известный алгоритм	Неизвестная последовательность шагов

В уроке 6 мы вручную писали цикл while response.tool_calls. Агент - это тот же цикл, но реализованный с контролем состояния, памятью и обработкой ошибок.

ReAct

Большинство агентов в LangChain работают по паттерну ReAct (Reasoning + Acting):

Вопрос → Рассуждение (Thought) → Действие (Action) → Наблюдение (Observation) → ...→ Ответ

Агент повторяет цикл "рассуждение → вызов инструмента → анализ результата" до тех пор, пока не решит, что у него достаточно информации для финального ответа.

Пример внутреннего процесса агента:

Вопрос: Какая погода в Москве и сколько это в Фаренгейтах?

Мысль: Нужно узнать погоду в Москве.
Действие: get_weather("москва")
Наблюдение: "Облачно, +5°C"

Мысль: Теперь нужно конвертировать 5°C в Фаренгейты. Формула: F = C * 9/5 + 32
Действие: calculate("5 * 9/5 + 32")
Наблюдение: "41.0"

Мысль: У меня есть все данные для ответа.
Ответ: В Москве облачно, +5°C (41°F).

create_agent

Начиная с LangChain v1, рекомендуемый способ создавать агентов с помощью langchain.agents.

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

load_dotenv()
model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)


@tool
def get_weather(city: str) -> str:
    """Возвращает текущую погоду в городе."""
    data = {
        "москва":  "Облачно, +5°C, ветер 12 км/ч",
        "лондон":  "Дождь, +8°C",
        "токио":   "Солнечно, +18°C",
    }
    return data.get(city.lower(), f"Нет данных для '{city}'")


@tool
def calculate(expression: str) -> str:
    """Вычисляет математическое выражение. Пример: '5 * 9/5 + 32'"""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"Ошибка: {e}"

tools = [get_weather, calculate]

# Создаем агента
agent = create_agent(model, tools)

# Запускаем
result = agent.invoke({
    "messages": [{"role": "user", "content": "Какая погода в Москве? Переведи в Фаренгейты."}]
})

# Финальный ответ — последнее сообщение
print(result["messages"][-1].content)

create_agent возвращает граф (LangGraph StateGraph), который управляет циклом tool use автоматически.

Системный промпт агента

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

С помощью него задают:

Роль и контекст агента
Правила использования инструментов
Формат ответов
Ограничения поведения

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

load_dotenv()
model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)


@tool
def get_weather(city: str) -> str:
    """Возвращает текущую погоду в городе."""
    data = {"москва": "Облачно, +5°C", "лондон": "Дождь, +8°C"}
    return data.get(city.lower(), f"Нет данных для '{city}'")


@tool
def calculate(expression: str) -> str:
    """Вычисляет математическое выражение."""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"Ошибка: {e}"


tools = [get_weather, calculate]

system_prompt = """Ты ассистент-аналитик данных.

Правила работы:
- Всегда показывай промежуточные вычисления
- Если данных недостаточно — спрашивай уточнения
- Отвечай на русском языке
- При работе с числами используй инструмент calculate, не считай в уме
"""

agent = create_agent(
    model,
    tools,
    system_prompt=system_prompt,
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Какая погода в Москве? Переведи в Фаренгейты."}]}
)
print(result["messages"][-1].content)

Память агента

По умолчанию агент stateless, каждый вызов начинается с чистого листа. Для сохранения контекста между вызовами используй checkpointer.

thread_id работает так же, как session_id в уроке 3, разные значения и изолированные истории.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain.agents import create_agent # LangChain facade
# Под капотом: StateGraph + compile() → PregelApp (LangGraph)
from langgraph.checkpoint.memory import MemorySaver  # create_agent строится на LangGraph, checkpointer из него

load_dotenv()
model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)


@tool
def get_weather(city: str) -> str:
    """Возвращает текущую погоду в городе."""
    data = {"москва": "Облачно, +5°C", "лондон": "Дождь, +8°C"}
    return data.get(city.lower(), "Нет данных")


@tool
def calculate(expression: str) -> str:
    """Вычисляет математическое выражение."""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"Ошибка: {e}"


# MemorySaver хранит историю в памяти процесса (сбрасывается при перезапуске)
memory = MemorySaver()

agent = create_agent(
    model,
    [get_weather, calculate],
    checkpointer=memory,
)

# thread_id изолирует разных пользователей / сессии

def chat(message: str) -> str:
    result = agent.invoke(
        {"messages": [{"role": "user", "content": message}]},
        config={"configurable": {"thread_id": "user_42"}},
    )
    return result["messages"][-1].content

print(chat("Меня зовут Алексей, я живу в Москве."))
# "Привет, Алексей!"

print(chat("Какая погода в моём городе?"))
# Агент помнит, что Алексей живёт в Москве
# "В Москве сейчас облачно, +5°C."

print(chat("А если бы я жил в Лондоне?"))
# Контекст сохраняется — агент понимает, о чём речь

Стриминг шагов агента

Агент может выполнять много шагов и ждать финального ответа неудобно. .stream() позволяет видеть процесс в реальном времени.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage
from langchain.agents import create_agent
from langgraph.checkpoint.memory import MemorySaver

load_dotenv()
model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)


@tool
def get_weather(city: str) -> str:
    """Возвращает текущую погоду в городе."""
    data = {"москва": "Облачно, +5°C", "токио": "Солнечно, +18°C"}
    return data.get(city.lower(), f"Нет данных для '{city}'")


@tool
def calculate(expression: str) -> str:
    """Вычисляет математическое выражение."""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"Ошибка: {e}"


memory = MemorySaver()
agent = create_agent(model, [get_weather, calculate], checkpointer=memory)


for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Какая погода в Москве и Токио? Где теплее и насколько?"}]},
    config={"configurable": {"thread_id": "stream_demo"}},
    stream_mode="updates",
):
    # chunk — словарь с ключом "model" или "tools"
    if "model" in chunk:
        message = chunk["model"]["messages"][-1]
        if isinstance(message, AIMessage) and message.content:
            print(f"Агент: {message.content}")
        if hasattr(message, "tool_calls") and message.tool_calls:
            for tc in message.tool_calls:
                print(f"  → вызов: {tc['name']}({tc['args']})")

    if "tools" in chunk:
        for msg in chunk["tools"]["messages"]:
            if isinstance(msg, ToolMessage):
                print(f"  ← результат: {msg.content}")

print()

Вывод будет примерно таким:

  → вызов: get_weather({'city': 'Москва'})
  → вызов: get_weather({'city': 'Токио'})
  ← результат: Облачно, +5°C
  ← результат: Солнечно, +18°C
Агент: Вот текущая погода:

**Москва:** Облачно, +5°C
**Токио:** Солнечно, +18°C

**Теплее в Токио на 13°C** (18°C - 5°C = 13°C)

Кроме того, в Токио не только теплее, но и солнечная погода, в то время как в Москве облачно.

Контроль количества шагов

По умолчанию агент не ограничен по числу шагов. В продакшне это нужно контролировать.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain.agents import create_agent
from langgraph.checkpoint.memory import MemorySaver

load_dotenv()
model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)


@tool
def get_weather(city: str) -> str:
    """Возвращает текущую погоду в городе."""
    data = {"москва": "Облачно, +5°C"}
    return data.get(city.lower(), f"Нет данных для '{city}'")


tools = [get_weather]
memory = MemorySaver()

agent = create_agent(
    model,
    tools,
    checkpointer=memory,
)

try:
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "Какая погода в Москве?"}]},
        config={
            "configurable": {"thread_id": "safe_session"},
            "recursion_limit": 20,  # максимум шагов
        },
    )
    from langchain_core.messages import AIMessage
    final = next(
        m for m in reversed(result["messages"])
        if isinstance(m, AIMessage) and m.content
    )
    print(final.content)
    # "В Москве сейчас облачно, +5°C."
except Exception as e:
    if "recursion" in str(e).lower():
        print("Агент превысил лимит шагов")
    else:
        raise

Middleware в LangChain

В LangChain v1 появился механизм middleware, прослойки, которые оборачивают агента и выполняются при каждом вызове. Они заменяют хуки (pre_model_hook/post_model_hook) и позволяют добавлять сквозную логику без изменения основного кода.

Middleware - это полноценные обёртки, а не просто callback. Они могут перехватить исключение, изменить входные данные или прервать выполнение.

from langchain.agents.middleware import SummarizationMiddleware, ModelFallbackMiddleware

# Автосжатие контекста при переполнении
agent_with_summary = SummarizationMiddleware(
    agent,
    model=model,
    max_tokens=4000,
)

# Автоматические ретраи при ошибках API
agent_with_retry = ModelFallbackMiddleware(
    agent,
    fallback_models=[ChatOpenAI(model="gpt-4o-mini")],
)

Основные встроенные middleware:

SummarizationMiddleware - сжимает длинную историю диалога до суммари, когда контекст превышает max_tokens
ModelFallbackMiddleware - при ошибке переключается на резервную модель с автоматическими ретраями
ModelCallLimitMiddleware - ограничивает максимальное количество обращений к модели
ToolCallLimitMiddleware - ограничивает число вызовов инструментов
PIIMiddleware - редактирует персональные данные во входных/выходных сообщениях
HumanInTheLoopMiddleware - добавляет точку подтверждения перед выполнением инструментов

Middleware комбинируются, каждый следующий оборачивает предыдущий.

from langchain.agents.middleware import (
    SummarizationMiddleware,
    ModelFallbackMiddleware,
    ToolCallLimitMiddleware,
)

production_agent = ToolCallLimitMiddleware(
    ModelFallbackMiddleware(
        SummarizationMiddleware(agent, model=model, max_tokens=4000),
        fallback_models=[ChatOpenAI(model="gpt-4o-mini")],
    ),
    max_tool_calls=10,
)

Когда агент, а когда цепочка

Ошибка использовать агента там, где достаточно цепочки. Агент сложнее в отладке, менее предсказуем и дороже по токенам.

Используйте цепочку (chain), если:

Последовательность шагов известна заранее
Вызовы инструментов детерминированы
Нужна максимальная предсказуемость и контроль
Производительность критична

Используйте агента, если:

Количество шагов неизвестно заранее
Нужно принимать решения на основе результатов предыдущих шагов
Пользователь может задавать произвольные вопросы
Нужна гибкость

Начинайте с цепочки. Переходите к агенту, когда цепочка становится слишком сложной.

Распространенные ошибки

1. Агент без системного промпта работает непредсказуемо

agent = create_agent(model, tools)  # нет системного промпта

Без промпта агент сам выбирает стратегию. Всегда задавайте роль и правила работы.

2. Бесконечный цикл из-за плохих инструментов

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

3. Потеря контекста без thread_id

# Каждый вызов это новый контекст, даже с checkpointer
result1 = agent.invoke(messages, config={"configurable": {"thread_id": "1"}})
result2 = agent.invoke(messages, config={"configurable": {"thread_id": "2"}})
# thread_id "1" и "2", разные сессии

Один пользователь = один постоянный thread_id.

4. Инструменты, которые делают несколько действий

Инструмент, который делает слишком много, сложно правильно использовать. Делите на маленькие специализированные функции, так модель точнее выбирает нужный.

5. Читать только последнее сообщение

# Неправильно: финальный ответ не всегда последний AIMessage
result["messages"][-1]  # может оказаться ToolMessage!

# Правильно: ищем последний AIMessage с непустым content
from langchain_core.messages import AIMessage
final = next(
    m for m in reversed(result["messages"])
    if isinstance(m, AIMessage) and m.content
)

Практическое задание

Создайте агента-помощника по программированию.

Требования:

1) Реализуйте инструменты:

search_docs(library: str, topic: str) -> str - поиск в документации (возвращайте захардкоженные примеры)
run_code_check(code: str) -> dict - проверка кода: синтаксис, базовые проблемы (используйте ast.parse)
get_library_version(library: str) -> str - версия библиотеки, захардкодьте несколько примеров: requests, fastapi, langchain и др.
suggest_alternative(approach: str) -> str - предлагает альтернативные подходы

2) Добавьте память MemorySaver с изоляцией по thread_id

3) Напишите системный промпт, агент это строгий senior разработчик, который:

Всегда проверяет код перед советом
Предпочитает современные подходы
Указывает на лучшие практики

4) Реализуйте стриминг с выводом вызовов инструментов

5) Протестируйте на трёх связанных вопросах подряд, убедитесь, что агент помнит контекст

Итоги урока

Мы научились делать агентов с памятью и инструментами, они работают с произвольными запросами. Но часто нужно работать с конкретными документами или корпоративной базой знаний. В следующем уроке разберем RAG (Retrieval-Augmented Generation), как загружать документы, разбивать их на чанки, использовать векторную БД и делать семантический поиск перед генерацией.

<< Урок 6

Урок 8 >>

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

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