خط أنابيب RAG المحلي

ابنِ نظام التوليد المعزز بالاسترجاع الكامل الذي يعمل بالكامل على جهازك. لا تغادر أي بيانات شبكتك أبداً.

نظرة عامة على البنية

┌─────────────────────────────────────────────────────────────────┐
│                    خط أنابيب RAG المحلي الكامل                   │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  المستندات ──► التقطيع ──► تضمينات محلية ──► مخزن المتجهات     │
│                            (nomic-embed)     (FAISS/Chroma)     │
│                                                                 │
│  الاستعلام ──► تضمين ──► استرجاع ──► LLM محلي ──► الإجابة       │
│               (nomic)    (top-k)    (llama3.2)                  │
│                                                                 │
│  كل شيء يعمل محلياً - خصوصية بيانات كاملة                       │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

التنفيذ الكامل

import ollama
import numpy as np
from typing import List

class LocalRAG:
    """خط أنابيب RAG محلي بالكامل مع Ollama."""

    def __init__(
        self,
        embedding_model: str = "nomic-embed-text",
        llm_model: str = "llama3.2",
        chunk_size: int = 500,
        chunk_overlap: int = 50
    ):
        self.embedding_model = embedding_model
        self.llm_model = llm_model
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.chunks: list[str] = []
        self.embeddings: list[list[float]] = []

    def add_documents(self, documents: list[str]):
        """قطّع وضمّن المستندات."""
        # قطّع المستندات
        for doc in documents:
            doc_chunks = self._chunk_text(doc)
            self.chunks.extend(doc_chunks)

        # ضمّن جميع القطع
        if self.chunks:
            response = ollama.embed(
                model=self.embedding_model,
                input=self.chunks
            )
            self.embeddings = response["embeddings"]

    def _chunk_text(self, text: str) -> list[str]:
        """قسّم النص إلى قطع متداخلة."""
        chunks = []
        start = 0
        while start < len(text):
            end = start + self.chunk_size
            chunk = text[start:end]
            if chunk.strip():
                chunks.append(chunk)
            start = end - self.chunk_overlap
        return chunks

    def _cosine_similarity(self, a: list[float], b: list[float]) -> float:
        """احسب تشابه جيب التمام."""
        a, b = np.array(a), np.array(b)
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

    def retrieve(self, query: str, top_k: int = 3) -> list[str]:
        """استرجع القطع ذات الصلة للاستعلام."""
        # ضمّن الاستعلام
        query_embedding = ollama.embed(
            model=self.embedding_model,
            input=query
        )["embeddings"][0]

        # احسب التشابهات
        similarities = [
            (i, self._cosine_similarity(query_embedding, emb))
            for i, emb in enumerate(self.embeddings)
        ]

        # رتب واحصل على top-k
        similarities.sort(key=lambda x: x[1], reverse=True)
        top_indices = [i for i, _ in similarities[:top_k]]

        return [self.chunks[i] for i in top_indices]

    def query(self, question: str, top_k: int = 3) -> str:
        """أجب على سؤال باستخدام RAG."""
        # استرجع السياق ذا الصلة
        context_chunks = self.retrieve(question, top_k)
        context = "\n\n".join(context_chunks)

        # ولّد الإجابة
        prompt = f"""Answer the question based on the following context.
If the answer isn't in the context, say "I don't have enough information."

Context:
{context}

Question: {question}

Answer:"""

        response = ollama.chat(
            model=self.llm_model,
            messages=[{"role": "user", "content": prompt}]
        )

        return response["message"]["content"]

# الاستخدام
rag = LocalRAG()

# أضف مستنداتك
documents = [
    """Python is a high-level programming language created by Guido van Rossum.
    It was first released in 1991. Python emphasizes code readability with its
    notable use of significant indentation. Python supports multiple programming
    paradigms including procedural, object-oriented, and functional programming.""",

    """Machine learning is a subset of artificial intelligence that focuses on
    building systems that learn from data. Common ML algorithms include decision
    trees, neural networks, and support vector machines. Python is the most
    popular language for machine learning due to libraries like scikit-learn,
    TensorFlow, and PyTorch."""
]

rag.add_documents(documents)

