RAG на эмбеддингах

RAG (Retrieval-Augmented Generation) — архитектура, в которой модель отвечает не «из памяти», а по вашим документам: корпус нарезается на фрагменты и векторизуется, при вопросе находятся ближайшие по смыслу фрагменты, и уже по ним генерируется ответ. Так нейросеть получает доступ к внутренней базе знаний без дообучения.

Весь цикл закрывают два эндпоинта Mixen:

• POST /v1/embeddings — текст в вектор (и для индекса, и для вопроса);
• POST /v1/chat/completions — судья-оценщик на этапе реранка и финальный ответ по контексту.

Rerank

Отдельного rerank-эндпоинта у нас нет. Вместо специализированной rerank-модели используется LLM-as-judge: релевантность найденных фрагментов оценивает дешёвая чат-модель (шаг 3). На качестве это почти не сказывается, а стоит копейки.

Как устроен пайплайн

1. Индекс — документы режутся на фрагменты и один раз превращаются в векторы. Единственный «тяжёлый» проход: он амортизируется и дальше не повторяется.
2. Поиск — вопрос тоже векторизуется; по косинусной близости берутся top-10–20 кандидатов.
3. Реранк — судья-модель читает кандидатов и оставляет top-3–5 действительно релевантных.
4. Ответ — финальная модель отвечает по оставшемуся контексту и цитирует фрагменты номерами.

На запросе платите только за эмбеддинг вопроса, вызов судьи и генерацию ответа.

Требования

pip install -U openai numpy httpx   # httpx понадобится в async-секции ниже

Код рассчитан на Python 3.10+.

Шаг 1. Индекс

Пример корпуса — фрагменты документации Mixen; на их место встанут ваши документы. Векторы считаются батчем — один сетевой рейс на всю пачку вместо поштучных запросов.

import hashlib
import json
from pathlib import Path

import numpy as np
from openai import OpenAI

client = OpenAI(
    base_url="https://api.mixen.ai/v1",
    api_key="ваш-api-ключ",
)

EMBED_MODEL = "text-embedding-3-small"  # 1536 измерений, ~2.7 ₽ за 1M входных токенов

chunks = [
    "Mixen — агрегатор нейросетей: текст, изображения, видео и аудио в одном кабинете, оплата в рублях.",
    "Баланс пополняется через СБП, банковскую карту, Telegram Stars или криптовалюту; списания идут за фактически потраченные токены.",
    "Публичный API совместим с OpenAI: тот же SDK, меняется только base URL — https://api.mixen.ai/v1.",
    "Ключи API создаются в кабинете, начинаются с mxn- и ограничены 60 запросами в минуту.",
    "POST /v1/chat/completions — основной эндпоинт: история сообщений, стриминг через SSE, картинки и файлы на входе.",
    "POST /v1/embeddings возвращает по одному вектору на каждый входной текст, порядок сохраняется — удобно батчить.",
]


def embed(texts: list[str]) -> np.ndarray:
    """Эмбеддинги батчем: один вызов API на всю пачку текстов."""
    resp = client.embeddings.create(model=EMBED_MODEL, input=texts)
    return np.array([item.embedding for item in resp.data])


