Build an Agentic Loop From Scratch in Python

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/).

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()

agentic-loop-python/requirements.txt

@@ -0,0 +1,3 @@
+openai>=1.0
+pytest>=8.0
+numpy>=1.26

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]

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