How to Use the OpenAI API in a RAG Pipeline (Python)
Learn how to use the OpenAI API in a RAG pipeline with Python: embeddings, the Responses API, streaming, structured outputs, chat history, and costs.
Introduction
Calling the OpenAI API in a RAG pipeline looks like one line of code.
client.responses.create(...), and you're done.
Until a user stares at a blank screen while the whole answer generates.
Or asks "what about express?" and your app has no idea what "that" refers to.
Or your monthly bill arrives.
In the previous article, Build a Simple RAG Application Using Python, we built a working RAG app from scratch.
It loaded documents, chunked them, embedded them, and answered questions with citations.
This time, we focus on the part that talks to OpenAI.
In this tutorial, we'll use the OpenAI API to turn that basic RAG app into something closer to a real product.
By the end, you'll know how to:
· Use the Embeddings API and the Responses API together
· Stream answers token by token
· Return structured JSON answers with verified citations
· Add conversation history, including follow-up questions
· Track tokens and estimate the cost of every question
· Handle API errors and retries properly
· Decide when OpenAI's built-in file search is enough
If you haven't built the app from Article 7 yet, do that first.
We'll reuse its index, so we won't repeat chunking and indexing here.
Quick Answer
To use the OpenAI API in a RAG pipeline, embed your document chunks and each question with the Embeddings API, retrieve the most similar chunks, and send them with the question to the Responses API. Then add streaming, structured outputs, chat history, and token tracking for production use.
The workflow looks like this:
User Question (+ chat history)
↓
Rewrite Follow-Up Into a Standalone Query
↓
Embed the Query (Embeddings API)
↓
Retrieve Top-k Chunks
↓
Generate the Answer (Responses API)
↓
Stream Text or Return Structured JSON
↓
Log Token Usage
Two OpenAI APIs do all the work.
The Embeddings API handles retrieval.
The Responses API handles generation.
What We're Building
We'll add one new file next to Article 7's rag.py:
openai_rag.py
It has two commands.
Ask one question and stream the answer:
python openai_rag.py ask "Can I get a refund after 30 days?"
Start a multi-turn chat:
python openai_rag.py chat
Add --json to either command to get a structured answer instead of streamed text.
After every run, the app prints how many tokens it used and a rough cost.
Architecture
Every question goes through two phases.
RETRIEVAL (every question)
User question + chat history
│
▼
Rewrite follow-up ──► responses.create
│
▼
Embed the query ──► embeddings.create
│
▼
Top-k chunks from index/
GENERATION
instructions + history + context + question
│
▼
responses.create
┌─────┴──────┐
▼ ▼
stream=True responses.parse
│ │
▼ ▼
Live text RAGAnswer JSON
└─────┬──────┘
▼
usage: input, cached, output tokens
1. Prerequisites
You'll need:
· The project from Article 7, with rag.py and a docs/ folder
· Python 3.10 or newer
· An OpenAI API key, from the OpenAI API keys page
We'll use the same two models as Article 7.
Embedding model
text-embedding-3-small
It returns 1,536-dimensional vectors by default, according to OpenAI's embeddings guide.
Chat model
gpt-6-luna
OpenAI's models page describes GPT-6 Luna as its option for cost-sensitive, high-volume workloads.
That fits RAG well.
Retrieval has already found the facts, so the model mostly needs to read and summarize.
Both models can be changed with the EMBED_MODEL and CHAT_MODEL environment variables.
2. Set Up the Project
Move into your Article 7 project:
cd simple-rag
Activate the virtual environment on macOS or Linux:
source .venv/bin/activate
On Windows:
.venv\Scripts\activate
Install or upgrade the dependencies:
pip install --upgrade openai numpy pydantic
We add pydantic for structured outputs.
This tutorial was tested against version 3.29 of the official OpenAI Python SDK.
Check yours:
python -c "import openai; print(openai.__version__)"
Make sure your API key is set:
export OPENAI_API_KEY="sk-..."
Build the index if you haven't already:
python rag.py index docs/
Your project now looks like this:
simple-rag/
├── .venv/
├── docs/
│ ├── refunds.md
│ └── shipping.md
├── index/
│ ├── vectors.npy
│ └── chunks.json
├── rag.py
└── openai_rag.py
We'll build openai_rag.py piece by piece.
The complete file is in Final Code.
3. The Two OpenAI APIs in a RAG Pipeline
A RAG pipeline uses OpenAI in two different places.
Embeddings API
client.embeddings.create() turns text into vectors.
You call it once per chunk at indexing time, then once per question.
Responses API
client.responses.create() generates the answer.
OpenAI's Python SDK README calls it the primary API for interacting with its models.
It takes your system rules as instructions and the conversation as input.
The final text is available as response.output_text.
What about Chat Completions?
client.chat.completions.create() still works.
The SDK README describes it as the previous standard, supported indefinitely.
If your existing RAG app uses it, you don't have to rush a rewrite.
For new code, we'll use the Responses API, because streaming events, structured outputs, conversation state, and file search are all built around it.
4. Embed the Question With the OpenAI Embeddings API
Article 7's rag.py already embedded every chunk in batches.
At question time, we only need to embed one query.
Add the imports and client:
import json
import os
from pathlib import Path
import numpy as np
import openai
from openai import OpenAI
from pydantic import BaseModel
EMBED_MODEL = os.getenv("EMBED_MODEL", "text-embedding-3-small")
CHAT_MODEL = os.getenv("CHAT_MODEL", "gpt-6-luna")
INDEX_DIR = Path("index")
NO_ANSWER = "I don't know based on the provided documents."
client = OpenAI(max_retries=3, timeout=60.0) # reads OPENAI_API_KEY
TOTALS = {"embed": 0, "input": 0, "cached": 0, "output": 0}
TOTALS will collect token counts across every API call.
Now the embedding function:
def embed(texts: list[str]) -> np.ndarray:
response = client.embeddings.create(model=EMBED_MODEL, input=texts)
TOTALS["embed"] += response.usage.prompt_tokens
matrix = np.array([item.embedding for item in response.data], dtype=np.float32)
return matrix / np.linalg.norm(matrix, axis=1, keepdims=True)
Look at this line:
TOTALS["embed"] += response.usage.prompt_tokens
Every embeddings response tells you how many tokens it consumed.
Embedding a short question costs very little, but it's still worth counting.
Retrieval is the same as Article 7:
def load_index() -> tuple[np.ndarray, list[dict]]:
vectors = np.load(INDEX_DIR / "vectors.npy")
records = json.loads((INDEX_DIR / "chunks.json").read_text(encoding="utf-8"))
return vectors, records
def retrieve(query: str, k: int = 4, min_score: float = 0.2) -> list[dict]:
vectors, records = load_index()
scores = vectors @ embed([query])[0]
top = np.argsort(scores)[::-1][:k]
return [{**records[i], "score": float(scores[i])} for i in top if scores[i] >= min_score]
If you want the theory behind this step, read What Are Embeddings?
Should you use the dimensions parameter?
The text-embedding-3 models accept a dimensions parameter that returns shorter vectors:
client.embeddings.create(model=EMBED_MODEL, input=texts, dimensions=512)
Shorter vectors take less storage and make search faster.
OpenAI's embeddings guide says they can be shortened without losing their concept-representing properties, though you trade away some accuracy.
If you use it, use the same value for chunks and questions.
Then rebuild the whole index.
5. Write the RAG Prompt for the Responses API
The prompt is where retrieval meets generation.
Add the instructions:
INSTRUCTIONS = f"""You are a support assistant for our company documents.
Answer using ONLY the context in the latest user message.
Cite the chunk ids you used in square brackets, like [docs/refunds.md#0].
If the context does not contain the answer, reply exactly: {NO_ANSWER}
Keep answers under 120 words."""
Then a function that builds the input list:
def build_input(question: str, hits: list[dict], history: list[dict]) -> list[dict]:
context = "\n\n---\n\n".join(f"[{h['id']}]\n{h['text']}" for h in hits)
turn = {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"}
return history + [turn]
The order is deliberate :
instructions (never changes)
history (grows one turn at a time)
context (changes every question)
question (changes every question)
Stable content goes first.
Changing content goes last.
That order matters for prompt caching, which we'll cover in the cost section.
Notice that the retrieved context is only attached to the latest message.
We don't keep old context in the history.
Old chunks would cost tokens on every later turn, and the next question gets its own fresh retrieval anyway.
6. Stream RAG Answers With the OpenAI API
Without streaming, the user sees nothing until the whole answer is generated.
With streaming, the first words appear almost immediately.
The OpenAI streaming guide shows the pattern: set stream=True and loop over typed events.
Add this:
def stream_answer(question: str, hits: list[dict], history: list[dict]) -> str:
stream = client.responses.create(
model=CHAT_MODEL,
instructions=INSTRUCTIONS,
input=build_input(question, hits, history),
reasoning={"effort": "low"},
max_output_tokens=1000,
stream=True,
)
parts = []
for event in stream:
if event.type == "response.output_text.delta":
print(event.delta, end="", flush=True)
parts.append(event.delta)
elif event.type == "response.completed":
track(event.response.usage)
print()
return "".join(parts)
Two events matter here.
response.output_text.delta
A new piece of text.
We print it right away and keep a copy, because we need the full answer for chat history.
response.completed
The final event.
It carries the complete response object, including token usage.
Two more parameters are worth explaining.
reasoning={"effort": "low"}
Current GPT-6 models can reason before they answer.
In RAG, the hard work is already done by retrieval, so low effort is usually enough.
Raise it if your answers need multi-step reasoning across chunks.
max_output_tokens=1000
This caps how much the model can generate, including reasoning tokens.
It's a safety net for cost.
Don't set it too low, or the response can end up incomplete.
Now the track() helper:
def track(usage) -> None:
if usage:
TOTALS["input"] += usage.input_tokens
TOTALS["cached"] += usage.input_tokens_details.cached_tokens
TOTALS["output"] += usage.output_tokens
Every response reports input tokens, cached input tokens, and output tokens.
7. Return Structured Answers With Citations
Streaming is great for people.
But sometimes code consumes the answer.
A web UI might show sources as clickable chips.
An API might return JSON to a mobile app.
For that, use Structured Outputs.
OpenAI's guide says Structured Outputs makes the model's response adhere to your JSON Schema.
In Python, you define that schema with a Pydantic model.
Add this:
class RAGAnswer(BaseModel):
answer: str
cited_ids: list[str]
found_in_context: bool
Then call responses.parse() instead of responses.create():
def structured_answer(question: str, hits: list[dict], history: list[dict]) -> RAGAnswer:
response = client.responses.parse(
model=CHAT_MODEL,
instructions=INSTRUCTIONS,
input=build_input(question, hits, history),
text_format=RAGAnswer,
reasoning={"effort": "low"},
max_output_tokens=1000,
)
track(response.usage)
result = response.output_parsed
if result is None: # refusal or incomplete output
return RAGAnswer(answer=NO_ANSWER, cited_ids=[], found_in_context=False)
retrieved = {h["id"] for h in hits}
result.cited_ids = [c for c in result.cited_ids if c in retrieved]
return result
response.output_parsed is already a RAGAnswer object.
No json.loads(), no missing keys.
Look at these two lines:
retrieved = {h["id"] for h in hits}
result.cited_ids = [c for c in result.cited_ids if c in retrieved]
A schema guarantees the shape of the answer.
It doesn't guarantee the content is true.
So we keep only citations that point to chunks we actually retrieved.
If the model invents docs/made-up.md#9, it gets dropped.
Why check for None?
The structured outputs guide calls out two edge cases: a safety refusal, and output that stops early because it hit max_output_tokens.
In either case there's no parsed object, so we fall back to the safe answer.
8. Add Conversation History to Your RAG Chatbot
Real users don't ask perfect standalone questions.
They ask:
"How long does shipping take?"
And then:
"What about express?"
The second question alone retrieves badly.
"What about express?" has no useful words to search with.
So we fix the query before retrieval.
Rewrite follow-up questions
Add this:
REWRITE = """Rewrite the user's latest message as one standalone search query.
Use the chat history to replace words like "it", "that" or "there".
Return only the query."""
def rewrite_question(question: str, history: list[dict]) -> str:
if not history:
return question
response = client.responses.create(
model=CHAT_MODEL,
instructions=REWRITE,
input=history + [{"role": "user", "content": question}],
reasoning={"effort": "low"},
max_output_tokens=300,
)
track(response.usage)
return response.output_text.strip() or question
With history, "What about express?" becomes something like:
How long does express shipping take?
That query finds the shipping chunk.
The first question of a conversation skips the rewrite, so it costs nothing extra.
Put the pieces together
Now one function handles a whole turn:
def ask(question: str, history: list[dict], k: int = 4, as_json: bool = False) -> str:
query = rewrite_question(question, history)
hits = retrieve(query, k=k)
if not hits:
print(NO_ANSWER)
return NO_ANSWER
if as_json:
result = structured_answer(question, hits, history)
print(result.model_dump_json(indent=2))
return result.answer
return stream_answer(question, hits, history)
We search with the rewritten query.
But we still send the user's original question to the model, together with the history, so the answer sounds natural.
Keep the history short
The chat loop stores plain questions and answers:
def chat(k: int = 4, as_json: bool = False, max_turns: int = 6) -> None:
history: list[dict] = []
print("Ask about your documents. Type 'exit' to quit.")
while True:
try:
question = input("\nYou: ").strip()
except (EOFError, KeyboardInterrupt):
break
if question.lower() in {"", "exit", "quit"}:
break
print("Assistant: ", end="", flush=True)
reply = ask(question, history, k=k, as_json=as_json)
history += [{"role": "user", "content": question}, {"role": "assistant", "content": reply}]
history = history[-max_turns * 2:] # keep the last few turns only
Look at this line:
history = history[-max_turns * 2:]
Every turn you keep is sent again with every new request.
Six turns is plenty for a support chat.
What about previous_response_id?
The Responses API can store conversation state for you.
Pass previous_response_id=response.id and OpenAI chains the turns.
Or use the Conversations API for long-lived threads.
Both are explained in OpenAI's conversation state guide.
We manage history manually here for two reasons.
You can see exactly what's sent.
And you control what gets dropped.
The guide also notes that with previous_response_id, all previous input tokens in the chain are still billed as input tokens.
Server-side state is convenient, but it isn't free.
9. Control OpenAI API Costs in RAG
RAG costs come from three places:
Embeddings → once per chunk at indexing, once per question
Input → instructions + history + retrieved chunks + question
Output → the answer (and any reasoning tokens)
In most RAG apps, input dominates.
Every question sends several chunks of context.
Here's how to keep each part under control.
Measure first
Add a small report:
PRICE_IN = float(os.getenv("PRICE_IN", "0.10")) # USD per 1M input tokens
PRICE_OUT = float(os.getenv("PRICE_OUT", "0.50")) # USD per 1M output tokens
def report() -> None:
cost = (TOTALS["input"] * PRICE_IN + TOTALS["output"] * PRICE_OUT) / 1_000_000
print(
f"\n[tokens] embed={TOTALS['embed']} input={TOTALS['input']} "
f"(cached {TOTALS['cached']}) output={TOTALS['output']} | chat cost ~${cost:.6f}"
)
The defaults match GPT-6 Luna's standard prices at the time of writing: $0.10 per million input tokens and $0.50 per million output tokens.
Prices change, so check the API pricing page and override them with environment variables.
This is a rough estimate.
It ignores caching discounts and cache-write fees.
Send fewer chunks
k is your biggest cost lever.
Going from k=8 to k=4 roughly halves the context you send.
Test with your real questions and use the smallest k that still finds the right chunk.
Keep chunks focused
Smaller, cleaner chunks mean less irrelevant text in every prompt.
See How Document Chunking Works in RAG for how to tune them.
Let prompt caching work
OpenAI's prompt caching guide says caching is enabled by default for supported models.
When requests share the same prompt prefix, the cached part is billed at a reduced rate.
On current models, the reusable prefix has to be at least 1,024 tokens.
So our short instructions won't be cached on their own.
Longer instructions and a growing chat history will.
That's why build_input() keeps stable content first and the changing context last.
Check cached in the report to see if it's working.
Cap the output
max_output_tokens and "Keep answers under 120 words" both limit output tokens.
Low reasoning effort keeps reasoning tokens down too.
Skip calls you don't need
If no chunk passes min_score, ask() returns "I don't know" without calling the model.
The first question in a chat skips the rewrite call.
The cheapest API call is the one you don't make.
10. Handle OpenAI API Errors and Retries
APIs fail.
Networks drop, rate limits hit, and keys expire.
The OpenAI Python SDK already retries some failures for you.
According to the SDK README, connection errors, 408, 409, 429, and 5xx errors are retried twice by default, with exponential backoff.
We raised that to three when creating the client:
client = OpenAI(max_retries=3, timeout=60.0)
The timeout stops a stuck request from hanging your app.
When retries run out, the SDK raises a typed exception.
We catch the common ones in the CLI:
try:
...
except openai.AuthenticationError:
raise SystemExit("Authentication failed. Check OPENAI_API_KEY.")
except openai.RateLimitError as e:
raise SystemExit(f"Rate limited or out of quota after retries: {e}")
except openai.APIConnectionError:
raise SystemExit("Could not reach the OpenAI API. Check your network.")
except FileNotFoundError:
raise SystemExit("No index found. Run: python rag.py index docs/")
finally:
report()
finally makes sure the token report prints even when something fails.
11. What Actually Happened?
Here's the full lifecycle of a follow-up question in chat mode:
User types "What about express?"
↓
rewrite_question() turns it into a standalone query
↓
embed() + retrieve() find the shipping chunk
↓
build_input() adds history + context + question
↓
responses.create(stream=True) streams the answer
↓
response.completed reports token usage
↓
The turn is added to history, oldest turns are dropped
The retrieval code barely changed from Article 7.
Everything new happened around the model call.
That's where most real-world RAG work lives.
12. Final Code
Here's the complete openai_rag.py:
"""openai_rag.py - the generation side of RAG with the OpenAI API.
Streaming answers, structured outputs with citations, chat history,
and token tracking. Uses the index built by rag.py (Article 7).
"""
import argparse
import json
import os
from pathlib import Path
import numpy as np
import openai
from openai import OpenAI
from pydantic import BaseModel
EMBED_MODEL = os.getenv("EMBED_MODEL", "text-embedding-3-small")
CHAT_MODEL = os.getenv("CHAT_MODEL", "gpt-6-luna")
INDEX_DIR = Path("index")
NO_ANSWER = "I don't know based on the provided documents."
# USD per 1M tokens. Check https://developers.openai.com/api/docs/pricing before trusting these.
PRICE_IN = float(os.getenv("PRICE_IN", "0.10"))
PRICE_OUT = float(os.getenv("PRICE_OUT", "0.50"))
client = OpenAI(max_retries=3, timeout=60.0) # reads OPENAI_API_KEY
TOTALS = {"embed": 0, "input": 0, "cached": 0, "output": 0}
# 1. Embeddings ----------------------------------------------------------
def embed(texts: list[str]) -> np.ndarray:
response = client.embeddings.create(model=EMBED_MODEL, input=texts)
TOTALS["embed"] += response.usage.prompt_tokens
matrix = np.array([item.embedding for item in response.data], dtype=np.float32)
return matrix / np.linalg.norm(matrix, axis=1, keepdims=True)
# 2. Retrieval (same idea as Article 7) ----------------------------------
def load_index() -> tuple[np.ndarray, list[dict]]:
vectors = np.load(INDEX_DIR / "vectors.npy")
records = json.loads((INDEX_DIR / "chunks.json").read_text(encoding="utf-8"))
return vectors, records
def retrieve(query: str, k: int = 4, min_score: float = 0.2) -> list[dict]:
vectors, records = load_index()
scores = vectors @ embed([query])[0]
top = np.argsort(scores)[::-1][:k]
return [{**records[i], "score": float(scores[i])} for i in top if scores[i] >= min_score]
# 3. Prompt --------------------------------------------------------------
INSTRUCTIONS = f"""You are a support assistant for our company documents.
Answer using ONLY the context in the latest user message.
Cite the chunk ids you used in square brackets, like [docs/refunds.md#0].
If the context does not contain the answer, reply exactly: {NO_ANSWER}
Keep answers under 120 words."""
def build_input(question: str, hits: list[dict], history: list[dict]) -> list[dict]:
context = "\n\n---\n\n".join(f"[{h['id']}]\n{h['text']}" for h in hits)
turn = {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"}
return history + [turn]
def track(usage) -> None:
if usage:
TOTALS["input"] += usage.input_tokens
TOTALS["cached"] += usage.input_tokens_details.cached_tokens
TOTALS["output"] += usage.output_tokens
# 4. Streaming answer ----------------------------------------------------
def stream_answer(question: str, hits: list[dict], history: list[dict]) -> str:
stream = client.responses.create(
model=CHAT_MODEL,
instructions=INSTRUCTIONS,
input=build_input(question, hits, history),
reasoning={"effort": "low"},
max_output_tokens=1000,
stream=True,
)
parts = []
for event in stream:
if event.type == "response.output_text.delta":
print(event.delta, end="", flush=True)
parts.append(event.delta)
elif event.type == "response.completed":
track(event.response.usage)
print()
return "".join(parts)
# 5. Structured answer ---------------------------------------------------
class RAGAnswer(BaseModel):
answer: str
cited_ids: list[str]
found_in_context: bool
def structured_answer(question: str, hits: list[dict], history: list[dict]) -> RAGAnswer:
response = client.responses.parse(
model=CHAT_MODEL,
instructions=INSTRUCTIONS,
input=build_input(question, hits, history),
text_format=RAGAnswer,
reasoning={"effort": "low"},
max_output_tokens=1000,
)
track(response.usage)
result = response.output_parsed
if result is None: # refusal or incomplete output
return RAGAnswer(answer=NO_ANSWER, cited_ids=[], found_in_context=False)
retrieved = {h["id"] for h in hits}
result.cited_ids = [c for c in result.cited_ids if c in retrieved]
return result
# 6. Conversation history -----------------------------------------------
REWRITE = """Rewrite the user's latest message as one standalone search query.
Use the chat history to replace words like "it", "that" or "there".
Return only the query."""
def rewrite_question(question: str, history: list[dict]) -> str:
if not history:
return question
response = client.responses.create(
model=CHAT_MODEL,
instructions=REWRITE,
input=history + [{"role": "user", "content": question}],
reasoning={"effort": "low"},
max_output_tokens=300,
)
track(response.usage)
return response.output_text.strip() or question
def ask(question: str, history: list[dict], k: int = 4, as_json: bool = False) -> str:
query = rewrite_question(question, history)
hits = retrieve(query, k=k)
if not hits:
print(NO_ANSWER)
return NO_ANSWER
if as_json:
result = structured_answer(question, hits, history)
print(result.model_dump_json(indent=2))
return result.answer
return stream_answer(question, hits, history)
def chat(k: int = 4, as_json: bool = False, max_turns: int = 6) -> None:
history: list[dict] = []
print("Ask about your documents. Type 'exit' to quit.")
while True:
try:
question = input("\nYou: ").strip()
except (EOFError, KeyboardInterrupt):
break
if question.lower() in {"", "exit", "quit"}:
break
print("Assistant: ", end="", flush=True)
reply = ask(question, history, k=k, as_json=as_json)
history += [{"role": "user", "content": question}, {"role": "assistant", "content": reply}]
history = history[-max_turns * 2:] # keep the last few turns only
# 7. Cost report ---------------------------------------------------------
def report() -> None:
cost = (TOTALS["input"] * PRICE_IN + TOTALS["output"] * PRICE_OUT) / 1_000_000
print(
f"\n[tokens] embed={TOTALS['embed']} input={TOTALS['input']} "
f"(cached {TOTALS['cached']}) output={TOTALS['output']} | chat cost ~${cost:.6f}"
)
# CLI --------------------------------------------------------------------
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="RAG with the OpenAI API")
sub = parser.add_subparsers(dest="command", required=True)
one = sub.add_parser("ask")
one.add_argument("question")
talk = sub.add_parser("chat")
for p in (one, talk):
p.add_argument("-k", type=int, default=4)
p.add_argument("--json", action="store_true", help="structured output")
args = parser.parse_args()
try:
if args.command == "ask":
ask(args.question, history=[], k=args.k, as_json=args.json)
else:
chat(k=args.k, as_json=args.json)
except openai.AuthenticationError:
raise SystemExit("Authentication failed. Check OPENAI_API_KEY.")
except openai.RateLimitError as e:
raise SystemExit(f"Rate limited or out of quota after retries: {e}")
except openai.APIConnectionError:
raise SystemExit("Could not reach the OpenAI API. Check your network.")
except FileNotFoundError:
raise SystemExit("No index found. Run: python rag.py index docs/")
finally:
report()
The important pattern is:
Retrieve: question (+ history) → standalone query → embedding → top-k chunks
Generate: instructions + history + context + question → Responses API
Deliver: streamed text for people, parsed JSON for code
Measure: usage tokens on every call
13. Run the OpenAI RAG App
Ask a single question:
python openai_rag.py ask "Can I get a refund after 30 days?"
The answer streams in word by word, then the report prints.
Expected output (your wording and token counts will differ):
After 30 days, refunds are issued as store credit only [docs/refunds.md#0].
[tokens] embed=... input=... (cached 0) output=... | chat cost ~$0.0000...
Ask for structured output:
python openai_rag.py ask "Can I get a refund after 30 days?" --json
Expected shape:
{
"answer": "After 30 days, refunds are issued as store credit only.",
"cited_ids": [
"docs/refunds.md#0"
],
"found_in_context": true
}
Start a chat:
python openai_rag.py chat
Try a follow-up:
You: How long does shipping take?
Assistant: Standard shipping takes 3 to 5 business days within India [docs/shipping.md#0].
You: What about express?
Assistant: Express shipping takes 1 to 2 business days and costs extra [docs/shipping.md#0].
The second answer only works because of the rewrite step.
Type exit to quit and see the token report for the whole session.
14. Common OpenAI API Errors in RAG
openai.AuthenticationError
Your API key is missing or wrong in this terminal.
Check it:
echo $OPENAI_API_KEY
openai.RateLimitError
You hit a rate limit, or your account is out of credit.
The SDK already retried.
If it keeps happening, check your usage and limits in the OpenAI dashboard.
openai.NotFoundError mentioning the model
The model name is wrong or not available to your account.
Check the spelling against the models page, or set CHAT_MODEL to a model you have access to.
openai.BadRequestError about an unsupported parameter
Some parameters only work with some models.
If your chosen model rejects reasoning, remove that argument.
Structured answer is always "I don't know"
output_parsed is None, usually because the response hit max_output_tokens.
Raise the limit or lower the reasoning effort.
No index found
You haven't built the index yet.
Run:
python rag.py index docs/
15. Common OpenAI RAG Mistakes
1. Re-Sending Old Context Every Turn
Storing retrieved chunks in the chat history makes every later request bigger.
Store plain questions and answers, and retrieve fresh context per turn.
2. Retrieving With the Raw Follow-Up
"What about it?" is a terrible search query.
Rewrite it with history first.
3. Trusting Citations Blindly
A valid JSON schema doesn't mean valid sources.
Check cited ids against what you actually retrieved.
4. Not Logging Token Usage
You can't reduce a cost you don't measure.
Log usage on every call, per user if you can.
5. Putting Dynamic Content First
Timestamps or user names at the top of your instructions change the prompt prefix on every call.
That can stop prompt caching from working.
Keep stable text first.
6. Unlimited History
Every turn you keep is billed again on every request.
Cap the history length.
7. Mixing Embedding Models
Questions and chunks must use the same embedding model, and the same dimensions value.
Change either one, and rebuild the index.
Custom RAG vs OpenAI File Search
OpenAI also offers a hosted option.
File search is a built-in tool in the Responses API.
You upload files to an OpenAI vector store, and the model searches them when it needs to.
A minimal version looks like this:
store = client.vector_stores.create(name="support-docs")
with open("docs/refunds.md", "rb") as f:
client.vector_stores.files.upload_and_poll(vector_store_id=store.id, file=f)
response = client.responses.create(
model=CHAT_MODEL,
input="Can I get a refund after 30 days?",
tools=[{"type": "file_search", "vector_store_ids": [store.id]}],
)
print(response.output_text)
So which should you use?
Custom Pipeline (This Tutorial)
Best for: control, debugging, and keeping your data and vectors where you choose.
Pros: you choose chunking, scores, filters, prompts, and the vector database. Every step is visible.
Cons: more code to write and maintain.
OpenAI File Search
Best for: quick prototypes and teams that don't want to run retrieval infrastructure.
Pros: OpenAI handles parsing, chunking, embedding, and search. Answers come with file citations.
Cons: less control over retrieval, your files are stored with OpenAI, and file search has its own storage and per-call pricing on top of model tokens.
A practical approach:
Prototype with file search
↓
Measure answer quality on real questions
↓
Move to a custom pipeline when you need more control
Either way, the generation techniques in this article still apply.
Responses API vs Chat Completions for RAG
You'll see both in RAG tutorials.
Responses API
Best for: new projects.
Pros: typed streaming events, responses.parse() for structured outputs, optional server-side conversation state, and built-in tools like file search.
Cons: older tutorials and libraries may not use it yet.
Chat Completions API
Best for: existing apps that already work.
Pros: widely known, and the SDK README says it's supported indefinitely.
Cons: built-in tools like file search are designed around the Responses API.
If you're starting fresh, use the Responses API.
16. Try It Yourself
Don't just copy the code.
Add one feature:
--show-sources
When the flag is set, print each retrieved chunk id and score before the answer, plus the rewritten query in chat mode.
Example:
query: How long does express shipping take?
docs/shipping.md#0 0.58
docs/refunds.md#0 0.21
Bonus
Add a --max-cost flag.
Stop the chat when the estimated cost passes the limit.
If both work, you understand how retrieval, generation, and cost connect.
FAQs
How do I use the OpenAI API for RAG?
Use the Embeddings API to turn your document chunks and each question into vectors, then retrieve the most similar chunks.
Send those chunks and the question to the Responses API with instructions to answer only from the context.
Should I use the Responses API or Chat Completions for RAG?
For new projects, use the Responses API.
OpenAI's Python SDK calls it the primary API, and it supports streaming events, structured outputs, conversation state, and file search.
Chat Completions still works and is supported, so existing apps don't need an urgent rewrite.
Does OpenAI have built-in RAG?
Yes.
The file search tool in the Responses API searches files you upload to an OpenAI vector store and returns answers with file citations.
It's a good fit for prototypes, while a custom pipeline gives you more control over chunking, retrieval, and where your data lives.
Which OpenAI embedding model is best for RAG?
Start with text-embedding-3-small.
Try text-embedding-3-large only if your own test questions show better retrieval.
Use the same model for chunks and questions, and re-embed everything when you switch.
How do I add chat history to a RAG chatbot?
Keep the last few user and assistant messages and send them with each request.
Before retrieval, rewrite follow-up questions into standalone search queries using that history, so "What about express?" becomes a query that actually finds the right chunk.
How can I reduce OpenAI API costs in a RAG app?
Send fewer and smaller chunks, cap chat history and output tokens, and keep stable instructions at the start of the prompt so prompt caching can apply.
Log the usage field on every response so you know where tokens go.
Can I stream RAG answers with the OpenAI API?
Yes.
Pass stream=True to client.responses.create() and print each response.output_text.delta event as it arrives.
The final response.completed event includes token usage.
Key Takeaways
You've now upgraded a basic RAG app into a more production-ready one.
Remember these points:
· A RAG pipeline uses two OpenAI APIs: Embeddings for retrieval and Responses for generation.
· Streaming with stream=True makes answers feel fast.
· responses.parse() with a Pydantic model returns structured answers you can use in code.
· Always check citations against the chunks you actually retrieved.
· Rewrite follow-up questions before retrieval, or chat history won't help search.
· Keep history short and store plain turns, not old context.
· Log usage on every call, and keep stable prompt content first for caching.
· File search is a quick hosted alternative; a custom pipeline gives you control.
The biggest lesson is simple:
The model call is one line. Everything around it, streaming, structure, history, and cost, is what makes a RAG app usable.
Conclusion
In Article 7, we proved RAG is just search first, generate second.
In this article, we made the "generate" part work like a real product.
Answers stream instead of making users wait.
Code gets clean JSON with checked citations.
Follow-up questions work.
And every question shows you what it cost.
None of this needed a framework.
But you may have noticed how much glue code we wrote for loading, splitting, prompting, and history.
That's exactly the work frameworks try to take off your hands.
Next, we'll rebuild the same pipeline with LangChain and see what it simplifies, and what it hides.
That's where things get more interesting.
Continue Learning
You're currently at Article 8 in the RAG series.
Previous Articles
1. What Is RAG? Retrieval-Augmented Generation Explained
Learn the basic RAG architecture and understand how retrieval and generation work together.
Understand how text is converted into numerical representations for semantic retrieval.
Learn how vector databases store and search embeddings.
4. Semantic Search vs Keyword Search
Understand the difference between exact-term matching and meaning-based retrieval.
5. How Document Chunking Works in RAG
Learn why documents are split into chunks and how chunking affects retrieval quality.
7. Build a Simple RAG Application Using Python
Build a complete RAG app from scratch with OpenAI embeddings and NumPy.
Next Articles
9. LangChain for RAG: A Beginner's Guide
Rebuild this pipeline with LangChain and see how it handles loading, chunking, retrieval, and generation.
10. LlamaIndex Explained: Build LLM Applications with Your Data
Explore another popular framework for building document-based RAG applications.
11. Build a Chat with PDF Application Using RAG
Handle real PDFs, page numbers, and citations in a chat interface.
12. PostgreSQL and pgvector: Build a RAG System with Vector Search
Move the index from NumPy files into a real vector database.
Have a Question?
If something isn't clear or your RAG app isn't working as expected, leave a comment below.
We'll do our best to explain it with simple words, practical examples, and code you can actually understand.
Happy Coding! 🚀
Follow TheTechLane
Keep learning AI, development, and modern developer technologies with TheTechLane.
Learn • Build • Share
Follow @thetechlanehq for more developer guides, tutorials, and practical technology content.