Public Python API

This reference covers the stable, user-facing facades and component factories of Muffakir. See the relevant build guides for end-to-end recipes.

MuffakirRAG

Builds a full RAG pipeline from a configuration. Use ask, get_similar_documents, and add_documents in application code.

Muffakir.Muffakir.MuffakirRAG

Multilingual RAG (Retrieval-Augmented Generation) library.

A configurable RAG library providing document processing, retrieval, and question-answering capabilities.

ask(question: str, **kwargs: Any) -> Dict[str, Any]

Ask a question and get an intelligent answer.

Parameters:

Name	Type	Description	Default
question	str	The question in Arabic or English	required
**kwargs	Any	Additional parameters to override defaults per query (thread-safe) - k (int): Number of documents to retrieve - retrieval_method (str): Retrieval method to use	{}

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Response containing answer and metadata

get_similar_documents(query: str, k: Optional[int] = None, method: Optional[str] = None) -> List[Document]

Retrieve similar documents for a given query.

Parameters:

Name	Type	Description	Default
query	str	Search query	required
k	int	Number of documents to retrieve	None
method	str	Retrieval method to use	None

Returns:

Type	Description
List[Document]	List[Document]: Similar documents

get_similar_documents_with_trace(query: str, k: Optional[int] = None, method: Optional[str] = None) -> RetrievalTelemetryResult

Run the configured retrieval architecture and return telemetry.

This is intentionally separate from the legacy list-returning method so existing library callers keep the same interface.

add_documents(documents: Union[List[Document], List[str]]) -> bool

Add new documents to the knowledge base.

Parameters:

Name	Type	Description	Default
documents	Union[List[Document], List[str]]	List of Document objects or file paths	required

Returns:

Name	Type	Description
bool	bool	Success status

get_config() -> Dict[str, Any]

Get current configuration.

MuffakirRetrieval

Runs the retrieval part of a pipeline without answer generation. It is useful for retrieval-only evaluation and debugging.

Muffakir.MuffakirRetrieval.MuffakirRetrieval

Retrieval-only RAG facade: exercises embedding/chunking/retrieval-method/ reranking/query-transformation exactly like MuffakirRAG, but never builds or calls a main generation LLM, AnswerGenerator, or HallucinationsCheck.

EvalRunner-compatible: implements get_similar_documents() with the same signature MuffakirRAG.get_similar_documents() has, so it is a drop-in rag for retrieval-metric-only evaluation.

get_similar_documents(query: str, k: Optional[int] = None) -> List[Document]

Transform (if enabled) -> retrieve -> rerank (if enabled). No LLM generation call anywhere in this path.

get_similar_documents_with_trace(query: str, k: Optional[int] = None) -> RetrievalTelemetryResult

Telemetry-aware retrieval used by EvalRunner.

The existing get_similar_documents method remains list-returning for public API compatibility.

VectorDB

Five vector-store backends accessible through a uniform interface. See the Retrieval and vector stores guide for backend selection, retrieval methods, and custom backend recipes.

VectorDB.factory.create_vector_db(provider: str = 'chroma', embedding_provider: Optional[EmbeddingProvider] = None, **kwargs: Any) -> BaseVectorDBManager

Factory function to instantiate Vector Database providers.

Parameters:

Name	Type	Description	Default
provider	str	'chroma', 'qdrant', 'pinecone', 'faiss', or 'milvus'.	'chroma'
embedding_provider	EmbeddingProvider	Injected embedding provider.	None
**kwargs	Any	Provider-specific configuration options (path, collection_name, api_key, etc.).	{}

Returns:

Name	Type	Description
BaseVectorDBManager	BaseVectorDBManager	An instance of BaseVectorDBManager.

VectorDB.base.BaseVectorDBManager

Bases: ABC

Abstract Base Class for Vector Database Managers in Muffakir RAG.

All VectorDB backends must inherit from this class and implement: - vector_store: property returning the underlying LangChain VectorStore - add_documents(): ingest documents into the backend - load_all_documents(): retrieve all persisted documents from the backend

add_documents(documents: List[Document]) -> None abstractmethod

Add documents to the vector store.

similarity_search(query: str, k: int = 2) -> List[Document]

Perform similarity vector search.

max_marginal_relevance_search(query: str, k: int = 2, fetch_k: int = 12) -> List[Document]

Perform Maximal Marginal Relevance (MMR) search.

load_all_documents() -> List[Document] abstractmethod

Retrieve all documents from the persistent vector store.

Used by HybridRAG for BM25 in-memory indexing. Each backend must implement this to fetch documents from its own storage layer, since _all_documents is ephemeral and lost on process restart.

Returns:

Type	Description
List[Document]	List[Document]: All documents currently stored in this backend.

get_all_documents() -> List[Document]

Returns all stored documents for local BM25 Hybrid indexing.

Delegates to load_all_documents() if the in-memory cache is empty (e.g. after a process restart when the backend has persisted data).

VectorDB.ChromaDBManager.ChromaDBManager

Bases: BaseVectorDBManager

Chroma Vector Database Provider.

add_documents(documents: List[Document]) -> None

Add documents to the Chroma collection and persist if supported.

load_all_documents() -> List[Document]

Retrieve all documents from the Chroma collection. Used for BM25 hybrid indexing after process restart.

get_collection_count() -> int

Return the number of documents in the Chroma collection.

VectorDB.FAISSDBManager.FAISSDBManager

Bases: BaseVectorDBManager

FAISS (Facebook AI Similarity Search) Vector Database Provider. Supports in-memory vector storage with optional disk persistence.

add_documents(documents: List[Document]) -> None

Add documents to FAISS index and save to disk.

load_all_documents() -> List[Document]

FAISS is an in-memory index with no native document storage retrieval. Returns the in-memory cache populated during add_documents(). For BM25 hybrid search after restart, re-index documents explicitly.

VectorDB.QdrantDBManager.QdrantDBManager

Bases: BaseVectorDBManager

