Query, Retrieve, and Generate
Write the query pipeline that embeds the user question, retrieves the top-k chunks, formats an augmented prompt, calls the LLM, and returns a cited answer.
Query, Retrieve, and Generate is a free AI Engineering Academy lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Engineering Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.
The Query Pipeline: End to End
The query pipeline is the online half of RAG — the code that runs in real time when a user asks a question. It connects all the components built during indexing: the embedding model, the vector store, the prompt template, and the LLM. A well-implemented query pipeline completes in under 500ms for most workloads and produces grounded, cited answers. In this lesson we build each step from scratch.
Step 1: Embed the User Query
The first step is to convert the user's natural language question into a vector embedding using the same model used during indexing. This embedding encodes the semantic meaning of the question and will be compared against document chunk embeddings in the vector store. Keep this step fast — use a lightweight model like text-embedding-3-small and cache embeddings for repeated identical queries.
from openai import OpenAI
client = OpenAI()
def embed_query(question: str) -> list:
response = client.embeddings.create(
model='text-embedding-3-small',
input=[question]
)
return response.data[0].embedding
user_question = 'What is our remote work policy?'
query_vector = embed_query(user_question)
print(f'Query embedded: {len(query_vector)}-dim vector')Step 2: Retrieve Top-K Chunks
Send the query vector to the vector store to find the K most semantically similar chunks. The returned matches are ranked by cosine similarity score (typically 0.0 to 1.0, higher is better). The ideal K value balances context richness against context window cost: K=5 is a common starting point. You can also apply metadata filters here to restrict retrieval to a specific department, document type, or date range.
def retrieve_chunks(query_vector, index, top_k=5, filters=None):
query_params = {
'vector': query_vector,
'top_k': top_k,
'include_metadata': True
}
if filters:
query_params['filter'] = filters
results = index.query(**query_params)
chunks = []
for match in results.matches:
chunks.append({
'score': match.score,
'text': match.metadata['text'],
'source': match.metadata.get('source', ''),
'page': match.metadata.get('page', '')
})
return chunksStep 3: Score Threshold Filtering
Not all retrieved chunks are genuinely relevant — some may have low similarity scores but still rank in the top-K because the query is outside the index's coverage. Apply a minimum score threshold to filter out low-confidence matches. If all retrieved chunks fall below the threshold, return a 'no information found' response rather than sending irrelevant context to the LLM, which would produce a worse answer than refusing gracefully.
MIN_SCORE_THRESHOLD = 0.75
def filter_by_score(chunks, threshold=MIN_SCORE_THRESHOLD):
relevant = [c for c in chunks if c['score'] >= threshold]
if not relevant:
print(f'No chunks above threshold {threshold}. Scores: {[c["score"] for c in chunks]}')
return relevant
retrieved = retrieve_chunks(query_vector, index, top_k=5)
filtered = filter_by_score(retrieved)
if not filtered:
print('Responding: no relevant information found')Step 4: Format the Context Block
Assemble the retrieved chunks into a structured context block that the LLM will read. Label each chunk with its source so the model can cite it accurately. Add a separator between chunks for clarity. Keep total context within your token budget — count tokens with tiktoken and truncate or drop lower-scoring chunks if you exceed the limit. The context block is inserted into the prompt between the system instruction and the user question.
def format_context(chunks):
parts = []
for i, chunk in enumerate(chunks, start=1):
source_label = chunk['source']
if chunk.get('page'):
source_label += f", page {chunk['page']}"
parts.append(
f'[Document {i} | Source: {source_label}]\n{chunk["text"]}'
)
return '\n\n---\n\n'.join(parts)
context = format_context(filtered)
print(f'Context block: {len(context)} characters')Step 5: Build the Augmented Prompt
Combine the context block, the system instruction, and the user question into the final prompt. The system message tells the model to use only the provided context and to cite sources. The user message contains the formatted context followed by the question. This clear separation prevents the model from mixing context content with the question and makes the boundary between retrieved data and user input unambiguous.
def build_prompt(question, context):
system_message = (
'You are a helpful assistant. Answer the question using ONLY '
'the information in the provided documents. '
'Cite the document number(s) used, like [Doc 1]. '
'If the documents do not contain the answer, say so.'
)
user_message = (
f'Documents:\n\n{context}\n\n'
f'Question: {question}'
)
return system_message, user_messageStep 6: Call the LLM and Get the Answer
Send the assembled prompt to the LLM using the Chat Completions API. Use a low temperature (0.0 to 0.3) for factual Q&A to get consistent, grounded answers. Higher temperatures produce more creative responses but increase the risk of the model adding information beyond what is in the context. Parse the response and return both the answer text and the retrieved sources so your application can display citations to the user.
def generate_answer(question, context, sources):
system_msg, user_msg = build_prompt(question, context)
response = client.chat.completions.create(
model='gpt-4o',
temperature=0.1, # low temperature for factual Q&A
messages=[
{'role': 'system', 'content': system_msg},
{'role': 'user', 'content': user_msg}
]
)
answer = response.choices[0].message.content
return {
'answer': answer,
'sources': sources,
'tokens_used': response.usage.total_tokens
}Putting It All Together
The complete query pipeline calls these steps in sequence. Each step is a pure function you can test independently, and the data flows cleanly from one step to the next. Adding logging at each step makes the pipeline observable — you can see exactly which chunks were retrieved, what score they had, how the context was assembled, and how many tokens were used. This visibility is essential for debugging and improving retrieval quality.
def answer_question(user_question, vector_index):
# Step 1: Embed query
q_vector = embed_query(user_question)
# Step 2: Retrieve
chunks = retrieve_chunks(q_vector, vector_index, top_k=5)
# Step 3: Filter low-confidence matches
chunks = filter_by_score(chunks, threshold=0.70)
if not chunks:
return {'answer': 'I do not have information about that topic.', 'sources': []}
# Step 4 & 5: Format and build prompt
context = format_context(chunks)
sources = [c['source'] for c in chunks]
# Step 6: Generate
return generate_answer(user_question, context, sources)Latency Optimization
The query pipeline has two I/O bound steps: the embedding call and the LLM call. Run them without unnecessary waits: the embedding call is fast (<100ms), the LLM call is slow (500ms-3s). To reduce perceived latency, stream the LLM response so tokens appear as they are generated rather than waiting for the full response. Cache the embedding of repeated identical queries to avoid redundant API calls.
async def answer_question_streaming(question, index):
q_vector = embed_query(question)
chunks = retrieve_chunks(q_vector, index, top_k=5)
chunks = filter_by_score(chunks)
if not chunks:
yield 'I do not have information about that topic.'
return
context = format_context(chunks)
system_msg, user_msg = build_prompt(question, context)
stream = await client.chat.completions.create(
model='gpt-4o',
stream=True,
messages=[
{'role': 'system', 'content': system_msg},
{'role': 'user', 'content': user_msg}
]
)
async for chunk in stream:
delta = chunk.choices[0].delta.content or ''
yield deltaLogging for Observability
Production RAG pipelines need structured logging so you can diagnose when retrieval fails or the LLM gives a bad answer. Log the query, retrieved chunk IDs and scores, context token count, answer, and latency for every request. Store these logs in a database or observability platform. When users report bad answers, you can replay the exact query and inspect which chunks were retrieved and why they were insufficient.
import time
import logging
import json
def answer_question_with_logging(question, index):
start = time.time()
q_vector = embed_query(question)
chunks = retrieve_chunks(q_vector, index, top_k=5)
chunks = filter_by_score(chunks)
context = format_context(chunks)
result = generate_answer(question, context, [c['source'] for c in chunks])
latency_ms = (time.time() - start) * 1000
log_entry = {
'question': question,
'num_chunks_retrieved': len(chunks),
'chunk_scores': [c['score'] for c in chunks],
'tokens_used': result.get('tokens_used'),
'latency_ms': round(latency_ms)
}
logging.info(json.dumps(log_entry))
return resultCaching Query Embeddings
If your application receives many repeated or near-identical queries — such as FAQ bots where users often ask the same questions — caching query embeddings is a simple, high-impact optimization. Hash the query string, check a Redis cache for the corresponding embedding, and only call the embedding API on a cache miss. Embedding cache hit rates of 30-60% are common in production FAQ and support chatbots, eliminating substantial API cost and reducing latency by 50-100ms per cached query.
import hashlib
import json
import redis
r = redis.Redis(host='localhost', port=6379)
EMBED_CACHE_TTL = 86400 # 24 hours
def embed_query_cached(question):
cache_key = 'embed:' + hashlib.sha256(question.encode()).hexdigest()
cached = r.get(cache_key)
if cached:
return json.loads(cached) # cache hit
# Cache miss: call the API
vector = embed_query(question)
r.setex(cache_key, EMBED_CACHE_TTL, json.dumps(vector))
return vectorQuick Check
Test your understanding of AI Engineering concepts from this lesson.
Lesson Recap
In this lesson you learned: the six-step query pipeline (embed query, retrieve chunks, filter by score, format context, build prompt, generate answer), score threshold filtering to handle queries outside the index coverage, and production enhancements including streaming responses, structured logging, and latency optimization. Next up we learn how to evaluate whether your complete RAG system is actually working correctly.
Frequently asked questions
Is the “Query, Retrieve, and Generate” lesson free?
Yes — the full text of “Query, Retrieve, and Generate” is free to read here on the web, and the AI Engineering Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Engineering Academy course, upgrade to CoddyKit PRO.
What will I learn in “Query, Retrieve, and Generate”?
Write the query pipeline that embeds the user question, retrieves the top-k chunks, formats an augmented prompt, calls the LLM, and returns a cited answer. You practise AI Engineering Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.
Do I need any experience to start AI Engineering Academy?
No prior experience is required. AI Engineering Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.
How long does the “Query, Retrieve, and Generate” lesson take?
Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.
Can I write and run code in this AI Engineering Academy lesson?
Yes. Every AI Engineering Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.