RAG برای اسناد فارسی: chunk، embedding، جستجوی کسینوسی و پاسخ با ارجاع — با کد

مدل زبانی چیزی دربارهٔ آیین‌نامهٔ داخلی شرکت تو، قراردادهای سال گذشته یا مستندات محصولت نمی‌داند. RAG (Retrieval-Augmented Generation) راه استاندارد حل این مشکل است: اسناد را به تکه‌های کوچک تقسیم می‌کنی، برای هر تکه یک بردار عددی (embedding) می‌سازی، موقع پرسش نزدیک‌ترین تکه‌ها را پیدا می‌کنی و همان‌ها را همراه سؤال به مدل می‌دهی تا با استناد جواب بدهد. این مقاله همهٔ چهار مرحله را برای متن فارسی، در یک فایل پایتون بدون فریم‌ورک، پیاده می‌کند و بعد دربارهٔ چیزهایی حرف می‌زند که RAG فارسی را از دمو به محصول می‌رساند.

بدون LangChain و بدون دیتابیس برداری شروع می‌کنیم؛ با numpy تا حدود صد هزار تکه، جستجو زیر چند ده میلی‌ثانیه است و هر خط کد را می‌فهمی. وقتی به سقف رسیدی، همان بردارها را به pgvector یا Qdrant منتقل می‌کنی.

مرحلهٔ ۱ — تکه‌کردن متن فارسی

اندازهٔ chunk مهم‌ترین تصمیم RAG است. تکهٔ خیلی کوچک بافت را از دست می‌دهد و تکهٔ خیلی بزرگ، هزینهٔ هر پاسخ را بالا می‌برد و نویز اضافه می‌کند. برای اسناد فارسی معمولی (آیین‌نامه، مستند فنی، پاسخ‌های پشتیبانی) ۶۰۰ تا ۹۰۰ کاراکتر با یک جمله هم‌پوشانی نقطهٔ شروع خوبی است. سه نکتهٔ مخصوص فارسی:

• روی مرز جمله ببُر، نه روی تعداد کاراکتر ثابت. علائم پایان جمله در فارسی شامل «.»، «؟» و «!» است و خیلی از متن‌ها با خط جدید جمله را تمام می‌کنند. بریدن وسط جمله، معنی را برای embedding خراب می‌کند.
• نیم‌فاصله را نگه دار. بعضی ابزارها موقع تمیزکاری، نیم‌فاصله (U+200C) را حذف یا به فاصله تبدیل می‌کنند؛ «می‌شود» به «می شود» تبدیل می‌شود و کیفیت هم جستجو و هم پاسخ پایین می‌آید.
• هم‌پوشانی یک جمله‌ای کافی است تا اطلاعاتی که دقیقاً روی مرز افتاده، در هر دو تکه باشد.

مرحلهٔ ۲ — embedding

endpoint POST /v1/embeddings همان قرارداد OpenAI را دارد: model و input (رشته یا لیست رشته) می‌گیرد و برای هر ورودی یک بردار برمی‌گرداند. برای مدل‌های embedding هزینه فقط بر اساس توکن ورودی است و در مقایسه با chat بسیار کم است؛ embedding کردن یک سند صد صفحه‌ای معمولاً چند صد تومان می‌شود.

بردارها را بعد از دریافت نرمال می‌کنیم (طول یک). با این کار شباهت کسینوسی به یک ضرب داخلی ساده تبدیل می‌شود و جستجو با یک ضرب ماتریسی انجام می‌شود.

مرحلهٔ ۳ و ۴ — جستجو و پاسخ با ارجاع

پرسش کاربر با همان مدل embedding می‌شود، ضرب ماتریسی امتیاز همهٔ تکه‌ها را یک‌جا می‌دهد، و چهار تکهٔ برتر با شماره و نام منبع در prompt می‌روند. system prompt سه چیز را اجبار می‌کند: فقط از تکه‌ها جواب بده، بعد از هر ادعا شمارهٔ تکه را در کروشه بیاور، و اگر جواب در تکه‌ها نیست صریح بگو. همین سه جمله بیشترِ توهم مدل را حذف می‌کند.

