Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -225,3 +225,6 @@ test_providers.py
# Internal analysis artifacts (not learning material)
analysis/
analysis_progress.md

# macOS resource fork / metadata files
._*
119 changes: 119 additions & 0 deletions agents/s01_agent_loop_ollama.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,119 @@
#!/usr/bin/env python3
"""
s01_agent_loop_ollama.py - The Agent Loop (Ollama version)

Same pattern as s01_agent_loop.py but uses Ollama via its
OpenAI-compatible API instead of Anthropic.

while stop_reason == "tool_calls":
response = LLM(messages, tools)
execute tools
append results

+----------+ +--------+ +---------+
| User | ---> | Ollama | ---> | Tool |
| prompt | | | | execute |
+----------+ +---+----+ +----+----+
^ |
| tool_result |
+---------------+
(loop continues)

Requirements:
pip install openai python-dotenv
ollama pull qwen2.5-coder:7b # or any tool-capable model
"""

import json
import os
import subprocess

from dotenv import load_dotenv
from openai import OpenAI

load_dotenv(override=True)

client = OpenAI(
base_url=os.getenv("OLLAMA_BASE_URL", "http://localhost:11434/v1"),
api_key=os.getenv("OLLAMA_API_KEY", "ollama"), # Ollama ignores the key
)
MODEL = os.getenv("OLLAMA_MODEL_ID", "qwen2.5-coder:7b")

SYSTEM = f"You are a coding agent at {os.getcwd()}. Use bash to solve tasks. Act, don't explain."

# OpenAI tool format (different from Anthropic's)
TOOLS = [{
"type": "function",
"function": {
"name": "bash",
"description": "Run a shell command.",
"parameters": {
"type": "object",
"properties": {"command": {"type": "string"}},
"required": ["command"],
},
},
}]


def run_bash(command: str) -> str:
dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
if any(d in command for d in dangerous):
return "Error: Dangerous command blocked"
try:
r = subprocess.run(command, shell=True, cwd=os.getcwd(),
capture_output=True, text=True, timeout=120)
out = (r.stdout + r.stderr).strip()
return out[:50000] if out else "(no output)"
except subprocess.TimeoutExpired:
return "Error: Timeout (120s)"


# -- The core pattern: same while loop, OpenAI message format --
def agent_loop(messages: list):
while True:
response = client.chat.completions.create(
model=MODEL,
messages=messages,
tools=TOOLS,
tool_choice="auto",
)
msg = response.choices[0].message

# Append assistant turn (convert to dict for mutability)
messages.append(msg.model_dump(exclude_unset=False))

# If no tool calls, we're done
if response.choices[0].finish_reason != "tool_calls" or not msg.tool_calls:
return

# Execute each tool call, collect results
for tool_call in msg.tool_calls:
args = json.loads(tool_call.function.arguments)
print(f"\033[33m$ {args['command']}\033[0m")
output = run_bash(args["command"])
print(output[:200])
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": output,
})


if __name__ == "__main__":
print(f"\033[90mUsing model: {MODEL} via {client.base_url}\033[0m\n")
history = [{"role": "system", "content": SYSTEM}]
while True:
try:
query = input("\033[36ms01-ollama >> \033[0m")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):
break
history.append({"role": "user", "content": query})
agent_loop(history)
last = history[-1]
content = last.get("content") or ""
if content:
print(content)
print()
158 changes: 158 additions & 0 deletions agents/s02_tool_use_ollama.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,158 @@
#!/usr/bin/env python3
"""
s02_tool_use_ollama.py - Tools (Ollama version)

Same dispatch-map pattern as s02_tool_use.py but uses Ollama
via its OpenAI-compatible API.

+----------+ +--------+ +------------------+
| User | ---> | Ollama | ---> | Tool Dispatch |
| prompt | | | | { |
+----------+ +---+----+ | bash: run_bash |
^ | read: run_read |
| | write: run_wr |
+----------+ edit: run_edit |
tool_result| } |
+------------------+

Key insight: "The loop didn't change at all. I just added tools."
"""

import json
import os
import subprocess
from pathlib import Path

from dotenv import load_dotenv
from openai import OpenAI

