Как создать сложную LLM-симуляцию с памятью на Python: разбор кейса Noodle Shop с GLM 4.7 и Qwen

Почему ваш LLM-агент забывает все через пять минут?

Вы потратили неделю на настройку промптов, подключили RAG, но ваш виртуальный официант путает заказы после третьего клиента. Стандартный контекст в 128K токенов - это не память, а просто буфер. Как только он переполняется, агент теряет нить разговора. Проблема не в моделях, а в архитектуре.

Мы решим это, построив систему, где агенты ведут личный дневник. Каждое взаимодействие, каждая деталь записывается, сжимается и извлекается по мере необходимости. Забудьте про одноразовые чат-боты. Мы создаем долгоживущих цифровых сотрудников.

GLM 4.7 против Qwen 2.5: выбор ядра для симуляции

На 26.02.2026 у нас есть два реалистичных кандидата для локального запуска: GLM-4.7-30B и Qwen2.5-35B. Не 400-миллиардные монстры, которые требуют отдельный дата-центр, а модели, которые можно запихнуть в сервер с парой A100.

GLM 4.7 я беру для критичных агентов, где нужен четкий структурированный вывод. Qwen 2.5 отлично работает как "мыслитель" для анализа дневниковых записей. Если вы сомневаетесь в выборе железа, мой гайд про выбор LLM под 128 ГБ VRAM расставит все по местам.

Архитектура памяти: дневник, который пишет сам себя

Вот как это работает в Noodle Shop. У агента "Официант" есть три типа памяти:

• Рабочая память: текущий диалог с клиентом (последние 10 реплик).
• Дневник: сжатая запись каждого события ("клиент А заказал острый рамен, предпочитает мало лапши").
• Профиль: извлеченные предпочтения ("клиент А - постоянный, всегда просит дополнительный яйцо").

Когда контекст рабочей памяти подходит к лимиту, система запускает процесс диаризации. LLM получает сырой диалог и инструкцию: "Сжато опиши ключевые события в JSON". Результат летит в векторную базу. При новом взаимодействии, сначала ищутся релевантные записи из дневника, затем формируется финальный промпт.

1 Код класса MemoryDiary

Не делайте так: хранить все в одном текстовом файле. Через час симуляции ваш промпт будет весить 5 мегабайт, а инференс замедлится до ползания.

import json
from datetime import datetime
from typing import List, Dict, Any
import numpy as np
from sentence_transformers import SentenceTransformer  # Актуально на 26.02.2026: используем 'all-MiniLM-L12-v2'

class MemoryDiary:
    def __init__(self, embedding_model_name='all-MiniLM-L12-v2'):
        self.entries = []  # Список записей дневника
        self.embeddings = []  # Векторные представления
        self.embedder = SentenceTransformer(embedding_model_name)
        self.index = None  # Можете использовать FAISS для скорости

    def add_entry(self, agent_name: str, event_description: str, metadata: Dict = None):
        """Добавляет запись в дневник и создает эмбеддинг."""
        entry = {
            "id": len(self.entries),
            "timestamp": datetime.now().isoformat(),
            "agent": agent_name,
            "event": event_description,
            "metadata": metadata or {}
        }
        self.entries.append(entry)
        # Создаем эмбеддинг для поиска
        text_to_embed = f"{agent_name}: {event_description}"
        embedding = self.embedder.encode(text_to_embed)
        self.embeddings.append(embedding)
        return entry

    def query_similar(self, query: str, top_k: int = 3) -> List[Dict]:
        """Ищет похожие события в дневнике."""
        query_embedding = self.embedder.encode(query)
        # Простой косинусный поиск (замените на FAISS для производства)
        similarities = []
        for idx, emb in enumerate(self.embeddings):
            sim = np.dot(query_embedding, emb) / (np.linalg.norm(query_embedding) * np.linalg.norm(emb))
            similarities.append((sim, idx))
        similarities.sort(reverse=True)
        result = []
        for sim, idx in similarities[:top_k]:
            self.entries[idx]['similarity_score'] = float(sim)
            result.append(self.entries[idx])
        return result

    def compress_old_memories(self, llm_client, max_entries: int = 50):
        """Если записей слишком много, сжимаем старые через LLM."""
        if len(self.entries) <= max_entries:
            return
        # Берем самые старые записи
        old_entries = self.entries[:10]
        prompt = f"""Сожми следующие события в одну краткую запись:
        {json.dumps(old_entries, ensure_ascii=False)}
        Верни JSON: {{"summary": "сжатое описание", "key_facts": ["факт1", "факт2"]}}"""
        # Используем LLM для сжатия
        compressed = llm_client.generate_json(prompt)
        # Заменяем старые записи одной сжатой
        new_entry = {
            "id": self.entries[0]['id'],
            "timestamp": datetime.now().isoformat(),
            "agent": "system",
            "event": compressed['summary'],
            "metadata": {"type": "compressed", "key_facts": compressed['key_facts']}
        }
        # Удаляем старые и добавляем сжатую
        self.entries = self.entries[10:]  # Удаляем 10 старых
        self.embeddings = self.embeddings[10:]
        self.add_entry(new_entry["agent"], new_entry["event"], new_entry["metadata"])