def sha256(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def build_index(chunks: list[str], cache_file: str = "embedding-cache.json") -> np.ndarray:
    """Кеш векторов на диске: ключ — sha256 чанка, значение — вектор.

    Правка одного документа не пересчитывает весь корпус: в API уезжают
    только те фрагменты, которых нет в кеше.
    """
    cache: dict[str, list[float]] = (
        json.loads(Path(cache_file).read_text()) if Path(cache_file).exists() else {}
    )
    missing = [c for c in chunks if sha256(c) not in cache]
    if missing:
        for chunk, vector in zip(missing, embed(missing)):
            cache[sha256(chunk)] = vector.tolist()
        Path(cache_file).write_text(json.dumps(cache, ensure_ascii=False))
    return np.array([cache[sha256(c)] for c in chunks])


matrix = build_index(chunks)  # (n_chunks, 1536)

Эмбеддинг-модели обслуживаются эндпоинтом /v1/embeddings и видны в каталоге GET /v1/models — ищите по полю type: "embeddings": text-embedding-3-small (1536 измерений), text-embedding-3-large (3072, точнее и дороже — ~17.5 ₽ за 1M токенов) и легаси-алиас text-embedding-ada-002, который ведёт на small.

Шаг 2. Поиск

Косинусная близость на NumPy — скалярное произведение нормированных векторов. Внешняя векторная БД для старта не нужна.

def retrieve(question: str, top_k: int = 10) -> list[tuple[str, float]]:
    """Top-k ближайших фрагментов по косинусу; вернём текст и оценку."""
    q = embed([question])[0]
    q /= np.linalg.norm(q)                                       # нормируем вопрос
    m = matrix / np.linalg.norm(matrix, axis=1, keepdims=True)    # и матрицу индекса
    scores = m @ q
    order = np.argsort(scores)[::-1][:top_k]
    return [(chunks[i], float(scores[i])) for i in order]


for text, score in retrieve("Какими способами можно пополнить баланс?", top_k=3):
    print(f"{score:.3f}  {text}")

Забирайте с запасом (10–20 кандидатов): дальше включается судья и сужает выдачу до 3–5 фрагментов.

Шаг 3. Реранк через LLM-as-judge

Эмбеддинги хорошо ловят тему, но путают фрагменты, близкие по смыслу и разные по делу. Судья — обычная дешёвая чат-модель, которая читает тексты и скорит их полезность для конкретного вопроса.

import re

JUDGE_MODEL = "z-ai/glm-5.3-flash"  # судья: 8.4 ₽ / 1M входных, 27.9 ₽ / 1M выходных


def _parse_scores(text: str) -> list[float]:
    """JSON из ответа судьи; запасной путь — вытянуть массив регуляркой."""
    try:
        return json.loads(text)["scores"]
    except (json.JSONDecodeError, KeyError, TypeError):
        match = re.search(r"\[[^\]]*\]", text)  # JSON может приехать в markdown-обёртке
        try:
            return json.loads(match.group()) if match else []
        except json.JSONDecodeError:
            return []


def judge_rerank(question: str, candidates: list[str], top_k: int = 4) -> list[str]:
    """Скорим кандидатов моделью-судьёй и оставляем лучшие top_k."""
    listing = "\n\n".join(f"[{i + 1}] {c}" for i, c in enumerate(candidates))
    resp = client.chat.completions.create(
        model=JUDGE_MODEL,
        max_tokens=1500,
        messages=[
            {"role": "system", "content":
                "Ты — оценщик релевантности. Проранжируй фрагменты по тому, "
                "насколько каждый помогает ответить на вопрос. Ответь строго "
                "JSON с ключом scores — список чисел от 0 до 1, по одному на "
                "каждый фрагмент в исходном порядке."},
            {"role": "user", "content": f"Вопрос: {question}\n\nФрагменты:\n\n{listing}"},
        ],
    )
    scores = _parse_scores(resp.choices[0].message.content or "")
    if len(scores) != len(candidates):
        return candidates[:top_k]  # судья промахнулся с форматом — режем как есть
    ranked = sorted(zip(candidates, scores), key=lambda pair: pair[1], reverse=True)
    return [c for c, _ in ranked[:top_k]]

Reasoning ест бюджет ответа

glm-5.3-flash — рассуждающая модель: перед JSON она думает, и рассуждение тарифицируется и съедает лимит max_tokens. Обрезанный ответ приходит с finish_reason: "length" и парсер уходит в запасной путь. Ставьте max_tokens от 1500. Даже так вызов судьи на десятке фрагментов стоит ~0.01–0.03 ₽ — можно звать его на каждый запрос.

Когда судья окупается:

• в выдаче много фрагментов, близких по теме, но разных по существу;
• точность критична: поддержка, юрдоки, внутренние регламенты — лучше меньше, но верно;
• корпус большой, и top-k приходится держать на уровне 15–20.

Когда можно пропустить:

• корпус маленький (десятки фрагментов) и косинусы уже хорошо разделяют выдачу;
• латентность важнее точности — каждый лишний вызов добавляет задержку.

Шаг 4. Ответ

Финальная модель получает только отобранные фрагменты и жёсткую инструкцию цитировать источники.

ANSWER_MODEL = "anthropic/claude-sonnet-5"  # 213.9 ₽ / 1M входных, 1069.5 ₽ / 1M выходных

SYSTEM_PROMPT = (
    "Отвечай только по приведённым фрагментам. После каждого факта указывай "
    "номер фрагмента в квадратных скобках: [1]. Если ответа во фрагментах нет — "
    "ответь «В документах ответа нет»."
)


def answer(question: str, evidence: list[str]) -> str:
    context = "\n\n".join(f"[{i + 1}] {c}" for i, c in enumerate(evidence))
    resp = client.chat.completions.create(
        model=ANSWER_MODEL,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Фрагменты документов:\n\n{context}\n\nВопрос: {question}"},
        ],
    )
    return resp.choices[0].message.content


