LangGraph — это фреймворк для создания сложных мультиагентных систем с управлением состояниями. В этом мануале мы создадим отказоустойчивый RAG-агент, работающий в Docker-контейнере с сохранением состояния между запусками.
Почему это важно?
- Сохранение контекста: Без персистентности агент «забывает» всё после каждого запроса
- Отказоустойчивость: При падении сервиса состояние восстанавливается из последней контрольной точки
- Масштабирование: Поддержка множества параллельных сессий (threads) для разных пользователей
Архитектура решения
Шаг 1: Структура проекта
Создадим следующую структуру директорий:
text
langgraph-rag-fault-tolerant/ ├── docker-compose.yml ├── .env ├── langgraph.json ├── agents/ │ └── rag_agent.py ├── requirements.txt └── Dockerfile
bash
mkdir -p langgraph-rag-fault-tolerant/agents cd langgraph-rag-fault-tolerant
Шаг 2: Определение графа состояний RAG-агента
Создадим агента с проверкой релевантности документов — это критически важно для RAG-систем, чтобы избежать галлюцинаций при нерелевантных запросах .
agents/rag_agent.py
python
from typing import TypedDict, Annotated, List, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Qdrant
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
import os
# === 1. Определение состояния ===
class RAGState(TypedDict):
"""Состояние графа — сохраняется в checkpoint"""
messages: Annotated[List[BaseMessage], add_messages]
documents: List[str]
is_relevant: bool
retry_count: int
# === 2. Инициализация LLM и Vector Store ===
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embeddings = OpenAIEmbeddings()
# Подключение к Qdrant (можно заменить на ваш векторный storage)
vector_store = Qdrant.from_existing_collection(
embeddings=embeddings,
collection_name="rag_docs",
url=os.getenv("QDRANT_URL", "http://localhost:6333")
)
# === 3. Определение узлов графа ===
def retrieve(state: RAGState):
"""Узел: поиск релевантных документов"""
last_message = state["messages"][-1].content
docs = vector_store.similarity_search(last_message, k=3)
return {"documents": [doc.page_content for doc in docs]}
def check_relevance(state: RAGState):
"""Узел: проверка релевантности найденных документов"""
if not state["documents"]:
return {"is_relevant": False}
prompt = f"""Оцени, отвечают ли найденные документы на вопрос пользователя.
Документы: {state['documents'][:500]}
Вопрос: {state['messages'][-1].content}
Ответь только ДА или НЕТ:"""
response = llm.invoke(prompt)
is_relevant = "да" in response.content.lower()
return {"is_relevant": is_relevant}
def generate_answer(state: RAGState):
"""Узел: генерация ответа на основе документов"""
context = "\n\n".join(state["documents"])
prompt = f"""На основе контекста ответь на вопрос пользователя.
Контекст: {context}
Вопрос: {state['messages'][-1].content}
Если в контексте нет информации, скажи об этом честно."""
response = llm.invoke(prompt)
return {"messages": [AIMessage(content=response.content)]}
def fallback_response(state: RAGState):
"""Узел: ответ без RAG (когда документы не релевантны)"""
response = llm.invoke(
f"Ответь на вопрос используя свои знания (документы не найдены): {state['messages'][-1].content}"
)
return {"messages": [AIMessage(content=response.content)], "retry_count": state.get("retry_count", 0) + 1}
def should_continue(state: RAGState) -> Literal["generate_answer", "fallback_response", "retrieve"]:
"""Условный переход: маршрутизация на основе релевантности"""
if state["is_relevant"]:
return "generate_answer"
elif state.get("retry_count", 0) < 2:
# Повторный поиск с переформулировкой (можно добавить query rewriting)
return "retrieve"
else:
return "fallback_response"
# === 4. Сборка графа ===
def create_rag_graph(checkpointer):
"""Создание графа с persistence"""
graph = StateGraph(RAGState)
# Добавление узлов
graph.add_node("retrieve", retrieve)
graph.add_node("check_relevance", check_relevance)
graph.add_node("generate_answer", generate_answer)
graph.add_node("fallback_response", fallback_response)
# Определение переходов
graph.set_entry_point("retrieve")
graph.add_edge("retrieve", "check_relevance")
graph.add_conditional_edges(
"check_relevance",
should_continue,
{
"generate_answer": "generate_answer",
"fallback_response": "fallback_response",
"retrieve": "retrieve"
}
)
graph.add_edge("generate_answer", END)
graph.add_edge("fallback_response", END)
# Компиляция с checkpointer'ом — это и есть отказоустойчивость!
return graph.compile(checkpointer=checkpointer)
# === 5. Инициализация при запуске ===
def get_app():
"""Фабрика для LangGraph Server"""
# Подключение к PostgreSQL для чекпоинтов
conn_string = os.getenv(
"DATABASE_URL",
"postgresql://postgres:postgres@postgres:5432/langgraph"
)
# Инициализация PostgresSaver с таблицей чекпоинтов
checkpointer = PostgresSaver.from_conn_string(conn_string)
checkpointer.setup() # Создание таблиц, если их нет
return create_rag_graph(checkpointer)
# Экспорт для LangGraph Server
app = get_app()
Шаг 3: Конфигурация LangGraph
langgraph.json — конфиг для LangGraph CLI:
json
{
"dependencies": ["."],
"graphs": {
"rag_agent": "./agents/rag_agent.py:app"
},
"env": ".env",
"checkpointer": {
"type": "postgres",
"conn_string": "${DATABASE_URL}"
}
}
Шаг 4: Docker-конфигурация
requirements.txt
text
langgraph>=0.2.0 langgraph-checkpoint-postgres>=0.1.0 langchain-openai>=0.1.0 langchain-community>=0.2.0 qdrant-client>=1.9.0 psycopg2-binary>=2.9.0
Dockerfile
dockerfile
FROM python:3.11-slim
WORKDIR /app
# Установка системных зависимостей
RUN apt-get update && apt-get install -y \
gcc \
libpq-dev \
&& rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Копирование кода агента
COPY agents/ ./agents/
COPY langgraph.json .
# Установка LangGraph CLI
RUN pip install langgraph-cli
EXPOSE 8123
# Запуск LangGraph сервера
CMD ["langgraph", "serve", "--host", "0.0.0.0", "--port", "8123"]
docker-compose.yml — полный стек с отказоустойчивостью:
yaml
version: '3.8'
services:
postgres:
image: postgres:15
environment:
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
POSTGRES_DB: langgraph
volumes:
- postgres_data:/var/lib/postgresql/data
ports:
- "5432:5432"
healthcheck:
test: ["CMD-SHELL", "pg_isready -U postgres"]
interval: 10s
timeout: 5s
retries: 5
restart: unless-stopped
deploy:
resources:
limits:
memory: 512M
redis:
image: redis:7-alpine
command: redis-server --appendonly yes
volumes:
- redis_data:/data
ports:
- "6379:6379"
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 10s
timeout: 5s
retries: 5
restart: unless-stopped
deploy:
resources:
limits:
memory: 256M
qdrant:
image: qdrant/qdrant:latest
volumes:
- qdrant_data:/qdrant/storage
ports:
- "6333:6333"
restart: unless-stopped
langgraph-api:
build: .
environment:
OPENAI_API_KEY: ${OPENAI_API_KEY}
DATABASE_URL: postgresql://postgres:postgres@postgres:5432/langgraph
REDIS_URL: redis://redis:6379
QDRANT_URL: http://qdrant:6333
ports:
- "8123:8123"
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_healthy
qdrant:
condition: service_started
restart: unless-stopped
deploy:
resources:
limits:
memory: 1G
cpus: '1.0'
reservations:
memory: 512M
cpus: '0.5'
volumes:
postgres_data:
redis_data:
qdrant_data:
.env
bash
OPENAI_API_KEY=sk-your-key-here # Опционально для LangSmith LANGSMITH_API_KEY=your-langsmith-key LANGSMITH_TRACING=true
Шаг 5: Запуск и тестирование
5.1 Запуск стека
bash
# Поднятие всех сервисов docker-compose up -d # Проверка статуса docker-compose ps # Просмотр логов API docker-compose logs -f langgraph-api
5.2 Тестирование через API
bash
# Создание нового потока (thread) — аналог сессии пользователя
curl -X POST http://localhost:8123/threads \
-H "Content-Type: application/json" \
-d '{}'
# Отправка запроса с привязкой к thread_id
curl -X POST http://localhost:8123/runs/stream \
-H "Content-Type: application/json" \
-d '{
"assistant_id": "rag_agent",
"thread_id": "user_123_session_1",
"input": {
"messages": [
{"role": "human", "content": "Что такое reward hacking?"}
]
}
}'
5.3 Проверка отказоустойчивости
bash
# 1. Отправляем запрос с длительной обработкой # 2. Перезапускаем API сервис docker-compose restart langgraph-api # 3. Отправляем тот же запрос с тем же thread_id # Система восстановит состояние и продолжит выполнение!
Шаг 6: Мониторинг состояния
Подключитесь к PostgreSQL и проверьте сохранённые контрольные точки:
bash
docker-compose exec postgres psql -U postgres -d langgraph
sql
-- Просмотр всех контрольных точек
SELECT
thread_id,
checkpoint_id,
parent_checkpoint_id,
checkpoint ->> 'messages' as messages
FROM checkpoints
ORDER BY checkpoint_id DESC
LIMIT 10;
-- Статистика по потокам
SELECT
thread_id,
COUNT(*) as checkpoint_count,
MAX(ts) as last_update
FROM checkpoints
GROUP BY thread_id;
Продвинутые сценарии отказоустойчивости
TTL для автоматической очистки чекпоинтов
python
from langgraph.checkpoint.postgres import PostgresSaver from datetime import timedelta checkpointer = PostgresSaver.from_conn_string(conn_string) checkpointer.setup() # Установка TTL — состояние хранится 7 дней checkpointer.set_ttl(thread_id="user_123", ttl=timedelta(days=7))
Redis для распределённых блокировок
Добавьте в docker-compose.yml:
yaml
redis:
image: redis:7-alpine
command: redis-server --appendonly yes --requirepass ${REDIS_PASSWORD}
В коде агента используйте Redis для предотвращения конкурентных запусков одного потока:
python
import redis
r = redis.Redis(host='redis', port=6379, decode_responses=True)
def with_lock(thread_id: str):
"""Декоратор для блокировки выполнения"""
def decorator(func):
def wrapper(*args, **kwargs):
lock_key = f"lock:{thread_id}"
if r.setnx(lock_key, "locked"):
r.expire(lock_key, 60) # timeout 60 сек
try:
return func(*args, **kwargs)
finally:
r.delete(lock_key)
else:
raise Exception(f"Thread {thread_id} already processing")
return wrapper
return decorator
Устранение неполадок
| Проблема | Решение |
|---|---|
PostgresSaver не найден | Установите langgraph-checkpoint-postgres |
| Чекпоинты не сохраняются | Проверьте DATABASE_URL и права на запись |
| «No checkpoint found for thread» | Убедитесь, что thread_id передан во всех запросах |
| GIL блокирует параллельные запросы | Используйте uvicorn --workers 4 для асинхронной обработки |
Заключение
Вы создали отказоустойчивый RAG-агент на LangGraph в Docker со следующими характеристиками:
✅ Персистентность состояния — PostgreSQL хранит контрольные точки, позволяя восстановить выполнение после рестарта
✅ Масштабируемость — Redis и thread_id обеспечивают изоляцию сессий пользователей
✅ Отказоустойчивость — Healthchecks и restart policies в Docker Compose
✅ RAG-логика — Проверка релевантности предотвращает галлюцинации при нерелевантных запросах
Дальнейшее улучшение
- Горизонтальное масштабирование — добавьте
langgraph-apiреплики с балансировщиком - Асинхронные узлы — перепишите узлы на async для лучшей производительности
- Мониторинг — интегрируйте Prometheus + Grafana для метрик чекпоинтов
- Шифрование — добавьте шифрование состояния в Postgres для sensitive data