Поисковый контур с генерацией ответов: что реально ломается
Chunking, качество эмбеддингов, latency retrieval, hallucination под нагрузкой.
Это инженерный материал о том, как считать и контролировать затраты на языковые модели. Он не описывает выполненный клиентский проект и не содержит универсального обещания экономии: пороги, цены и качество нужно измерять на собственной системе.
Опишите один и тот же период и одинаковый набор задач: число запросов, входные и выходные токены, используемые модели, долю ошибок и задержку. Разделяйте стоимость модели, инфраструктуры, хранения и повторных запросов. Без такой базы оптимизация легко превращается в сравнение несопоставимых величин.
Повторяющиеся запросы можно обслуживать из кэша, если это не нарушает актуальность и права доступа. Сравнивайте не только долю попаданий, но и качество ответа, срок действия записи, размер контекста и стоимость проверки сходства. Порог сходства выбирайте на контрольной выборке, а не переносите из другого проекта.
class SemanticCache:
def __init__(self, threshold=0.95):
self.threshold = threshold
self.cache = [] # [(embedding, response)]
def get(self, query_embedding):
for emb, response in self.cache:
sim = cosine_similarity(query_embedding, emb)
if sim >= self.threshold:
return response
return None
def set(self, embedding, response):
self.cache.append((embedding, response))
if len(self.cache) > 10_000:
self.cache.pop(0) # LRU
Простая классификация и сложное извлечение могут требовать разных моделей. Составьте явные классы сложности, определите допустимое качество для каждого класса и проверяйте маршрутизатор на отложенной выборке. Названия моделей и пороги должны соответствовать доступному вам API и регулярно пересматриваться.
COMPLEXITY_PROMPT = """
Оцени сложность задачи: 'low', 'medium', 'high'.
Low: простая классификация, короткий текст.
Medium: извлечение сущностей, средний текст.
High: сложные зависимости, длинный контекст, рассуждения.
Ответь одним словом.
"""
async def route_request(task, context):
complexity = await classify_complexity(task, context)
model = MODEL_MAP[complexity]
return await call_llm(model, task, context)
Для задач без требования к немедленному ответу сравните пакетную обработку с обычными запросами. Зафиксируйте допустимую задержку, лимиты провайдера, порядок повторов и приоритеты очереди. Условия конкретного API, скидки и сроки выполнения нужно проверять по его актуальной документации.
Если задача повторяется и имеет устойчивую разметку, сравните вызов универсальной модели с узкой моделью или дообучением. До запуска определите набор проверки, минимально допустимое качество, стоимость подготовки данных и план отката. Уменьшение цены не компенсирует ошибки в критическом контуре.
Соберите небольшую контрольную выборку и меняйте только один фактор за раз. Для каждого варианта запишите стоимость на единицу работы, задержку, долю повторов, ошибки и результаты ручной проверки. Публикуйте числа только вместе с периодом, единицами измерения и способом расчёта.
Такой порядок помогает найти экономию, не выдавая локальный эксперимент за результат для любого продукта.