question = "Как пополнить баланс?"
candidates = [c for c, _ in retrieve(question, top_k=10)]
print(answer(question, judge_rerank(question, candidates)))

Не хотите Anthropic — openai/gpt-5.6-sol по той же цене (213.9 и 1069.5 ₽ за 1M входных и выходных).

Держите префикс промпта стабильным: системная инструкция и каркас сообщения не меняются от запроса к запросу, поэтому попадают в промпт-кэш. Повторные запросы читают вход по 21.4 ₽ за 1M вместо 213.9 ₽ — в 10 раз дешевле; кэш живёт 5 минут и продлевается каждым попаданием. Практическое правило: постоянную инструкцию — в начало промпта, меняющийся контекст — в конец.

Стриминг ответа

Наш API отдаёт реальные дельты: первый текст прилетает задолго до конца генерации. Для длинных ответов оборачивайте шаг 4 в генератор.

def generate_stream(question: str, evidence: list[str]):
    """Ответ льётся дельтами по мере генерации."""
    context = "\n\n".join(f"[{i + 1}] {c}" for i, c in enumerate(evidence))
    stream = client.chat.completions.create(
        model=ANSWER_MODEL,
        stream=True,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Фрагменты документов:\n\n{context}\n\nВопрос: {question}"},
        ],
    )
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content


for piece in generate_stream(question, judge_rerank(question, candidates)):
    print(piece, end="", flush=True)

Async-пайплайн

В боте или веб-сервере блокировать воркер на три похода в API — плохая идея. Ниже тот же конвейер на AsyncOpenAI: три последовательных ожидания, а параллелить здесь нечего — каждый этап требует результата предыдущего (что параллелится — в разделе про multi-query).

import asyncio

from openai import AsyncOpenAI

aclient = AsyncOpenAI(
    base_url="https://api.mixen.ai/v1",
    api_key="ваш-api-ключ",
)


async def rag_pipeline(question: str, top_k: int = 10, final_k: int = 4) -> str:
    """Индекс → поиск → судья → ответ, целиком асинхронно."""
    q = np.array((await aclient.embeddings.create(
        model=EMBED_MODEL, input=question,
    )).data[0].embedding)
    q /= np.linalg.norm(q)
    scores = (matrix / np.linalg.norm(matrix, axis=1, keepdims=True)) @ q
    found = [chunks[i] for i in np.argsort(scores)[::-1][:top_k]]

    listing = "\n\n".join(f"[{i + 1}] {c}" for i, c in enumerate(found))
    judged = await aclient.chat.completions.create(
        model=JUDGE_MODEL,
        max_tokens=1500,
        messages=[
            {"role": "system", "content":
                "Ты — оценщик релевантности. Проранжируй фрагменты по пользе "
                "для ответа на вопрос. Ответь строго JSON с ключом scores — "
                "список чисел от 0 до 1 на каждый фрагмент."},
            {"role": "user", "content": f"Вопрос: {question}\n\nФрагменты:\n\n{listing}"},
        ],
    )
    jscores = _parse_scores(judged.choices[0].message.content or "")
    if len(jscores) == len(found):
        found = [c for c, s in sorted(zip(found, jscores),
                                      key=lambda pair: pair[1], reverse=True)][:final_k]

    context = "\n\n".join(f"[{i + 1}] {c}" for i, c in enumerate(found))
    final = await aclient.chat.completions.create(
        model=ANSWER_MODEL,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Фрагменты документов:\n\n{context}\n\nВопрос: {question}"},
        ],
    )
    return final.choices[0].message.content


