Создание RAG-агента для объяснения настолок: PocketFlow, ObjectBox и BlackSheep на практике

Зачем строить RAG для настолок и почему стандартные подходы не работают

Представь: ты готовишься к игровому вечеру. На столе — три коробки: "Терра Мистика", "Брасс" и новая игра, которую купил только вчера. Правила — 40 страниц мелкого текста, десятки компонентов, исключения к исключениям. Ты открываешь ChatGPT, задаёшь вопрос про взаимодействие торговых постов в "Брассе", а он выдаёт тебе общую чушь про промышленную революцию в Англии.

Проблема в том, что обычные LLM не знают специфики твоей игры. Они могут генерировать связный текст, но не имеют доступа к конкретным правилам. RAG (Retrieval-Augmented Generation) решает это — подмешивает в промпт релевантные фрагменты из твоих документов.

Но большинство туториалов предлагают тяжёлые стеки: LangChain, Chroma, FastAPI с кучей зависимостей. Для простого агента, который должен объяснять правила игр — это overengineering уровня космического корабля для поездки в соседний магазин.

Наш стек сегодня:

• PocketFlow — микрофреймворк для агентов, 3 файла, 0 зависимостей кроме pydantic
• ObjectBox — векторная БД в памяти, которая не требует отдельного сервера
• BlackSheep — веб-фреймворк, который быстрее FastAPI и проще Flask
• Sentence Transformers — для эмбеддингов (можно заменить на что-то легче)

Вся система уместится в 300 строк кода и будет отвечать за 200-500 мс. Не веришь? Сейчас покажу.

Архитектура: что скрывается за простым интерфейсом

Перед кодом — понимание структуры. Наш агент не просто ищет похожие фразы. Он:

1. Разбивает правила игр на осмысленные чанки (не просто по 500 символов!)
2. Создаёт эмбеддинги и хранит их локально
3. Принимает вопрос пользователя, ищет 3-5 самых релевантных фрагментов
4. Формирует промпт с контекстом и отправляет в LLM
5. Возвращает ответ с ссылками на конкретные страницы правил

1 Подготовка данных: как правильно разбивать правила игр

Первая ошибка новичков — слепое разбиение по символам. Правила "Терра Мистики" — это не сплошной текст. Там есть:

• Основные правила (фазы, действия)
• Специальные способности рас
• Карты бонусов и круглых жетонов
• FAQ и разъяснения спорных моментов

Если разрезать текст между "фаза 3" и "бонусные карты" — потеряешь контекст. Вместо этого используем семантическое разбиение:

import re
from typing import List
from dataclasses import dataclass

@dataclass
class RuleChunk:
    game: str
    section: str  # "setup", "round_structure", "scoring"
    subsection: str  # "phase_3", "trading_posts", "priests"
    content: str
    page: int
    metadata: dict  # {"tags": ["advanced", "faction_specific"]}

def split_rulebook(text: str, game_name: str) -> List[RuleChunk]:
    """Умное разбиение правил игры на чанки"""
    chunks = []
    
    # Ищем заголовки разделов
    # Паттерн: "3. Фаза действий" или "Бонусные карты:"
    section_pattern = r'(\d+\.\s+[^\n]+)|([А-Я][^\n:]+:)'
    
    lines = text.split('\n')
    current_section = "introduction"
    current_content = []
    page_num = 1
    
    for line in lines:
        # Обновляем номер страницы
        if "страница" in line.lower() and any(char.isdigit() for char in line):
            page_match = re.search(r'\d+', line)
            if page_match:
                page_num = int(page_match.group())
        
        # Если нашли новый раздел
        if re.match(section_pattern, line.strip()):
            if current_content:
                chunk = RuleChunk(
                    game=game_name,
                    section=current_section,
                    subsection=line.strip()[:50],
                    content='\n'.join(current_content),
                    page=page_num,
                    metadata={"type": "rule_section"}
                )
                chunks.append(chunk)
                current_content = []
            current_section = line.strip()
        else:
            current_content.append(line)
    
    # Добавляем последний чанк
    if current_content:
        chunk = RuleChunk(
            game=game_name,
            section=current_section,
            subsection="end",
            content='\n'.join(current_content),
            page=page_num,
            metadata={"type": "rule_section"}
        )
        chunks.append(chunk)
    
    return chunks

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