# rag.py — RAG فارسی: chunk → embedding → جستجوی کسینوسی → پاسخ با ارجاع
# pip install openai numpy
import os, re, json, sys
import numpy as np
from openai import OpenAI

client = OpenAI(
    base_url=os.getenv("UTTAPEN_BASE_URL", "https://api.uttapen.ir/v1"),
    api_key=os.environ["UTTAPEN_API_KEY"],
)
EMBED_MODEL = os.environ["EMBED_MODEL"]           # شناسهٔ مدل embedding را از /models بردار
CHAT_MODEL = os.getenv("CHAT_MODEL", "openai/gpt-5-mini")
BATCH = int(os.getenv("EMBED_BATCH", "32"))

# ---------- ۱) chunk کردن متن فارسی ----------
_SENT = re.compile(r"(?<=[\.\!\?؟।\n])\s+")

def chunk_text(text: str, max_chars=900, overlap_sents=1) -> list[str]:
    text = text.replace("\r", "")
    sents = [s.strip() for s in _SENT.split(text) if s.strip()]
    chunks, cur = [], []
    for s in sents:
        if sum(len(x) for x in cur) + len(s) > max_chars and cur:
            chunks.append(" ".join(cur))
            cur = cur[-overlap_sents:]        # هم‌پوشانی: جملهٔ آخر به chunk بعدی هم می‌رود
        cur.append(s)
    if cur:
        chunks.append(" ".join(cur))
    return chunks

# ---------- ۲) embedding ----------
def embed(texts: list[str]) -> np.ndarray:
    out = []
    for i in range(0, len(texts), BATCH):
        resp = client.embeddings.create(model=EMBED_MODEL, input=texts[i:i + BATCH])
        out.extend(d.embedding for d in sorted(resp.data, key=lambda d: d.index))
    vecs = np.asarray(out, dtype=np.float32)
    return vecs / (np.linalg.norm(vecs, axis=1, keepdims=True) + 1e-9)   # نرمال → dot = cosine

def build_index(doc_paths: list[str]) -> dict:
    chunks, meta = [], []
    for p in doc_paths:
        for j, c in enumerate(chunk_text(open(p, encoding="utf-8").read())):
            chunks.append(c)
            meta.append({"source": os.path.basename(p), "chunk": j})
    return {"chunks": chunks, "meta": meta, "vecs": embed(chunks)}

# ---------- ۳) جستجو ----------
def search(index: dict, query: str, k=4) -> list[tuple[float, int]]:
    q = embed([query])[0]
    scores = index["vecs"] @ q                    # کسینوس، چون هر دو نرمال‌اند
    top = np.argsort(-scores)[:k]
    return [(float(scores[i]), int(i)) for i in top]

# ---------- ۴) پاسخ با ارجاع ----------
def answer(index: dict, question: str) -> str:
    hits = search(index, question)
    context = "\n\n".join(
        f"[{n}] ({index['meta'][i]['source']} #{index['meta'][i]['chunk']})\n{index['chunks'][i]}"
        for n, (_, i) in enumerate(hits, 1)
    )
    resp = client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[
            {"role": "system", "content":
             "فقط بر اساس قطعه‌های داده‌شده جواب بده. بعد از هر ادعا شمارهٔ قطعه را در کروشه بیاور، مثل [2]. "
             "اگر جواب در قطعه‌ها نیست، صریح بگو «در اسناد نیامده»."},
            {"role": "user", "content": f"قطعه‌ها:\n{context}\n\nپرسش: {question}"},
        ],
        max_tokens=400,
        temperature=0,
    )
    return resp.choices[0].message.content

if __name__ == "__main__":
    idx = build_index(sys.argv[1:-1])
    np.save("index.npy", idx["vecs"]); json.dump({"chunks": idx["chunks"], "meta": idx["meta"]},
                                                open("index.json", "w"), ensure_ascii=False)
    print(answer(idx, sys.argv[-1]))

