دليل شامل لبناء نظام RAG: إنشاء ذكاء اصطناعي جاهز للإنتاج باستخدام ArXiv Paper Curator

⏱️ وقت القراءة المتوقع: 12 دقيقة

مقدمة: بناء أنظمة RAG جاهزة للإنتاج

لقد أصبح الجيل المعزز بالاسترجاع (RAG) حجر الأساس لتطبيقات الذكاء الاصطناعي الحديثة، مما يمكن النماذج اللغوية الكبيرة من الوصول إلى قواعد المعرفة الواسعة والاستدلال عليها. ومع ذلك، فإن معظم الدروس التعليمية تركز على أمثلة بسيطة تفشل في معالجة تحديات العالم الحقيقي مثل قابلية التوسع والمراقبة ونشر الإنتاج.

يسد مشروع ArXiv Paper Curator من Jam With AI هذه الفجوة من خلال توفير منهج كامل لمدة 6 أسابيع لبناء أنظمة RAG جاهزة للإنتاج. سيرشدك هذا الدليل عبر الرحلة الكاملة، من الإعداد الأساسي إلى استراتيجيات المراقبة والتخزين المؤقت المتقدمة.

🎯 ما ستبنيه

بنهاية هذا الدليل، ستكون قد بنيت نظام RAG متطور يقوم بـ:

• جمع ومعالجة الأوراق البحثية من ArXiv تلقائياً
• تنفيذ البحث المختلط الذي يجمع بين البحث الكلمي BM25 والتشابه الشعاعي
• توفير قدرات سؤال وجواب ذكية باستخدام النماذج اللغوية المحلية
• تضمين مراقبة الإنتاج مع إمكانية الملاحظة الشاملة
• تطبيق استراتيجيات التخزين المؤقت للأداء الأمثل
• تقديم واجهة ويب للاستكشاف التفاعلي

🏗️ نظرة عامة على معمارية النظام

يتبع ArXiv Paper Curator معمارية معيارية جاهزة للإنتاج:

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   جمع البيانات   │    │     المعالجة    │    │  البحث و RAG   │
│   (ArXiv API)   │───▶│  (PDF + نص)    │───▶│ (فهرس مختلط)   │
└─────────────────┘    └─────────────────┘    └─────────────────┘
         │                       │                       │
         ▼                       ▼                       ▼
┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   PostgreSQL    │    │   OpenSearch    │    │    Gradio UI    │
│ (البيانات الوصفية)│    │  (فهرس البحث)   │    │ (واجهة المستخدم) │
└─────────────────┘    └─────────────────┘    └─────────────────┘

📋 المتطلبات المسبقة

قبل البدء، تأكد من وجود:

• Docker & Docker Compose: للخدمات المحتواة
• Python 3.11+: لتشغيل التطبيق
• 8GB+ RAM: لتشغيل جميع الخدمات محلياً
• 10GB+ مساحة تخزين: للأوراق والفهارس

🚀 دليل التنفيذ أسبوعياً

الأسبوع الأول: أساس البنية التحتية

يؤسس الأسبوع الأول البنية التحتية الأساسية باستخدام Docker Compose لتنسيق خدمات متعددة.

إعداد البيئة

# استنساخ المستودع
git clone https://github.com/jamwithai/arxiv-paper-curator.git
cd arxiv-paper-curator

# بدء جميع الخدمات
make start

# التحقق من الحالة
make health

معمارية الخدمات الأساسية

يستخدم النظام عدة خدمات محتواة:

1. PostgreSQL: يخزن البيانات الوصفية للأوراق والعلاقات
2. OpenSearch: يوفر قدرات البحث الكلمي والشعاعي
3. FastAPI: يخدم نقاط النهاية لواجهة برمجة التطبيقات RESTful
4. Gradio: يقدم واجهة الويب التفاعلية

تكوين البيئة

# متغيرات التكوين الرئيسية
DATABASE_URL="postgresql://postgres:password@localhost:5432/arxiv_papers"
OPENSEARCH_URL="http://localhost:9200"
JINA_API_KEY="your-jina-api-key"  # للتضمينات
LANGFUSE_SECRET_KEY="your-key"     # للمراقبة

الأسبوع الثاني: خط أنابيب جمع البيانات

يركز الأسبوع الثاني على بناء خط أنابيب قوي لجمع البيانات يجلب ويعالج ويخزن الأوراق البحثية تلقائياً.

تكامل ArXiv API

يستخدم النظام ArXiv API لجلب الأوراق بناءً على الفئات واستعلامات البحث:

async def fetch_papers_from_arxiv(
    query: str = "cat:cs.AI",
    max_results: int = 100,
    start_date: Optional[datetime] = None
) -> List[ArxivPaper]:
    """
    جلب الأوراق من ArXiv API مع البيانات الوصفية الشاملة
    """
    base_url = "http://export.arxiv.org/api/query"
    params = {
        "search_query": query,
        "start": 0,
        "max_results": max_results,
        "sortBy": "submittedDate",
        "sortOrder": "descending"
    }

