Как вы передаете состояние (state) между шагами агента?

Краткий тезис

Передача состояния между шагами агента — ключевой механизм, обеспечивающий согласованность и контекст в многошаговых рассуждениях. Лучшая практика — использование LangGraph State Schema с TypedDict и явными полями (messages, retrieved_chunks, intermediate_answers). Необходимо хранить только минимально необходимые данные, избегая дублирования и переполнения контекста. Правильное управление состоянием напрямую влияет на производительность, отказоустойчивость и интерпретируемость агента.

1. Термин: State (состояние) агента

Состояние агента — это совокупность данных, которые агент накапливает и использует в процессе выполнения задачи. В Agentic RAG состояние включает историю диалога, результаты поиска, промежуточные рассуждения, выбранные инструменты и их результаты.

Почему это важно

• Агент — это не stateless-функция; он принимает решения на основе предыдущих шагов.
• Без правильной передачи состояния агент «забывает» контекст, что приводит к бессвязным ответам.
• Состояние должно быть сериализуемым (для логирования, отладки, восстановления после сбоев) и детерминированным (воспроизводимость).

2. Основные подходы к передаче состояния

Лучшая практика для RAG|Agentic RAG — LangGraph State Schema (или аналоги в других фреймворках), так как она обеспечивает чёткую структуру, контроль версий и интеграцию с графом выполнения.

3. LangGraph State Schema: TypedDict и редьюсеры

LangGraph — библиотека для построения агентов в виде графов. Состояние описывается через TypedDict (или dataclass) с аннотациями типов. Каждое поле может иметь редьюсер — функцию, определяющую, как обновлять значение при добавлении нового.

Пример базовой схемы

from typing import TypedDict, Annotated, Sequence
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    retrieved_chunks: list[str]
    intermediate_answers: list[str]
    current_tool: str
    is_finished: bool

Ключевые моменты

• messages — история диалога; редьюсер add_messages автоматически добавляет новые сообщения к существующим.
• retrieved_chunks — найденные фрагменты документов; можно перезаписывать или дополнять.
• intermediate_answers — ответы на подвопросы (например, после вызова инструмента).
• current_tool — имя текущего инструмента (для логирования).
• is_finished — флаг завершения.

Редьюсеры — это функции, которые получают текущее значение и новое, и возвращают обновлённое. Встроенные: add_messages, add (для чисел), replace (по умолчанию). Можно писать свои.

4. Типичные поля состояния в Agentic RAG

Важно не хранить всё подряд. Каждое поле должно быть обосновано необходимостью для принятия решений агентом.

5. Проблемы и best practices

Проблемы

1. Разрастание состояния — если хранить все промежуточные результаты, состояние может стать огромным, замедляя передачу и увеличивая затраты на LLM (контекстное окно).
2. Конфликты при параллелизме — если агент запускает несколько веток (например, параллельные вызовы инструментов), состояние может быть повреждено.
3. Сериализация — не все объекты (например, модели, соединения) можно сериализовать. Нужно хранить только данные.
4. Восстановление после сбоя — если агент упал, состояние должно быть сохранено для перезапуска.

Best practices

• Минимализм: хранить только то, что нужно для следующих шагов. Например, не хранить полные тексты чанков, только их ID или краткое содержание.
• Явные поля: использовать TypedDict, а не общий dict с произвольными ключами.
• Редьюсеры для агрегации: если нужно накапливать данные (например, список ошибок), используйте редьюсер add или кастомный.
• Чекпоинты (checkpointing): в LangGraph встроена возможность сохранять состояние после каждого шага (persistence), что позволяет восстанавливать выполнение.
• Тестирование: писать unit-тесты на обновление состояния (проверять, что редьюсеры работают корректно).

6. Пример кода: передача состояния в LangGraph

from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, add_messages
from langchain_core.messages import HumanMessage, AIMessage

# Определяем состояние
class AgentState(TypedDict):
    messages: Annotated[Sequence, add_messages]
    retrieved_chunks: list[str]
    intermediate_answers: list[str]

# Функция-узел: поиск документов
def retrieve(state: AgentState) -> dict:
    query = state["messages"][-1].content
    chunks = vector_store.similarity_search(query, k=3)
    return {"retrieved_chunks": [chunk.page_content for chunk in chunks]}

# Функция-узел: генерация ответа
def generate(state: AgentState) -> dict:
    context = "\n".join(state["retrieved_chunks"])
    prompt = f"Context: {context}\nQuestion: {state['messages'][-1].content}"
    answer = llm.invoke(prompt)
    return {"messages": [AIMessage(content=answer)], "intermediate_answers": [answer]}

# Строим граф
builder = StateGraph(AgentState)
builder.add_node("retrieve", retrieve)
builder.add_node("generate", generate)
builder.set_entry_point("retrieve")
builder.add_edge("retrieve", "generate")
builder.set_finish_point("generate")
graph = builder.compile()

# Запуск
initial_state = {"messages": [HumanMessage(content="Что такое RAG?")],
                 "retrieved_chunks": [],
                 "intermediate_answers": []}
result = graph.invoke(initial_state)
print(result["messages"][-1].content)

Объяснение

• Состояние передаётся между узлами автоматически.
• Узел retrieve возвращает только retrieved_chunks, остальные поля остаются без изменений.
• Узел generate добавляет новое сообщение в messages и записывает промежуточный ответ.
• Редьюсер add_messages гарантирует, что история не перезаписывается, а дополняется.

7. Сравнение фреймворков для передачи состояния

Вывод для сложных Agentic RAG-систем с многошаговыми рассуждениями LangGraph — наиболее гибкий и контролируемый вариант.

8. Пет-проект для закрепления

Задача Реализовать агента, который отвечает на вопросы по документации LangChain, используя многошаговый поиск и проверку фактов.

Инструменты Python, LangGraph, LangChain, ChromaDB (векторное хранилище), OpenAI API.

Шаги:

1. Загрузить документацию LangChain (например, несколько страниц) и разбить на чанки.
2. Создать эмбеддинги и индекс в ChromaDB.
3. Определить состояние агента с полями: messages, retrieved_chunks, verified_claims, confidence.
4. Построить граф из узлов:
   • retrieve — поиск top-3 чанков.
   • verify — LLM проверяет, достаточно ли информации; если нет — возвращает need_more.
   • generate — формирует финальный ответ.
5. Добавить условное ребро: если verify вернул need_more, перейти к узлу refine_query (переформулировка запроса) и затем снова к retrieve.
6. Реализовать чекпоинты (сохранение состояния после каждого шага) для отладки.
7. Протестировать на вопросах разной сложности.

Ожидаемый результат Рабочий агент, который может самостоятельно уточнять запрос, если первого поиска недостаточно. Состояние логируется в файл для анализа.

9. Связь с другими вопросами

10. Навигация

• Предыдущий: 143
• Следующий: 145
• Индекс: 00. Индекс разборов

Навигация

• Предыдущий: 143
• Следующий: 145
• Индекс: 00. Индекс разборов