{"title":"RAG Pipeline Optimization","description":"","content":"Your RAG system works. It retrieves relevant chunks, generates grounded answers, and your demo got a round of applause. Then you deploy it to real users and everything falls apart.\n\nA single query takes 4 seconds. Users stare at a loading spinner and leave. Your monthly API bill hits $2,000 for just 500 daily users. And when your team tries a load test at 50 concurrent users, the system buckles, response times spike to 15 seconds, and some requests time out entirely.\n\nThe frustrating part is that the answer quality is good. The system works, it is just too slow, too expensive, and too fragile for production. You need to cut latency in half, reduce costs by 40%, and handle 10x more traffic, without degrading the answers your users depend on.\n\nThis is the optimization chapter. We are not building new features. We are taking the RAG pipeline you built in 5.2 and making it production-ready.\n\n---\n\n# Where Does the Time Go?\n\nBefore optimizing anything, you need to know what is slow. Most developers guess wrong. They assume the LLM generation step is the bottleneck (it often is, but not always), and they miss the other stages that quietly eat hundreds of milliseconds each.\n\nA typical RAG query flows through five stages. Here is the breakdown with typical latencies:\n\n\n| Stage | What Happens | Typical Latency | % of Total |\n|-------|-------------|----------------|------------|\n| Query preprocessing | Rewriting, expansion, classification | 50-200ms | 5-10% |\n| Embedding | Converting query text to a vector | 50-150ms | 5-8% |\n| Retrieval | Vector search + optional re-ranking | 100-500ms | 10-20% |\n| Context assembly | Formatting chunks into prompt | 5-20ms | <1% |\n| LLM generation | Generating the final answer | 1,000-3,000ms | 60-80% |\n\n\nThe numbers tell an important story. LLM generation dominates, but the other stages add up. If your retrieval takes 400ms and your embedding takes 150ms, that is 550ms before the LLM even starts thinking. At 50 concurrent users, those stages compete for resources and the latencies multiply.\n\n### Profiling Your Pipeline\n\nLet's build a simple profiler that measures each stage. This is the first thing you should do before any optimization work.\n\n\n```python\nimport time\nfrom dataclasses import dataclass, field\nfrom contextlib import contextmanager\n\n@dataclass\nclass PipelineProfile:\n \"\"\"Tracks timing for each stage of the RAG pipeline.\"\"\"\n stages: dict = field(default_factory=dict)\n start_time: float = 0.0\n\n @contextmanager\n def track(self, stage_name: str):\n \"\"\"Context manager to time a pipeline stage.\"\"\"\n start = time.perf_counter()\n yield\n elapsed = time.perf_counter() - start\n self.stages[stage_name] = elapsed\n\n def total(self) -> float:\n return sum(self.stages.values())\n\n def report(self) -> str:\n total = self.total()\n lines = [f\"Total: {total:.3f}s\"]\n for stage, duration in self.stages.items():\n pct = (duration / total * 100) if total > 0 else 0\n lines.append(f\" {stage}: {duration:.3f}s ({pct:.1f}%)\")\n return \"\\n\".join(lines)\n\ndef profiled_rag_query(question: str, collection, openai_client) -> tuple:\n \"\"\"Run a RAG query with timing for each stage.\"\"\"\n profile = PipelineProfile()\n\n # Stage 1: Query preprocessing\n with profile.track(\"preprocessing\"):\n processed_query = question.strip().lower()\n\n # Stage 2: Embedding\n with profile.track(\"embedding\"):\n embed_response = openai_client.embeddings.create(\n model=\"text-embedding-3-small\",\n input=processed_query,\n )\n query_vector = embed_response.data[0].embedding\n\n # Stage 3: Retrieval\n with profile.track(\"retrieval\"):\n results = collection.query(\n query_embeddings=[query_vector],\n n_results=5,\n )\n chunks = results[\"documents\"][0]\n\n # Stage 4: Context assembly\n with profile.track(\"context_assembly\"):\n context = \"\\n\\n\".join(\n f\"[Chunk {i+1}]: {chunk}\"\n for i, chunk in enumerate(chunks)\n )\n\n # Stage 5: LLM generation\n with profile.track(\"generation\"):\n response = openai_client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": (\n \"Answer the question using the provided context. \"\n \"Be concise and accurate.\"\n ),\n },\n {\n \"role\": \"user\",\n \"content\": f\"Context:\\n{context}\\n\\nQuestion: {question}\",\n },\n ],\n )\n answer = response.choices[0].message.content\n\n print(profile.report())\n return answer, profile\n```\n\n\nRun this on 20 representative queries. You will see patterns: maybe embedding is consistently slow because you are using a large model, maybe retrieval spikes on certain query types, maybe generation latency varies wildly based on answer length. These patterns tell you where to focus your optimization effort.\n\n---\n\n# Caching: Eliminating Redundant Work\n\nThe fastest computation is the one you never do. Caching is the single highest-impact optimization for most RAG systems because users ask similar questions repeatedly. A customer support RAG system might get \"How do I reset my password?\" fifty times a day. Without caching, you embed the query, search the vector store, and call the LLM fifty times. With caching, you do it once.\n\n### Level 1: Exact Query Cache\n\nThe simplest cache matches exact query strings to previous results. If someone asks the identical question, return the cached answer immediately.\n\n\n```python\nimport hashlib\nimport json\nimport time\nfrom pathlib import Path\n\nclass QueryCache:\n \"\"\"Simple disk-backed exact-match query cache.\"\"\"\n\n def __init__(self, cache_dir: str = \".rag_cache\", ttl_seconds: int = 3600):\n self.cache_dir = Path(cache_dir)\n self.cache_dir.mkdir(exist_ok=True)\n self.ttl = ttl_seconds\n\n def _key(self, query: str) -> str:\n \"\"\"Generate a cache key from the query string.\"\"\"\n return hashlib.sha256(query.strip().lower().encode()).hexdigest()\n\n def get(self, query: str) -> str | None:\n \"\"\"Look up a cached answer. Returns None on miss.\"\"\"\n key = self._key(query)\n cache_file = self.cache_dir / f\"{key}.json\"\n if not cache_file.exists():\n return None\n\n data = json.loads(cache_file.read_text())\n # Check TTL\n if time.time() - data[\"timestamp\"] > self.ttl:\n cache_file.unlink()\n return None\n\n return data[\"answer\"]\n\n def set(self, query: str, answer: str):\n \"\"\"Store an answer in the cache.\"\"\"\n key = self._key(query)\n cache_file = self.cache_dir / f\"{key}.json\"\n cache_file.write_text(json.dumps({\n \"query\": query,\n \"answer\": answer,\n \"timestamp\": time.time(),\n }))\n\n# Usage\ncache = QueryCache(ttl_seconds=3600) # 1-hour TTL\n\ndef cached_rag_query(question: str, collection, openai_client) -> str:\n # Check cache first\n cached = cache.get(question)\n if cached:\n print(\"Cache hit!\")\n return cached\n\n # Cache miss, run full pipeline\n answer, _ = profiled_rag_query(question, collection, openai_client)\n cache.set(question, answer)\n return answer\n```\n\n\nExact match caching has a major limitation: \"How do I reset my password?\" and \"How can I reset my password?\" are treated as different queries. That is where semantic caching comes in.\n\n### Level 2: Semantic Cache\n\nA semantic cache embeds the query and checks if any previous query is semantically similar enough to reuse its answer. This catches paraphrases, rephrasing, and minor variations.\n\n\n```python\nimport numpy as np\n\nclass SemanticCache:\n \"\"\"\n Cache that matches semantically similar queries\n using embedding similarity.\n \"\"\"\n\n def __init__(\n self,\n openai_client,\n similarity_threshold: float = 0.95,\n max_entries: int = 10000,\n ):\n self.client = openai_client\n self.threshold = similarity_threshold\n self.max_entries = max_entries\n self.entries = [] # list of (embedding, query, answer)\n\n def _embed(self, text: str) -> list[float]:\n response = self.client.embeddings.create(\n model=\"text-embedding-3-small\",\n input=text,\n )\n return response.data[0].embedding\n\n def _cosine_similarity(self, a: list[float], b: list[float]) -> float:\n a_np = np.array(a)\n b_np = np.array(b)\n return float(np.dot(a_np, b_np) / (\n np.linalg.norm(a_np) * np.linalg.norm(b_np)\n ))\n\n def get(self, query: str) -> tuple[str, float] | None:\n \"\"\"\n Find a semantically similar cached query.\n Returns (answer, similarity_score) or None.\n \"\"\"\n query_embedding = self._embed(query)\n\n best_match = None\n best_score = 0.0\n\n for entry_embedding, entry_query, entry_answer in self.entries:\n score = self._cosine_similarity(query_embedding, entry_embedding)\n if score > best_score:\n best_score = score\n best_match = (entry_answer, best_score)\n\n if best_match and best_score >= self.threshold:\n return best_match\n\n return None\n\n def set(self, query: str, answer: str):\n \"\"\"Add a query-answer pair to the semantic cache.\"\"\"\n embedding = self._embed(query)\n self.entries.append((embedding, query, answer))\n\n # Evict oldest entries if cache is full\n if len(self.entries) > self.max_entries:\n self.entries = self.entries[-self.max_entries:]\n```\n\n\nThe threshold parameter is critical. Set it too low (0.85) and you return cached answers for questions that are not really the same. Set it too high (0.99) and the cache rarely hits. A threshold of 0.95 is a good starting point, meaning two queries need to be very similar for a cache hit.\n\nIn production, you would store the cache embeddings in the same vector database you use for document retrieval. ChromaDB, Pinecone, and other vector stores can handle this efficiently. The implementation above uses a linear scan for clarity, but it would be too slow for caches with thousands of entries.\n\n### Level 3: Embedding Cache\n\nIf you are calling an embedding API, you are paying per token and waiting for a network round trip on every query. An embedding cache stores previously computed embeddings so you never embed the same text twice.\n\n\n```python\nclass EmbeddingCache:\n \"\"\"Cache for computed embeddings to avoid redundant API calls.\"\"\"\n\n def __init__(self, openai_client):\n self.client = openai_client\n self.cache = {} # text_hash -> embedding\n\n def embed(self, text: str) -> list[float]:\n \"\"\"Get embedding, using cache if available.\"\"\"\n key = hashlib.sha256(text.encode()).hexdigest()\n\n if key in self.cache:\n return self.cache[key]\n\n response = self.client.embeddings.create(\n model=\"text-embedding-3-small\",\n input=text,\n )\n embedding = response.data[0].embedding\n self.cache[key] = embedding\n return embedding\n```\n\n\nThis is especially useful during ingestion, where you might re-process documents that have not changed. But it also helps at query time for repeated or similar queries.\n\n---\n\n# Pre-computation: Shifting Work to Ingestion Time\n\nEvery millisecond you spend at query time is a millisecond your user waits. Pre-computation moves expensive operations from query time to ingestion time, where latency does not matter.\n\n### Pre-computed Chunk Summaries\n\nInstead of sending raw chunks to the LLM, you can pre-generate summaries during ingestion. Each chunk gets a short summary (1-2 sentences) that captures its key information. At query time, you retrieve chunks and send their summaries instead of the full text, dramatically reducing the number of tokens in the LLM prompt.\n\n\n```python\ndef precompute_summaries(\n chunks: list[str],\n openai_client,\n batch_size: int = 10,\n) -> list[dict]:\n \"\"\"\n Pre-compute summaries for document chunks at ingestion time.\n Returns list of {chunk, summary} dicts.\n \"\"\"\n results = []\n\n for i in range(0, len(chunks), batch_size):\n batch = chunks[i:i + batch_size]\n for chunk in batch:\n response = openai_client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": (\n \"Summarize this text in 1-2 sentences. \"\n \"Preserve all key facts, names, and numbers.\"\n ),\n },\n {\"role\": \"user\", \"content\": chunk},\n ],\n max_tokens=100,\n )\n summary = response.choices[0].message.content\n results.append({\"chunk\": chunk, \"summary\": summary})\n\n return results\n```\n\n\nThis technique reduces the tokens sent to the LLM by 60-80%. If your average chunk is 300 tokens and the summary is 50 tokens, sending 5 summaries costs 250 tokens instead of 1,500. At $0.60 per million output tokens for gpt-4o-mini, this adds up fast across thousands of queries.\n\nThe trade-off is ingestion cost. You are calling the LLM once per chunk during ingestion. For 10,000 chunks, that might cost $5-15. But you pay this once, and every query benefits.\n\n### Pre-computed Metadata Enrichment\n\nAnother pre-computation strategy is enriching each chunk with metadata at ingestion time: topic labels, entity mentions, question-answer pairs that the chunk could answer.\n\n\n```python\ndef enrich_chunk_metadata(chunk: str, openai_client) -> dict:\n \"\"\"\n Extract metadata from a chunk at ingestion time\n for faster filtering at query time.\n \"\"\"\n response = openai_client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": (\n \"Analyze this text and return JSON with:\\n\"\n '- \"topics\": list of 2-3 topic labels\\n'\n '- \"entities\": list of named entities mentioned\\n'\n '- \"questions\": list of 2-3 questions this text answers\\n'\n \"Return valid JSON only.\"\n ),\n },\n {\"role\": \"user\", \"content\": chunk},\n ],\n response_format={\"type\": \"json_object\"},\n )\n return json.loads(response.choices[0].message.content)\n```\n\n\nWith pre-computed topics, you can filter chunks before vector search (\"only search chunks about billing\") which reduces the search space and improves both speed and relevance.\n\n---\n\n# Retrieval Optimization\n\nThe retrieval stage has two costs: the time spent searching, and the tokens sent to the LLM. Optimizing both makes a significant difference.\n\n### Reducing Chunks Sent to the LLM\n\nAs we covered in 5.3, re-ranking improves retrieval quality. But it also enables a powerful optimization: retrieve more chunks, re-rank them, and send only the top 3 instead of the top 5. Fewer chunks means fewer tokens, which means faster and cheaper generation.\n\n\n```python\ndef optimized_retrieval(\n question: str,\n collection,\n openai_client,\n initial_k: int = 15,\n final_k: int = 3,\n) -> list[str]:\n \"\"\"\n Retrieve more chunks than needed, then re-rank\n and keep only the best ones.\n \"\"\"\n # Step 1: Broad retrieval\n results = collection.query(\n query_texts=[question],\n n_results=initial_k,\n )\n candidates = results[\"documents\"][0]\n\n # Step 2: LLM-based re-ranking (lightweight)\n scored_chunks = []\n for chunk in candidates:\n response = openai_client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": (\n \"Rate how relevant this text is to the question. \"\n \"Return only a number from 0 to 10.\"\n ),\n },\n {\n \"role\": \"user\",\n \"content\": (\n f\"Question: {question}\\n\\nText: {chunk}\"\n ),\n },\n ],\n max_tokens=5,\n )\n try:\n score = float(response.choices[0].message.content.strip())\n except ValueError:\n score = 5.0\n scored_chunks.append((score, chunk))\n\n # Step 3: Keep only the top-scoring chunks\n scored_chunks.sort(key=lambda x: x[0], reverse=True)\n return [chunk for _, chunk in scored_chunks[:final_k]]\n```\n\n\nThis trades a small amount of re-ranking time (calling gpt-4o-mini with short prompts) for a large reduction in generation tokens. In practice, you would use a cross-encoder model for re-ranking instead of the LLM, as we discussed in Module 4.4. A cross-encoder re-ranks 15 chunks in 50-100ms, which is much faster and cheaper than 15 LLM calls.\n\n### Adaptive Retrieval\n\nNot every question needs the same number of chunks. A simple factual question (\"What is the default timeout?\") might need just one chunk, while a complex question (\"Compare the performance characteristics of our three database options\") might need five.\n\n\n```python\ndef adaptive_retrieval(\n question: str,\n collection,\n openai_client,\n) -> list[str]:\n \"\"\"\n Adjust the number of retrieved chunks based on\n query complexity.\n \"\"\"\n # Classify query complexity\n response = openai_client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": (\n \"Classify this question as 'simple' (one fact needed), \"\n \"'moderate' (2-3 facts needed), or \"\n \"'complex' (many facts or comparison needed). \"\n \"Return only the classification word.\"\n ),\n },\n {\"role\": \"user\", \"content\": question},\n ],\n max_tokens=10,\n )\n complexity = response.choices[0].message.content.strip().lower()\n\n # Map complexity to chunk count\n chunk_counts = {\"simple\": 2, \"moderate\": 4, \"complex\": 6}\n n_chunks = chunk_counts.get(complexity, 4)\n\n results = collection.query(\n query_texts=[question],\n n_results=n_chunks,\n )\n return results[\"documents\"][0]\n```\n\n\nAdaptive retrieval reduces the average number of tokens per query. If 60% of your queries are simple and only need 2 chunks instead of 5, you save 60% of retrieval tokens on those queries, which translates directly to cost and latency savings.\n\n---\n\n# Async and Parallel Retrieval\n\nWhen your RAG pipeline performs multiple I/O operations (embedding the query, searching the vector store, re-ranking, calling the LLM), doing them sequentially wastes time. Each operation involves a network round trip, and your CPU sits idle while waiting for responses.\n\n### Parallel Embedding and Keyword Search\n\nIf your pipeline uses hybrid retrieval (as we built in 5.2 and covered in 5.3), you can run the vector search and keyword search in parallel. They do not depend on each other.\n\n\n```python\nimport asyncio\nfrom openai import AsyncOpenAI\n\nasync def parallel_hybrid_search(\n question: str,\n collection,\n bm25_index,\n openai_client: AsyncOpenAI,\n) -> dict:\n \"\"\"\n Run vector search and keyword search in parallel,\n then merge results.\n \"\"\"\n async def vector_search():\n # Embed and search\n response = await openai_client.embeddings.create(\n model=\"text-embedding-3-small\",\n input=question,\n )\n query_vector = response.data[0].embedding\n results = collection.query(\n query_embeddings=[query_vector],\n n_results=10,\n )\n return results[\"documents\"][0]\n\n async def keyword_search():\n # BM25 search (CPU-bound but fast)\n scores = bm25_index.get_scores(question.split())\n top_indices = sorted(\n range(len(scores)),\n key=lambda i: scores[i],\n reverse=True,\n )[:10]\n return [bm25_index.corpus[i] for i in top_indices]\n\n # Run both searches concurrently\n vector_results, keyword_results = await asyncio.gather(\n vector_search(),\n keyword_search(),\n )\n\n return {\n \"vector_results\": vector_results,\n \"keyword_results\": keyword_results,\n }\n```\n\n\nBy running embedding + vector search concurrently with keyword search, you cut the retrieval stage time roughly in half. Instead of 150ms (embedding) + 100ms (vector search) + 80ms (keyword search) = 330ms, you get max(250ms, 80ms) = 250ms.\n\n### Streaming Generation\n\nThe LLM generation step is the longest, and your user should not have to wait for the entire response before seeing anything. Streaming sends tokens to the user as they are generated, dramatically improving perceived latency.\n\n\n```python\nfrom openai import OpenAI\n\ndef streaming_rag_query(\n question: str,\n context: str,\n openai_client: OpenAI,\n):\n \"\"\"\n Generate a RAG response with streaming.\n Tokens appear as they're generated.\n \"\"\"\n stream = openai_client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": (\n \"Answer the question using the provided context. \"\n \"Be concise and accurate.\"\n ),\n },\n {\n \"role\": \"user\",\n \"content\": f\"Context:\\n{context}\\n\\nQuestion: {question}\",\n },\n ],\n stream=True,\n )\n\n full_response = []\n for chunk in stream:\n if chunk.choices[0].delta.content:\n token = chunk.choices[0].delta.content\n full_response.append(token)\n print(token, end=\"\", flush=True)\n\n print() # Newline after streaming\n return \"\".join(full_response)\n```\n\n\nStreaming does not reduce total latency (the LLM still generates all the tokens), but it reduces time-to-first-token from 2-3 seconds to 200-500ms. This is the single most impactful perceived performance improvement you can make.\n\n---\n\n# Cost Reduction Strategies\n\nRAG costs come from three sources: embedding API calls, vector database operations, and LLM generation. Generation dominates, typically accounting for 70-85% of total cost.\n\n### Model Routing\n\nNot every query needs your most expensive model. Simple factual lookups can use a cheaper model while complex reasoning questions use the premium one.\n\n\n```python\ndef route_to_model(\n question: str,\n context: str,\n openai_client: OpenAI,\n) -> str:\n \"\"\"\n Route simple queries to a cheaper model,\n complex queries to a more capable one.\n \"\"\"\n # Classify query complexity\n classify_response = openai_client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": (\n \"Classify this as 'simple' (direct fact lookup) or \"\n \"'complex' (requires reasoning, comparison, or synthesis). \"\n \"Return only the word.\"\n ),\n },\n {\"role\": \"user\", \"content\": question},\n ],\n max_tokens=10,\n )\n complexity = classify_response.choices[0].message.content.strip().lower()\n\n # Route to appropriate model\n model = \"gpt-4o-mini\" if complexity == \"simple\" else \"gpt-4o\"\n\n response = openai_client.chat.completions.create(\n model=model,\n messages=[\n {\n \"role\": \"system\",\n \"content\": (\n \"Answer the question using the provided context.\"\n ),\n },\n {\n \"role\": \"user\",\n \"content\": f\"Context:\\n{context}\\n\\nQuestion: {question}\",\n },\n ],\n )\n return response.choices[0].message.content\n```\n\n\nThe classification call itself is cheap (a few tokens with gpt-4o-mini). If 70% of your queries are simple, you save the cost difference between gpt-4o and gpt-4o-mini on those queries. With gpt-4o at roughly $10/million output tokens and gpt-4o-mini at $0.60/million, that is a significant reduction.\n\n### Cost Optimization Summary\n\nHere is a table showing the impact of each optimization on a typical RAG system processing 10,000 queries per day:\n\n\n| Optimization | Latency Impact | Cost Impact | Implementation Effort |\n|-------------|---------------|-------------|----------------------|\n| Exact query cache | -90% on cache hits | -90% on cache hits | Low (hours) |\n| Semantic cache | -85% on cache hits | -85% on cache hits | Medium (1-2 days) |\n| Embedding cache | -30ms per cached embed | -20% on embedding costs | Low (hours) |\n| Chunk pre-summarization | -30-50% generation time | -40-60% generation cost | Medium (1 day) |\n| Adaptive retrieval | -20% average tokens | -20% generation cost | Medium (1 day) |\n| Streaming | -80% perceived TTFT | None | Low (hours) |\n| Parallel retrieval | -30% retrieval time | None | Medium (1 day) |\n| Model routing | None | -40-60% generation cost | Medium (1 day) |\n\n\nThe most impactful combination for most teams: **semantic caching + chunk pre-summarization + model routing + streaming**. Together, these typically deliver 50%+ latency reduction and 40%+ cost reduction.\n\n---\n\n# Putting It All Together: The Optimized Pipeline\n\nLet's see how all these techniques combine into a single optimized pipeline.\n\n\n```mermaid\nflowchart TD\n Q[User Query]:::primary\n\n Q --> SC{Semantic Cache
Hit?}\n SC -->|Yes| CACHED[Return Cached Answer
~10ms]:::green\n\n SC -->|No| PARALLEL\n\n subgraph PARALLEL[Parallel Stage]\n EMB[Embed Query
cached if seen]:::orange\n KW[Keyword Search
BM25]:::orange\n end\n\n EMB --> VS[Vector Search]:::teal\n VS --> MERGE[Merge + Re-rank]:::teal\n KW --> MERGE\n\n MERGE --> ADAPT[Adaptive Chunk
Selection]:::primary\n\n ADAPT --> ROUTE{Model Router}\n ROUTE -->|Simple| MINI[gpt-4o-mini
Stream]:::green\n ROUTE -->|Complex| FULL[gpt-4o
Stream]:::orange\n\n MINI --> ANS[Answer + Cache Update]:::green\n FULL --> ANS\n\n classDef primary fill:#00ceff,stroke:#000,color:#000\n classDef orange fill:#ffa94d,stroke:#000,color:#000\n classDef teal fill:#38d9a9,stroke:#000,color:#000\n classDef green fill:#69db7c,stroke:#000,color:#000\n```\n\n\nThe flow works like this:\n\n1. Every query first hits the semantic cache. If a similar question was answered recently, return the cached answer in roughly 10ms.\n2. On a cache miss, embedding and keyword search run in parallel.\n3. Results merge and get re-ranked with a cross-encoder.\n4. Adaptive chunk selection picks only the chunks actually needed for this query complexity.\n5. The model router sends simple queries to gpt-4o-mini and complex queries to gpt-4o.\n6. Generation streams tokens to the user for fast perceived response.\n7. The answer gets stored in the semantic cache for future queries.\n\n### Before and After\n\nHere is what a typical optimization journey looks like:\n\n\n```mermaid\nflowchart LR\n subgraph Before[\"Before Optimization\"]\n B1[4.2s latency]:::red\n B2[$0.15 per query]:::red\n B3[50 max concurrent]:::red\n end\n\n subgraph After[\"After Optimization\"]\n A1[1.8s latency
avg with cache: 0.4s]:::green\n A2[$0.06 per query]:::green\n A3[200+ concurrent]:::green\n end\n\n Before --> |Semantic cache
Async retrieval
Model routing
Streaming| After\n\n classDef red fill:#ff8787,stroke:#000,color:#000\n classDef green fill:#69db7c,stroke:#000,color:#000\n```\n\n\nThe exact numbers depend on your workload, but a 50-60% latency reduction and 40-60% cost reduction is typical when you apply the full optimization stack.\n\n---\n\n# Monitoring and Continuous Optimization\n\nOptimization is not a one-time effort. Query patterns change, new documents get added, and model pricing evolves. You need ongoing monitoring to catch regressions and find new optimization opportunities.\n\n**Key metrics to track:**\n\n- **P50 and P95 latency** per pipeline stage (not just total)\n- **Cache hit rate** (target: 30-50% for semantic cache)\n- **Tokens per query** (average and P95)\n- **Cost per query** (broken down by embedding, retrieval, generation)\n- **Answer quality scores** from your evaluation pipeline (covered in 5.5)\n\nThe most common regression is cache hit rate dropping because the query distribution changed. If a new feature launch brings in a flood of new question types, your cache hit rate plummets and latency spikes. Monitoring alerts you to this so you can adjust your caching strategy or warm the cache proactively.\n\n---\n\n### References\n\n- [OpenAI Embeddings Guide](undefined)\n- [ChromaDB Documentation](undefined)\n- [GPTCache: Semantic Caching for LLMs](undefined)\n- [Async Python with asyncio (Official Docs)](undefined)\n- [OpenAI API Streaming Guide](undefined)","pageType":"ai-engineering"}

RAG Pipeline Optimization

Ashish Pratap Singh

Get Premium