معالجة PDF باستخدام Docling

يستخدم المشروع مكتبة Docling من IBM لمعالجة PDF المتطورة التي تحافظ على بنية المستند:

from docling.document_converter import DocumentConverter

def process_pdf_with_docling(pdf_path: str) -> ProcessedDocument:
    """
    استخراج المحتوى المنظم من الأوراق البحثية
    """
    converter = DocumentConverter()
    result = converter.convert(pdf_path)
    
    return ProcessedDocument(
        title=result.document.title,
        abstract=result.document.abstract,
        sections=result.document.sections,
        figures=result.document.figures,
        tables=result.document.tables
    )

تصميم مخطط قاعدة البيانات

مخطط PostgreSQL محسن للبيانات الوصفية للأوراق البحثية:

-- جدول الأوراق الأساسي
CREATE TABLE papers (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    arxiv_id VARCHAR(50) UNIQUE NOT NULL,
    title TEXT NOT NULL,
    abstract TEXT,
    authors TEXT[],
    categories TEXT[],
    published_date TIMESTAMP,
    pdf_url TEXT,
    content_extracted BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMP DEFAULT NOW()
);

-- جدول علاقة المؤلفين
CREATE TABLE paper_authors (
    paper_id UUID REFERENCES papers(id),
    author_name TEXT,
    author_order INTEGER,
    PRIMARY KEY (paper_id, author_name)
);

الأسبوع الثالث: البحث الكلمي باستخدام BM25

ينفذ الأسبوع الثالث البحث الكلمي المتطور باستخدام خوارزمية BM25 من OpenSearch، مما يوفر الأساس لنظام البحث المختلط.

تكوين فهرس OpenSearch

# إعدادات الفهرس المحسنة لـ BM25
index_settings = {
    "settings": {
        "number_of_shards": 1,
        "number_of_replicas": 0,
        "analysis": {
            "analyzer": {
                "scientific_analyzer": {
                    "type": "custom",
                    "tokenizer": "standard",
                    "filter": [
                        "lowercase",
                        "scientific_stemmer",
                        "stop_words_filter"
                    ]
                }
            },
            "filter": {
                "scientific_stemmer": {
                    "type": "stemmer",
                    "language": "english"
                },
                "stop_words_filter": {
                    "type": "stop",
                    "stopwords": "_english_"
                }
            }
        }
    },
    "mappings": {
        "properties": {
            "title": {
                "type": "text",
                "analyzer": "scientific_analyzer",
                "boost": 3.0
            },
            "abstract": {
                "type": "text",
                "analyzer": "scientific_analyzer",
                "boost": 2.0
            },
            "content": {
                "type": "text",
                "analyzer": "scientific_analyzer"
            }
        }
    }
}

بناء استعلام BM25 المتقدم

def build_bm25_query(
    query: str,
    filters: Optional[Dict] = None,
    boost_recent: bool = True
) -> Dict:
    """
    بناء استعلامات BM25 متطورة مع تعزيز الحقول
    """
    base_query = {
        "query": {
            "bool": {
                "must": [
                    {
                        "multi_match": {
                            "query": query,
                            "fields": [
                                "title^3",      # تعزيز مطابقات العنوان
                                "abstract^2",   # تعزيز مطابقات الملخص
                                "content",      # مطابقة المحتوى القياسية
                                "authors"       # تضمين مطابقة المؤلف
                            ],
                            "type": "best_fields",
                            "fuzziness": "AUTO"
                        }
                    }
                ]
            }
        }
    }
    
    # إضافة تعزيز الحداثة
    if boost_recent:
        base_query["query"]["bool"]["should"] = [
            {
                "function_score": {
                    "functions": [
                        {
                            "gauss": {
                                "published_date": {
                                    "origin": "now",
                                    "scale": "30d",
                                    "decay": 0.5
                                }
                            }
                        }
                    ]
                }
            }
        ]
    
    return base_query

الأسبوع الرابع: البحث المختلط باستخدام التضمينات

يقدم الأسبوع الرابع التضمينات الشعاعية والبحث المختلط، حيث يجمع بين التشابه الدلالي ومطابقة الكلمات المفتاحية لأداء استرجاع فائق.

استراتيجية تقسيم النص

ينفذ النظام تقسيماً واعياً بالأقسام يحافظ على بنية المستند:

