LangGraph на практике: создаём агента, который помнит всё

Почему ваши агенты забывают всё через 5 минут (и как это исправить)

Вы создали умного агента на LangChain. Он отвечает на вопросы, ищет в документах, даже шутит иногда. Но есть проблема: каждый новый запрос - это новая жизнь. Он не помнит, что вы только что обсуждали. Не знает ваших предпочтений. Каждый раз начинаете с чистого листа.

Знакомо? Это классическая проблема агентов без состояния. Они работают в режиме "один запрос - один ответ". Контекстное окно LLM - это оперативная память, которая сбрасывается после каждого вызова.

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

Что такое LangGraph и зачем он нужен

LangGraph - это фреймворк для построения stateful (состоятельных) мультиагентных приложений. Если LangChain - это конструктор цепочек, то LangGraph - это конструктор графов с циклами, условиями и состоянием.

Представьте себе:

• Узлы (nodes) - это функции, которые что-то делают
• Рёбра (edges) - это переходы между узлами
• Состояние (state) - это общая память, которая передаётся между узлами
• Условия (conditional edges) - это развилки: "если ответ готов, иди на выход, иначе продолжай думать"

Вот почему это важно: обычные цепочки линейны. Запрос → обработка → ответ. Графы могут содержать циклы: "подумай → проверь → если недостаточно хорошо, подумай ещё". Именно так работают сложные агенты.

Что мы будем строить: агент-персональный помощник

Наш агент будет:

1. Запоминать предпочтения пользователя (любит краткие ответы, ненавидит эмодзи, предпочитает технические детали)
2. Вести историю диалога в структурированном виде
3. Использовать память при генерации ответов
4. Иметь саморефлексию: анализировать, достаточно ли он понял запрос

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

1 Настраиваем окружение и устанавливаем зависимости

Первое, что ломает большинство туториалов - устаревшие версии пакетов. На февраль 2026 актуальны:

# Создаём виртуальное окружение
python -m venv langgraph_env
source langgraph_env/bin/activate  # На Windows: langgraph_env\Scripts\activate

# Устанавливаем актуальные версии
pip install langgraph==0.2.0  # Последняя стабильная на февраль 2026
pip install langchain==0.2.0  # Совместимая версия
pip install langchain-openai==0.1.0  # Для работы с OpenAI
pip install pydantic==2.7.0  # Для типизации состояния

2 Определяем состояние агента: что он будет помнить

Состояние в LangGraph - это Pydantic модель. Все узлы графа читают и пишут в это состояние. Вот как НЕ надо делать:

# ПЛОХО: словарь без структуры
state = {
    "messages": [],
    "user_prefs": {},
    "history": []
}
# Почему плохо: нет типизации, легко ошибиться в ключах

А вот правильный подход:

from typing import List, Dict, Any, Optional, Annotated
from typing_extensions import TypedDict
from pydantic import BaseModel, Field
from datetime import datetime

# Модель для эпизодической памяти
class MemoryEntry(BaseModel):
    timestamp: datetime = Field(default_factory=datetime.now)
    event_type: str  # "user_preference", "conversation", "reflection"
    content: Dict[str, Any]
    confidence: float = 0.8
    source: str = "dialog"

# Состояние нашего агента
class AgentState(TypedDict):
    # Текущий запрос пользователя
    user_input: str
    
    # История сообщений для контекста LLM
    messages: Annotated[List[Dict], "Список сообщений в формате чата"]
    
    # Долговременная память (эпизодическая)
    memory: Annotated[List[MemoryEntry], "Структурированные записи памяти"]
    
    # Извлечённые предпочтения пользователя
    user_preferences: Annotated[Dict[str, Any], "Предпочтения пользователя"]
    
    # Текущий ответ агента
    agent_response: str
    
    # Флаги для управления потоком
    needs_clarification: bool
    response_complete: bool

Почему такая сложность? Потому что простая текстовая история - это мусор для LLM. Структурированная память позволяет:

• Быстро искать релевантные воспоминания
• Фильтровать по типу события
• Учитывать уверенность (confidence) в записи
• Автоматически очищать старые записи

3 Создаём узлы графа: мозг нашего агента

Узел в LangGraph - это функция, которая принимает состояние и возвращает обновлённое состояние. Давайте создадим первый узел - анализатор запроса.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import StateGraph, END
import json

# Инициализируем LLM (используем актуальную на 2026 модель)
llm = ChatOpenAI(
    model="gpt-4o-mini",  # Компактная и эффективная версия
    temperature=0.3,
    api_key="your-api-key-here"  # Замените на свой ключ
)

