RankifyAgent API

Intelligent model recommendation system for Rankify.

Overview

RankifyAgent provides: - Programmatic recommendations via recommend() and RankifyRecommender - Conversational interface via RankifyAgent class - Model registry with metadata for all available models

Quick Example

from rankify.agent import recommend, RankifyAgent

# Quick recommendation
result = recommend(task="qa", gpu=True)
print(result.retriever.name)  # "BGE"

# Conversational agent
agent = RankifyAgent(backend="azure")
response = agent.chat("I need a fast search system")
print(response.message)

API Reference

rankify.agent

RankifyAgent - Intelligent Model Recommendation System for Rankify.

This module provides: - RankifyAgent: Conversational AI assistant for model selection - RankifyRecommender: Programmatic recommendation API - Model Registry: Metadata for all available models

Example Usage

from rankify.agent import RankifyAgent, recommend

# Quick recommendation
result = recommend(task="qa", gpu=True)
print(result.retriever.name)
print(result.reranker.name)

# Conversational interface
agent = RankifyAgent(backend="azure")
response = agent.chat("I need a fast search system for production")
print(response.message)
print(response.code_snippet)

RETRIEVER_REGISTRY = {'bm25': ModelMetadata(name='BM25', method='bm25', description='Classic sparse retrieval using BM25 algorithm. Fast, no GPU needed.', speed=(Speed.VERY_FAST), accuracy=(Accuracy.GOOD), gpu_required=False, memory_mb=500, best_for=['keyword search', 'exact match', 'large corpus', 'low latency'], languages=['en', 'multilingual']), 'dpr-multi': ModelMetadata(name='DPR (Multi-Encoder)', method='dpr', description='Dense Passage Retrieval with separate query/passage encoders.', speed=(Speed.MEDIUM), accuracy=(Accuracy.VERY_GOOD), gpu_required=True, memory_mb=2000, best_for=['semantic search', 'qa', 'open-domain qa'], model_path='facebook-dpr-ctx_encoder-multiset-base'), 'dpr-single': ModelMetadata(name='DPR (Single-Encoder)', method='dpr', description='DPR with single encoder for both query and passage.', speed=(Speed.FAST), accuracy=(Accuracy.GOOD), gpu_required=True, memory_mb=1500, best_for=['semantic search', 'faster inference'], model_path='facebook-dpr-question_encoder-single-nq-base'), 'ance': ModelMetadata(name='ANCE', method='ance', description='Approximate Nearest Neighbor Negative Contrastive Estimation.', speed=(Speed.MEDIUM), accuracy=(Accuracy.EXCELLENT), gpu_required=True, memory_mb=2500, best_for=['high accuracy retrieval', 'qa', 'passage ranking'], model_path='castorini/ance-msmarco-passage'), 'bge': ModelMetadata(name='BGE', method='bge', description='BAAI General Embedding - state-of-the-art dense retriever.', speed=(Speed.FAST), accuracy=(Accuracy.STATE_OF_THE_ART), gpu_required=True, memory_mb=1500, best_for=['semantic search', 'qa', 'high accuracy', 'multilingual'], languages=['en', 'zh', 'multilingual'], model_path='BAAI/bge-base-en-v1.5'), 'colbert': ModelMetadata(name='ColBERT', method='colbert', description='Contextualized Late Interaction over BERT - token-level matching.', speed=(Speed.SLOW), accuracy=(Accuracy.STATE_OF_THE_ART), gpu_required=True, memory_mb=4000, best_for=['high precision', 'passage retrieval', 'fine-grained matching'], model_path='colbert-ir/colbertv2.0'), 'contriever': ModelMetadata(name='Contriever', method='contriever', description='Unsupervised dense retriever, good zero-shot performance.', speed=(Speed.MEDIUM), accuracy=(Accuracy.VERY_GOOD), gpu_required=True, memory_mb=1800, best_for=['zero-shot retrieval', 'domain transfer', 'no training data'], model_path='facebook/contriever-msmarco'), 'hyde': ModelMetadata(name='HyDE (Hypothetical Document Embeddings)', method='hyde', description='Generates hypothetical documents to improve retrieval.', speed=(Speed.SLOW), accuracy=(Accuracy.EXCELLENT), gpu_required=True, memory_mb=3000, api_required=True, best_for=['complex queries', 'sparse training data', 'query expansion']), 'online': ModelMetadata(name='Online Retriever', method='online', description='Web search-based retrieval using search APIs.', speed=(Speed.MEDIUM), accuracy=(Accuracy.VERY_GOOD), gpu_required=False, memory_mb=100, api_required=True, api_provider='serper', best_for=['real-time data', 'current events', 'web search']), 'diver-bge': ModelMetadata(name='Diver (BGE Large)', method='diver-dense', description='BAAI/bge-large-en-v1.5 via the Diver dense retrieval framework.', speed=(Speed.MEDIUM), accuracy=(Accuracy.STATE_OF_THE_ART), gpu_required=True, memory_mb=3000, best_for=['semantic search', 'high accuracy', 'BEIR benchmarks'], model_path='bge'), 'diver-sbert': ModelMetadata(name='Diver (SBERT all-mpnet-base-v2)', method='diver-dense', description='sentence-transformers/all-mpnet-base-v2 via the Diver framework.', speed=(Speed.FAST), accuracy=(Accuracy.VERY_GOOD), gpu_required=True, memory_mb=1500, best_for=['sentence similarity', 'semantic search'], model_path='sbert'), 'diver-nomic': ModelMetadata(name='Diver (Nomic Embed)', method='diver-dense', description='nomic-ai/nomic-embed-text-v1 via the Diver framework.', speed=(Speed.FAST), accuracy=(Accuracy.VERY_GOOD), gpu_required=True, memory_mb=1500, best_for=['long context', 'semantic search', 'document retrieval'], model_path='nomic'), 'diver-e5': ModelMetadata(name='Diver (E5-Mistral-7B)', method='diver-dense', description='intfloat/e5-mistral-7b-instruct — instruction-tuned LLM encoder in the Diver framework.', speed=(Speed.SLOW), accuracy=(Accuracy.STATE_OF_THE_ART), gpu_required=True, memory_mb=16000, best_for=['zero-shot retrieval', 'instruction following', 'complex queries'], model_path='e5'), 'diver-sf': ModelMetadata(name='Diver (SFR-Embedding-Mistral)', method='diver-dense', description='Salesforce/SFR-Embedding-Mistral — Salesforce Mistral-based bi-encoder in the Diver framework.', speed=(Speed.SLOW), accuracy=(Accuracy.STATE_OF_THE_ART), gpu_required=True, memory_mb=16000, best_for=['high accuracy retrieval', 'complex queries', 'BEIR benchmarks'], model_path='sf'), 'diver-rader': ModelMetadata(name='Diver (RaDeR)', method='diver-dense', description='Raderspace/RaDeR_Qwen_25_7B — reasoning-aware dense retriever in the Diver framework.', speed=(Speed.SLOW), accuracy=(Accuracy.EXCELLENT), gpu_required=True, memory_mb=16000, best_for=['reasoning-intensive queries', 'multi-hop QA', 'math-related retrieval'], model_path='rader'), 'diver-grit': ModelMetadata(name='Diver (GritLM-7B)', method='diver-dense', description='GritLM/GritLM-7B — generative representation model in the Diver framework.', speed=(Speed.VERY_SLOW), accuracy=(Accuracy.STATE_OF_THE_ART), gpu_required=True, memory_mb=16000, best_for=['generative retrieval', 'LLM-quality embeddings', 'long context'], model_path='grit'), 'diver-model': ModelMetadata(name='Diver Retriever-4B', method='diver-dense', description='AQ-MedAI/Diver-Retriever-4B — the flagship Diver diverse-evidence retrieval model.', speed=(Speed.SLOW), accuracy=(Accuracy.STATE_OF_THE_ART), gpu_required=True, memory_mb=8000, best_for=['diverse evidence retrieval', 'BEIR benchmarks', 'medical QA'], model_path='diver'), 'diver-inst-l': ModelMetadata(name='Diver (Instructor-Large)', method='diver-dense', description='hkunlp/instructor-large — instruction-following encoder in the Diver framework.', speed=(Speed.MEDIUM), accuracy=(Accuracy.VERY_GOOD), gpu_required=True, memory_mb=3000, best_for=['instruction following', 'domain-specific retrieval'], model_path='inst-l'), 'diver-m2': ModelMetadata(name='Diver (M2-BERT-32K)', method='diver-dense', description='togethercomputer/m2-bert-80M-32k-retrieval — long-context retrieval in the Diver framework.', speed=(Speed.MEDIUM), accuracy=(Accuracy.VERY_GOOD), gpu_required=True, memory_mb=2000, best_for=['long-context retrieval', '32k sequence length'], model_path='m2'), 'reasonir': ModelMetadata(name='ReasonIR-8B', method='reasonir', description='reasonir/ReasonIR-8B — SOTA reasoning-intensive retriever on the BRIGHT benchmark. No model_id needed.', speed=(Speed.VERY_SLOW), accuracy=(Accuracy.STATE_OF_THE_ART), gpu_required=True, memory_mb=16000, best_for=['reasoning-intensive queries', 'BRIGHT benchmark', 'complex multi-hop QA', 'science queries']), 'reason-embed-qwen3-8b': ModelMetadata(name='ReasonEmbed Qwen3-8B', method='reason-embed', description="hanhainebula/reason-embed-qwen3-8b-0928 — Qwen3-8B for reasoning retrieval. Use model_id='qwen3-8b'.", speed=(Speed.VERY_SLOW), accuracy=(Accuracy.STATE_OF_THE_ART), gpu_required=True, memory_mb=16000, best_for=['reasoning-intensive retrieval', 'complex queries'], model_path='qwen3-8b'), 'reason-embed-qwen3-4b': ModelMetadata(name='ReasonEmbed Qwen3-4B', method='reason-embed', description="hanhainebula/reason-embed-qwen3-4b-0928 — balanced Qwen3-4B for reasoning retrieval. Use model_id='qwen3-4b'.", speed=(Speed.SLOW), accuracy=(Accuracy.EXCELLENT), gpu_required=True, memory_mb=8000, best_for=['reasoning retrieval', 'balanced accuracy/speed'], model_path='qwen3-4b'), 'reason-embed-llama-8b': ModelMetadata(name='ReasonEmbed LLaMA-3.1-8B', method='reason-embed', description="hanhainebula/reason-embed-llama-3.1-8b-0928 — LLaMA-3.1-8B for reasoning retrieval. Use model_id='llama-8b'.", speed=(Speed.SLOW), accuracy=(Accuracy.EXCELLENT), gpu_required=True, memory_mb=16000, best_for=['reasoning retrieval', 'open-source LLaMA backbone'], model_path='llama-8b'), 'bge-reasoner-embed': ModelMetadata(name='BGE Reasoner Embed (Qwen3-8B)', method='bge-reasoner-embed', description='BAAI/bge-reasoner-embed-qwen3-8b-0923 — BGE reasoning-augmented retriever. No model_id needed.', speed=(Speed.SLOW), accuracy=(Accuracy.EXCELLENT), gpu_required=True, memory_mb=16000, best_for=['reasoning-augmented retrieval', 'BEIR benchmarks', 'complex queries'])} module-attribute

