Создаём AI Dungeon Master на Python: разбор open-source движка для игр с управлением состоянием через JSON

Зачем вообще нужен свой Dungeon Master?

Потому что все существующие решения либо платные, либо ограничены, либо глупые. Хочется контроля. Хочется, чтобы ИИ-мастер не забыл, что дракону пять ходов назад отрубили хвост, и чтобы история развивалась логично, а не скакала по случайным тропинкам. Главная боль – управление контекстом. LLM имеет ограниченное окно внимания, а состояние игры накапливается. Нужен механизм, который будет хранить историю, выжимать из нее суть и подавать модели в переработанном виде.

Корень проблемы: LLM не помнит, кто она и что происходит

Дашь модели чистый промпт – она начнет историю заново. Кинешь всю переписку – упрется в лимит токенов или начнет глючить. Нужен буфер. Нужна память. И самое главное – нужна единая точка истины для состояния игры. Все, от инвентаря персонажа до погоды в мире, должно жить не в промптах, а в структурированном хранилище. Отсюда и выросла идея движка, где ядро – это JSON-объект, который вертится между логикой игры, интерфейсом и LLM.

Решение: движок-посредник с JSON в сердце

Архитектура проста до безобразия. Есть класс DungeonMaster. В нем три главных компонента: State Manager (работает с JSON-состоянием), LLM Client (отправляет запросы к API) и Memory Loop (сжимает историю диалога). Интерфейс на Pygame рисует состояние и ловит действия игрока. Вся магия в том, что движок не завязан на конкретного провайдера LLM. Поддерживается любой OpenAI-совместимый API, будь то облачный сервис или локальный Ollama с моделью Llama 3.2 или новейшей DeepSeek-R1. Это дает свободу и экономит деньги.

1 Настраиваем окружение и задумываемся о структуре JSON

Создаем виртуальное окружение и ставим актуальные на 08.03.2026 библиотеки. openai версии 1.30.0+, ollama 0.5.0+, pygame 2.6.0+. Проект будет работать и на старых версиях, но зачем? Новые версии часто исправляют критические баги с потоковым выводом.

python -m venv venv
source venv/bin/activate  # или venv\\Scripts\\activate.bat на Windows
pip install openai ollama pygame httpx

Теперь самое важное – дизайн JSON-состояния. Он должен быть гибким, но не раздутым. Пример скелета:

{
  "world": {
    "name": "Эриндар",
    "time": "полдень",
    "weather": "ясно",
    "current_location": "Таверна 'Ржавый котёл'",
    "locations_visited": ["Стартовый лес", "Деревня Бри"],
    "active_quests": ["Найти пропавшего кузнеца"]
  },
  "player": {
    "name": "Аэлин",
    "race": "эльф",
    "class": "следопыт",
    "health": 85,
    "max_health": 100,
    "inventory": ["меч", "3 золотые монеты", "карта"],
    "abilities": ["Следопытство", "Маскировка"]
  },
  "npcs": {
    "tavern_keeper": {
      "name": "Барни",
      "disposition": "дружелюбный",
      "known_info": ["слышал о банде гоблинов на востоке"]
    }
  },
  "history": [
    "Вы вошли в таверну. Бармен кивает вам.",
    "Вы спросили о пропавшем кузнеце. Бармен почесал бороду."
  ],
  "last_actions": ["игрок вошел в таверну", "игрок спросил о кузнеце"],
  "summary": "Игрок только что прибыл в таверну и начал расследование."
}

Поле summary – ключевое. Это сжатое описание последних событий, которое будет вставляться в промпт вместо полной истории. Его будет обновлять наш цикл памяти.

2 Пишем ядро: класс DungeonMaster с провайдер-агностик клиентом

Класс должен уметь работать с разными LLM. Создаем базовый клиент, который определяет, локальная это модель через Ollama или облачная через OpenAI. На 08.03.2026 многие облачные API (Anthropic, Google Gemini) поддерживают OpenAI-совместимый формат, так что это универсальный подход.

import json
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
import httpx

class LLMClient(ABC):
    """Абстрактный клиент для работы с LLM."""
    @abstractmethod
    async def generate(self, prompt: str, system_prompt: str = "") -> str:
        pass