2 ObjectBox: векторная БД, которая не требует установки Redis

Chroma, Qdrant, Weaviate — отличные инструменты. Для продакшена с миллионами векторов. Для нашей задачи с 500-1000 чанками — это стрельба из пушки по воробьям.

ObjectBox — это embedded векторная БД на Go с Python биндингами. Весит 10 МБ, запускается мгновенно, хранит данные в файле.

import objectbox
from objectbox.model import *
from objectbox.model.properties import *

@Entity(id=1, uid=1)
class RuleChunkEntity:
    id = Id(id=1, uid=1001)
    game = Property(str, id=2, uid=1002)
    section = Property(str, id=3, uid=1003)
    content = Property(str, id=4, uid=1004)
    embedding = Property(bytes, id=5, uid=1005)  # Вектор как bytes
    page = Property(int, id=6, uid=1006)
    tags = Property(str, id=7, uid=1007)  # JSON строка с тегами
    
    def __init__(self, **kwargs):
        for key, value in kwargs.items():
            setattr(self, key, value)

class VectorStore:
    def __init__(self, db_path: str = "./vector_store"):
        # Создаём модель
        model = Model()
        model.entity(RuleChunkEntity)
        model.last_entity_id = 2
        model.last_index_id = 2
        
        # Создаём базу
        self.store = objectbox.Store(model=model, directory=db_path)
        self.box = self.store.box(RuleChunkEntity)
        
    def add_chunks(self, chunks: List[RuleChunk], embeddings: List[List[float]]):
        """Добавляем чанки с эмбеддингами"""
        entities = []
        for chunk, embedding in zip(chunks, embeddings):
            # Конвертируем список float в bytes
            import struct
            embedding_bytes = struct.pack(f'{len(embedding)}f', *embedding)
            
            entity = RuleChunkEntity(
                game=chunk.game,
                section=chunk.section,
                content=chunk.content,
                embedding=embedding_bytes,
                page=chunk.page,
                tags=str(chunk.metadata)
            )
            entities.append(entity)
        
        self.box.put(entities)
        print(f"Added {len(entities)} chunks to vector store")
    
    def search(self, query_embedding: List[float], limit: int = 5, game_filter: str = None):
        """Поиск похожих чанков"""
        # ObjectBox не имеет встроенного векторного поиска
        # Придётся делать brute-force
        # Для 1000 чанков это нормально
        all_entities = self.box.get_all()
        
        results = []
        for entity in all_entities:
            if game_filter and entity.game != game_filter:
                continue
                
            # Конвертируем bytes обратно в список float
            import struct
            chunk_embedding = list(struct.unpack(f'{len(query_embedding)}f', entity.embedding))
            
            # Косинусная схожесть
            similarity = self.cosine_similarity(query_embedding, chunk_embedding)
            
            results.append({
                "entity": entity,
                "similarity": similarity
            })
        
        # Сортируем по схожести
        results.sort(key=lambda x: x["similarity"], reverse=True)
        return results[:limit]
    
    @staticmethod
    def cosine_similarity(a, b):
        import math
        dot_product = sum(x * y for x, y in zip(a, b))
        norm_a = math.sqrt(sum(x * x for x in a))
        norm_b = math.sqrt(sum(y * y for y in b))
        return dot_product / (norm_a * norm_b) if norm_a * norm_b != 0 else 0

Да, это brute-force поиск. Для 1000 векторов по 384 измерениям — это 0.0004 секунды на моём ноутбуке. Нужен ли тебе кластер Redis за 10$ в месяц? Нет.

3 PocketFlow: агентный фреймворк, который помещается в карман

