Production-ready AI-агент с нуля: ReAct, Advanced RAG и работа с инструментами

Почему простые RAG-системы не работают в продакшене

Вы уже пробовали собирать AI-агента на базе простого RAG (Retrieval-Augmented Generation)? Скорее всего, столкнулись с классическими проблемами: агент "галлюцинирует", не умеет работать с инструментами, теряет контекст в длинных диалогах. Это происходит потому, что наивный RAG — это всего лишь поиск по векторной базе и генерация ответа. В реальных продакшен-системах нужен агент, который умеет:

• Планировать действия (Reasoning)
• Взаимодействовать с внешними системами (Acting)
• Использовать контекст из нескольких источников
• Обрабатывать ошибки и восстанавливаться
• Работать в многопользовательском режиме

Архитектура production-ready агента: ReAct + Advanced RAG

ReAct (Reasoning + Acting) — это парадигма, где агент циклически выполняет: Мысль → Действие → Наблюдение. В сочетании с Advanced RAG мы получаем систему, которая не просто ищет в базе знаний, а умеет планировать сложные запросы, использовать инструменты и адаптироваться к ситуации.

1 Настройка Advanced RAG с гибридным поиском

Простые эмбеддинги часто теряют ключевые слова. Гибридный поиск сочетает семантический (векторный) и лексический (BM25) поиск:

from langchain.retrievers import BM25Retriever, EnsembleRetriever
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# 1. Векторный поиск
vectorstore = Chroma(
    embedding_function=OpenAIEmbeddings(),
    persist_directory="./chroma_db"
)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 2. Лексический поиск (BM25)
texts = ["Документ 1", "Документ 2", ...]  # Ваши документы
bm25_retriever = BM25Retriever.from_texts(texts)
bm25_retriever.k = 5

# 3. Ансамбль ретриверов
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.7, 0.3]  # Больший вес векторному поиску
)

# 4. Добавляем реранкинг для улучшения качества
from langchain.llms import OpenAI
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

llm = OpenAI(temperature=0)
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=ensemble_retriever
)

2 Реализация ReAct-агента с инструментами

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

from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import Tool
from langchain import hub
from langchain.llms import OpenAI

# 1. Определяем инструменты
def search_knowledge_base(query: str) -> str:
    """Поиск в базе знаний"""
    docs = compression_retriever.get_relevant_documents(query)
    return "\n".join([doc.page_content for doc in docs])

def execute_sql_query(query: str) -> str:
    """Выполнение SQL запроса"""
    # Интеграция с вашей БД
    return "Результаты из БД"

def call_external_api(endpoint: str, params: dict) -> str:
    """Вызов внешнего API"""
    # Реализация HTTP-запроса
    return "Ответ API"

# 2. Создаем инструменты LangChain
tools = [
    Tool(
        name="KnowledgeBaseSearch",
        func=search_knowledge_base,
        description="Используй для поиска информации в базе знаний"
    ),
    Tool(
        name="SQLQuery",
        func=execute_sql_query,
        description="Используй для выполнения SQL запросов к базе данных"
    ),
    Tool(
        name="ExternalAPI",
        func=call_external_api,
        description="Используй для вызова внешних API"
    )
]

# 3. Загружаем промпт для ReAct
prompt = hub.pull("hwchase17/react")

# 4. Создаем агента
llm = OpenAI(temperature=0, model="gpt-4")
agent = create_react_agent(llm, tools, prompt)

# 5. Исполнитель агента
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True,
    max_iterations=5,  # Защита от бесконечных циклов
    early_stopping_method="generate"
)

3 Продакшен-оркестрация и управление состоянием

Агент должен работать в асинхронном режиме, сохранять историю диалогов и масштабироваться:

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from celery import Celery
import redis
import json

# Конфигурация Redis для хранения состояния
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# Celery для асинхронных задач
celery_app = Celery('agent_tasks', broker='redis://localhost:6379/0')

class AgentRequest(BaseModel):
    session_id: str
    query: str
    user_id: str

class AgentSession:
    def __init__(self, session_id: str):
        self.session_id = session_id
        self.history = []
        
    def add_to_history(self, role: str, content: str):
        self.history.append({"role": role, "content": content})
        # Сохраняем в Redis с TTL 24 часа
        redis_client.setex(
            f"agent_session:{self.session_id}",
            86400,
            json.dumps(self.history)
        )
    
    @classmethod
    def load_session(cls, session_id: str):
        data = redis_client.get(f"agent_session:{session_id}")
        if data:
            history = json.loads(data)
            session = cls(session_id)
            session.history = history
            return session
        return cls(session_id)

# FastAPI приложение
app = FastAPI(title="AI Agent API")

