RAG у простому Python: чанкування, TF-IDF, косинусна схожість та формулювання запитів
Створіть генерацію з покращенням через пошук, використовуючи лише математичний модуль: розбиття на частини за принципом ковзної вікна, вектори TF-IDF, ранжування за косинусом, чітко сформульований запит та його обмеження.
Джерела: Building RAG From First Principles with just plain python .
Типовий посібник з RAG пропонує вам встановити п’ять бібліотек та викликати .from_documents(). Через десять хвилин демо-версія працює, але ви нічого не розумієте. Коли процес пошуку повертає непотрібні дані або модель створює хибні висновки, ви застрягаєте, адже міркування виконала бібліотека замість вас.
Тут конвеєр побудований навпаки — простою мовою Python, з використанням лише модуля math та одного виклику LLM у текстовому форматі наприкінці. Жодних моделей ембеддингу, жодного FAISS, жодного Pinecone. Розуміння механізмів дозволяє вам, коли ви згодом будете використовувати LangChain чи справжню векторну базу даних, розуміти, що відбувається під капотом. У кінцевому підсумку у вас є працездатний конвеєр, а ще важливіше — уявлення про кожну цифру, яку він генерує.
Проблема, яку вирішує RAG
Знання моделі мови застигають після завершення навчання та зберігаються виключно у її вагах. Щоб вона могла відповідати на запитання щодо внутрішніх корпоративних документів, кодової бази чи сьогоднішніх новин, обидві очевидні опції виявляються незадовільними:
- Поновити навчання моделі. Це повільно та дорого, а зміни, внесені завтра, потребують ще одного циклу навчання.
- Вставити все у запит. Цей підхід не є масштабованим. Навіть вікно обсягом у мільйон токенів не може вмістити 50 000 документів, а дослідження моделей з довгим контекстом неодноразово фіксують ефект „втрати інформації посередині“, коли дані, заховані глибоко в великому запиті, використовуються гірше, навіть якщо вони туди поміщаються. Крім того, додаткові токени призводять до більшої затримки та вищих витрат за кожен запит.
Справжній секрет RAG є простішим, ніж здається. Документи залишаються у зовнішньому сховищі, і під час обробки запиту завантажуються та передаються моделі лише кілька фрагментів, які є релевантними для цього запиту. Штучний інтелект ніколи не зберігає ваші дані — він щоразу отримує новий, релевантний фрагмент.
Отже, будь-яка система RAG складається з двох частин, об’єднаних разом: функції пошуку, яка знаходить потрібний текст у значно більшому наборі даних для конкретного запиту, та функції генерації, яка передає цей текст разом із запитом у штучний інтелект та отримує відповідь. У наступних розділах буде розглянуто, як створюються обидві ці частини з базових елементів.
Крок 1: розділення на фрагменти
Перетворення цілого 50-сторінкового PDF у один вектор знищує майже всі корисні сигнали, оскільки деталі, які могли б відповісти на запит, зсереднюються та стають нічим. Першою задачею є розділення документів на невеликі, конкретні фрагменти.
Найпростіший метод — це склянне вікно фіксованого розміру, яке підраховується за кількістю слів, з перекриттям між сусідніми частинами:
def chunk_text(text: str, chunk_size: int = 200, overlap: int = 50) -> list[str]:
if overlap >= chunk_size:
raise ValueError("overlap must be smaller than chunk_size")
words = text.split()
chunks = []
start = 0
while start < len(words):
end = start + chunk_size
chunks.append(" ".join(words[start:end]))
if end >= len(words):
break
start = end - overlap # step back so chunks share context
return chunks
Ця клозула-захист — це не декоративний елемент. Якщо значення overlap досягає або перевищує chunk_size, позиція початку припиняє змінюватися, і цикл ніколи не закінчується. Наявність великого значення overlap „для більшого контексту“ сприяє легким помилкам, тому функція відмовляється виконуватися замість того, щоб просто застигнути.
Значення overlap має більше значення, ніж здається. Без нього речення, яке відповідає на запитання, може перетинати межу блоку та розділятися, через що жоден з блоків не міститиме повної ідеї, яку можна було б отримати. За наявності overlap ця прикордонна область залишається цілою принаймні в одному блоку.
Один важливий момент, про який варто знати: фіксовані вікна слів ігнорують структуру речень та абзаців, тому вони можуть розділити речення чи рядок у таблиці навпіл. Розділення з урахуванням речень чи абзаців уникає цього, але створює нерівномірні розміри частин, оскільки речення з 10 та 150 словами стають однією частиною. Це спотворює бюджет токенів та впливає на те, наскільки сильно будь-яка окрема частина впливає на подальші обчислення. Жоден підхід до розділення тексту не є без компромісів, і вибір полягає у тому, яку проблему ви готові виправляти.
Крок 2: перетворення тексту на числа за допомогою TF-IDF
Комп’ютер не може безпосередньо порівнювати схожість двох рядків. Кожен наступний крок потребує векторів — списків чисел, з якими можна виконувати арифметичні операції. Класичним способом створення векторів без залежностей є TF-IDF — частота терміну помножена на зворотну частоту документа.
- TF підраховує, як часто слово з’являється в цьому окремому фрагменті. Високе значення для слова „nирка“ свідчить про те, що фрагмент стосується нирок.
- IDF вимірює, наскільки рідкісним є слово в усій колекції. Слова на кшталт „the“ та „is“ зустрічаються скрізь і нічого не говорять про те, який фрагмент ви читаєте, тому їхній показник близький до нуля. Рідкісне слово в фрагменті є набагато сильнішим сигналом.
Код створює словник з усіх фрагментів, обчислює IDF для кожного терміна, а потім перетворює кожен фрагмент на вектор TF-IDF:
import math
from collections import Counter
def build_vocabulary(chunks: list[str]) -> list[str]:
vocab = set()
for chunk in chunks:
vocab.update(chunk.lower().split())
return sorted(vocab)
def compute_idf(chunks: list[str], vocab: list[str]) -> dict[str, float]:
num_chunks = len(chunks)
idf = {}
for word in vocab:
containing = sum(1 for c in chunks if word in c.lower().split())
idf[word] = math.log(num_chunks / (1 + containing))
return idf
def compute_tf(chunk: str, vocab: list[str]) -> dict[str, float]:
words = chunk.lower().split()
counts = Counter(words)
total = len(words)
return {word: counts.get(word, 0) / total for word in vocab}
def vectorize(chunk: str, vocab: list[str], idf: dict[str, float]) -> list[float]:
tf = compute_tf(chunk, vocab)
return [tf[word] * idf[word] for word in vocab]
Спробуйте застосувати цей підхід до трьох фрагментів, і щось стане зрозумілим. Слово, яке присутнє майже в кожному фрагменті, наприклад „the“, має IDF, близький до нуля, тож його повторення нічого не додає до вектора. Рідкісне слово, специфічне для певної теми, отримує значно вищий IDF. TF-IDF — це буквально формула для визначення того, чи повідомляє слово щось унікальне про фрагмент.
Недоліком є масштаб. У разі 100 000 документів словник може містити понад 500 000 унікальних слів, тож кожен вектор фрагмента, навіть трьохсловного, складається з 500 000 чисел, переважно нулів. Такі розріджені вектори складно зберігати та порівнювати. Саме тому в промислових системах переходять на щільні нейронні ембеддинги з розмірністю 384–1536, у яких майже немає нулів.
Крок 3: вимірювання релевантності за допомогою косинусової схожості
Якщо розглядати фрагменти як вектори, наступне питання — наскільки фрагмент є релевантним для запиту. Чиста відстань не є правильним інструментом, оскільки для тексту кут між векторами має більше значення, ніж їхня довжина. Фрагмент, у якому слово „кіт“ згадується один раз, та фрагмент, у якому воно повторюється двадцять разів, все одно стосуються приблизно тієї самої теми, і косинусна схожість ставиться до них однаково, не зважаючи на те, який з них „гучніший“.
Для реалізації потрібен скалярний добуток, модуль та співвідношення між ними:
def dot_product(a, b):
return sum(x * y for x, y in zip(a, b))
def magnitude(v):
return math.sqrt(sum(x * x for x in v))
def cosine_similarity(a, b):
denom = magnitude(a) * magnitude(b)
return 0.0 if denom == 0 else dot_product(a, b) / denom
Косинусна схожість у принципі коливається від -1 до 1, і за цією налаштовкою TF-IDF вона майже завжди залишається у діапазоні від 0 до 1:
- 1.0 означає, що вектори спрямовані точно в одному напрямку – максимальна схожість.
- 0.0 означає, що вони ортогональні та не мають спільних значущих вимірів; простіше кажучи, у запиті та фрагмента немає спільних слов.
Оцінка 0 – це не аномалія. Це означає, що система правильно визначила: фрагмент не має жодного відношення до запиту.
Один момент щодо точності: формула згладженого IDF може дати трохи негативне значення слову, яке зустрічається у кожному фрагменті, тому косинусна схожість іноді може трохи опуститися нижче нуля. Для реального тексту це майже не має значення, але стверджувати, що вона ніколи не буває негативною, – це перебільшення.
Слабкою стороною TF-IDF з використанням косинусової схожості є синоніми. Запит, який містить слово „automobile“, у порівнянні з документом, де використано слово „car“, отримує оцінку майже нульову, адже враховується лише буквальне збіг словникового запасу, без урахування значення. Моделі щільних ембеддингів, які розташовують синоніми поруч один з одним, створені саме для подолання цієї проблеми.
Крок 4: механізм пошуку
Після того, як векторизація та оцінка схожості готові, процес пошуку полягає у тому, щоб один раз векторизувати кожен фрагмент, векторизувати вхідний запит, оцінити кожен фрагмент щодо нього, відсортувати результати та зберегти кращі k фрагментів:
class VectorStore:
def __init__(self):
self.chunks, self.vocab, self.idf, self.vectors = [], [], {}, []
def index(self, documents, chunk_size=50, overlap=10):
all_chunks = []
for doc in documents:
all_chunks.extend(chunk_text(doc, chunk_size, overlap))
self.chunks = all_chunks
self.vocab = build_vocabulary(all_chunks)
self.idf = compute_idf(all_chunks, self.vocab)
self.vectors = [vectorize(c, self.vocab, self.idf) for c in all_chunks]
def search(self, query, top_k=3):
query_vector = vectorize(query, self.vocab, self.idf)
scored = [(c, cosine_similarity(query_vector, v))
for c, v in zip(self.chunks, self.vectors)]
scored.sort(key=lambda pair: pair[1], reverse=True)
return scored[:top_k]
Одна деталь може мовчки зіпсувати все. Запит потрібно перетворювати за допомогою словникового запасу та значень IDF, обчислених з документів; використання словникового запасу, отриманого лише з самого запиту, є помилкою. Косинусна схожість передбачає вектори однакової довжини, де позиція i позначає одне й те саме слово в кожному з них. Якщо надати запиту власний словниковий запас, то на позиції 5 в одному векторі може бути слово „heart“, а в іншому — „and“, що робить оцінку схожості структурно беззмістовною, а не просто трохи неточною.
Крок 5: доповнення запиту
У цьому кроці не потрібна математика — лише дисципліна. Ви поєднуєте отримані фрагменти з запитом користувача в єдиний запит, який чітко вказує моделі, як поводитися:
def build_augmented_prompt(query: str, retrieved_chunks: list[str]) -> str:
context = "\n\n".join(
f"[Source {i+1}]: {chunk}" for i, chunk in enumerate(retrieved_chunks)
)
return f"""Answer the question using ONLY the context below.
If the answer is not contained in the context, say "I don't know based on the provided context."
Cite which source number(s) you used.
Context:
{context}
Question: {query}
Answer:"""
Два аспекти тут не є суто декоративними.
„Використовувати ЛИШЕ наведений нижче контекст“. Без цього чіткого правила модель поєднує отримані фрагменти з тим, що пам’ятає з процесу навчання, і тоді вже неможливо визначити, яка частина відповіді походить з надійних документів, а яка — це припущення. Таке поєднання, надане з впевненістю, завдає більшої шкоди, ніж повне ігнорування RAG, адже здається, що є джерело, хоча його насправді немає.
Нумеровані позначки джерел та інструкція щодо посилань. Вони дозволяють відстежувати походження відповідей. Коли відповідь є неправильною, що трапляється, позначені джерела показують, який фрагмент ввів модель у оману — чи був сам фрагмент застарілим або неправильним, чи модель неправильно інтерпретувала правильний фрагмент. Без позначок виявлення причини поганої відповіді є переважно припущенням.
Крок 6: об’єднання всього в єдине ціле
Повний процес включає отримання даних, їх покращення та один виклик LLM. Ключовим рішенням у проектуванні є те, щоб індексування та пошук залишалися двома окремими етапами, а не однією функцією, яка виконує обидва завдання:
class RagPipeline:
def __init__(self, client, model="openai/gpt-oss-20b"):
self.store = VectorStore()
self.client = client
self.model = model
def build_index(self, documents: list[str]):
"""Run once, whenever documents are loaded or updated."""
self.store.index(documents, chunk_size=50, overlap=10)
def answer(self, query: str, top_k: int = 3) -> str:
"""Run on every user query — never re-indexes."""
results = self.store.search(query, top_k=top_k)
chunks = [c for c, _ in results]
prompt = build_augmented_prompt(query, chunks)
response = self.client.chat.completions.create(
model=self.model,
max_tokens=500,
messages=[{"role": "user", "content": prompt}],
)
return response.choices[0].message.content
Якби індексування відбувалося всередині функції на кожен запит, кожен запит знову розбивав би текст, перебудовував словниковий запас та перераховував усі вектори TF-IDF для всього збірника, хоча ніщо з цього не змінюється між запитами. Для 10 000 документів та 1 000 запитів це означає 999 ідентичних процедур перебудови. Краще створити індекс один раз, а потім виконувати запити багато разів — саме такий підхід рекомендують FAISS, Chroma та Pinecone. Це правильна архітектура з самого початку, а не передчасна оптимізація.
У чому проблема цієї системи
Ці обмеження варто описувати чесно, адже це не баги. Вони безпосередньо випливають з вищезазначених рішень щодо проектування.
- Відсутнє семантичне розуміння. „Автомобіль“ та „кар“ ніколи не будуть збігатися, оскільки TF-IDF бачить лише буквальне перекриття слів у словниковому запасі.