ReAct agents

Question: How many hours do I have to request a refund on an annual plan?

Thought:     I need the refund window. I should search the policy.
Action:      search("annual plan refund window")
Observation: "Annual plans can be refunded within 30 days of purchase."

Thought:     30 days. I need to convert days to hours.
Action:      calculator("30 * 24")
Observation: 720

Thought:     I have the answer.
Answer:      You have 720 hours (30 days) from purchase.

import json
import re

KNOWLEDGE = {
    "refund": "Annual plans can be refunded within 30 days of purchase. Monthly plans are not refunded.",
    "backup": "Backups run every night at 2 AM and are kept for 90 days.",
}


def search(query: str) -> str:
    hits = [text for key, text in KNOWLEDGE.items() if key in query.lower()]
    return hits[0] if hits else "No results."


def calculator(expression: str) -> str:
    if not re.fullmatch(r"[\d\s+\-*/().]+", expression):
        return "Error: only numbers and + - * / ( ) are allowed."
    return str(eval(expression, {"__builtins__": {}}))    # safe only because of the check above


TOOLS = {"search": search, "calculator": calculator}
print(search("annual plan refund window"))
print(calculator("30 * 24"))
print(calculator("__import__('os').remove('x')"))

def fake_model(messages: list[dict]) -> dict:
    observations = [m["content"] for m in messages if m["role"] == "tool"]
    if not observations:
        return {"thought": "I need the refund window; search the policy.",
                "tool": "search", "args": {"query": "annual plan refund window"}}
    days = re.search(r"within (\d+) days", " ".join(observations))
    if days and len(observations) == 1:
        return {"thought": f"The window is {days.group(1)} days; convert to hours.",
                "tool": "calculator", "args": {"expression": f"{days.group(1)} * 24"}}
    return {"thought": "I have everything I need.",
            "answer": f"You have {observations[-1]} hours ({days.group(1)} days) from purchase."}

def run_agent(question: str, model=fake_model, max_steps: int = 5) -> str:
    messages = [{"role": "user", "content": question}]
    for step in range(1, max_steps + 1):
        decision = model(messages)
        print(f"[{step}] thought: {decision['thought']}")
        if "answer" in decision:
            return decision["answer"]
        name, args = decision["tool"], decision["args"]
        if name not in TOOLS:
            observation = f"Error: unknown tool {name}"
        else:
            observation = TOOLS[name](**args)
        print(f"    action: {name}({json.dumps(args)})")
        print(f"    observation: {observation}")
        messages.append({"role": "assistant", "content": json.dumps(decision)})
        messages.append({"role": "tool", "content": observation})
    return "Stopped: step limit reached without an answer."


print("answer:", run_agent("How many hours do I have to request a refund on an annual plan?"))

SCHEMAS = [
    {"type": "function", "function": {
        "name": "search", "description": "Search the product policy documents.",
        "parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}}},
    {"type": "function", "function": {
        "name": "calculator", "description": "Evaluate an arithmetic expression like '30 * 24'.",
        "parameters": {"type": "object", "properties": {"expression": {"type": "string"}}, "required": ["expression"]}}},
]
SYSTEM = ("You answer questions about our product. Use search for facts and calculator for arithmetic. "
          "Never guess numbers. When you have the answer, reply without calling tools.")


def run_live_agent(question: str, max_steps: int = 6) -> str:
    messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": question}]
    for _ in range(max_steps):
        response = client.chat.completions.create(model="gpt-4.1-mini", messages=messages, tools=SCHEMAS)
        message = response.choices[0].message
        if not message.tool_calls:
            return message.content                          # final answer
        messages.append(message)
        for call in message.tool_calls:
            args = json.loads(call.function.arguments)
            result = TOOLS[call.function.name](**args)
            messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
    return "Stopped: step limit reached."

def stubborn_model(messages: list[dict]) -> dict:
    """A badly behaved model that keeps searching for the same thing."""
    return {"thought": "Let me search again.", "tool": "search", "args": {"query": "pricing"}}


def run_guarded(question: str, model, max_steps: int = 6, max_repeats: int = 2) -> str:
    messages = [{"role": "user", "content": question}]
    seen: dict[str, int] = {}
    for step in range(1, max_steps + 1):
        decision = model(messages)
        if "answer" in decision:
            return decision["answer"]
        key = f"{decision['tool']}:{json.dumps(decision['args'], sort_keys=True)}"
        seen[key] = seen.get(key, 0) + 1
        if seen[key] > max_repeats:
            return f"Stopped at step {step}: repeated action {key} {seen[key]} times."
        observation = TOOLS[decision["tool"]](**decision["args"])
        messages.append({"role": "tool", "content": observation})
    return "Stopped: step limit reached."


print(run_guarded("How much does it cost?", stubborn_model))