Domain RAG — `domain_rag` Module — Public API Reference

from fennec_community.rag.types.domain_rag import DomainDefinition, PRESET_DOMAINS, DomainSpecificRAG, DomainMetrics

from fennec_community.rag.types.domain_rag import DomainDefinition

DomainDefinition(
    name: str,
    terms: List[str],
    display_name: str = "",
    language: str = "ar",
    synonyms: Dict[str, List[str]] = {},
    system_prompt: str = "",
    rejection_message: str = "",
    min_relevance: float = 0.10,
    boost_weight: float = 1.5,
    metadata: Dict[str, Any] = {},
)

from fennec_community.rag.types.domain_rag import DomainDefinition

nutrition = DomainDefinition(
    name="nutrition",
    display_name="التغذية والصحة",
    language="ar",
    terms=["بروتين", "كربوهيدرات", "فيتامين", "سعرات", "حمية"],
    synonyms={"سعرات": ["كالوري", "طاقة"], "حمية": ["رجيم", "دايت"]},
    system_prompt="أنت خبير تغذية. أجب بدقة علمية وأضف تحذيرات عند الضرورة.",
    rejection_message="هذا السؤال خارج نطاق التغذية والصحة.",
    min_relevance=0.15,
    boost_weight=1.8,
)

domain.all_terms -> Set[str]

print(nutrition.all_terms)
# {'بروتين', 'كربوهيدرات', 'فيتامين', 'سعرات', 'حمية', 'كالوري', 'طاقة', 'رجيم', 'دايت'}

domain.add_terms(*terms: str) -> None

domain.add_terms("أوميغا 3", "ألياف", "معادن")
print(len(domain.all_terms))  # updated count

domain.add_synonym(term: str, *synonyms: str) -> None

domain.add_synonym("بروتين", "بروتينات", "ببتيد")
print(domain.synonyms["بروتين"])  # ['بروتينات', 'ببتيد']

domain.to_dict() -> Dict

import json
print(json.dumps(domain.to_dict(), ensure_ascii=False, indent=2))

DomainDefinition.from_dict(data: Dict) -> DomainDefinition

data = json.loads(raw_json)
domain = DomainDefinition.from_dict(data)

domain.save(path: str) -> None

domain.save("domains/nutrition.json")

DomainDefinition.load(path: str) -> DomainDefinition

domain = DomainDefinition.load("domains/nutrition.json")
rag = DomainSpecificRAG(my_rag, domain=domain)

from fennec_community.rag.types.domain_rag import PRESET_DOMAINS

from fennec_community.rag.types.domain_rag import PRESET_DOMAINS, DomainSpecificRAG

# Use directly by key string
rag = DomainSpecificRAG(my_rag, domain="medical")

# Inspect a preset
print(PRESET_DOMAINS["legal"].terms)

# Extend a preset before use
medical = DomainDefinition.from_dict(PRESET_DOMAINS["medical"].to_dict())
medical.add_terms("تصوير مقطعي", "رنين مغناطيسي")
rag = DomainSpecificRAG(my_rag, domain=medical)

from fennec_community.rag.types.domain_rag import DomainSpecificRAG

DomainSpecificRAG(
    rag_system: Any,
    domain: Union[str, DomainDefinition],
    domain_terms: Optional[List[str]] = None,
    llm: Optional[Any] = None,
    strict_mode: bool = False,
    auto_expand_terms: bool = False,
)

from fennec_community.rag.types.domain_rag import DomainSpecificRAG, DomainDefinition

# Pattern 1 — preset by key
rag = DomainSpecificRAG(my_rag, domain="medical", strict_mode=True)

# Pattern 2 — preset + extra terms
rag = DomainSpecificRAG(
    my_rag,
    domain="medical",
    domain_terms=["تصوير مقطعي", "رنين مغناطيسي"],
)

# Pattern 3 — custom DomainDefinition
nutrition = DomainDefinition(
    name="nutrition",
    terms=["بروتين", "فيتامين", "سعرات"],
    language="ar",
)
rag = DomainSpecificRAG(my_rag, domain=nutrition, auto_expand_terms=True)

