Context engineering | Курс LangChain Agents урок 4

Цель урока: научиться управлять тем, что агент видит на каждом шаге, через динамический промпт и хуки middleware.

Теория

Почему агенты ломаются

Когда агент делает что-то не то, обычно одна из двух причин:

модель недостаточно подходит для задачи
модель получила неправильный контекст

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

Context engineering - это практика подавать модели правильную информацию в правильный момент. В этом и заключается основная работа при разработке агентов.

Три типа контекста

Тип	Что контролируете	Сохраняется?
Model context	Промпт, инструменты, формат ответа	Нет
Tool context	Что инструмент читает и пишет в стейт	Да
Life-cycle context	Что происходит между вызовами	Да

В этом уроке разбираем model context, это то, что модель видит при каждом вызове.

Три источника данных

Динамический контекст может приходить из разных мест:

State - текущий разговор, история сообщений. Меняется с каждым шагом.
Runtime Context - статичная конфигурация сессии: user_id, права доступа, подключения к базам данных. Задаётся при запуске агента.
Store - долгосрочная память, данные между сессиями. Разберём в уроке 7.

Middleware

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

Два стиля хуков:

Node-style - функции, которые запускаются в конкретный момент:

before_model - перед каждым вызовом модели
after_model - после каждого ответа модели
before_agent - один раз в начале
after_agent - один раз в конце

Wrap-style - оборачивают вызов и решают, запускать ли его вообще:

wrap_model_call - вокруг вызова модели (можно добавить retry, кэш)
wrap_tool_call - вокруг вызова инструмента

`@dynamic_prompt`

Самый частый сценарий это менять системный промпт в зависимости от контекста. Для этого есть декоратор @dynamic_prompt:

from langchain.agents.middleware import dynamic_prompt, ModelRequest

@dynamic_prompt
def my_prompt(request: ModelRequest) -> str:
    message_count = len(request.state["messages"])
    return f"Ты помощник. В разговоре {message_count} сообщений."

Функция получает объект request с доступом к текущему стейту и runtime context и возвращает строку промпта.

Примеры кода

Динамический промпт по длине разговора

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest
from langchain_openai import ChatOpenAI

load_dotenv()


@dynamic_prompt
def adaptive_prompt(request: ModelRequest) -> str:
    message_count = len(request.state["messages"])
    base = "Ты аналитик-исследователь. Отвечай структурированно."

    if message_count > 10:
        base += " Разговор длинный, будь лаконичен."
    elif message_count == 1:
        base += " Это начало разговора, задай уточняющие вопросы если нужно."

    return base

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

agent = create_agent(
    model=model,
    tools=[],
    middleware=[adaptive_prompt],
)

response = agent.invoke({
    "messages": [{"role": "user", "content": "Расскажи об Anthropic."}]
})
print(response["messages"][-1].content)

Хуки для логирования

import os
from dotenv import load_dotenv
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model, after_model
from langchain_openai import ChatOpenAI
from langgraph.runtime import Runtime
from typing import Any

load_dotenv()