print(asyncio.run(rag_pipeline("Что делать при ошибке 429?")))

Метаданные

Голый текст фрагмента отвечает только на «что найдено». Для интерфейсов важнее «откуда»: заведите dataclass с источником и разделом.

from dataclasses import dataclass


@dataclass
class Chunk:
    text: str
    source: str   # файл, страница или URL происхождения
    section: str   # тематический раздел: «оплата», «лимиты», «модели»...


corpus = [
    Chunk("Баланс пополняется через СБП, карту, Telegram Stars или криптовалюту.",
          source="docs/pricing.md", section="оплата"),
    Chunk("Списания за генерацию идут с баланса по факту потраченных токенов.",
          source="docs/pricing.md", section="оплата"),
    Chunk("Ключи API начинаются с mxn- и ограничены 60 запросами в минуту.",
          source="docs/rate-limits.md", section="лимиты"),
    Chunk("При ошибке 429 повторяйте запрос с нарастающей задержкой.",
          source="docs/rate-limits.md", section="лимиты"),
]

# индекс строится по .text; порядок зафиксирован списком corpus
matrix = build_index([c.text for c in corpus])


def retrieve_filtered(question: str, section: str | None, top_k: int = 10) -> list[Chunk]:
    """Фильтр по разделу — ДО судьи: кандидатов меньше, оценка точнее и дешевле."""
    pool = [i for i, c in enumerate(corpus) if section is None or c.section == section]
    q = embed([question])[0]
    q /= np.linalg.norm(q)
    m = matrix / np.linalg.norm(matrix, axis=1, keepdims=True)
    scores = m @ q
    pool.sort(key=lambda i: scores[i], reverse=True)
    return [corpus[i] for i in pool[:top_k]]

В ответе теперь можно ссылаться на источник — [1] docs/pricing.md вместо безымянного [1]: пользователь видит, откуда модель взяла факт, а вы быстрее находите устаревший документ.

Multi-query RAG

Пользователь формулирует вопрос своими словами, и одна формулировка может не пересечься со словарём документа. Решение: LLM переформулирует вопрос несколькими способами, поиск идёт по каждому варианту, кандидаты объединяются с дедупликацией.

async def rephrase(question: str) -> list[str]:
    """Три варианта вопроса: синонимами, общее и конкретнее."""
    resp = await aclient.chat.completions.create(
        model=JUDGE_MODEL,
        max_tokens=800,
        messages=[{"role": "user", "content":
            "Переформулируй вопрос тремя разными способами: синонимами, "
            "более общо и более конкретно. Ответ — ровно три строки, без "
            "нумерации.\n\n"
            f"Вопрос: {question}"}],
    )
    lines = (resp.choices[0].message.content or "").splitlines()
    cleaned = [l.strip().lstrip("0123456789.-) ") for l in lines if l.strip()]
    return [question, *cleaned[:3]]


async def multi_query_candidates(question: str, per_query: int = 5) -> list[str]:
    """Поиск по каждой формулировке; объединение кандидатов без дублей."""
    variants = await rephrase(question)
    resp = await aclient.embeddings.create(model=EMBED_MODEL, input=variants)
    m = matrix / np.linalg.norm(matrix, axis=1, keepdims=True)
    merged: dict[str, None] = {}  # dict вместо set — сохраняем порядок появления
    for item in resp.data:  # порядок ответа совпадает с порядком input
        q = np.array(item.embedding)
        q /= np.linalg.norm(q)
        for i in np.argsort(m @ q)[::-1][:per_query]:
            merged.setdefault(chunks[i], None)
    return list(merged)


