Контекстное окно AI-агентов: сжатие, кэширование, бюджетирование

Управление контекстным окном AI-агентов — сжатие, кэширование, summarization | QantCore

🪟

Контекстное окно AI-агентов: управление, сжатие, кэширование

Как эффективно управлять контекстным окном AI-агентов: стратегии сжатия истории диалога, prompt caching, автоматическая суммаризация, sliding window и token budgeting. Практические примеры с OpenAI, Anthropic и LangChain.

advanced ⏱ 20 мин

🚨 1. Проблема контекстного окна: лимиты, стоимость и деградация

Контекстное окно — это максимальное количество токенов, которое языковая модель может обработать за один запрос. Для AI-агентов, которые ведут длинные диалоги или работают с большими объёмами данных, управление контекстным окном становится критической задачей. Давайте разберём три ключевые проблемы.

1.1 Жёсткие лимиты моделей

Каждая модель имеет фиксированный размер контекстного окна. Когда вы превышаете лимит, API возвращает ошибку — и агент падает. Вот актуальные лимиты популярных моделей (июль 2026):

Модель	Контекстное окно	~ страниц текста
GPT-4o	128 000 токенов	~250
GPT-4o-mini	128 000 токенов	~250
Claude 3.5 Sonnet	200 000 токенов	~400
Claude 3 Opus	200 000 токенов	~400
Gemini 1.5 Pro	2 000 000 токенов	~4 000
GPT-3.5 Turbo	16 385 токенов	~30

⚠️ Важно: Даже если модель поддерживает 128K токенов, качество ответов часто деградирует при заполнении контекста свыше 50-70% от лимита. Это известная проблема «lost in the middle» — модель хуже извлекает информацию из середины длинного контекста.

1.2 Стоимость длинного контекста

Вы платите за каждый входной токен. Для GPT-4o это $2.50 за 1M входных токенов. Если агент на каждом шаге отправляет 100K токенов истории — это $0.25 за один вызов. При 100 вызовах в день вы тратите $25/день или $750/месяц только на контекст. Без оптимизации затраты растут линейно с длиной диалога.

1.3 Деградация качества ответов

Исследования показывают: когда контекст перегружен старой и нерелевантной информацией, модель начинает:

Игнорировать последние инструкции (recency bias против длины)
Путать факты из разных частей диалога
Галлюцинировать на основе устаревших данных в контексте
Давать менее точные и более общие ответы

💡 Правило эффективного контекста: держите контекст на уровне 20-40% от лимита модели. Это даёт запас для ответа и снижает стоимость на 60-80%. Для GPT-4o оптимально — 25 000–50 000 токенов.

📏 2. Подсчёт токенов и мониторинг (tiktoken, usage tracking)

Прежде чем управлять контекстом, нужно научиться его измерять. Токенизация — это разбиение текста на минимальные смысловые единицы. Один токен ≈ 0.75 слова для английского и ≈ 0.5 слова для русского языка. Начнём с базового инструментария.

2.1 Установка и базовый подсчёт с tiktoken

Библиотека tiktoken от OpenAI — это быстрый токенизатор на Rust с Python-биндингами. Он использует тот же алгоритм BPE, что и сами модели.

    BASH
    pip install tiktoken
  

    PYTHON
import tiktoken
import json

# Загружаем кодировку для GPT-4o / GPT-4 / GPT-3.5
encoding = tiktoken.get_encoding("cl100k_base")

# Простой подсчёт токенов для строки
text_ru = "Привет! Как эффективно управлять контекстным окном AI-агентов?"
tokens = encoding.encode(text_ru)
print(f"Русский текст: {len(tokens)} токенов")

text_en = "Hello! How to efficiently manage AI agent context windows?"
tokens_en = encoding.encode(text_en)
print(f"Английский текст: {len(tokens_en)} токенов")

# Декодируем обратно для проверки
decoded = encoding.decode(tokens)
print(f"Декодировано: {decoded}")
  

2.2 Подсчёт токенов для всего диалога (messages)

OpenAI использует специальную формулу для подсчёта токенов в сообщениях чата: каждый role добавляет служебные токены. Вот функция, которая считает точно так же, как API:

    PYTHON
