RAG сжигает деньги — я создал систему контроля затрат, чтобы это исправить.

Большинство систем RAG оптимизируют результаты, ориентируясь на релевантность, а не на стоимость. Я разработал готовый к внедрению слой контроля затрат, объединяющий семантическое кэширование, маршрутизацию запросов и контроль бюджета, который снижает затраты на LLM на 85% без ущерба для качества ответов.

Делиться

Вкратце:

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

Системы RAG терпят неудачу не только в плане качества. Они также могут стать неэффективными с точки зрения затрат, причем зачастую это происходит не сразу.

Каждый дополнительный полученный токен имеет свою цену. В моей системе избыточная выборка контекста составляла от 3 до 8 раз больше, чем фактически требовалось для запросов.

Во многих базовых реализациях повторяющиеся запросы обрабатываются независимо, без повторного использования предыдущих результатов.

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

Благодаря семантическому кэшированию (до 98,5% попаданий в предварительно подготовленном, прогретом кэше), маршрутизации запросов (около 81% запросов перенаправлялись на более дешевую модель в тестовом наборе) и уровню бюджета токенов с автоматическим выключателем, система достигла снижения затрат до 85,8% при 10 000 запросов в день, сохраняя при этом качество ответа в оцениваемой конфигурации.

Эти результаты основаны на локальных тестах, выполненных в базовой конфигурации, описанной ниже.

Система, которая прекрасно работала, но незаметно истощала деньги.

Я создал систему RAG, которая работала безупречно: я запускал одни и те же запросы через один и тот же конвейер и каждый раз получал одинаковые результаты. В ходе тестирования никаких проблем не возникало, задержка была стабильной, а ответы — корректными.

Затем я просмотрел логи токенов.

В моей системе даже простые вопросы, такие как «Что такое RAG?» или «Определите семантический поиск», затрагивали самую ресурсоемкую модель. Каждый повторный запрос оплачивался полностью, даже если я ответил на тот же самый вопрос десять минут назад. Каждый запрос извлекал десять фрагментов данных, в то время как основную работу выполняли два запроса.

Система не была сломана. Она просто была финансово слепа. А в больших масштабах это различие перестает иметь значение.

Запустить конвейер RAG на локальном ноутбуке несложно. Но стандартная схема: получить данные, запросить информацию, вызвать систему — оставляет огромные операционные пробелы. Во многих руководствах по внедрению RAG основное внимание часто не уделяется анализу производственных затрат. В реальном мире необходимо следить за эффективностью вычислительных ресурсов и токенов. Тратите ли вы бюджет на повторную обработку того же самого запроса, который поступил на сервер три минуты назад? Действительно ли для простого поиска фактоидов необходим тот же самый ресурсоемкий и дорогостоящий путь модели, что и для многошагового запроса на логическое рассуждение?

Я уже создал слой проектирования контекста для своей предыдущей системы [7], который контролировал то, что входит в контекстное окно, по соображениям качества. Но качество и стоимость — это разные области отказа. Вы можете иметь идеальный контроль контекста и все равно заплатить в 8 раз больше, чем нужно.

Это тот уровень контроля затрат, который я построил поверх существующего — с реальными цифрами и кодом, который вы можете запустить.

Все приведенные ниже результаты получены в ходе реального запуска системы (Python 3.12.6, Windows 11, только ЦП, без ГП), за исключением случаев, когда явно указано, что они были рассчитаны.

Почему RAG намеренно действует в финансовом плане слепо.

RAG был разработан для решения проблемы качества поиска [1]. Он никогда не был разработан для решения проблемы стоимости. Это не критика — это просто другой уровень стека.

Но в процессе производства эти два слоя сталкиваются. А такое столкновение обходится дорого.

Существует три конкретных типа отказов.

Ошибка 1: Избыточная загрузка контекстного окна.

В большинстве реализаций по умолчанию выбираются 10 самых больших фрагментов. «На всякий случай».

Проблема в том, что на практике ответ содержится в 2-3 фрагментах. Остальные 7-8 — это шум, избыточный контекст, который добавляет токены, не добавляя информации. Вы каждый раз платите за эти токены.

При обработке 500 токенов на запрос, с поиском топ-10, где 7 фрагментов не требуются:

 Unnecessary tokens per query: ~350 At 10,000 requests/day: 3,500,000 unnecessary tokens/day At $0.015/1K tokens: $52.50/day in pure waste Monthly: $1,575 in unnecessary context

