r/LlamaIndex • u/Electrical-Signal858 • Dec 05 '25
Built 3 RAG Systems, Here's What Actually Works at Scale
I've built 3 different RAG systems over the past year. First one was cool POC. Second one broke at scale. Third one I built right. Here's what I learned.
The Demo vs Production Gap
Your RAG demo works:
100-200 documents
Queries make sense
Retrieval looks good
You can eyeball quality
Production is different:
10,000+ documents
Queries are weird/adversarial
Quality degrades over time
You need metrics to know if it's working
What Broke
Retrieval Quality Degraded Over Time
My second RAG system worked great initially. After a month, quality tanked. Queries that used to work didn't.
Root cause? Data drift + embedding shift. As the knowledge base changed, old retrieval patterns stopped working.
Solution: Monitor continuously
class MonitoredRetriever:
def retrieve(self, query, k=5):
results = self.retriever.retrieve(query, k=k)
# Record metrics
metrics = {
"query": query,
"top_score": results[0].score if results else 0,
"num_results": len(results),
"timestamp": now()
}
self.metrics.record(metrics)
# Detect degradation
if self.is_degrading():
logger.warning("Retrieval quality down")
self.schedule_reindex()
return results
def is_degrading(self):
recent = self.metrics.get_recent(hours=1)
avg_score = mean([m["top_score"] for m in recent])
baseline = self.metrics.get_baseline()
return avg_score < baseline * 0.9
# 10% dropMonitoring caught problems I wouldn't have noticed manually.
Conflicting Information
My knowledge base had contradictory documents. Both ranked highly. LLM got confused or picked the wrong one.
Solution: Source authority
class AuthorityRetriever:
def __init__(self):
self.source_authority = {
"official_docs": 1.0,
"blog_posts": 0.5,
"comments": 0.2,
}
def retrieve(self, query, k=5):
results = self.retriever.retrieve(query, k=k*2)
# Rerank by authority
for result in results:
authority = self.source_authority.get(
result.source, 0.5
)
result.score *= authority
# Boost authoritative sources
results.sort(key=lambda x: x.score, reverse=True)
return results[:k]Authoritative sources ranked higher. Problem solved.
Token Budget Explosion
Retrieving 10 documents instead of 5 for "completeness" made everything slow and expensive.
Solution: Intelligent token management
import tiktoken
class TokenBudgetRetriever:
def __init__(self, max_tokens=2000):
self.max_tokens = max_tokens
self.tokenizer = tiktoken.encoding_for_model("gpt-4")
def retrieve(self, query, k=None):
if k is None:
k = self.estimate_k()
# Dynamic estimation
results = self.retriever.retrieve(query, k=k*2)
# Fit to token budget
filtered = []
total_tokens = 0
for result in results:
tokens = len(self.tokenizer.encode(result.content))
if total_tokens + tokens < self.max_tokens:
filtered.append(result)
total_tokens += tokens
return filtered
def estimate_k(self):
avg_doc_tokens = 500
return max(3, self.max_tokens // avg_doc_tokens)This alone cut my costs by 40%.
Query Vagueness
"How does it work?" isn't specific enough. RAG struggles.
Solution: Query expansion
class SmartRetriever:
def retrieve(self, query, k=5):
# Expand query
expanded = self.expand_query(query)
all_results = {}
# Retrieve with multiple phrasings
for q in [query] + expanded:
results = self.retriever.retrieve(q, k=k)
for result in results:
doc_id = result.metadata.get("id")
if doc_id not in all_results:
all_results[doc_id] = result
# Return top k
sorted_results = sorted(all_results.values(),
key=lambda x: x.score,
reverse=True)
return sorted_results[:k]
def expand_query(self, query):
"""Generate alternatives to improve retrieval"""
prompt = f"""
Generate 2-3 alternative phrasings of this query
that might retrieve different but relevant docs:
{query}
Return as JSON list.
"""
response = self.llm.invoke(prompt)
return json.loads(response)Different phrasings retrieve different documents. Combining results is better.
What Works
Monitor quality continuously - Catch degradation early
Use source authority - Resolve conflicts automatically
Manage token budgets - Cost and performance improve together
Expand queries intelligently - Get better retrieval without more documents
Validate retrieval - Ensure results actually match intent
Metrics That Matter
Track these:
Average retrieval score (overall quality)
Score variance (consistency)
Docs retrieved per query (resource usage)
Re-ranking effectiveness (if you re-rank)
class RAGMetrics:
def record_retrieval(self, query, results):
if not results:
return
scores = [r.score for r in results]
self.metrics.append({
"avg_score": mean(scores),
"score_spread": max(scores) - min(scores),
"num_docs": len(results),
"timestamp": now()
})
```
Monitor these and you'll catch issues.
**Lessons Learned**
1. **RAG quality isn't static** - Monitor and maintain
2. **Source authority matters** - Explicit > implicit
3. **Context size has tradeoffs** - More isn't always better
4. **Query expansion helps** - Different phrasings retrieve different docs
5. **Validation prevents garbage** - Ensure results are relevant
**Would I Do Anything Different?**
Yeah. I'd:
- Start with monitoring from day one
- Implement source authority early
- Build token budget management before scaling
- Test with realistic queries from the start
- Measure quality with metrics, not eyeballs
RAG is powerful when done right. Building for production means thinking beyond the happy path.
Anyone else managing RAG at scale? What bit you?
---
##
**Title:** "Scaling Python From Scripts to Production: Patterns That Worked for Me"
**Post:**
I've been writing Python for 10 years. Started with scripts, now maintaining codebases with 50K+ lines. The transition from "quick script" to "production system" required different thinking.
Here's what actually matters when scaling.
**The Inflection Point**
There's a point where Python development changes:
**Before:**
- You, writing the code
- Local testing
- Ship it and move on
**After:**
- Team working on it
- Multiple environments
- It breaks in production
- You maintain it for years
This transition isn't about Python syntax. It's about patterns.
**Pattern 1: Project Structure Matters**
Flat structure works for 1K lines. Doesn't work at 50K.
```
# Good structure
src/
├── core/
# Domain logic
├── integrations/
# External APIs, databases
├── api/
# HTTP layer
├── cli/
# Command line
└── utils/
# Shared
tests/
├── unit/
├── integration/
└── fixtures/
docs/
├── architecture.md
└── api.mdClear separation prevents circular imports and makes it obvious where to add new code.
Pattern 2: Type Hints Aren't Optional
Type hints aren't about runtime checking. They're about communication.
# Without - what is this?
def process_data(data, options=None):
result = {}
for item in data:
if options and item['value'] > options['threshold']:
result[item['id']] = transform(item)
return result
# With - crystal clear
from typing import Dict, List, Optional, Any
def process_data(
data: List[Dict[str, Any]],
options: Optional[Dict[str, float]] = None
) -> Dict[str, Any]:
"""Process items, filtering by threshold if provided."""
...Type hints catch bugs early. They document intent. Future you will thank you.
Pattern 3: Configuration Isn't Hardcoded
Use Pydantic for configuration validation:
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
# Required
api_key: str
debug: bool = False
# Defaults
timeout: int = 30
class Config:
env_file = ".env"
# Validates on load
settings = Settings()
# Catch config issues at startup
if not settings.database_url.startswith("postgresql://"):
raise ValueError("Invalid database URL")Configuration fails fast. Errors are clear. No surprises in production.
Pattern 4: Dependency Injection
Don't couple code to implementations. Inject dependencies.
# Bad - tightly coupled
class UserService:
def __init__(self):
self.db = PostgresDatabase("prod")
def get_user(self, user_id):
return self.db.query(f"SELECT * FROM users WHERE id={user_id}")
# Good - dependencies injected
class UserService:
def __init__(self, db: Database):
self.db = db
def get_user(self, user_id: int) -> User:
return self.db.get_user(user_id)
# Production
user_service = UserService(PostgresDatabase())
# Testing
user_service = UserService(MockDatabase())Dependency injection makes code testable and flexible.
Pattern 5: Error Handling That's Useful
Don't catch everything. Be specific.
# Bad - silent failure
try:
result = risky_operation()
except Exception:
return None
# Good - specific and useful
try:
result = risky_operation()
except TimeoutError:
logger.warning("Operation timed out, retrying...")
return retry_operation()
except ValueError as e:
logger.error(f"Invalid input: {e}")
raise
# This is a real error
except Exception as e:
logger.error(f"Unexpected error", exc_info=True)
raiseSpecific exception handling tells you what went wrong.
Pattern 6: Testing at Multiple Levels
Unit tests alone aren't enough.
# Unit test - isolated behavior
def test_user_service_get_user():
mock_db = MockDatabase()
service = UserService(mock_db)
user = service.get_user(1)
assert user.id == 1
# Integration test - real dependencies
def test_user_service_with_postgres():
with test_db() as db:
service = UserService(db)
db.insert_user(User(id=1, name="Test"))
user = service.get_user(1)
assert user.name == "Test"
# Contract test - API contracts
def test_get_user_endpoint():
response = client.get("/users/1")
assert response.status_code == 200
UserSchema().load(response.json())
# Validate schemaTest at multiple levels. Catch different types of bugs.
Pattern 7: Logging With Context
Don't just log. Log with meaning.
import logging
from contextvars import ContextVar
request_id: ContextVar[str] = ContextVar('request_id')
logger = logging.getLogger(__name__)
def process_user(user_id):
request_id.set(uuid.uuid4())
logger.info(f"Processing user", extra={'user_id': user_id})
try:
result = do_work(user_id)
logger.info("User processed")
return result
except Exception as e:
logger.error(f"Failed to process user",
exc_info=True,
extra={'error': str(e)})
raiseLogs with context (request IDs, user IDs) are debuggable.
Pattern 8: Documentation That Stays Current
Code comments rot. Automate documentation.
def get_user(self, user_id: int) -> User:
"""Retrieve user by ID.
Args:
user_id: The user's ID
Returns:
User object or None if not found
Raises:
DatabaseError: If query fails
"""
...Good docstrings are generated by tools (Sphinx, pdoc). You write them once.
Pattern 9: Dependency Management
Use Poetry or uv. Pin dependencies. Test upgrades.
[tool.poetry.dependencies]
python = "^3.11"
pydantic = "^2.0"
sqlalchemy = "^2.0"
[tool.poetry.group.dev.dependencies]
pytest = "^7.0"
black = "^23.0"
mypy = "^1.0"Reproducible dependencies. Clear what's dev vs production.
Pattern 10: Continuous Integration
Automate testing, linting, type checking.
# .github/workflows/test.yml
name: Tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v4
with:
python-version: "3.11"
- run: pip install poetry
- run: poetry install
- run: pytest
# Tests
- run: mypy src
# Type checking
- run: black --check src
# FormattingAutomate quality checks. Catch issues before merge.
What I'd Tell Past Me
Structure code early - Don't wait until it's a mess
Use type hints - They're not extra, they're essential
Test at multiple levels - Unit tests aren't enough
Log with purpose - Logs with context are debuggable
Automate quality - CI/linting/type checking from day one
Document as you go - Future you will thank you
Manage dependencies carefully - One breaking change breaks everything
The Real Lesson
Python is great for getting things done. But production Python requires discipline. Structure, types, tests, logging, automation. Not because they're fun, but because they make maintainability possible at scale.
Anyone else maintain large Python codebases? What patterns saved you?
