Agentic RAG Infinity Reranker

Run hybrid LanceDB search with Cohere embeddings and a local Infinity reranker (BAAI/bge-reranker-base on localhost:7997) behind a Claude agent.

Run agentic RAG over the Agno docs with LanceDB hybrid search, Cohere embeddings, and a local Infinity reranker on port 7997.

agentic_rag_infinity_reranker.py

"""
Agentic Rag Infinity Reranker
=============================

Demonstrates agentic RAG with an Infinity reranker backend (relocated integration example).
"""

import asyncio
import importlib

from agno.agent import Agent
from agno.knowledge.embedder.cohere import CohereEmbedder
from agno.knowledge.knowledge import Knowledge
from agno.knowledge.reranker.infinity import InfinityReranker
from agno.models.anthropic import Claude
from agno.vectordb.lancedb import LanceDb, SearchType

# ---------------------------------------------------------------------------
# Setup
# ---------------------------------------------------------------------------
knowledge = Knowledge(
    # Use LanceDB as the vector database, store embeddings in the `agno_docs_infinity` table
    vector_db=LanceDb(
        uri="tmp/lancedb",
        table_name="agno_docs_infinity",
        search_type=SearchType.hybrid,
        embedder=CohereEmbedder(id="embed-v4.0"),
        # Use Infinity reranker for local, fast reranking
        reranker=InfinityReranker(
            model="BAAI/bge-reranker-base",  # You can change this to other models
            host="localhost",
            port=7997,
            top_n=5,  # Return top 5 reranked documents
        ),
    ),
)

# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
    model=Claude(id="claude-3-7-sonnet-latest"),
    # Agentic RAG is enabled by default when `knowledge` is provided to the Agent.
    knowledge=knowledge,
    # search_knowledge=True gives the Agent the ability to search on demand
    # search_knowledge is True by default
    search_knowledge=True,
    instructions=[
        "Include sources in your response.",
        "Always search your knowledge before answering the question.",
        "Provide detailed and accurate information based on the retrieved documents.",
    ],
    markdown=True,
)


def test_infinity_connection():
    """Test if Infinity server is running and accessible"""
    try:
        infinity_client = importlib.import_module("infinity_client")
        _ = infinity_client.Client(base_url="http://localhost:7997")
        print("[OK] Successfully connected to Infinity server at localhost:7997")
        return True
    except Exception as e:
        print(f"[ERROR] Failed to connect to Infinity server: {e}")
        print(
            "\nPlease make sure Infinity server is running. See setup instructions above."
        )
        return False


# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    print("Agentic RAG with Infinity Reranker Example")
    print("=" * 50)

    # Load knowledge base
    print("\nLoading knowledge base...")
    asyncio.run(
        knowledge.ainsert_many(
            urls=[
                "https://docs.agno.com/agents/overview.md",
                "https://docs.agno.com/tools/overview.md",
                "https://docs.agno.com/knowledge/overview.md",
            ]
        )
    )

    # Test Infinity connection first
    if not test_infinity_connection():
        exit(1)

    print("\nStarting agent interaction...")
    print("=" * 50)

    # Example questions to test the reranking capabilities
    questions = [
        "What are Agents and how do they work?",
        "How do I use tools with agents?",
        "What is the difference between knowledge and tools?",
    ]

    for i, question in enumerate(questions, 1):
        print(f"\n[Question {i}] {question}")
        print("-" * 40)
        agent.print_response(question, stream=True)
        print("\n" + "=" * 50)

    print("\nExample completed!")
    print("\nThe Infinity reranker helped improve the relevance of retrieved documents")
    print("by reranking them based on semantic similarity to your queries.")

Run the Example

1. Set up your virtual environment

   Mac

   uv venv --python 3.12
   source .venv/bin/activate
   

   uv venv --python 3.12
   .venv\Scripts\activate
   
2. Install dependencies

   uv pip install -U agno anthropic cohere infinity-client lancedb pyarrow
   
3. Export your API keys

   Mac/Linux

   export ANTHROPIC_API_KEY="your_anthropic_api_key_here"
   export CO_API_KEY="your_co_api_key_here"
   

   $Env:ANTHROPIC_API_KEY="your_anthropic_api_key_here"
   $Env:CO_API_KEY="your_co_api_key_here"
   
4. Start Infinity

   Install Infinity and start the reranker on port 7997:

   uv pip install -U "infinity-emb[all]"
   infinity_emb v2 --model-id BAAI/bge-reranker-base --port 7997
   
5. Run the example

   Save the code above as agentic_rag_infinity_reranker.py, then run:

   python agentic_rag_infinity_reranker.py
   

Full source: cookbook/07_knowledge/05_integrations/rag/agentic_rag_infinity_reranker.py