اجرا:

export UTTAPEN_API_KEY=sk-up-...
export EMBED_MODEL=<شناسهٔ مدل embedding از /models>
python rag.py docs/ayin-name.txt docs/faq.txt "حداقل مبلغ شارژ چقدر است؟"

خروجی برای یک سند نمونه دربارهٔ قوانین کیف پول، چیزی شبیه این می‌شود (متن نمونه است و به مدل بستگی دارد):

حداقل مبلغ شارژ پنجاه هزار تومان است [2]. مبلغ بعد از تأیید پرداخت بلافاصله به کیف پول اضافه می‌شود [2].

فایل‌های index.npy و index.json کنار هم ذخیره می‌شوند تا دفعهٔ بعد بدون embedding مجدد، مستقیم جستجو کنی. برای یک سرویس واقعی، همین دو فایل را در startup بارگذاری کن و فقط تابع answer را پشت یک endpoint بگذار.

هزینهٔ هر پاسخ از کجا می‌آید

در RAG، هزینهٔ embedding یک‌بار پرداخت می‌شود و ناچیز است؛ هزینهٔ اصلی در مرحلهٔ پاسخ است، چون هر بار k تکه به‌عنوان ورودی به مدل chat می‌رود. با چهار تکهٔ ۹۰۰ کاراکتری فارسی (حدود ۱٬۵۰۰ تا ۲٬۰۰۰ توکن) و ۳۰۰ توکن پاسخ، هر جواب با openai/gpt-5-mini در حدود ۱۵۰ تومان و با openai/gpt-5-nano زیر ۳۰ تومان درمی‌آید. سه اهرم داری: k را کم کن، chunk را کوچک‌تر کن، یا مدل ارزان‌تر بگذار. برای بیشتر پرسش‌وپاسخ‌های داخلی، مدل ارزان با تکه‌های خوب از مدل گران با تکه‌های بد بهتر جواب می‌دهد.

عدد دقیق هر پاسخ در هدر X-Uttapen-Cost-Toman برمی‌گردد. اگر می‌خواهی پاسخ را استریم کنی (برای رابط چت داخلی طبیعی است)، تابع answer را با stream=True بازنویسی کن؛ الگویش در مستندات استریم هست.

چیزهایی که RAG فارسی را واقعاً خوب می‌کند

آزمون embedding قبل از انتخاب. کیفیت مدل‌های embedding در فارسی به‌اندازهٔ انگلیسی یکنواخت نیست. سی سؤال واقعی بنویس و برای هر کدام مشخص کن جواب در کدام تکه است. برای هر مدل کاندید، index بساز و ببین در چند درصد سؤال‌ها تکهٔ درست در چهار نتیجهٔ اول هست. این عدد (recall@4) معیار انتخاب توست، نه اسم مدل.

جستجوی ترکیبی. برای اصطلاح‌های دقیق — شمارهٔ ماده، کد محصول، اسم خاص — جستجوی واژگانی (BM25 یا حتی LIKE ساده) گاهی از embedding بهتر است. یک راه ساده: نتایج هر دو روش را بگیر، امتیازها را نرمال کن و جمع بزن. کتابخانهٔ rank_bm25 برای این کار کافی است و برای فارسی فقط به یک tokenizer ساده روی فاصله و نیم‌فاصله نیاز دارد.

بازنویسی پرسش. سؤال کاربر اغلب کوتاه و مبهم است («سقفش چقدره؟»). قبل از embedding، با یک مدل ارزان پرسش را با توجه به تاریخچهٔ گفتگو به یک جملهٔ کامل و مستقل بازنویسی کن. هزینه‌اش چند تومان است و recall را محسوس بالا می‌برد.

عنوان در تکه. اگر سند ساختار دارد (فصل، ماده، تیتر)، عنوان بخش را به ابتدای هر تکه اضافه کن. تکهٔ «ماده ۱۲ — استرداد وجه: ...» خیلی بهتر از تکه‌ای که با «در این صورت مبلغ...» شروع می‌شود، پیدا می‌شود.