import tiktoken
from typing import List, Dict

def count_tokens_messages(messages: List[Dict], model: str = "gpt-4o") -> int:
    """Точный подсчёт токенов для списка сообщений чата."""
    try:
        encoding = tiktoken.encoding_for_model(model)
    except KeyError:
        encoding = tiktoken.get_encoding("cl100k_base")

    tokens_per_message = 3  # служебные токены на каждое сообщение
    tokens_per_name = 1    # если есть поле "name"

    num_tokens = 3  # priming tokens: every reply is primed with <|start|>assistant<|message|>

    for msg in messages:
        num_tokens += tokens_per_message
        for key, value in msg.items():
            num_tokens += len(encoding.encode(value))
            if key == "name":
                num_tokens += tokens_per_name

    return num_tokens

# Пример использования
history = [
    {"role": "system", "content": "Ты — AI-ассистент для разработчиков."},
    {"role": "user", "content": "Как работает контекстное окно?"},
    {"role": "assistant", "content": "Контекстное окно — это лимит токенов..."},
]
total = count_tokens_messages(history)
print(f"Всего токенов в диалоге: {total}")
print(f"Лимит GPT-4o: 128 000, заполнено: {total/128000*100:.1f}%")
  

2.3 Мониторинг использования токенов через API

Каждый ответ от OpenAI содержит детальную статистику в объекте usage. Собирайте эту информацию для аудита:

    PYTHON
from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Привет!"}]
)

usage = response.usage
print(f"📥 Входные токены:  {usage.prompt_tokens}")
print(f"📤 Выходные токены: {usage.completion_tokens}")
print(f"💰 Всего токенов:   {usage.total_tokens}")

# Расчёт стоимости (GPT-4o: $2.50/1M input, $10.00/1M output)
cost = (usage.prompt_tokens / 1_000_000) * 2.50 + \
       (usage.completion_tokens / 1_000_000) * 10.00
print(f"💵 Стоимость запроса: ${cost:.6f}")
  

🔧 Совет для production: логируйте usage.total_tokens на каждом шаге агента и стройте графики в Grafana/Datadog. При достижении порога (например, 80% от лимита) триггерьте сжатие контекста или переключение на более ёмкую модель. Это предотвращает падение агента в production.

🪟 3. Sliding Window: простая стратегия с ограничением истории

Sliding Window — это самая простая и надёжная стратегия управления контекстом. Идея: мы храним только последние N сообщений или последние K токенов, а всё что старше — отбрасываем. Это решает проблему бесконечного роста истории ценой потери старого контекста.

3.1 Базовая реализация: ограничение по количеству сообщений

    PYTHON
from typing import List, Dict

class SlidingWindowChat:
    """Простой sliding window: храним последние max_messages сообщений."""

    def __init__(self, system_prompt: str, max_messages: int = 20):
        self.system_prompt = system_prompt
        self.max_messages = max_messages
        self.history: List[Dict] = []

    def add_message(self, role: str, content: str):
        self.history.append({"role": role, "content": content})

        # Оставляем только последние max_messages
        if len(self.history) > self.max_messages:
            trimmed = self.history[:-self.max_messages]
            self.history = self.history[-self.max_messages:]
            print(f"🪟 Удалено {len(trimmed)} старых сообщений")

    def get_context(self) -> List[Dict]:
        """Возвращает контекст: system prompt + последние сообщения."""
        return [
            {"role": "system", "content": self.system_prompt}
        ] + self.history

# Пример работы
chat = SlidingWindowChat(
    system_prompt="Ты — AI-агент для работы с кодом.",
    max_messages=6
)

for i in range(10):
    chat.add_message("user", f"Сообщение пользователя №{i+1}")
    chat.add_message("assistant", f"Ответ ассистента №{i+1}")

print(f"Всего в контексте: {len(chat.get_context())} сообщений (system + история)")
# → 7 сообщений: 1 system + 6 последних из истории
  

3.2 Продвинутая версия: ограничение по токенам