RERANKER_REGISTRY = {'monot5-base-msmarco': ModelMetadata(name='MonoT5 Base', method='monot5', description='T5-based pointwise reranker, trained on MS MARCO.', speed=(Speed.MEDIUM), accuracy=(Accuracy.EXCELLENT), gpu_required=True, memory_mb=900, best_for=['qa', 'general reranking', 'passage ranking'], model_path='castorini/monot5-base-msmarco'), 'monot5-3b-msmarco': ModelMetadata(name='MonoT5 3B', method='monot5', description='Large MonoT5 reranker for highest accuracy.', speed=(Speed.SLOW), accuracy=(Accuracy.STATE_OF_THE_ART), gpu_required=True, memory_mb=12000, best_for=['highest accuracy', 'offline processing'], model_path='castorini/monot5-3b-msmarco'), 'monobert': ModelMetadata(name='MonoBERT', method='monobert', description='BERT-based pointwise reranker.', speed=(Speed.MEDIUM), accuracy=(Accuracy.VERY_GOOD), gpu_required=True, memory_mb=1500, best_for=['qa', 'document ranking'], model_path='castorini/monobert-large-msmarco'), 'flashrank-minilm': ModelMetadata(name='FlashRank MiniLM', method='flashrank', description='Ultra-fast ONNX-based reranker, CPU-friendly.', speed=(Speed.VERY_FAST), accuracy=(Accuracy.GOOD), gpu_required=False, memory_mb=50, best_for=['low latency', 'cpu deployment', 'edge devices', 'production'], model_path='ms-marco-MiniLM-L-12-v2'), 'flashrank-tinybert': ModelMetadata(name='FlashRank TinyBERT', method='flashrank', description='Smallest FlashRank model, fastest inference.', speed=(Speed.VERY_FAST), accuracy=(Accuracy.BASIC), gpu_required=False, memory_mb=20, best_for=['ultra-low latency', 'mobile', 'embedded'], model_path='ms-marco-TinyBERT-L-2-v2'), 'upr-t5-base': ModelMetadata(name='UPR (T5 Base)', method='upr', description='Unsupervised Passage Reranker using T5.', speed=(Speed.MEDIUM), accuracy=(Accuracy.GOOD), gpu_required=True, memory_mb=900, best_for=['zero-shot reranking', 'no training data'], model_path='t5-base'), 'rankgpt-gpt4': ModelMetadata(name='RankGPT (GPT-4)', method='rankgpt-api', description='LLM-based listwise reranking using GPT-4.', speed=(Speed.SLOW), accuracy=(Accuracy.STATE_OF_THE_ART), gpu_required=False, memory_mb=100, api_required=True, api_provider='openai', best_for=['highest accuracy', 'complex queries', 'reasoning-based ranking'], model_path='gpt-4o'), 'rankgpt-llama': ModelMetadata(name='RankGPT (LLaMA)', method='rankgpt', description='Local LLM-based listwise reranking using LLaMA.', speed=(Speed.VERY_SLOW), accuracy=(Accuracy.EXCELLENT), gpu_required=True, memory_mb=16000, best_for=['high accuracy', 'privacy', 'no api costs'], model_path='meta-llama/Meta-Llama-3.1-8B-Instruct'), 'inranker-base': ModelMetadata(name='InRanker Base', method='inranker', description='In-context reranking with instruction-tuned models.', speed=(Speed.MEDIUM), accuracy=(Accuracy.VERY_GOOD), gpu_required=True, memory_mb=1500, best_for=['instruction following', 'zero-shot'], model_path='unicamp-dl/InRanker-base'), 'echorank': ModelMetadata(name='EchoRank', method='echorank', description='Pairwise reranking with echo-based verification.', speed=(Speed.MEDIUM), accuracy=(Accuracy.VERY_GOOD), gpu_required=True, memory_mb=2000, best_for=['verification', 'fact checking']), 'rankt5-base': ModelMetadata(name='RankT5 Base', method='rankt5', description='T5-based listwise reranker.', speed=(Speed.MEDIUM), accuracy=(Accuracy.EXCELLENT), gpu_required=True, memory_mb=1500, best_for=['listwise ranking', 'multi-document'], model_path='Soyoung97/RankT5-base'), 'listt5-base': ModelMetadata(name='ListT5 Base', method='listt5', description='List-aware T5 reranker.', speed=(Speed.MEDIUM), accuracy=(Accuracy.EXCELLENT), gpu_required=True, memory_mb=1500, best_for=['list ranking', 'document sets'], model_path='Soyoung97/ListT5-base'), 'colbert-reranker': ModelMetadata(name='ColBERT Reranker', method='colbert_ranker', description='ColBERT-based reranking for fine-grained matching.', speed=(Speed.SLOW), accuracy=(Accuracy.STATE_OF_THE_ART), gpu_required=True, memory_mb=3000, best_for=['fine-grained matching', 'high precision']), 'transformer-reranker': ModelMetadata(name='Transformer Reranker', method='transformer_ranker', description='General transformer-based cross-encoder reranker.', speed=(Speed.MEDIUM), accuracy=(Accuracy.VERY_GOOD), gpu_required=True, memory_mb=1500, best_for=['general reranking', 'cross-encoder']), 'cohere-rerank': ModelMetadata(name='Cohere Reranker', method='apiranker', description="Cohere's reranking API - fast and accurate.", speed=(Speed.FAST), accuracy=(Accuracy.EXCELLENT), gpu_required=False, memory_mb=50, api_required=True, api_provider='cohere', best_for=['production', 'easy deployment', 'no gpu'], model_path='cohere'), 'jina-rerank': ModelMetadata(name='Jina Reranker', method='apiranker', description="Jina AI's reranking API.", speed=(Speed.FAST), accuracy=(Accuracy.VERY_GOOD), gpu_required=False, memory_mb=50, api_required=True, api_provider='jina', best_for=['production', 'multilingual'], model_path='jina'), 'voyage-rerank': ModelMetadata(name='Voyage Reranker', method='apiranker', description="Voyage AI's reranking API.", speed=(Speed.FAST), accuracy=(Accuracy.EXCELLENT), gpu_required=False, memory_mb=50, api_required=True, api_provider='voyage', best_for=['production', 'high quality'], model_path='voyage'), 'splade-reranker': ModelMetadata(name='SPLADE Reranker', method='splade_reranker', description='Sparse lexical and dense hybrid reranker.', speed=(Speed.FAST), accuracy=(Accuracy.VERY_GOOD), gpu_required=True, memory_mb=1200, best_for=['hybrid retrieval', 'keyword + semantic']), 'sentence-transformer-reranker': ModelMetadata(name='Sentence Transformer Reranker', method='sentence_transformer_reranker', description='Reranking with sentence transformer embeddings.', speed=(Speed.FAST), accuracy=(Accuracy.GOOD), gpu_required=True, memory_mb=1000, best_for=['semantic similarity', 'embedding-based']), 'llm2vec-reranker': ModelMetadata(name='LLM2Vec Reranker', method='llm2vec', description='LLM embeddings converted for reranking.', speed=(Speed.SLOW), accuracy=(Accuracy.EXCELLENT), gpu_required=True, memory_mb=8000, best_for=['llm-based', 'high accuracy']), 'twolar': ModelMetadata(name='TWOLAR', method='twolar', description='Two-stage list-aware reranking.', speed=(Speed.MEDIUM), accuracy=(Accuracy.EXCELLENT), gpu_required=True, memory_mb=2000, best_for=['list-aware', 'multi-stage'])} module-attribute