@before_model
def log_before(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    last = state["messages"][-1]
    print(f"[before_model] сообщений: {len(state['messages'])}, последнее: {last.content[:60]}")
    return None  # None - продолжаем нормальный цикл

@after_model
def log_after(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    last = state["messages"][-1]
    print(f"[after_model] ответ модели: {last.content[:60]}")
    return None

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

agent = create_agent(
    model=model,
    tools=[],
    middleware=[log_before, log_after],
)

agent.invoke({
    "messages": [{"role": "user", "content": "Привет!"}]
})

Хук возвращает None, чтобы продолжить нормальный цикл. Если вернуть словарь, он будет применён к стейту.

Runtime Context: передаём конфигурацию сессии

Runtime Context позволяет передать агенту данные, которые не меняются в течение сессии: user_id, роль пользователя, настройки.

import os
from dataclasses import dataclass
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest
from langchain_openai import ChatOpenAI

load_dotenv()

@dataclass
class UserContext:
    user_id: str
    role: str  # "admin" или "viewer"
    language: str

@dynamic_prompt
def role_based_prompt(request: ModelRequest) -> str:
    ctx = request.runtime.context
    if ctx is None:
        return "Ты помощник."

    base = f"Ты аналитик. Разговариваешь с пользователем {ctx.user_id}."

    if ctx.role == "admin":
        base += " Пользователь имеет доступ ко всем данным."
    else:
        base += " Показывай только публичные данные."

    if ctx.language == "ru":
        base += " Отвечай на русском языке."

    return base

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

agent = create_agent(
    model=model,
    tools=[],
    middleware=[role_based_prompt],
    context_schema=UserContext,
)

# Передаём контекст при вызове
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Какие данные мне доступны?"}]},
    context=UserContext(user_id="user-42", role="admin", language="ru"),
)
print(response["messages"][-1].content)

Сквозной проект: динамический промпт для агента-аналитика

import asyncio
import os
import sys
from dataclasses import dataclass
from pathlib import Path
from dotenv import load_dotenv
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import dynamic_prompt, before_model, ModelRequest
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from typing import Any

load_dotenv()

SERVER_PATH = str(Path(__file__).parent.parent / "agents_course" / "tools_server.py")

@dynamic_prompt
def analyst_prompt(request: ModelRequest) -> str:
    message_count = len(request.state["messages"])
    base = (
        "Ты аналитик-исследователь. Собирай информацию через инструменты "
        "и давай структурированный ответ с выводами."
    )
    if message_count > 6:
        base += " Разговор длинный, резюмируй только ключевые факты."
    return base

@before_model
def log_step(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    print(f"  [шаг] сообщений в стейте: {len(state['messages'])}")
    return None

async def main():
    client = MultiServerMCPClient({
        "analyst_tools": {
            "transport": "stdio",
            "command": sys.executable,
            "args": [SERVER_PATH],
        }
    })

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

    agent = create_agent(
        model=model,
        tools=tools,
        middleware=[analyst_prompt, log_step],
        checkpointer=InMemorySaver(),
    )

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

    response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "Найди информацию об Anthropic."}]},
        config,
    )
    print(response["messages"][-1].content)

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

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

Хук случайно возвращает словарь

@before_model
def my_hook(state: AgentState, runtime: Runtime):
    print("log")
    # Неправильно: функция без return неявно возвращает None,
    # но если забыть return и написать что-то другое:
    result = do_something()
    return result  # если result не None, он применится к стейту

# Правильно: явно возвращать None если изменений нет
@before_model
def my_hook(state: AgentState, runtime: Runtime) -> dict | None:
    print("log")
    return None

Порядок middleware имеет значение

# Неправильно: логирование стоит после модификации промпта,
# видит уже изменённое состояние
agent = create_agent(model=model, tools=[], middleware=[modify_prompt, log_before])

# Правильно: логирование до изменений, чтобы видеть исходное состояние
agent = create_agent(model=model, tools=[], middleware=[log_before, modify_prompt])

Middleware выполняются в порядке списка. Если хотите залогировать исходное состояние, ставьте логгер первым.

request.messages не существует в Python

# Неправильно: request.messages - это JS API, в Python его нет
@dynamic_prompt
def my_prompt(request: ModelRequest) -> str:
    message_count = len(request.messages)  # AttributeError
    return f"Сообщений: {message_count}"

# Правильно: используйте request.state["messages"]
@dynamic_prompt
def my_prompt(request: ModelRequest) -> str:
    message_count = len(request.state["messages"])
    return f"Сообщений: {message_count}"

Задание

Добавьте к агенту конвертации валют из уроков 1-3 два middleware:

@dynamic_prompt, который меняет промпт в зависимости от того, сколько конвертаций уже сделано (можно отслеживать через количество сообщений)
@after_model, который логирует стоимость запроса через state["messages"][-1].usage_metadata (если модель возвращает метаданные использования)

Запустите диалог из трёх запросов и убедитесь в LangSmith, что промпт менялся между запросами.

<< урок 3

урок 5 >>

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

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