class OpenAIClient(LLMClient):
    def __init__(self, api_key: str, base_url: str = "https://api.openai.com/v1", model: str = "gpt-4.5-turbo"):
        # Используем новейшую доступную модель на 08.03.2026
        self.client = httpx.AsyncClient(headers={"Authorization": f"Bearer {api_key}"})
        self.base_url = base_url
        self.model = model

    async def generate(self, prompt: str, system_prompt: str = "") -> str:
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})

        try:
            response = await self.client.post(
                f"{self.base_url}/chat/completions",
                json={
                    "model": self.model,
                    "messages": messages,
                    "temperature": 0.8,
                    "max_tokens": 500,
                },
                timeout=30.0
            )
            response.raise_for_status()
            data = response.json()
            return data["choices"][0]["message"]["content"]
        except Exception as e:
            return f"Ошибка API: {e}"

class OllamaClient(LLMClient):
    def __init__(self, base_url: str = "http://localhost:11434", model: str = "llama3.2:latest"):
        self.client = httpx.AsyncClient()
        self.base_url = base_url
        self.model = model  # На 08.03.2026 актуальны llama3.2, deepseek-r1:latest, command-r-plus

    async def generate(self, prompt: str, system_prompt: str = "") -> str:
        try:
            payload = {
                "model": self.model,
                "prompt": prompt,
                "system": system_prompt,
                "stream": False,
                "options": {"temperature": 0.8, "num_predict": 500}
            }
            response = await self.client.post(f"{self.base_url}/api/generate", json=payload, timeout=60.0)
            response.raise_for_status()
            data = response.json()
            return data.get("response", "Нет ответа")
        except Exception as e:
            return f"Ошибка Ollama: {e}"

Зачем абстракция? Потому что завтра вы захотите перейти с платного GPT на бесплатную локальную модель. И не захотите переписывать половину кода. (Совет: если боитесь асинхронности, можно использовать потоки, но об этом позже).

3 Собираем State Manager и Memory Loop – мозги системы

State Manager загружает, сохраняет и обновляет JSON. Memory Loop – это отдельная функция, которая раз в N ходов сжимает историю. Не делайте сжатие после каждого действия – это дорого. Лучше накапливать 10-15 сообщений, затем отправлять их модели с инструкцией "создай краткое резюме последних событий". Это резюме потом кладется в поле summary, а старая история частично очищается.

class StateManager:
    def __init__(self, filepath: str = "game_state.json"):
        self.filepath = filepath
        self.state = self._load_state()

    def _load_state(self) -> Dict[str, Any]:
        try:
            with open(self.filepath, 'r', encoding='utf-8') as f:
                return json.load(f)
        except FileNotFoundError:
            # Возвращаем состояние по умолчанию
            return {
                "world": {}, "player": {}, "npcs": {},
                "history": [], "last_actions": [], "summary": ""
            }

    def save_state(self):
        with open(self.filepath, 'w', encoding='utf-8') as f:
            json.dump(self.state, f, ensure_ascii=False, indent=2)

    def update_state(self, new_data: Dict[str, Any]):
        # Рекурсивное обновление словаря, чтобы не затереть несвязанные поля
        def deep_update(target, source):
            for key, value in source.items():
                if key in target and isinstance(target[key], dict) and isinstance(value, dict):
                    deep_update(target[key], value)
                else:
                    target[key] = value
        deep_update(self.state, new_data)
        self.save_state()

class MemoryLoop:
    def __init__(self, llm_client: LLMClient, max_history_length: int = 10):
        self.llm_client = llm_client
        self.max_history = max_history_length

    async def summarize_history(self, history: list) -> str:
        if len(history) < 5:
            return " "  # Не суммируем слишком короткую историю

        prompt = "Создай очень краткое резюме следующих событий (максимум 3 предложения):\n"
        for event in history[-self.max_history:]:
            prompt += f"- {event}\n"
        prompt += "\nРезюме:"

        summary = await self.llm_client.generate(prompt, system_prompt="Ты помощник, который создает сжатые резюме.")
        return summary.strip()

4 Соединяем всё в главном классе и добавляем многопоточность

Pygame работает в главном потоке. Если мы будем ждать ответа от LLM в этом же потоке, интерфейс зависнет. Решение – вызов LLM в отдельном потоке или асинхронной задаче. Я предпочитаю asyncio с asyncio.run_in_executor для совместимости с Pygame, который не асинхронный от природы. Это выглядит страшновато, но работает.