Ограничение по количеству сообщений неточно: одно сообщение может содержать 10 токенов, другое — 10 000. Правильнее использовать лимит по токенам:

    PYTHON
import tiktoken
import json
from typing import List, Dict

class TokenWindowChat:
    """Sliding window с ограничением по токенам (более точный контроль)."""

    def __init__(self, system_prompt: str, max_tokens: int = 8000,
                 model: str = "gpt-4o"):
        self.system_prompt = system_prompt
        self.max_tokens = max_tokens
        self.encoding = tiktoken.encoding_for_model(model)
        self.history: List[Dict] = []

    def _count_tokens(self, messages: List[Dict]) -> int:
        text = json.dumps(messages, ensure_ascii=False)
        return len(self.encoding.encode(text))

    def add_message(self, role: str, content: str):
        self.history.append({"role": role, "content": content})

        # Вычисляем токены для system prompt
        system_tokens = self._count_tokens([
            {"role": "system", "content": self.system_prompt}
        ])

        # Удаляем старые сообщения с начала, пока общее число токенов > лимита
        total = system_tokens + self._count_tokens(self.history)
        dropped = 0
        while total > self.max_tokens and len(self.history) > 2:
            self.history.pop(0)
            dropped += 1
            total = system_tokens + self._count_tokens(self.history)

        if dropped:
            print(f"🪟 Обрезано {dropped} сообщений (токенов: {total}/<{self.max_tokens})")

    def get_context(self) -> List[Dict]:
        return [{"role": "system", "content": self.system_prompt}] + self.history
  

3.3 Плюсы и минусы Sliding Window

✅ Плюсы	❌ Минусы
Предельно простая реализация	Безвозвратная потеря старого контекста
Гарантированное соблюдение лимита	Агент «забывает» важные ранние инструкции
Предсказуемая стоимость	Не различает важные и неважные сообщения
Минимальные накладные расходы	Неэффективно для длинных многошаговых задач

⚠️ Когда использовать: Sliding Window хорош для простых чат-ботов и краткосрочных задач (до 20-30 шагов). Для агентов с долгой памятью и сложными цепочками действий нужна суммаризация (секция 4).

📝 4. Автоматическая суммаризация диалога (LLM-based summarization)

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

4.1 Принцип работы: сжатие с сохранением смысла

Алгоритм: когда история превышает порог (например, 60% от лимита), мы берём самую старую часть сообщений, отправляем их дешёвой модели (например, GPT-4o-mini) с промптом «Суммаризируй этот диалог», и заменяем старые сообщения на одно summary-сообщение.

    PYTHON
from openai import OpenAI
import tiktoken
import json
from typing import List, Dict, Optional