RAG_METHOD_REGISTRY = {'zero-shot': ModelMetadata(name='Zero-Shot RAG', method='zero-shot', description='Direct answer generation without examples.', speed=(Speed.FAST), accuracy=(Accuracy.GOOD), gpu_required=False, memory_mb=100, best_for=['simple qa', 'fast responses']), 'basic-rag': ModelMetadata(name='Basic RAG', method='basic-rag', description='Standard RAG with context injection.', speed=(Speed.FAST), accuracy=(Accuracy.VERY_GOOD), gpu_required=False, memory_mb=100, best_for=['general qa', 'document qa']), 'chain-of-thought-rag': ModelMetadata(name='Chain-of-Thought RAG', method='chain-of-thought-rag', description='RAG with step-by-step reasoning.', speed=(Speed.MEDIUM), accuracy=(Accuracy.EXCELLENT), gpu_required=False, memory_mb=100, best_for=['complex questions', 'multi-step reasoning', 'math']), 'self-consistency-rag': ModelMetadata(name='Self-Consistency RAG', method='self-consistency-rag', description='Multiple generations with majority voting.', speed=(Speed.SLOW), accuracy=(Accuracy.STATE_OF_THE_ART), gpu_required=False, memory_mb=100, best_for=['highest accuracy', 'critical applications', 'verification']), 'react-rag': ModelMetadata(name='ReAct RAG', method='react-rag', description='Reasoning and acting interleaved.', speed=(Speed.VERY_SLOW), accuracy=(Accuracy.EXCELLENT), gpu_required=False, memory_mb=100, best_for=['multi-hop qa', 'tool use', 'complex reasoning']), 'fid-rag': ModelMetadata(name='Fusion-in-Decoder RAG', method='fid-rag', description='All contexts fused in decoder at once.', speed=(Speed.MEDIUM), accuracy=(Accuracy.EXCELLENT), gpu_required=True, memory_mb=4000, best_for=['multi-document', 'information synthesis'], model_path='google/fid-nq-base'), 'in-context-ralm': ModelMetadata(name='In-Context RALM', method='in-context-ralm', description='In-context retrieval-augmented language modeling.', speed=(Speed.MEDIUM), accuracy=(Accuracy.EXCELLENT), gpu_required=False, memory_mb=100, best_for=['few-shot learning', 'context adaptation'])} module-attribute

__all__ = ['RankifyAgent', 'AgentResponse', 'RankifyRecommender', 'RecommendationResult', 'PipelineConfig', 'recommend', 'ModelMetadata', 'TaskType', 'Speed', 'Accuracy', 'RETRIEVER_REGISTRY', 'RERANKER_REGISTRY', 'RAG_METHOD_REGISTRY', 'get_all_retrievers', 'get_all_rerankers', 'get_all_rag_methods', 'get_model', 'filter_models'] module-attribute

ModelMetadata dataclass

Metadata for a single model.

Source code in rankify/agent/model_registry.py

@dataclass
class ModelMetadata:
    """Metadata for a single model."""
    name: str
    method: str
    description: str
    speed: Speed
    accuracy: Accuracy
    gpu_required: bool
    memory_mb: int
    best_for: List[str]
    languages: List[str] = field(default_factory=lambda: ["en"])
    api_required: bool = False
    api_provider: Optional[str] = None
    model_path: Optional[str] = None

    def matches_constraints(self, constraints: Dict[str, Any]) -> bool:
        """Check if model matches user constraints."""
        if constraints.get("gpu") is False and self.gpu_required:
            return False
        if constraints.get("max_memory_mb") and self.memory_mb > constraints["max_memory_mb"]:
            return False
        if constraints.get("api_only") is True and not self.api_required:
            return False
        if constraints.get("no_api") is True and self.api_required:
            return False
        if constraints.get("language") and constraints["language"] not in self.languages:
            return False
        return True

    def score_for_task(self, task: TaskType) -> float:
        """Score model suitability for a task (0-1)."""
        task_keywords = {
            TaskType.QUESTION_ANSWERING: ["qa", "question", "answer", "reading comprehension"],
            TaskType.SEARCH: ["search", "retrieval", "ranking", "information retrieval"],
            TaskType.SUMMARIZATION: ["summarization", "long context", "multi-document"],
            TaskType.CONVERSATIONAL: ["conversational", "dialogue", "chat"],
            TaskType.DOMAIN_SPECIFIC: ["domain", "specialized", "technical"],
        }
        keywords = task_keywords.get(task, [])
        matches = sum(1 for k in keywords if k in " ".join(self.best_for).lower())
        return min(1.0, matches / max(1, len(keywords)) + 0.3)  # Base score of 0.3

matches_constraints(constraints)

Check if model matches user constraints.

Source code in rankify/agent/model_registry.py

def matches_constraints(self, constraints: Dict[str, Any]) -> bool:
    """Check if model matches user constraints."""
    if constraints.get("gpu") is False and self.gpu_required:
        return False
    if constraints.get("max_memory_mb") and self.memory_mb > constraints["max_memory_mb"]:
        return False
    if constraints.get("api_only") is True and not self.api_required:
        return False
    if constraints.get("no_api") is True and self.api_required:
        return False
    if constraints.get("language") and constraints["language"] not in self.languages:
        return False
    return True

score_for_task(task)

Score model suitability for a task (0-1).

Source code in rankify/agent/model_registry.py

