In part 2 I built the data layer — a relational database for bookings and customers, and a vector store for the policy documents. This part builds the agent that uses that vector store: the Policy Agent. It answers policy questions — refunds, terms, fares — and the whole point is that every answer is grounded in the company's actual documents, not the model's memory.
Why policy questions are hard
Policy is exactly where human support struggles, and where a grounded AI earns its place. People forget the details or answer inconsistently. The documents are long and change often. The awkward questions need two or three sections combined and read carefully. And for anything compliance-related, "roughly right" isn't good enough — you need the real wording, and ideally a citation back to it.
That's what RAG is for. Retrieval-augmented generation means the model retrieves the relevant policy text first and answers from that — it doesn't get to improvise. For policy, that's the difference between a system you can trust and one you can't.
How the Policy Agent works
The flow is straightforward: take the question, find the policy chunks that actually match it, hand those to the model as context, and have it answer from them — with a citation back to the source.
Customer Query
↓
Query Embedding (Vector Generation)
↓
Vector Similarity Search
↓
Document Chunk Retrieval (Top-K)
↓
Re-ranking & Filtering
↓
Context Assembly
↓
LLM Prompt Construction
↓
Response Generation
↓
Citation Addition
↓
Final ResponseThe critical insight for enterprise AI consulting is that each stage requires careful optimization. A poorly implemented RAG system suffers from irrelevant retrievals, context overflow, or hallucinated responses. Professional-grade implementation demands attention to chunking strategy, embedding model selection, retrieval algorithms, and prompt engineering.
Key architectural decisions:
- Vector Database Choice: a managed store like Pinecone is what you'd reach for at scale; Chroma or Weaviate suit smaller deployments. The UKConnect build uses neither — one policy document fits in a pickled vector store, as Part 2 covered
- Embedding Model: OpenAI's text-embedding-3-large (3072 dimensions) provides excellent semantic understanding for policy documents, though sentence-transformers offer cost-effective alternatives
- Chunking Strategy: Semantic chunking with 500-token overlap preserves context across chunk boundaries, critical for policy documents with interconnected clauses
- Re-ranking Pipeline: a cross-encoder re-ranker should recover relevant chunks that pure vector similarity ranks too low — worth reaching for when compliance accuracy matters, though I haven't measured the gain on a corpus this small
If you're picking a vector database for something real, I'd prioritise query latency and filtering capability over raw storage capacity. Policy systems tend to need filtering by document version, department, or regulatory jurisdiction during retrieval, and that requirement narrows the field faster than scale does.
Document Processing & Chunking Strategy
The foundation of any RAG implementation is document processing. Poor chunking destroys retrieval quality - chunks that are too large dilute semantic meaning, while chunks that are too small lose critical context. Enterprise policy documents present unique challenges: legal language spans multiple paragraphs, cross-references connect distant sections, and hierarchical structures (sections, subsections, clauses) must be preserved.
The chunking strategy uses semantic boundaries rather than arbitrary token counts:
import tiktoken
from typing import List, Dict
import re
class PolicyDocumentProcessor:
def __init__(self, chunk_size: int = 500, overlap: int = 100):
self.chunk_size = chunk_size
self.overlap = overlap
self.encoding = tiktoken.encoding_for_model("gpt-4")
def extract_hierarchy(self, text: str) -> Dict[str, str]:
"""Extract document structure metadata."""
hierarchy = {
'section': None,
'subsection': None,
'article': None
}
# Match section headers (e.g., "Section 3.1: Refund Policy")
section_match = re.search(
r'Section\\s+(\\d+(?:\\.\\d+)?):?\\s*([^\\n]+)',
text,
re.IGNORECASE
)
if section_match:
hierarchy['section'] = section_match.group(2).strip()
# Match article numbers for legal documents
article_match = re.search(
r'Article\\s+(\\d+)',
text,
re.IGNORECASE
)
if article_match:
hierarchy['article'] = article_match.group(1)
return hierarchy
def semantic_chunk(self, document: str, metadata: Dict) -> List[Dict]:
"""Chunk document at semantic boundaries with context preservation."""
chunks = []
# Split on double newlines (paragraph boundaries)
paragraphs = re.split(r'\\n\\n+', document)
current_chunk = []
current_tokens = 0
current_hierarchy = {}
for para in paragraphs:
para_tokens = len(self.encoding.encode(para))
# Extract hierarchy for this paragraph
para_hierarchy = self.extract_hierarchy(para)
if any(para_hierarchy.values()):
current_hierarchy.update(para_hierarchy)
# Check if adding this paragraph exceeds chunk size
if current_tokens + para_tokens > self.chunk_size and current_chunk:
# Create chunk with metadata
chunk_text = '\\n\\n'.join(current_chunk)
chunks.append({
'text': chunk_text,
'metadata': {
**metadata,
**current_hierarchy,
'token_count': current_tokens
}
})
# Start new chunk with overlap
# Keep last paragraph for context continuity
if len(current_chunk) > 1:
current_chunk = [current_chunk[-1], para]
current_tokens = len(self.encoding.encode(
'\\n\\n'.join(current_chunk)
))
else:
current_chunk = [para]
current_tokens = para_tokens
else:
current_chunk.append(para)
current_tokens += para_tokens
# Add final chunk
if current_chunk:
chunks.append({
'text': '\\n\\n'.join(current_chunk),
'metadata': {
**metadata,
**current_hierarchy,
'token_count': current_tokens
}
})
return chunks
def process_policy_document(
self,
filepath: str,
doc_metadata: Dict
) -> List[Dict]:
"""Process a complete policy document into searchable chunks."""
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
# Remove excessive whitespace while preserving structure
content = re.sub(r'\\n{3,}', '\\n\\n', content)
# Chunk the document
chunks = self.semantic_chunk(content, doc_metadata)
# Add chunk IDs for citation
for idx, chunk in enumerate(chunks):
chunk['chunk_id'] = f"{doc_metadata['document_id']}_chunk_{idx}"
return chunksThis approach preserves semantic coherence while maintaining manageable chunk sizes. The overlap strategy ensures that concepts split across paragraphs remain retrievable. I haven't benchmarked this against fixed-size chunking on the UKConnect corpus — it's one policy document, which is too small for the comparison to mean anything.
Critical implementation considerations:
- Token counting accuracy: Use the same tokenizer as your embedding model to avoid chunk size mismatches
- Metadata enrichment: Capture document version, effective date, and regulatory references during processing
- Structure preservation: Maintain hierarchical context (section numbers, article references) in metadata for filtering
- Update handling: Implement version control to deprecate outdated policy chunks when documents are updated
For organizations with complex policy frameworks spanning hundreds of documents, professional AI consulting ensures chunking strategies align with document structure and retrieval requirements. We've seen enterprises waste months on RAG implementations that fail due to poor document processing fundamentals.
Vector Embedding Implementation
Vector embeddings transform text into high-dimensional numerical representations where semantic similarity corresponds to geometric proximity. This mathematical foundation enables semantic search - the ability to find relevant content even when query terms don't exactly match document language.
Scaled up, an embedding pipeline needs batch processing, caching, and error recovery. Here's the shape of one — note this sketch uses managed services rather than the Gemini embeddings and pickle store the UKConnect repo actually runs:
import openai
from typing import List, Dict
import numpy as np
from pinecone import Pinecone, ServerlessSpec
import hashlib
import json
class EmbeddingPipeline:
def __init__(
self,
openai_api_key: str,
pinecone_api_key: str,
index_name: str = "policy-embeddings"
):
self.openai_client = openai.OpenAI(api_key=openai_api_key)
self.pc = Pinecone(api_key=pinecone_api_key)
self.index_name = index_name
self.embedding_model = "text-embedding-3-large"
self.embedding_dimension = 3072
# Initialize or connect to Pinecone index
self._initialize_index()
def _initialize_index(self):
"""Create Pinecone index with optimal configuration."""
existing_indexes = [idx.name for idx in self.pc.list_indexes()]
if self.index_name not in existing_indexes:
self.pc.create_index(
name=self.index_name,
dimension=self.embedding_dimension,
metric='cosine', # Cosine similarity for text embeddings
spec=ServerlessSpec(
cloud='aws',
region='us-east-1' # Choose region near your application
)
)
self.index = self.pc.Index(self.index_name)
def generate_embeddings(
self,
texts: List[str],
batch_size: int = 100
) -> List[List[float]]:
"""Generate embeddings with batching and error handling."""
embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
try:
response = self.openai_client.embeddings.create(
model=self.embedding_model,
input=batch,
encoding_format="float"
)
batch_embeddings = [
item.embedding for item in response.data
]
embeddings.extend(batch_embeddings)
except Exception as e:
print(f"Embedding generation failed for batch {i}: {e}")
# Retry individual texts in failed batch
for text in batch:
try:
response = self.openai_client.embeddings.create(
model=self.embedding_model,
input=[text],
encoding_format="float"
)
embeddings.append(response.data[0].embedding)
except Exception as inner_e:
print(f"Failed to embed text: {inner_e}")
# Add zero vector as placeholder
embeddings.append([0.0] * self.embedding_dimension)
return embeddings
def upsert_chunks(self, chunks: List[Dict], namespace: str = "policies"):
"""Upload document chunks with embeddings to Pinecone."""
# Extract texts for embedding
texts = [chunk['text'] for chunk in chunks]
# Generate embeddings
embeddings = self.generate_embeddings(texts)
# Prepare vectors for Pinecone
vectors = []
for chunk, embedding in zip(chunks, embeddings):
# Create unique ID from content hash
content_hash = hashlib.md5(
chunk['text'].encode()
).hexdigest()[:8]
vector_id = chunk.get('chunk_id', content_hash)
# Prepare metadata (Pinecone has size limits)
metadata = {
'text': chunk['text'][:1000], # Truncate for storage
'full_text_hash': content_hash,
**{k: v for k, v in chunk['metadata'].items()
if v is not None}
}
vectors.append({
'id': vector_id,
'values': embedding,
'metadata': metadata
})
# Upsert in batches
batch_size = 100
for i in range(0, len(vectors), batch_size):
batch = vectors[i:i + batch_size]
self.index.upsert(
vectors=batch,
namespace=namespace
)
def query_similar(
self,
query: str,
top_k: int = 5,
namespace: str = "policies",
filter_dict: Dict = None
) -> List[Dict]:
"""Find most similar document chunks to query."""
# Generate query embedding
query_embedding = self.generate_embeddings([query])[0]
# Query Pinecone
results = self.index.query(
vector=query_embedding,
top_k=top_k,
include_metadata=True,
namespace=namespace,
filter=filter_dict # e.g., {'section': 'refund_policy'}
)
# Format results
matches = []
for match in results.matches:
matches.append({
'id': match.id,
'score': match.score,
'text': match.metadata.get('text', ''),
'metadata': {
k: v for k, v in match.metadata.items()
if k != 'text'
}
})
return matchesThis sketch covers the features embedding generation needs before you'd trust it with real volume:
- Batch processing: Reduces API calls and improves throughput for large document sets
- Error recovery: Handles individual embedding failures without losing entire batches
- Metadata management: Preserves document context while respecting Pinecone's metadata size limits
- Content hashing: Enables deduplication and change detection for document updates
- Namespace organization: Separates different policy domains (HR, legal, technical) for targeted retrieval
My working assumption — from the retrieval work I've done since, on clinical guidelines rather than rail policies — is that embedding quality matters more than retrieval algorithm sophistication. Modern embedding models give good zero-shot performance on domain content without fine-tuning. For organisations with highly specialised terminology, fine-tuning the embedding model on domain data is the next lever to reach for, though I can't put a number on the gain for your corpus and neither can anyone who hasn't measured it on your corpus.
Search & Retrieval Optimization
Vector similarity search provides a strong baseline, but enterprise AI systems require additional optimization layers. Raw vector search suffers from several limitations: it can't filter by document metadata efficiently, doesn't understand query intent, and treats all semantic similarity equally regardless of document importance.
The retrieval pipeline implements a three-stage process:
from typing import List, Dict, Optional
import cohere
class PolicyRetrieval:
def __init__(
self,
embedding_pipeline: EmbeddingPipeline,
cohere_api_key: str
):
self.embedding_pipeline = embedding_pipeline
self.cohere_client = cohere.Client(cohere_api_key)
def hybrid_search(
self,
query: str,
top_k: int = 20,
rerank_top_k: int = 5,
filter_metadata: Optional[Dict] = None
) -> List[Dict]:
"""Three-stage retrieval: vector search, re-ranking, filtering."""
# Stage 1: Vector similarity search with over-retrieval
# Retrieve more candidates than needed for re-ranking
vector_results = self.embedding_pipeline.query_similar(
query=query,
top_k=top_k,
filter_dict=filter_metadata
)
if not vector_results:
return []
# Stage 2: Re-ranking with cross-encoder model
# Cohere's re-ranker uses bidirectional attention
documents = [result['text'] for result in vector_results]
rerank_response = self.cohere_client.rerank(
model='rerank-english-v3.0',
query=query,
documents=documents,
top_n=rerank_top_k,
return_documents=True
)
# Map re-ranked results back to original metadata
reranked_results = []
for result in rerank_response.results:
original_result = vector_results[result.index]
reranked_results.append({
**original_result,
'rerank_score': result.relevance_score,
'original_rank': result.index
})
# Stage 3: Apply business logic filters
filtered_results = self._apply_business_filters(
reranked_results,
query
)
return filtered_results
def _apply_business_filters(
self,
results: List[Dict],
query: str
) -> List[Dict]:
"""Apply domain-specific filtering logic."""
filtered = []
for result in results:
metadata = result.get('metadata', {})
# Filter out deprecated policy versions
if metadata.get('status') == 'deprecated':
continue
# Ensure minimum relevance threshold
if result.get('rerank_score', 0) < 0.5:
continue
# Check document recency for time-sensitive policies
# (Implementation would check effective_date in metadata)
filtered.append(result)
return filtered
def explain_retrieval(self, query: str, results: List[Dict]) -> Dict:
"""Generate retrieval explanation for debugging and transparency."""
explanation = {
'query': query,
'num_results': len(results),
'top_scores': [r.get('rerank_score') for r in results[:3]],
'source_documents': [
{
'id': r['id'],
'section': r['metadata'].get('section', 'unknown'),
'score': r.get('rerank_score')
}
for r in results
]
}
return explanationThis hybrid approach significantly improves retrieval quality:
- Over-retrieval strategy: Fetch 20 candidates for re-ranking to ensure relevant documents aren't missed by vector search alone
- Cross-encoder re-ranking: Cohere's re-ranker evaluates query-document pairs jointly, capturing subtle semantic relationships that vector similarity misses
- Business logic filters: Implement domain-specific rules (document freshness, deprecation status, regulatory requirements)
- Transparency mechanisms: Provide retrieval explanations for debugging and compliance auditing
Hybrid retrieval should beat pure vector search on queries where an exact term matters — a policy code, a station name — because lexical matching catches what embeddings smooth over. I've not measured that here. What I'd say with more confidence is that retrieval optimisation is iterative: you need retrieval quality metrics and real user queries before any threshold you pick is more than a guess.
LLM Integration & Prompt Engineering
The final stage of our RAG pipeline combines retrieved documents with sophisticated prompt engineering to generate accurate, well-cited policy responses. This is where many enterprise AI implementations fail - poor prompts lead to hallucinations, missing citations, or overly generic responses that don't leverage the retrieved context effectively.
The prompt engineering strategy follows these principles:
from typing import List, Dict
import openai
class PolicyAgent:
def __init__(self, openai_api_key: str, retrieval_system: PolicyRetrieval):
self.client = openai.OpenAI(api_key=openai_api_key)
self.retrieval = retrieval_system
self.model = "gpt-4-turbo-preview"
def construct_prompt(
self,
query: str,
context_chunks: List[Dict]
) -> str:
"""Build sophisticated RAG prompt with grounding instructions."""
# Format retrieved context with citations
context_sections = []
for idx, chunk in enumerate(context_chunks, 1):
metadata = chunk['metadata']
section_ref = metadata.get('section', 'Unknown Section')
context_sections.append(
f"[Source {idx}: {section_ref}]\\n{chunk['text']}\\n"
)
context_text = "\\n".join(context_sections)
prompt = f"""You are a Policy Assistant for an enterprise customer support system. Your role is to provide accurate, helpful answers to customer policy questions based ONLY on the provided policy documents.
RETRIEVED POLICY CONTEXT:
{context_text}
CUSTOMER QUESTION:
{query}
INSTRUCTIONS:
1. Answer the question using ONLY information from the provided policy context
2. If the context doesn't contain enough information to answer fully, say so explicitly
3. Cite specific sources using [Source N] notation for all factual claims
4. Use clear, customer-friendly language while maintaining accuracy
5. If policies conflict or have exceptions, explain them clearly
6. Do not make assumptions or add information not present in the context
7. For time-sensitive policies, mention any effective dates if present
RESPONSE FORMAT:
- Start with a direct answer to the customer's question
- Provide supporting details with source citations
- End with any relevant caveats or additional information the customer should know
Your response:"""
return prompt
def generate_response(
self,
query: str,
temperature: float = 0.1,
max_tokens: int = 800
) -> Dict:
"""Generate policy response with full RAG pipeline."""
# Retrieve relevant policy chunks
retrieved_chunks = self.retrieval.hybrid_search(
query=query,
top_k=20,
rerank_top_k=5
)
if not retrieved_chunks:
return {
'response': "I apologize, but I couldn't find relevant policy information to answer your question. Please contact our support team for assistance.",
'sources': [],
'confidence': 'low'
}
# Construct prompt with retrieved context
prompt = self.construct_prompt(query, retrieved_chunks)
# Generate response with low temperature for consistency
completion = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": "You are a precise, helpful policy assistant. Always ground your responses in the provided context and cite sources."
},
{
"role": "user",
"content": prompt
}
],
temperature=temperature,
max_tokens=max_tokens
)
response_text = completion.choices[0].message.content
# Extract source citations from response
import re
cited_sources = re.findall(r'\\[Source (\\d+)\\]', response_text)
unique_sources = list(set(cited_sources))
# Build source references
source_refs = []
for source_num in unique_sources:
idx = int(source_num) - 1
if idx < len(retrieved_chunks):
chunk = retrieved_chunks[idx]
source_refs.append({
'source_number': source_num,
'section': chunk['metadata'].get('section', 'Unknown'),
'document_id': chunk.get('id', 'Unknown'),
'relevance_score': chunk.get('rerank_score', 0)
})
return {
'response': response_text,
'sources': source_refs,
'confidence': self._assess_confidence(retrieved_chunks, cited_sources),
'retrieval_explanation': self.retrieval.explain_retrieval(
query,
retrieved_chunks
)
}
def _assess_confidence(
self,
retrieved_chunks: List[Dict],
cited_sources: List[str]
) -> str:
"""Heuristic confidence assessment based on retrieval quality."""
if not retrieved_chunks:
return 'low'
top_score = retrieved_chunks[0].get('rerank_score', 0)
num_citations = len(set(cited_sources))
if top_score > 0.8 and num_citations >= 2:
return 'high'
elif top_score > 0.6 and num_citations >= 1:
return 'medium'
else:
return 'low'This implementation incorporates several advanced prompt engineering techniques:
- Explicit grounding instructions: Prevents hallucination by constraining the LLM to retrieved context
- Structured citation format: Makes source tracking programmatically verifiable for compliance
- Customer-friendly language directive: Balances accuracy with accessibility
- Caveat handling: Encourages the model to surface edge cases and exceptions
- Low temperature setting: Reduces variability for consistent policy responses
- Confidence scoring: Provides quality indicators for routing low-confidence queries to human agents
Prompt engineering is an empirical discipline: test prompts against diverse query types, measure citation accuracy, and iterate. In a later RAG project of mine, query expansion alone was worth 0.030–0.035 factual F1 on a measured benchmark — a real number, from a different corpus, offered here only to show the scale these adjustments move at rather than to promise you the same.
Testing & Quality Assurance
Enterprise RAG systems require rigorous testing to ensure accuracy, consistency, and compliance. Unlike traditional software where test cases are deterministic, LLM-based systems exhibit variability that demands statistical evaluation approaches.
The testing framework covers multiple dimensions:
import pytest
from typing import List, Dict
import json
class RAGTestSuite:
def __init__(self, policy_agent: PolicyAgent):
self.agent = policy_agent
self.test_cases = self._load_test_cases()
def _load_test_cases(self) -> List[Dict]:
"""Load curated test cases with ground truth answers."""
return [
{
'query': 'What is the refund window for cancelled bookings?',
'expected_sources': ['refund_policy'],
'expected_facts': ['48 hours', 'full refund'],
'category': 'refund'
},
{
'query': 'Can I change my booking date after confirmation?',
'expected_sources': ['modification_policy'],
'expected_facts': ['24 hours notice', 'one-time change'],
'category': 'modification'
},
# ... more test cases
]
def test_retrieval_precision(self):
"""Test that correct policy sections are retrieved."""
results = []
for test_case in self.test_cases:
retrieved = self.agent.retrieval.hybrid_search(
test_case['query'],
rerank_top_k=3
)
retrieved_sections = [
chunk['metadata'].get('section', '')
for chunk in retrieved
]
# Check if expected sources appear in top 3
precision = any(
expected in str(retrieved_sections)
for expected in test_case['expected_sources']
)
results.append({
'query': test_case['query'],
'precision': precision,
'retrieved_sections': retrieved_sections
})
precision_rate = sum(r['precision'] for r in results) / len(results)
assert precision_rate >= 0.90, f"Retrieval precision {precision_rate} below 90% threshold"
def test_factual_accuracy(self):
"""Test that generated responses contain expected facts."""
results = []
for test_case in self.test_cases:
response = self.agent.generate_response(test_case['query'])
response_text = response['response'].lower()
# Check if expected facts appear in response
facts_present = [
fact.lower() in response_text
for fact in test_case['expected_facts']
]
accuracy = sum(facts_present) / len(facts_present)
results.append({
'query': test_case['query'],
'accuracy': accuracy,
'missing_facts': [
fact for fact, present in
zip(test_case['expected_facts'], facts_present)
if not present
]
})
avg_accuracy = sum(r['accuracy'] for r in results) / len(results)
assert avg_accuracy >= 0.85, f"Factual accuracy {avg_accuracy} below 85% threshold"
def test_citation_completeness(self):
"""Test that responses include proper source citations."""
for test_case in self.test_cases:
response = self.agent.generate_response(test_case['query'])
assert len(response['sources']) > 0, \\
f"No sources cited for query: {test_case['query']}"
# Verify citations appear in response text
import re
citations = re.findall(r'\\[Source \\d+\\]', response['response'])
assert len(citations) > 0, \\
f"Sources provided but not cited in text: {test_case['query']}"
def test_consistency(self, iterations: int = 5):
"""Test response consistency across multiple generations."""
for test_case in self.test_cases[:3]: # Test subset for speed
responses = []
for _ in range(iterations):
response = self.agent.generate_response(test_case['query'])
responses.append(response['response'])
# Check that key facts appear consistently
for expected_fact in test_case['expected_facts']:
appearances = sum(
expected_fact.lower() in resp.lower()
for resp in responses
)
consistency_rate = appearances / iterations
assert consistency_rate >= 0.8, \\
f"Fact '{expected_fact}' inconsistent across generations"
def test_hallucination_detection(self):
"""Test queries designed to trigger hallucinations."""
hallucination_queries = [
"What is the refund policy for Mars trips?", # Non-existent product
"How much does a premium subscription cost?", # If not in docs
]
for query in hallucination_queries:
response = self.agent.generate_response(query)
# Low confidence or explicit "I don't know" expected
assert (
response['confidence'] == 'low' or
"don't have" in response['response'].lower() or
"couldn't find" in response['response'].lower()
), f"Potential hallucination for query: {query}"This testing framework validates several critical quality dimensions:
- Retrieval Precision: are the correct policy sections found? 90%+ is the bar I'd set before trusting it — a target, not a result I measured
- Factual Accuracy: do generated responses contain the expected factual elements? Again 85%+ is a threshold I'd want to clear, not a score I recorded
- Citation Completeness: Validates that all responses include proper source attribution
- Response Consistency: Tests that the same query produces semantically consistent answers across multiple generations
- Hallucination Resistance: Evaluates system behavior on queries that should trigger "I don't know" responses
Anything that actually ships needs continuous evaluation against real user queries: user satisfaction ratings, escalation rates to human agents, citation accuracy audits. I'd expect demonstrable testing rigour to help with stakeholder buy-in and regulatory sign-off too — that one is reasoning from how these conversations tend to go, not something I've carried through an approval process myself.
Production Deployment Considerations
Moving from prototype to production requires addressing scalability, monitoring, and maintenance:
- Latency optimization: cache frequent queries and pre-compute embeddings for common policy sections if you're chasing sub-2-second responses
- Cost management: Monitor embedding and LLM API costs; implement query batching and result caching to reduce expenses at scale
- Version control: Track policy document versions in vector database metadata to enable rollback and audit trails
- Monitoring dashboards: Instrument retrieval precision, response latency, citation accuracy, and user satisfaction scores
- Human-in-the-loop: Route low-confidence responses to human review; use feedback to improve retrieval and prompts
Most of the hard work in a real deployment is here, not in the core RAG. A system that's technically clever but slow, expensive, or unmonitored doesn't survive contact with production. This is the part I spend the most time on with clients.
Next in the series
The Policy Agent answers questions. Next I build its opposite number — the Ticket Agent, which doesn't just answer but acts: searching, booking, modifying and cancelling against the database.
Production-ready RAG — grounded, cited, fast enough, and monitored — is what I build at twentytwotensors. Get in touch if you need it.