RAG: как подключить свои данные к языковой модели и получить умного ассистента

Пользователь: «Как настроить деплой на VPS?»
         ↓
Эмбеддинг запроса → вектор вопроса
         ↓
Векторный поиск → топ-5 чанков из документации
         ↓
Промпт: [чанк 1] [чанк 2] [чанк 3] + «Как настроить деплой на VPS?»
         ↓
LLM → ответ, основанный на вашей документации

pip install chromadb openai tiktoken langchain langchain-openai langchain-community

import os
import glob
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

DOCS_DIR = "./docs"       # папка с вашими .md файлами
CHROMA_DIR = "./chroma_db"  # куда сохранять базу

def load_markdown_files(directory: str) -> list[dict]:
    """Загружает все .md файлы, возвращает список {text, source}."""
    documents = []
    for filepath in glob.glob(f"{directory}/**/*.md", recursive=True):
        with open(filepath, "r", encoding="utf-8") as f:
            text = f.read()
        documents.append({"text": text, "source": filepath})
    return documents

def build_index(docs_dir: str, chroma_dir: str):
    # Загружаем документы
    raw_docs = load_markdown_files(docs_dir)
    print(f"Загружено документов: {len(raw_docs)}")

    # Нарезаем на чанки
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=800,       # символов в чанке
        chunk_overlap=100,    # перекрытие — чтобы не потерять контекст на границах
        separators=["\n## ", "\n### ", "\n\n", "\n", " "],
    )

    chunks = []
    metadatas = []
    for doc in raw_docs:
        parts = splitter.split_text(doc["text"])
        chunks.extend(parts)
        metadatas.extend([{"source": doc["source"]}] * len(parts))

    print(f"Чанков после нарезки: {len(chunks)}")

    # Создаём векторную базу
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma.from_texts(
        texts=chunks,
        embedding=embeddings,
        metadatas=metadatas,
        persist_directory=chroma_dir,
    )
    print(f"Индекс сохранён в {chroma_dir}")
    return vectorstore

if __name__ == "__main__":
    build_index(DOCS_DIR, CHROMA_DIR)

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.schema import HumanMessage, SystemMessage

CHROMA_DIR = "./chroma_db"

def load_vectorstore(chroma_dir: str):
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    return Chroma(
        persist_directory=chroma_dir,
        embedding_function=embeddings,
    )

def ask(question: str, vectorstore, k: int = 5) -> str:
    # Ищем релевантные чанки
    results = vectorstore.similarity_search(question, k=k)
    context = "\n\n---\n\n".join([doc.page_content for doc in results])
    sources = list({doc.metadata.get("source", "") for doc in results})

    # Строим промпт
    system_prompt = """Ты — ассистент по документации. 
Отвечай только на основе предоставленного контекста.
Если ответа нет в контексте — честно скажи об этом.
Не придумывай информацию."""

    user_prompt = f"""Контекст из документации:
{context}

Вопрос: {question}"""

    # Генерируем ответ
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    response = llm.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=user_prompt),
    ])

    return response.content, sources

if __name__ == "__main__":
    vs = load_vectorstore(CHROMA_DIR)
    answer, sources = ask("Как настроить автодеплой на VPS?", vs)
    print(answer)
    print("\nИсточники:", sources)

# Устанавливаем Ollama
curl -fsSL https://ollama.com/install.sh | sh

# Скачиваем модели
ollama pull nomic-embed-text   # эмбеддинги
ollama pull llama3.2           # или qwen2.5, mistral

from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.llms import Ollama
from langchain_community.vectorstores import Chroma

def build_local_index(docs_dir: str, chroma_dir: str):
    raw_docs = load_markdown_files(docs_dir)

    from langchain.text_splitter import RecursiveCharacterTextSplitter
    splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=100)
    chunks, metadatas = [], []
    for doc in raw_docs:
        parts = splitter.split_text(doc["text"])
        chunks.extend(parts)
        metadatas.extend([{"source": doc["source"]}] * len(parts))

    # Локальные эмбеддинги через Ollama
    embeddings = OllamaEmbeddings(model="nomic-embed-text")
    vectorstore = Chroma.from_texts(
        texts=chunks,
        embedding=embeddings,
        metadatas=metadatas,
        persist_directory=chroma_dir,
    )
    return vectorstore

def ask_local(question: str, vectorstore) -> str:
    results = vectorstore.similarity_search(question, k=5)
    context = "\n\n".join([doc.page_content for doc in results])

    llm = Ollama(model="llama3.2")
    prompt = f"""Контекст: {context}

Вопрос: {question}

Отвечай только на основе контекста. Если ответа нет — скажи об этом."""

    return llm.invoke(prompt)

-- Включаем расширение
CREATE EXTENSION IF NOT EXISTS vector;

