LLM-Powered AI Agents - Tool Use and Planning

from openai import OpenAI
import json, requests, subprocess

client = OpenAI()

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# TOOLS -- functions the agent can call
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

def web_search(query: str) -> str:
    '''Search the web using DuckDuckGo API.'''
    try:
        r = requests.get("https://api.duckduckgo.com/", params={"q": query, "format": "json"}, timeout=5)
        data = r.json()
        results = data.get("RelatedTopics", [])[:3]
        return json.dumps([{"title": r.get("Text", ""), "url": r.get("FirstURL", "")} for r in results if isinstance(r, dict)])
    except Exception as e:
        return f"Search error: {e}"

def execute_python(code: str) -> str:
    '''Execute Python code safely and return output.'''
    try:
        result = subprocess.run(
            ["python", "-c", code],
            capture_output=True, text=True, timeout=10
        )
        output = result.stdout or result.stderr
        return output[:1000]  # truncate long outputs
    except subprocess.TimeoutExpired:
        return "Timeout: code took > 10 seconds"
    except Exception as e:
        return f"Error: {e}"

def read_file(path: str) -> str:
    '''Read contents of a file.'''
    try:
        with open(path, "r") as f:
            return f.read()[:2000]
    except Exception as e:
        return f"Error reading file: {e}"

def write_file(path: str, content: str) -> str:
    '''Write content to a file.'''
    try:
        with open(path, "w") as f:
            f.write(content)
        return f"File written: {path}"
    except Exception as e:
        return f"Error writing file: {e}"

TOOL_REGISTRY = {
    "web_search": web_search,
    "execute_python": execute_python,
    "read_file": read_file,
    "write_file": write_file,
}

TOOLS_SCHEMA = [
    {"type": "function", "function": {"name": "web_search", "description": "Search the web for current information", "parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}}},
    {"type": "function", "function": {"name": "execute_python", "description": "Execute Python code and return output", "parameters": {"type": "object", "properties": {"code": {"type": "string"}}, "required": ["code"]}}},
    {"type": "function", "function": {"name": "read_file", "description": "Read a file from disk", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}}},
    {"type": "function", "function": {"name": "write_file", "description": "Write content to a file", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}}},
]

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# REACT AGENT LOOP
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
class ReActAgent:
    '''ReAct: Reason + Act. Interleaves thinking and tool use.'''

    def __init__(self, model: str = "gpt-4o", max_steps: int = 10):
        self.model = model
        self.max_steps = max_steps
        self.messages = []

    def run(self, task: str) -> str:
        self.messages = [
            {"role": "system", "content": '''You are a capable AI agent with access to tools.
Think step by step. Use tools to gather information, do calculations, or take actions.
When you have completed the task and have a final answer, respond directly without calling tools.'''},
            {"role": "user", "content": task}
        ]

        for step in range(self.max_steps):
            response = client.chat.completions.create(
                model=self.model,
                messages=self.messages,
                tools=TOOLS_SCHEMA,
                tool_choice="auto",
            )

            msg = response.choices[0].message
            self.messages.append(msg)

            # No tool calls = agent is done
            if not msg.tool_calls:
                return msg.content

            # Execute all tool calls
            for tool_call in msg.tool_calls:
                fn_name = tool_call.function.name
                fn_args = json.loads(tool_call.function.arguments)

                print(f"  [Step {step+1}] Calling {fn_name}({json.dumps(fn_args)[:100]})")
                fn = TOOL_REGISTRY.get(fn_name)
                result = fn(**fn_args) if fn else f"Unknown tool: {fn_name}"

                self.messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": str(result)[:2000],
                })

        return "Max steps reached"

# Example: multi-step agent task
agent = ReActAgent()
result = agent.run('''
1. Calculate the compound interest on $10,000 at 7% annually for 30 years
2. Search for the current S&P 500 average annual return
3. Write a Python script that compares these two investment scenarios
4. Save the script to investment_comparison.py
''')
print(result)