Understanding the Landscape: LlamaIndex, Django, and FastAPI
Modern developers face a critical architectural decision when building data-driven applications: which framework best serves the project's core purpose? LlamaIndex, Django, and FastAPI occupy distinct niches in the Python ecosystem, yet they increasingly overlap as AI-powered features become standard requirements. This tutorial provides a complete, hands-on comparison to help you make informed decisions.
What Is LlamaIndex?
LlamaIndex is a data orchestration framework designed specifically for Large Language Model (LLM) applications. It acts as an intermediary between your data sources and LLMs, providing indexing, retrieval, and query synthesis capabilities. Rather than being a web framework, LlamaIndex is a data engine that connects structured and unstructured data to AI models.
Core concepts include:
- Documents — raw data units (PDFs, database rows, API responses)
- Nodes — chunked, processed pieces of documents ready for indexing
- Indexes — data structures optimized for LLM retrieval (vector stores, tree indexes, keyword tables)
- Query Engines — interfaces that combine retrieval and LLM synthesis
- Agents — autonomous decision-making loops that select tools and orchestrate multi-step reasoning
# Basic LlamaIndex pipeline
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.llms.openai import OpenAI
# 1. Load documents from a directory
documents = SimpleDirectoryReader('./data').load_data()
# 2. Create an index over those documents
index = VectorStoreIndex.from_documents(documents)
# 3. Configure the LLM
llm = OpenAI(model="gpt-4", temperature=0.1)
# 4. Create a query engine with custom settings
query_engine = index.as_query_engine(
llm=llm,
similarity_top_k=5,
response_mode="compact"
)
# 5. Execute a natural language query
response = query_engine.query(
"What are the key revenue drivers mentioned in the Q4 report?"
)
print(response)
What Is Django?
Django is a full-stack, batteries-included web framework built around the Model-View-Template (MVT) architectural pattern. It provides an ORM, authentication system, admin interface, form handling, and template engine out of the box. Django excels at building monolithic web applications where server-rendered HTML, database modeling, and rapid development velocity are priorities.
Core concepts include:
- Models — Python classes that define database schema via Django ORM
- Views — functions or classes that handle HTTP requests and return responses
- Templates — Django Template Language files for server-side rendering
- URLconf — declarative URL routing system
- Admin — automatically generated administrative interface
# Django models.py — defining database schema
from django.db import models
from django.contrib.auth.models import User
class Project(models.Model):
STATUS_CHOICES = [
('planning', 'Planning'),
('active', 'Active'),
('completed', 'Completed'),
('archived', 'Archived'),
]
name = models.CharField(max_length=200)
description = models.TextField(blank=True)
owner = models.ForeignKey(User, on_delete=models.CASCADE, related_name='projects')
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='planning')
budget = models.DecimalField(max_digits=12, decimal_places=2, null=True, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
ordering = ['-created_at']
indexes = [
models.Index(fields=['status', 'owner']),
]
def __str__(self):
return f"{self.name} ({self.get_status_display()})"
# Django views.py — handling requests
from django.shortcuts import render, get_object_or_404, redirect
from django.contrib.auth.decorators import login_required
from django.contrib import messages
from .models import Project
from .forms import ProjectForm
@login_required
def project_detail(request, project_id):
project = get_object_or_404(
Project.objects.select_related('owner'),
id=project_id
)
tasks = project.tasks.filter(completed=False).order_by('priority')
context = {
'project': project,
'active_tasks': tasks,
'completion_percentage': project.get_completion_rate(),
}
return render(request, 'projects/detail.html', context)
@login_required
def project_update(request, project_id):
project = get_object_or_404(Project, id=project_id, owner=request.user)
if request.method == 'POST':
form = ProjectForm(request.POST, instance=project)
if form.is_valid():
form.save()
messages.success(request, 'Project updated successfully.')
return redirect('project_detail', project_id=project.id)
else:
form = ProjectForm(instance=project)
return render(request, 'projects/form.html', {'form': form, 'project': project})
What Is FastAPI?
FastAPI is a modern, high-performance web framework optimized for building RESTful and GraphQL APIs. It leverages Python type hints to provide automatic request validation, serialization, and interactive API documentation (OpenAPI/Swagger). FastAPI is built on Starlette for web handling and Pydantic for data validation, making it ideal for microservices, real-time applications, and AI model serving endpoints.
Core concepts include:
- Path Operations — decorated functions tied to specific HTTP methods and URL paths
- Pydantic Models — type-safe request/response schemas with automatic validation
- Dependency Injection — reusable logic via FastAPI's dependency system
- Middleware — hooks into the request/response lifecycle
- Background Tasks — asynchronous post-response processing
# FastAPI application with AI model serving
from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
import asyncio
app = FastAPI(
title="Document Analysis API",
version="1.0.0",
description="Serves LlamaIndex-powered document queries"
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
# --- Pydantic schemas ---
class DocumentUpload(BaseModel):
filename: str = Field(..., min_length=1, max_length=255)
content: str = Field(..., min_length=10)
tags: Optional[List[str]] = Field(default_factory=list)
class QueryRequest(BaseModel):
question: str = Field(..., min_length=3, max_length=1000)
document_ids: Optional[List[str]] = None
max_tokens: int = Field(default=500, ge=50, le=4000)
class QueryResponse(BaseModel):
answer: str
sources: List[str]
confidence_score: float = Field(..., ge=0.0, le=1.0)
processing_time_ms: int
# --- Dependency for index access ---
async def get_index():
# In production, this would connect to a persistent vector store
from app.services.index_manager import IndexManager
manager = IndexManager()
try:
index = await manager.get_or_create_index()
return index
except Exception as e:
raise HTTPException(status_code=503, detail=f"Index unavailable: {str(e)}")
# --- API endpoints ---
@app.post("/documents", status_code=201)
async def ingest_document(
upload: DocumentUpload,
background_tasks: BackgroundTasks,
index = Depends(get_index)
):
"""Ingest a new document into the LlamaIndex pipeline."""
document_id = f"doc_{datetime.utcnow().timestamp()}"
# Store raw document and trigger async indexing
background_tasks.add_task(
index.insert_document,
document_id=document_id,
text=upload.content,
metadata={"filename": upload.filename, "tags": upload.tags}
)
return {
"document_id": document_id,
"status": "queued",
"estimated_processing": "15-30 seconds"
}
@app.get("/documents/{document_id}/status")
async def check_processing_status(document_id: str):
"""Check the indexing status of a document."""
from app.services.index_manager import IndexManager
manager = IndexManager()
status = await manager.get_document_status(document_id)
if status is None:
raise HTTPException(status_code=404, detail="Document not found")
return status
@app.post("/query", response_model=QueryResponse)
async def query_documents(
request: QueryRequest,
index = Depends(get_index)
):
"""Execute a natural language query against indexed documents."""
import time
start_time = time.time()
# Build query engine with filters if document_ids specified
query_engine = index.as_query_engine(
similarity_top_k=8,
response_mode="tree_summarize"
)
# Execute query
response = await query_engine.aquery(request.question)
processing_time = int((time.time() - start_time) * 1000)
return QueryResponse(
answer=str(response),
sources=[node.metadata.get('filename', 'unknown') for node in response.source_nodes],
confidence_score=getattr(response, 'confidence', 0.85),
processing_time_ms=processing_time
)
@app.get("/health")
async def health_check():
"""Lightweight health endpoint for orchestrators."""
return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}
Why This Comparison Matters
🚀 Deploy your AI agent in 10 minutes
Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.
Try it free →These three frameworks represent fundamentally different architectural philosophies, yet they increasingly intersect in production AI applications. Understanding their distinctions prevents costly architectural mistakes:
- Wrong framework choice leads to rewrites — Building a real-time API in Django requires significant customization; building a monolithic CMS in FastAPI means reinventing Django's admin
- AI integration patterns differ dramatically — LlamaIndex plugs into FastAPI naturally as a service, but into Django as part of the data layer
- Team composition matters — Frontend-heavy teams benefit from Django's templates; backend/ML teams thrive with FastAPI's API-first design; data engineers gravitate toward LlamaIndex's data pipeline approach
- Performance profiles diverge — FastAPI's async capabilities handle high-concurrency AI workloads; Django's sync ORM requires careful tuning for LLM response streaming
Practical Integration Patterns
Pattern 1: FastAPI + LlamaIndex — The AI Microservice
This is the most natural pairing for AI-powered APIs. FastAPI serves as the HTTP layer while LlamaIndex handles all LLM orchestration internally. The key advantage is FastAPI's native async support, which aligns perfectly with LlamaIndex's async query capabilities.
# Complete runnable AI microservice combining FastAPI + LlamaIndex
# app/main.py
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from llama_index.core import VectorStoreIndex, Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.node_parser import SentenceWindowNodeParser
import chromadb
from chromadb.config import Settings as ChromaSettings
import asyncio
import json
app = FastAPI(title="RAG Microservice")
# Configure global LlamaIndex settings
Settings.llm = OpenAI(model="gpt-4o", temperature=0.2)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.node_parser = SentenceWindowNodeParser.from_defaults(
window_size=3,
window_overlap=1,
)
# Initialize persistent vector store
chroma_client = chromadb.PersistentClient(
path="./vector_store",
settings=ChromaSettings(anonymized_telemetry=False)
)
class SearchRequest(BaseModel):
query: str
top_k: int = 5
stream: bool = False
class SearchResult(BaseModel):
text: str
score: float
metadata: dict
@app.on_event("startup")
async def startup_event():
"""Warm up the index on application startup."""
global query_engine
collection = chroma_client.get_or_create_collection("documents")
vector_store = ChromaVectorStore(chroma_collection=collection)
index = VectorStoreIndex.from_vector_store(vector_store)
query_engine = index.as_query_engine(
similarity_top_k=5,
streaming=True,
)
print(f"✅ Index warmed up with {collection.count()} documents")
@app.post("/search", response_model=list[SearchResult])
async def search_documents(request: SearchRequest):
"""Execute semantic search with optional streaming."""
if request.stream:
# Streaming response for real-time token delivery
async def token_generator():
streaming_response = await query_engine.astream(request.query)
async for token in streaming_response:
yield f"data: {json.dumps({'token': token})}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(
token_generator(),
media_type="text/event-stream"
)
# Non-streaming response
response = await query_engine.aquery(request.query)
return [
SearchResult(
text=node.text,
score=node.score,
metadata=node.metadata
)
for node in response.source_nodes
]
@app.post("/ingest")
async def ingest_document(content: str, filename: str = "untitled"):
"""Add a document to the index in real-time."""
from llama_index.core import Document
doc = Document(text=content, metadata={"filename": filename})
index.insert_document(doc)
return {
"status": "indexed",
"filename": filename,
"chunk_count": len(doc.text.split('\n'))
}
Pattern 2: Django + LlamaIndex — AI-Enhanced Monolith
Django's strength is its ORM and admin interface. Integrating LlamaIndex allows you to augment traditional CRUD operations with AI-powered features like semantic search, automated categorization, or intelligent report generation. The key pattern is treating LlamaIndex indexes as Django services that consume model data.
# Django service layer integrating LlamaIndex
# app/services/knowledge_base.py
from django.core.cache import cache
from django.db.models import Q
from llama_index.core import VectorStoreIndex, Document
from llama_index.core.retrievers import QueryFusionRetriever
from llama_index.llms.openai import OpenAI
from typing import List, Dict, Any
import hashlib
class KnowledgeBaseService:
"""
Singleton service that maintains a LlamaIndex over Django model data.
Uses Django's cache framework for index persistence.
"""
def __init__(self):
self.llm = OpenAI(model="gpt-4o", temperature=0.1)
self._index = None
def _build_index_from_database(self):
"""Construct LlamaIndex documents from Django ORM queries."""
from apps.articles.models import Article
from apps.support.models import FAQ
documents = []
# Index published articles
articles = Article.objects.filter(
status='published'
).select_related('author').prefetch_related('tags')
for article in articles:
documents.append(Document(
text=f"Title: {article.title}\n\n{article.content}",
metadata={
"source_type": "article",
"id": str(article.id),
"author": article.author.get_full_name(),
"published": article.published_at.isoformat(),
"tags": [tag.name for tag in article.tags.all()],
}
))
# Index FAQ entries
faqs = FAQ.objects.filter(is_active=True)
for faq in faqs:
documents.append(Document(
text=f"Q: {faq.question}\nA: {faq.answer}",
metadata={
"source_type": "faq",
"id": str(faq.id),
"category": faq.category.name,
}
))
return VectorStoreIndex.from_documents(documents)
def get_index(self):
"""Get or create index with Django cache integration."""
cache_key = "llama_index_version"
current_hash = self._compute_data_hash()
cached_hash = cache.get(cache_key)
if self._index is None or cached_hash != current_hash:
self._index = self._build_index_from_database()
cache.set(cache_key, current_hash, timeout=3600) # 1 hour
return self._index
def _compute_data_hash(self):
"""Compute hash of underlying data to detect staleness."""
from apps.articles.models import Article
from apps.support.models import FAQ
article_count = Article.objects.filter(status='published').count()
faq_count = FAQ.objects.filter(is_active=True).count()
latest_update = Article.objects.filter(
status='published'
).order_by('-updated_at').values_list('updated_at', flat=True).first()
hash_input = f"{article_count}:{faq_count}:{latest_update}"
return hashlib.md5(hash_input.encode()).hexdigest()
def semantic_search(self, query: str, top_k: int = 5) -> List[Dict[str, Any]]:
"""Execute semantic search across all indexed content."""
index = self.get_index()
retriever = QueryFusionRetriever(
index.as_retriever(similarity_top_k=top_k),
num_queries=3,
llm=self.llm,
)
results = retriever.retrieve(query)
return [
{
"text": node.text[:500] + "..." if len(node.text) > 500 else node.text,
"score": node.score,
"source_type": node.metadata.get("source_type"),
"source_id": node.metadata.get("id"),
"metadata": node.metadata,
}
for node in results
]
def generate_synthesis(self, query: str, context_nodes) -> str:
"""Generate a coherent response from retrieved context."""
index = self.get_index()
query_engine = index.as_query_engine(
llm=self.llm,
response_mode="compact",
similarity_top_k=3
)
response = query_engine.query(query)
return str(response)
# Usage in Django views
# app/views/search.py
from django.http import JsonResponse
from django.views.decorators.http import require_GET
from django.contrib.auth.decorators import login_required
from .services.knowledge_base import KnowledgeBaseService
knowledge_service = KnowledgeBaseService()
@require_GET
@login_required
def ai_search_view(request):
query = request.GET.get('q', '').strip()
if not query or len(query) < 3:
return JsonResponse({"error": "Query too short"}, status=400)
try:
results = knowledge_service.semantic_search(query, top_k=5)
synthesis = knowledge_service.generate_synthesis(query, results)
return JsonResponse({
"query": query,
"ai_synthesis": synthesis,
"sources": results,
"source_count": len(results),
})
except Exception as e:
return JsonResponse({"error": str(e)}, status=500)
Pattern 3: Full Stack — FastAPI Backend + Django Admin + LlamaIndex Engine
For sophisticated applications, you can combine all three: FastAPI handles public API traffic with async efficiency, Django manages internal tools and admin interfaces, and LlamaIndex powers the AI intelligence layer consumed by both. This pattern is increasingly common in enterprise AI platforms.
# Architecture: FastAPI (API) ←→ LlamaIndex (AI Engine) ←→ Django (Admin/Management)
# The LlamaIndex service becomes a shared internal service
# shared_services/llama_engine.py — Used by BOTH FastAPI and Django
from llama_index.core import VectorStoreIndex
from llama_index.core.storage.storage_context import StorageContext
from llama_index.vector_stores.qdrant import QdrantVectorStore
from qdrant_client import QdrantClient
import asyncio
from typing import Optional
class LlamaEngineService:
"""
Singleton AI engine shared across the platform.
FastAPI calls this for real-time queries.
Django management commands call this for indexing.
"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._initialized = False
return cls._instance
async def initialize(self):
if self._initialized:
return
# Connect to persistent Qdrant vector store
self.client = QdrantClient(host="qdrant-service", port=6333)
self.vector_store = QdrantVectorStore(
client=self.client,
collection_name="platform_knowledge",
)
self.storage_context = StorageContext.from_defaults(
vector_store=self.vector_store
)
self.index = VectorStoreIndex.from_vector_store(
self.vector_store
)
self._initialized = True
async def query(self, question: str, filters: Optional[dict] = None) -> dict:
"""Execute query — called by FastAPI endpoints."""
await self.initialize()
query_engine = self.index.as_query_engine(
similarity_top_k=10,
streaming=False,
)
response = await query_engine.aquery(question)
return {
"answer": str(response),
"sources": [
{
"text": node.text,
"metadata": node.metadata,
"score": node.score,
}
for node in response.source_nodes
],
}
async def ingest_batch(self, documents: list) -> int:
"""Batch ingest — called by Django management commands."""
await self.initialize()
from llama_index.core import Document as LlamaDoc
llama_docs = [
LlamaDoc(text=doc["text"], metadata=doc.get("metadata", {}))
for doc in documents
]
self.index.insert_documents(llama_docs)
return len(llama_docs)
# FastAPI side — lightweight async endpoint
# fastapi_app/endpoints/ai.py
from fastapi import APIRouter, Depends
from shared_services.llama_engine import LlamaEngineService
router = APIRouter(prefix="/ai", tags=["AI"])
engine = LlamaEngineService()
@router.post("/ask")
async def ask_question(question: str):
"""Public-facing AI query endpoint."""
result = await engine.query(question)
return {
"question": question,
"answer": result["answer"],
"sources_count": len(result["sources"]),
}
# Django side — management command for batch indexing
# django_app/management/commands/index_content.py
from django.core.management.base import BaseCommand
from django.db.models import Count
from apps.content.models import Article, Documentation
from shared_services.llama_engine import LlamaEngineService
import asyncio
class Command(BaseCommand):
help = "Index all published content into the LlamaIndex engine"
def handle(self, *args, **options):
# Gather content from Django ORM
articles = Article.objects.filter(status='published')
docs = Documentation.objects.filter(is_active=True)
batch = []
for article in articles:
batch.append({
"text": f"{article.title}\n{article.content}",
"metadata": {
"django_model": "Article",
"id": article.id,
"author": article.author.username,
}
})
for doc in docs:
batch.append({
"text": doc.content,
"metadata": {
"django_model": "Documentation",
"id": doc.id,
"version": doc.version,
}
})
# Push to shared engine
engine = LlamaEngineService()
count = asyncio.run(engine.ingest_batch(batch))
self.stdout.write(
self.style.SUCCESS(f"✅ Indexed {count} documents into Qdrant")
)
Framework Selection Decision Matrix
Use this decision framework when choosing your primary framework:
- Choose LlamaIndex as your core when your application's primary value is AI-powered data synthesis, multi-source retrieval, or agent-based reasoning. LlamaIndex is not a web framework — pair it with FastAPI or Django for HTTP serving
- Choose Django as your core when you need server-rendered HTML, rapid CRUD development, built-in admin, authentication, and you're building a traditional web application that happens to have AI features
- Choose FastAPI as your core when you're building APIs first — especially async APIs that serve mobile apps, SPAs, or act as AI model serving endpoints with streaming requirements
Best Practices Across All Three
1. Dependency Injection Patterns
All three frameworks benefit from clean dependency management. FastAPI has a built-in dependency injection system. Django uses class-based views and service layers. LlamaIndex uses global Settings objects.
# FastAPI dependency injection
async def get_llama_service():
service = LlamaService()
await service.initialize()
try:
yield service
finally:
await service.cleanup()
@app.get("/query")
async def query(q: str, service = Depends(get_llama_service)):
return await service.process(q)
# Django class-based view with dependency injection
class LlamaMixin:
"""Mixin that injects LlamaIndex service into Django views."""
def setup(self, request, *args, **kwargs):
super().setup(request, *args, **kwargs)
self.llama_service = get_llama_service()
def dispatch(self, request, *args, **kwargs):
return super().dispatch(request, *args, **kwargs)
class SearchView(LlamaMixin, TemplateView):
template_name = "search/results.html"
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context['results'] = self.llama_service.search(
self.request.GET.get('q', '')
)
return context
# LlamaIndex global Settings pattern
from llama_index.core import Settings
Settings.llm = OpenAI(model="gpt-4o")
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-large")
Settings.chunk_size = 1024
Settings.chunk_overlap = 128
2. Caching and Index Persistence
LlamaIndex indexes are computationally expensive to rebuild. Cache them aggressively using vector store persistence (Qdrant, Pinecone, ChromaDB). Django's cache framework works well for HTTP-level caching. FastAPI benefits from in-memory caching with async locks.
# Production-grade index caching with FastAPI
import asyncio
from typing import Optional
from datetime import datetime, timedelta
class IndexCache:
def __init__(self, ttl_seconds: int = 3600):
self._index: Optional[VectorStoreIndex] = None
self._last_built: Optional[datetime] = None
self._ttl = ttl_seconds
self._lock = asyncio.Lock()
async def get_index(self) -> VectorStoreIndex:
now = datetime.utcnow()
# Fast path — index exists and is fresh
if self._index and self._last_built:
if (now - self._last_built).seconds < self._ttl:
return self._index
# Slow path — rebuild with lock to prevent stampedes
async with self._lock:
# Double-check after acquiring lock
if self._index and self._last_built:
if (now - self._last_built).seconds < self._ttl:
return self._index
# Perform expensive rebuild
self._index = await self._build_index()
self._last_built = now
return self._index
async def _build_index(self):
# Expensive operation — connect to vector store, load documents
documents = await self._load_all_documents()
return VectorStoreIndex.from_documents(documents)
3. Error Handling and Graceful Degradation
When integrating LLMs via LlamaIndex, failures are inevitable (rate limits, token limits, model downtime). Implement circuit breakers and fallback responses across all frameworks.
# Universal error handler for LLM operations
from functools import wraps
import time
from typing import Callable, Any
class CircuitBreaker:
def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 60):
self.failure_count = 0
self.last_failure_time = None
self.threshold = failure_threshold
self.recovery = recovery_timeout
self.state = "closed" # closed, open, half-open
def __call__(self, func: Callable) -> Callable:
@wraps(func)
async def wrapper(*args, **kwargs):
if self.state == "open":
if time.time() - self.last_failure_time > self.recovery:
self.state = "half-open"
else:
raise ServiceUnavailableError(
"Circuit breaker open — LLM service unavailable"
)
try:
result = await func(*args, **kwargs)
if self.state == "half-open":
self.state = "closed"
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.threshold:
self.state = "open"
# Return fallback response instead of crashing
return {
"answer": "I'm currently unable to process this query. "
"Please try again shortly.",
"fallback": True,
"error": str(e),
}
return wrapper
# Usage in FastAPI
llama_circuit_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=120)
@llama_circuit_breaker
async def safe_llama_query(query: str) -> dict:
"""Query with automatic circuit breaking."""
response = await query_engine.aquery(query)
return {"answer": str(response), "fallback": False}
# Usage in Django
from django.core.cache import cache
def cached_llama_query_with_fallback(query: str) -> dict:
"""Django-style query with cache fallback."""
cache_key = f"llama_response_{hash(query)}"
cached = cache.get(cache_key)
if cached:
return cached
try:
response = query_engine.query(query)
result = {"answer": str(response), "cached": False}
cache.set(cache_key, result, timeout=300)
return result
except Exception:
# Check for stale cache as fallback
stale = cache.get(cache_key, version=2)
if stale:
return {**stale, "stale_fallback": True}
return {"answer": "Service temporarily unavailable", "fallback": True}
4. Streaming Responses
LLM responses can take seconds to generate. Streaming provides immediate user feedback. FastAPI excels here with SSE (Server-Sent Events). Django requires additional configuration with ASGI servers. LlamaIndex provides native streaming support.
# FastAPI streaming with LlamaIndex
from fastapi.responses import StreamingResponse
import asyncio
@app.post("/stream-query")
async def stream_query(question: str):
async def event_generator():
# Send initial metadata
yield f"data: {json.dumps({'type': 'start', 'timestamp': datetime.utcnow().isoformat()})}\n\n"
# Stream LlamaIndex response tokens
streaming_response = await query_engine.astream(question)
full_response = ""
async for token in streaming_response:
full_response += token
yield f"data: {json.dumps({'type': 'token', 'content': token})}\n\n"
# Send completion with full response and sources
yield f"data: {json.dumps({'type': 'complete', 'full_text': full_response, 'sources': [str(n) for n in streaming_response.source_nodes]})}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
"Connection": "keep-alive",
}
)
# Django streaming (requires Daphne/ASGI)
# In views.py with Django 4.2+
from django.http import StreamingHttpResponse
from django.views.decorators.async import async_csrf_exempt
@async_csrf_exempt
async def django_stream_view(request):
async def content_stream():
response = await query_engine.astream(request.GET.get('q', ''))
async for chunk in response:
yield chunk.encode('utf-8')
return StreamingHttpResponse(
content_stream(),
content_type='text/plain',
)
Security Considerations
Each framework presents distinct security challenges when integrated with LLMs:
- Prompt injection prevention — Sanitize user inputs before they reach LlamaIndex query engines. Implement input validation in FastAPI/Pydantic schemas or Django forms
- Data exfiltration — LlamaIndex retrievers may leak sensitive data from your indexes. Implement document-level access control before indexing
- Token usage abuse — Rate limit LLM queries at the framework level. FastAPI middleware, Django throttling, or LlamaIndex callback handlers can track token consumption
# Input sanitization across frameworks
import re
from typing import Optional
class PromptSanitizer:
"""Shared sanitization logic usable in FastAPI and Django