class SectionAwareChunker:
    def __init__(self, max_chunk_size: int = 512, overlap: int = 50):
        self.max_chunk_size = max_chunk_size
        self.overlap = overlap
    
    def chunk_document(self, document: ProcessedDocument) -> List[DocumentChunk]:
        """
        إنشاء قطع تحترم حدود الأقسام
        """
        chunks = []
        
        # معالجة كل قسم بشكل منفصل
        for section in document.sections:
            section_chunks = self._chunk_section(
                section.content,
                section.title,
                section.level
            )
            chunks.extend(section_chunks)
        
        return chunks
    
    def _chunk_section(
        self, 
        content: str, 
        section_title: str, 
        level: int
    ) -> List[DocumentChunk]:
        """
        تقسيم محتوى القسم مع الحفاظ على السياق
        """
        sentences = self._split_sentences(content)
        chunks = []
        current_chunk = []
        current_length = 0
        
        for sentence in sentences:
            sentence_length = len(sentence.split())
            
            if (current_length + sentence_length > self.max_chunk_size 
                and current_chunk):
                # إنشاء قطعة مع سياق القسم
                chunk_text = f"{section_title}\n\n" + " ".join(current_chunk)
                chunks.append(DocumentChunk(
                    text=chunk_text,
                    section=section_title,
                    level=level,
                    word_count=current_length
                ))
                
                # بدء قطعة جديدة مع التداخل
                overlap_sentences = current_chunk[-self.overlap:]
                current_chunk = overlap_sentences + [sentence]
                current_length = sum(len(s.split()) for s in current_chunk)
            else:
                current_chunk.append(sentence)
                current_length += sentence_length
        
        # إضافة القطعة الأخيرة
        if current_chunk:
            chunk_text = f"{section_title}\n\n" + " ".join(current_chunk)
            chunks.append(DocumentChunk(
                text=chunk_text,
                section=section_title,
                level=level,
                word_count=current_length
            ))
        
        return chunks

توليد التضمينات الشعاعية

يستخدم النظام خدمة التضمين من Jina AI للحصول على تمثيلات شعاعية عالية الجودة:

class JinaEmbeddingService:
    def __init__(self, api_key: str, model: str = "jina-embeddings-v2-base-en"):
        self.api_key = api_key
        self.model = model
        self.client = httpx.AsyncClient()
    
    async def generate_embeddings(
        self, 
        texts: List[str], 
        batch_size: int = 32
    ) -> List[List[float]]:
        """
        توليد التضمينات في دفعات للكفاءة
        """
        all_embeddings = []
        
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            batch_embeddings = await self._embed_batch(batch)
            all_embeddings.extend(batch_embeddings)
        
        return all_embeddings
    
    async def _embed_batch(self, texts: List[str]) -> List[List[float]]:
        """
        معالجة دفعة واحدة من النصوص
        """
        response = await self.client.post(
            "https://api.jina.ai/v1/embeddings",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": self.model,
                "input": texts,
                "encoding_format": "float"
            }
        )
        
        result = response.json()
        return [item["embedding"] for item in result["data"]]

تنفيذ البحث المختلط

يجمع البحث المختلط بين BM25 والتشابه الشعاعي مع دمج النقاط المتطور:

class HybridSearchEngine:
    def __init__(
        self, 
        opensearch_client, 
        embedding_service,
        bm25_weight: float = 0.6,
        vector_weight: float = 0.4
    ):
        self.opensearch = opensearch_client
        self.embeddings = embedding_service
        self.bm25_weight = bm25_weight
        self.vector_weight = vector_weight
    
    async def search(
        self, 
        query: str, 
        k: int = 10,
        filters: Optional[Dict] = None
    ) -> List[SearchResult]:
        """
        تنفيذ البحث المختلط مع دمج النقاط
        """
        # توليد تضمين الاستعلام
        query_embedding = await self.embeddings.generate_embeddings([query])
        query_vector = query_embedding[0]
        
        # بناء الاستعلام المختلط
        hybrid_query = {
            "size": k * 2,  # استرجاع المزيد لإعادة الترتيب
            "query": {
                "bool": {
                    "should": [
                        # مكون BM25
                        {
                            "multi_match": {
                                "query": query,
                                "fields": ["title^3", "abstract^2", "content"],
                                "type": "best_fields"
                            }
                        },
                        # مكون التشابه الشعاعي
                        {
                            "knn": {
                                "embedding": {
                                    "vector": query_vector,
                                    "k": k
                                }
                            }
                        }
                    ]
                }
            }
        }
        
        # إضافة المرشحات إذا تم توفيرها
        if filters:
            hybrid_query["query"]["bool"]["filter"] = self._build_filters(filters)
        
        # تنفيذ البحث
        response = await self.opensearch.search(
            index="papers_hybrid",
            body=hybrid_query
        )
        
        # معالجة النتائج وإعادة الترتيب
        results = self._process_results(response["hits"]["hits"])
        reranked_results = self._rerank_results(results, query, k)
        
        return reranked_results
    
    def _rerank_results(
        self, 
        results: List[SearchResult], 
        query: str, 
        k: int
    ) -> List[SearchResult]:
        """
        تطبيق إعادة ترتيب متطورة بناءً على إشارات متعددة
        """
        # حساب النقاط المعيارية
        max_bm25 = max(r.bm25_score for r in results) if results else 1.0
        max_vector = max(r.vector_score for r in results) if results else 1.0
        
        for result in results:
            # تعيير النقاط الفردية
            norm_bm25 = result.bm25_score / max_bm25
            norm_vector = result.vector_score / max_vector
            
            # حساب النقاط المختلطة
            result.hybrid_score = (
                self.bm25_weight * norm_bm25 + 
                self.vector_weight * norm_vector
            )
            
            # تطبيق إشارات ترتيب إضافية
            result.hybrid_score *= self._calculate_quality_multiplier(result)
        
        # ترتيب حسب النقاط المختلطة وإرجاع أفضل k
        return sorted(results, key=lambda x: x.hybrid_score, reverse=True)[:k]

