RAG: How to Connect Your Data to a Language Model and Get a Smart Assistant

User: “How to set up a deploy on VPS?”
↓
Embedding query → question vector
↓
Vector search → top 5 chunks from documentation
↓
Prompt: [chunk 1] [chunk 2] [chunk 3] + "How to set up a depot for VPS?"
↓
LLM Answer Based on Your Documentation

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)

# Installing Ollama
curl -fsSL https://ollama.com/install.sh | sh

# Downloading models.
ollama pull nomic-embed-text #embedding
ollama pull llama3.2 # or 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)

- Enable expansion.
Create Extension IF NOT EXISTS vector

- Creating a table for chunks
CREATE TABLE documents
id SERIAL PRIMARY KEY,
content TEXT NOT NULL,
source TEXT,
embedding vector(1536) - text-embedding-3-small size
created at TIMESTAMPTZ DEFAULT NOW()
);

Index for fast search (HNSW – the best speed/accuracy ratio)
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]  # убираем пустые

Add full text search to the table
Alter table documents ADD COLUMN ts tsvector
GENERATED ALWAYS AS (to tsvector('russian', content) STORED;
CREATE INDEX on documents USING gin(ts)

Hybrid query: weighted sum of BM25 and cosine similarity
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)

Indexation:
Selected embedding model (the same for indexing and search)
Size of chunks 300-1000 tokens, overlap 10-15%
Slicing by semantic boundaries (headings, paragraphs)
● Metadata: source, section, date
Incremental re-indexation implemented

Search:
k = 5-10 chanks when searching
The similarity metric corresponds to the model (cosine for most)
Hybrid search is considered if there are exact terms/names

Generation:
● The system prompt clearly requires a response only by context.
● The model honestly says "don't know" if there's no answer in context
● Sources are shown to the user

Infrastructure:
Vector base with persistence (not in-memory)
● Log all requests and found chunky
● Quality monitoring on test questions

Safety:
API keys in environment variables
● If the data is private, a local LLM (Ollama) is used.
● No leakage of system prompt in response