Agentic RAG

from agno.vectordb.lancedb import LanceDb, SearchType
from agno.embedder.openai import OpenAIEmbedder

# Configure vector database
vector_db = LanceDb(
    table_name="mcp-docs-knowledge-base",  # Table name for storing vectors
    uri="tmp/lancedb",                     # Local storage path
    search_type=SearchType.vector,         # Vector similarity search
    embedder=OpenAIEmbedder(id="text-embedding-3-small")  # Embedding model
)

from agno.knowledge.url import UrlKnowledge
from agno.vectordb.lancedb import LanceDb, SearchType
from agno.embedder.openai import OpenAIEmbedder

def load_knowledge_base(urls: list[str] = None):
    """Load URLs into vector database."""
    knowledge_base = UrlKnowledge(
        urls=urls or [],
        vector_db=LanceDb(
            table_name="mcp-docs-knowledge-base",
            uri="tmp/lancedb",
            search_type=SearchType.vector,
            embedder=OpenAIEmbedder(id="text-embedding-3-small"),
        ),
    )
    knowledge_base.load()  # Downloads, chunks, embeds, and stores
    return knowledge_base

from agno.agent import Agent, RunResponseEvent
from agno.models.openai import OpenAIChat
from typing import Iterator

def agentic_rag_response(
    urls: list[str] = None, 
    query: str = ""
) -> Iterator[RunResponseEvent]:
    """Create RAG agent and stream response."""
    # Load knowledge base
    knowledge_base = load_knowledge_base(urls)
    
    # Create agent with knowledge
    agent = Agent(
        model=OpenAIChat(id="gpt-4o"),
        knowledge=knowledge_base,
        search_knowledge=True,  # Enable knowledge search
        markdown=True,
    )
    
    # Stream response
    response: Iterator[RunResponseEvent] = agent.run(query, stream=True)
    return response

# The knowledge base automatically:
# 1. Fetches URL content
# 2. Chunks text (typically 512-1024 tokens)
# 3. Generates embeddings via OpenAI API
# 4. Stores in LanceDB with metadata

knowledge_base.load()  # All steps happen here

import os
from phoenix.otel import register

# Set environment variables for Arize Phoenix
os.environ["PHOENIX_CLIENT_HEADERS"] = f"api_key={os.getenv('ARIZE_PHOENIX_API_KEY')}"
os.environ["PHOENIX_COLLECTOR_ENDPOINT"] = "https://app.phoenix.arize.com"

# Configure the Phoenix tracer
tracer_provider = register(
    project_name="default",
    auto_instrument=True,  # Automatically trace OpenAI calls
)

import streamlit as st
from typing import Iterator

st.set_page_config(page_title="Agentic RAG", layout="wide")
st.title("Agentic RAG with Agno & GPT-4o")

# Sidebar: URL management
with st.sidebar:
    st.markdown("### 🧠 Knowledge Base URLs")
    
    if "urls" not in st.session_state:
        st.session_state.urls = [""]
    
    # URL input fields
    for i, url in enumerate(st.session_state.urls):
        st.session_state.urls[i] = st.text_input(
            f"URL {i+1}",
            value=url,
            key=f"url_{i}",
            label_visibility="collapsed"
        )
    
    # Add URL button
    if st.button("➕"):
        if st.session_state.urls and st.session_state.urls[-1].strip() != "":
            st.session_state.urls.append("")
    
    # Load knowledge base
    if st.button("Load Knowledge Base"):
        urls = [u for u in st.session_state.urls if u.strip()]
        urls = list(dict.fromkeys(urls))  # Remove duplicates
        
        if urls:
            with st.spinner("Loading knowledge base..."):
                try:
                    knowledge_base = load_knowledge_base(urls)
                    st.session_state.docs_loaded = True
                    st.session_state.loaded_urls = urls.copy()
                    st.success(f"Loaded {len(urls)} URL(s)!")
                except Exception as e:
                    st.error(f"Error: {str(e)}")
        else:
            st.warning("Please add at least one URL.")
    
    # Reset button
    if st.button("🔄 Reset KB"):
        st.session_state.docs_loaded = False
        if 'loaded_urls' in st.session_state:
            del st.session_state['loaded_urls']
        st.success("Knowledge base reset!")
        st.rerun()

# Chat interface
query = st.chat_input("Ask a question")

if query:
    if not st.session_state.get('docs_loaded', False):
        st.warning("Please load the knowledge base first.")
    else:
        loaded_urls = st.session_state.loaded_urls
        response = agentic_rag_response(loaded_urls, query)
        
        st.markdown("#### Answer")
        answer = ""
        answer_placeholder = st.empty()
        
        # Stream response
        for content in response:
            if hasattr(content, 'event') and content.event == "RunResponseContent":
                answer += content.content
                answer_placeholder.markdown(answer)

# This happens automatically when agent.run() is called
# with search_knowledge=True:

# 1. Query embedding
query_vector = embedder.embed(query)

# 2. Vector search in LanceDB
results = vector_db.search(
    query_vector,
    limit=5,  # Top 5 most similar chunks
    metric="cosine"  # Cosine similarity
)

# 3. Context is automatically added to prompt
# 4. LLM generates response with context

git clone https://github.com/Arindam200/awesome-ai-apps.git
cd rag_apps/agentic_rag
uv sync

OPENAI_API_KEY=your_openai_api_key_here
ARIZE_PHOENIX_API_KEY=your_phoenix_api_key_here  # Optional

uv run streamlit run main.py