# Узел 1: Анализ запроса и извлечение интента
def analyze_query(state: AgentState) -> AgentState:
    """Анализирует запрос пользователя, извлекает интент и проверяет, нужны ли уточнения."""
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", """Ты - анализатор запросов. Определи:
        1. Что хочет пользователь (основной интент)
        2. Нужны ли уточнения (да/нет)
        3. Какие сущности упомянуты
        
        Ответ в формате JSON:
        {
            "intent": "описание интента",
            "needs_clarification": true/false,
            "clarification_questions": ["вопрос1", "вопрос2"],
            "entities": ["сущность1", "сущность2"]
        }"""),
        ("human", "{user_input}")
    ])
    
    chain = prompt | llm
    result = chain.invoke({"user_input": state["user_input"]})
    
    try:
        analysis = json.loads(result.content)
    except json.JSONDecodeError:
        # Если LLM вернул не JSON, используем fallback
        analysis = {
            "intent": "general_query",
            "needs_clarification": False,
            "clarification_questions": [],
            "entities": []
        }
    
    # Создаём запись в памяти
    memory_entry = MemoryEntry(
        event_type="query_analysis",
        content=analysis,
        confidence=0.9
    )
    
    # Обновляем состояние
    return {
        "needs_clarification": analysis["needs_clarification"],
        "memory": state["memory"] + [memory_entry]
    }

Теперь создадим узел для поиска в памяти:

# Узел 2: Поиск релевантных воспоминаний
def retrieve_memories(state: AgentState) -> AgentState:
    """Ищет релевантные записи в памяти на основе текущего запроса."""
    
    if not state["memory"]:
        return {"memory": state["memory"]}  # Память пуста
    
    # Простой поиск по ключевым словам (в реальном проекте используйте векторный поиск)
    query = state["user_input"].lower()
    relevant_memories = []
    
    for entry in state["memory"]:
        # Ищем в содержимом записи
        entry_str = json.dumps(entry.content).lower()
        
        # Проверяем релевантность (упрощённо)
        words = query.split()
        matches = sum(1 for word in words if word in entry_str)
        
        if matches > 0:
            relevant_memories.append(entry)
    
    # Ограничиваем количество воспоминаний для контекста
    relevant_memories = relevant_memories[-5:]  # Последние 5 релевантных
    
    return {
        "memory": state["memory"],  # Оригинальная память не меняется
        "relevant_memories": relevant_memories  # Новое поле в состоянии
    }

4 Собираем граф: соединяем узлы рёбрами

Теперь самое интересное - создаём граф и определяем, как узлы связаны между собой.

# Создаём граф
workflow = StateGraph(AgentState)

# Добавляем узлы
workflow.add_node("analyze", analyze_query)
workflow.add_node("retrieve", retrieve_memories)
workflow.add_node("generate", generate_response)
workflow.add_node("clarify", ask_clarification)
workflow.add_node("update_memory", update_user_preferences)

# Определяем начальную точку
workflow.set_entry_point("analyze")

# Добавляем рёбра (переходы)
workflow.add_edge("analyze", "retrieve")

# Условное ребро: если нужны уточнения, идём в clarify, иначе в generate
workflow.add_conditional_edges(
    "retrieve",
    # Функция-роутер
    lambda state: "clarify" if state["needs_clarification"] else "generate",
    {
        "clarify": "clarify",
        "generate": "generate"
    }
)

# После генерации ответа обновляем память
workflow.add_edge("generate", "update_memory")
workflow.add_edge("clarify", "update_memory")

# Конец графа
workflow.add_edge("update_memory", END)

# Компилируем граф в исполняемый объект
app = workflow.compile()

Обратите внимание на conditional_edges - это развилки. Граф решает, куда идти дальше на основе состояния. Это то, что отличает графы от линейных цепочек.

5 Добавляем генерацию ответа с учётом памяти

Самый важный узел - генератор ответов. Именно здесь память влияет на результат.

def generate_response(state: AgentState) -> AgentState:
    """Генерирует ответ с учётом памяти и предпочтений пользователя."""
    
    # Подготавливаем контекст из памяти
    memory_context = ""
    if "relevant_memories" in state and state["relevant_memories"]:
        memory_context = "\nРелевантные воспоминания:\n"
        for i, memory in enumerate(state["relevant_memories"], 1):
            memory_context += f"{i}. {memory.content}\n"
    
    # Учитываем предпочтения пользователя
    preferences = state.get("user_preferences", {})
    style_hint = ""
    if preferences.get("prefers_concise", False):
        style_hint = "Отвечай кратко, без лишних деталей. "
    if preferences.get("hates_emojis", False):
        style_hint += "Не используй эмодзи. "
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", f"""Ты - полезный AI-ассистент. У тебя есть память о прошлых взаимодействиях.
        
        {style_hint}
        
        Учитывай следующую информацию из памяти:
        {memory_context}
        
        Отвечай на русском языке, будь полезным и точным."""),
        # Добавляем историю диалога
        *state["messages"][-6:],  # Последние 6 сообщений
        ("human", "{user_input}")
    ])
    
    chain = prompt | llm
    response = chain.invoke({"user_input": state["user_input"]})
    
    return {
        "agent_response": response.content,
        "messages": state["messages"] + [
            {"role": "user", "content": state["user_input"]},
            {"role": "assistant", "content": response.content}
        ]
    }

6 Запускаем и тестируем агента

Теперь давайте посмотрим на нашего агента в действии:

# Инициализируем начальное состояние
initial_state = AgentState(
    user_input="Привет! Меня зовут Алексей. Я предпочитаю краткие ответы без эмодзи.",
    messages=[],
    memory=[],
    user_preferences={},
    agent_response="",
    needs_clarification=False,
    response_complete=False
)

# Запускаем граф
result = app.invoke(initial_state)

print("Ответ агента:", result["agent_response"])
print("\nЗапись в памяти:")
for entry in result["memory"]:
    print(f"- {entry.event_type}: {entry.content}")

Агент должен ответить что-то вроде: "Привет, Алексей! Запомнил, что вы предпочитаете краткие ответы без эмодзи."

Теперь второй запрос:

# Второй запрос - агент должен помнить предпочтения
second_state = AgentState(
    user_input="Расскажи о LangGraph",
    messages=result["messages"],  # Продолжаем историю
    memory=result["memory"],      # Сохраняем память
    user_preferences=result.get("user_preferences", {}),
    agent_response="",
    needs_clarification=False,
    response_complete=False
)

result2 = app.invoke(second_state)
print("\nВторой ответ (должен быть кратким и без эмодзи):")
print(result2["agent_response"])

Где всё ломается: типичные ошибки и как их избежать

Я видел десятки сломанных реализаций LangGraph. Вот самые частые проблемы:

Продвинутые техники: куда двигаться дальше

Базовая реализация работает. Но в продакшене нужно больше:

1. Векторная память с семантическим поиском

Вместо простого текстового поиска используйте эмбеддинги. Каждую запись MemoryEntry эмбеддите и храните в Qdrant или Pinecone. Тогда поиск "мне нравятся технические детали" найдёт записи про "предпочитает подробные объяснения".

2. Иерархическая память

Разные типы воспоминаний храните по-разному:

• Краткосрочная: последние 10 сообщений в оперативном контексте
• Среднесрочная: предпочтения пользователя, извлечённые факты
• Долгосрочная: важные события, подтверждённые предпочтения

3. Саморефлексия и очистка памяти

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

def reflect_on_memory(state: AgentState) -> AgentState:
    """Анализирует память, удаляет противоречия, обобщает факты."""
    # Например: если есть 5 записей "пользователь любит кофе", 
    # создаём одну обобщённую запись с confidence=0.95
    # А противоречивые записи ("любит кофе" vs "не любит кофе") помечаем для уточнения
    pass

Почему бесконечный контекст не решает проблему памяти

Сейчас у некоторых моделей контекстное окно достигает 1 миллиона токенов. Кажется, что можно просто запихнуть всю историю в промпт и забыть о сложных системах памяти. Но это иллюзия.

Во-первых, цена. 1M токенов в GPT-4o стоит около $30 за запрос. Во-вторых, качество. LLM плохо извлекают информацию из середины длинного контекста (эффект "потеря в середине"). В-третьих, релевантность. 99% истории диалога не нужны для текущего запроса.

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

Что в итоге получилось

Мы создали агента на LangGraph, который:

• Имеет структурированное состояние с типизацией
• Запоминает предпочтения пользователя между запросами
• Использует граф с условными переходами для сложной логики
• Может запрашивать уточнения при неясных запросах
• Масштабируется: можно добавлять новые узлы (поиск в интернете, вызов API, работа с файлами)

Полный код проекта доступен в GitHub (ссылка анонимная, как просили).

Хотите глубже? Посмотрите как проектировать современных AI-агентов или разбор почему ломаются LLM-агенты.

А если нужно встроить такого агента в продакшен - обратите внимание на LangSmith для мониторинга и отладки. Как это сделали Vodafone и Fastweb для автоматизации поддержки клиентов.