الأسبوع الخامس: نظام RAG الكامل

يدمج الأسبوع الخامس كل شيء في نظام RAG كامل مع هندسة المحفزات المتطورة وتوليد الاستجابات.

تكامل النموذج اللغوي مع Ollama

يستخدم النظام Ollama للاستنتاج المحلي للنماذج اللغوية، مما يوفر الخصوصية والتحكم:

class OllamaRAGService:
    def __init__(
        self, 
        base_url: str = "http://localhost:11434",
        model: str = "llama3.1:8b"
    ):
        self.base_url = base_url
        self.model = model
        self.client = httpx.AsyncClient()
    
    async def generate_answer(
        self, 
        query: str, 
        context_chunks: List[DocumentChunk],
        max_context_length: int = 4000
    ) -> RAGResponse:
        """
        توليد إجابات سياقية باستخدام المستندات المسترجعة
        """
        # إعداد السياق مع القطع الذكي
        context = self._prepare_context(context_chunks, max_context_length)
        
        # بناء محفز RAG
        prompt = self._build_rag_prompt(query, context)
        
        # توليد الاستجابة
        response = await self._call_ollama(prompt)
        
        # المعالجة اللاحقة والتحقق
        processed_response = self._post_process_response(
            response, 
            query, 
            context_chunks
        )
        
        return processed_response
    
    def _build_rag_prompt(self, query: str, context: str) -> str:
        """
        بناء محفزات RAG متطورة مع تعريف الدور
        """
        return f"""أنت مساعد بحثي خبير متخصص في الذكاء الاصطناعي وأدبيات علوم الحاسوب. مهمتك هي تقديم إجابات دقيقة وشاملة بناءً على الأوراق البحثية المقدمة.

**السياق من الأوراق البحثية:**
{context}

**سؤال المستخدم:**
{query}

**التعليمات:**
1. اعتمد إجابتك بشكل أساسي على السياق المقدم
2. إذا لم يحتو السياق على معلومات كافية، اذكر ذلك بوضوح
3. قم بتضمين مراجع محددة للأوراق عند تقديم الادعاءات
4. قدم التفاصيل التقنية عند الاقتضاء
5. عند مناقشة المنهجيات، اشرحها بوضوح
6. سلط الضوء على أي قيود أو تحذيرات مذكورة في الأوراق

**الإجابة:**"""

    def _prepare_context(
        self, 
        chunks: List[DocumentChunk], 
        max_length: int
    ) -> str:
        """
        اختيار وتنسيق قطع السياق بذكاء
        """
        # ترتيب القطع حسب نقاط الصلة
        sorted_chunks = sorted(chunks, key=lambda x: x.score, reverse=True)
        
        context_parts = []
        current_length = 0
        
        for chunk in sorted_chunks:
            chunk_text = f"[{chunk.paper_title}]\n{chunk.text}\n---\n"
            chunk_length = len(chunk_text.split())
            
            if current_length + chunk_length > max_length:
                break
                
            context_parts.append(chunk_text)
            current_length += chunk_length
        
        return "\n".join(context_parts)

هندسة المحفزات المتقدمة

يتضمن النظام محفزات متخصصة لأنواع مختلفة من الاستعلامات:

class PromptTemplates:
    COMPARATIVE_ANALYSIS = """بناءً على الأوراق البحثية المقدمة، قارن وقابل بين النُهج/الطرق/المفاهيم التالية: {concepts}

يرجى هيكلة استجابتك كما يلي:
1. **نظرة عامة**: مقدمة موجزة لكل مفهوم
2. **أوجه التشابه الرئيسية**: ما هي القواسم المشتركة بين هذه النُهج؟
3. **الاختلافات الرئيسية**: كيف تختلف في المنهجية أو الافتراضات أو النتائج؟
4. **مقارنة الأداء**: إذا كانت متاحة، قارن فعاليتها
5. **حالات الاستخدام**: متى ستختار واحداً على الآخر؟
6. **القيود**: ما هي قيود كل نهج؟

الأوراق المرجعية: {paper_titles}"""

    METHODOLOGY_EXPLANATION = """اشرح المنهجية الموصوفة في الأوراق البحثية لـ: {topic}

يرجى تقديم:
1. **تعريف المشكلة**: ما هي المشكلة التي يتم معالجتها؟
2. **نظرة عامة على النهج**: وصف المنهجية عالي المستوى
3. **التفاصيل التقنية**: شرح خطوة بخطوة للطريقة
4. **اعتبارات التنفيذ**: الجوانب العملية للتنفيذ
5. **طرق التقييم**: كيف تم التحقق من النهج؟
6. **ملخص النتائج**: النتائج الرئيسية ومقاييس الأداء
7. **القيود والعمل المستقبلي**: القيود المعترف بها والخطوات التالية

بناءً على: {paper_titles}"""

    TREND_ANALYSIS = """تحليل الاتجاهات والتطورات في: {research_area}

بناءً على الأوراق المقدمة، ناقش:
1. **السياق التاريخي**: كيف تطور هذا المجال؟
2. **الحالة الحالية**: ما هي النُهج المهيمنة اليوم؟
3. **الأنماط الناشئة**: ما هي الاتجاهات الجديدة المرئية؟
4. **الابتكارات الرئيسية**: ما هي المساهمات الاختراقية المذكورة؟
5. **التحديات المفتوحة**: ما هي المشاكل التي لا تزال بلا حل؟
6. **الاتجاهات المستقبلية**: ما هي اتجاهات البحث المقترحة؟

الأوراق المحللة: {paper_titles}"""