def score_for_task(self, task: TaskType) -> float:
    """Score model suitability for a task (0-1)."""
    task_keywords = {
        TaskType.QUESTION_ANSWERING: ["qa", "question", "answer", "reading comprehension"],
        TaskType.SEARCH: ["search", "retrieval", "ranking", "information retrieval"],
        TaskType.SUMMARIZATION: ["summarization", "long context", "multi-document"],
        TaskType.CONVERSATIONAL: ["conversational", "dialogue", "chat"],
        TaskType.DOMAIN_SPECIFIC: ["domain", "specialized", "technical"],
    }
    keywords = task_keywords.get(task, [])
    matches = sum(1 for k in keywords if k in " ".join(self.best_for).lower())
    return min(1.0, matches / max(1, len(keywords)) + 0.3)  # Base score of 0.3

TaskType

Bases: Enum

Supported task types.

Source code in rankify/agent/model_registry.py

class TaskType(Enum):
    """Supported task types."""
    QUESTION_ANSWERING = "qa"
    SEARCH = "search"
    SUMMARIZATION = "summarization"
    CONVERSATIONAL = "conversational"
    DOMAIN_SPECIFIC = "domain_specific"

Speed

Bases: Enum

Model speed categories.

Source code in rankify/agent/model_registry.py

class Speed(Enum):
    """Model speed categories."""
    VERY_FAST = "very_fast"  # < 10ms per query
    FAST = "fast"            # < 50ms per query
    MEDIUM = "medium"        # < 200ms per query
    SLOW = "slow"            # < 1s per query
    VERY_SLOW = "very_slow"  # > 1s per query

Accuracy

Bases: Enum

Model accuracy categories.

Source code in rankify/agent/model_registry.py

class Accuracy(Enum):
    """Model accuracy categories."""
    BASIC = "basic"
    GOOD = "good"
    VERY_GOOD = "very_good"
    EXCELLENT = "excellent"
    STATE_OF_THE_ART = "sota"

RankifyRecommender

Intelligent model recommender for Rankify.

Recommends optimal retrievers, rerankers, and RAG methods based on user constraints and task requirements.

Source code in rankify/agent/recommender.py

class RankifyRecommender:
    """
    Intelligent model recommender for Rankify.

    Recommends optimal retrievers, rerankers, and RAG methods
    based on user constraints and task requirements.
    """

    def __init__(self):
        self.retrievers = RETRIEVER_REGISTRY
        self.rerankers = RERANKER_REGISTRY
        self.rag_methods = RAG_METHOD_REGISTRY

    def recommend(
        self,
        task: str = "qa",
        constraints: Optional[Dict[str, Any]] = None,
        include_reranker: bool = True,
        include_rag: bool = False,
        top_k: int = 3,
    ) -> RecommendationResult:
        """
        Get recommended models for a given task and constraints.

        Args:
            task: Task type - "qa", "search", "summarization", "conversational", "domain_specific"
            constraints: Optional constraints dict with keys like:
                - gpu: bool - Whether GPU is available
                - max_memory_mb: int - Maximum memory budget
                - max_latency_ms: int - Maximum latency budget
                - api_only: bool - Only recommend API-based models
                - no_api: bool - Only recommend local models
                - language: str - Required language support
            include_reranker: Whether to recommend a reranker
            include_rag: Whether to recommend a RAG method
            top_k: Number of alternatives to include

        Returns:
            RecommendationResult with recommended models and alternatives
        """
        constraints = constraints or {}
        task_type = self._parse_task_type(task)

        # Get best retriever
        retrievers = self._filter_and_score(
            self.retrievers, constraints, task_type
        )
        best_retriever = retrievers[0] if retrievers else None

        # Get best reranker
        best_reranker = None
        rerankers = []
        if include_reranker:
            rerankers = self._filter_and_score(
                self.rerankers, constraints, task_type
            )
            best_reranker = rerankers[0] if rerankers else None

        # Get best RAG method
        best_rag = None
        rag_methods = []
        if include_rag:
            rag_methods = self._filter_and_score(
                self.rag_methods, constraints, task_type
            )
            best_rag = rag_methods[0] if rag_methods else None

        # Build pipeline config
        pipeline_config = None
        if best_retriever:
            pipeline_config = PipelineConfig(
                retriever=best_retriever.method,
                retriever_params={"model_name": best_retriever.model_path} if best_retriever.model_path else {},
                reranker=best_reranker.method if best_reranker else None,
                reranker_params={"model_name": best_reranker.model_path} if best_reranker and best_reranker.model_path else {},
                rag_method=best_rag.method if best_rag else None,
            )

        # Generate explanation
        explanation = self._generate_explanation(
            task_type, constraints, best_retriever, best_reranker, best_rag
        )

        return RecommendationResult(
            retriever=best_retriever,
            reranker=best_reranker,
            rag_method=best_rag,
            pipeline_config=pipeline_config,
            explanation=explanation,
            alternatives={
                "retrievers": retrievers[1:top_k+1] if len(retrievers) > 1 else [],
                "rerankers": rerankers[1:top_k+1] if len(rerankers) > 1 else [],
                "rag_methods": rag_methods[1:top_k+1] if len(rag_methods) > 1 else [],
            }
        )

    def recommend_for_latency(
        self,
        max_latency_ms: int,
        task: str = "qa",
        include_reranker: bool = True,
    ) -> RecommendationResult:
        """Recommend models optimized for latency budget."""
        speed_constraint = self._latency_to_speed(max_latency_ms)
        constraints = {"max_speed": speed_constraint}
        return self.recommend(task=task, constraints=constraints, include_reranker=include_reranker)

    def recommend_for_accuracy(
        self,
        task: str = "qa",
        gpu_available: bool = True,
    ) -> RecommendationResult:
        """Recommend highest accuracy models regardless of speed."""
        constraints = {"gpu": gpu_available, "prefer_accuracy": True}
        return self.recommend(task=task, constraints=constraints, include_reranker=True, include_rag=True)

    def recommend_for_production(
        self,
        gpu_available: bool = False,
        api_allowed: bool = True,
    ) -> RecommendationResult:
        """Recommend production-ready models (fast, reliable)."""
        constraints = {
            "gpu": gpu_available,
            "api_only": api_allowed,
            "prefer_speed": True,
        }
        return self.recommend(task="search", constraints=constraints, include_reranker=True)

    def _parse_task_type(self, task: str) -> TaskType:
        """Parse task string to TaskType enum."""
        task_map = {
            "qa": TaskType.QUESTION_ANSWERING,
            "question_answering": TaskType.QUESTION_ANSWERING,
            "search": TaskType.SEARCH,
            "retrieval": TaskType.SEARCH,
            "summarization": TaskType.SUMMARIZATION,
            "summarize": TaskType.SUMMARIZATION,
            "conversational": TaskType.CONVERSATIONAL,
            "chat": TaskType.CONVERSATIONAL,
            "domain": TaskType.DOMAIN_SPECIFIC,
            "domain_specific": TaskType.DOMAIN_SPECIFIC,
        }
        return task_map.get(task.lower(), TaskType.QUESTION_ANSWERING)

    def _filter_and_score(
        self,
        registry: Dict[str, ModelMetadata],
        constraints: Dict[str, Any],
        task: TaskType,
    ) -> List[ModelMetadata]:
        """Filter models by constraints and score by task suitability."""
        results = []

        for name, model in registry.items():
            if model.matches_constraints(constraints):
                score = self._compute_score(model, constraints, task)
                results.append((score, model))

        # Sort by score descending
        results.sort(key=lambda x: x[0], reverse=True)
        return [model for _, model in results]

    def _compute_score(
        self,
        model: ModelMetadata,
        constraints: Dict[str, Any],
        task: TaskType,
    ) -> float:
        """Compute overall score for a model."""
        score = 0.0

        # Task suitability (0-1)
        score += model.score_for_task(task) * 0.4

        # Accuracy score (0-1)
        accuracy_scores = {
            Accuracy.BASIC: 0.2,
            Accuracy.GOOD: 0.4,
            Accuracy.VERY_GOOD: 0.6,
            Accuracy.EXCELLENT: 0.8,
            Accuracy.STATE_OF_THE_ART: 1.0,
        }
        score += accuracy_scores.get(model.accuracy, 0.5) * 0.3

        # Speed score (0-1), inverse if prefer_accuracy
        speed_scores = {
            Speed.VERY_FAST: 1.0,
            Speed.FAST: 0.8,
            Speed.MEDIUM: 0.5,
            Speed.SLOW: 0.3,
            Speed.VERY_SLOW: 0.1,
        }
        speed_weight = 0.2 if constraints.get("prefer_accuracy") else 0.3
        if constraints.get("prefer_speed"):
            speed_weight = 0.4
        score += speed_scores.get(model.speed, 0.5) * speed_weight

        return score

    def _latency_to_speed(self, latency_ms: int) -> Speed:
        """Convert latency budget to speed requirement."""
        if latency_ms < 20:
            return Speed.VERY_FAST
        elif latency_ms < 100:
            return Speed.FAST
        elif latency_ms < 500:
            return Speed.MEDIUM
        elif latency_ms < 2000:
            return Speed.SLOW
        else:
            return Speed.VERY_SLOW

    def _generate_explanation(
        self,
        task: TaskType,
        constraints: Dict[str, Any],
        retriever: Optional[ModelMetadata],
        reranker: Optional[ModelMetadata],
        rag: Optional[ModelMetadata],
    ) -> str:
        """Generate human-readable explanation for recommendation."""
        parts = []

        if retriever:
            parts.append(f"**Retriever: {retriever.name}**")
            parts.append(f"  - {retriever.description}")
            parts.append(f"  - Speed: {retriever.speed.value}, Accuracy: {retriever.accuracy.value}")
            if retriever.gpu_required:
                parts.append("  - ⚠️ Requires GPU")

        if reranker:
            parts.append(f"\n**Reranker: {reranker.name}**")
            parts.append(f"  - {reranker.description}")
            parts.append(f"  - Speed: {reranker.speed.value}, Accuracy: {reranker.accuracy.value}")
            if reranker.api_required:
                parts.append(f"  - 🔌 Requires API ({reranker.api_provider})")

        if rag:
            parts.append(f"\n**RAG Method: {rag.name}**")
            parts.append(f"  - {rag.description}")

        return "\n".join(parts)