-- Создаём таблицу для чанков
CREATE TABLE documents (
    id          SERIAL PRIMARY KEY,
    content     TEXT NOT NULL,
    source      TEXT,
    embedding   vector(1536),  -- размер под text-embedding-3-small
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

-- Индекс для быстрого поиска (HNSW — лучшее соотношение скорость/точность)
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

import psycopg2
import numpy as np
from openai import OpenAI

client = OpenAI()

def get_embedding(text: str) -> list[float]:
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

def insert_chunk(conn, content: str, source: str):
    embedding = get_embedding(content)
    with conn.cursor() as cur:
        cur.execute(
            "INSERT INTO documents (content, source, embedding) VALUES (%s, %s, %s)",
            (content, source, embedding)
        )
    conn.commit()

def search_similar(conn, query: str, k: int = 5) -> list[dict]:
    query_embedding = get_embedding(query)
    with conn.cursor() as cur:
        cur.execute("""
            SELECT content, source,
                   1 - (embedding <=> %s::vector) AS similarity
            FROM documents
            ORDER BY embedding <=> %s::vector
            LIMIT %s
        """, (query_embedding, query_embedding, k))
        rows = cur.fetchall()
    return [
        {"content": r[0], "source": r[1], "similarity": r[2]}
        for r in rows
    ]

# Пример нарезки с сохранением заголовка раздела
import re

def chunk_markdown_with_headers(text: str, source: str) -> list[dict]:
    """Нарезает Markdown, сохраняя заголовок раздела в метаданных."""
    chunks = []
    current_header = ""
    current_text = []

    for line in text.split("\n"):
        if line.startswith("## "):
            # Сохраняем накопленный текст
            if current_text:
                chunks.append({
                    "text": "\n".join(current_text).strip(),
                    "source": source,
                    "section": current_header,
                })
                current_text = []
            current_header = line.lstrip("# ").strip()
        current_text.append(line)

    # Последний раздел
    if current_text:
        chunks.append({
            "text": "\n".join(current_text).strip(),
            "source": source,
            "section": current_header,
        })

    return [c for c in chunks if len(c["text"]) > 50]  # убираем пустые

-- Добавляем полнотекстовый поиск в таблицу
ALTER TABLE documents ADD COLUMN ts tsvector
    GENERATED ALWAYS AS (to_tsvector('russian', content)) STORED;
CREATE INDEX ON documents USING gin(ts);

-- Гибридный запрос: взвешенная сумма BM25 и косинусного сходства
SELECT content, source,
       (0.5 * ts_rank(ts, query) + 0.5 * (1 - (embedding <=> $1::vector))) AS score
FROM documents, to_tsquery('russian', $2) query
WHERE ts @@ query OR (embedding <=> $1::vector) < 0.4
ORDER BY score DESC
LIMIT 5;

import hashlib
import json
import os

HASH_FILE = ".index_hashes.json"

def file_hash(filepath: str) -> str:
    with open(filepath, "rb") as f:
        return hashlib.md5(f.read()).hexdigest()

def get_changed_files(docs_dir: str) -> list[str]:
    """Возвращает только изменившиеся файлы."""
    hashes = {}
    if os.path.exists(HASH_FILE):
        with open(HASH_FILE) as f:
            hashes = json.load(f)

    changed = []
    current = {}
    for filepath in glob.glob(f"{docs_dir}/**/*.md", recursive=True):
        h = file_hash(filepath)
        current[filepath] = h
        if hashes.get(filepath) != h:
            changed.append(filepath)

    with open(HASH_FILE, "w") as f:
        json.dump(current, f)

    return changed

pip install ragas

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_recall

# dataset — список {question, answer, contexts, ground_truth}
result = evaluate(dataset, metrics=[faithfulness, answer_relevancy, context_recall])
print(result)

Индексация:
☐ Выбрана модель эмбеддингов (одна и та же для индексации и поиска)
☐ Размер чанков 300–1000 токенов, перекрытие 10–15%
☐ Нарезка по смысловым границам (заголовки, абзацы)
☐ Метаданные: source, section, дата
☐ Реализована инкрементальная переиндексация

Поиск:
☐ k=5–10 чанков при поиске
☐ Метрика схожести соответствует модели (косинусная для большинства)
☐ Рассмотрен гибридный поиск если есть точные термины/имена

Генерация:
☐ Системный промпт явно требует отвечать только по контексту
☐ Модель честно говорит «не знаю», если ответа нет в контексте
☐ Источники показываются пользователю

Инфраструктура:
☐ Векторная база с persistence (не in-memory)
☐ Логируются все запросы и найденные чанки
☐ Есть мониторинг качества на тестовых вопросах

Безопасность:
☐ API-ключи в переменных среды
☐ Если данные приватные — используется локальная LLM (Ollama)
☐ Нет утечки системного промпта в ответе