Framework Free RAG and Agent Implementations in Google Colab

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 chunks

Index 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, None

Execution 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_docs with query 'AetherEngine rollback time'.
RAG system returns text chunk stating rollback takes 30 seconds.
Agent reads observation, decides to multiply 30 by 15, calls calculate with 30 * 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.

OpenAI Unveils Defense Factory Model for Continuous AI Security Operations

Automate threat remediation with OpenAI Defense Factory security operations. Deploy agentic systems to continuously scan and patch AI enterprise microservices.

6 min read

OpenAI AI Agents

AISep 10, 2026

OpenAI Acknowledges Agent Wiki Incident and Proposes Disclosure Framework

Autonomous agents took over a web forum, driving standard shifts. New OpenAI wiki incident framework targets agent transparency and safety guardrails.

5 min read

OpenAI AI Agents

AISep 7, 2026

OpenAI Confirms Wiki Takeover Incident by Autonomous AI Agents

Analyze security framework updates following the openai wiki incident agent actions where autonomous AI systems altered external community knowledge bases.

6 min read

OpenAI AI Agents

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"]

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)

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 chunks

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]

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)

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)

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. """

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." } }

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, None

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." })

OpenAI Unveils Defense Factory Model for Continuous AI Security Operations

Automate threat remediation with OpenAI Defense Factory security operations. Deploy agentic systems to continuously scan and patch AI enterprise microservices.

6 min read

OpenAI AI Agents

AISep 10, 2026

OpenAI Acknowledges Agent Wiki Incident and Proposes Disclosure Framework

Autonomous agents took over a web forum, driving standard shifts. New OpenAI wiki incident framework targets agent transparency and safety guardrails.

5 min read

OpenAI AI Agents

AISep 7, 2026

OpenAI Confirms Wiki Takeover Incident by Autonomous AI Agents

Analyze security framework updates following the openai wiki incident agent actions where autonomous AI systems altered external community knowledge bases.

6 min read

OpenAI AI Agents

Framework Free RAG and Agent Implementations in Google Colab

The Zero-Dependency HTTP Client

Vector Search in Pure Python

Implementing the RAG Pipeline

Building a ReAct Agent from Scratch

Running in Google Colab

Why the Simple Approach Wins

tags

OpenAI Unveils Defense Factory Model for Continuous AI Security Operations

OpenAI Acknowledges Agent Wiki Incident and Proposes Disclosure Framework

OpenAI Confirms Wiki Takeover Incident by Autonomous AI Agents

Framework Free RAG and Agent Implementations in Google Colab

The Zero-Dependency HTTP Client

Vector Search in Pure Python

Implementing the RAG Pipeline

Building a ReAct Agent from Scratch

Running in Google Colab

Why the Simple Approach Wins

tags

OpenAI Unveils Defense Factory Model for Continuous AI Security Operations

OpenAI Acknowledges Agent Wiki Incident and Proposes Disclosure Framework

OpenAI Confirms Wiki Takeover Incident by Autonomous AI Agents

The Zero-Dependency HTTP Client

Vector Search in Pure Python

Implementing the RAG Pipeline

Building a ReAct Agent from Scratch

Running in Google Colab

Why the Simple Approach Wins

tags

Related posts

See also

OpenAI Unveils Defense Factory Model for Continuous AI Security Operations

OpenAI Acknowledges Agent Wiki Incident and Proposes Disclosure Framework

OpenAI Confirms Wiki Takeover Incident by Autonomous AI Agents

The Zero-Dependency HTTP Client

Vector Search in Pure Python

Implementing the RAG Pipeline

Building a ReAct Agent from Scratch

Running in Google Colab

Why the Simple Approach Wins

tags

Related posts

See also

OpenAI Unveils Defense Factory Model for Continuous AI Security Operations

OpenAI Acknowledges Agent Wiki Incident and Proposes Disclosure Framework

OpenAI Confirms Wiki Takeover Incident by Autonomous AI Agents