recommend(task='qa', constraints=None, include_reranker=True, include_rag=False, top_k=3)

Get recommended models for a given task and constraints.

Parameters:

Name	Type	Description	Default
task	str	Task type - "qa", "search", "summarization", "conversational", "domain_specific"	'qa'
constraints	Optional[Dict[str, Any]]	Optional constraints dict with keys like: - gpu: bool - Whether GPU is available - max_memory_mb: int - Maximum memory budget - max_latency_ms: int - Maximum latency budget - api_only: bool - Only recommend API-based models - no_api: bool - Only recommend local models - language: str - Required language support	None
include_reranker	bool	Whether to recommend a reranker	True
include_rag	bool	Whether to recommend a RAG method	False
top_k	int	Number of alternatives to include	3

Returns:

Type	Description
RecommendationResult	RecommendationResult with recommended models and alternatives

Source code in rankify/agent/recommender.py

def recommend(
    self,
    task: str = "qa",
    constraints: Optional[Dict[str, Any]] = None,
    include_reranker: bool = True,
    include_rag: bool = False,
    top_k: int = 3,
) -> RecommendationResult:
    """
    Get recommended models for a given task and constraints.

    Args:
        task: Task type - "qa", "search", "summarization", "conversational", "domain_specific"
        constraints: Optional constraints dict with keys like:
            - gpu: bool - Whether GPU is available
            - max_memory_mb: int - Maximum memory budget
            - max_latency_ms: int - Maximum latency budget
            - api_only: bool - Only recommend API-based models
            - no_api: bool - Only recommend local models
            - language: str - Required language support
        include_reranker: Whether to recommend a reranker
        include_rag: Whether to recommend a RAG method
        top_k: Number of alternatives to include

    Returns:
        RecommendationResult with recommended models and alternatives
    """
    constraints = constraints or {}
    task_type = self._parse_task_type(task)

    # Get best retriever
    retrievers = self._filter_and_score(
        self.retrievers, constraints, task_type
    )
    best_retriever = retrievers[0] if retrievers else None

    # Get best reranker
    best_reranker = None
    rerankers = []
    if include_reranker:
        rerankers = self._filter_and_score(
            self.rerankers, constraints, task_type
        )
        best_reranker = rerankers[0] if rerankers else None

    # Get best RAG method
    best_rag = None
    rag_methods = []
    if include_rag:
        rag_methods = self._filter_and_score(
            self.rag_methods, constraints, task_type
        )
        best_rag = rag_methods[0] if rag_methods else None

    # Build pipeline config
    pipeline_config = None
    if best_retriever:
        pipeline_config = PipelineConfig(
            retriever=best_retriever.method,
            retriever_params={"model_name": best_retriever.model_path} if best_retriever.model_path else {},
            reranker=best_reranker.method if best_reranker else None,
            reranker_params={"model_name": best_reranker.model_path} if best_reranker and best_reranker.model_path else {},
            rag_method=best_rag.method if best_rag else None,
        )

    # Generate explanation
    explanation = self._generate_explanation(
        task_type, constraints, best_retriever, best_reranker, best_rag
    )

    return RecommendationResult(
        retriever=best_retriever,
        reranker=best_reranker,
        rag_method=best_rag,
        pipeline_config=pipeline_config,
        explanation=explanation,
        alternatives={
            "retrievers": retrievers[1:top_k+1] if len(retrievers) > 1 else [],
            "rerankers": rerankers[1:top_k+1] if len(rerankers) > 1 else [],
            "rag_methods": rag_methods[1:top_k+1] if len(rag_methods) > 1 else [],
        }
    )

recommend_for_latency(max_latency_ms, task='qa', include_reranker=True)

Recommend models optimized for latency budget.

Source code in rankify/agent/recommender.py

def recommend_for_latency(
    self,
    max_latency_ms: int,
    task: str = "qa",
    include_reranker: bool = True,
) -> RecommendationResult:
    """Recommend models optimized for latency budget."""
    speed_constraint = self._latency_to_speed(max_latency_ms)
    constraints = {"max_speed": speed_constraint}
    return self.recommend(task=task, constraints=constraints, include_reranker=include_reranker)

recommend_for_accuracy(task='qa', gpu_available=True)

Recommend highest accuracy models regardless of speed.

Source code in rankify/agent/recommender.py

def recommend_for_accuracy(
    self,
    task: str = "qa",
    gpu_available: bool = True,
) -> RecommendationResult:
    """Recommend highest accuracy models regardless of speed."""
    constraints = {"gpu": gpu_available, "prefer_accuracy": True}
    return self.recommend(task=task, constraints=constraints, include_reranker=True, include_rag=True)

recommend_for_production(gpu_available=False, api_allowed=True)

Recommend production-ready models (fast, reliable).

Source code in rankify/agent/recommender.py

def recommend_for_production(
    self,
    gpu_available: bool = False,
    api_allowed: bool = True,
) -> RecommendationResult:
    """Recommend production-ready models (fast, reliable)."""
    constraints = {
        "gpu": gpu_available,
        "api_only": api_allowed,
        "prefer_speed": True,
    }
    return self.recommend(task="search", constraints=constraints, include_reranker=True)

RecommendationResult dataclass

Result of a model recommendation.

Source code in rankify/agent/recommender.py

@dataclass
class RecommendationResult:
    """Result of a model recommendation."""
    retriever: ModelMetadata
    reranker: Optional[ModelMetadata] = None
    rag_method: Optional[ModelMetadata] = None
    pipeline_config: Optional[PipelineConfig] = None
    explanation: str = ""
    alternatives: Dict[str, List[ModelMetadata]] = field(default_factory=dict)

    def build_pipeline(self):
        """Build pipeline from recommendation."""
        from rankify.agent.pipeline_builder import PipelineBuilder
        builder = PipelineBuilder()
        return builder.build(self.pipeline_config)

build_pipeline()

Build pipeline from recommendation.

Source code in rankify/agent/recommender.py

def build_pipeline(self):
    """Build pipeline from recommendation."""
    from rankify.agent.pipeline_builder import PipelineBuilder
    builder = PipelineBuilder()
    return builder.build(self.pipeline_config)

