Pythonagentic_loop.py
Copy
1
import os
2
import json
3
import math
4
from typing import Any
5
import openai
6
 
7
client = openai.OpenAI(api_key=os.environ["OPENAI_API_KEY"])
8
 
9
# ── Tool definitions ────────────────────────────────────────────────
10
TOOLS = [
11
    {
12
        "type": "function",
13
        "function": {
14
            "name": "get_weather",
15
            "description": "Get current weather for a city",
16
            "parameters": {
17
                "type": "object",
18
                "properties": {
19
                    "city": {"type": "string"},
20
                    "units": {"type": "string", "enum": ["celsius", "fahrenheit"], "default": "celsius"},
21
                },
22
                "required": ["city"],
23
            },
24
        },
25
    },
26
    {
27
        "type": "function",
28
        "function": {
29
            "name": "calculate",
30
            "description": "Evaluate a mathematical expression",
31
            "parameters": {
32
                "type": "object",
33
                "properties": {
34
                    "expression": {"type": "string", "description": "Math expression, e.g. '2 + 2 * 3'"},
35
                },
36
                "required": ["expression"],
37
            },
38
        },
39
    },
40
    {
41
        "type": "function",
42
        "function": {
43
            "name": "search_products",
44
            "description": "Search for products by category and max price",
45
            "parameters": {
46
                "type": "object",
47
                "properties": {
48
                    "category": {"type": "string"},
49
                    "max_price": {"type": "number"},
50
                },
51
                "required": ["category"],
52
            },
53
        },
54
    },
55
]
56
 
57
# ── Tool implementations ────────────────────────────────────────────
58
def get_weather(city: str, units: str = "celsius") -> dict:
59
    """Simulated weather tool."""
60
    weather_db = {
61
        "tokyo": {"temp": 22, "condition": "Partly cloudy", "humidity": 65},
62
        "london": {"temp": 15, "condition": "Rainy", "humidity": 80},
63
        "new york": {"temp": 18, "condition": "Sunny", "humidity": 50},
64
    }
65
    data = weather_db.get(city.lower(), {"temp": 20, "condition": "Clear", "humidity": 55})
66
    temp = data["temp"] if units == "celsius" else data["temp"] * 9/5 + 32
67
    return {"city": city, "temperature": temp, "units": units, **data}
68
 
69
def calculate(expression: str) -> dict:
70
    """Safe math expression evaluator."""
71
    try:
72
        allowed = set("0123456789+-*/.() ")
73
        if not all(c in allowed for c in expression):
74
            return {"error": "Invalid characters in expression"}
75
        result = eval(expression, {"__builtins__": {}}, {"sqrt": math.sqrt, "pi": math.pi})
76
        return {"result": result, "expression": expression}
77
    except Exception as e:
78
        return {"error": str(e)}
79
 
80
def search_products(category: str, max_price: float = 1000.0) -> dict:
81
    """Simulated product search."""
82
    products = {
83
        "laptop": [{"name": "ThinkPad X1", "price": 899}, {"name": "MacBook Air M2", "price": 999}],
84
        "phone": [{"name": "Pixel 8", "price": 599}, {"name": "iPhone 15", "price": 799}],
85
    }
86
    results = [p for p in products.get(category.lower(), []) if p["price"] <= max_price]
87
    return {"category": category, "results": results[:3]}
88
 
89
TOOL_FUNCTIONS: dict[str, Any] = {
90
    "get_weather": get_weather,
91
    "calculate": calculate,
92
    "search_products": search_products,
93
}
94
 
95
# ── Agentic loop ────────────────────────────────────────────────────
96
def run_agent(user_message: str, max_iterations: int = 10) -> str:
97
    messages = [{"role": "user", "content": user_message}]
98
    
99
    for iteration in range(max_iterations):
100
        response = client.chat.completions.create(
101
            model="gpt-4o-mini",
102
            messages=messages,
103
            tools=TOOLS,
104
            tool_choice="auto",
105
        )
106
        
107
        choice = response.choices[0]
108
        messages.append(choice.message.to_dict() if hasattr(choice.message, 'to_dict') else {
109
            "role": "assistant",
110
            "content": choice.message.content,
111
            "tool_calls": [tc.model_dump() for tc in (choice.message.tool_calls or [])],
112
        })
113
        
114
        if choice.finish_reason == "stop":
115
            return choice.message.content
116
        
117
        if choice.finish_reason == "tool_calls":
118
            # Process all tool calls (may be parallel)
119
            for tc in choice.message.tool_calls:
120
                fn_name = tc.function.name
121
                fn_args = json.loads(tc.function.arguments)
122
                print(f"  [Tool] {fn_name}({fn_args})")
123
                
124
                if fn_name in TOOL_FUNCTIONS:
125
                    result = TOOL_FUNCTIONS[fn_name](**fn_args)
126
                else:
127
                    result = {"error": f"Unknown tool: {fn_name}"}
128
                
129
                messages.append({
130
                    "role": "tool",
131
                    "tool_call_id": tc.id,
132
                    "content": json.dumps(result),
133
                })
134
    
135
    return "Max iterations reached"
136
 
137
# Test
138
print(run_agent("What's the weather in Tokyo and London? Also, what's 144 * 3.14159?"))
139
print()
140
print(run_agent("Find me laptops under $950"))
141