Это число рассчитано на основе указанных предположений, а не измерено от начала до конца.

Режим сбоя 2: Отсутствие уровня кэширования

Два пользователя задают вопрос «Что такое RAG?» с интервалом в десять минут, и система выдает одно и то же векторное представление, извлекает одни и те же фрагменты и возвращает один и тот же ответ.

Вы оплачиваете полную стоимость программы LLM дважды.

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

Режим отказа 3: Отсутствие маршрутизации модели

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

Даже когда задают вопрос: «Что означает аббревиатура LLM?»

Для решения этого вопроса не нужны GPT-4.5 или Claude Opus. Не требуется многошаговое логическое рассуждение. Не требуется контекстное окно в 200 000 символов. Необходима быстрая и недорогая модель, которая должна завершиться за 200 мс.

Используя предположения о ценообразовании в этой конфигурации, модель самого высокого уровня примерно в 90 раз дороже за токен, чем модель самого низкого уровня [2]. Учитывая, что 81% запросов в бенчмарке представляют собой простые поиски фактов, неправильная маршрутизация приводит к существенному и предотвратимому увеличению стоимости обслуживания.

Эти закономерности могут проявляться и в более простых конфигурациях RAG, особенно если не предусмотрены оптимизации с учетом стоимости.

Полный код: https://github.com/Emmimal/rag-cost-control-layer/

Реальные издержки в масштабах производства

Прежде чем что-либо строить, я хотел честно увидеть цифры.

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

 Scale Naive cost/day Optimized cost/day Saving 100 req/day $1.20 $0.18 84.6% 1,000 req/day $12.00 $1.71 85.7% 10,000 req/day $120.00 $17.00 85.8% 

При ежемесячной нагрузке 10 000 запросов в день: 3600 долларов в исходном случае против 510 долларов в оптимизированном. Экономия составляет 3090 долларов каждый месяц.

(Все цифры рассчитаны на основе указанных ценовых предположений, а не по результатам реальных вызовов API.)

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

Архитектура: четыре слоя, одна система.

Уровень контроля затрат состоит из четырех компонентов, каждый из которых нацелен на устранение различных видов отказов в системе.

Каждый слой выполняет одну единственную задачу. Вместе они обеспечивают экономичность системы на каждом этапе принятия решений.

Компонент 1: Семантический кэш

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

Как это работает

Семантическое кэширование для конвейеров LLM — это устоявшийся подход — такие инструменты, как GPTCache [8], продемонстрировали, что кэширование по семантическому сходству, а не по точному совпадению строк, может значительно сократить количество вызовов LLM. Данная реализация следует тому же принципу, используя встраиватель TF-IDF на чистом Python без внешних зависимостей.

Каждый входящий запрос встраивается с помощью векторизатора TF-IDF [3]. Кэш содержит список предыдущих пар запрос-ответ, каждая со своим встраиванием. Когда поступает новый запрос:

1. Вставьте запрос
2. Вычислить косинусное сходство по всем кэшированным эмбеддингам.
3. Если наилучшее сходство ≥ пороговое значение (по умолчанию 0,75): вернуть кэшированный ответ.
4. В случае ошибки: вызвать LLM, сохранить результат.

 class SemanticCache: def get(self, query: str) -> Optional[str]: query = self._validate(query) if query is None: return None with self._lock: self.stats.total_requests += 1 if not self._entries: self.stats.cache_misses += 1 return None q_vec = self._embedder.embed(query) best, best_sim = self._find_best(q_vec) if best is not None and best_sim >= self.threshold: best.hit_count += 1 self.stats.cache_hits += 1 self.stats.total_cost_saved_usd += self.cost_per_llm_call_usd return best.response self.stats.cache_misses += 1 return None

Для обеспечения потокобезопасности в кэше используется RLock . Встраивание каждого запроса кэшируется и пересчитывается только при изменении словаря, поэтому время поиска остается стабильным даже при больших размерах кэша.

Настройка порога

Значение по умолчанию 0,75 настроено для оценки сходства TF-IDF. Эмбеддинги Sentence-Transformer, как правило, дают более высокие показатели сходства для одного и того же совпадения, поэтому с помощью OpenAI text-embedding-3-small пороговое значение обычно смещается в сторону 0,92–0,95.

 Lower threshold → more cache hits → risk of wrong answer for edge cases Higher threshold → fewer hits → more conservative but more accurate