# استعلم من النظام
answer = rag.query("Who created Python and when?")
print(answer)

استخدام LangChain مع ChromaDB

from langchain_ollama import ChatOllama, OllamaEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# هيئ النماذج المحلية
embeddings = OllamaEmbeddings(model="nomic-embed-text")
llm = ChatOllama(model="llama3.2")

# مستندات نموذجية
documents = [
    "Ollama is an open-source tool for running large language models locally.",
    "LangChain is a framework for developing applications powered by LLMs.",
    "ChromaDB is an open-source vector database for AI applications.",
    "RAG combines retrieval with generation for more accurate responses."
]

# قسّم المستندات
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
splits = text_splitter.create_documents(documents)

# أنشئ مخزن متجهات (مستمر محلياً)
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# أنشئ المسترجع
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# طلب RAG
prompt = ChatPromptTemplate.from_template("""
Answer based only on this context:

{context}

Question: {question}
""")

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# ابنِ السلسلة
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# استعلم
answer = rag_chain.invoke("What is Ollama?")
print(answer)

تحميل المستندات من الملفات

from langchain_community.document_loaders import (
    TextLoader,
    PyPDFLoader,
    DirectoryLoader
)

# حمّل ملف نصي واحد
text_loader = TextLoader("document.txt")
text_docs = text_loader.load()

# حمّل PDF
pdf_loader = PyPDFLoader("document.pdf")
pdf_docs = pdf_loader.load()

# حمّل جميع الملفات من مجلد
dir_loader = DirectoryLoader(
    "./documents",
    glob="**/*.txt",
    loader_cls=TextLoader
)
all_docs = dir_loader.load()

# أضف لمخزن المتجهات
vectorstore.add_documents(all_docs)

تدفق ردود RAG

from langchain_ollama import ChatOllama, OllamaEmbeddings
from langchain_community.vectorstores import Chroma

embeddings = OllamaEmbeddings(model="nomic-embed-text")
llm = ChatOllama(model="llama3.2")

# افترض أن vectorstore أُنشئ مسبقاً
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

def stream_rag_response(question: str):
    """تدفق رد RAG رمزاً برمز."""
    # استرجع السياق
    docs = vectorstore.similarity_search(question, k=3)
    context = "\n\n".join(doc.page_content for doc in docs)

    prompt = f"""Answer based on this context:

{context}

Question: {question}
Answer:"""

    # تدفق الرد
    for chunk in llm.stream(prompt):
        print(chunk.content, end="", flush=True)
    print()

stream_rag_response("What are the benefits of local LLMs?")

التقييم

def evaluate_rag(rag_system, test_questions: list[dict]) -> dict:
    """قيّم نظام RAG على أسئلة اختبار."""
    results = {
        "total": len(test_questions),
        "correct": 0,
        "details": []
    }

    for item in test_questions:
        question = item["question"]
        expected = item["expected_answer"]

        answer = rag_system.query(question)

        # مطابقة كلمات مفتاحية بسيطة (استخدم RAGAS للإنتاج)
        is_correct = any(
            keyword.lower() in answer.lower()
            for keyword in expected.split()
        )

        results["details"].append({
            "question": question,
            "answer": answer,
            "expected": expected,
            "correct": is_correct
        })

        if is_correct:
            results["correct"] += 1

    results["accuracy"] = results["correct"] / results["total"]
    return results

# اختبر
test_set = [
    {"question": "Who created Python?", "expected_answer": "Guido van Rossum"},
    {"question": "When was Python released?", "expected_answer": "1991"}
]

eval_results = evaluate_rag(rag, test_set)
print(f"الدقة: {eval_results['accuracy']:.1%}")

المزايا الرئيسية لـ RAG المحلي

الميزة	RAG محلي	RAG سحابي
خصوصية البيانات	كاملة	البيانات تُرسل للسحابة
التكلفة	العتاد فقط	رسوم لكل رمز
زمن الاستجابة	ثابت	متغير
الإنترنت	غير مطلوب	مطلوب
التخصيص	تحكم كامل	محدود

خط أنابيب RAG الخاص بك يعمل الآن بالكامل محلياً. بعد ذلك، سنضيف قدرات استدعاء الوظائف. :::