rag.query(
    query: str,
    context: Optional[Dict] = None,
) -> Dict[str, Any]

result = rag.query("ما هي الجرعة اليومية الموصى بها من فيتامين د؟")

print(result["answer"])
print(f"Domain relevance : {result['domain_relevance']:.2%}")
print(f"Matched terms    : {result['matched_terms']}")
print(f"Docs used        : {result['docs_count']}")
print(f"Latency          : {result['latency_ms']} ms")

await rag.aquery(
    query: str,
    context: Optional[Dict] = None,
) -> Dict[str, Any]

import asyncio

async def main():
    result = await rag.aquery("ما هي أسباب ارتفاع ضغط الدم؟")
    print(result["answer"])

asyncio.run(main())

rag.add_document(
    text: str,
    metadata: Optional[Dict] = None,
) -> Any

rag.add_document(
    "فيتامين د ضروري لصحة العظام وامتصاص الكالسيوم. الجرعة اليومية الموصى بها للبالغين هي 600-800 وحدة دولية.",
    metadata={"source": "who_guidelines_2024", "page": 12},
)

rag.add_terms(*terms: str) -> None

rag.add_terms("أوميغا 3", "حمض الفوليك", "الكولاجين")

rag.add_synonym(term: str, *synonyms: str) -> None

rag.add_synonym("فيتامين د", "كوليكالسيفيرول", "D3")

rag.get_metrics() -> Dict[str, Any]

import json
print(json.dumps(rag.get_metrics(), indent=2, ensure_ascii=False))

{
  "domain": "medical",
  "total_queries": 42,
  "rejected": 3,
  "acceptance_rate": "92.9%",
  "avg_relevance": 0.612,
  "avg_docs_per_query": 7.2,
  "errors": 0,
  "total_terms": 14
}

rag.save_domain(path: str) -> None

rag.add_terms("تصوير مقطعي", "رنين مغناطيسي")
rag.save_domain("domains/medical_extended.json")

# Later — reload
domain = DomainDefinition.load("domains/medical_extended.json")
rag2 = DomainSpecificRAG(my_rag, domain=domain)

await rag.agenerate(query: str, **kwargs) -> str

answer = await rag.agenerate("ما هو تأثير الكافيين على ضغط الدم؟")
print(answer)

await rag.aretrieve(query: str, **kwargs) -> List[Dict]

docs = await rag.aretrieve("فيتامين د والكالسيوم")
for doc in docs:
    print(f"[{doc['score']:.3f}] {doc['text'][:100]}")

async for token in rag.astream(
    query: str,
    context: Optional[Dict] = None,
):
    print(token, end="", flush=True)

import asyncio

async def stream():
    print("Answer: ", end="")
    async for token in rag.astream("كيف يؤثر السكر على الجهاز المناعي؟"):
        print(token, end="", flush=True)
    print()

asyncio.run(stream())

async with DomainSpecificRAG(my_rag, domain="medical") as rag:
    result = await rag.aquery("ما هي أعراض السكري؟")
    print(result["answer"])
# __aexit__ is a no-op — used as a scope delimiter

repr(rag)

DomainSpecificRAG(domain='medical', terms=14, strict=True, queries=42)

from fennec_community.rag.types.domain_rag import DomainMetrics

metrics.record(
    relevance: float,
    docs: int,
    rejected: bool = False,
) -> None

metrics.summary() -> Dict[str, Any]

print(rag.metrics.summary())
# {'total_queries': 10, 'rejected': 2, 'acceptance_rate': '80.0%',
#  'avg_relevance': 0.523, 'avg_docs_per_query': 6.0, 'errors': 0}

{
    "answer": "سؤالك خارج نطاق الطب والصحة.",
    "domain": "medical",
    "domain_display": "الطب والصحة",
    "domain_relevance": 0.042,
    "matched_terms": [],
    "enhanced_query": "",
    "docs_count": 0,
    "latency_ms": 0.0,
    "rejected": True,
    "sources": [],
}

