Ответы по своим документам (RAG)
У вас будет программа, которая отвечает по вашим документам и ставит ссылки на источник [1] [2].
Нужно: ключ fetch_... (создать), баланс (пополнить), Python 3.10+, документы в .txt или .md (PDF и Word сначала переведите в текст, например через pypdf).
Поиск работает у вас на компьютере и ничего не стоит: эндпоинта эмбеддингов в Zukko нет. Zukko пишет сам ответ.
- 1Сложите документы
Создайте папку
docsи положите туда.txtили.md. Подпапки можно. - 2Установите библиотекиbash
pip install openai rank_bm25 snowballstemmer - 3Сохраните код
Скопируйте код из раздела «Готовый код» ниже в
rag.pyрядом с папкойdocs. - 4Запуститеbash
export ZUKKO_API_KEY="fetch_..." python rag.pyЗадайте вопрос. В ответе будут номера источников и файлы, откуда взяты фрагменты. Добавили документ — перезапустите программу.
1Готовый код
import os
import re
import sys
from pathlib import Path
import snowballstemmer
from openai import APIConnectionError, APIStatusError, OpenAI
from rank_bm25 import BM25Plus
DOCS_DIR = Path(os.environ.get("DOCS_DIR", "docs")) # папка с файлами .txt и .md
MODEL = os.environ.get("ZUKKO_MODEL", "claude-haiku-5-5")
CHUNK_SIZE = 1200 # примерный размер фрагмента в символах
TOP_K = 5 # сколько фрагментов отправлять модели
В бота: создайте index один раз при запуске и вызывайте ask в обработчике сообщений, как в статье Telegram-бот. ask синхронная — в асинхронном боте: await asyncio.to_thread(ask, index, text).
2Другие способы
Всё в запрос — до ~200–300 страниц
Без поиска: склейте документы и отправьте с вопросом. Каждый вопрос оплачивает всю базу и отвечается дольше.
from pathlib import Path
import os
from openai import OpenAI
client = OpenAI(base_url="https://api.zukko.pro/v1", api_key=os.environ["ZUKKO_API_KEY"], timeout=300)
docs = "\n\n".join(
f"=== {p.name} ===\n{p.read_text(encoding='utf-8')}" for p in sorted(Path("docs").glob("*.md"))
)
question = "Сколько дней отпуска положено новому сотруднику?"
response = client.chat.completions.create(
model="claude-sonnet-5",
messages=[
# документы — в начало: эта часть одинаковая и попадает в кэш промпта
{"role": "system", "content": "Отвечай только по документам ниже и называй файл-источник. "
"Если ответа нет — скажи об этом.\n\n" + docs},
{"role": "user", "content": question},
],
)
print(response.choices[0].message.content)Поиск по смыслу — когда в вопросах другие слова, чем в документах
Например, «уволиться» и «расторжение трудового договора». Замените класс Index в rag.py на этот код (работает локально, без видеокарты):
# pip install sentence-transformers
from sentence_transformers import SentenceTransformer
encoder = SentenceTransformer("intfloat/multilingual-e5-small") # скачается один раз, понимает русский
class Index:
def __init__(self, chunks: list[dict]):
self.chunks = chunks
# у моделей e5 тексты помечают «passage: », а вопросы — «query: »
self.vectors = encoder.encode(
["passage: " + c["text"] for c in chunks], normalize_embeddings=True
)
def search(self, query: str, k: int = TOP_K) -> list[dict]:
q = encoder.encode(["query: " + query], normalize_embeddings=True)[0]
scores = self.vectors @ q # чем больше число, тем ближе смысл
best = scores.argsort()[::-1][:k]
return [self.chunks[i] for i in best]Векторы для тысяч фрагментов считаются несколько минут — сохраните их в файл (numpy.save).
3Если не работает
| Что видите | Что сделать |
|---|---|
| «В папке docs нет файлов» | Переведите файлы в .txt/.md, проверьте путь или задайте DOCS_DIR |
| «В документах ничего не нашлось», хотя ответ есть | Перефразируйте вопрос. Часто — перейдите на поиск по смыслу |
| Ответ неполный | Увеличьте TOP_K до 8–10 или уменьшите CHUNK_SIZE |
| Модель отвечает не по документам | Не убирайте правило про источники, попробуйте claude-sonnet-5 |
| «Всё в запрос» отвечает очень долго | Перейдите на поиск по словам, поставьте таймаут клиента 300 с |
| 401, 402, 429 | См. Ошибки API |