From e38de6fa6140b4774d79cbb28383828744091051 Mon Sep 17 00:00:00 2001 From: mostafaibrahim17 Date: Mon, 7 Sep 2026 19:33:41 +0300 Subject: [PATCH] added the agentic loop article code --- agentic-loop-python/README.md | 3 + agentic-loop-python/agent.py | 157 ++++++++++++++++++++++ agentic-loop-python/requirements.txt | 3 + agentic-loop-python/search.py | 28 ++++ agentic-loop-python/test_search.py | 35 +++++ agentic-loop-python/tools.py | 187 +++++++++++++++++++++++++++ 6 files changed, 413 insertions(+) create mode 100644 agentic-loop-python/README.md create mode 100644 agentic-loop-python/agent.py create mode 100644 agentic-loop-python/requirements.txt create mode 100644 agentic-loop-python/search.py create mode 100644 agentic-loop-python/test_search.py create mode 100644 agentic-loop-python/tools.py diff --git a/agentic-loop-python/README.md b/agentic-loop-python/README.md new file mode 100644 index 0000000000..e4b8c447dc --- /dev/null +++ b/agentic-loop-python/README.md @@ -0,0 +1,3 @@ +# Build an Agentic Loop from Scratch in Python + +This folder provides the code examples for the Real Python tutorial [Build an Agentic Loop from Scratch in Python](https://realpython.com/agentic-loop/). \ No newline at end of file diff --git a/agentic-loop-python/agent.py b/agentic-loop-python/agent.py new file mode 100644 index 0000000000..dc188cd6ad --- /dev/null +++ b/agentic-loop-python/agent.py @@ -0,0 +1,157 @@ +"""An agentic loop that fixes and optimizes Python code.""" + +import json + +from openai import OpenAI +from tools import ( + OPTIMIZATION_MENU, + TOOL_DEFINITIONS, + TOOL_REGISTRY, + read_code, +) + +client = OpenAI() +MODEL = "gpt-4o-mini" +MAX_ITERATIONS = 10 + + +def call_tools(tool_calls): + """Execute each tool call and return the results.""" + results = [] + for call in tool_calls: + name = call.function.name + if name not in TOOL_REGISTRY: + output = f"Error: Unknown tool '{name}'" + else: + try: + args = json.loads(call.function.arguments) + args.pop("reasoning", None) + output = TOOL_REGISTRY[name](**args) + except Exception as err: + output = f"Error: {err}" + results.append( + { + "role": "tool", + "tool_call_id": call.id, + "content": str(output), + } + ) + return results + + +def run_phase(system_prompt, goal, phase_name): + """Run one phase of the agentic loop.""" + print(f"\n{'=' * 50}") + print(f" {phase_name}") + print(f"{'=' * 50}\n") + + messages = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": goal}, + ] + + last_good = read_code("search.py") + + for step in range(1, MAX_ITERATIONS + 1): + print(f"[Step {step}]") + + response = client.chat.completions.create( + model=MODEL, + messages=messages, + tools=TOOL_DEFINITIONS, + ) + + message = response.choices[0].message + messages.append(message) + + if message.content: + print(f" Think: {message.content[:200]}") + + if not message.tool_calls: + print(f"\n{phase_name} complete.\n") + return message.content + + for call in message.tool_calls: + name = call.function.name + args = json.loads(call.function.arguments) + reasoning = args.pop("reasoning", None) + if reasoning: + print(f" Think: {reasoning}") + short = str(args)[:80] + print(f" Act: {name}({short})") + + results = call_tools(message.tool_calls) + messages.extend(results) + + for call in message.tool_calls: + if call.function.name == "write_code": + test_out = TOOL_REGISTRY["run_tests"]() + if "failed" in test_out.lower(): + TOOL_REGISTRY["write_code"]("search.py", last_good) + messages.append( + { + "role": "user", + "content": ( + "Your change broke a test." + " I reverted search.py. " + "Try a different approach." + ), + }, + ) + else: + last_good = read_code("search.py") + break + + for r in results: + preview = r["content"][:120] + print(f" Observe: {preview}") + + print(f"\n{phase_name}: iteration cap reached.\n") + return None + + +def main(): + """Run both phases of the agent.""" + source = read_code("search.py") + print(f"Target program loaded ({len(source)} chars)") + + # Phase 1: Fix the correctness bug + phase1_system = ( + "You are a Python debugging agent. Fix bugs " + "in search.py by reading the code, running " + "the test suite, and writing a corrected " + "version. Only modify search.py. Stop when " + "all tests pass." + ) + phase1_goal = ( + "The test suite in test_search.py has a " + "failing test. Start by running the test " + "suite to see which test fails and why. " + "Then find the bug in search.py and fix it." + ) + run_phase(phase1_system, phase1_goal, "Phase 1: Fix") + + # Phase 2: Optimize performance + phase2_system = ( + "You are a Python optimization agent. Speed " + "up search_similar() in search.py. Run the " + "benchmark first, then optimize using one of " + "the techniques below. After optimizing, run " + "both the tests and the benchmark to verify " + "correctness and improvement.\n\n" + f"{OPTIMIZATION_MENU}" + ) + phase2_goal = ( + "Optimize search_similar() in search.py for " + "speed. Measure before and after. Keep all " + "tests passing." + ) + run_phase( + phase2_system, + phase2_goal, + "Phase 2: Optimize", + ) + + +if __name__ == "__main__": + main() diff --git a/agentic-loop-python/requirements.txt b/agentic-loop-python/requirements.txt new file mode 100644 index 0000000000..39cd4dbef7 --- /dev/null +++ b/agentic-loop-python/requirements.txt @@ -0,0 +1,3 @@ +openai>=1.0 +pytest>=8.0 +numpy>=1.26 diff --git a/agentic-loop-python/search.py b/agentic-loop-python/search.py new file mode 100644 index 0000000000..8593cf8662 --- /dev/null +++ b/agentic-loop-python/search.py @@ -0,0 +1,28 @@ +import random + + +def cosine_similarity(vec_a, vec_b): + """Return cosine similarity between two vectors.""" + dot = sum(a * b for a, b in zip(vec_a, vec_b)) + mag_a = sum(a**2 for a in vec_a) ** 0.5 + mag_b = sum(b**2 for b in vec_b) ** 0.5 + return dot / (mag_a * mag_b) + + +def load_vectors(count=500, dimensions=128): + """Generate random vectors for the demo.""" + random.seed(42) + return [ + [random.uniform(-1, 1) for _ in range(dimensions)] + for _ in range(count) + ] + + +def search_similar(query, stored, top_k=5): + """Find the top_k most similar vectors to query.""" + scores = [] + for i, candidate in enumerate(stored): + score = cosine_similarity(query, candidate) + scores.append((i, score)) + scores.sort(key=lambda x: x[1], reverse=True) + return scores[:top_k] diff --git a/agentic-loop-python/test_search.py b/agentic-loop-python/test_search.py new file mode 100644 index 0000000000..46edee885e --- /dev/null +++ b/agentic-loop-python/test_search.py @@ -0,0 +1,35 @@ +"""Test suite for the vector similarity search module.""" + +import pytest + +from search import ( + cosine_similarity, + load_vectors, + search_similar, +) + + +def test_identical_vectors(): + vec = [1.0, 2.0, 3.0] + assert cosine_similarity(vec, vec) == pytest.approx(1.0) + + +def test_orthogonal_vectors(): + vec_a = [1.0, 0.0] + vec_b = [0.0, 1.0] + assert cosine_similarity(vec_a, vec_b) == pytest.approx(0.0) + + +def test_zero_vector(): + vec = [1.0, 2.0, 3.0] + zero = [0.0, 0.0, 0.0] + result = cosine_similarity(vec, zero) + assert result == 0.0 + + +def test_search_returns_top_k(): + stored = load_vectors(count=50, dimensions=8) + query = stored[0] + results = search_similar(query, stored, top_k=3) + assert len(results) == 3 + assert results[0][0] == 0 diff --git a/agentic-loop-python/tools.py b/agentic-loop-python/tools.py new file mode 100644 index 0000000000..c38f22bf34 --- /dev/null +++ b/agentic-loop-python/tools.py @@ -0,0 +1,187 @@ +"""Tools the agent can call during its loop.""" + +import subprocess + + +TOOL_DEFINITIONS = [ + { + "type": "function", + "function": { + "name": "run_tests", + "description": ( + "Run the pytest suite on test_search.py and return the output." + ), + "parameters": { + "type": "object", + "properties": { + "reasoning": { + "type": "string", + "description": ( + "One short sentence explaining " + "why you're running the tests " + "now." + ), + }, + }, + "required": ["reasoning"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "run_benchmark", + "description": ( + "Benchmark search_similar() and return the elapsed time." + ), + "parameters": { + "type": "object", + "properties": { + "reasoning": { + "type": "string", + "description": ( + "One short sentence explaining " + "why you're running the " + "benchmark now." + ), + }, + }, + "required": ["reasoning"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "read_code", + "description": ("Read and return a Python file's contents."), + "parameters": { + "type": "object", + "properties": { + "filename": { + "type": "string", + "description": "File to read.", + }, + "reasoning": { + "type": "string", + "description": ( + "One short sentence explaining " + "why you're reading this file " + "now." + ), + }, + }, + "required": ["filename", "reasoning"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "write_code", + "description": ("Overwrite a Python file with new code."), + "parameters": { + "type": "object", + "properties": { + "filename": { + "type": "string", + "description": ("File to write."), + }, + "content": { + "type": "string", + "description": ("The new file contents."), + }, + "reasoning": { + "type": "string", + "description": ( + "One short sentence explaining " + "what this change does and " + "why." + ), + }, + }, + "required": [ + "filename", + "content", + "reasoning", + ], + }, + }, + }, +] + +OPTIMIZATION_MENU = ( + "Available optimization techniques:\n" + "1. Set/dict swap: Replace list lookups " + "with set lookups\n" + "2. Built-in replacement: Use built-ins " + "instead of loops\n" + "3. functools.cache: Cache repeated " + "function calls\n" + "4. NumPy vectorization: Replace Python " + "loops with NumPy array operations" +) + + +def run_tests(): + """Run pytest and return the captured output.""" + result = subprocess.run( + [ + "python", + "-m", + "pytest", + "test_search.py", + "-v", + ], + capture_output=True, + text=True, + ) + return result.stdout + result.stderr + + +def run_benchmark(): + """Time search_similar() and return the result.""" + setup = ( + "from search import load_vectors, " + "search_similar; " + "stored = load_vectors(500, 128); " + "query = stored[0]" + ) + result = subprocess.run( + [ + "python", + "-m", + "timeit", + "-n", + "10", + "-r", + "3", + "-s", + setup, + "search_similar(query, stored)", + ], + capture_output=True, + text=True, + ) + return result.stdout.strip() + + +def read_code(filename): + """Read and return the contents of a file.""" + with open(filename) as f: + return f.read() + + +def write_code(filename, content): + """Write new contents to a file.""" + with open(filename, "w") as f: + f.write(content) + return f"Wrote {len(content)} characters to {filename}" + + +TOOL_REGISTRY = { + "run_tests": run_tests, + "run_benchmark": run_benchmark, + "read_code": read_code, + "write_code": write_code, +}