PipelineConfig dataclass

Configuration for a complete RAG pipeline.

Source code in rankify/agent/recommender.py

@dataclass
class PipelineConfig:
    """Configuration for a complete RAG pipeline."""
    retriever: str
    retriever_params: Dict[str, Any] = field(default_factory=dict)
    reranker: Optional[str] = None
    reranker_params: Dict[str, Any] = field(default_factory=dict)
    rag_method: Optional[str] = None
    rag_params: Dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary."""
        return {
            "retriever": self.retriever,
            "retriever_params": self.retriever_params,
            "reranker": self.reranker,
            "reranker_params": self.reranker_params,
            "rag_method": self.rag_method,
            "rag_params": self.rag_params,
        }

to_dict()

Convert to dictionary.

Source code in rankify/agent/recommender.py

def to_dict(self) -> Dict[str, Any]:
    """Convert to dictionary."""
    return {
        "retriever": self.retriever,
        "retriever_params": self.retriever_params,
        "reranker": self.reranker,
        "reranker_params": self.reranker_params,
        "rag_method": self.rag_method,
        "rag_params": self.rag_params,
    }

RankifyAgent

Conversational AI agent for Rankify model selection.

Supports multiple LLM backends: - Azure OpenAI - OpenAI - LiteLLM (100+ providers) - Local LLMs via transformers

Example

agent = RankifyAgent(backend="azure")
response = agent.chat("I need to build a QA system with no GPU")
print(response.message)
print(response.recommendation)

Source code in rankify/agent/agent.py

class RankifyAgent:
    """
    Conversational AI agent for Rankify model selection.

    Supports multiple LLM backends:
    - Azure OpenAI
    - OpenAI
    - LiteLLM (100+ providers)
    - Local LLMs via transformers

    Example:
        ```python
        agent = RankifyAgent(backend="azure")
        response = agent.chat("I need to build a QA system with no GPU")
        print(response.message)
        print(response.recommendation)
        ```
    """

    SYSTEM_PROMPT = """You are RankifyAgent, an expert AI assistant for the Rankify framework.
Rankify is a comprehensive Python toolkit for Retrieval, Re-Ranking, and Retrieval-Augmented Generation (RAG).

Your job is to help users select the best models for their use case. You have access to:
- **Sparse Retrievers**: BM25 (fast, no GPU, exact match)
- **Dense Retrievers**: DPR, ANCE, BGE, ColBERT, Contriever
- **Diver Dense Retrievers** (method="diver-dense"): Many bi-encoder and LLM-based variants selectable via model_id
- **Reasoning-Augmented Retrievers** (SOTA on BRIGHT benchmark): ReasonIR-8B, ReasonEmbed, BGE-Reasoner-Embed
- **Online Retriever**: Web search via APIs (real-time data)
- **HyDE**: Hypothetical Document Embedding for complex queries
- 23 reranking methods (MonoT5, FlashRank, RankGPT, InRanker, ColBERT, API rerankers, etc.)
- 7 RAG methods (Basic RAG, Chain-of-Thought, Self-Consistency, ReAct, FiD, etc.)

**Diver Dense Retriever Guide (method="diver-dense"):**
Valid model_ids (must have corpus_path):
- `bge` → BAAI/bge-large-en-v1.5
- `sbert` → sentence-transformers/all-mpnet-base-v2
- `nomic` → nomic-ai/nomic-embed-text-v1
- `diver` → AQ-MedAI/Diver-Retriever-4B (flagship diverse evidence model)
- `inst-l` → hkunlp/instructor-large
- `inst-xl` → hkunlp/instructor-xl
- `e5` → intfloat/e5-mistral-7b-instruct (LLM-based)
- `sf` → Salesforce/SFR-Embedding-Mistral (LLM-based)
- `rader` → Raderspace/RaDeR_Qwen_25_7B (reasoning-augmented)
- `grit` → GritLM/GritLM-7B (generative representation)
- `m2` → togethercomputer/m2-bert-80M-32k-retrieval (long context, 32k)
- `contriever` → facebook/contriever-msmarco

Example: `Retriever(method="diver-dense", model_id="diver", corpus_path="data/corpus.jsonl", n_docs=10)`

**Reasoning-Augmented Retrievers:**
- `Retriever(method="reasonir", corpus_path=...)` → reasonir/ReasonIR-8B (SOTA BRIGHT benchmark)
- `Retriever(method="reason-embed", model_id="qwen3-8b"|"qwen3-4b"|"llama-8b", corpus_path=...)`
- `Retriever(method="bge-reasoner-embed", corpus_path=...)` → BAAI/bge-reasoner-embed-qwen3-8b

When helping users, consider:
1. Their task type (QA, search, summarization, conversational, reasoning-intensive)
2. Hardware constraints (GPU availability, memory)
3. Latency requirements
4. Whether they can use APIs or need local models
5. Language requirements