load_dotenv(override=True)

WORKDIR = Path.cwd()
client = OpenAI(
base_url=os.getenv("OLLAMA_BASE_URL", "http://localhost:11434/v1"),
api_key=os.getenv("OLLAMA_API_KEY", "ollama"),
)
MODEL = os.getenv("OLLAMA_MODEL_ID", "qwen2.5-coder:7b")

SYSTEM = f"You are a coding agent at {WORKDIR}. Use tools to solve tasks. Act, don't explain."


def safe_path(p: str) -> Path:
path = (WORKDIR / p).resolve()
if not path.is_relative_to(WORKDIR):
raise ValueError(f"Path escapes workspace: {p}")
return path


def run_bash(command: str) -> str:
dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
if any(d in command for d in dangerous):
return "Error: Dangerous command blocked"
try:
r = subprocess.run(command, shell=True, cwd=WORKDIR,
capture_output=True, text=True, timeout=120)
out = (r.stdout + r.stderr).strip()
return out[:50000] if out else "(no output)"
except subprocess.TimeoutExpired:
return "Error: Timeout (120s)"


def run_read(path: str, limit: int = None) -> str:
try:
lines = safe_path(path).read_text().splitlines()
if limit and limit < len(lines):
lines = lines[:limit] + [f"... ({len(lines) - limit} more lines)"]
return "\n".join(lines)[:50000]
except Exception as e:
return f"Error: {e}"


def run_write(path: str, content: str) -> str:
try:
fp = safe_path(path)
fp.parent.mkdir(parents=True, exist_ok=True)
fp.write_text(content)
return f"Wrote {len(content)} bytes to {path}"
except Exception as e:
return f"Error: {e}"


def run_edit(path: str, old_text: str, new_text: str) -> str:
try:
fp = safe_path(path)
content = fp.read_text()
if old_text not in content:
return f"Error: Text not found in {path}"
fp.write_text(content.replace(old_text, new_text, 1))
return f"Edited {path}"
except Exception as e:
return f"Error: {e}"


# -- The dispatch map: {tool_name: handler} --
TOOL_HANDLERS = {
"bash": lambda **kw: run_bash(kw["command"]),
"read_file": lambda **kw: run_read(kw["path"], kw.get("limit")),
"write_file": lambda **kw: run_write(kw["path"], kw["content"]),
"edit_file": lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]),
}

# OpenAI function format (vs Anthropic's input_schema format)
TOOLS = [
{"type": "function", "function": {
"name": "bash", "description": "Run a shell command.",
"parameters": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}}},
{"type": "function", "function": {
"name": "read_file", "description": "Read file contents.",
"parameters": {"type": "object", "properties": {"path": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["path"]}}},
{"type": "function", "function": {
"name": "write_file", "description": "Write content to file.",
"parameters": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}}},
{"type": "function", "function": {
"name": "edit_file", "description": "Replace exact text in file.",
"parameters": {"type": "object", "properties": {"path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}}, "required": ["path", "old_text", "new_text"]}}},
]


def agent_loop(messages: list):
while True:
response = client.chat.completions.create(
model=MODEL, messages=messages,
tools=TOOLS, tool_choice="auto",
)
msg = response.choices[0].message
messages.append(msg.model_dump(exclude_unset=False))

if response.choices[0].finish_reason != "tool_calls" or not msg.tool_calls:
return

for tool_call in msg.tool_calls:
args = json.loads(tool_call.function.arguments)
handler = TOOL_HANDLERS.get(tool_call.function.name)
output = handler(**args) if handler else f"Unknown tool: {tool_call.function.name}"
print(f"> {tool_call.function.name}: {str(output)[:200]}")
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(output),
})


if __name__ == "__main__":
print(f"\033[90mUsing model: {MODEL} via {client.base_url}\033[0m\n")
history = [{"role": "system", "content": SYSTEM}]
while True:
try:
query = input("\033[36ms02-ollama >> \033[0m")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):
break
history.append({"role": "user", "content": query})
agent_loop(history)
last = history[-1]
content = last.get("content") or ""
if content and last.get("role") == "assistant":
print(content)
print()
Loading