Правильный пороговый уровень зависит от предметной области. Для узких систем (например, ботов поддержки одного продукта или внутренних баз знаний) пороговое значение 0,70–0,75 может быть оптимальным. Для более широких систем обычно требуются более высокие пороговые значения, часто 0,90 и выше.

Реальные эталонные показатели

Выполнение 200 запросов с реалистичным соотношением (60% простых, 30% стандартных, 10% сложных, 20% повторяющихся):

 Hit rate: 98.5% Avg hit latency: ~4 ms Avg miss latency: ~4–5 ms p95 hit latency: ~5–7 ms Cost saved (200 queries): $0.788

В ходе тестирования показатель попаданий составил 98,5%, поскольку 40% запросов предварительно заполняются кэшем, имитируя прогретую производственную систему после первоначального нарастания трафика.

Разница в задержке более существенна: ~4 мс для попадания в кэш по сравнению с ~700 мс для вызова LLM — примерно 175-кратное улучшение на запрос, еще до учета экономии средств.

Производственные заметки

• max_size=1000 с вытеснением LRU по умолчанию. Увеличьте значение для систем с высокой нагрузкой.
• ttl_seconds=3600 рекомендуется для областей, где факты меняются. Установите значение None для стабильных баз знаний.
• Встраивание TF-IDF работает без каких-либо внешних зависимостей. Для создания производственных сценариев с реальным семантическим сходством замените его на API-встраивание — один из методов интерфейса, описанный в коде.

Компонент 2: Маршрутизатор запросов

Не все запросы заслуживают одной и той же модели. Маршрутизатор классифицирует каждый входящий запрос по сложности и направляет его в соответствующий уровень — автоматически, менее чем за 0,025 мс.

Три сигнала, один балл

Показатель сложности представляет собой взвешенную комбинацию трех независимых сигналов:

Показатель длины (вес: 0,20) . Нормализованное количество токенов. Запрос из 5 слов и запрос из 50 слов — это разные задачи. Насыщается при 80 токенах.

 def _length_score(self, query: str) -> float: return min(len(query.split()) / 80.0, 1.0)

Плотность сущностей (вес: 0,30) — отношение количества слов, написанных заглавными буквами, цифр и технической пунктуации к общему количеству токенов. Запросы с высокой плотностью сущностей, как правило, более специфичны и сложны.

 def _entity_score(self, query: str) -> float: tokens = query.split() if not tokens: return 0.0 hits = sum( 1 for t in tokens if (t[0].isupper() and len(t) > 1) or re.search(r"d", t) or re.search(r"[:>/%]", t) ) return min(hits / len(tokens), 1.0)

Глубина рассуждений имеет наибольший вес (0,50). Она вычисляется на основе ключевых слов, связанных с рассуждениями, таких как «сравнить», «противопоставить», «анализировать», «почему», «компромисс», «проектирование» и «архитектура». Двух совпадений достаточно, чтобы получить максимальный балл.

 REASONING_KEYWORDS: frozenset[str] = frozenset({ "compare", "contrast", "analyze", "why", "trade-off", "design", "architecture", "failure mode", "evaluate", "relationship between", "when should", "how should", ... }) def _reasoning_score(self, query: str) -> float: q_lower = query.lower() hits = sum(1 for kw in REASONING_KEYWORDS if kw in q_lower) return min(hits / 2.0, 1.0)

Быстрый путь: обнаружение фактоидов

Перед выставлением оценки маршрутизатор обнаруживает шаблоны фактов, такие как «Что такое X», «Определить X» и «Перечислить X». Эти запросы направляются напрямую по протоколу SIMPLE с фиксированной оценкой 0,10, минуя полную оценку.

 FACTOID_PATTERNS = [ re.compile(r"^(what is|what are|who is|where is)b", re.I), re.compile(r"^(define|definition of|meaning of)b", re.I), re.compile(r"^(list|name|give me)b.{0,40}$", re.I), ]

Маршрутизация на практике

Из результатов моей демонстрации:

 [Query 01] What is RAG? Tier: simple (score: 0.10) → gpt-4o-mini [Query 04] How does hybrid retrieval differ from pure vector search? Tier: standard (score: 0.306) → gpt-4o [Query 06] Compare the cost and latency trade-offs of agentic RAG versus standard Tier: standard (score: 0.611) → gpt-4o

