"""A minimal agent loop. Standard library only. Python 3.9+. Run with the mock model (no API key needed): python agent.py "What is (12 + 8) * 3?" Run with a real model (any OpenAI-compatible chat API): export OPENAI_API_KEY=... # your key export LLM_MODEL=... # a model name your provider offers export LLM_BASE_URL=https://api.openai.com/v1 # optional python agent.py --real "What is (12 + 8) * 3?" """ import ast import json import operator import os import re import sys import urllib.request # ---------- 1. Tools: plain Python functions plus a short description ---------- _OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, ast.USub: operator.neg} def _eval(node): if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)): return node.value if isinstance(node, ast.BinOp) and type(node.op) in _OPS: return _OPS[type(node.op)](_eval(node.left), _eval(node.right)) if isinstance(node, ast.UnaryOp) and type(node.op) in _OPS: return _OPS[type(node.op)](_eval(node.operand)) raise ValueError("unsupported expression") def calculator(expression: str) -> str: """Safely evaluate basic arithmetic such as '(12 + 8) * 3'.""" tree = ast.parse(expression, mode="eval") return str(_eval(tree.body)) FACTS = { "react": "ReAct (2022) mixes reasoning steps and tool actions in one loop.", "compound ai system": "A system that uses several parts: model calls, retrievers, tools.", "workflow": "LLMs and tools are run along a path the developer wrote in code.", "agent": "The LLM decides its own next step and which tool to use.", } def lookup(term: str) -> str: """Look up a short definition in a tiny local knowledge base.""" key = term.strip().lower() if key in FACTS: return FACTS[key] raise KeyError(f"no entry for '{term}'. Known terms: {', '.join(FACTS)}") TOOLS = { "calculator": {"fn": calculator, "args": ["expression"], "doc": "Evaluate basic arithmetic."}, "lookup": {"fn": lookup, "args": ["term"], "doc": "Get a short definition of a term."}, } # ---------- 2. The model: any object with .complete(messages) -> str ---------- # The model must answer with JSON, in one of two shapes: # {"thought": "...", "action": "calculator", "input": {"expression": "2+2"}} # {"thought": "...", "final": "The answer is 4"} def system_prompt() -> str: lines = ["You are an agent. Think, then act. Reply with ONE JSON object only.", 'To use a tool: {"thought": "...", "action": "", "input": {...}}', 'To finish: {"thought": "...", "final": ""}', "Tools:"] for name, t in TOOLS.items(): lines.append(f"- {name}({', '.join(t['args'])}): {t['doc']}") return "\n".join(lines) class MockLLM: """A rule-based stand-in for a real model. It lets the loop run offline. It is NOT smart. It only shows the shape of the conversation.""" def complete(self, messages): last = messages[-1] if last["role"] == "user": question = last["content"] expr = re.search(r"[\d(][\d\s+\-*/().^]*[\d)]", question) if expr and any(c in expr.group(0) for c in "+-*/^"): return json.dumps({"thought": "This is arithmetic. I will use the calculator.", "action": "calculator", "input": {"expression": expr.group(0).replace("^", "**")}}) m = re.search(r"(?:what is|define|explain)\s+(?:a |an |the )?([a-z\- ]+?)\??$", question.strip().lower()) if m: return json.dumps({"thought": "I need a definition. I will look it up.", "action": "lookup", "input": {"term": m.group(1)}}) return json.dumps({"thought": "No tool fits.", "final": "I cannot help with that."}) # last message is a tool observation obs = last["content"] if obs.startswith("ERROR"): return json.dumps({"thought": "The tool failed. I will report it.", "final": f"Sorry, the tool failed. {obs}"}) return json.dumps({"thought": "I have what I need.", "final": obs}) class OpenAICompatibleLLM: """Calls any chat-completions API that follows the OpenAI format.""" def __init__(self): self.key = os.environ["OPENAI_API_KEY"] self.model = os.environ["LLM_MODEL"] self.base = os.environ.get("LLM_BASE_URL", "https://api.openai.com/v1") def complete(self, messages): body = json.dumps({"model": self.model, "messages": messages, "temperature": 0}).encode() req = urllib.request.Request( self.base + "/chat/completions", data=body, headers={"Content-Type": "application/json", "Authorization": "Bearer " + self.key}) with urllib.request.urlopen(req, timeout=60) as r: return json.load(r)["choices"][0]["message"]["content"] # ---------- 3. The loop ---------- def run_agent(question, llm, max_steps=5, verbose=True): messages = [{"role": "system", "content": system_prompt()}, {"role": "user", "content": question}] for step in range(1, max_steps + 1): raw = llm.complete(messages) messages.append({"role": "assistant", "content": raw}) try: msg = json.loads(raw) except json.JSONDecodeError: obs = "ERROR: reply was not valid JSON. Reply with one JSON object." messages.append({"role": "tool", "content": obs}) if verbose: print(f"[{step}] bad JSON -> asking again") continue if verbose: print(f"[{step}] thought: {msg.get('thought', '')}") if "final" in msg: if verbose: print(f"[{step}] final: {msg['final']}") return msg["final"] name, args = msg.get("action"), msg.get("input", {}) if name not in TOOLS: obs = f"ERROR: unknown tool '{name}'. Tools: {', '.join(TOOLS)}" else: try: obs = TOOLS[name]["fn"](**args) except Exception as e: # tool errors go back to the model obs = f"ERROR: {type(e).__name__}: {e}" if verbose: print(f"[{step}] action: {name}({args}) -> {obs}") messages.append({"role": "tool", "content": obs}) return "Stopped: step limit reached." if __name__ == "__main__": argv = sys.argv[1:] real = "--real" in argv argv = [a for a in argv if a != "--real"] q = " ".join(argv) or "What is (12 + 8) * 3?" run_agent(q, OpenAICompatibleLLM() if real else MockLLM())