Qdrant Vector Database Provider using langchain_qdrant.

add_documents(documents: List[Document]) -> None

Ingest documents into Qdrant store.

load_all_documents() -> List[Document]

Retrieve all documents from the Qdrant collection. Used for BM25 hybrid indexing after process restart.

VectorDB.PineconeDBManager.PineconeDBManager

Bases: BaseVectorDBManager

Pinecone Vector Database Provider using langchain_pinecone.

add_documents(documents: List[Document]) -> None

Ingest documents into Pinecone index.

load_all_documents() -> List[Document]

Pinecone does not support bulk document retrieval from the SDK. Returns the in-memory cache populated during add_documents().

VectorDB.MilvusDBManager.MilvusDBManager

Bases: BaseVectorDBManager

Milvus Vector Database Provider using langchain_milvus. Supports both local file-based Milvus Lite (./milvus_local.db) and remote cluster (http://localhost:19530).

add_documents(documents: List[Document]) -> None

Ingest documents into Milvus store.

load_all_documents() -> List[Document]

Milvus does not support simple bulk retrieval without a query. Returns the in-memory cache populated during add_documents(). For BM25 hybrid search after restart, re-index documents explicitly.

MuffakirEvaluation

Evaluates a compatible RAG instance or web search agent against a Q&A dataset. See the Evaluation guide for metric formulas, Arabic text matching, and output report analysis.

Muffakir.MuffakirEvaluation.MuffakirEvaluation

Custom RAG evaluation facade.

Option 1: inject an already-built MuffakirRAG via evaluate(rag=..., dataset=...). Dataset must match MuffakirSyntheticData / QAPair schema.

evaluate(rag: Any, dataset: DatasetInput, save: bool = False, output_dir: Optional[str] = None) -> EvaluationReport

Evaluate an injected MuffakirRAG on a QAPair-compatible dataset.

Parameters:

Name	Type	Description	Default
rag	Any	Initialized MuffakirRAG instance	required
dataset	DatasetInput	path | DataFrame | list[dict] with QAPair columns	required
save	bool	if True, write report to output_dir	False
output_dir	Optional[str]	override config output_dir when saving	None

get_config() -> Dict[str, Any]

Evaluation

Evaluation loop execution, report schemas, metric computation, dataset loaders, and refusal detection.

Runner and dataset

Evaluation.runner.EvalRunner

Evaluation loop over QAPair samples against an injected MuffakirRAG.

Sequential by default (max_workers=1). Pass max_workers>1 to evaluate samples concurrently via a thread pool — safe because each sample's LLM/retrieval calls are I/O-bound and every shared collaborator (rag, llm_provider, the metric objects) is either stateless per call or has had its shared-state races fixed (see MuffakirRAG.ask()).

run(pairs: List[QAPair]) -> EvaluationReport

Evaluation.dataset.load_evaluation_dataset(dataset: DatasetInput, fail_fast: bool = True, max_samples: Optional[int] = None) -> List[QAPair]

Load and validate an evaluation dataset into QAPair objects.

Accepts file path, DataFrame, or list of dicts. Required columns: question, answer, context. Optional: chunk_id, source_file.

Models and reports

Evaluation.models.EvaluationReport dataclass

Sklearn-like evaluation results object.

summary() -> Dict[str, float]

Flat aggregate metrics dict (like sklearn classification_report averages).

to_dataframe() -> 'pd.DataFrame'

save(output_dir: str) -> None

Evaluation.models.EvalSampleResult dataclass

Evaluation.models.RetrievalScores dataclass

Evaluation.models.GenerationScores dataclass

Metrics and judges

Evaluation.metrics.generation.FaithfulnessMetric

Faithfulness via ContextGroundingChecker (1.0 grounded, 0.0 otherwise).

score(answer: str, context: str, query: str = '') -> Optional[float]

Return 1.0 if grounded, 0.0 if hallucinated.

Raises whatever ContextGroundingChecker.check() raises on failure (e.g. HallucinationCheckError) instead of swallowing it — the caller (Evaluation/runner.py::_evaluate_one) already distinguishes typed, systemic failures (abort the whole run) from per-sample bugs (exclude just this sample) and must see the real exception to do so.

Evaluation.metrics.generation.AnswerCorrectnessMetric

LLM-as-judge answer correctness against gold answer (score 0–1).

score(question: str, gold_answer: str, predicted_answer: str) -> Optional[float]

Return correctness score (0.0-1.0).

A prompt-formatting bug propagates untyped (per-sample exclusion via the caller's dispatcher). A judge LLM-call failure is classified into a typed ProviderError and raised — a live provider outage will fail identically for every remaining sample, so the caller (Evaluation/runner.py::_evaluate_one) aborts the whole run instead of silently excluding N samples one by one.

Evaluation.metrics.generation.LLMJudgeRatingMetric

Semantic answer correctness against a reference answer (integer 1–5).

score(gold_answer: str, predicted_answer: str) -> int

Return an integer rating from 1 through 5.

Malformed judge output raises ValueError so the runner records a per-sample evaluation error instead of inventing a valid-looking rating. Provider-call failures are promoted to ProviderError and abort the run, matching the other LLM-backed generation metrics.

Observability helpers

Evaluation.refusals.detect_answer_refusal(answer: str) -> Optional[str]

Return a stable reason for an explicit no-answer response, if any.

We intentionally require the whole (short) answer to be a refusal. This avoids flagging a legitimate answer that merely quotes the phrase while discussing it.

MuffakirComposer

Automated architecture search and hyperparameter optimization for RAG pipelines. See the Composer search guide for search space definitions, Pareto frontier analysis, and checkpoint recovery.

Composer.composer.MuffakirComposer

Automated Architecture Search for RAG Pipelines.

Provides a scikit-learn-like interface for finding optimal RAG pipeline configurations by systematically evaluating different combinations of: - Query expansion methods - Retrieval strategies - Reranking approaches - Top-k values

Example

composer = MuffakirComposer(config={
    "data_dir": "/path/to/documents",
    "api_key": "your-api-key",
    "llm_provider": "together",
    "llm_model": "meta-llama/Llama-3-8b-chat-hf",
})

report = composer.fit(
    search_space={
        "query_expansion": ["none", "multi_query", "hyde"],
        "retrieval": ["similarity_search", "hybrid"],
        "reranking": ["none", "cross_encoder"],
        "k": [3, 5, 10],
    },
    n_jobs=4,
)

print(report.best_config)
print(report.best_score)

Attributes:

Name	Type	Description
config		Base configuration dictionary
search_space		ConfigSpace instance defining the search space

fit(search_space: Optional[Dict[str, List]] = None, eval_dataset: Optional[Union[str, List[QAPair]]] = None, strategy: str = 'grid', n_jobs: int = 4, metrics: Optional[List[str]] = None, metric_weights: Optional[Dict[str, float]] = None, max_eval_samples: int = 50, save_report: bool = True, report_path: str = './muffakir_report.json', checkpoint_dir: str = './muffakir_checkpoints/', resume: bool = True, max_trials: Optional[int] = None, max_runtime_minutes: Optional[float] = None, custom_pricing: Optional[Dict[str, Dict[str, float]]] = None, enable_trace: bool = True, trace_dir: Optional[str] = None, trace_queue: Optional[Any] = None) -> ComposerReport

Run fit inside a whole-run observation and writer lifecycle.

ComposerUI supplies its own cross-process queue so it can begin tracing before constructing Composer. Direct SDK callers get an equivalent session created here and drained in finally.

clear_checkpoint(checkpoint_dir: str = './muffakir_checkpoints/') -> None

Clear checkpoint for fresh run.

Parameters:

Name	Type	Description	Default
checkpoint_dir	str	Checkpoint directory to clear	'./muffakir_checkpoints/'

Composer

Architecture search space, checkpoint management, and trial reporting.

Configuration space and defaults

Composer.config_space.ConfigSpace

Manages the search space for architecture optimization.

Provides utilities for: - Validating search space configurations - Generating all combinations (grid search) - Computing total number of trials

stages: List[str] property

Get list of stage names in the search space.

total_combinations: int property

Calculate total number of pipeline combinations.

__init__(search_space: Optional[Dict[str, List[Any]]] = None)

Initialize ConfigSpace.

Parameters:

Name	Type	Description	Default
search_space	Optional[Dict[str, List[Any]]]	Custom search space dictionary. If None, uses DEFAULT_SEARCH_SPACE.	None

get_options(stage: str) -> List[Any]

Get available options for a specific stage.

Parameters:

Name	Type	Description	Default
stage	str	Stage name	required

Returns:

Type	Description
List[Any]	List of available options for that stage

generate_combinations() -> List[Dict[str, Any]]

Generate all possible pipeline configurations.

Uses itertools.product to create the Cartesian product of all stage options.

Returns:

Type	Description
List[Dict[str, Any]]	List of configuration dictionaries, one per combination

generate_combinations_with_ids() -> List[tuple]

Generate combinations with trial IDs.

Returns:

Type	Description
List[tuple]	List of (trial_id, config_dict) tuples

filter_combinations(completed_ids: set, combinations: Optional[List[tuple]] = None) -> List[tuple]

Filter out already completed combinations.

Parameters:

Name	Type	Description	Default
completed_ids	set	Set of trial IDs that have been completed	required
combinations	Optional[List[tuple]]	Optional pre-generated combinations with IDs	None

Returns:

Type	Description
List[tuple]	List of (trial_id, config_dict) tuples for remaining trials

summary() -> str

Get a text summary of the search space.

Composer.config_space.DEFAULT_SEARCH_SPACE: Dict[str, List[Any]] = {'query_expansion': ['none', 'multi_query', 'hyde', 'step_back'], 'retrieval': ['similarity_search', 'max_marginal_relevance', 'hybrid'], 'reranking': ['none', 'semantic_similarity', 'cross_encoder'], 'k': [3, 5, 10]} module-attribute

Checkpoint management

Composer.checkpoint.CheckpointManager

Manages checkpoint storage for architecture search trials.

Saves completed trial results to a JSON file so that the search can be resumed after a crash or interruption.

Writes are atomic (temp file + os.replace) so a crash mid-write never corrupts the existing checkpoint — exactly the scenario this feature exists to survive.

Attributes:

Name	Type	Description
checkpoint_dir		Directory to store checkpoint files
checkpoint_file		Path to the main checkpoint JSON file

save_trial(trial: TrialResult, search_space: Optional[Dict[str, Any]] = None) -> None

Save a completed trial to the checkpoint file.

Replaces an existing entry with the same trial_id (idempotent) and writes atomically so a crash never corrupts the checkpoint.

Parameters:

Name	Type	Description	Default
trial	TrialResult	The completed TrialResult to save	required
search_space	Optional[Dict[str, Any]]	The search space this trial's ID was generated from. Stored once and carried forward on every subsequent save (pass it every call — cheap, small, constant per run) so a resumed run can detect a search space change via validate_search_space.	None

clear() -> None

Clear the checkpoint file for a fresh run.

Deletes the checkpoint file if it exists.

Search results and reporting

Composer.results.report.ComposerReport dataclass

Aggregated report from MuffakirComposer architecture search.

Attributes:

Name	Type	Description
trials	List[TrialResult]	List of all trial results
best_trial	Optional[TrialResult]	The trial with highest composite score
search_space	Dict[str, List]	The search space that was explored
total_duration_ms	float	Total execution time in milliseconds
metrics_used	List[str]	List of metrics that were computed
created_at	datetime	Timestamp when report was created

to_dataframe() -> pd.DataFrame

Convert results to pandas DataFrame.

Returns:

Type	Description
DataFrame	pd.DataFrame: DataFrame with all trial results

get_top_n_trials(n: int = 5) -> List[TrialResult]

Get top N trials by composite score.

get_pareto_frontier(objectives: Optional[List[Tuple[str, str]]] = None) -> List[TrialResult]

Return the non-dominated set of successful trials across the given objectives.

Each objective is a (field_name, direction) pair, direction is "max" or "min". Default objectives are quality (composite_score, maximize) vs latency (latency_ms, minimize) — both fields exist on every TrialResult today. A cost-based objective can be added later as an extra tuple without changing this method's signature, once per-trial cost exists.

A trial is on the frontier unless some other trial is at least as good on every objective and strictly better on at least one (i.e. it is not dominated by any other trial).

get_failure_clusters() -> Dict[str, List[int]]

Group failed trials' trial_ids by error_code (or error_type, or "unknown"). See Trace.clustering.cluster_failures.

Composer.results.trial.TrialResult dataclass

Stores the results of a single pipeline configuration trial.

Attributes:

Name	Type	Description
trial_id	int	Unique identifier for this trial
config	Dict[str, Any]	Pipeline configuration dictionary
metrics	Dict[str, float]	Computed evaluation metrics (recall, faithfulness, etc.)
composite_score	float	Weighted average of all metrics
latency_ms	float	Total execution time in milliseconds
error	Optional[str]	Error message if trial failed, None otherwise
error_code	Optional[str]	Stable machine-readable failure code (e.g. "PROVIDER_TIMEOUT"), set from a MuffakirError's error_code when the trial fails
error_type	Optional[str]	The failing exception's class name (e.g. "ProviderTimeoutError")
completed_at	datetime	Timestamp when trial completed
token_usage	Dict[str, int]	Aggregated prompt/completion/total token counts across every distinct LLMProvider used by this trial (generation, query-transform override if any, and the eval judge). Always populated when usage was observable, even if pricing for it is unknown.
cost_usd	Optional[float]	Best-effort dollar cost for this trial's LLM calls, priced via Pricing.price_map.PriceMap. None when no component's (provider, model) could be priced at all — see Composer/evaluation.py.

is_successful: bool property

Check if trial completed successfully (no error).

to_dict() -> Dict[str, Any]

Convert to dictionary for serialization.

from_dict(data: Dict[str, Any]) -> TrialResult classmethod

Create TrialResult from dictionary.

to_json() -> str

Serialize to JSON string.

from_json(json_str: str) -> TrialResult classmethod

Deserialize from JSON string.

To export a winning trial into standalone, executable Python code or a deployable project, see Export a trial to Python.

LLMProvider

Creates a LangChain chat model from a provider, model, credential, and optional endpoint. Use it directly when you need an LLM without constructing a RAG pipeline. See the LLM provider guide for provider recipes and configuration.

LLMProvider.LLMProvider.LLMProvider

Facade for building LangChain chat models from a unified configuration.

Validates arguments, dispatches to the concrete provider builder (Groq / Together / OpenRouter / OpenAI / Anthropic / Gemini / Ollama / Azure OpenAI / Custom OpenAI-compatible), and exposes get_llm() / call() over the resulting model. Heavy SDK imports happen lazily inside the concrete builders, so optional dependencies stay optional.

get_llm() -> Any

Get the underlying LLM client.

call(*args: Any, **kwargs: Any) -> Any

Proxy method to invoke the LLM.

Prefers invoke (the LangChain standard interface) and falls back to calling the model object directly for legacy clients.

get_usage_totals() -> dict

Sum of prompt/completion/total tokens recorded across every call so far.

get_per_sample_usage_totals() -> dict

Sum of prompt/completion/total tokens per (trial_id, sample_index).

get_sample_usage_totals(trial_id: int, sample_index: int) -> dict

Convenience accessor: usage totals for one sample, or all-zero if unseen.

reset_usage() -> None

Clear recorded token-usage history for this provider instance.

LLMProvider.LLMProvider.create_llm_provider(provider: Union[ProviderName, str] = 'openai', api_key: Optional[str] = None, model: str = 'gpt-4o-mini', temperature: float = 0.5, max_tokens: int = 300, base_url: Optional[str] = None, **kwargs: Any) -> LLMProvider

Factory function to easily instantiate an LLMProvider.

DocumentParser

Factory and standardized models for document parsing, layout analysis, and OCR. See the Document parsers guide for detailed provider recipes and configuration.

DocumentParser.create_document_parser(provider: str, **kwargs: Any) -> BaseDocumentParser

Factory function to initialize the requested document parser provider.

Parameters:

Name	Type	Description	Default
provider	str	Provider name (e.g. 'azure', 'docling', 'llama_parse'). Aliases 'llamaparse' and 'llama-parse' are also accepted.	required
**kwargs	Any	Provider-specific configuration arguments.	{}

Returns:

Name	Type	Description
BaseDocumentParser	BaseDocumentParser	Instantiated document parser.

Raises:

Type	Description
ConfigurationError	if provider is unknown or unsupported (also a ValueError subclass, for backward compatibility).

DocumentParser.models.ParsedDocument dataclass

Standardized output from any document parser. This is the SINGLE contract between parsing and the rest of the SDK.

Text contract: text is the concatenation of all page/section texts joined by "\n\n". Every concrete parser must comply with this so downstream chunking sees consistent whitespace regardless of provider.

__post_init__() -> None

Warn (do not raise) when the parsed text is empty — usually indicates a parse problem.

DocumentParser.models.PageContent dataclass

Represents a single page of parsed content.

DocumentParser.base.BaseDocumentParser

Bases: ABC

Abstract base for all document parsing/OCR providers.

Contract for concrete providers:

• parse_file(file_path) must raise FileNotFoundError if the path does not exist, and return a fully-populated ParsedDocument otherwise.
• parse_directory(directory_path) must return a list of successfully parsed ParsedDocument objects. Per-file failures must be logged and summarized at the end of the batch (the list must still be returned so partial progress is not lost).
• supported_extensions() returns lowercase, dot-prefixed extensions (e.g. [".pdf", ".txt"]).
• The shared ParsedDocument.text contract joins pages/sections with "\n\n"; concrete providers must comply.
• Both methods accept an optional base_dir keyword: when given, the resolved path must stay within it or a ConfigurationError is raised (path-traversal containment). Defaults to None (no containment check), preserving existing behavior for callers that don't opt in — intended for callers that expose these parsers over an API boundary.

Library note: concrete providers must NEVER call logging.basicConfig — that mutates the host application's root logger. Use getLogger(__name__) only.

parse_file(file_path: str) -> ParsedDocument abstractmethod

Parse a single file and return a :class:ParsedDocument.

Raises:

Type	Description
FileNotFoundError	if file_path does not exist.

parse_directory(directory_path: str) -> List[ParsedDocument] abstractmethod

Parse all supported files in a directory.

Returns the list of successfully parsed documents; per-file failures are logged and summarized but do not abort the batch.

supported_extensions() -> List[str] abstractmethod

Return list of (lowercase, dot-prefixed) file extensions this parser handles.

supports(file_path: str) -> bool

Return True if file_path has an extension this parser supports.

Centralizes the extension check so concrete providers do not reimplement it in parse_directory.

Embedding

Factory and base classes for local, cloud, and custom embedding providers with dual-layer caching and hardware acceleration. See the Embeddings guide for complete recipes.

Embedding.create_embedding_provider(provider: str = 'sentence_transformers', model_name: Optional[str] = None, api_key: Optional[str] = None, cache_dir: str = '.embedding_cache', batch_size: int = 32, custom_embeddings: Optional[Embeddings] = None, device: str = 'auto', **kwargs: Any) -> BaseEmbeddingProvider

Factory function to instantiate pluggable Embedding Providers.

Parameters:

Name	Type	Description	Default
provider	str	'sentence_transformers' / 'huggingface' / 'local', 'openai', 'cohere', or 'custom'.	'sentence_transformers'
model_name	str	Model identifier.	None
api_key	str	Provider API key.	None
cache_dir	str	Directory path for disk caching.	'.embedding_cache'
batch_size	int	Batch size for bulk document encoding.	32
custom_embeddings	Embeddings	Custom injected LangChain Embeddings instance.	None

Returns:

Name	Type	Description
BaseEmbeddingProvider	BaseEmbeddingProvider	An instance inheriting from BaseEmbeddingProvider.

Raises:

Type	Description
ValueError	if provider is unknown or empty.

Embedding.EmbeddingProvider

EmbeddingProvider(model_name: str = 'mohamed2811/Muffakir_Embedding', provider: str = 'sentence_transformers', api_key: Optional[str] = None, cache_dir: str = '.embedding_cache', batch_size: int = 32, custom_embeddings: Optional[Embeddings] = None, device: str = 'auto', **kwargs: Any) -> BaseEmbeddingProvider

Backward-compatible entry point for initializing embedding providers. Delegates directly to create_embedding_provider factory.

Embedding.base.BaseEmbeddingProvider

Bases: Embeddings, ABC

Abstract Base Class for Embedding Providers.

Provides two layers of caching across all providers

1. An in-memory LRU cache (bounded, thread-safe) for hot lookups.
2. An on-disk JSON cache for persistence across runs.

Concrete providers implement _embed_documents_raw and _embed_query_raw; this base handles caching, batching, and the LangChain Embeddings interface.

embed_query(text: str) -> List[float]

LangChain protocol: embed a single query (delegates to :meth:embed_single).

Every VectorDB backend's similarity_search() calls into this method (it's the embedding_function= handed to e.g. Chroma) -- timing it here, once, captures query-embedding time centrally without touching any of the call sites in RetrieveMethods.py or the concrete VectorDB providers. See Embedding.timing.EmbeddingTimingTracker.

embed_documents(texts: List[str]) -> List[List[float]]

LangChain protocol: embed a list of documents (delegates to :meth:embed).

embed_single(text: str) -> List[float]

Embed a single query with caching.

embed(texts: List[str]) -> List[List[float]]

Embed multiple documents with batching and caching.

Uncached texts are batched (size = self.batch_size) and sent to the provider's _embed_documents_raw; results are cached and merged back into the output list preserving the original order.

Pricing

LLM pricing catalog and cost estimation engine. See the Cost and pricing guide for usage recipes and custom overrides.

Pricing.price_map.PriceMap

Looks up per-token input/output pricing for a (provider, model) pair.

Lookup order in get_price: user-supplied provider-qualified key, legacy user-supplied model-only key, then the same two forms in the LiteLLM map. Provider-qualified keys prevent collisions when two APIs use the same model identifier, while model-only overrides remain supported.

load(url: str = DEFAULT_URL, timeout: float = 10.0) -> None

Fetch the litellm price map once. Never raises: on any failure, fetch_failed is set True and get_price/compute_cost will return None for anything not covered by a custom override.

get_price(provider: str, model: str) -> Optional[Dict[str, float]]

Return {"input_cost_per_token": ..., "output_cost_per_token": ...} or None.

compute_cost(provider: str, model: str, prompt_tokens: int, completion_tokens: int) -> Optional[float]

Return the dollar cost for the given token counts, or None if pricing is unknown.

to_dict() -> Dict[str, Any]

Serialize a picklable/JSON-able snapshot (for checkpoint storage / worker processes).

from_dict(snapshot: Dict[str, Any]) -> PriceMap classmethod

Rehydrate a PriceMap from a snapshot produced by to_dict() -- no network call.

PromptManager

Bilingual prompt template manager and placeholder validation engine. See the Prompt management guide for the complete prompt catalog and override recipes.

PromptManager.PromptManager.MuffakirPrompt

A centralized manager for all prompt templates used in the pipeline, supporting localized prompt loading, dynamic listing, and dynamic edits.

get_prompt(key: str) -> str

Get a specific prompt template by key.

update_prompt(key: str, template: str) -> None

Update or add a prompt template in-memory. Validates placeholders to warn the developer of potential omissions.

validate_prompt(key: str, template: str) -> None

Strictly validate template without changing this manager.

get_all_prompts(language: Optional[str] = None) -> Dict[str, str]

Return all prompts for the selected language. If language is specified and is different from self.language, load it dynamically on-demand.

print_all_prompts(language: Optional[str] = None) -> None

Print all loaded prompts for the selected language in a clear, formatted terminal layout.

QueryTransformer

Five pluggable strategies for rewriting or expanding a query before vector retrieval. See the Query transformation guide for strategy recipes and comparisons.

QueryTransformer.QueryTransformer.QueryTransformer

Main Orchestrator class for query transformation in Muffakir RAG.

Supports pluggable transformation strategies (Query Rewriting, Multi-Query Expansion, etc.) and integrates directly with LLMProvider and MuffakirPrompt.

name: str property

transform_query(original_query: str, conversation_history: Optional[List[Dict[str, str]]] = None) -> Union[str, List[str]]

Transform the input query into an optimized representation for vector search.

QueryTransformer.factory.create_query_transformer(strategy: str = 'rewrite', llm_provider: Optional[LLMProvider] = None, prompt_manager: Optional[MuffakirPrompt] = None, **kwargs: Any) -> BaseQueryTransformer

Factory function to initialize a Query Transformer strategy.

Parameters:

Name	Type	Description	Default
strategy	str	Strategy name ('rewrite', 'multi_query', 'decomposition', 'hyde', 'step_back', etc.). Default is 'rewrite'.	'rewrite'
llm_provider	LLMProvider	Instantiated LLM provider.	None
prompt_manager	MuffakirPrompt	Instantiated prompt manager.	None
**kwargs	Any	Strategy-specific parameters.	{}

Returns:

Name	Type	Description
BaseQueryTransformer	BaseQueryTransformer	Strategy instance.

Raises:

Type	Description
ValueError	If llm_provider is missing or strategy is unknown.

QueryTransformer.base.BaseQueryTransformer

Bases: ABC

Abstract base class for all Query Transformation strategies. Supported strategies include Query Rewriting, Multi-Query Expansion, HyDE, etc.

name: str abstractmethod property

Human-readable strategy name (e.g. 'rewrite', 'multi_query').

transform(query: str, conversation_history: Optional[List[Dict[str, str]]] = None) -> Union[str, List[str]] abstractmethod

Transform the input query into an optimized representation (or list of representations) for RAG vector database retrieval.

Parameters:

Name	Type	Description	Default
query	str	The raw input user query.	required
conversation_history	list of dict	Past conversation messages e.g. [{'role': 'user', 'content': '...'}, {'role': 'assistant', 'content': '...'}]	None

Returns:

Type	Description
Union[str, List[str]]	Union[str, List[str]]: Transformed query string or list of query strings.

QueryTransformer.rewriter.QueryRewriter

Bases: BaseQueryTransformer

Query Rewriting Strategy.

Transforms a noisy, contextual, or poorly phrased user message into a clean, direct, keyword-rich standalone query optimized for vector database retrieval.

Features: - Temperature = 0.0 for predictable, deterministic rewrites. - Prompts loaded directly from MuffakirPrompt (PromptManager). - Fully bilingual (Arabic & English). - Robust fallback to original query on exception.

transform(query: str, conversation_history: Optional[List[Dict[str, str]]] = None) -> str

QueryTransformer.multi_query.MultiQueryExpansion

Bases: BaseQueryTransformer

Multi-Query Expansion Strategy.

Expands a single user query into 3 to 5 distinct variations using synonyms, different phrasing, and related terminology to optimize vector database retrieval.

Features: - Temperature = 0.2 for slight creative variance in synonyms. - Pydantic structured output with robust fallback parser. - Ensures the original query is always preserved as the first element. - Prompts loaded directly from MuffakirPrompt (PromptManager). - Fully bilingual (Arabic & English).

transform(query: str, conversation_history: Optional[List[Dict[str, str]]] = None) -> List[str]

QueryTransformer.query_decomposition.QueryDecomposition

Bases: BaseQueryTransformer

Query Decomposition Strategy.

Splits a complex, compound, or multi-step user query into distinct, independent sub-queries optimized for isolated vector database retrieval.

Features: - Temperature = 0.0 for deterministic, precise splitting logic. - Pydantic structured output with robust fallback line parser. - Preserves simple queries as single-item lists without over-decomposing. - Prompts loaded directly from MuffakirPrompt (PromptManager). - Fully bilingual (Arabic & English).

transform(query: str, conversation_history: Optional[List[Dict[str, str]]] = None) -> List[str]

QueryTransformer.hyde.HyDEQueryTransformer

Bases: BaseQueryTransformer

Hypothetical Document Embeddings (HyDE) Strategy.

Generates a plausible, hypothetical answer or document snippet for the user query. The hypothetical document is then used as the query text for vector database embedding search, bridging the structural gap between questions and answer passages.

Features: - Temperature = 0.3 for natural document formatting and phrasing variance. - Prompts loaded directly from MuffakirPrompt (PromptManager). - Fully bilingual (Arabic & English). - Robust fallback to original query string on exception.

transform(query: str, conversation_history: Optional[List[Dict[str, str]]] = None) -> str

QueryTransformer.step_back.StepBackQueryTransformer

Bases: BaseQueryTransformer

Step-Back Prompting Strategy.

Takes a highly technical or specific user query and generates a broader, high-level 'step-back' question to retrieve foundational domain concepts along with specific facts.

Features: - Temperature = 0.0 for deterministic abstraction logic. - Pydantic structured output with fallback parser. - Returns a list containing both the original specific query and the step-back query: [original_query, step_back_query] so vector retrieval searches for both concepts. - Prompts loaded directly from MuffakirPrompt (PromptManager). - Fully bilingual (Arabic & English).

transform(query: str, conversation_history: Optional[List[Dict[str, str]]] = None) -> List[str]

Reranker

Six reranking strategies accessible through a uniform interface and open registry. See the Reranking guide for strategy recipes and comparisons.

Reranker.Reranker.Reranker

Backward-compatible Reranker wrapper. Delegates to create_reranker() factory and wraps as BaseReranker instance.

rerank(query: str, documents, top_k=None)

Reranker.factory.create_reranker(method: str = 'semantic_similarity', embedding_provider: Optional[Any] = None, llm_provider: Optional[Any] = None, prompt_manager: Optional[Any] = None, model_name: Optional[str] = None, **kwargs: Any) -> BaseReranker

Instantiate a registered reranking strategy.

Reranker.factory.register_reranker(name: str, factory: RerankerFactory, *, label: Optional[str] = None, description: str = '', aliases: Tuple[str, ...] = (), dependency: Optional[str] = None, configuration: str = 'none', replace: bool = False) -> None

Register a reranker factory for SDK users and extensions.

Factories receive the same keyword context as :func:create_reranker. Names and aliases are normalized to lowercase.

Reranker.factory.list_reranker_specs() -> Tuple[RerankerSpec, ...]

Return registered strategies in stable registration order.

Reranker.base.BaseReranker

Bases: ABC

Abstract Base Class for all Reranking strategies in Muffakir RAG.

score(query: str, documents: List[Document]) -> List[Tuple[Document, float]] abstractmethod

Score and rank documents against a query.

Parameters:

Name	Type	Description	Default
query	str	The user query.	required
documents	List[Document]	Candidate documents to rank.	required

Returns:

Type	Description
List[Tuple[Document, float]]	List[Tuple[Document, float]]: Documents with scores, sorted descending.

rerank(query: str, documents: List[Document], top_k: Optional[int] = None) -> List[Document]

Rerank documents and return the top-k most relevant ones.

Parameters:

Name	Type	Description	Default
query	str	The user query.	required
documents	List[Document]	Retrieved documents to rerank.	required
top_k	int	Number of documents to return. Defaults to all.	None

Returns:

Type	Description
List[Document]	List[Document]: Reranked documents.

Reranker.semantic_similarity.SemanticSimilarityReranker

Bases: BaseReranker

Semantic Similarity Reranker. Scores documents using cosine similarity between query and document embeddings.

score(query: str, documents: List[Document]) -> List[Tuple[Document, float]]

Reranker.bm25.BM25Reranker

Bases: BaseReranker

BM25 (Okapi BM25) Reranker. Ranks documents using sparse keyword frequency scoring. Requires: rank_bm25 (pip install rank-bm25).

score(query: str, documents: List[Document]) -> List[Tuple[Document, float]]

Reranker.cross_encoder.CrossEncoderReranker

Bases: BaseReranker

Cross-Encoder Reranker. Uses a bi-directional cross-encoder model to jointly score (query, document) pairs. This is the highest quality reranking method — the model reads both query and document together, producing a precise relevance score.

Default model: 'BAAI/bge-reranker-base' (backward compatible). Requires: sentence_transformers (pip install sentence-transformers).

score(query: str, documents: List[Document]) -> List[Tuple[Document, float]]

Reranker.pointwise.PointwiseReranker

Bases: BaseReranker

Pointwise Learning-to-Rank (L2R) Reranker.

Each document is scored independently against the query using a Cross-Encoder as the scoring function. This is a practical, training-free implementation of pointwise L2R where the relevance score is the cross-encoder output probability.

Optionally supports a relevance threshold to filter out irrelevant documents (score below threshold are treated as irrelevant and placed at the end).

Default model: 'BAAI/bge-reranker-base' (backward compatible). Requires: sentence_transformers (pip install sentence-transformers).

score(query: str, documents: List[Document]) -> List[Tuple[Document, float]]

Score each document independently against the query. Sigmoid activation ensures scores are in [0.0, 1.0] probability range.

Reranker.llm.LLMReranker

Bases: BaseReranker

LLM-Based Reranker using Pydantic Structured Output.

Uses the configured LLM to independently score each (query, document) pair on a continuous relevance scale of 0.0 to 1.0.

Architecture: - Primary: Uses llm.with_structured_output(RerankerScore) for JSON-enforced structured output. This works with OpenAI, Anthropic, Groq, and other function-calling capable providers. - Fallback: If the provider does not support structured output, falls back to plain text generation + regex score extraction.

The prompt is loaded from PromptManager (reranker_scoring key), ensuring full bilingual support (Arabic & English). Temperature is forced to 0.0 for deterministic, consistent scoring.

Why continuous score vs. binary (yes/no)? - Binary grading (yes/no) is appropriate for filtering decisions (e.g., hallucination checking, context relevance). - Continuous scoring (0.0–1.0) is required for reranking because we need to sort documents by relative relevance, not just keep/discard them.

score(query: str, documents: List[Document]) -> List[Tuple[Document, float]]

Score each document independently against the query. Prefers structured output; falls back to regex parsing.

Reranker.remote.RemoteReranker

Bases: BaseReranker

Rerank through a Cohere-compatible, index-based HTTP response.

score(query: str, documents: List[Document]) -> List[Tuple[Document, float]]

MuffakirSearch

Search RAG facade integrating pluggable web search engines (Tavily, Firecrawl, SerpAPI) with LLM answer generation for zero-corpus search and QA. See Adaptive web search for full configuration recipes, evaluation, and telemetry.

Muffakir.MuffakirSearch.MuffakirSearch

Search RAG facade: integrates pluggable WebSearch providers (Firecrawl, Tavily, SerpAPI) with LLM answer generation.

search(query: str) -> Dict[str, Any]

Perform a search using the configured web search provider and LLM.

ask(question: str, **kwargs) -> Dict[str, Any]

EvalRunner-compatible adapter (matches MuffakirRAG.ask()'s call and return shape) so a web-search-only run can be scored by the existing evaluation pipeline unchanged. k/retrieval_method kwargs are accepted for interface compatibility but unused — there is no local retrieval in web-search-only mode.

get_similar_documents(query: str, k: Optional[int] = None) -> List[Any]

Web-search-only mode has no local retrieval corpus — retrieval metrics (recall/precision/mrr/ndcg) are not meaningful here. Fail loudly with a clear message rather than an ambiguous AttributeError.

get_config() -> Dict[str, Any]

Get current configuration.

WebSearch

Pluggable web search engine factory, provider base classes, standardized result models, and concrete search backends. See the Adaptive web search guide for provider comparison, parameter details, and custom backend development.

WebSearch.factory.create_web_search_provider(provider: str = 'firecrawl', **kwargs: Any) -> BaseWebSearchProvider

Factory function to instantiate pluggable Web Search providers.

Parameters:

Name	Type	Description	Default
provider	str	'firecrawl', 'tavily', or 'serpapi'.	'firecrawl'
**kwargs	Any	Provider-specific configuration (api_key, max_results, max_depth, etc.).	{}

Returns:

Name	Type	Description
BaseWebSearchProvider	BaseWebSearchProvider	An instance inheriting from BaseWebSearchProvider.

WebSearch.base.BaseWebSearchProvider

Bases: ABC

Abstract base for all web search / scraping providers.

search(query: str) -> WebSearchResult abstractmethod

Run a web search for the given query and return normalized results.

WebSearch.models.WebSearchResult dataclass

Standardized output from any web search provider. This is the SINGLE contract between search providers and the Search orchestrator.

WebSearch.tavily.TavilyWebSearchProvider

Bases: BaseWebSearchProvider

Tavily AI-optimized web search provider.

search(query: str) -> WebSearchResult

WebSearch.firecrawl.FirecrawlWebSearchProvider

Bases: BaseWebSearchProvider

Firecrawl deep-research provider. Uses FirecrawlApp.deep_research to gather web content and sources.

search(query: str) -> WebSearchResult

WebSearch.serpapi.SerpAPIWebSearchProvider

Bases: BaseWebSearchProvider

SerpAPI Google search provider.

search(query: str) -> WebSearchResult

SyntheticData

Generates Q&A evaluation datasets from a document corpus using an LLM. See the Synthetic data guide for configuration and workflow details.

Muffakir.MuffakirSyntheticData.MuffakirSyntheticData

Backward-compatible wrapper for MuffakirSyntheticData. Maintains exact original interface while delegating to SyntheticDataPipeline.

generate_dataset(custom_prompt: Optional[str] = None, max_chunks: Optional[int] = None) -> Any

Generate synthetic dataset and return pandas DataFrame.

SyntheticData.pipeline.SyntheticDataPipeline

Main Orchestrator for Synthetic Data Generation in Muffakir RAG.

Coordinates document parsing, chunking, LLM Q&A generation with retry logic, Pydantic schema validation, periodic checkpointing, and dataset export.

run(custom_prompt: Optional[str] = None, max_chunks: Optional[int] = None) -> Tuple[pd.DataFrame, GenerationStats]

Run the complete synthetic dataset generation pipeline.

Returns:

Type	Description
Tuple[DataFrame, GenerationStats]	Tuple[pd.DataFrame, GenerationStats]: The generated DataFrame and run stats.

SyntheticData.models.QAPair

Bases: BaseModel

A single validated Q&A pair generated from a document chunk.

SyntheticData.models.GenerationStats

Bases: BaseModel

Statistics returned after synthetic dataset generation.

SyntheticData.models.SyntheticDataConfig

Bases: BaseModel

Validated configuration model for Synthetic Data generation.

Trace

Observability helpers, trace schema dataclasses, and the TraceWriter that records per-sample pipeline traces during Composer runs. See Traces and observability for the full guide to trace file layout, field reference, and diagnostic patterns.

Observability helpers

Trace.observability.observation_context(queue: Any, *, trial_id: Optional[int] = None, sample_index: Optional[int] = None, attempt: Optional[int] = None, secrets: Optional[Iterable[Optional[str]]] = None) -> Iterator[None]

Install queue and attribution metadata for the current thread.

Trace.observability.observe_stage(stage: str, component: str, *, provider: Optional[str] = None, model: Optional[str] = None, defer_propagated_error: bool = True) -> StageObservation

Trace.observability.emit_stage_outcome(stage: str, component: str, *, outcome: str = 'success', duration_ms: Optional[float] = None, recovery: Optional[str] = None, provider: Optional[str] = None, model: Optional[str] = None, error: Optional[BaseException] = None) -> None

Emit one completed stage attempt to the active trace queue.

Trace.observability.mark_current_stage_error(error: BaseException, recovery: str = 'fallback') -> None

Mark a caught problem as the current stage's single recovered outcome.

Trace.observability.record_exception_outcome(error: BaseException, *, recovery: str, fallback_stage: str = 'trial_execution', fallback_component: str = 'executor') -> None

Finalize a propagated exception once retry/fatal disposition is known.

Trace.observability.sanitize_error_message(value: Any, secrets: Optional[Iterable[str]] = None) -> str

Return a bounded diagnostic message with common credential shapes removed.

Trace schema

Trace.models.RunManifest dataclass

Trace.models.TrialRecord dataclass

Trace.models.SampleTraceRecord dataclass

Storage

Trace.writer.TraceWriter

Drains a queue (plain queue.Queue, or a multiprocessing.Manager().Queue() shared with ProcessPoolExecutor workers) on a single background thread.

Usage

writer = TraceWriter(trace_dir, work_queue=optional_manager_queue) writer.write_manifest(manifest.to_dict()) writer.write_trial_started(active_trial_dict) writer.write_trial(trial_record.to_dict()) # can also be called writer.write_sample(sample_record.to_dict()) # from a worker process ... writer.close()

write_manifest(manifest: Dict[str, Any]) -> None

write_trial(record: Dict[str, Any]) -> None

write_trial_started(record: Dict[str, Any]) -> None

write_sample(record: Dict[str, Any]) -> None

write_operation(record: Dict[str, Any]) -> None

close(timeout: float = 30.0) -> None

Drain remaining queued records and stop the background thread.