قواعد استرجاع RAG

يوفر NeMo Guardrails دعماً مدمجاً للتوليد المعزز بالاسترجاع (RAG) مع حواجز تضمن أن الردود مبنية على قاعدة معرفتك مع منع الهلوسات.

إعداد قاعدة المعرفة

التكوين

# config/config.yml
models:
  - type: main
    engine: openai
    model: gpt-4o

  - type: embeddings
    engine: openai
    model: text-embedding-3-small

knowledge_base:
  - type: file
    path: ./kb/
    formats: [md, txt, pdf]

  - type: vector_db
    provider: chroma
    collection: company_docs
    connection_string: ./chroma_db

rails:
  retrieval:
    flows:
      - retrieve relevant chunks
      - check retrieval relevance
      - ground response in context

فهرسة المستندات

from nemoguardrails import LLMRails, RailsConfig
from nemoguardrails.kb.index import KnowledgeBaseIndex

config = RailsConfig.from_path("./config")

# فهرسة المستندات
index = KnowledgeBaseIndex(config)

# إضافة المستندات
index.add_documents([
    {
        "content": "منتجنا يدعم Python 3.9+...",
        "metadata": {"source": "docs/requirements.md", "category": "technical"}
    },
    {
        "content": "التسعير يبدأ من 99$/شهر...",
        "metadata": {"source": "docs/pricing.md", "category": "business"}
    }
])

# بناء الفهرس
index.build()

قواعد الاسترجاع

تدفق الاسترجاع الأساسي

# rails/retrieval.co
define flow retrieve relevant chunks
  """استرجاع السياق قبل توليد الاستجابة."""
  $chunks = execute search_knowledge_base(
    query=$user_message,
    top_k=5
  )

  if len($chunks) == 0
    bot inform no relevant information
    stop

  # تخزين في السياق للتوليد
  $context.retrieved_chunks = $chunks

define bot inform no relevant information
  "ليس لدي معلومات محددة حول ذلك في قاعدة معرفتي. دعني أقدم توجيهاً عاماً بدلاً من ذلك."

التحقق من الصلة

# actions/retrieval.py
from nemoguardrails.actions import action
from typing import List, Dict
import numpy as np

@action(name="search_knowledge_base")
async def search_knowledge_base(
    query: str,
    top_k: int = 5,
    relevance_threshold: float = 0.7
) -> List[Dict]:
    """
    البحث في قاعدة المعرفة مع تصفية الصلة.
    """
    # الحصول على التضمينات (باستخدام النموذج المكون)
    query_embedding = await get_embedding(query)

    # البحث في قاعدة البيانات المتجهة
    results = await vector_db.search(
        embedding=query_embedding,
        top_k=top_k * 2  # استرجاع المزيد للتصفية
    )

    # التصفية حسب الصلة
    relevant = [
        r for r in results
        if r["score"] >= relevance_threshold
    ][:top_k]

    return relevant


@action(name="check_retrieval_quality")
async def check_retrieval_quality(
    query: str,
    chunks: List[Dict]
) -> Dict:
    """
    تقييم إذا كانت القطع المسترجعة كافية
    للإجابة على الاستعلام.
    """
    if not chunks:
        return {"sufficient": False, "reason": "no_chunks"}

    # التحقق من التغطية
    avg_relevance = np.mean([c["score"] for c in chunks])

    if avg_relevance < 0.6:
        return {
            "sufficient": False,
            "reason": "low_relevance",
            "avg_score": avg_relevance
        }

    # التحقق من التنوع (تجنب القطع المتكررة)
    unique_sources = len(set(c["metadata"]["source"] for c in chunks))

    return {
        "sufficient": True,
        "avg_relevance": avg_relevance,
        "source_diversity": unique_sources,
        "chunk_count": len(chunks)
    }

التحقق من التأسيس

# rails/grounding.co
define flow ground response in context
  """التأكد من أن الاستجابة مؤسسة على السياق المسترجع."""
  # أولاً توليد الاستجابة مع السياق
  $response = execute generate_grounded_response(
    query=$user_message,
    context=$context.retrieved_chunks
  )

  # التحقق من التأسيس
  $grounding_check = execute verify_grounding(
    response=$response,
    context=$context.retrieved_chunks
  )

  if not $grounding_check.is_grounded
    # إعادة التوليد مع تأسيس أكثر صرامة
    $response = execute regenerate_with_citations(
      query=$user_message,
      context=$context.retrieved_chunks,
      ungrounded_claims=$grounding_check.ungrounded_claims
    )

  bot say $response

إجراء التحقق من التأسيس

# actions/grounding.py
from nemoguardrails.actions import action
from typing import List, Dict

@action(name="verify_grounding")
async def verify_grounding(
    response: str,
    context: List[Dict]
) -> Dict:
    """
    التحقق من أن ادعاءات الاستجابة مؤسسة على السياق.
    يستخدم نموذج NLI (استدلال اللغة الطبيعية).
    """
    from transformers import pipeline

    nli = pipeline("text-classification", model="roberta-large-mnli")

    # استخراج الادعاءات من الاستجابة
    claims = extract_claims(response)

    # دمج السياق
    context_text = " ".join([c["content"] for c in context])

    ungrounded = []
    for claim in claims:
        # التحقق إذا كان الادعاء مستتبعاً من السياق
        result = nli(f"{context_text} [SEP] {claim}")

        if result[0]["label"] != "ENTAILMENT":
            ungrounded.append({
                "claim": claim,
                "verdict": result[0]["label"],
                "confidence": result[0]["score"]
            })

    return {
        "is_grounded": len(ungrounded) == 0,
        "grounded_ratio": (len(claims) - len(ungrounded)) / len(claims) if claims else 1.0,
        "ungrounded_claims": ungrounded
    }


