LangChain and LlamaIndex add bloat. They wrap HTTP requests, string formatting, and math in custom classes. Debugging framework harder than fixing prompt. Build without libraries for control.
RAG pipeline needs chunking, vector math, prompt formatting. Agent needs loop to parse commands and call functions. Standard Python libraries give control, avoid dependency conflicts, keep notebooks light.
Build RAG and ReAct agent using Python standard library.
The Zero-Dependency HTTP Client
LLM APIs use HTTP. Use built-in urllib.request and json. Works with OpenAI, DeepSeek, OpenRouter, Ollama. Client code:
import json
import urllib.request
class LLMClient:
def __init__(self, api_key, base_url="https://api.openai.com/v1", model="gpt-4o-mini"):
self.api_key = api_key
self.base_url = base_url
self.model = model
def request(self, endpoint, payload):
url = f"{self.base_url}/{endpoint}"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
}
req = urllib.request.Request(
url,
data=json.dumps(payload).encode("utf-8"),
headers=headers,
method="POST"
)
try:
with urllib.request.urlopen(req) as response:
return json.loads(response.read().decode("utf-8"))
except Exception as e:
print(f"API Request failed: {e}")
raise
def chat(self, messages, temperature=0.0):
payload = {
"model": self.model,
"messages": messages,
"temperature": temperature
}
response = self.request("chat/completions", payload)
return response["choices"][0]["message"]["content"]
def embed(self, text, model="text-embedding-3-small"):
payload = {
"model": model,
"input": text
}
response = self.request("embeddings", payload)
return response["data"][0]["embedding"]Class replaces OpenAI package. Handles serialization, headers, parsing. Drop into Colab.
Vector Search in Pure Python
Vector databases overkill for small datasets. While production setups require choosing a vector database for RAG, you can run search in memory for smaller projects. Use cosine similarity to find relevant chunks. Code:
def dot_product(v1, v2):
return sum(x * y for x, y in zip(v1, v2))
def magnitude(v):
return sum(x * x for x in v) ** 0.5
def cosine_similarity(v1, v2):
mag_v1 = magnitude(v1)
mag_v2 = magnitude(v2)
if not mag_v1 or not mag_v2:
return 0.0
return dot_product(v1, v2) / (mag_v1 * mag_v2)Math fast for hundreds of chunks. Scale to thousands with numpy matrix operations. Avoids database.
Implementing the RAG Pipeline
RAG pipeline steps: chunk text, embed chunks, retrieve matches. To ensure accuracy, run through a RAG evaluation checklist before shipping. Chunking function splits by paragraph or word limit:
def chunk_text(text, max_words=150):
paragraphs = text.split("\n\n")
chunks = []
current_chunk = []
current_word_count = 0
for para in paragraphs:
para = para.strip()
if not para:
continue
words = para.split()
if current_word_count + len(words) > max_words:
chunks.append(" ".join(current_chunk))
current_chunk = words
current_word_count = len(words)
else:
current_chunk.extend(words)
current_word_count += len(words)
if current_chunk:
chunks.append(" ".join(current_chunk))
return chunksIndex stores text and embeddings. Search returns top matches:
class SimpleIndex:
def __init__(self, client):
self.client = client
self.documents = []
def add_text(self, text):
chunks = chunk_text(text)
for chunk in chunks:
embedding = self.client.embed(chunk)
self.documents.append({
"text": chunk,
"embedding": embedding
})
def search(self, query, top_k=2):
query_embedding = self.client.embed(query)
results = []
for doc in self.documents:
similarity = cosine_similarity(query_embedding, doc["embedding"])
results.append((similarity, doc["text"]))
# Sort by similarity score in descending order
results.sort(key=lambda x: x[0], reverse=True)
return results[:top_k]Sample data for index:
knowledge_base = """
Project Aether is our internal deployment platform built on Kubernetes.
It uses a custom controller called AetherEngine to manage resource scaling.
AetherEngine monitors CPU load and memory pressure every 5 seconds.
Deployments are defined using YAML files placed in the /deploy directory of a repository.
The platform automatically detects changes and runs a rolling update.
If a deployment fails, AetherEngine rolls back to the last stable state within 30 seconds.
Database connections are managed via a built-in proxy named AetherProxy.
AetherProxy handles connection pooling and rotates credentials automatically every 24 hours.
Developers do not need to store database passwords in their application code.
"""
# Initialize client and index
api_key = "your-api-key"
client = LLMClient(api_key=api_key)
index = SimpleIndex(client)
# Build index
index.add_text(knowledge_base)Query index, inject context into prompt, call LLM:
def ask_rag(query, index, client):
# Retrieve context
matches = index.search(query, top_k=2)
context = "\n-\n".join([text for score, text in matches])
# Construct prompt
system_prompt = (
"You are a helpful technical assistant. Answer the user's question "
"using only the provided context. If the answer cannot be found in "
"the context, say 'I do not know'."
)
user_prompt = f"Context:\n{context}\n\nQuestion: {query}\nAnswer:"
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
]
return client.chat(messages)
# Example query
response = ask_rag("How often does AetherProxy rotate credentials?", index, client)
print(response)Code retrieves context, formats prompt, returns answer. No hidden layers. Print variables to inspect.
Building a ReAct Agent from Scratch
Agents run tools to gather facts. ReAct loop: receive request, think, call tool, observe, repeat. System prompt enforces format:
AGENT_SYSTEM_PROMPT = """
You are an assistant that can use tools to answer questions.
You have access to the following tools:
{tools_description}
You must respond using the following format:
Thought: [Your reasoning about what to do next]
Action: [tool_name] [tool_input]
Observation: [The result of the tool execution]
If you have enough information to answer the user, write:
Thought: I have the final answer.
Final Answer: [Your final response to the user]
Do not make up observations. Only write 'Thought' and 'Action'. Wait for the observation from the environment.
"""Define calculator and RAG search tools:
def calculate(expression):
# Basic math evaluator
try:
# Clean expression
allowed_chars = "0123456789+-*/(). "
clean_expr = "".join([c for c in expression if c in allowed_chars])
return str(eval(clean_expr))
except Exception as e:
return f"Error evaluating expression: {e}"
def search_docs(query):
# Connects the agent to our RAG index
results = index.search(query, top_k=1)
if not results:
return "No relevant documentation found."
return results[0][1]
# Tool registry
tools = {
"calculate": {
"func": calculate,
"description": "Evaluates simple mathematical expressions. Input: mathematical expression string."
},
"search_docs": {
"func": search_docs,
"description": "Searches Project Aether documentation. Input: search query string."
}
}Parser extracts tool name and input from LLM response:
import re
def parse_action(text):
# Look for Action: tool_name tool_input
action_match = re.search(r"Action:\s*(\w+)\s+(.+)", text)
if action_match:
return action_match.group(1), action_match.group(2).strip()
return None, NoneExecution loop runs conversation, feeds tool output back, repeats until final answer. Without strict validation, agents risk falling into an LLM agent infinite loop:
def run_agent(user_query, client, max_steps=5):
# Format tools description
tools_desc = ""
for name, info in tools.items():
tools_desc += f"- {name}: {info['description']}\n"
system_message = AGENT_SYSTEM_PROMPT.format(tools_description=tools_desc)
messages = [
{"role": "system", "content": system_message},
{"role": "user", "content": user_query}
]
print(f"Query: {user_query}\n")
for step in range(max_steps):
response = client.chat(messages, temperature=0.0)
print(response)
print("-" * 40)
# Check for final answer
if "Final Answer:" in response:
break
# Parse action
tool_name, tool_input = parse_action(response)
if tool_name and tool_name in tools:
# Execute tool
observation = tools[tool_name]["func"](tool_input)
print(f"Observation: {observation}\n")
# Append response and observation to history
messages.append({"role": "assistant", "content": response})
messages.append({"role": "user", "content": f"Observation: {observation}"})
else:
# If the LLM failed to format the action correctly, guide it back
messages.append({"role": "assistant", "content": response})
messages.append({
"role": "user",
"content": "Invalid format. You must choose an Action or provide a Final Answer."
})Test agent with query requiring search and math:
query = "How many seconds does it take for AetherEngine to roll back a failed deployment, multiplied by 15?"
run_agent(query, client)Execution steps:
- Agent reads query, decides to search documentation.
- Agent calls
search_docswith query 'AetherEngine rollback time'. - RAG system returns text chunk stating rollback takes 30 seconds.
- Agent reads observation, decides to multiply 30 by 15, calls
calculatewith30 * 15. - Calculator returns
450. - Agent returns final answer: 'It takes 450 seconds.'
Running in Google Colab
To run in Google Colab, manage API keys securely. Do not hardcode keys. Use Colab Secret Manager:
- Click key icon (Secrets) in left sidebar.
- Add new secret named
OPENAI_API_KEY. - Enter API key in value field.
- Enable notebook access for key.
Retrieve key in code:
from google.colab import userdata
api_key = userdata.get('OPENAI_API_KEY')Replace api_key = "your-api-key" with snippet.
Why the Simple Approach Wins
Frameworks add complexity. Custom loops offer benefits:
- Debugging: Print exact API strings. No hidden prompts or templates.
- Speed: No heavy imports. Direct execution path.
- Understanding: Writing math and parsing logic shows how systems work. Easy to adjust thresholds.
Simple code is easier to maintain, modify, and scale.
tags
rag, ai-agents, python, llms, machine-learning
Related posts
- Building Minimalist Vector Search Engines from Scratch
- Designing ReAct Agent Loops for Production LLMs
- Optimizing Context Windows in Low-Resource Environments
- Standard Library Networking Patterns in Python AI Tooling
- Managing Prompt Drift in Autonomous Systems
- Debugging LLM Output Formats Without Framework Parsers
- Local LLM Integration: Using Ollama with Pure Python
- Effective Chunking Strategies for Technical Documentation RAG
- Implementing Safe Sandbox Environments for Python Code Agents
- Cosine Similarity vs Euclidean Distance in Small-Scale Embeddings