class SummarizingChat:
    """Чат с автоматической суммаризацией длинной истории."""

    SUMMARY_PROMPT = """Суммаризируй следующий фрагмент диалога. Сохрани:
1. Ключевые решения и выводы
2. Важные факты и цифры
3. Действия, которые были выполнены
4. Контекст текущей задачи

Не включай: приветствия, вежливости, повторяющуюся информацию.

Диалог для суммаризации:
{conversation}

Суммаризация (на русском):"""

    def __init__(
        self,
        system_prompt: str,
        max_tokens: int = 8000,
        summary_trigger_ratio: float = 0.6,  # 60% от лимита — триггер суммаризации
        model: str = "gpt-4o",
        summarizer_model: str = "gpt-4o-mini",  # дешёвая модель для суммаризации
        client: Optional[OpenAI] = None
    ):
        self.system_prompt = system_prompt
        self.max_tokens = max_tokens
        self.summary_trigger = int(max_tokens * summary_trigger_ratio)
        self.model = model
        self.summarizer_model = summarizer_model
        self.client = client or OpenAI()
        self.encoding = tiktoken.encoding_for_model(model)
        self.history: List[Dict] = []
        self.summary: Optional[str] = None

    def _count_tokens(self, messages: List[Dict]) -> int:
        text = json.dumps(messages, ensure_ascii=False)
        return len(self.encoding.encode(text))

    def _summarize(self, messages: List[Dict]) -> str:
        """Отправляем старые сообщения дешёвой модели для суммаризации."""
        conv_text = "\n".join(
            f"[{m['role']}] {m['content']}" for m in messages
        )

        response = self.client.chat.completions.create(
            model=self.summarizer_model,
            messages=[{
                "role": "user",
                "content": self.SUMMARY_PROMPT.format(conversation=conv_text)
            }],
            temperature=0.3,
            max_tokens=1000
        )
        summary_text = response.choices[0].message.content
        tokens_before = self._count_tokens(messages)
        tokens_after = len(self.encoding.encode(summary_text))
        print(f"📝 Суммаризация: {tokens_before} → {tokens_after} токенов (сжатие {(1 - tokens_after/tokens_before)*100:.0f}%)")
        return summary_text

    def add_message(self, role: str, content: str):
        self.history.append({"role": role, "content": content})
        self._maybe_summarize()

    def _maybe_summarize(self):
        """Проверяем, не пора ли суммаризировать."""
        total = self._count_tokens(self.history)

        if total < self.summary_trigger or len(self.history) < 6:
            return

        # Берём старейшие 50% сообщений для суммаризации
        split_point = len(self.history) // 2
        old_messages = self.history[:split_point]

        # Суммаризируем и заменяем на одно summary-сообщение
        new_summary = self._summarize(old_messages)
        self.summary = new_summary
        self.history = self.history[split_point:]

        print(f"📝 Контекст после суммаризации: {len(self.history)} живых сообщений + summary")

    def get_context(self) -> List[Dict]:
        """Собирает финальный контекст: system + summary + история."""
        messages = [{"role": "system", "content": self.system_prompt}]

        if self.summary:
            messages.append({
                "role": "system",
                "content": f"[Резюме предыдущего диалога]\n{self.summary}"
            })

        messages.extend(self.history)
        return messages
  

4.2 Каскадная (рекурсивная) суммаризация

Для очень длинных диалогов одной суммаризации недостаточно. Используйте каскадный подход: когда накапливается несколько summary, суммаризируйте и их. Это похоже на map-reduce для контекста.

    PYTHON
class CascadingSummarizer:
    """Многоуровневая суммаризация: summary → meta-summary."""

    def __init__(self, client: OpenAI, max_summaries: int = 5):
        self.client = client
        self.max_summaries = max_summaries
        self.summaries: List[str] = []       # summary первого уровня
        self.meta_summary: Optional[str] = None # meta-summary второго уровня

    def add_summary(self, summary: str):
        self.summaries.append(summary)

        if len(self.summaries) >= self.max_summaries:
            combined = "\n---\n".join(self.summaries)
            response = self.client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{
                    "role": "user",
                    "content": f"Объедини эти резюме диалогов в одно мета-резюме:\n\n{combined}"
                }],
                temperature=0.3,
                max_tokens=1500
            )
            self.meta_summary = response.choices[0].message.content
            self.summaries = []  # сбрасываем первый уровень
            print("📝📝 Создано мета-резюме (уровень 2)")

    def get_context_block(self) -> str:
        parts = []
        if self.meta_summary:
            parts.append(f"[Мета-резюме всей истории]\n{self.meta_summary}")
        if self.summaries:
            recent = "\n---\n".join(self.summaries[-3:])
            parts.append(f"[Недавние резюме]\n{recent}")
        return "\n\n".join(parts)
  

💡 Совет: используйте GPT-4o-mini для суммаризации — она в 20 раз дешевле GPT-4o ($0.15/1M токенов против $2.50/1M). Даже агрессивная суммаризация на каждом 10-м шаге добавляет менее 1% к общей стоимости диалога. Дополнительно можно использовать structured output (JSON mode) для гарантированного формата резюме.

💾 5. Prompt Caching (Anthropic, OpenAI) — кэширование системных промптов

Prompt Caching — это функция, при которой провайдер кэширует неизменяемую часть вашего промпта (обычно system prompt, инструкции, примеры few-shot) и не тарифицирует её повторную обработку. Это даёт экономию 50-90% на входных токенах при повторяющихся запросах.

5.1 Anthropic Prompt Caching