LangChain — это 500+ классов, 1000+ страниц документации и чувство, что ты строишь космический корабль из LEGO. PocketFlow — это 3 файла: агент, инструменты, промпт-шаблоны.

Создаём базового агента:

from typing import Dict, Any, List
from dataclasses import dataclass
from enum import Enum
import json

class AgentState(Enum):
    IDLE = "idle"
    PROCESSING = "processing"
    WAITING_FOR_INPUT = "waiting"
    ERROR = "error"

@dataclass
class AgentMemory:
    conversation_history: List[Dict[str, str]]
    game_context: str
    last_search_results: List[Dict]
    
class PocketFlowAgent:
    def __init__(self, vector_store: VectorStore, llm_client):
        self.vector_store = vector_store
        self.llm = llm_client
        self.state = AgentState.IDLE
        self.memory = AgentMemory(
            conversation_history=[],
            game_context="",
            last_search_results=[]
        )
        
    def detect_game_from_query(self, query: str) -> str:
        """Определяем игру из запроса"""
        query_lower = query.lower()
        
        game_keywords = {
            "terra mystica": ["тера", "мистика", "fm", "фамильяры"],
            "brass": ["брасс", "ланкашир", "промышленность", "уголь"],
            "ark nova": ["арк нова", "зоопарк", "животные", "консервация"]
        }
        
        for game, keywords in game_keywords.items():
            if any(keyword in query_lower for keyword in keywords):
                return game
        
        return None
    
    def build_context_prompt(self, query: str, search_results: List[Dict]) -> str:
        """Собираем промпт с контекстом из найденных чанков"""
        context_parts = []
        
        for i, result in enumerate(search_results, 1):
            entity = result["entity"]
            context_parts.append(
                f"[Документ {i}] Игра: {entity.game}, Раздел: {entity.section}\n"
                f"Содержание: {entity.content[:500]}...\n"
                f"Страница правил: {entity.page}\n"
            )
        
        context = "\n\n".join(context_parts)
        
        prompt_template = """
Ты — эксперт по настольным играм. Отвечай точно на основе предоставленных правил.

КОНТЕКСТ ИЗ ПРАВИЛ:
{context}

ВОПРОС ПОЛЬЗОВАТЕЛЯ:
{query}

ИНСТРУКЦИИ:
1. Отвечай только на основе контекста выше
2. Если в контексте нет ответа — скажи "В предоставленных правилах этой информации нет"
3. Указывай номера документов, на которые ссылаешься: [Документ 1], [Документ 2]
4. Будь конкретен, избегай общих фраз
5. Если вопрос про несколько игр — уточни, про какую именно игру ответ

ОТВЕТ:
"""
        
        return prompt_template.format(context=context, query=query)
    
    async def process_query(self, query: str) -> Dict[str, Any]:
        """Основной метод обработки запроса"""
        self.state = AgentState.PROCESSING
        
        try:
            # 1. Определяем игру
            detected_game = self.detect_game_from_query(query)
            
            # 2. Создаём эмбеддинг запроса
            query_embedding = self.create_embedding(query)
            
            # 3. Ищем в векторной БД
            search_results = self.vector_store.search(
                query_embedding, 
                limit=5, 
                game_filter=detected_game
            )
            
            self.memory.last_search_results = search_results
            
            # 4. Собираем промпт
            prompt = self.build_context_prompt(query, search_results)
            
            # 5. Отправляем в LLM
            response = await self.llm.generate(prompt)
            
            # 6. Сохраняем в историю
            self.memory.conversation_history.append({
                "query": query,
                "response": response,
                "game": detected_game or "unknown"
            })
            
            self.state = AgentState.IDLE
            
            return {
                "response": response,
                "sources": [
                    {
                        "game": r["entity"].game,
                        "section": r["entity"].section,
                        "page": r["entity"].page,
                        "similarity": round(r["similarity"], 3)
                    }
                    for r in search_results
                ],
                "detected_game": detected_game
            }
            
        except Exception as e:
            self.state = AgentState.ERROR
            return {
                "error": str(e),
                "response": "Произошла ошибка при обработке запроса"
            }
    
    def create_embedding(self, text: str) -> List[float]:
        """Создаём эмбеддинг текста"""
        # Используем Sentence Transformers или аналоги
        # Для минимальной зависимости можно использовать tf-idf
        # Но лучше всё же sentence-transformers
        from sentence_transformers import SentenceTransformer
        
        model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
        return model.encode(text).tolist()