Это основа. Теперь нужно заставить LLM заполнять этот дневник. Для этого мы создадим агента с двумя режимами: "диалог" и "рефлексия".

2 Настройка JSON вывода в GLM 4.7 и Qwen 2.5

Звучит просто, но 90% разработчиков ломаются на этом этапе. Модель то выдает JSON, то вдруг начинает рассказывать историю своей жизни. Решение - использовать нативные JSON-режимы, которые появились в 2025 году.

import requests
import json

class GLMClient:
    def __init__(self, api_url: str, api_key: str = None):
        self.api_url = api_url
        self.api_key = api_key
        
    def generate_json(self, prompt: str, system_prompt: str = None) -> Dict:
        """Вызов GLM 4.7 с гарантированным JSON выводом."""
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}" if self.api_key else ""
        }
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})
        
        payload = {
            "model": "glm-4-7",  # Актуально на 26.02.2026
            "messages": messages,
            "temperature": 0.1,  # Низкая температура для стабильности
            "max_tokens": 1024,
            "response_format": {"type": "json_object"}  # Ключевой параметр!
        }
        
        response = requests.post(self.api_url, headers=headers, json=payload, timeout=30)
        response.raise_for_status()
        result = response.json()
        content = result['choices'][0]['message']['content']
        # Парсим JSON, даже если в ответе есть лишний текст
        try:
            return json.loads(content)
        except json.JSONDecodeError:
            # Экстренный парсинг: ищем JSON между ```json и ```
            import re
            json_match = re.search(r'```json\n(.*?)\n```', content, re.DOTALL)
            if json_match:
                return json.loads(json_match.group(1))
            # Если все плохо, возвращаем ошибку
            raise ValueError(f"GLM не вернул валидный JSON: {content[:200]}")

class QwenClient:
    def __init__(self, model_path: str):
        # Локальный запуск через transformers
        from transformers import AutoTokenizer, AutoModelForCausalLM
        self.tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
        self.model = AutoModelForCausalLM.from_pretrained(
            model_path,
            device_map="auto",
            torch_dtype=torch.float16,
            trust_remote_code=True
        )
        
    def generate_json(self, prompt: str) -> Dict:
        """Локальный вызов Qwen 2.5. Требует мощной видеокарты."""
        # Формируем промпт с инструкцией JSON
        json_prompt = f"""
Ты - ассистент, который всегда отвечает в формате JSON.

Запрос: {prompt}

Ответь ТОЛЬКО в виде валидного JSON, без пояснений.
JSON должен содержать поля: "action", "details", "confidence".
"""
        inputs = self.tokenizer(json_prompt, return_tensors="pt").to(self.model.device)
        with torch.no_grad():
            outputs = self.model.generate(**inputs, max_new_tokens=512, temperature=0.1)
        response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
        # Извлекаем JSON из ответа
        response = response.split("JSON:")[-1].strip()
        return json.loads(response)

3 Собираем симуляцию Noodle Shop

Теперь склеиваем все компоненты. Сценарий: клиент заходит, делает заказ, возвращается через неделю. Официант должен помнить его предпочтения.