«Что такое RAG?» — это классический факт из учебника. Он сразу же указывает на оптимальную модель и приводит к более простому решению. Вопрос «Сравните компромиссы между стоимостью и задержкой…» получает оценку 0,611 только за счет ключевых слов — это многомерный аналитический вопрос, для которого действительно необходима более сильная модель.

Эталонный тест: Распределение в масштабе

Выполнение 500 запросов к реалистичным параметрам:

 Simple: 81.0% → gpt-4o-mini ($0.000165/1K tokens) Standard: 16.4% → gpt-4o ($0.005/1K tokens) Complex: 2.6% → gpt-4.5 ($0.015/1K tokens) Total saved vs always-expensive: $3.41 (500 queries) Avg routing latency: <0.025 ms

В тестовом наборе запросов 81% трафика направляется к модели с более низкой стоимостью. Накладные расходы маршрутизатора составляют <0,025 мс на каждое решение, что на практике незначительно.

Отсутствует уровень модели — Безопасность производства

Критически важное исправление для производственной среды: если в вашем model_map отсутствует уровень, маршрутизатор не завершает работу с KeyError . Он безопасно переключается на уровень STANDARD :

 # Merge supplied map with defaults — missing keys fall back safely self.model_map = {**DEFAULT_MODEL_MAP, **(model_map or {})}

Это важно при развертывании в среде, где доступны только определенные модели. Система корректно реагирует на снижение производительности, а не аварийно завершает работу.

Компонент 3: Уровень бюджета токенов

Кэш и маршрутизатор сокращают количество и стоимость вызовов LLM. Уровень бюджета токенов обрабатывает распределение токенов для каждого вызова, предотвращает скрытое переполнение и регистрирует использование токенов.

Это напрямую основано на концепции моей системы контекстного проектирования [7], но расширяет ее за счет явного отслеживания стоимости для каждого слота.

Распределение на основе слотов

Каждый запрос резервирует токены в фиксированном порядке приоритета:

 # Reserve in priority order: fixed → history → docs → output ctx.budget.reserve("system_prompt", 200) # 1. Never negotiable ctx.budget.reserve_text("history", history) # 2. Makes multi-turn coherent ctx.budget.reserve_text("retrieved_docs", docs) # 3. What's left after fixed costs ctx.budget.reserve("output", min(512, ctx.budget.remaining())) # 4. Generation space

Порядок размещения фиксирован. Системный запрос рассматривается как накладные расходы, история сохраняет согласованность, а полученные документы представляют собой сжимаемый слой при ограниченном пространстве. Количество токенов для текстовых слотов оценивается в 1 токен ≈ 4 символа для английской прозы [6].

Если порядок документов неверен, они удаляются до того, как будет учтена история изменений. Контролер бюджета явно следит за соблюдением этого правила.

Отслеживание затрат по каждому слоту

Стоимость каждого бронирования регистрируется:

 self._slots[slot_name] = SlotUsage( name=slot_name, reserved_tokens=granted, cost_usd=granted * self._cost_per_token, )

После генерации вы записываете фактические данные:

 ctx.record_actual(actual_tokens=620, cost_usd=0.0031)

record_actual является идемпотентным. Повторные вызовы игнорируются после предупреждения, что предотвращает двойной учет в реестре расходов.

Защита отрицательного жетона

Исправление производственных ошибок, которое кажется тривиальным, но имеет значение:

 def reserve(self, slot_name: str, tokens: int) -> int: if tokens <= 0: logger.debug("reserve(%s, %d) — non-positive tokens rejected", slot_name, tokens) return 0

Если в вышестоящем процессе произойдет ошибка в расчетах и количество токенов окажется отрицательным, бюджет не станет отрицательным и не испортит все последующие вычисления. Он зафиксирует это в журнале и вернет 0.

Компонент 4: Учет затрат и автоматический выключатель.

Это недостающий слой, который защищает вашу систему от главного производственного кошмара: неконтролируемого роста затрат.

Слепое пятно производства

Вы добавляете использование инструмента к своему агенту RAG. Агент входит в цикл повторных попыток — вызов инструмента завершается неудачей, агент повторяет попытку, повторная попытка также завершается неудачей, он повторяет попытку снова. Каждый цикл представляет собой полный вызов LLM с полной стоимостью. Цикл работает в течение 6 часов ночью, пока вы спите.

