Domain-Specific RAG

Overview

DomainSpecificRAG gates a backend behind a vocabulary-based relevance check: queries that don't sufficiently match a domain's terms/synonyms can be rejected (in strict_mode) before any retrieval or generation happens, and answers are steered toward a domain-appropriate system prompt/language.

DomainDefinition

@dataclass
class DomainDefinition:
    name: str
    terms: List[str]
    display_name: str = ""
    language: str = "ar"
    synonyms: Dict[str, List[str]] = field(default_factory=dict)
    system_prompt: str = ""
    rejection_message: str = ""
    min_relevance: float = 0.1
    boost_weight: float = 1.5
    metadata: Dict[str, Any] = field(default_factory=dict)

my_domain = DomainDefinition(
    name="nutrition",
    display_name="Nutrition & Health",
    language="en",
    terms=["protein", "carbohydrate", "vitamin", "calories", "diet"],
    synonyms={"calories": ["energy", "kcal"]},
    system_prompt="You are a nutrition expert. Be scientifically precise.",
    rejection_message="This question is outside the scope of nutrition.",
    min_relevance=0.15,
)

PRESET_DOMAINS

A module-level constant importable alongside DomainDefinition; it supplies ready-made domains (e.g. "medical") so DomainSpecificRAG(sys, domain="medical") works without constructing a DomainDefinition manually. The exact preset contents were not enumerated in the extracted signatures — inspect domain_defination.py directly for the full list if you need to audit preset term lists.

DomainMetrics

@dataclass
class DomainMetrics:
    total_queries: int = 0
    rejected_queries: int = 0
    total_docs_used: int = 0
    errors: int = 0
    avg_relevance: float = 0.0

    def record(self, relevance: float, docs: int, rejected: bool = False): ...
    def summary(self) -> Dict[str, Any]: ...

Running counters updated on every query; summary() is what DomainSpecificRAG.get_metrics() returns.

DomainSpecificRAG

Constructor

def __init__(
    self,
    rag_system,
    domain: str | DomainDefinition,
    domain_terms: Optional[List[str]] = None,
    llm=None,
    strict_mode: bool = False,
    auto_expand_terms: bool = False,
)

Four ways to construct, per the class docstring

# 1. Preset domain
rag = DomainSpecificRAG(sys, domain="medical")

# 2. Simple domain via terms
rag = DomainSpecificRAG(sys, domain="nutrition", domain_terms=["protein", "vitamin"])

# 3. Full custom domain
domain = DomainDefinition(name="cooking", terms=["recipe", "ingredients"], system_prompt="You are a chef...")
rag = DomainSpecificRAG(sys, domain=domain)

# 4. Load from JSON
domain = DomainDefinition.load("cooking.json")
rag = DomainSpecificRAG(sys, domain=domain)

query(query, context=None) -> Dict[str, Any]

Synchronous entry point (internally delegates the heavy lifting to the async pipeline — see below — but is a plain sync method itself for compatibility).

aquery(query, context=None) -> Dict[str, Any] — pipeline

1. _measure_relevance(query) — multi-layer scoring against domain.all_terms; returns (score in [0,1], matched_terms).
2. If score < domain.min_relevance and strict_mode is on → _rejection_response(relevance) is returned immediately (no retrieval or generation call is made).
3. _enhance_query(query, matched_terms) — expands the query with the matched domain terms to steer retrieval.
4. _retrieve(query) — async retrieval normalised to List[Dict].
5. _rank_by_domain(docs) — combines the backend's own relevance score with domain-term boosting (boost_weight).
6. _generate(query, docs, extra_context) — builds a system prompt via _build_system_prompt (domain system_prompt + a hard language instruction from _LANG_INSTRUCTIONS) and calls the backend/LLM.
7. Metrics are recorded via DomainMetrics.record() on every call.

_LANG_INSTRUCTIONS

A class-level dict mapping ar/en/fr/de/es to an explicit "answer only in this language" instruction string, injected both into the system prompt (_build_system_prompt) and directly into the query text (_inject_language) — the latter exists specifically, per its docstring, as a "reliable fix" for cases where the backend doesn't respect system-prompt language instructions.

astream(query, context=None)

Streams the final answer. If rejected under strict_mode, the rejection_message is streamed immediately without any retrieval/generation call. Otherwise it delegates to rag_system.astream if available, or falls back to generating the full answer first and streaming it word-by-word.

async for token in domain_rag.astream("your question here"):
    print(token, end="", flush=True)

Other methods

Complete example

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

domain = DomainDefinition(
    name="legal",
    terms=["contract", "liability", "clause", "jurisdiction"],
    system_prompt="You are a legal assistant. Cite sources when possible.",
    rejection_message="This falls outside legal topics I can help with.",
    min_relevance=0.2,
)

rag = DomainSpecificRAG(backend, domain=domain, strict_mode=True)
result = rag.query("What does a force majeure clause typically cover?")
print(result["answer"])
print(rag.get_metrics())

Best practices

• Use strict_mode=True for user-facing assistants that must refuse off-topic questions; leave it False for internal tools where deprioritizing (not blocking) off-domain content is preferable.
• Prefer explicit synonyms over auto_expand_terms=True in production — LLM-driven vocabulary expansion is nondeterministic and happens silently.

Common mistakes

• Forgetting that min_relevance only blocks queries in strict_mode; in non-strict mode, low-relevance queries still proceed, just without a domain-term retrieval boost.
• Passing a bare domain name that isn't in PRESET_DOMAINS without also supplying domain_terms — _resolve_domain needs one or the other to build a usable DomainDefinition.

Related APIs

• Federated RAG — combine several DomainSpecificRAG instances as sources for automatic domain routing.