الأسبوع السادس: مراقبة الإنتاج والتخزين المؤقت

يركز الأسبوع السادس على جاهزية الإنتاج مع المراقبة الشاملة واستراتيجيات التخزين المؤقت الذكية.

تكامل Langfuse للإمكانية الملاحظة

class RAGMonitoringService:
    def __init__(self, langfuse_client):
        self.langfuse = langfuse_client
    
    async def trace_rag_request(
        self, 
        query: str, 
        search_results: List[SearchResult],
        llm_response: RAGResponse,
        execution_time: float
    ) -> None:
        """
        تتبع شامل لخط أنابيب RAG
        """
        trace = self.langfuse.trace(
            name="rag_query",
            input={"query": query},
            output={"answer": llm_response.answer},
            metadata={
                "execution_time_ms": execution_time * 1000,
                "num_results": len(search_results),
                "model_used": llm_response.model,
                "context_length": llm_response.context_length
            }
        )
        
        # تتبع أداء البحث
        search_span = trace.span(
            name="search_retrieval",
            input={"query": query},
            output={"num_results": len(search_results)},
            metadata={
                "search_type": "hybrid",
                "bm25_weight": 0.6,
                "vector_weight": 0.4
            }
        )
        
        # تتبع توليد النموذج اللغوي
        generation_span = trace.span(
            name="llm_generation",
            input={"prompt_length": len(llm_response.prompt)},
            output={"response_length": len(llm_response.answer)},
            metadata={
                "model": llm_response.model,
                "temperature": llm_response.temperature,
                "tokens_used": llm_response.token_count
            }
        )
        
        # تتبع مقاييس الجودة
        self._track_quality_metrics(trace, query, llm_response)
    
    def _track_quality_metrics(
        self, 
        trace, 
        query: str, 
        response: RAGResponse
    ) -> None:
        """
        تتبع مؤشرات جودة الاستجابة
        """
        metrics = {
            "has_citations": len(response.citations) > 0,
            "answer_length": len(response.answer.split()),
            "confidence_score": response.confidence_score,
            "context_utilization": response.context_utilization_score
        }
        
        trace.score(
            name="response_quality",
            value=self._calculate_quality_score(metrics)
        )

استراتيجية التخزين المؤقت الذكية

class RAGCacheService:
    def __init__(self, redis_client, ttl: int = 3600):
        self.redis = redis_client
        self.ttl = ttl
    
    async def get_cached_response(
        self, 
        query: str, 
        filters: Optional[Dict] = None
    ) -> Optional[CachedRAGResponse]:
        """
        استرجاع الاستجابات المخزنة مؤقتاً مع مطابقة التشابه الدلالي
        """
        # توليد مفتاح التخزين المؤقت
        cache_key = self._generate_cache_key(query, filters)
        
        # التحقق من المطابقة الدقيقة أولاً
        cached = await self.redis.get(cache_key)
        if cached:
            return CachedRAGResponse.parse_raw(cached)
        
        # التحقق من الاستعلامات المشابهة باستخدام تشابه التضمين
        similar_response = await self._find_similar_cached_query(query)
        if similar_response and similar_response.similarity_score > 0.85:
            return similar_response
        
        return None
    
    async def cache_response(
        self, 
        query: str, 
        response: RAGResponse,
        filters: Optional[Dict] = None
    ) -> None:
        """
        تخزين الاستجابات مؤقتاً مع البيانات الوصفية للاسترجاع الذكي
        """
        cache_key = self._generate_cache_key(query, filters)
        
        cached_response = CachedRAGResponse(
            query=query,
            response=response,
            filters=filters,
            cached_at=datetime.utcnow(),
            access_count=1,
            quality_score=response.quality_score
        )
        
        # التخزين مع TTL بناءً على نقاط الجودة
        ttl = self._calculate_adaptive_ttl(response.quality_score)
        
        await self.redis.setex(
            cache_key,
            ttl,
            cached_response.json()
        )
        
        # تحديث فهرس تضمين الاستعلام للبحث بالتشابه
        await self._index_query_embedding(query, cache_key)
    
    def _calculate_adaptive_ttl(self, quality_score: float) -> int:
        """
        تعديل TTL للتخزين المؤقت بناءً على جودة الاستجابة
        """
        base_ttl = self.ttl
        quality_multiplier = min(2.0, max(0.5, quality_score * 2))
        return int(base_ttl * quality_multiplier)