{
    "answer": "الجرعة اليومية الموصى بها من فيتامين د للبالغين هي 600–800 وحدة دولية...",
    "domain": "medical",
    "domain_display": "الطب والصحة",
    "domain_relevance": 0.743,
    "matched_terms": ["فيتامين", "جرعة"],
    "enhanced_query": "ما هي الجرعة اليومية الموصى بها من فيتامين د؟ عقار",
    "docs_count": 8,
    "latency_ms": 312.5,
    "sources": [
        {"text": "فيتامين د ضروري لصحة العظام...", "score": 1.234, "metadata": {}},
        ...
    ],
}

rag.query(query, context) / await rag.aquery(query, context)
│
├─ _measure_relevance(query)
│    ├─ Layer 1: direct word-set intersection with all_terms
│    ├─ Layer 2: partial string containment check
│    └─ score = (direct_hits * 1.0 + partial_hits * 0.5) / query_words
│              * boost_weight   →  capped at 1.0
│
├─ strict_mode and score < min_relevance
│    └─ metrics.record(rejected=True) → return _rejection_response()
│
├─ _enhance_query(query, matched_terms)
│    ├─ no matched terms → prepend "[display_name]" to query
│    └─ matched terms   → append up to 6 synonyms (2 per term, top-3 terms)
│                         → trimmed to 600 chars
│
├─ _retrieve(enhanced_query)            [async, top_k=15]
│    ├─ rag_system.retrieve() called in thread pool
│    ├─ normalises tuples, dicts, and object formats uniformly
│    └─ returns List[Dict{text, score, metadata}]
│
├─ _rank_by_domain(docs)
│    ├─ for each doc: domain_hits = doc_words ∩ all_terms
│    ├─ domain_bonus = min(hits / len(all_terms), 0.4) * boost_weight
│    ├─ combined_score = rag_score + domain_bonus
│    ├─ sort descending by combined_score
│    ├─ strict_mode: filter docs with domain_hits == 0
│    └─ return top-10 docs
│
├─ _generate(query, docs, context)      [async]
│    ├─ build system_prompt: domain.system_prompt + language-lock instruction
│    ├─ inject language instruction directly into query string
│    ├─ call rag_system.generate(lang_query, ctx) in thread pool
│    └─ auto-build sources list if not in response
│
├─ metrics.record(relevance, docs_count)
│
└─ return enriched Dict (answer + 8 diagnostic fields)

query = "ما هي جرعة الدواء؟"
all_terms = {"دواء", "علاج", "مرض", "أعراض", "تشخيص", "جرعة", "مريض", "طبيب", "عقار", "دواية"}

query_words = {"ما", "هي", "جرعة", "الدواء؟"}
direct_matches = {"جرعة", "دواء"}  (normalised)
partial_matches = {"دواية"}        (contains "دوا")

score_raw = (2 * 1.0 + 1 * 0.5) / 4 = 0.625
score = min(0.625 * 1.5, 1.0) = 0.9375  ✅ accepted

domain.language = "ar"
→ lang_instruction = "يجب أن تكون إجابتك باللغة العربية حصراً."

system_prompt  = domain.system_prompt + "\n" + lang_instruction
enhanced_query = lang_instruction + "\n" + original_query

# Both channels used to maximise LLM compliance

import asyncio
import json
from fennec_community.rag.types.domain_rag import DomainSpecificRAG, DomainDefinition, PRESET_DOMAINS
from fennec_community.rag.core import RAGSystem

my_rag = RAGSystem(
    vector_db=my_vector_db,
    llm=my_llm,
    chunker=my_chunker,
    context_manager=my_ctx_mgr,
    config=config,
    enable_query_expansion=True,
    query_expansion_variants=3,
)
# ── 1. Use a preset domain ────────────────────────────────────────────────
rag_medical = DomainSpecificRAG(
    rag_system=my_rag,
    domain="medical",
    strict_mode=True,
)

