We are going to build a document Q&A agent that retrieves relevant passages from private text files before generating an answer. If you need to ground LLM outputs in your own data without sending documents to a third-party vector database service, this gives you a fully self-contained pipeline using Oxlo.ai for both embeddings and inference.
What you'll need
- Python 3.10 or newer
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK and NumPy:
pip install openai numpy - A plain-text file to search. The code below assumes
handbook.txtlives in the same directory.
Step 1: Configure the Oxlo.ai client
Set up the OpenAI-compatible client pointing at Oxlo.ai. You will use this single client for both embeddings and chat completion calls.
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ.get("OXLO_API_KEY")
)
Step 2: Chunk your documents
Split the text into overlapping word windows so semantically similar passages stay intact. A simple fixed-size splitter is enough for most internal docs.
def chunk_text(text, chunk_size=300, overlap=50):
words = text.split()
chunks = []
for i in range(0, len(words), chunk_size - overlap):
chunk = " ".join(words[i:i + chunk_size])
chunks.append(chunk)
return chunks
with open("handbook.txt", "r", encoding="utf-8") as f:
raw_text = f.read()
chunks = chunk_text(raw_text)
print(f"Created {len(chunks)} chunks")
Step 3: Embed chunks with Oxlo.ai
Send the chunks through Oxlo.ai's embeddings endpoint and store the vectors in a NumPy array. I am using BGE-Large because it performs well on retrieval tasks.
import numpy as np
def embed_texts(texts):
response = client.embeddings.create(
model="bge-large",
input=texts
)
return np.array([item.embedding for item in response.data])
chunk_embeddings = embed_texts(chunks)
print(f"Embedding matrix shape: {chunk_embeddings.shape}")
Step 4: Build cosine-similarity retrieval
Normalize the vectors so a dot product equals cosine similarity, then write a function that returns the top-k chunks for any query.
def normalize(vectors):
norms = np.linalg.norm(vectors, axis=1, keepdims=True)
return vectors / norms
chunk_norms = normalize(chunk_embeddings)
def retrieve(query, k=3):
q_emb = embed_texts([query])
q_norm = normalize(q_emb)
scores = np.dot(chunk_norms, q_norm.T).flatten()
top_idx = np.argsort(scores)[-k:][::-1]
return [(chunks[i], float(scores[i])) for i in top_idx]
Step 5: Write the RAG system prompt
Lock the model into using only retrieved context. The prompt below requires inline citations and a fallback when no answer exists.
SYSTEM_PROMPT = """You are a precise research assistant. Answer the user's question using ONLY the retrieved context provided below.
Rules:
1. Cite the source passage number in square brackets, like [1] or [2].
2. If the context does not contain the answer, say "I don't have enough information to answer that."
3. Keep answers concise and directly supported by the text.
Retrieved context:
{context}
"""
Step 6: Assemble the agent
Combine retrieval and generation into one function. The query is embedded, the top chunks are injected into the prompt, and the response is generated by Llama 3.3 70B on Oxlo.ai.
def ask_agent(question):
passages = retrieve(question, k=3)
context_blocks = []
for idx, (text, score) in enumerate(passages, 1):
context_blocks.append(f"[{idx}] {text}")
context = "\n\n".join(context_blocks)
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT.format(context=context)},
{"role": "user", "content": question},
],
temperature=0.1
)
return {
"answer": response.choices[0].message.content,
"sources": passages
}
Run it
Save the complete script below as rag_agent.py, set your API key, and run it against any text file.
import os
import numpy as np
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ["OXLO_API_KEY"]
)
def chunk_text(text, chunk_size=300, overlap=50):
words = text.split()
chunks = []
for i in range(0, len(words), chunk_size - overlap):
chunk = " ".join(words[i:i + chunk_size])
chunks.append(chunk)
return chunks
def embed_texts(texts):
response = client.embeddings.create(model="bge-large", input=texts)
return np.array([item.embedding for item in response.data])
def normalize(vectors):
norms = np.linalg.norm(vectors, axis=1, keepdims=True)
return vectors / norms
def retrieve(chunks, chunk_norms, query, k=3):
q_emb = embed_texts([query])
q_norm = normalize(q_emb)
scores = np.dot(chunk_norms, q_norm.T).flatten()
top_idx = np.argsort(scores)[-k:][::-1]
return [(chunks[i], float(scores[i])) for i in top_idx]
SYSTEM_PROMPT = """You are a precise research assistant. Answer the user's question using ONLY the retrieved context provided below.
Rules:
1. Cite the source passage number in square brackets, like [1] or [2].
2. If the context does not contain the answer, say "I don't have enough information to answer that."
3. Keep answers concise and directly supported by the text.
Retrieved context:
{context}
"""
def ask_agent(chunks, chunk_norms, question):
passages = retrieve(chunks, chunk_norms, question, k=3)
context_blocks = [f"[{idx}] {text}" for idx, (text, _) in enumerate(passages, 1)]
context = "\n\n".join(context_blocks)
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT.format(context=context)},
{"role": "user", "content": question},
],
temperature=0.1
)
return response.choices[0].message.content, passages
if __name__ == "__main__":
with open("handbook.txt", "r", encoding="utf-8") as f:
text = f.read()
chunks = chunk_text(text)
embeddings = embed_texts(chunks)
chunk_norms = normalize(embeddings)
answer, sources = ask_agent(chunks, chunk_norms, "What is the refund policy?")
print("Answer:", answer)
print("\nTop sources:")
for txt, score in sources:
print(f"- {txt[:100]}... (score: {score:.3f})")
Example output:
Answer: Full refunds are available within 30 days of purchase with a valid receipt [1]. After 30 days, store credit is issued at the manager's discretion [2].
Top sources:
- [1] Customers may return any item within 30 days for a full refund. A valid receipt is required for all returns... (score: 0.891)
- [2] Returns after 30 days are handled on a case-by-case basis and typically result in store credit... (score: 0.845)
- [3] All electronics must be unopened and accompanied by the original packaging to qualify for a refund... (score: 0.712)
Next steps
Replace the NumPy retriever with a persistent store such as Chroma or pgvector if you need to scale beyond a few thousand chunks. You can also switch the chat model to qwen-3-32b on Oxlo.ai if you want stronger multilingual reasoning, or to deepseek-v3.2 for a free-tier coding assistant that cites your internal docs.
Because Oxlo.ai uses flat per-request pricing, your embedding and chat costs stay constant regardless of how long your chunks are. For high-volume RAG pipelines with large context windows, that can cut costs significantly compared to token-based providers. See the details at https://oxlo.ai/pricing.
Top comments (0)