🔧 نقاط النهاية للواجهة البرمجية والاستخدام

يوفر النظام واجهات برمجة تطبيقات RESTful شاملة:

نقاط النهاية الأساسية

# فحص الحالة
GET /health

# إدارة الأوراق
GET /api/v1/papers              # قائمة الأوراق مع الترقيم
GET /api/v1/papers/{id}         # الحصول على ورقة محددة
POST /api/v1/papers/ingest      # تفعيل الجمع

# نقاط نهاية البحث
POST /api/v1/search             # البحث الكلمي BM25
POST /api/v1/hybrid-search      # البحث المختلط (BM25 + شعاعي)
POST /api/v1/ask                # سؤال وجواب RAG

# نقاط نهاية التحليلات
GET /api/v1/analytics/search    # مقاييس أداء البحث
GET /api/v1/analytics/rag       # مقاييس نظام RAG

مثال على استخدام واجهة برمجة التطبيقات

import httpx

async def query_rag_system():
    async with httpx.AsyncClient() as client:
        # تنفيذ البحث المختلط
        search_response = await client.post(
            "http://localhost:8000/api/v1/hybrid-search",
            json={
                "query": "آليات الانتباه في المحولات",
                "k": 5,
                "filters": {
                    "categories": ["cs.AI", "cs.LG"],
                    "date_range": {
                        "start": "2023-01-01",
                        "end": "2024-12-31"
                    }
                }
            }
        )
        
        # طرح سؤال RAG
        rag_response = await client.post(
            "http://localhost:8000/api/v1/ask",
            json={
                "question": "كيف تعمل آليات الانتباه في المحولات وما هي قيودها؟",
                "context_k": 10,
                "model": "llama3.1:8b",
                "temperature": 0.7
            }
        )
        
        return rag_response.json()

🎛️ واجهة الويب Gradio

يتضمن النظام واجهة ويب متطورة مبنية باستخدام Gradio:

import gradio as gr

def create_rag_interface():
    with gr.Blocks(title="ArXiv Paper Curator") as interface:
        gr.Markdown("# 🤖 نظام ArXiv Paper Curator RAG")
        
        with gr.Tab("🔍 البحث في الأوراق"):
            with gr.Row():
                search_query = gr.Textbox(
                    label="استعلام البحث",
                    placeholder="أدخل مصطلحات البحث..."
                )
                search_type = gr.Radio(
                    ["كلمي (BM25)", "دلالي (شعاعي)", "مختلط"],
                    value="مختلط",
                    label="نوع البحث"
                )
            
            search_button = gr.Button("بحث", variant="primary")
            search_results = gr.JSON(label="نتائج البحث")
        
        with gr.Tab("💬 طرح الأسئلة"):
            with gr.Row():
                with gr.Column(scale=2):
                    question = gr.Textbox(
                        label="السؤال",
                        placeholder="اسأل عن الأوراق البحثية...",
                        lines=3
                    )
                    
                    with gr.Row():
                        model_choice = gr.Dropdown(
                            ["llama3.1:8b", "mistral:7b", "codellama:13b"],
                            value="llama3.1:8b",
                            label="النموذج"
                        )
                        temperature = gr.Slider(
                            0.0, 1.0, 0.7,
                            label="درجة الحرارة"
                        )
                
                with gr.Column(scale=1):
                    context_k = gr.Slider(
                        1, 20, 10,
                        label="أوراق السياق"
                    )
                    include_citations = gr.Checkbox(
                        True,
                        label="تضمين الاستشهادات"
                    )
            
            ask_button = gr.Button("طرح السؤال", variant="primary")
            
            with gr.Row():
                answer = gr.Textbox(
                    label="الإجابة",
                    lines=10,
                    max_lines=20
                )
                context_papers = gr.JSON(
                    label="الأوراق المصدر"
                )
        
        with gr.Tab("📊 التحليلات"):
            refresh_button = gr.Button("تحديث التحليلات")
            
            with gr.Row():
                total_papers = gr.Number(label="إجمالي الأوراق")
                indexed_papers = gr.Number(label="الأوراق المفهرسة")
                cache_hit_rate = gr.Number(label="معدل نجاح التخزين المؤقت (%)")
            
            performance_chart = gr.Plot(label="أداء الاستعلامات")
    
    return interface

# تشغيل الواجهة
if __name__ == "__main__":
    interface = create_rag_interface()
    interface.launch(server_name="0.0.0.0", server_port=7860)

