RAG на чыстым Python: разбіўка, TF-IDF, касінуснае падабенства і промпт
Будуем генерацыю з пошукам толькі з модулем math: разбіўка слізгальным акном, вектары TF-IDF, касінуснае ранжаванне, абгрунтаваны промпт і яго межы.
Крыніцы: Building RAG From First Principles with just plain python .
У стандартным нарадчыках па RAG прасіцца установіць пяць кнігакошак і вызваць .from_documents(). Чэрэз дзесяць хвілін працюе дамавай, але нічога з яго не разумееш. Калі працэс выкарыстоўвання дадзеных вяртае некалькісныя рэзультаты або мадэль стварае халюцынаціі, ты застрагаеш, бо кнігакошак выконваў усю працу заместа табе.
У даным случае пайплайн ствараецца наадворот — простым Python, выкарыстоўваючы толькі модуль math і аднай вызову LLM у фармаце простага тэксту ў канцы. Ніякіх модэляў эмбеддынгу, ні FAISS, ні Pinecone. Розумеючы механізмы, ты зможаш лепшая разумець, што відбываецца, калі пазней будзеш выкарыстоўваць LangChain або справжню вектарную базу дадзеных. У канцы ў табе будзе працуючы пайплайн, а таксама розуменне кожной цыфры, якая ў яму выраховываецца.
Проблема, якую рашае RAG
Знанні мадэлі языка застойваюцца, калі завершаецца навчанне, і ўсё яны знаходзяцца толькі ў ўжоўвешчаннях мадэлі. Ёй неабходна можлівасць адпавядаць на запытанні пра внутршняе дакументацыю компаніі, кодавую базу чыста пра новыні сёння зранку, але два очывіцыя варыянты не задовольняюць:
- Перэнавчыць модэль. Гэта дзейсніца повольна і дорога, а змяны, якія з’явяцца завтра, заставяюць пачынаць усё значыну.
- Увесты ўсё у запыт. Гэты падход не ўзмацнівае эфектыўнасць. Нават вікно на мільйон токеноў не можа разам паклаць 50 000 дакументоў, а даследжэнняя ўрожайных модэляў з дзейсным кантэкстам раз за разам апісваюць эфект „загублення ў серэдзіне“, калі інфармацыя, захаваная глыбока ў вялікім запыце, выкарыстоўваецца гэрш, нават калі ёй і ўсё ж месца. Дадатковыя токены таксама прыводзяць да большага часу адпаведзі і вышэйшых віткоў за кожны вызов.
Рэальны секрэт RAG ўсё просты, як можа здавацца. Дакументы застаюцься у зовнішнім сховішчы, а пад час обработкі запіту выкарыстоўваюцца толькі тыя фрагменты, якіе стосуюцца гэтага запіту. LLM ніколі не мае вашых дадзеных — ён кожны раз отрымае новы, актуальны фрагмент.
Такім чынам, кожная сістэма RAG складаецца з двух частак: выкарыстоўвання дадзеных, якая шукае правы тэкст у вельмі большай колькасці інфармацыі па заданаму запіту, і генераванне, якая перадае гэты тэкст разам з запітам у LLM і атрымляе адпаведны адказ. У наступных частках будуць показаны способы стварэння гэтых двух частак з базовых элементаў.
Шаг 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 паказвае, як часта з’являецца слова ў даным фрагменце. Высокі показнік для слова "kidney" указвае на тое, што фрагмент стосуецца ніркаў.
- 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 з сяродзіннасцю косайну ёсць сынонімы. Запіт, у яком указана фраза "автомабіль", пры аналізе дакумента, у яком напісана фраза "машына", дае рэзультат майжэ нулевы, таму што уваажаецца толькі буквальна перакрыцча слоў, а значэнне не бераецася ў расчыт. Мадэлі густых укладанняў, якія размешчаюць сынонімы падалей адзін ад другога, створаны самэ для таго, каб заполніць гэты праслойк.
Крок 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 магчыма будзе «серца» ў аднам векторы і «і» у іншым, што робіць значэнне балу супадання структурна беззначным, а не проста трохі некальканым.
Шаг 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 бачыць толькі буквальнае перакрыцце словніка.