Provide clear recommendations with explanations. Include code snippets when helpful.
Be concise but thorough. Ask clarifying questions if needed."""

    def __init__(
        self,
        backend: str = "azure",
        model_name: Optional[str] = None,
        api_key: Optional[str] = None,
        **kwargs
    ):
        """
        Initialize RankifyAgent.

        Args:
            backend: LLM backend - "azure", "openai", "litellm", "local"
            model_name: Model name (optional, uses defaults per backend)
            api_key: API key (optional, uses environment variables)
            **kwargs: Additional backend-specific arguments
        """
        self.backend = backend.lower()
        self.model_name = model_name
        self.api_key = api_key
        self.kwargs = kwargs
        self.recommender = RankifyRecommender()
        self.conversation_history: List[Dict[str, str]] = []

        # Set up the LLM client
        self._setup_client()

    def _setup_client(self):
        """Set up the LLM client based on backend."""
        if self.backend == "azure":
            self._setup_azure()
        elif self.backend == "openai":
            self._setup_openai()
        elif self.backend == "litellm":
            self._setup_litellm()
        elif self.backend == "local":
            self._setup_local()
        else:
            raise ValueError(f"Unknown backend: {self.backend}")

    def _setup_azure(self):
        """Set up Azure OpenAI client."""
        try:
            from openai import AzureOpenAI
        except ImportError:
            raise ImportError("Please install openai: pip install openai")

        self.client = AzureOpenAI(
            api_key=self.api_key or os.environ.get("AZURE_OPENAI_API_KEY"),
            api_version=os.environ.get("AZURE_API_VERSION", "2024-05-01-preview"),
            azure_endpoint=os.environ.get("AZURE_OPENAI_ENDPOINT"),
        )
        self.model_name = self.model_name or os.environ.get("AZURE_DEPLOYMENT_NAME", "gpt-4o")
        self._call_llm = self._call_azure

    def _setup_openai(self):
        """Set up OpenAI client."""
        try:
            from openai import OpenAI
        except ImportError:
            raise ImportError("Please install openai: pip install openai")

        self.client = OpenAI(
            api_key=self.api_key or os.environ.get("OPENAI_API_KEY"),
        )
        self.model_name = self.model_name or "gpt-4o-mini"
        self._call_llm = self._call_openai

    def _setup_litellm(self):
        """Set up LiteLLM for 100+ providers."""
        try:
            import litellm
        except ImportError:
            raise ImportError("Please install litellm: pip install litellm")

        self.litellm = litellm
        self.model_name = self.model_name or "gpt-4o-mini"
        self._call_llm = self._call_litellm

    def _setup_local(self):
        """Set up local LLM via transformers."""
        try:
            from transformers import pipeline
        except ImportError:
            raise ImportError("Please install transformers: pip install transformers")

        self.model_name = self.model_name or "microsoft/Phi-3-mini-4k-instruct"
        self.pipeline = pipeline(
            "text-generation",
            model=self.model_name,
            device_map="auto",
            **self.kwargs
        )
        self._call_llm = self._call_local

    def _call_azure(self, messages: List[Dict[str, str]]) -> str:
        """Call Azure OpenAI."""
        response = self.client.chat.completions.create(
            model=self.model_name,
            messages=messages,
            temperature=0.7,
            max_tokens=1000,
        )
        return response.choices[0].message.content

    def _call_openai(self, messages: List[Dict[str, str]]) -> str:
        """Call OpenAI."""
        response = self.client.chat.completions.create(
            model=self.model_name,
            messages=messages,
            temperature=0.7,
            max_tokens=1000,
        )
        return response.choices[0].message.content

    def _call_litellm(self, messages: List[Dict[str, str]]) -> str:
        """Call LiteLLM."""
        response = self.litellm.completion(
            model=self.model_name,
            messages=messages,
            temperature=0.7,
            max_tokens=1000,
        )
        return response.choices[0].message.content

    def _call_local(self, messages: List[Dict[str, str]]) -> str:
        """Call local LLM."""
        # Format messages for local model
        prompt = "\n".join([
            f"{m['role'].upper()}: {m['content']}" 
            for m in messages
        ])
        prompt += "\nASSISTANT:"

        output = self.pipeline(
            prompt,
            max_new_tokens=500,
            do_sample=True,
            temperature=0.7,
        )
        return output[0]["generated_text"].split("ASSISTANT:")[-1].strip()

    def chat(self, user_message: str) -> AgentResponse:
        """
        Have a conversation with the agent.

        Args:
            user_message: User's message/question

        Returns:
            AgentResponse with message, optional recommendation, and code snippet
        """
        # Add user message to history
        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })

        # Detect if this is a recommendation request
        intent = self._detect_intent(user_message)

        # If asking for recommendation, get one first
        recommendation = None
        context = ""
        if intent.get("wants_recommendation"):
            recommendation = self._get_recommendation_from_intent(intent)
            context = self._format_recommendation_context(recommendation)

        # Prepare messages for LLM
        messages = [
            {"role": "system", "content": self.SYSTEM_PROMPT + context}
        ] + self.conversation_history

        # Get LLM response
        response_text = self._call_llm(messages)

        # Add assistant response to history
        self.conversation_history.append({
            "role": "assistant",
            "content": response_text
        })

        # Generate code snippet if recommendation was made
        code_snippet = None
        if recommendation:
            code_snippet = self._generate_code_snippet(recommendation)

        return AgentResponse(
            message=response_text,
            recommendation=recommendation,
            code_snippet=code_snippet,
        )

    def recommend(
        self,
        task: str = "qa",
        gpu: bool = True,
        api_allowed: bool = True,
        include_rag: bool = False,
    ) -> RecommendationResult:
        """
        Get a direct recommendation without conversation.

        Args:
            task: Task type - "qa", "search", "summarization"
            gpu: Whether GPU is available
            api_allowed: Whether API-based models are allowed
            include_rag: Whether to include RAG method

        Returns:
            RecommendationResult with recommended models
        """
        constraints = {"gpu": gpu}
        if not api_allowed:
            constraints["no_api"] = True

        return self.recommender.recommend(
            task=task,
            constraints=constraints,
            include_rag=include_rag,
        )

    def _detect_intent(self, message: str) -> Dict[str, Any]:
        """Detect user intent from message."""
        message_lower = message.lower()

        intent = {
            "wants_recommendation": False,
            "task": None,
            "constraints": {}
        }

        # Check for recommendation keywords
        recommendation_keywords = [
            "recommend", "suggest", "best", "which", "what should",
            "help me choose", "need a", "looking for", "want to build"
        ]
        if any(kw in message_lower for kw in recommendation_keywords):
            intent["wants_recommendation"] = True

        # Detect task type
        if any(w in message_lower for w in ["qa", "question", "answer", "reading"]):
            intent["task"] = "qa"
        elif any(w in message_lower for w in ["search", "find", "retriev", "lookup"]):
            intent["task"] = "search"
        elif any(w in message_lower for w in ["summar", "condense", "tldr"]):
            intent["task"] = "summarization"
        elif any(w in message_lower for w in ["chat", "convers", "dialog"]):
            intent["task"] = "conversational"
        else:
            intent["task"] = "qa"  # Default

        # Detect constraints
        if any(w in message_lower for w in ["no gpu", "cpu only", "without gpu"]):
            intent["constraints"]["gpu"] = False
        elif "gpu" in message_lower:
            intent["constraints"]["gpu"] = True

        if any(w in message_lower for w in ["fast", "low latency", "quick", "speed"]):
            intent["constraints"]["prefer_speed"] = True

        if any(w in message_lower for w in ["accurate", "best quality", "highest quality"]):
            intent["constraints"]["prefer_accuracy"] = True

        if any(w in message_lower for w in ["local", "no api", "offline", "privacy"]):
            intent["constraints"]["no_api"] = True

        if any(w in message_lower for w in ["production", "deploy", "scale"]):
            intent["constraints"]["prefer_speed"] = True

        return intent

    def _get_recommendation_from_intent(
        self, 
        intent: Dict[str, Any]
    ) -> RecommendationResult:
        """Get recommendation based on detected intent."""
        return self.recommender.recommend(
            task=intent.get("task", "qa"),
            constraints=intent.get("constraints", {}),
            include_reranker=True,
            include_rag=True,
        )

    def _format_recommendation_context(
        self, 
        recommendation: RecommendationResult
    ) -> str:
        """Format recommendation as context for LLM."""
        parts = ["\n\n[RECOMMENDATION CONTEXT]"]

        if recommendation.retriever:
            parts.append(f"Best Retriever: {recommendation.retriever.name}")
            parts.append(f"  Method: {recommendation.retriever.method}")

        if recommendation.reranker:
            parts.append(f"Best Reranker: {recommendation.reranker.name}")
            parts.append(f"  Method: {recommendation.reranker.method}")

        if recommendation.rag_method:
            parts.append(f"Best RAG Method: {recommendation.rag_method.name}")

        return "\n".join(parts)

    def _generate_code_snippet(
        self, 
        recommendation: RecommendationResult
    ) -> str:
        """Generate code snippet from recommendation."""
        lines = ["from rankify.retrievers.retriever import Retriever"]

        if recommendation.reranker:
            lines.append("from rankify.models.reranking import Reranking")

        if recommendation.rag_method:
            lines.append("from rankify.generator.generator import Generator")

        lines.append("")

        # Retriever
        if recommendation.retriever:
            r = recommendation.retriever
            lines.append(f"# Retriever: {r.name}")
            if r.method == "bm25":
                lines.append(f'retriever = Retriever(method="{r.method}", n_docs=100)')
            else:
                lines.append(f'retriever = Retriever(method="{r.method}", n_docs=100)')

        # Reranker
        if recommendation.reranker:
            rr = recommendation.reranker
            lines.append(f"\n# Reranker: {rr.name}")
            if rr.model_path:
                lines.append(f'reranker = Reranking(method="{rr.method}", model_name="{rr.model_path}")')
            else:
                lines.append(f'reranker = Reranking(method="{rr.method}")')

        # RAG
        if recommendation.rag_method:
            rag = recommendation.rag_method
            lines.append(f"\n# RAG Method: {rag.name}")
            lines.append(f'generator = Generator(method="{rag.method}", model_name="gpt-4o-mini", backend="openai")')

        # Usage
        lines.append("\n# Usage")
        lines.append("documents = retriever.retrieve(documents)")
        if recommendation.reranker:
            lines.append("documents = reranker.rank(documents)")
        if recommendation.rag_method:
            lines.append("answers = generator.generate(documents)")

        return "\n".join(lines)

    def clear_history(self):
        """Clear conversation history."""
        self.conversation_history = []

    def get_available_models(self, model_type: str = "all") -> Dict[str, List[str]]:
        """
        Get list of available models.

        Args:
            model_type: "retriever", "reranker", "rag", or "all"

        Returns:
            Dict of model names by type
        """
        result = {}

        if model_type in ["retriever", "all"]:
            result["retrievers"] = list(RETRIEVER_REGISTRY.keys())

        if model_type in ["reranker", "all"]:
            result["rerankers"] = list(RERANKER_REGISTRY.keys())

        if model_type in ["rag", "all"]:
            result["rag_methods"] = list(RAG_METHOD_REGISTRY.keys())

        return result

__init__(backend='azure', model_name=None, api_key=None, **kwargs)

Initialize RankifyAgent.

Parameters:

Name	Type	Description	Default
backend	str	LLM backend - "azure", "openai", "litellm", "local"	'azure'
model_name	Optional[str]	Model name (optional, uses defaults per backend)	None
api_key	Optional[str]	API key (optional, uses environment variables)	None
**kwargs		Additional backend-specific arguments	{}

Source code in rankify/agent/agent.py

def __init__(
    self,
    backend: str = "azure",
    model_name: Optional[str] = None,
    api_key: Optional[str] = None,
    **kwargs
):
    """
    Initialize RankifyAgent.

    Args:
        backend: LLM backend - "azure", "openai", "litellm", "local"
        model_name: Model name (optional, uses defaults per backend)
        api_key: API key (optional, uses environment variables)
        **kwargs: Additional backend-specific arguments
    """
    self.backend = backend.lower()
    self.model_name = model_name
    self.api_key = api_key
    self.kwargs = kwargs
    self.recommender = RankifyRecommender()
    self.conversation_history: List[Dict[str, str]] = []

    # Set up the LLM client
    self._setup_client()

chat(user_message)

Have a conversation with the agent.

Parameters:

Name	Type	Description	Default
user_message	str	User's message/question	required

Returns:

Type	Description
AgentResponse	AgentResponse with message, optional recommendation, and code snippet

Source code in rankify/agent/agent.py

def chat(self, user_message: str) -> AgentResponse:
    """
    Have a conversation with the agent.

    Args:
        user_message: User's message/question

    Returns:
        AgentResponse with message, optional recommendation, and code snippet
    """
    # Add user message to history
    self.conversation_history.append({
        "role": "user",
        "content": user_message
    })

    # Detect if this is a recommendation request
    intent = self._detect_intent(user_message)

    # If asking for recommendation, get one first
    recommendation = None
    context = ""
    if intent.get("wants_recommendation"):
        recommendation = self._get_recommendation_from_intent(intent)
        context = self._format_recommendation_context(recommendation)

    # Prepare messages for LLM
    messages = [
        {"role": "system", "content": self.SYSTEM_PROMPT + context}
    ] + self.conversation_history

    # Get LLM response
    response_text = self._call_llm(messages)

    # Add assistant response to history
    self.conversation_history.append({
        "role": "assistant",
        "content": response_text
    })

    # Generate code snippet if recommendation was made
    code_snippet = None
    if recommendation:
        code_snippet = self._generate_code_snippet(recommendation)

    return AgentResponse(
        message=response_text,
        recommendation=recommendation,
        code_snippet=code_snippet,
    )

recommend(task='qa', gpu=True, api_allowed=True, include_rag=False)

Get a direct recommendation without conversation.

Parameters:

Name	Type	Description	Default
task	str	Task type - "qa", "search", "summarization"	'qa'
gpu	bool	Whether GPU is available	True
api_allowed	bool	Whether API-based models are allowed	True
include_rag	bool	Whether to include RAG method	False

Returns:

Type	Description
RecommendationResult	RecommendationResult with recommended models

Source code in rankify/agent/agent.py

def recommend(
    self,
    task: str = "qa",
    gpu: bool = True,
    api_allowed: bool = True,
    include_rag: bool = False,
) -> RecommendationResult:
    """
    Get a direct recommendation without conversation.

    Args:
        task: Task type - "qa", "search", "summarization"
        gpu: Whether GPU is available
        api_allowed: Whether API-based models are allowed
        include_rag: Whether to include RAG method

    Returns:
        RecommendationResult with recommended models
    """
    constraints = {"gpu": gpu}
    if not api_allowed:
        constraints["no_api"] = True

    return self.recommender.recommend(
        task=task,
        constraints=constraints,
        include_rag=include_rag,
    )

clear_history()

Clear conversation history.

Source code in rankify/agent/agent.py

def clear_history(self):
    """Clear conversation history."""
    self.conversation_history = []

get_available_models(model_type='all')

Get list of available models.

Parameters:

Name	Type	Description	Default
model_type	str	"retriever", "reranker", "rag", or "all"	'all'

Returns:

Type	Description
Dict[str, List[str]]	Dict of model names by type

Source code in rankify/agent/agent.py

def get_available_models(self, model_type: str = "all") -> Dict[str, List[str]]:
    """
    Get list of available models.

    Args:
        model_type: "retriever", "reranker", "rag", or "all"

    Returns:
        Dict of model names by type
    """
    result = {}

    if model_type in ["retriever", "all"]:
        result["retrievers"] = list(RETRIEVER_REGISTRY.keys())

    if model_type in ["reranker", "all"]:
        result["rerankers"] = list(RERANKER_REGISTRY.keys())

    if model_type in ["rag", "all"]:
        result["rag_methods"] = list(RAG_METHOD_REGISTRY.keys())

    return result

AgentResponse dataclass

Response from the agent.

Source code in rankify/agent/agent.py

@dataclass
class AgentResponse:
    """Response from the agent."""
    message: str
    recommendation: Optional[RecommendationResult] = None
    code_snippet: Optional[str] = None

get_all_retrievers()

Get all available retrievers.

Source code in rankify/agent/model_registry.py

def get_all_retrievers() -> Dict[str, ModelMetadata]:
    """Get all available retrievers."""
    return RETRIEVER_REGISTRY

get_all_rerankers()

Get all available rerankers.

Source code in rankify/agent/model_registry.py

def get_all_rerankers() -> Dict[str, ModelMetadata]:
    """Get all available rerankers."""
    return RERANKER_REGISTRY

get_all_rag_methods()

Get all available RAG methods.

Source code in rankify/agent/model_registry.py

def get_all_rag_methods() -> Dict[str, ModelMetadata]:
    """Get all available RAG methods."""
    return RAG_METHOD_REGISTRY

get_model(model_type, name)

Get a specific model by type and name.

Source code in rankify/agent/model_registry.py

def get_model(model_type: str, name: str) -> Optional[ModelMetadata]:
    """Get a specific model by type and name."""
    registries = {
        "retriever": RETRIEVER_REGISTRY,
        "reranker": RERANKER_REGISTRY,
        "rag": RAG_METHOD_REGISTRY,
    }
    registry = registries.get(model_type)
    if registry:
        return registry.get(name)
    return None

filter_models(model_type, constraints, task=None)

Filter models by constraints and optionally score for task.

Source code in rankify/agent/model_registry.py

def filter_models(
    model_type: str,
    constraints: Dict[str, Any],
    task: Optional[TaskType] = None,
) -> List[ModelMetadata]:
    """Filter models by constraints and optionally score for task."""
    registries = {
        "retriever": RETRIEVER_REGISTRY,
        "reranker": RERANKER_REGISTRY,
        "rag": RAG_METHOD_REGISTRY,
    }
    registry = registries.get(model_type, {})

    results = []
    for name, model in registry.items():
        if model.matches_constraints(constraints):
            results.append(model)

    # Sort by task suitability if task provided
    if task:
        results.sort(key=lambda m: m.score_for_task(task), reverse=True)

    return results

recommend(task='qa', gpu=True, api_allowed=True, include_rag=False)

Quick recommendation function.

Parameters:

Name	Type	Description	Default
task	str	"qa", "search", "summarization", "conversational"	'qa'
gpu	bool	Whether GPU is available	True
api_allowed	bool	Whether API-based models are allowed	True
include_rag	bool	Whether to include RAG method in recommendation	False

Returns:

Type	Description
RecommendationResult	RecommendationResult with best models for the use case

Source code in rankify/agent/recommender.py

def recommend(
    task: str = "qa",
    gpu: bool = True,
    api_allowed: bool = True,
    include_rag: bool = False,
) -> RecommendationResult:
    """
    Quick recommendation function.

    Args:
        task: "qa", "search", "summarization", "conversational"
        gpu: Whether GPU is available
        api_allowed: Whether API-based models are allowed
        include_rag: Whether to include RAG method in recommendation

    Returns:
        RecommendationResult with best models for the use case
    """
    recommender = RankifyRecommender()
    constraints = {"gpu": gpu}
    if not api_allowed:
        constraints["no_api"] = True
    return recommender.recommend(
        task=task,
        constraints=constraints,
        include_rag=include_rag,
    )