If you've spent time tweaking prompts to fix a flaky RAG app or a wandering agent, you've probably noticed the same thing most of us do eventually: rewording the prompt fixes the demo case and breaks two others. The actual bug usually isn't the prompt. It's what the model was handed before the prompt even ran — retrieved documents that contradict each other, conversation history crowding out the facts that matter, or the one correct chunk buried in the middle of a context blob where positional bias guarantees the model will underweight it.
This is a full implementation walkthrough of a context pipeline — the retrieval, ranking, assembly, memory, and reuse logic that decides what a model actually sees. Every stage below includes working code you can adapt directly, plus the specific failure modes each stage exists to prevent.
Architecture Overview
raw query
│
▼
[1] Query Rewriting — fix ambiguous/short queries before retrieval
│
▼
[2] Hybrid Retrieval — dense + sparse search, merged
│
▼
[3] Reranking — precision pass over retrieved candidates
│
▼
[4] Deduplication — collapse near-identical/contradictory chunks
│
▼
[5] Context Assembly — token-budgeted, position-aware final input
│
▼
[6] Reuse Check — skip stages if a near-identical request exists
│
▼
[7] Generation — the actual model call
│
▼
[8] Evaluation Logging — per-stage metrics for regression testing
Prompt engineering lives entirely in box 7. Everything else is context engineering, and it's usually where production bugs actually hide. Let's build each stage.
1. Query Rewriting
Users type short, underspecified queries. Retrievers do better with explicit, expanded ones.
def rewrite_query(user_query: str, conversation_history: list[str]) -> str:
# Cheap, deterministic expansion first — cheaper than an LLM call,
# and it resolves a surprising fraction of ambiguous short queries.
is_short = len(user_query.split()) < 4
has_pronoun = any(p in user_query.lower() for p in ["it", "that", "this", "those"])
if (is_short or has_pronoun) and conversation_history:
return f"{conversation_history[-1]} {user_query}"
return user_query
In production, this step often escalates to a small, fast LLM call for genuinely ambiguous queries — but gate that escalation behind the heuristic above, since most queries don't need it and every extra model call adds latency you don't want to pay by default.
def rewrite_query_with_llm_fallback(user_query, history, needs_escalation_fn, llm_rewrite_fn):
rewritten = rewrite_query(user_query, history)
if needs_escalation_fn(rewritten):
return llm_rewrite_fn(rewritten, history)
return rewritten
2. Hybrid Retrieval
Pure vector search misses exact matches — IDs, product codes, proper nouns — that embeddings tend to smear together in semantic space. Combine dense and sparse retrieval, and merge with reciprocal rank fusion rather than naive score averaging (scores from different retrieval methods aren't on comparable scales):
def reciprocal_rank_fusion(result_lists, k=60):
scores = {}
for results in result_lists:
for rank, doc in enumerate(results):
scores[doc.id] = scores.get(doc.id, 0) + 1 / (k + rank + 1)
ranked_ids = sorted(scores, key=scores.get, reverse=True)
doc_lookup = {doc.id: doc for results in result_lists for doc in results}
return [doc_lookup[i] for i in ranked_ids]
def hybrid_retrieve(query: str, vector_store, keyword_index, k: int = 20):
dense_results = vector_store.similarity_search(query, k=k)
sparse_results = keyword_index.search(query, k=k)
return reciprocal_rank_fusion([dense_results, sparse_results])[:k]
Retrieve generously here — k=20 or more — over-fetching is cheap, and the next two stages exist specifically to cut the candidate set down intelligently rather than relying on the retriever to be precise on the first pass.
3. Reranking
This is the step most RAG tutorials skip, and it's usually the highest-leverage fix available for citation and relevance bugs.
def rerank(query: str, candidates: list, reranker_model, top_n: int = 6):
scored = reranker_model.score(query, [c.text for c in candidates])
ranked = sorted(zip(candidates, scored), key=lambda x: x[1], reverse=True)
return [doc for doc, score in ranked[:top_n]]
A cross-encoder reranker — or an LLM-based relevance judge for lower-volume use cases — trades a small amount of latency for a substantial jump in precision. Retrieval optimizes for recall across the whole corpus; reranking optimizes for precision over the candidates you already pulled. Skipping this step and taking top-k directly from a vector store is one of the most common causes of noisy, contradictory context reaching generation.
4. Deduplication
Retrieved candidates frequently include near-duplicates — different versions of the same document, or restated facts across multiple sources — that consume token budget without adding information, and can actively confuse a model when they disagree with each other.
def deduplicate(candidates: list, similarity_fn, threshold: float = 0.92,
recency_key=None):
kept = []
for doc in candidates:
duplicate_of = None
for existing in kept:
if similarity_fn(doc.text, existing.text) > threshold:
duplicate_of = existing
break
if duplicate_of is None:
kept.append(doc)
elif recency_key and recency_key(doc) > recency_key(duplicate_of):
# Keep the more recent version when duplicates represent
# different versions of the same underlying document.
kept.remove(duplicate_of)
kept.append(doc)
return kept
If your source data has any notion of versioning or supersession — updated policies, revised documentation — this stage needs to know about it explicitly. Embedding similarity alone can't tell a model which of two near-identical documents is current; that's a metadata problem, not a generation problem, and it needs to be solved before the model ever sees the candidates.
5. Context Assembly — With an Explicit Token Budget
This is the part most implementations get wrong by default: concatenate everything relevant and let the model sort it out. Do it deliberately instead.
def assemble_context(system_prompt, retrieved_docs, history, user_query,
max_tokens=8000, token_counter=len):
budget = {
"system": int(max_tokens * 0.10),
"history": int(max_tokens * 0.25),
"retrieved": int(max_tokens * 0.55),
"query": int(max_tokens * 0.10),
}
def trim_to_budget(text_blocks, limit):
kept, used = [], 0
for block in text_blocks:
t = token_counter(block)
if used + t > limit:
break
kept.append(block)
used += t
return kept
# Keep the most recent history turns, trimmed from the oldest end
trimmed_history = trim_to_budget(history[::-1], budget["history"])[::-1]
# Highest-ranked retrieved docs first — and positioned first in the
# final context, since models under-attend to mid-context information.
trimmed_docs = trim_to_budget(
[d.text for d in sorted(retrieved_docs, key=lambda d: -d.score)],
budget["retrieved"],
)
return {
"system": system_prompt,
"context": trimmed_docs, # placed early in the final prompt
"history": trimmed_history, # placed after context
"query": user_query, # placed last
}
Two things matter here more than anything else in this pipeline: placement and budget. Put the most decision-critical retrieved facts near the start or end of the assembled context, never buried in the middle — long-context evaluations consistently show models under-attend to mid-context information, an effect commonly referred to as "lost in the middle." And give every section a hard token budget instead of letting whichever section happens to be largest (usually conversation history, in long-running sessions) silently crowd out the rest.
6. Reuse Layer — Don't Repeat Work You've Already Done
Once retrieval, reranking, and assembly are dialed in, the next optimization most teams miss entirely is reuse. If two queries are semantically close, there's often no need to run the full pipeline twice.
def get_or_run_pipeline(query, cache, run_full_pipeline_fn, generate_fn,
full_reuse_threshold=0.98, partial_reuse_threshold=0.92):
match = cache.find_similar(query, threshold=partial_reuse_threshold)
if match is None:
result = run_full_pipeline_fn(query)
cache.store(query, result)
return result
if match.similarity > full_reuse_threshold:
return match.cached_response # full reuse
else:
# Reuse retrieval + ranking work, regenerate only the final answer
return generate_fn(query, context=match.cached_context)
This is a simplified version of the pattern used by semantic caching layers and by newer, purpose-built projects like Remem, an open-source work-reuse engine designed specifically for RAG and agent pipelines. Instead of a binary cache hit/miss, it evaluates similarity and decides per-request whether to fully reuse a prior response, reuse retrieved context while regenerating the answer, or run the pipeline fresh. For high-traffic apps with repetitive query patterns — support bots, internal knowledge assistants, coding copilots — this kind of tiered reuse is frequently a bigger cost and latency win than swapping to a cheaper model, and unlike a model swap, it doesn't trade off against output quality on the requests that do need a fresh run.
7. Memory for Multi-Turn Sessions
Full conversation replay doesn't scale — it resends the same tokens on every turn, growing latency and cost linearly with conversation length, and it's exactly the mechanism that let history quietly crowd out retrieved context in the assembly stage above. Summarize older turns once a session passes a length threshold:
def manage_conversation_memory(history, summarizer_fn, max_raw_turns=6):
if len(history) <= max_raw_turns:
return history
older_turns = history[:-max_raw_turns]
recent_turns = history[-max_raw_turns:]
summary = summarizer_fn(older_turns)
return [f"[Earlier conversation summary]: {summary}"] + recent_turns
For assistants that need precise recall of specific earlier facts rather than just the gist — a coding assistant remembering an exact variable name from twenty turns ago, for instance — summarization alone will lose detail. In that case, extract discrete facts into a separate structured store instead of relying on summarized prose, and retrieve relevant facts back into context the same way you'd retrieve external documents.
8. Evaluation Harness
A minimal harness for catching regressions before they hit users — run this against a fixed test set on every change to retrieval, ranking, or assembly logic, the same way you'd run unit tests against any other service:
def evaluate_pipeline(test_cases, pipeline_fn, groundedness_checker):
results = []
for case in test_cases:
output = pipeline_fn(case["query"])
retrieved_ids = {d.id for d in output["retrieved_docs"]}
expected_ids = set(case["expected_ids"])
results.append({
"query": case["query"],
"retrieval_recall": len(retrieved_ids & expected_ids) / max(len(expected_ids), 1),
"groundedness": groundedness_checker(output["answer"], output["context"]),
"latency_ms": output["latency_ms"],
})
return results
Prompt changes are cheap to iterate on in a chat window. Pipeline regressions are not, and they're much harder to spot by eyeballing a handful of outputs — a retrieval regression can look, three stages downstream, like a confusing but plausible-sounding wrong answer with no obvious connection to its actual cause.
Putting It Together
def handle_query(user_query, history, vector_store, keyword_index,
reranker, cache, system_prompt):
rewritten = rewrite_query(user_query, history)
cached = cache.find_similar(rewritten, threshold=0.92)
if cached and cached.similarity > 0.98:
return cached.cached_response
candidates = hybrid_retrieve(rewritten, vector_store, keyword_index)
reranked = rerank(rewritten, candidates, reranker)
deduped = deduplicate(reranked, similarity_fn=cosine_sim, recency_key=lambda d: d.updated_at)
managed_history = manage_conversation_memory(history, summarizer_fn=summarize)
context = assemble_context(system_prompt, deduped, managed_history, rewritten)
result = generate(context)
cache.store(rewritten, result)
return result
Every box in this pipeline is a place where a bug can hide — and, importantly, every box is testable independently of the others. That's the real advantage of building a context pipeline instead of a single prompt string: you get a system you can debug stage by stage, instead of a black box you can only tune by trial and error and hope the regression doesn't come back somewhere else.
Common Pitfalls Checklist
Run through this list before assuming a bug is a prompt problem:
☐ Is retrieval returning near-duplicate or superseded documents with no recency signal distinguishing them?
☐ Is there a reranking step, or is the system taking raw top-k from vector search?
☐ Does every context section have an explicit token budget, or can one section (usually history) silently crowd out the rest?
☐ Is the most decision-critical retrieved content positioned near the start/end of context, or is it buried in the middle?
☐ Is conversation history growing unbounded, or is it summarized/extracted past a length threshold?
☐ Is the pipeline re-running in full for near-duplicate queries that could partially or fully reuse prior work?
☐ Are retrieval and groundedness evaluated independently, or only judged by eyeballing final output quality?
Takeaways
- Query rewriting, hybrid retrieval, and reranking fix more RAG problems than prompt tuning does — start there when debugging.
- Deduplicate semantically, not just exactly, and resolve document recency explicitly rather than relying on embedding similarity to imply it.
- Give context sections explicit token budgets, and place the most important information near the start or end of the assembled context, never the middle.
- Add a reuse layer once your pipeline is stable — repeated or near-duplicate queries are common in most production apps, and re-running the full pipeline every time is wasted latency and cost.
- Evaluate retrieval, ranking, and groundedness as separate, independently-testable metrics, not just final output quality.
Prompt engineering is still part of the job. It's just the last five percent of it — and treating it as the whole job is exactly how six weeks of debugging effort ends up pointed at the wrong layer of the system.
Top comments (1)
Great perspective. "Context engineering" feels like a much more accurate description of what production AI teams actually do. The biggest gains rarely come from tweaking prompts—they come from delivering the right context at the right time through retrieval, memory, tool outputs, and structured state.
I also like the emphasis on pipelines rather than individual prompts. Building reliable LLM applications is increasingly about context quality, evaluation, and orchestration instead of chasing prompt tricks. The prompt is just one component; the surrounding system is what ultimately determines reliability. Great read!