# ── 2. Index domain documents ─────────────────────────────────────────────
rag_medical.add_document(
    "فيتامين د ضروري لامتصاص الكالسيوم. الجرعة الموصى بها: 600–800 وحدة دولية يومياً.",
    metadata={"source": "who_guidelines", "page": 5},
)
rag_medical.add_document(
    "ارتفاع ضغط الدم (فوق 140/90) يستدعي تعديل نمط الحياة أو العلاج الدوائي.",
    metadata={"source": "cardiology_manual", "chapter": 3},
)

# ── 3. Synchronous query ──────────────────────────────────────────────────
result = rag_medical.query("ما هي الجرعة اليومية الموصى بها من فيتامين د؟")
print(result["answer"])
print(f"Relevance  : {result['domain_relevance']:.2%}")
print(f"Matched    : {result['matched_terms']}")
print(f"Docs used  : {result['docs_count']}")
print(f"Latency    : {result['latency_ms']} ms")

# ── 4. Off-topic query (strict_mode=True rejects it) ─────────────────────
result_rej = rag_medical.query("من هو مخترع الهاتف؟")
print(result_rej["answer"])        # → rejection_message
print(result_rej.get("rejected"))  # → True

# ── 5. Custom domain ──────────────────────────────────────────────────────
nutrition = DomainDefinition(
    name="nutrition",
    display_name="التغذية والصحة",
    language="ar",
    terms=["بروتين", "كربوهيدرات", "فيتامين", "سعرات", "حمية"],
    synonyms={"سعرات": ["كالوري", "طاقة"], "حمية": ["رجيم", "دايت"]},
    system_prompt="أنت خبير تغذية. أجب بدقة علمية.",
    rejection_message="هذا السؤال خارج نطاق التغذية والصحة.",
    min_relevance=0.15,
)

rag_nutrition = DomainSpecificRAG(
    my_rag,
    domain=nutrition,
    strict_mode=False,
    auto_expand_terms=True,   # LLM adds 10 more terms at startup
)

# ── 6. Add runtime vocabulary ─────────────────────────────────────────────
rag_nutrition.add_terms("أوميغا 3", "حمض الفوليك")
rag_nutrition.add_synonym("فيتامين", "فيتامينات", "vitamins")

# ── 7. Async query ────────────────────────────────────────────────────────
async def async_demo():
    result = await rag_nutrition.aquery("كيف يؤثر البروتين على بناء العضلات؟")
    print(result["answer"])

asyncio.run(async_demo())

# ── 8. Async generate (answer only) ──────────────────────────────────────
async def gen_demo():
    answer = await rag_nutrition.agenerate("ما هي أفضل مصادر الكربوهيدرات؟")
    print(answer)

asyncio.run(gen_demo())

# ── 9. Streaming ──────────────────────────────────────────────────────────
async def stream_demo():
    print("Streaming: ", end="")
    async for token in rag_nutrition.astream("ما الفرق بين الكربوهيدرات البسيطة والمعقدة؟"):
        print(token, end="", flush=True)
    print()

asyncio.run(stream_demo())

# ── 10. Metrics ───────────────────────────────────────────────────────────
print(json.dumps(rag_medical.get_metrics(), indent=2, ensure_ascii=False))
print(json.dumps(rag_nutrition.get_metrics(), indent=2, ensure_ascii=False))

# ── 11. Save and reload domain ────────────────────────────────────────────
rag_nutrition.save_domain("domains/nutrition_v2.json")
loaded = DomainDefinition.load("domains/nutrition_v2.json")
rag_v2  = DomainSpecificRAG(my_rag, domain=loaded)

# ── 12. Inspect a preset ──────────────────────────────────────────────────
print(PRESET_DOMAINS["legal"].terms)
print(PRESET_DOMAINS["financial"].system_prompt)

# ── 13. Async context manager ─────────────────────────────────────────────
async def ctx_demo():
    async with DomainSpecificRAG(my_rag, domain="academic") as rag:
        result = await rag.aquery("ما هي الفرضية الأساسية في هذا البحث؟")
        print(result["answer"])

asyncio.run(ctx_demo())