Anthropic был первым, кто запустил prompt caching. Вы помечаете блоки промпта атрибутом cache_control, и Claude кэширует их на 5 минут (таймер сбрасывается при каждом использовании). Стоимость cache read — всего 10% от базовой цены входных токенов.

    PYTHON
import anthropic

client = anthropic.Anthropic()

# Длинный system prompt (типичный для AI-агента)
SYSTEM_PROMPT = """Ты — AI-агент для анализа кода. Твои правила:

1. Всегда проверяй безопасность кода перед выполнением
2. Используй Python 3.11+ с type hints
3. Для файловых операций используй pathlib
4. Логируй все действия через structlog
5. При ошибках предлагай 3 варианта исправления
6. ...

[много других инструкций]"""

# Создаём запрос с кэшированием system prompt
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=4096,
    system=[
        {
            "type": "text",
            "text": SYSTEM_PROMPT,
            # 👇 помечаем для кэширования
            "cache_control": {"type": "ephemeral"}
        }
    ],
    messages=[
        {"role": "user", "content": "Напиши функцию для парсинга JSON."}
    ]
)

# Проверяем, что попало в кэш
if response.usage.cache_creation_input_tokens:
    print(f"💾 Создан кэш: {response.usage.cache_creation_input_tokens} токенов")
if response.usage.cache_read_input_tokens:
    print(f"💾 Чтение из кэша: {response.usage.cache_read_input_tokens} токенов")
    # Эти токены стоят в 10 раз дешевле!
print(f"💰 Обычные входные: {response.usage.input_tokens} токенов")
  

5.2 OpenAI Prompt Caching (автоматический)

OpenAI внедрил автоматическое кэширование — вам не нужно ничего помечать. Система сама кэширует префиксы промптов, которые повторяются между запросами. Кэш живёт 5-10 минут, скидка — 50% на закэшированные токены.

    PYTHON
from openai import OpenAI

client = OpenAI()

# OpenAI кэширует автоматически — достаточно повторить запрос
# с тем же длинным system prompt в течение 5-10 минут

def query_with_cache(user_message: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},  # кэшируется
            {"role": "user", "content": user_message}
        ]
    )

    usage = response.usage
    # Проверяем детали кэширования (доступно в usage)
    cached = getattr(usage, 'prompt_tokens_details', None)
    if cached and getattr(cached, 'cached_tokens', 0):
        print(f"💾 Закэшировано: {cached.cached_tokens} токенов (скидка 50%)")

    return response.choices[0].message.content

# Первый запрос — полная цена за system prompt
query_with_cache("Задача 1: напиши функцию сортировки")
# Второй запрос — system prompt уже в кэше, скидка 50%
query_with_cache("Задача 2: оптимизируй эту функцию")
  

5.3 Стратегия размещения кэшируемых блоков

Ключевое правило prompt caching: кэшируются только префиксы — непрерывные блоки от начала промпта. Если вы измените сообщение в середине, кэш инвалидируется. Поэтому располагайте статический контент (правила, примеры, инструкции) в начале, а динамический (история диалога) — в конце.

Стратегия	Anthropic	OpenAI
Активация	Явная (`cache_control`)	Автоматическая
Срок жизни кэша	5 мин (сбрасывается при использовании)	5-10 мин
Скидка	90% на чтение кэша	50% на закэшированные токены
Мин. токенов для кэша	1024 (Claude 3.5+), 2048 (Claude 3)	1024
Макс. кэшируемых блоков	4 блока на запрос	Автоматически

🔧 Практический совет: для AI-агентов с длинным system prompt (5000+ токенов) prompt caching окупается с первого повторного вызова. При 100 запросах в день экономия на Anthropic составляет ~$4.50/день только на system prompt. Размещайте few-shot примеры и документацию API в начале промпта — они тоже попадут в кэш.

💰 6. Token Budgeting и приоритизация контекста

Token Budgeting — это стратегия, при которой мы не просто обрезаем контекст по лимиту, а осознанно распределяем токеновый «бюджет» между разными типами контента по их важности. Важные сообщения (решения, выводы, ключевые факты) сохраняются, а второстепенные — сжимаются или удаляются.