@action(name="generate_grounded_response")
async def generate_grounded_response(
    query: str,
    context: List[Dict]
) -> str:
    """توليد استجابة مؤسسة بصرامة على السياق."""
    context_text = "\n\n".join([
        f"[المصدر: {c['metadata'].get('source', 'unknown')}]\n{c['content']}"
        for c in context
    ])

    prompt = f"""بناءً فقط على السياق التالي، أجب على سؤال المستخدم.
إذا كان السياق لا يحتوي على معلومات كافية، قل ذلك.
دائماً اذكر مصادرك باستخدام تنسيق [المصدر: اسم الملف].

السياق:
{context_text}

السؤال: {query}

الإجابة (مؤسسة على السياق فقط):"""

    # استخدام LLM المكون
    response = await llm.generate(prompt)
    return response


@action(name="regenerate_with_citations")
async def regenerate_with_citations(
    query: str,
    context: List[Dict],
    ungrounded_claims: List[Dict]
) -> str:
    """إعادة التوليد مع إزالة الادعاءات غير المؤسسة وإضافة الاستشهادات."""
    context_text = "\n\n".join([
        f"[المصدر: {c['metadata'].get('source', 'unknown')}]\n{c['content']}"
        for c in context
    ])

    ungrounded_text = "\n".join([c["claim"] for c in ungrounded_claims])

    prompt = f"""بناءً فقط على السياق التالي، أجب على سؤال المستخدم.
لا تتضمن هذه الادعاءات غير المتحققة: {ungrounded_text}
كل بيان يجب أن يكون له استشهاد مثل [المصدر: اسم الملف].

السياق:
{context_text}

السؤال: {query}

الإجابة (مع الاستشهادات):"""

    response = await llm.generate(prompt)
    return response

خط أنابيب RAG الإنتاجي

# خط أنابيب RAG كامل مع الحواجز
from nemoguardrails import LLMRails, RailsConfig
from dataclasses import dataclass
from typing import Optional

@dataclass
class RAGResult:
    answer: str
    sources: list
    grounding_score: float
    was_regenerated: bool

class GuardedRAGPipeline:
    """RAG إنتاجي مع NeMo Guardrails."""

    def __init__(self, config_path: str):
        self.config = RailsConfig.from_path(config_path)
        self.rails = LLMRails(self.config)

    async def query(
        self,
        question: str,
        user_context: dict = None
    ) -> RAGResult:
        """
        تنفيذ استعلام RAG محمي.
        """
        context = user_context or {}
        context["query"] = question

        # تنفيذ خط أنابيب القواعد
        response = await self.rails.generate_async(
            messages=[{"role": "user", "content": question}],
            context=context
        )

        # استخراج البيانات الوصفية من السياق
        retrieved_chunks = response.get("context", {}).get("retrieved_chunks", [])
        grounding_check = response.get("context", {}).get("grounding_check", {})

        return RAGResult(
            answer=response["content"],
            sources=[c["metadata"]["source"] for c in retrieved_chunks],
            grounding_score=grounding_check.get("grounded_ratio", 1.0),
            was_regenerated=grounding_check.get("was_regenerated", False)
        )

# الاستخدام
pipeline = GuardedRAGPipeline("./config")

result = await pipeline.query(
    "ما لغات البرمجة التي يدعمها منتجكم؟"
)

print(f"الإجابة: {result.answer}")
print(f"المصادر: {result.sources}")
print(f"التأسيس: {result.grounding_score:.1%}")

تدفق منع الهلوسة

# rails/anti_hallucination.co
define flow prevent hallucination
  """خط أنابيب كامل لمنع الهلوسة."""

  # الخطوة 1: الاسترجاع
  $chunks = execute search_knowledge_base(query=$user_message, top_k=5)

  # الخطوة 2: التحقق إذا كان لدينا سياق كافٍ
  $quality = execute check_retrieval_quality(query=$user_message, chunks=$chunks)

  if not $quality.sufficient
    if $quality.reason == "no_chunks"
      bot admit lack of knowledge
    else
      bot provide hedged response
    stop

  # الخطوة 3: توليد استجابة مؤسسة
  $response = execute generate_grounded_response(
    query=$user_message,
    context=$chunks
  )

  # الخطوة 4: التحقق من التأسيس
  $check = execute verify_grounding(response=$response, context=$chunks)

  if $check.grounded_ratio < 0.8
    # الكثير من الادعاءات غير المؤسسة
    $response = execute regenerate_with_citations(
      query=$user_message,
      context=$chunks,
      ungrounded_claims=$check.ungrounded_claims
    )

  # الخطوة 5: إضافة مؤشر الثقة
  $confidence = "high" if $check.grounded_ratio > 0.9 else "moderate"
  bot respond with confidence indicator $response $confidence

define bot admit lack of knowledge
  "ليس لدي معلومات متحققة حول هذا الموضوع في قاعدة معرفتي. أوصي بالتحقق من وثائقنا الرسمية أو الاتصال بالدعم."

define bot provide hedged response
  "بناءً على معلومات محدودة وجدتها، على الرغم من أنني أوصي بالتحقق: "

أفضل ممارسة RAG: تحقق دائماً من التأسيس قبل إرجاع الردود. الاستجابة التي تعترف بعدم اليقين أفضل من تلك التي تذكر بثقة معلومات غير متحققة.

التالي: تعمق في إطار عمل Guardrails AI للتحقق من صحة المخططات. :::