question = "Как заплатить за нейросети?"
found = asyncio.run(multi_query_candidates(question))
evidence = judge_rerank(question, found)

Переформулировка идёт той же дешёвой моделью и добавляет к запросу примерно ещё 0.01 ₽. Побочный эффект — устойчивость к опечаткам и жаргону: хотя бы одна из формулировок обычно попадает в документ.

Полный пример

Скрипт шагов 1–4 целиком — скопируйте и подставьте свой корпус.

"""RAG по документам через Mixen: индекс → поиск → судья → ответ."""
import hashlib
import json
import re
from pathlib import Path

import numpy as np
from openai import OpenAI

client = OpenAI(base_url="https://api.mixen.ai/v1", api_key="ваш-api-ключ")

EMBED_MODEL = "text-embedding-3-small"
JUDGE_MODEL = "z-ai/glm-5.3-flash"
ANSWER_MODEL = "anthropic/claude-sonnet-5"

SYSTEM_PROMPT = (
    "Отвечай только по приведённым фрагментам. После каждого факта указывай "
    "номер фрагмента в квадратных скобках: [1]. Если ответа во фрагментах "
    "нет — ответь «В документах ответа нет»."
)

chunks = [
    "Mixen — агрегатор нейросетей: текст, изображения, видео и аудио в одном кабинете, оплата в рублях.",
    "Баланс пополняется через СБП, банковскую карту, Telegram Stars или криптовалюту; списания идут за фактически потраченные токены.",
    "Публичный API совместим с OpenAI: тот же SDK, меняется только base URL — https://api.mixen.ai/v1.",
    "Ключи API создаются в кабинете, начинаются с mxn- и ограничены 60 запросами в минуту.",
    "POST /v1/chat/completions — основной эндпоинт: история сообщений, стриминг через SSE, картинки и файлы на входе.",
    "POST /v1/embeddings возвращает по одному вектору на каждый входной текст, порядок сохраняется.",
]


def sha256(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def embed(texts: list[str]) -> np.ndarray:
    resp = client.embeddings.create(model=EMBED_MODEL, input=texts)
    return np.array([item.embedding for item in resp.data])


def build_index(cache_file: str = "embedding-cache.json") -> np.ndarray:
    cache: dict[str, list[float]] = (
        json.loads(Path(cache_file).read_text()) if Path(cache_file).exists() else {}
    )
    missing = [c for c in chunks if sha256(c) not in cache]
    if missing:
        for chunk, vector in zip(missing, embed(missing)):
            cache[sha256(chunk)] = vector.tolist()
        Path(cache_file).write_text(json.dumps(cache, ensure_ascii=False))
    return np.array([cache[sha256(c)] for c in chunks])


def retrieve(question: str, top_k: int = 10) -> list[str]:
    q = embed([question])[0]
    q /= np.linalg.norm(q)
    m = matrix / np.linalg.norm(matrix, axis=1, keepdims=True)
    return [chunks[i] for i in np.argsort(m @ q)[::-1][:top_k]]


def _parse_scores(text: str) -> list[float]:
    try:
        return json.loads(text)["scores"]
    except (json.JSONDecodeError, KeyError, TypeError):
        match = re.search(r"\[[^\]]*\]", text)
        try:
            return json.loads(match.group()) if match else []
        except json.JSONDecodeError:
            return []


def judge_rerank(question: str, candidates: list[str], top_k: int = 4) -> list[str]:
    listing = "\n\n".join(f"[{i + 1}] {c}" for i, c in enumerate(candidates))
    resp = client.chat.completions.create(
        model=JUDGE_MODEL,
        max_tokens=1500,
        messages=[
            {"role": "system", "content":
                "Ты — оценщик релевантности. Проранжируй фрагменты по пользе "
                "для ответа на вопрос. Ответь строго JSON с ключом scores — "
                "список чисел от 0 до 1 на каждый фрагмент."},
            {"role": "user", "content": f"Вопрос: {question}\n\nФрагменты:\n\n{listing}"},
        ],
    )
    scores = _parse_scores(resp.choices[0].message.content or "")
    if len(scores) != len(candidates):
        return candidates[:top_k]
    ranked = sorted(zip(candidates, scores), key=lambda pair: pair[1], reverse=True)
    return [c for c, _ in ranked[:top_k]]


def answer(question: str, evidence: list[str]) -> str:
    context = "\n\n".join(f"[{i + 1}] {c}" for i, c in enumerate(evidence))
    resp = client.chat.completions.create(
        model=ANSWER_MODEL,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Фрагменты документов:\n\n{context}\n\nВопрос: {question}"},
        ],
    )
    return resp.choices[0].message.content