class NoodleShopAgent:
    def __init__(self, name: str, llm_client, memory_diary: MemoryDiary):
        self.name = name
        self.llm = llm_client
        self.memory = memory_diary
        self.conversation_buffer = []  # Рабочая память
        
    def interact(self, customer_input: str, customer_id: str = "unknown") -> str:
        """Основной цикл взаимодействия."""
        # 1. Поиск в памяти
        relevant_memories = self.memory.query_similar(f"клиент {customer_id}: {customer_input}")
        
        # 2. Формирование контекста
        memory_context = "\n".join([f"- {m['event']}" for m in relevant_memories[:3]])
        
        # 3. Промпт для диалога
        prompt = f"""
Ты - {self.name}, официант в ресторане лапши.

Контекст из памяти о клиенте {customer_id}:
{memory_context if memory_context else "Это новый клиент."}

Текущий разговор (последние реплики):
{self._format_buffer()}

Клиент говорит: "{customer_input}"

Ответь естественно, как официант. Учти контекст из памяти.
"""
        
        # 4. Генерация ответа
        response = self.llm.generate(prompt)  # Упрощенный вызов
        
        # 5. Обновление буфера
        self.conversation_buffer.append(f"Клиент: {customer_input}")
        self.conversation_buffer.append(f"Официант: {response}")
        if len(self.conversation_buffer) > 10:  # Ограничиваем буфер
            self.conversation_buffer = self.conversation_buffer[-10:]
        
        # 6. Запись в дневник (асинхронно)
        self._record_to_diary(customer_input, response, customer_id)
        
        return response
    
    def _record_to_diary(self, customer_input: str, response: str, customer_id: str):
        """Записывает событие в дневник через LLM-рефлексию."""
        reflection_prompt = f"""
Опиши ключевое событие из диалога в JSON.
Диалог:
Клиент: {customer_input}
Официант: {response}

Верни JSON: {{
  "event_summary": "краткое описание",
  "customer_preference": "выявленное предпочтение клиента",
  "action_taken": "что сделал официант"
}}"""
        
        try:
            event_data = self.llm.generate_json(reflection_prompt)
            self.memory.add_entry(
                self.name,
                event_data["event_summary"],
                {
                    "customer_id": customer_id,
                    "preference": event_data["customer_preference"],
                    "action": event_data["action_taken"]
                }
            )
        except Exception as e:
            print(f"Ошибка записи в дневник: {e}")

Ошибки, которые сломают вашу симуляцию

Я видел, как эти ошибки убивали десятки проектов:

1. Слепая вера в эмбеддинги. Sentence Transformer кодирует текст, но не понимает смысл. Фраза "я ненавижу острый рамен" и "обожаю острый рамен" будут иметь похожие эмбеддинги. Добавляйте ключевые слова вручную.
2. Отсутствие сжатия памяти. Через 1000 диалогов ваш дневник превратится в помойку. Реализуйте функцию compress_old_memories, которую я показал выше. Или используйте готовые решения типа Mem0, о которых я писал в статье "Зашариваем память".
3. Игнорирование задержек. Каждый вызов LLM для рефлексии добавляет 2-3 секунды. Клиент не будет ждать. Делайте запись в дневник асинхронно, после отправки ответа.
4. Смешение языков. GLM 4.7 отлично работает с английским, Qwen 2.5 силен в китайском. Если ваши промпты на русском, а модель обучалась на другом распределении языков - ждите галлюцинаций. Тестируйте.

Что дальше? Неочевидный трюк с компрессией памяти

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

Когда клиент спрашивает "что вы мне рекомендовали в прошлый раз?", первый поиск идет по предпочтениям. Когда спрашивает "когда я последний раз был здесь?", поиск идет по временной шкале. Одна и та же запись "клиент А заказал острый рамен 20.02.2026" попадает в оба индекса.

Это требует больше места, но экономит время на промпт-инжиниринг. Вместо того, чтобы учить LLM "вспоминать даты", вы просто даете ей готовый факт из специализированного индекса.

Если ваш бюджет позволяет, используйте GLM 4.7 для критичных операций (обработка заказов, сжатие памяти), а Qwen 2.5 - для анализа настроения клиентов и генерации персонализированных предложений. Их можно запустить параллельно, как описывал в кейсе про перевод RAG-агента с OpenAI.

И последнее: не гонитесь за размером контекста. 1 миллион токенов - это круто, но если ваша память организована как мусорный бак, вы просто быстрее его заполните. Лучше 10K токенов с умной системой поиска и сжатия, чем 100K сырых диалогов.

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