import asyncio
from concurrent.futures import ThreadPoolExecutor

class DungeonMaster:
    def __init__(self, llm_client: LLMClient, state_file: str = "game_state.json"):
        self.llm_client = llm_client
        self.state_manager = StateManager(state_file)
        self.memory_loop = MemoryLoop(llm_client)
        self.executor = ThreadPoolExecutor(max_workers=2)
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)

    def generate_response_sync(self, player_action: str) -> str:
        """Блокирующий метод для использования из Pygame."""
        # Запускаем асинхронную задачу в отдельном потоке
        future = asyncio.run_coroutine_threadsafe(self._generate_response(player_action), self.loop)
        return future.result(timeout=45)  # Ждем ответа до 45 секунд

    async def _generate_response(self, player_action: str) -> str:
        # 1. Обновляем состояние: добавляем действие в историю
        self.state_manager.update_state({
            "last_actions": [player_action],
            "history": self.state_manager.state.get("history", []) + [f"Игрок: {player_action}"]
        })

        # 2. Формируем промпт с актуальным состоянием
        state = self.state_manager.state
        prompt = f"""
Ты мастер игры в фэнтезийном мире. Вот текущее состояние:
Мир: {state['world']}
Игрок: {state['player']}
Последние события: {state.get('summary', 'Нет данных')}

Последнее действие игрока: {player_action}

Опиши, что происходит дальше, в 2-3 предложениях. Будь креативным, но учитывай контекст.
"""

        system_prompt = "Ты креативный и внимательный мастер настольной RPG. Ты следишь за логикой мира и персонажей."

        # 3. Запрос к LLM
        response = await self.llm_client.generate(prompt, system_prompt)

        # 4. Обновляем историю ответом мастера
        self.state_manager.update_state({
            "history": state.get("history", []) + [f"Мастер: {response}"]
        })

        # 5. Проверяем, не пора ли сжать историю
        if len(state.get("history", [])) > 12:
            summary = await self.memory_loop.summarize_history(state["history"])
            if summary:
                self.state_manager.update_state({
                    "summary": summary,
                    "history": state["history"][-6:]  # Оставляем только последние 6 реплик
                })

        return response

Теперь главный цикл Pygame может вызывать dm.generate_response_sync(action), не блокируя отрисовку. Да, это костыль с двумя циклами событий, но что поделать – таковы реалии.

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

• Токены убегают. Даже с циклом памяти нужно жестко ограничивать длину промпта. Всегда считайте приблизительное количество токенов в промпте (можно использовать библиотеку tiktoken для OpenAI или эмпирически – 1 токен ≈ 4 символа для английского).
• LLM игнорирует состояние. Модель может прочитать ваш JSON и сделать вид, что его нет. Виной слабый промпт. Нужно явно указывать: "Вот текущее состояние: ... УЧТИ ЭТО В СВОЕМ ОТВЕТЕ". Используйте грамматики, как в нашем руководстве, чтобы форсировать структуру ответа.
• Ollama падает под нагрузкой. Локальные модели требуют оперативной памяти. На 08.03.2026 для более-менее качественной игры нужна хотя бы модель на 7-13 миллиардов параметров (например, Llama 3.2 11B) и 16+ ГБ ОЗУ. Иначе генерация будет медленной, а игра – прерывистой.
• JSON-состояние становится монстром. Не храните в состоянии всё подряд. Выделите только изменяемые данные. Статические описания локаций держите в отдельных файлах.
• Асинхронность vs потоки. Если вы не готовы разбираться с asyncio, используйте простые потоки (threading.Thread). Но тогда остерегайтесь состояния гонки при обновлении JSON. Используйте блокировки (threading.Lock).

Частые вопросы (FAQ)

Последний совет – не пытайтесь сделать идеально с первого раза. Соберите минимальную рабочую версию: консольный ввод, вывод, простейшее состояние. Убедитесь, что цепочка (действие -> обновление JSON -> запрос к LLM -> ответ) работает. Затем постепенно наращивайте сложность. Иначе утонете в деталях и забросите проект. Удачи в создании своих миров!

Подписаться на канал