AI Agents Explained

import openai, json, os

client = openai.OpenAI(api_key=os.environ["OPENAI_API_KEY"])

# Define tools the agent can use
tools = [
    {"type": "function", "function": {
        "name": "search_web",
        "description": "Search the web for current information",
        "parameters": {"type": "object", "properties": {
            "query": {"type": "string", "description": "Search query"}
        }, "required": ["query"]}
    }},
    {"type": "function", "function": {
        "name": "calculate",
        "description": "Evaluate a math expression",
        "parameters": {"type": "object", "properties": {
            "expression": {"type": "string", "description": "Math expression"}
        }, "required": ["expression"]}
    }}
]

# Tool implementations
def search_web(query: str) -> str:
    # In production, call a real search API (SerpAPI, Tavily, etc.)
    return f"Search results for '{query}': [simulated results here]"

def calculate(expression: str) -> str:
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"Error: {e}"

TOOL_MAP = {"search_web": search_web, "calculate": calculate}

def run_agent(goal: str, max_steps: int = 10) -> str:
    messages = [
        {"role": "system", "content": "You are a research agent. Use tools to find information and answer questions. When done, reply with your final answer (no tool call)."},
        {"role": "user", "content": goal}
    ]

    for step in range(max_steps):
        response = client.chat.completions.create(
            model="gpt-4o-mini", messages=messages, tools=tools
        )
        msg = response.choices[0].message
        messages.append(msg)

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

        # Execute each tool call
        for call in msg.tool_calls:
            fn = TOOL_MAP[call.function.name]
            result = fn(**json.loads(call.function.arguments))
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": result
            })
            print(f"  Step {step+1}: {call.function.name}({call.function.arguments}) → {result[:100]}")

    return "Agent reached max steps without completing."

answer = run_agent("What is 15% of the average Fortune 500 company revenue?")

import json, os

MEMORY_FILE = "agent_memory.json"

class AgentMemory:
    def __init__(self):
        self.facts = self._load()

    def _load(self) -> list:
        if os.path.exists(MEMORY_FILE):
            with open(MEMORY_FILE) as f:
                return json.load(f)
        return []

    def save(self):
        with open(MEMORY_FILE, "w") as f:
            json.dump(self.facts, f, indent=2)

    def add(self, fact: str):
        self.facts.append(fact)
        self.save()

    def search(self, query: str) -> list:
        query_lower = query.lower()
        return [f for f in self.facts if query_lower in f.lower()]

    def context_string(self, query: str) -> str:
        relevant = self.search(query)
        if not relevant:
            return "No relevant memories found."
        return "Known facts:\n" + "\n".join(f"- {f}" for f in relevant[-10:])

memory = AgentMemory()
# memory.add("The client's budget is $50,000")
# memory.add("They prefer monthly billing")
# context = memory.context_string("budget")
# Inject context into the system prompt

class BudgetTracker:
    def __init__(self, max_tokens: int = 50000):
        self.max_tokens = max_tokens
        self.used = 0

    def track(self, usage):
        self.used += usage.total_tokens
        if self.used > self.max_tokens:
            raise Exception(f"Budget exceeded: {self.used}/{self.max_tokens} tokens")
        return self.used

budget = BudgetTracker(max_tokens=20000)
# After each API call:
# budget.track(response.usage)