Лакальны RAG-канвеер на Python без выдаткаў на API
Практычны ўрок: генерацыя з пошукам на CPU з sentence-transformers, numpy і Flan-T5, уключна з разбіўкай на фрагменты, пошукам і A/B-тэстам.
Крыніцы: Build a RAG System From Scratch — Hands-On, No API Costs
Полны ўраджэння на аднойчынных дадзеных паляцоўку можна стварыць і запускіць выключна на персональным камп’ютары, выкарыстоўваючы толькі CPU, бесплатныя адкрытага кода мадэлі, без ключа OpenAI, без карточкі крэдыту і без лімітаў на швыдкасць. Галоўная мета — зразумець кожны элемент, а не проста копіюваць «чорную скрынку».
Найлепшы спосаб — практычная праця: трэба заставіць адкрытым тэрмінал і файл на Python пад назвай rag.py, запускаць кожны блок і прыбліжна адгадваць рэзультат прычымання яго, перш чым выконваць. Пасля большай часткі крокаў прадстаўляюцца короткія вправы, і самэ ў іхіх адбываецца большая частка навучэння.
Проблема, якую рашае RAG
Мадэль мовы знае толькі тое, што было ў яе дадзеных для навучэння. Прыватныя нотаткі, корпоратыўная вікі, зустрэча мінулага тыдня чы будзь-які невядомы документ для яе ёсць недоступнымі. Калі запытаць яе пра іх, яна альбо з’являе незнаёмство, альбо, што горэ, з упэўненасцю выдумвае ўсё сама.
RAG дзеліць задачу на две часты:
- Адзыскваванне: знайсці тыя часткі вашых сабе дакументаў, якія стосуюцца запытання. Гэта проблема адзысквавання.
- Стварэнне: падаць гэтыя часткі модэлю з інструкцыяй адпаведзець на запытанне на ўжотым матэрыяле.
Модэль перастае здагадваліцца і пачынае чытаць. Усё інша — гэта тэхнічныя аспекты для правільнага выканання гэтых двух крокаў.
question ──► [embed] ──► [search your docs] ──► top chunks ──┐
▼
[LLM: "answer using this context"] ──► answer
Бесплатны ўстатку
Тры складнікі павылучаюць усю сістэму. sentence-transformers ператварае тэкст у векторы, выкарыстоўваючы невялікі мадэль, які працуе на CPU і завантажваецца толькі раз. numpy адпрацоўвае вычысленні сэмабільнасці, а весь «база дадзеных вектораў» запішваецца прыблізна ў пяць рэйкаў, таму працэс застаецца відным. transformers з Flan-T5 запішвае адпаведзь, так як ён налаштованы для выканання інструкцый, працуе на CPU і не патрабуе API.
Завантажыце все раз:
pip install sentence-transformers transformers torch numpy
Пры першым запуске кожны мадэль завантажвае сваія вагі з Hugging Face. Функцыя embedder мае розмер апошней 80 MB, а flan-t5-base — апошней 1 GB. Пасля чаго гэтыя даны кешуюцца і мадэль можа працаваць без падключэння да інтарнету. Калі ёсць обмежаны прыемна дыска чыста памяці, flan-t5-small (размерамі апошней 300 MB) можа заменіць базовую мадэль, хоць і з менш высокай якасцю, але з значна большой швальнасцю.
Шаг 1: корпус дакументаў, які мадэль не можа знайсці
RAG лёгкае разумець на прыкладзе дакументаў, якія мадэль не магла бачыць, таму скрыпт пачынаецца з мініятурнай вымышленай вікі компаніі. Створыце файл rag.py і запішыце туда дакументы:
# rag.py
DOCUMENTS = [
"""Nimbus is a fictional note-taking app launched in 2023. The free plan,
called Nimbus Lite, allows up to 50 notes and 1 GB of storage. There are no
collaboration features on the free plan.""",
У вікі ёсць яшчэ артыкулы:
"""Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed
monthly. Pro removes the note limit, gives 50 GB of storage, and unlocks
real-time collaboration with up to 5 people per note.""", """Nimbus stores all notes encrypted at rest using AES-256. End-to-end
encryption is only available on the Pro plan and must be enabled manually in
Settings > Security. Once enabled it cannot be turned off for that note.""", """The Nimbus mobile app supports offline editing. Changes made offline are
queued and sync automatically the next time the device is online. If two
devices edit the same note offline, Nimbus keeps both versions and flags a
conflict for the user to resolve.""", """Nimbus offers a 30-day refund policy on all paid plans, no questions asked.
Refunds are processed to the original payment method within 5 business days.
Annual plans cancelled after 30 days are not refundable but stay active until
the end of the billing period.""", """Nimbus support is available via email at help@nimbus.example and live chat.
Live chat is only staffed for Pro customers, Monday to Friday, 9am to 6pm UTC.
Free-plan users receive email support with a typical 48-hour response time.""",
]
Перш чым што-небудзь стварыць, падтвердзіце, што проблема існуе. У Python-шэле запытайце простую мадэль пра Nimbus:
from transformers import pipeline
gen = pipeline("text2text-generation", model="google/flan-t5-base")
print(gen("How much does Nimbus Pro cost?", max_new_tokens=50)[0]["generated_text"])
Ён выдае якісь рэзультат, а гэты рэзультат будзе некорэктным або порожнім, таму што гэта проста здогадка. Запамятавайце гэты адказ, калі ўступныя інструкцыі падкорэгтуюць яго.
Шаг 2: разбіўка на часткі
Цэлыя дакументы не падаюць у мадэль. Іх разбіваюць на меншыя часткі з двух прычын: адзінакоўка стае болей тачной, таму што знаходзіцца параграф, які адпавядае на запытанне, а не цэлы практычны кярэквенц, а таксама тым, што вікна кантэксту мадэлі ёсць обмежаным.
Дадзіце функцыю разбіўкі на часткі:
def chunk_text(text, chunk_size=60, overlap=15):
"""Split text into overlapping chunks of `chunk_size` words."""
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 by `overlap` so context isn't cut
return chunks
Потым створыце спіс чакункоў, фіксуючы, з каго документа кожны чакунк выйшаў:
# Build our chunk list, remembering which doc each chunk came from
chunks = []
for doc_id, doc in enumerate(DOCUMENTS):
for c in chunk_text(doc):
chunks.append({"doc_id": doc_id, "text": c})print(f"{len(DOCUMENTS)} documents -> {len(chunks)} chunks")
for c in chunks[:3]:
print("-", c["text"][:70], "...")
Перакрыцце існуе таму, што факт можа распадацца на два часткі, напрыклад, палітыка вярнення грошаў, якая роздзелена на «праз 5» і «рабочыя дні». Калі немае перакрыцця, жадна з частак не мае всіх дакументаў факту. Павтарэнне калькі слоў у кожным месцы перакрыцця дапамагае фактам застацца цэлымі пасля такога роздзелу.
Заданне: запрацаваце з параметрам chunk_size=20, ўнаследке чаго будуць стварацца больш маленькія часткі, а пасля спробаваце chunk_size=500, калі кожны з короткіх дакументаў стане адной часткай. Перад запускам спробаваце здагадацца, якая настройка дае болей точныя рэзультаты, але рызікуе втратай суседньего кантэксту. Маленькія часткі ёсць точнымі, але беднымі на кантэкст, і гэта прытрыманне знаходзіцца ў самай сутнасі процэса часткавання. Не існуе універсальна правільнага размеру.
Шаг 3: embeddingi
Умяшчэнне — гэта вектар цялоў, які кодуе значэнне тэксту. Фрагменты тэксту, якія маюць падобныя значэнні, стаюць вектарамі, якія выказваюць адносна тое жа направленне, нават якшо ў іх няма спяльных слоў; таму фраза "Скількі гэта?" знаходзіцца рыжым фразе "Калькуюцца якія цены?"
Імпортуйце модэль умяшчэння:
from sentence_transformers import SentenceTransformer
Потым уключыце кожны фрагмент:
embedder = SentenceTransformer("all-MiniLM-L6-v2")# Embed every chunk. normalize_embeddings=True makes the vectors unit-length,
# which lets us measure similarity with a simple dot product later.
chunk_texts = [c["text"] for c in chunks]
chunk_vectors = embedder.encode(chunk_texts, normalize_embeddings=True)print("vector shape:", chunk_vectors.shape) # (num_chunks, 384)
Кожны элемент цяпер прыяходзіць да 384-вымернага вектара, пры чым 384 — гэта проста колькасць цыфр, якія выкарыстоўваецца гэтым конкрэтным мадэлю для выражэння значэння.
Заданне: пазірце, як працюе значэнне, адмахнуўшыся невялікаг эксперымента:
import numpy as np
pairs = embedder.encode(
["the price of the pro plan", "how much does it cost", "the weather in Paris"],
normalize_embeddings=True,
)
print("price vs cost :", round(float(pairs[0] @ pairs[1]), 3)) # should be HIGH
print("price vs weather:", round(float(pairs[0] @ pairs[2]), 3)) # should be LOW
Выразенне a @ b ўособляе сабою дотычную скарбліву, якая для едынічных вектораў практычна доравнюе косінусу схожасці. Значэнне 1,0 абазаванае на тым, што векторы знаходзяцца ў той самай напрамленні, а 0 — на тым, што воны несуць ніякога зв’язку. Першая пара мае даўаць значэнне значна вышэ за другую, нават якфіразы не маюць спольных слоў; гэта і ёсць семантычны пошук. Паспрабуйце здагадацца прыблізныя значэнні обох чыслаў з точнасцю да 0,2 першым час запуску.
Крок 4: адзысканне
У кроці пошуку неабходна увключыць запіт, паўпоравняць я з кожным вектаром фрагменту і застаўці найболей прыдатныя. Адколькі векторы нормалізаваны, „паўпоравнянне“ ператвараецца на множэнне матрыц.
import numpy as np
Функцыя выкарыстоўвання:
def retrieve(question, k=3):
q_vec = embedder.encode([question], normalize_embeddings=True)[0]
scores = chunk_vectors @ q_vec # cosine similarity to every chunk
top_idx = np.argsort(scores)[::-1][:k] # indices of the k highest scores
return [(chunks[i]["text"], float(scores[i])) for i in top_idx]for text, score in retrieve("How much does Nimbus Pro cost?"):
print(f"[{score:.3f}] {text[:80]}...")
Цэнавака должна выклікватыся на верху з найлепшым рэйтынгам, нават як у запытанні было напісана слова „вартасць“, а ў дасяглінах — „вартасць 8 долераў у месяц“. Гэта значыць падтрымка сэнсу, а не проста падтрымка ключоўых слоў.
Занятка: спробуйце тры запытанні і паглядзіце, што будзе выклікванацца:
retrieve("Магчымае мне адзыскваць свяй грошы?")должна вернуць частку з інформацыяй пра варыцею, нават як слова „варыца“ ніколі не было ў запытанні.retrieve("Чы сябе мае мае даныя ў безпэці?")должна выклікваць частку, якая апісвае шыфраванне.retrieve("Чы ён все ўсё працюе без падключэння да інтарнету?")апэкавана вяртае частку, якая апісвае працу без інтарнету.
Багато людзей недаюць належнай цены гэтам этапу. Калі падчас выкарыстоўвання даных вяртаюцца некарэктны фрагменты, жадных хітрых падказак не можа вывяліць правільную адпаведзь. Якшто інфармацыя некачаная, то і рэзультат будзе некачаным, таму самэ ў першую чаргу трэба прыкласты зусілля да адлучэння бяга.
Шаг 5: стварэнне
Цяпер два часткі з’еднаюцца. Падготавіце падказку, якая будзе перадаць атрыманыя фрагменты ў Flan-T5 і паведаміць яго адпавядаць толькі на ўснове іх.
from transformers import pipeline
Функцыя генеравання:
generator = pipeline("text2text-generation", model="google/flan-t5-base")def rag_answer(question, k=3):
retrieved = retrieve(question, k=k)
context = "\n".join(text for text, _ in retrieved) prompt = f"""Answer the question using only the context below.
If the answer is not in the context, say you don't know.Context:
{context}Question: {question}
Answer:""" out = generator(prompt, max_new_tokens=80)[0]["generated_text"]
return out.strip(), retrievedanswer, sources = rag_answer("How much does Nimbus Pro cost?")
print("ANSWER:", answer)
print("\nBased on:")
for text, score in sources:
print(f" [{score:.3f}] {text[:70]}...")
Паруцьце рэзультат з некорэктным або порожнім адказам з 1-го крока. Модэль і запытанне ідэнтычныя, а ўсё, што змінілася — гэта наявнасць відпаведнага тэксту для чытання. Гэтае разлік і ёсць сумой значэння RAG.
Вядомасці: задайце запытанне, якое не пакрываюцца дасягнікамі, напрыклад, пра кліент для рабочага столу Linux для Nimbus, і вызывайце rag_answer з яго. Правільна налаштавана сістэма павінна з’явіць, што не ведае адказу. Калі ваша сістэма выдумвае адказ, гэта ёсць справжнім спосабам неяксасці, які называецца галюцынацыяю, нават пасля адзысквання інформацыі, і пра гэта пасля будзе сказана ў наступным раздзеле.
Готовы скрыпт
Цікаваце цэлы файл rag.py як адзін працюючы програма з інтэрактыўным цыклам. Ён пачынаецца з імпортаў:
# rag.py — a complete, local, no-API RAG system
import numpy as np
from sentence_transformers import SentenceTransformer
from transformers import pipeline
Далей ідзе весь тэкст:
DOCUMENTS = [
"""Nimbus is a fictional note-taking app launched in 2023. The free plan,
called Nimbus Lite, allows up to 50 notes and 1 GB of storage. There are no
collaboration features on the free plan.""",
"""Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed
monthly. Pro removes the note limit, gives 50 GB of storage, and unlocks
real-time collaboration with up to 5 people per note.""",
"""Nimbus stores all notes encrypted at rest using AES-256. End-to-end
encryption is only available on the Pro plan and must be enabled manually in
Settings > Security. Once enabled it cannot be turned off for that note.""",
"""The Nimbus mobile app supports offline editing. Changes made offline are
queued and sync automatically the next time the device is online. If two
devices edit the same note offline, Nimbus keeps both versions and flags a
conflict for the user to resolve.""",
"""Nimbus offers a 30-day refund policy on all paid plans, no questions asked.
Refunds are processed to the original payment method within 5 business days.
Annual plans cancelled after 30 days are not refundable but stay active until
the end of the billing period.""",
"""Nimbus support is available via email at help@nimbus.example and live chat.
Live chat is only staffed for Pro customers, Monday to Friday, 9am to 6pm UTC.
Free-plan users receive email support with a typical 48-hour response time.""",
]def chunk_text(text, chunk_size=60, overlap=15):
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
return chunksprint("Loading models (first run downloads them)...")
embedder = SentenceTransformer("all-MiniLM-L6-v2")
generator = pipeline("text2text-generation", model="google/flan-t5-base")# Index the documents once at startup
chunks = []
for doc_id, doc in enumerate(DOCUMENTS):
for c in chunk_text(doc):
chunks.append({"doc_id": doc_id, "text": c})
chunk_vectors = embedder.encode(
[c["text"] for c in chunks], normalize_embeddings=True
)def retrieve(question, k=3):
q_vec = embedder.encode([question], normalize_embeddings=True)[0]
scores = chunk_vectors @ q_vec
top_idx = np.argsort(scores)[::-1][:k]
return [(chunks[i]["text"], float(scores[i])) for i in top_idx]def rag_answer(question, k=3):
retrieved = retrieve(question, k=k)
context = "\n".join(text for text, _ in retrieved)
prompt = (
"Answer the question using only the context below. "
"If the answer is not in the context, say you don't know.\n\n"
f"Context:\n{context}\n\nQuestion: {question}\nAnswer:"
)
out = generator(prompt, max_new_tokens=80)[0]["generated_text"]
return out.strip()if __name__ == "__main__":
print("RAG ready. Ask about Nimbus (or type 'quit').\n")
while True:
q = input("You: ").strip()
if q.lower() in {"quit", "exit", ""}:
break
print("Nimbus bot:", rag_answer(q), "\n")
Пачаце з:
python rag.py
Потым спробуйце запіты на кшталт таго, калі трэба адзначыць вярненне грошаў, чы розпачынаючыя корыстувальнікі маюць можлівасць ведаць чат у рэальным часе, або які ў размерах безкоштовны прызначаны прыёмнік для зберагачвання дакументаў.
Падтверджэнне цянічнасці за дапамойкай тэста A/B
Адны эксперымент чытра паказвае сутнасць, ставячы той самы запіт з і без адпраўленага кантэксту. Спачатку — базовая функцыя:
def no_rag(question):
out = generator(f"Question: {question}\nAnswer:", max_new_tokens=80)
return out[0]["generated_text"].strip()
Потым параболе:
q = "Can free-plan Nimbus users use live chat support?"
print("WITHOUT context:", no_rag(q))
print("WITH context :", rag_answer(q))
Адказ без кантэксту — гэта толькі здагадка, тады як адказ з кантэкстам павінен паведамляць, што жывы чат доступны толькі для прыемнікаў Pro, а безплатныя корыстувачы атрымваюць пісьмо. Прабавіце тры разныя запытанні. Гэты прыклад A/B ёсць найболей пераканаючым дэманам RAG, таму зберагіце яго для пояснення ідэі іншым.
Апграды за практычным рынкам
Калі у вас ёсць працюючы сістэма, гэтыя апграды маюць наўрадзе найбольша значэнне, прыблізна за практычным рынкам:
- Адзначыць выкладнікі. Функцыя вялікай часткай вяртае
retrieved, таму выдрукуйце фрагменты пад кожным адказам. Справжнія продукты RAG завышаюць успех або парадуць неудачу самэлькі на гэтым. Адказ з выкладнікамі ёсць надзеяным, а адказ без ягох — гэта толькі здагадка, практычна адзначаная гарным шрыфтом.
k да 5 і задаце мінімальную оцэнку, адхоўваючы толькі тыя даныя, якія стояць вышэй за прыблізна 0.25. Калі жадны фрагмент не паспеў, адпаведзце "Я не ведаю" і вучоўнае аб’екта не вызывайце ўжо. эта мера пазбавляе ад большасці галюцынацый, падобных тым у вправе па кроку 5.chunk_vectors @ q_vec скануе кожны фрагмент, што падходзіць для сотняў, але ўповільнюе працу для мільйонаў. Chroma (pip install chromadb) стварае індэксы вектараў для шыранага пошуку і зберагае іх на дыску, таму ў момент запуску нічога не трэба перзберагаць. Канцэпцыі імбеддынгу, хранення і пошуку за косай лініёй застаюцца тымі ж, а Chroma проста выконвае іх у масштабных умовах.generator(...) на запіт до яго локального API. Безкоштовній рывень API ад Google Gemini або Groq таксама працюе, калі замініты лішэ радок генераціі в rag_answer. Змінюецца лішэ одна функція, тоды як працэс пошуку залішаецца незміннім.rag_answer і перэвірце правільнісць. Гэта не надто прывабліво, але саме гэта відрызняе прототып ад готового продукту, аджэ тыя, што не віміруецца, неможліво покрашчыты.Дыягнастыка проблем
RAG — гэта пошук плюс адаптавальны помачнік для напісання, якія з’едынаюцца, і практычна кожная проблема выклікаецца адной з эйх частак. Некарэктны адказ, калі быў запрашаны правільны фрагмент, — гэта проблема з генераванням адказу або запытам. Некарэктны адказ, калі правільны фрагмент ніколі не быў запрашаны, — гэта проблема з запрашам, якая стосуецца фрагментавання, усадків або параметра k. Перш чым што-небудзь зменіць, трэба выявіць, калія з частак не працуе. Функцыі retrieve() і rag_answer() залічваюцца окрэмымі саме для таго, каб кожную частку можна было пераглядзець окрэма, і гэта раз’еднанне ёсць найважлівейшым прыемам у гэтым апрабоўванні.
Усуненне проблем
- Памедзелы або нерэальны завантажэнне: мадэлі пад час першага запуску загружаюцца з Hugging Face і кэшуюцца пад
~/.cache/huggingface. Пасля кэшаванняя яны працуюць без падключэння да інтэрнэту.
google/flan-t5-small з chunk_size равным 40. Мадэль эмбедынгу ўстатківа мала, таму яна не будзе ствараць працявых прыгонкаў.pip install torch --index-url https://download.pytorch.org/whl/cpu.max_new_tokens, каліколькі Flan-T5 часта зупіняецца рана. Большы генератор, напрыклад, той, які выдае Ollama, дапамагае больш, чым будзь-якіія настройкі.Развязваўшы асанаванні, вы атрымаеце глыбэйшае розуменне тэхнікі RAG, чым большасць людзей за дапамогою простых бібліятэкаў.