فیلتر متادیتا. وقتی اسناد چند دسته‌اند (قرارداد، آیین‌نامه، FAQ)، دستهٔ سند را در meta نگه دار و قبل از جستجوی برداری، بر اساس آن فیلتر کن. numpy با یک mask بولی این کار را می‌کند.

تشخیص «در اسناد نیامده». اگر بالاترین امتیاز شباهت زیر آستانه‌ای بود (بسته به مدل، معمولاً بین ۰٫۳ تا ۰٫۵)، اصلاً به مدل chat نرو و همان جملهٔ ثابت را برگردان. هم ارزان‌تر است و هم از جواب‌های ساختگی جلوگیری می‌کند. آستانه را با همان سی سؤال آزمون کالیبره کن.

نگهداری index وقتی اسناد عوض می‌شوند

اسناد زنده‌اند: آیین‌نامه اصلاح می‌شود، صفحهٔ FAQ جواب جدید می‌گیرد، قرارداد قدیمی باطل می‌شود. اگر هر بار کل مجموعه را دوباره embedding کنی، هم وقت تلف می‌شود و هم پول. راه ساده این است که برای هر تکه یک hash از متنش (مثلاً sha1) کنار meta نگه داری. در به‌روزرسانی، اسناد را دوباره تکه کن، hash هر تکه را با index فعلی مقایسه کن، فقط تکه‌های جدید یا تغییرکرده را embedding کن و تکه‌هایی که دیگر در هیچ سندی نیستند را حذف کن. با numpy این یعنی چند خط mask و np.concatenate؛ با دیتابیس برداری یعنی upsert و delete با شناسهٔ تکه.

یک نکتهٔ پنهان: اگر مدل embedding را عوض کنی، همهٔ بردارها باید از نو ساخته شوند، چون بردارهای دو مدل مختلف با هم قابل مقایسه نیستند. نام مدل را کنار index ذخیره کن و در startup چک کن که با EMBED_MODEL فعلی یکی است؛ وگرنه نتایج بی‌سروصدا بی‌معنی می‌شوند و کسی متوجه نمی‌شود.

سه اشتباه رایج

اول، فرستادن کل سند به‌جای تکه‌های مرتبط، به این امید که مدل با context بزرگ خودش پیدا می‌کند. کار می‌کند ولی هر پرسش به اندازهٔ کل سند هزینه دارد و دقت با بزرگ شدن ورودی پایین می‌آید. دوم، اعتماد به شمارهٔ ارجاع بدون نمایش متن منبع؛ کاربر باید بتواند روی «[2]» کلیک کند و تکهٔ اصلی را ببیند، وگرنه ارجاع فقط تزئین است. سوم، آزمون نکردن با سؤال‌هایی که جوابشان در اسناد نیست؛ نیمی از ارزش RAG در «نه» گفتن درست است و این را فقط با سؤال‌های بی‌جواب می‌شود سنجید.

محدودیت‌ها

RAG جادو نیست. اگر جواب در اسناد نباشد، بهترین حالت این است که مدل بگوید نیست. اگر تکه‌ها بد بریده شده باشند، مدل خوب هم جواب بد می‌دهد. و برای اسنادی با جدول‌های پیچیده یا نمودار، متن ساده کافی نیست و به استخراج ساخت‌یافتهٔ جداگانه نیاز داری. از نظر داده هم یادت باشد متن تکه‌ها و پرسش به ارائه‌دهندهٔ مدل می‌رسد؛ gateway چیزی ذخیره نمی‌کند، ولی برای اسناد محرمانه سیاست ارائه‌دهندهٔ زیرین را در نظر بگیر.

کد این مقاله روی gateway اجرا شده: فرمت /v1/embeddings، batch کردن ورودی، ترتیب index در پاسخ و مسیر chat همان‌طور که نوشتیم کار می‌کنند. برای امتحان با اسناد خودت یک کلید از داشبورد بساز؛ با چند هزار تومان می‌توانی چند صد صفحه را index کنی و ده‌ها پرسش بزنی.