matrix = build_index()  # первый запуск считает векторы, дальше — из кеша

if __name__ == "__main__":
    question = "Какими способами можно пополнить баланс?"
    evidence = judge_rerank(question, retrieve(question, top_k=10))
    print(answer(question, evidence))

Экономика типового запроса (10 кандидатов судье, 4 фрагмента в контексте ответа):

Этап	Тариф	Величина
Эмбеддинг вопроса	small, ~2.7 ₽ / 1M входных	доли копейки
Судья	glm-5.3-flash, 8.4 / 27.9 ₽ за 1M входных / выходных	0.01–0.03 ₽
Ответ	claude-sonnet-5, 213.9 ₽ / 1M входных (кэш-чтение 21.4 ₽) и 1069.5 ₽ / 1M выходных	основная статья

Обычно запрос обходится в 0.05–0.2 ₽: дороже всего выход финальной модели (он в 5 раз дороже входа), поэтому короткие фрагменты и лаконичные ответы экономят напрямую. Индексация амортизируется — векторы считаются один раз и лежат в кеше по хэшу, повторные запуски бесплатны. Весь конвейер стоит меньше 1 ₽ за запрос даже без кэшей и судьи-сокращений.

Нарезка документов

Качество RAG решается на этапе чанкинга сильнее, чем выбором модели ответа:

• Размер — 200–500 токенов на фрагмент. Мельче — теряется контекст фразы; крупнее — в выдачу лезет шум и дорожает контекст.
• Перекрытие — 10–15% соседних фрагментов, чтобы мысль на границе не потерялась. Для русского берите 20%: богатая морфология раскидывает связанные предложения сильнее.
• Границы — режьте по заголовкам и абзацам, а не по символам посреди предложения.

Готовый сплиттер — из langchain-text-splitters (отдельный пакет, ядро Langchain не нужен):

pip install -U langchain-text-splitters

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1200,     # в символах: примерно 300–400 токенов русского текста
    chunk_overlap=240,   # 20% — запас на морфологию
    separators=["\n\n", "\n", ". ", " ", ""],  # от абзацев к словам
)
pieces = splitter.split_text(Path("docs/pricing.md").read_text(encoding="utf-8"))

Продакшен-стек

Матрицы NumPy хватает до десятков тысяч фрагментов. Когда корпус перерастает память или нужны фильтры и репликация — переезжайте на векторное хранилище:

Инструмент	Что это	Когда выбирать
pgvector	расширение Postgres	данные уже в Postgres — векторы лягут в ту же схему, без новой БД
Qdrant	self-hosted векторная БД	большие корпуса, фильтры по метаданным, гибридный поиск из коробки
Chroma	встраиваемая БД	прототипы и небольшие корпуса с персистентностью
FAISS	библиотека индексов	максимум скорости поиска, всё в памяти
Weaviate / Pinecone	управляемые сервисы	не хотите эксплуатировать инфраструктуру сами

Эмбеддинги при переезде не пересчитываются: тот же text-embedding-3-small и те же векторы грузятся в хранилище как есть.

Гибрид с BM25 закрывает слепую зону эмбеддингов — точные названия, артикулы, коды ошибок:

# pip install -U rank_bm25
from rank_bm25 import BM25Okapi

bm25 = BM25Okapi([c.lower().split() for c in chunks])