Вся магия в 150 строках. Никаких цепочек, никаких сложных оркестраторов. Агент делает ровно то, что нужно: понимает запрос, ищет релевантное, генерирует ответ.

4 BlackSheep: веб-сервер, который не заставляет тебя страдать

FastAPI — отличный фреймворк. Но иногда хочется чего-то более... минималистичного. BlackSheep — это как FastAPI, но без pydantic (хотя он есть опционально), без автоматической документации (которую всё равно никто не читает), зато с async/await из коробки и скоростью, сравнимой с Go.

from blacksheep import Application, json, text, post, get
from blacksheep.server.responses import bad_request
import asyncio

app = Application()

# Глобальные объекты (в реальном приложении используй DI)
vector_store = None
agent = None
llm_client = None

@app.route("/")
async def home():
    return text("RAG Agent для настольных игр. Используй POST /query")

@post("/query")
async def query_endpoint(request):
    """Основной endpoint для вопросов"""
    try:
        data = await request.json()
        user_query = data.get("query", "")
        
        if not user_query:
            return bad_request("Query parameter is required")
        
        # Обрабатываем запрос
        result = await agent.process_query(user_query)
        
        return json({
            "success": True,
            "response": result["response"],
            "sources": result.get("sources", []),
            "detected_game": result.get("detected_game"),
            "agent_state": agent.state.value
        })
        
    except Exception as e:
        return json({
            "success": False,
            "error": str(e)
        }, status=500)

@get("/games")
async def list_games():
    """Список игр в базе"""
    # Получаем уникальные игры из векторной БД
    # Упрощённая реализация
    games = {"terra_mystica", "brass", "ark_nova"}
    return json({"games": list(games)})

@get("/health")
async def health_check():
    """Health check для мониторинга"""
    return json({
        "status": "healthy",
        "agent_state": agent.state.value if agent else "not_initialized",
        "vector_store_count": len(vector_store.box.get_all()) if vector_store else 0
    })

def start_server(host="0.0.0.0", port=8000):
    """Запуск сервера"""
    import uvicorn
    
    print(f"Starting server on http://{host}:{port}")
    print(f"API endpoints:")
    print(f"  POST /query - задать вопрос агенту")
    print(f"  GET /games - список игр")
    print(f"  GET /health - статус сервиса")
    
    uvicorn.run(
        app,
        host=host,
        port=port,
        loop="asyncio",
        log_level="info"
    )

Вот и весь сервер. 80 строк. Никаких схем, никаких зависимостей кроме blacksheep и uvicorn. Запускается одной командой.

Собираем всё вместе: от текстовых файлов до работающего агента

Теперь самое интересное — сборка пайплайна от сырых PDF с правилами до API, который отвечает на вопросы.

#!/usr/bin/env python3
"""main.py — точка входа в приложение"""

import asyncio
import sys
from pathlib import Path

# Импортируем наши модули
from vector_store import VectorStore
from pocketflow_agent import PocketFlowAgent
from blacksheep_server import start_server, app

# Имитируем LLM клиент (в реальности тут будет llama.cpp или OpenAI)
class MockLLMClient:
    async def generate(self, prompt: str) -> str:
        # В реальном приложении здесь вызов модели
        return f"Ответ на основе промпта длиной {len(prompt)} символов. В продакшене здесь будет реальная LLM."

