Building Agentic RAG with Streaming Responses Using Agno and GPT-5

Building agentic RAG with streaming responses requires orchestrating URL ingestion, vector embedding storage, and real-time token generation—implemented in the Arindam200/awesome-ai-apps repository using Agno agents, LanceDB, and OpenAI's GPT-5 with Streamlit for the UI.

The Arindam200/awesome-ai-apps repository provides a production-ready reference implementation for developers exploring retrieval-augmented generation with interactive streaming capabilities. Located in rag_apps/agentic_rag/, this full-stack example demonstrates how to ingest web content, persist embeddings in a local vector database, and stream AI-generated answers character-by-character through a modern web interface.

Architecture Overview

The system follows a modular pipeline architecture that separates data ingestion from inference. Web URLs enter through the Streamlit sidebar, get processed into vector embeddings, and feed an Agno agent that streams responses back to the user interface.

The data flow mirrors the repository's documented architecture:

Web URLs → UrlKnowledge (Agno) → LanceDB → Agno Agent (GPT-5) → Streaming Response

This design ensures that knowledge base updates remain independent from the inference layer, allowing you to re-index content without modifying the agent configuration.

Core System Components

Streamlit Interface (main.py)

The user interface resides entirely in rag_apps/agentic_rag/main.py, where st.set_page_config initializes the application layout. The sidebar manages URL input states through st.session_state.urls, while the main chat area handles real-time message rendering. The "Load Knowledge Base" button triggers the ingestion pipeline, and the chat input captures user queries for the agentic RAG workflow.

Knowledge Ingestion with UrlKnowledge

The UrlKnowledge class (from the Agno framework) handles the heavy lifting of content extraction and chunking. When load_knowledge_base() executes (lines 36-50), it instantiates UrlKnowledge with the provided URL list and a LanceDB backend. Internally, this fetches each webpage, extracts readable text, splits content into semantic chunks, and prepares them for embedding generation.

LanceDB Vector Storage

Embeddings persist in LanceDB, configured as a fast, local vector store. The implementation uses LanceDb(table_name="mcp-docs-knowledge-base", uri="tmp/lancedb") (lines 43-48), storing embeddings locally in the tmp/lancedb directory. This eliminates external database dependencies while maintaining sub-second retrieval latency for similarity searches.

OpenAI Embeddings and GPT-5 Integration

The system uses two distinct OpenAI models: OpenAIEmbedder(id="text-embedding-3-small") for creating vector representations of web content (line 47), and OpenAIChat(id="gpt-5-2025-08-07") for generating final responses (line 61). This separation allows optimization of embedding costs while leveraging GPT-5's advanced reasoning capabilities for the agentic layer.

Real-Time Streaming Protocol

Streaming activates through the stream=True parameter in agent.run(query, stream=True) (line 68). Rather than returning a complete string, this yields a generator of RunResponseEvent objects. The UI filters for RunResponseContent events (lines 53-57), appending each token chunk to the displayed message in real-time, creating a ChatGPT-like typing effect.

Observability with Arize Phoenix

Optional telemetry integrates via phoenix.otel.register (lines 20-31), tracing requests, latency metrics, and token usage. This provides production visibility into retrieval accuracy and generation performance without impacting the streaming user experience.

Step-by-Step Implementation Flow

Step 1: URL Collection Users enter target URLs through the Streamlit sidebar, which stores them in session state for persistence across interactions.

Step 2: Knowledge Base Construction Clicking "Load Knowledge Base" invokes load_knowledge_base(urls), which:

• Instantiates UrlKnowledge with the URL list and LanceDB configuration
• Generates embeddings using OpenAIEmbedder
• Writes vectors to the local LanceDB table

Step 3: Agent Initialization When a query arrives, agentic_rag_response(loaded_urls, query) creates an Agent instance with knowledge=knowledge_base and search_knowledge=True enabled. This configures the agent to automatically retrieve relevant context before generation.

Step 4: Streamed Response Generation The agent processes the query with streaming enabled, yielding RunResponseEvent objects containing partial tokens. The UI iterates this generator, flushing each chunk to the frontend immediately upon arrival.

Practical Code Examples

Loading the Knowledge Base Programmatically

Import the ingestion logic directly from the main module to index content outside the UI:

from rag_apps.agentic_rag.main import load_knowledge_base

urls = [
    "https://docs.python.org/3/tutorial/",
    "https://openai.com/blog/gpt-4"
]

# Returns a UrlKnowledge object ready for queries

kb = load_knowledge_base(urls)

Executing a Streaming RAG Query

Access the agent directly for backend processing or testing:

from rag_apps.agentic_rag.main import agentic_rag_response

query = "How does GPT-4 handle function calling?"
response_stream = agentic_rag_response(urls, query)

# Print the streamed answer as it arrives

answer = ""
for event in response_stream:
    if hasattr(event, "event") and event.event == "RunResponseContent":
        answer += event.content
        print(event.content, end="", flush=True)

Complete End-to-End Script

For batch processing or automation, combine both functions in a standalone script:

import os
from dotenv import load_dotenv

# Load OPENAI_API_KEY and ARIZE_PHOENIX_API_KEY

load_dotenv()

from rag_apps.agentic_rag.main import (
    load_knowledge_base,
    agentic_rag_response,
)

# Define URLs to index

urls = [
    "https://openai.com/blog/gpt-4",
    "https://modelcontextprotocol.io/docs/learn/architecture.md",
]

# Build the knowledge base (runs once)

kb = load_knowledge_base(urls)

# Ask a question and stream the answer

question = "Explain the concept of Model Context Protocol."
for chunk in agentic_rag_response(urls, question):
    if hasattr(chunk, "event") and chunk.event == "RunResponseContent":
        print(chunk.content, end="", flush=True)

Running the Application Locally

Launch the complete application using uv (or pip):

uv run streamlit run rag_apps/agentic_rag/main.py

The command initializes the Streamlit server on localhost:8501, presenting a sidebar for URL management and a main chat interface. After loading your knowledge base, queries trigger immediate retrieval and stream responses token-by-token to the frontend.

Summary

• Agentic RAG combines vector retrieval with LLM reasoning, implemented here using Agno's Agent class with search_knowledge=True to automatically fetch context before generation.

• Streaming responses require setting stream=True in agent.run(), yielding RunResponseEvent objects that the UI consumes incrementally for real-time display.

• Local vector storage via LanceDB eliminates cloud dependencies; the tmp/lancedb URI provides persistent storage across application restarts without external infrastructure.

• Modular architecture separates ingestion (load_knowledge_base) from inference (agentic_rag_response), allowing independent scaling and testing of each component.

Frequently Asked Questions

What distinguishes agentic RAG from standard RAG implementations?

Standard RAG typically performs a single retrieval pass before generation, while agentic RAG enables the AI to iteratively search, evaluate, and retrieve additional context during the reasoning process. In the Arindam200/awesome-ai-apps implementation, the Agno agent can invoke multiple knowledge base searches autonomously if the initial retrieval proves insufficient, creating a more dynamic research capability than static context injection.

How does the streaming mechanism work in Agno?

The Agno framework implements streaming through Python generators. When you call agent.run(query, stream=True), it returns a generator that yields RunResponseEvent objects rather than a complete response string. Each event contains metadata about the generation stage; filtering for RunResponseContent events (as seen in lines 53-57 of main.py) provides the actual text tokens, which you can flush to the UI immediately using flush=True in your print or write statements.

Why was LanceDB chosen for this agentic RAG system?

LanceDB provides a serverless, embedded vector database that stores data locally in the tmp/lancedb directory. This choice eliminates network latency and external service dependencies, making the repository immediately runnable without Docker container orchestration or cloud API keys beyond OpenAI. The implementation uses LanceDb with OpenAIEmbedder integration, ensuring compatibility with the 1536-dimensional embeddings produced by the text-embedding-3-small model.

Can I customize the embedding model or LLM provider in this implementation?

Yes, the architecture supports swapping components through constructor parameters. Replace OpenAIEmbedder(id="text-embedding-3-small") (line 47) with any Agno-compatible embedder, such as Ollama for local embeddings. Similarly, change OpenAIChat(id="gpt-5-2025-08-07") (line 61) to Claude(model="claude-3-opus-20240229") or another supported chat model. The UrlKnowledge and Agent classes abstract the underlying provider, requiring no changes to the ingestion or streaming logic when switching models.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}