@celery_app.task
def process_agent_query(session_id: str, query: str):
    """Асинхронная обработка запроса агентом"""
    session = AgentSession.load_session(session_id)
    session.add_to_history("user", query)
    
    # Добавляем историю в контекст
    context = "\n".join([f"{h['role']}: {h['content']}" for h in session.history[-5:]])
    full_query = f"Context:\n{context}\n\nCurrent query: {query}"
    
    # Выполняем агента
    result = agent_executor.invoke({"input": full_query})
    
    session.add_to_history("assistant", result["output"])
    return result["output"]

@app.post("/query")
async def query_agent(request: AgentRequest, background_tasks: BackgroundTasks):
    """Эндпоинт для запросов к агенту"""
    # Запускаем асинхронную обработку
    task = process_agent_query.delay(request.session_id, request.query)
    
    return {
        "task_id": task.id,
        "status": "processing",
        "session_id": request.session_id
    }

@app.get("/result/{task_id}")
async def get_result(task_id: str):
    """Получение результата обработки"""
    task_result = celery_app.AsyncResult(task_id)
    
    if task_result.ready():
        return {"status": "completed", "result": task_result.result}
    else:
        return {"status": "processing"}

Промпт-инжиниринг для production-агентов

Промпты должны быть структурированными, детерминированными и содержать четкие инструкции:

REACT_PROMPT_TEMPLATE = """
Ты — профессиональный AI-агент. Твоя задача — решать проблемы пользователя, используя доступные инструменты.

Доступные инструменты:
{tools}

Формат ответа:
Мысль: Ты должен думать о том, что делать дальше
Действие: Название инструмента (если нужно)
Ввод действия: Входные данные для инструмента
Наблюдение: Результат выполнения инструмента
... (повторять пока не найдешь ответ)

Текущая сессия (последние 5 сообщений):
{history}

Пользователь: {input}

Начни!
"""

Мониторинг и логирование в продакшене

Без мониторинга ваш агент — черный ящик. Реализуйте комплексную систему логирования:

import structlog
from opentelemetry import trace
from prometheus_client import Counter, Histogram

# Метрики Prometheus
AGENT_REQUESTS = Counter('agent_requests_total', 'Total agent requests')
AGENT_ERRORS = Counter('agent_errors_total', 'Total agent errors')
AGENT_LATENCY = Histogram('agent_request_latency_seconds', 'Agent request latency')

# Структурированное логирование
logger = structlog.get_logger()

def log_agent_interaction(session_id: str, query: str, response: str, tools_used: list):
    """Логирование взаимодействия с агентом"""
    logger.info(
        "agent_interaction",
        session_id=session_id,
        query=query,
        response_length=len(response),
        tools_used=tools_used,
        user_feedback=None  # Можно добавить сбор фидбека
    )
    
    # Трейсинг OpenTelemetry
    tracer = trace.get_tracer(__name__)
    with tracer.start_as_current_span("agent_processing") as span:
        span.set_attribute("session_id", session_id)
        span.set_attribute("query", query)
        span.set_attribute("tools_count", len(tools_used))

# Декоратор для отслеживания
@AGENT_LATENCY.time()
def process_with_monitoring(session_id: str, query: str):
    AGENT_REQUESTS.inc()
    try:
        result = agent_executor.invoke({"input": query})
        return result
    except Exception as e:
        AGENT_ERRORS.inc()
        logger.error("agent_error", error=str(e), session_id=session_id)
        raise

Деплой и масштабирование

Для продакшена используйте Docker и оркестрацию:

# Dockerfile для AI-агента
FROM python:3.11-slim

WORKDIR /app

# Устанавливаем системные зависимости
RUN apt-get update && apt-get install -y \
    gcc \
    g++ \
    && rm -rf /var/lib/apt/lists/*

# Копируем зависимости
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Копируем код
COPY . .

# Запускаем приложение
CMD ["gunicorn", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", \
     "--bind", "0.0.0.0:8000", "main:app"]

# docker-compose.yml
version: '3.8'

services:
  agent-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - redis
      - celery-worker
    deploy:
      replicas: 3
      resources:
        limits:
          memory: 2G
        reservations:
          memory: 1G

  celery-worker:
    build: .
    command: celery -A main.celery_app worker --loglevel=info
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - redis
    deploy:
      replicas: 2

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

  prometheus:
    image: prom/prometheus
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - "9090:9090"

volumes:
  redis_data:

Частые ошибки и их решение

FAQ: Ответы на частые вопросы

Что дальше? Эволюция агентов

Современные агенты эволюционируют в сторону автономных систем. Изучите CodeAct — темную лошадку среди AI-агентов для понимания, как агенты могут писать и исполнять код. Для вдохновения посмотрите, как ИИ решает реальные проблемы в статье про "Air traffic control для больниц".

Production-ready AI-агент — это не просто скрипт, а полноценная распределенная система. Начните с простой архитектуры, добавьте мониторинг, и постепенно усложняйте. Главное — сохранять баланс между сложностью и поддерживаемостью.