Без автоматического выключателя вы проснетесь и обнаружите счет.

При наличии автоматического выключателя система автоматически ограничивает или блокирует подачу электроэнергии после достижения установленного вами почасового порога.

CostLedger: Отслеживание текущих расходов

 class CostLedger: def record(self, cost_usd, tokens, model_tier, request_id=""): event = SpendEvent(timestamp=time.time(), cost_usd=cost_usd, ...) with self._lock: self._events.append(event) self._total_lifetime_usd += cost_usd self._prune() # removes events older than 24 hours def hourly_spend(self) -> float: return self._window_spend(3600) def daily_spend(self) -> float: return self._window_spend(86400)

В реестре хранится скользящее окно событий расходования средств. _prune() удаляет события старше 24 часов, поддерживая ограничение объема памяти. Потокобезопасность обеспечивается за счет RLock .

CircuitBreaker: Три состояния [4, 5]

 CLOSED → Normal operation. All requests pass through. OPEN → Threshold breached. Requests blocked or downgraded. HALF_OPEN → Cooldown elapsed. One probe request allowed to test recovery.

 def _check_and_trip(self) -> None: if self.ledger.hourly_breach() or self.ledger.daily_breach(): self.breaker.trip()

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

Понижение рейтинга против блокировки

Два режима производства:

 enforcer = BudgetEnforcer( hourly_limit_usd=5.0, daily_limit_usd=50.0, downgrade_on_breach=True, # graceful degradation )

downgrade_on_breach=True : когда срабатывает автоматический выключатель, запросы перенаправляются на более дешевую модель вместо блокировки. Пользователи получают ухудшенное качество, а не ошибку. Для большинства производственных систем это правильный выбор.

downgrade_on_breach=False : запросы полностью блокируются с резервным сообщением. Используйте это для критически важных систем, где неправильный ответ хуже, чем его отсутствие.

Риск ложноположительных результатов — честное предупреждение

Это тот частный случай, который необходимо рассмотреть в статье. По результатам моих тестов:

 Strict threshold (hourly_limit=$0.001): → {'allowed': 0, 'downgraded': 0, 'blocked': 10} → 10/10 legitimate requests blocked Sensible threshold (hourly_limit=$5.00): → {'allowed': 10, 'downgraded': 0, 'blocked': 10} → Wait: that's wrong. Sensible threshold (hourly_limit=$5.00): → {'allowed': 10, 'downgraded': 0, 'blocked': 0} → 10/10 requests served correctly

Одна строка конфигурации. Катастрофическая разница.

Если установить слишком низкий hourly_limit , вы заблокируете собственный производственный трафик. Правило: установите лимит в 2–3 раза выше ожидаемого пикового значения, а не среднего. Средние затраты — это то, сколько всё стоит, когда всё в порядке. Лимиты защищают от скачков.

Из результатов бенчмарка: «Установите hourly_limit равным 2–3-кратному ожидаемому пиковому значению, а не среднему. Используйте downgrade_on_breach=True для плавного снижения производительности вместо блокировки пользователей».

Вся система соединений соединена между собой.

 class ProductionRAGPipeline: def __init__(self): self.cache = SemanticCache(threshold=0.75, ttl_seconds=3600) self.router = QueryRouter(simple_threshold=0.25, complex_threshold=0.65) self.enforcer = BudgetEnforcer( hourly_limit_usd=5.0, daily_limit_usd=50.0, per_request_limit_usd=0.10, downgrade_on_breach=True, ) def query(self, user_query: str, retrieved_context: str = "") -> dict: # Step 1: Cache lookup cached = self.cache.get(user_query) if cached is not None: return {"response": cached, "source": "CACHE HIT", "cost_usd": 0.0} # Step 2: Route to model tier routing = self.router.route(user_query) # Step 3: Token budget + cost enforcement with self.enforcer.request( model_tier=routing.tier.value, estimated_tokens=500, ) as ctx: if not ctx.allowed: return {"response": ctx.fallback_response, "source": "BLOCKED"} ctx.budget.reserve("system_prompt", 200) ctx.budget.reserve_text("history", "...") ctx.budget.reserve_text("retrieved_docs", retrieved_context) ctx.budget.reserve("output", min(512, ctx.budget.remaining())) response, tokens, cost = call_llm(user_query, ctx.model_tier) ctx.record_actual(actual_tokens=tokens, cost_usd=cost) # Step 4: Cache for future reuse self.cache.set(user_query, response) return {"response": response, "cost_usd": cost, "tier": routing.tier.value}

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

Что на самом деле показывает демоверсия

Запуск полного конвейера обработки 8 демонстрационных запросов (на основе моих реальных результатов):

 [Query 01] What is RAG? Source: LLM CALL | Tier: simple | Model: gpt-4o-mini Cost: $0.000015 | Saved: $0.007417 vs expensive model [Query 02] What is a vector database? Source: CACHE HIT | Saved: $0.0040 (LLM call avoided) [Query 06] Compare the cost and latency trade-offs of agentic RAG... Source: LLM CALL | Tier: standard | Model: gpt-4o Score: 0.611 | Cost: $0.000790 [Query 07] What is RAG? (repeated) Source: CACHE HIT | Saved: $0.0040 Run Summary: Total cost (8 queries): $0.001389 Total saved vs naive: $0.047668 Circuit breaker: closed

Запрос 01 и Запрос 07 — это один и тот же вопрос, заданный дважды. При втором запросе кэш возвращает результат за 0,5 мс и ничего не стоит. Это система работает именно так, как и задумано.

Запрос 06 — действительно сложный вопрос: он содержит слова «сравнение», «компромиссы» и ссылается на две архитектуры. Он получает оценку 0,611, направляется к gpt-4o и стоит 0,000790 долларов. Решение о маршрутизации верное.

Примечание о задержке: Все показатели задержки измерены с помощью имитированного вызова LLM. Реальная задержка составляет 200–800 мс на один вызов LLM в зависимости от провайдера и нагрузки. Время попадания в кэш остается примерно 4 мс независимо от этого.

Контрольные показатели: Какую реальную экономию это обеспечивает?

Все приведенные ниже цифры получены в результате реальных тестов производительности на моем компьютере (Python 3.12.6, Windows 11, только процессор).

Производительность семантического кэша

 Queries run: 200 Hit rate: 98.5% Avg hit latency: ~4 ms Avg miss latency: ~4–5 ms p95 hit latency: ~5–7 ms Cost saved (200 q): $0.788

Показатель попаданий в кэш, составляющий 98,5%, достигается при использовании прогретого кэша после нескольких часов трафика на определенном домене. Показатели попаданий при холодном запуске обычно начинаются с ~20–30% и улучшаются по мере заполнения кэша.

Распределение маршрутизатора запросов

 Queries run: 500 Simple: 81.0% → gpt-4o-mini Standard: 16.4% → gpt-4o Complex: 2.6% → gpt-4.5 Total saved: $3.41 Avg routing latency: <0.025 ms

81% запросов направляются по более дешевой модели. Этап маршрутизации добавляет менее 0,025 мс к каждому запросу и обеспечивает ощутимую экономию средств в масштабе.

Сравнение масштабов: наивный и оптимизированный подходы

В модели расчета стоимости наша базовая архитектура предполагает наихудший сценарий, полностью основанный на модели GPT-4.5 со средним количеством 800 токенов на запрос. В масштабе оптимизированная система предполагает консервативный показатель попадания в семантический кэш в 28% и направляет примерно 62% входящих запросов на более простые, недорогие модели.

 Scale Naive/day Opt/day Saving Monthly saving 100 req/day $1.20 $0.18 84.6% $30 1,000 req/day $12.00 $1.71 85.7% $309 10,000 req/day $120.00 $17.00 85.8% $3,090

Процент экономии стабилизируется на уровне ~85,8% при объеме запросов выше 1000 в день. При меньшем объеме фиксированные накладные расходы конвейера (генерация встраивания, вычисление маршрутизации) начинают иметь значение по сравнению с экономией.

Честные дизайнерские решения

TF-IDF против преобразователей предложений

Кэш использует чистый Python-встраиватель TF-IDF — без PyTorch, без преобразователей предложений и без фоновых потоков, зависающих в Windows. TF-IDF сопоставляет общие токены, а не семантическое значение.

Для одного и того же запроса, сформулированного разными словами («Что такое RAG?» против «Определение генерации с расширенными возможностями поиска»), сходство TF-IDF будет ниже, чем сходство, полученное с помощью преобразования предложений. Если ваши пользователи склонны перефразировать, а не повторять, показатель попаданий будет ниже, чем показывает этот сравнительный анализ.

Для замены на настоящий семантический эмбеддер — один из методов интерфейса:

 class OpenAIEmbedder: def fit(self, texts): pass def embed(self, text): import openai r = openai.embeddings.create(model="text-embedding-3-small", input=text) return r.data[0].embedding

Передайте это в SemanticCache , и ничего больше не изменится.

Пороговые значения маршрутизации являются эмпирическими.

Значения по умолчанию simple_threshold=0.25 и complex_threshold=0.65 откалиброваны на основе набора запросов RAG-домена. Для разных доменов, таких как юридический, медицинский или служба поддержки клиентов, требуются разные пороговые значения.

Распределение маршрутизации (81/16/2.6) отражает ориентированный на RAG набор запросов. Системы поддержки клиентов в значительной степени ориентированы на простые запросы, в то время как у помощников, ориентированных на исследования, выше доля сложных запросов.

В CostLedger отсутствует функция сохранения данных.

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

Если вы переходите к производственной среде с несколькими рабочими процессами или частыми перезапусками контейнеров, вам потребуется использовать Redis или легковесную базу данных для хранения этих данных. Сам интерфейс record() , hourly_spend() и daily_spend() — был намеренно разделен, чтобы вы могли заменить уровень хранения без переписывания логики вашего приложения.

Показатели задержки являются фиктивными.

Быстрая проверка фактов: демонстрация показывает задержки в 0,09–1,05 мс. Это отражает накладные расходы основного конвейера при имитации вызова LLM, а не реальную задержку API. В производственной среде реальный вызов LLM добавит 200–800 мс в зависимости от вашего провайдера, выбранной модели и текущей сетевой нагрузки.

Однако остальные метрики абсолютно реальны. Задержка попадания в кэш (~4 мс) реальна. Задержка принятия решения о маршрутизации (менее 0,025 мс) реальна. Накладные расходы на обеспечение соблюдения бюджета действительно незначительны. Единственный параметр, который здесь высмеивается, — это фактическое время кругового пути до поставщика LLM.

Это НЕ

Это не улучшение качества поиска. Если ваша базовая система RAG извлекает неправильные документы, этот слой не исправит это. Для улучшения качества поиска, переранжирования и сжатия контекста обратитесь к слою проектирования контекста, описанному в предыдущей статье.

Это не слой оптимизации задержки. Хотя кэш значительно снижает задержку при попадании в кэш, весь конвейер обработки данных добавляет незначительные, хотя и пренебрежимо малые, накладные расходы при промахе кэша.

Это не заменяет надлежащую систему мониторинга LLM. CostLedger служит своего рода ограничителем для отслеживания и контроля расходов, но в производственной среде вам все равно понадобятся надежные инструменты для ведения журналов, трассировки и мониторинга. Этот уровень обеспечивает прозрачность затрат, но не всеобъемлющий мониторинг.

Сборка: экономически эффективный производственный слой

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

Но системы RAG также терпят неудачу в плане стоимости. Большинство систем, ориентированных на производство, сосредоточены на качестве поиска. Эта проблема стоимости реже становится предметом внимания — и когда она возникает, то происходит незаметно. Нет ни ошибки, ни предупреждения, ни оповещения. Система продолжает работать безупречно. А счет просто продолжает расти.

Для решения этой проблемы описанная мной архитектура встраивает четыре отдельных защитных уровня между вашим конвейером получения данных и вызовом LLM:

• Семантический кэш — возвращает известные ответы менее чем за 4 мс, стоимость LLM равна 0 долларов.
• Маршрутизатор запросов — направляет 81% эталонного трафика к моделям, которые до 90 раз дешевле.
• Бюджет токенов — отслеживает каждый токен, предотвращает скрытое переполнение.
• Автоматический выключатель — автоматически ограничивает скорость передачи данных до того, как цикл повторных попыток приведет к дополнительным расходам.

Итог: совокупное снижение затрат на 85,8% при 10 000 запросах в день. В данной конфигурации оценки это соответствует приблизительной ежемесячной экономии в размере 3090 долларов, достигнутой без изменения базовой модели и без измеримого ухудшения качества ответов.

Самое приятное? Система работает на чистом Python. Никаких громоздких фреймворков, никаких sentence-transformers и никаких огромных внешних зависимостей. Она обеспечивает мгновенный запуск и корректный выход на всех платформах.

Полный код: https://github.com/Emmimal/rag-cost-control-layer/

RAG поможет вам найти правильные ответы.

Это позволит вам получить правильный счет.

Ссылки

[1] Льюис, П., Перес, Э., Пиктус, А., Петрони, Ф., Карпухин, В., Гоял, Н., Кюттлер, Х., Льюис, М., Их, В., Рокташель, Т., Ридель, С., и Киела, Д. (2020). Генерация с расширенным поиском для задач обработки естественного языка, требующих больших объемов знаний. Достижения в области нейронных информационных систем, 33, 9459–9474. https://arxiv.org/abs/2005.11401

[2] OpenAI. (2026). Цены на API OpenAI. https://openai.com/api/pricing/ (Цены могут меняться; уточняйте текущие тарифы на момент внедрения.)

[3] Педрегоса Ф., Варокво Г., Грамфор А., Мишель В., Тирион Б., Гризель О., Блондель М., Преттенхофер П., Вайс Р., Дюбур В., Вандерплас Ж., Пассос А., Курнапо Д., Брюхер М., Перро М. и Дюшенэ Э. (2011). Scikit-learn: машинное обучение на Python. Журнал исследований машинного обучения, 12, 2825–2830. https://jmlr.org/papers/v12/pedregosa11a.html (ссылка на реализацию TF-IDF.)

[4] Фаулер, М. (2002). Шаблоны архитектуры корпоративных приложений. Аддисон-Уэсли. (Шаблон автоматического выключателя.)

[5] Нюгард, М. (2007). Выпустите его! Разработка и развертывание готового к производству программного обеспечения. Прагматичная книжная полка. (Проектирование автоматического выключателя; оригинальная формулировка шаблона, использованного в этой реализации.)

[6] OpenAI. (2023). Подсчет токенов с помощью tiktoken. https://github.com/openai/tiktoken (Справочник по оценке токенов: 1 токен ≈ 4 символа для английской прозы.)

[7] Александр, Э.П. (2026). RAG недостаточно — я создал недостающий контекстный слой, который обеспечивает работу систем LLM. Towards Data Science. https://towardsdatascience.com/rag-isnt-enough-i-built-the-missing-context-layer-that-makes-llm-systems-work/ (Перекрестная ссылка: слой качества контекста; в этой статье рассматривается слой стоимости.)

[8] Банг, З. и др. (2023). GPTCache: семантический кэш с открытым исходным кодом для приложений LLM, обеспечивающий более быстрые ответы и экономию средств. https://github.com/zilliztech/GPTCache

Раскрытие информации

Весь код в этой статье написан мной и является оригинальной работой, разработанной и протестированной на Python 3.12.6, Windows 11, только на ЦП, без ГП. Система не использует внешние библиотеки машинного обучения — ни PyTorch, ни преобразователи предложений, ни numpy. Все компоненты работают только на стандартной библиотеке Python.

Результаты бенчмарков получены в ходе реальных запусков системы на моей локальной машине и полностью воспроизводимы путем клонирования репозитория и запуска файлов demo/demo.py и benchmarks/run_benchmarks.py . В демонстрационной версии используется смоделированный вызов LLM — показатели задержки ответов LLM (0,09–1,05 мс) отражают только смоделированный конвейер; реальная задержка API LLM составляет 200–800 мс в зависимости от провайдера и нагрузки. Задержка попадания в кэш (~4 мс) и задержка маршрутизации (менее 0,025 мс) измерены на основе реальной реализации на Python. Показатели стоимости сравнения масштабов (наивный и оптимизированный варианты) рассчитаны на основе известных входных данных о ценах и заявленных предположений, а не на основе реальных вызовов API.

Стоимость за 1000 токенов, использованная во всех расчетах: gpt-4o-mini ($0,000165), gpt-4o ($0,005), gpt-4.5 ($0,015). Эти цены отражают общедоступную информацию на момент написания и могут изменяться. Перед использованием этих данных для планирования бюджета проверьте текущие курсы на сайте https://openai.com/api/pricing/.

У меня нет никаких финансовых связей с OpenAI, Anthropic или любой другой компанией или инструментом, упомянутым в этой статье.

Эммимал П. Александр. Посмотреть все работы Эммимал П. Александра.

Источник: towardsdatascience.com