# ── 14. Raw retrieval (no generation) ────────────────────────────────────
async def retrieve_only():
    docs = await rag_medical.aretrieve("أعراض السكري")
    for doc in docs[:3]:
        print(f"[{doc['score']:.3f}] {doc['text'][:80]}")

asyncio.run(retrieve_only())

from fennec_community.llm import MistralInterface
from fennec_community.document_loaders import TextLoader 
from fennec_community.vector_database import FAISSVectorDatabase
from fennec_community.chunks import ArabicTextChunker
from fennec_community.context import ContextManager
from fennec_community.embeddings import OllamaEmbedder
from fennec_community.rag.core import RAGSystem , LoadedDocument
from fennec_community.rag.types.domain_rag import DomainSpecificRAG , DomainDefinition


import nest_asyncio
nest_asyncio.apply()
base_rag = RAGSystem(
    vector_db=vector_db, llm=llm,
    chunker=chunker, context_manager=context_manager,
)
l1 = LoadedDocument(
    page_content=" القلب هو العضو المركزي في الجهاز الدوري. يضخ الدم عبر الشرايين والأوردة. أمراض القلب تشمل: ارتفاع الضغط، الذبحة الصدرية، قصور القلب الاحتقاني.",
    metadata={"source": "heart.txt"})
l2 = LoadedDocument(
    page_content=" السكري مرض مزمن يؤثر على كيفية معالجة الجسم للسكر. النوع الأول يعتمد على الأنسولين، والنوع الثاني يرتبط بمقاومة الأنسولين.",
    metadata={"source": "diabetes.txt"})

base_rag.add_documents(docs=[l1, l2])

# --- 2. طريقة 1: مجال جاهز ---
medical_rag = DomainSpecificRAG(
    rag_system=base_rag,
    domain="medical",
    strict_mode=True,
)

# --- 3. طريقة 2: مجال مخصص كامل ---
medical_domain = DomainDefinition(
    name="medical",
    display_name="الطب والصحة",
    terms=["قلب", "ضغط", "سكري", "جراحة", "أنسولين", "شرايين",
           "أوردة", "دم", "علاج", "دواء", "مستشفى", "طبيب"],
    system_prompt="أنت طبيب متخصص تُجيب بدقة علمية عالية.",
    min_relevance=0.15,
    boost_weight=1.5,
    rejection_message="عذرًا، تخصصي في الطب فقط.",
    synonyms={
        "قلب": ["فؤاد", "عضلة القلب"],
        "ضغط": ["ضغط الدم", "hypertension"],
    },
    metadata={"version": "2.0", "reviewed_by": "dr-ahmed"},
)

custom_rag = DomainSpecificRAG(
    rag_system=base_rag,
    domain=medical_domain,
    strict_mode=True,
)

custom_rag.add_terms("نبضات القلب", "الأوعية الدموية")
custom_rag.add_synonym("علاج", "معالجة", "مداواة")

# --- 4. الاستعلام المتزامن ---
q1 = "ما هو علاج ارتفاع ضغط الدم؟"
result1 = custom_rag.query(q1)
print(f"❓ {q1}")
print(f"🎯 الصلة: {result1['domain_relevance']:.2%}")
print(f"🔍 المصطلحات: {result1['matched_terms']}")
print(f"📝 الإجابة: {result1['answer'][:200]}")

# سؤال خارج المجال — سيُرفض في strict_mode
q2 = "ما أفضل برنامج لتعلم البرمجة؟"
result2 = custom_rag.query(q2)
print(f"\n❓ {q2}")
print(f"❌ مرفوض: {result2.get('rejected', False)}")
print(f"💬 {result2['answer']}")

# --- 5. الإحصائيات ---
metrics = custom_rag.get_metrics()
print("\n📊 إحصائيات المجال:")
for k, v in metrics.items():
    print(f"  {k}: {v}")

# --- 6. حفظ المجال وتحميله ---
custom_rag.save_domain("medical_domain.json")
domain_from_file = DomainDefinition.load("medical_domain.json")
rag_from_file = DomainSpecificRAG(rag_system=base_rag, domain=domain_from_file)