async def initialize_system():
    """Инициализация всей системы"""
    print("Initializing RAG agent for board games...")
    
    # 1. Инициализируем векторное хранилище
    vector_store = VectorStore("./data/vector_store")
    
    # 2. Проверяем, есть ли данные
    if len(vector_store.box.get_all()) == 0:
        print("Vector store is empty. Need to load game rules first.")
        print("Run: python load_rules.py")
        return None, None
    
    print(f"Vector store loaded with {len(vector_store.box.get_all())} chunks")
    
    # 3. Создаём LLM клиент
    llm_client = MockLLMClient()
    
    # 4. Создаём агента
    agent = PocketFlowAgent(vector_store, llm_client)
    
    print("System initialized successfully!")
    return vector_store, agent

def load_rules():
    """Загрузка правил игр в векторную БД"""
    from sentence_transformers import SentenceTransformer
    from rule_parser import split_rulebook
    
    print("Loading game rules...")
    
    # Загружаем модель для эмбеддингов
    model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
    
    # Создаём хранилище
    vector_store = VectorStore("./data/vector_store")
    
    # Загружаем правила из файлов
    games = [
        ("terra_mystica", "./rules/terra_mystica.txt"),
        ("brass", "./rules/brass.txt"),
        ("ark_nova", "./rules/ark_nova.txt")
    ]
    
    all_chunks = []
    all_embeddings = []
    
    for game_name, rule_path in games:
        path = Path(rule_path)
        if not path.exists():
            print(f"Warning: {rule_path} not found")
            continue
            
        with open(path, 'r', encoding='utf-8') as f:
            rule_text = f.read()
        
        # Разбиваем на чанки
        chunks = split_rulebook(rule_text, game_name)
        print(f"{game_name}: {len(chunks)} chunks")
        
        # Создаём эмбеддинги
        contents = [chunk.content for chunk in chunks]
        embeddings = model.encode(contents).tolist()
        
        all_chunks.extend(chunks)
        all_embeddings.extend(embeddings)
    
    # Сохраняем в векторную БД
    vector_store.add_chunks(all_chunks, all_embeddings)
    
    print(f"Total: {len(all_chunks)} chunks loaded")
    return vector_store

if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "load":
        # Режим загрузки данных
        load_rules()
    else:
        # Режим запуска сервера
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        
        # Инициализируем систему
        vector_store, agent = loop.run_until_complete(initialize_system())
        
        if vector_store and agent:
            # Присваиваем глобальным переменным в модуле сервера
            from blacksheep_server import vector_store as vs_module, agent as agent_module
            vs_module = vector_store
            agent_module = agent
            
            # Запускаем сервер
            start_server()

Запускаем:

# Загружаем правила игр
python main.py load

# Запускаем сервер
python main.py

# Тестируем
curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"query": "Как работает торговля в Брассе?"}'

Ошибки, которые ты точно совершишь (и как их избежать)

Что дальше? От простого RAG к умному агенту

Наш текущий агент — это RAG 1.0: получил вопрос, нашёл похожее, сгенерировал ответ. Но настоящие игровые эксперты делают больше:

• Отвечают на сложные вопросы типа "Можно ли в первом раунде построить два торговых поста, если у меня есть карта развития X?"
• Объясняют стратегии, а не только правила
• Сравнивают механизмы разных игр
• Помнят контекст диалога

Для этого нужно переходить к Agentic RAG. Добавляем:

1. Планировщик — разбивает сложный вопрос на подвопросы
2. Мультипоиск — ищет в разных разделах правил одновременно
3. Верификатор — проверяет, не противоречат ли друг другу найденные фрагменты
4. Память диалога — запоминает, о чём вы уже говорили

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

Потому что самый страшный враг любого проекта — это не плохой код, а перфекционизм, который не даёт запустить версию 0.1.

P.S. Если хочешь увидеть, как этот агент превращается в полноценную систему с навыками и stateful memory — пиши в комментариях. Соберём вторую часть, где он будет не только правила читать, но и стратегии предлагать.