6.1 Распределение бюджета

Типичный бюджет для агента с окном 128K токенов:

Категория	Бюджет	Токенов	Стратегия при переполнении
System prompt	5-10%	6 000–12 000	Кэширование (секция 5)
Few-shot примеры	10-15%	12 000–19 000	Кэширование, ротация
История диалога	30-40%	38 000–51 000	Суммаризация (секция 4)
Инструменты/схемы	10-15%	12 000–19 000	Динамическая подгрузка
Резерв для ответа	20-25%	25 000–32 000	Всегда свободен

6.2 Реализация Token Budget Manager

    PYTHON
import tiktoken
import json
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable
from enum import Enum

class Priority(Enum):
    CRITICAL = 0   # никогда не удаляется (system prompt, решения)
    HIGH = 1       # важные факты, выводы
    MEDIUM = 2     # обычные сообщения диалога
    LOW = 3        # приветствия, вежливости, вода

@dataclass
class BudgetMessage:
    role: str
    content: str
    priority: Priority = Priority.MEDIUM
    tokens: int = 0  # будет вычислено

@dataclass
class TokenBudget:
    total_limit: int = 128000
    system_prompt_pct: float = 0.08
    few_shot_pct: float = 0.12
    history_pct: float = 0.35
    tools_pct: float = 0.12
    response_reserve_pct: float = 0.20

    @property
    def history_budget(self) -> int:
        return int(self.total_limit * self.history_pct)

class TokenBudgetManager:
    """Управляет контекстом на основе приоритетов."""

    def __init__(self, budget: TokenBudget, model: str = "gpt-4o"):
        self.budget = budget
        self.encoding = tiktoken.encoding_for_model(model)
        self.system_prompt: List[BudgetMessage] = []
        self.history: List[BudgetMessage] = []

    def set_system(self, content: str):
        self.system_prompt = [
            BudgetMessage(
                role="system", content=content,
                priority=Priority.CRITICAL,
                tokens=len(self.encoding.encode(content))
            )
        ]

    def add_message(self, role: str, content: str,
                    priority: Priority = Priority.MEDIUM,
                    classifier: Optional[Callable] = None):
        # Автоматическая классификация приоритета
        if classifier:
            priority = classifier(content)
        elif priority == Priority.MEDIUM:
            priority = self._auto_priority(content)

        msg = BudgetMessage(
            role=role, content=content,
            priority=priority,
            tokens=len(self.encoding.encode(content))
        )
        self.history.append(msg)
        self._enforce_budget()

    def _auto_priority(self, content: str) -> Priority:
        """Эвристическая классификация важности сообщения."""
        content_lower = content.lower()

        # Ключевые слова HIGH приоритета
        high_keywords = [
            "решение", "вывод", "ошибка", "exception",
            "результат", "итог", "важно", "critical",
            "деплой", "production", "финальный"
        ]
        if any(kw in content_lower for kw in high_keywords):
            return Priority.HIGH

        # LOW приоритет — короткие/вежливые сообщения
        if len(content) < 20:
            return Priority.LOW

        return Priority.MEDIUM

    def _enforce_budget(self):
        """Удаляет сообщения с низким приоритетом при превышении бюджета."""
        total_history = sum(m.tokens for m in self.history)

        if total_history <= self.budget.history_budget:
            return

        # Сначала удаляем LOW, потом MEDIUM — но не CRITICAL и HIGH
        for priority in [Priority.LOW, Priority.MEDIUM]:
            if total_history <= self.budget.history_budget:
                break

            to_remove = [
                i for i, msg in enumerate(self.history)
                if msg.priority == priority
            ]

            for idx in reversed(to_remove):
                if total_history <= self.budget.history_budget:
                    break
                removed = self.history.pop(idx)
                total_history -= removed.tokens

        dropped = sum(1 for m in self.history if m.priority == Priority.LOW)
        print(f"💰 Budget: {total_history}/{self.budget.history_budget} токенов истории")

    def get_context(self) -> List[Dict]:
        """Собирает финальный контекст."""
        messages = [{"role": m.role, "content": m.content}
                    for m in self.system_prompt + self.history]
        total = self.encoding.encode(json.dumps(messages, ensure_ascii=False))
        print(f"📊 Итоговый контекст: {len(total)} токенов из {self.budget.total_limit} доступных")
        return messages