🚀 النشر والتوسع

إعداد إنتاج Docker Compose

version: '3.8'

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://postgres:${POSTGRES_PASSWORD}@db:5432/arxiv_papers
      - OPENSEARCH_URL=http://opensearch:9200
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - opensearch
      - redis
    volumes:
      - ./data:/app/data
  
  db:
    image: postgres:15
    environment:
      - POSTGRES_DB=arxiv_papers
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
    volumes:
      - postgres_data:/var/lib/postgresql/data
  
  opensearch:
    image: opensearchproject/opensearch:2.8.0
    environment:
      - discovery.type=single-node
      - "OPENSEARCH_JAVA_OPTS=-Xms2g -Xmx2g"
    volumes:
      - opensearch_data:/usr/share/opensearch/data
  
  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data
  
  gradio:
    build: .
    command: python gradio_launcher.py
    ports:
      - "7860:7860"
    depends_on:
      - app

volumes:
  postgres_data:
  opensearch_data:
  redis_data:

نصائح لتحسين الأداء

1. تحسين قاعدة البيانات:
   • إنشاء فهارس على الحقول المستعلمة بكثرة
   • استخدام تجميع الاتصالات
   • تنفيذ نسخ القراءة للتوسع
2. تحسين البحث:
   • تحسين إعدادات مجموعة OpenSearch
   • استخدام قوالب الفهرس للتعيين المتسق
   • تنفيذ تخزين نتائج البحث مؤقتاً
3. تحسين التضمين:
   • معالجة توليد التضمين بالدفعات
   • تخزين التضمينات للمحتوى المعاد استخدامه مؤقتاً
   • استخدام التضمينات المكممة لكفاءة الذاكرة
4. تحسين النموذج اللغوي:
   • تنفيذ إحماء النموذج
   • استخدام تسريع GPU عند التوفر
   • تحسين قوالب المحفزات للكفاءة

🔍 المراقبة وإمكانية الملاحظة

المقاييس الرئيسية للتتبع

# مقاييس الأداء
search_latency = histogram("search_latency_seconds")
rag_latency = histogram("rag_latency_seconds")
cache_hit_rate = gauge("cache_hit_rate")

# مقاييس الجودة
answer_relevance = histogram("answer_relevance_score")
context_utilization = histogram("context_utilization_score")
user_satisfaction = histogram("user_satisfaction_score")

# مقاييس النظام
active_connections = gauge("active_database_connections")
opensearch_cluster_health = gauge("opensearch_cluster_health")
embedding_queue_size = gauge("embedding_queue_size")

تكوين لوحة معلومات Langfuse

يتكامل النظام مع Langfuse للحصول على إمكانية ملاحظة شاملة:

• تحليل التتبع: تتبع طلبات كاملة من الاستعلام إلى الاستجابة
• مراقبة الأداء: تحليل زمن الاستجابة حسب المكون
• تتبع الجودة: نقاط جودة الاستجابة وتقييم المستخدمين
• تتبع التكلفة: استخدام الرموز وتكاليف واجهة برمجة التطبيقات
• مراقبة الأخطاء: معدلات الفشل وتصنيف الأخطاء

🎯 الميزات المتقدمة والتوسيعات

1. الدعم متعدد الوسائط

توسيع النظام للتعامل مع الأشكال والجداول من الأوراق:

class MultiModalProcessor:
    async def process_figures(self, paper_id: str) -> List[Figure]:
        """استخراج ومعالجة الأشكال من الأوراق البحثية"""
        
    async def process_tables(self, paper_id: str) -> List[Table]:
        """استخراج ومعالجة الجداول مع الحفاظ على البنية"""

2. تحليل شبكة الاستشهادات

بناء رسوم بيانية للاستشهادات لاكتشاف الأوراق المحسن:

class CitationNetworkService:
    async def build_citation_graph(self) -> NetworkGraph:
        """بناء شبكات الاستشهاد لتوصية الأوراق"""
        
    async def find_influential_papers(self, topic: str) -> List[Paper]:
        """تحديد الأوراق عالية الاستشهاد في مجالات محددة"""

3. محرك التخصيص

تنفيذ تعلم تفضيلات المستخدم:

class PersonalizationService:
    async def learn_user_preferences(self, user_id: str, interactions: List[Interaction]):
        """تعلم تفضيلات المستخدم من أنماط البحث والقراءة"""
        
    async def personalize_search_results(self, user_id: str, results: List[SearchResult]) -> List[SearchResult]:
        """إعادة ترتيب النتائج بناءً على تفضيلات المستخدم"""

🏆 أفضل الممارسات والدروس المستفادة

1. إدارة جودة البيانات

• تنفيذ معالجة PDF قوية مع استراتيجيات البديل
• التحقق من المحتوى المستخرج قبل الفهرسة
• مراقبة معدلات نجاح الاستخراج وتحديد الأوراق الإشكالية
• التعامل مع المحتوى متعدد اللغات بشكل مناسب

2. تحسين جودة البحث