def hybrid_retrieve(question: str, top_k: int = 10) -> list[str]:
    """Эмбеддинги ловят смысл, BM25 — точные словоформы; кандидаты идут судье."""
    q = embed([question])[0]
    q /= np.linalg.norm(q)
    m = matrix / np.linalg.norm(matrix, axis=1, keepdims=True)
    by_vector = {chunks[i] for i in np.argsort(m @ q)[::-1][:top_k]}
    by_keyword = {chunks[i] for i in np.argsort(
        bm25.get_scores(question.lower().split()))[::-1][:top_k]}
    return list(by_vector | by_keyword)

И кеш эмбеддингов по хэшу чанка (шаг 1) переезжает без изменений — это просто прослойка перед любым хранилищем.

Практика

• Одна модель для индекса и запроса. Векторные пространства small и large несовместимы: смешаете — получите мусорные косинусы.
• Индексируйте батчами. Один вызов /v1/embeddings на пачку фрагментов — быстрее и дешевле поштучных.
• top-k — стартуйте с 10 на выходе поиска и сужайте до 3–5 после судьи.
• Отсечка по косинусу ~0.3. Ниже — фрагменты почти наверняка мимо; честное «не нашёл» лучше контекста из шума.
• Судья не на каждый запрос. Если top-10 и так хорошо разделены (заметный разрыв косинусов) — идите сразу в ответ и экономьте задержку.
• Русский корпус с претензиями на качество — text-embedding-3-large. Честная оговорка: специализированных мультиязычных моделей (e5, multilingual) у нас нет — large лучший мультиязычный вариант из наших.
• Стабильный префикс промпта — системная инструкция без изменений от запроса к запросу бьёт в промпт-кэш и режет цену входа модели ответа примерно в 10 раз.
• Кеш эмбеддингов по хэшу — переиндексация трогает только изменившиеся фрагменты, а не весь корпус.

Решение проблем

Батч эмбеддингов падает с ошибкой 400

Суммарный вход батча превысил лимит запроса. Режьте пачки по ~100 фрагментов, для длинных чанков — меньше:

def embed_safe(texts: list[str], batch_size: int = 100) -> np.ndarray:
    parts = [embed(texts[i:i + batch_size])
             for i in range(0, len(texts), batch_size)]
    return np.vstack(parts)

Судья возвращает не JSON

Три типовые причины. Первая — модель обернула ответ в markdown-блок с меткой json: регулярка в _parse_scores вытягивает массив и игнорирует обёртку. Вторая — reasoning съел max_tokens, ответ оборвался на середине (finish_reason: "length"): поднимите потолок до 1500 и выше. Третья — пояснительный текст вокруг массива; регулярка справляется и с ним. Если судья стабильно ломается, проверьте, что просите «строго JSON» в системном промпте.

Косинусы одинаковые или мусорные

Проверьте нормализацию: и вопрос, и строки матрицы делятся на свои нормы до скалярного произведения. Дальше — дубли фрагментов в индексе (они раздувают top-k) и одна ли модель эмбеддингов использована для индекса и для запроса. Смешение моделей даёт правдоподобные на вид, но бессмысленные оценки.

Модель галлюцинирует поверх контекста

Ужесточите системный промпт: «нет данных во фрагментах — так и скажи», без общих знаний. Снизьте число фрагментов до 3 после судьи, включите отсечку по косинусу и требуйте ссылку на номер после каждого факта — утверждение без опоры видно сразу.

Ответ идёт слишком долго

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

Слабое качество на русском

Перейдите на text-embedding-3-large, измельчите чанки до 200–300 токенов и поднимите перекрытие до 20%. Добавьте гибрид с BM25 — точные словоформы ловятся лексическим поиском лучше, чем эмбеддингами.

Следующие шаги

• Интеграции — Dify и n8n: готовый RAG-конвейер без кода.
• Чат и стриминг — все параметры POST /v1/chat/completions, включая режимы рассуждения и веб-поиск.
• Каталог моделей — модели и цены для этапа ответа.