# Пример использования
budget = TokenBudget(total_limit=8000)  # маленький бюджет для демонстрации
manager = TokenBudgetManager(budget)
manager.set_system("Ты — AI-агент для DevOps задач.")

manager.add_message("user", "Привет!")                          # → LOW
manager.add_message("assistant", "Здравствуйте! Чем могу помочь?") # → LOW
manager.add_message("user", "Ошибка деплоя в production — critical!")  # → HIGH
manager.add_message("assistant", "Фиксирую: ошибка 500 на эндпоинте /api/v2. Вывод: нужно откатить миграцию.") # → HIGH

context = manager.get_context()
# LOW-приоритетные сообщения будут удалены первыми при превышении бюджета
  

6.3 Комбинирование всех стратегий в одном пайплайне

Идеальный production-пайплайн комбинирует все пять стратегий:

    PYTHON
class ProductionContextPipeline:
    """Объединяет все стратегии в одном классе."""

    def __init__(self, system_prompt: str, model: str = "gpt-4o"):
        self.model = model
        self.encoding = tiktoken.encoding_for_model(model)

        # Компоненты
        self.token_counter = TokenBudgetManager(
            TokenBudget(total_limit=40000), model
        )
        self.summarizer = SummarizingChat(
            system_prompt, max_tokens=40000, model=model
        )
        self.slider = TokenWindowChat(
            system_prompt, max_tokens=40000, model=model
        )

        self.token_counter.set_system(system_prompt)

    def process(self, messages: List[Dict]) -> List[Dict]:
        """Основной пайплайн обработки контекста."""
        total_tokens = len(self.encoding.encode(
            json.dumps(messages, ensure_ascii=False)
        ))

        strategy = self._select_strategy(total_tokens)

        if strategy == "cache":
            print("💾 Использую prompt caching (system prompt)")
            pass  # system prompt кэшируется автоматически

        elif strategy == "sliding_window":
            print("🪟 Применяю sliding window")

        elif strategy == "summarize":
            print("📝 Запускаю суммаризацию")

        elif strategy == "budget":
            print("💰 Применяю token budgeting")

        return messages  # обработанный контекст

    def _select_strategy(self, total_tokens: int) -> str:
        """Выбор стратегии в зависимости от заполнения контекста."""
        ratio = total_tokens / 128000

        if ratio < 0.15:
            return "cache"           # кэширование — всегда активно
        elif ratio < 0.40:
            return "sliding_window"  # простое окно
        elif ratio < 0.65:
            return "budget"          # приоритетное распределение
        else:
            return "summarize"       # максимальное сжатие
  

🏆 Итоговая рекомендация: начните с простого sliding window, добавьте мониторинг токенов (секция 2). Когда агент начнёт «забывать» важное — внедрите суммаризацию (секция 4). Для production с высокой нагрузкой подключите prompt caching (секция 5) и token budgeting (эта секция). Комбинированный подход даёт экономию 60-80% на стоимости контекста при сохранении качества ответов.

🎯 Быстрый чек-лист: выбор стратегии

Ситуация	Рекомендуемая стратегия	Экономия
Короткие диалоги (до 20 шагов)	Sliding Window	0% (просто контроль)
Длинные диалоги с важным контекстом	Суммаризация (GPT-4o-mini)	40-60%
Много запросов с одним system prompt	Prompt Caching	50-90% на system
Разнородный контекст (код + диалог + данные)	Token Budgeting	30-50%
Production-агент с высокой нагрузкой	Все стратегии вместе	60-80%

Помните: управление контекстным окном — это не разовая настройка, а непрерывный процесс. Мониторьте usage.total_tokens на каждом шаге, анализируйте паттерны использования и адаптируйте стратегии под конкретные сценарии вашего агента. Правильное управление контекстом превращает агента из дорогой игрушки в эффективный production-инструмент.