• ضبط أوزان البحث المختلط بناءً على أنواع الاستعلامات
• تنفيذ توسيع الاستعلام لتذكر أفضل
• استخدام نماذج إعادة الترتيب لدقة محسنة
• اختبار A/B لاستراتيجيات استرجاع مختلفة

3. تصميم نظام RAG

• تصميم محفزات واعية بالسياق لأنواع استعلام مختلفة
• تنفيذ التحقق من الاستجابة لالتقاط الهلوسة
• استخدام تنسيقات الإخراج المنظمة للاستجابات المتسقة
• مراقبة وتحسين جودة الإجابة بشكل مستمر

4. جاهزية الإنتاج

• تنفيذ مراقبة شاملة من اليوم الأول
• التصميم للتوسع الأفقي مع خدمات عديمة الحالة
• استخدام التدهور الرشيق عند فشل المكونات
• تنفيذ قواطع الدائرة للتبعيات الخارجية

🔮 التحسينات المستقبلية

تتضمن خارطة طريق ArXiv Paper Curator عدة تطويرات مثيرة:

1. تقنيات RAG المتقدمة

• الضبط الدقيق المعزز بالاسترجاع: دمج الاسترجاع مع تدريب النماذج المتخصصة
• الاستدلال متعدد الخطوات: تنفيذ سلسلة الفكر للاستعلامات المعقدة
• التوليف عبر الأوراق: توليد رؤى من خلال ربط أوراق متعددة

2. تجربة المستخدم المحسنة

• التصورات التفاعلية: رسوم بيانية لعلاقات الأوراق وتحليل الاتجاهات
• ميزات التعاون: قوائم القراءة المشتركة والتعليقات التوضيحية
• تطبيق الهاتف المحمول: تطبيقات iOS/Android الأصلية للوصول أثناء التنقل

3. ميزات المساعد البحثي

• مراجعات الأدبيات التلقائية: توليد أوراق مسح شاملة
• تحديد فجوات البحث: تسليط الضوء على مناطق البحث غير المستكشفة
• مقارنة المنهجيات: تحليل جنباً إلى جنب لنُهج مختلفة

📚 موارد إضافية

موارد التعلم

• دفاتر أسبوعية: دفاتر Jupyter مفصلة لكل مرحلة تنفيذ
• شروحات الفيديو: أدلة التنفيذ خطوة بخطوة
• منتدى المجتمع: مجتمع نقاش ودعم نشط

الوثائق

• مرجع واجهة برمجة التطبيقات: مواصفة OpenAPI كاملة
• دليل المعمارية: وثائق تصميم النظام المفصلة
• دليل النشر: أفضل ممارسات نشر الإنتاج

المجتمع والدعم

• نقاشات GitHub: أسئلة وأجوبة المجتمع وطلبات الميزات
• خادم Discord: دردشة وتعاون في الوقت الفعلي
• ساعات مكتبية: جلسات أسبوعية مع فريق التطوير

🎉 الخلاصة

يمثل ArXiv Paper Curator نهجاً شاملاً لبناء أنظمة RAG جاهزة للإنتاج. من خلال اتباع هذا الدليل، تعلمت:

• تصميم معماريات RAG قابلة للتوسع تتعامل مع تعقيد العالم الحقيقي
• تنفيذ أنظمة البحث المختلط التي تجمع استراتيجيات استرجاع متعددة
• بناء خطوط أنابيب بيانات قوية للجمع المستمر للمحتوى
• نشر حلول المراقبة لإمكانية ملاحظة الإنتاج
• إنشاء واجهات بديهية لتفاعل المستخدم النهائي

المهارات والأنماط التي تعلمتها هنا قابلة للتطبيق مباشرة على أي مجال تحتاج فيه لبناء أنظمة ذكية يمكنها الاستدلال على مجموعات مستندات كبيرة. سواء كنت تعمل مع وثائق قانونية أو بحوث طبية أو وثائق تقنية، فالمبادئ تبقى نفسها.

الخطوات التالية

1. التجريب مع مجالات مختلفة: تطبيق النظام على حالة الاستخدام المحددة الخاصة بك
2. المساهمة في المشروع: مشاركة التحسينات والتوسيعات مع المجتمع
3. التوسع للإنتاج: نشر المثيل الخاص بك وجمع تعليقات المستخدمين الحقيقيين
4. البقاء محدثاً: متابعة المشروع للميزات والتحسينات الجديدة

تذكر أن بناء أنظمة الذكاء الاصطناعي للإنتاج عملية تكرارية. ابدأ بالأساسيات، قس كل شيء، وحسن باستمرار بناءً على أنماط الاستخدام الحقيقية.

---

مستعد لبناء نظام RAG الخاص بك؟ استنسخ المستودع وابدأ بالأسبوع الأول: ArXiv Paper Curator على GitHub

للحصول على آخر التحديثات ونقاشات المجتمع، انضم إلى خادم Discord الخاص بنا وتابع @jamwithai على Twitter.