S120 · LLM Agents — Function Calling, Tools, Planning

{
  "type": "function",
  "function": {
    "name": "get_weather",
    "description": "Get the current weather for a city. Returns temperature in Celsius and a short condition string.",
    "parameters": {
      "type": "object",
      "properties": {
        "city": { "type": "string", "description": "City name, e.g. 'Tokyo' or 'San Francisco'" },
        "units": { "type": "string", "enum": ["celsius", "fahrenheit"], "default": "celsius" }
      },
      "required": ["city"]
    }
  }
}

"""mini_agent.py — a ~120-line LLM agent with function calling.
 
Requires: pip install openai
Set OPENAI_API_KEY. Model can be gpt-4o-mini for cheap smoke tests.
"""
from __future__ import annotations
import json
import os
from openai import OpenAI
 
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
MODEL = "gpt-4o-mini"
MAX_STEPS = 8
 
 
# ---------- 1. Tools — plain Python functions ----------
def get_weather(city: str, units: str = "celsius") -> dict:
    """Fake weather — swap for a real API in prod."""
    fake = {"tokyo": 22, "london": 12, "san francisco": 17}.get(city.lower())
    if fake is None:
        return {"error": f"unknown city {city!r}"}
    temp = fake if units == "celsius" else fake * 9 / 5 + 32
    return {"temp": temp, "units": units, "condition": "partly cloudy"}
 
 
def convert_temp(value: float, from_unit: str, to_unit: str) -> float:
    if from_unit == to_unit:
        return value
    if from_unit == "celsius" and to_unit == "fahrenheit":
        return value * 9 / 5 + 32
    if from_unit == "fahrenheit" and to_unit == "celsius":
        return (value - 32) * 5 / 9
    return float("nan")
 
 
def finish(answer: str) -> str:
    """Emit the final answer and stop."""
    return answer
 
 
TOOLS_TABLE = {
    "get_weather":  get_weather,
    "convert_temp": convert_temp,
    "finish":       finish,
}
 
# ---------- 2. JSON schemas the LLM sees ----------
TOOLS_SCHEMA = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the current weather for a city. Returns temperature and condition.",
            "parameters": {
                "type": "object",
                "properties": {
                    "city":  {"type": "string", "description": "City name, e.g. 'Tokyo'"},
                    "units": {"type": "string", "enum": ["celsius", "fahrenheit"], "default": "celsius"},
                },
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "convert_temp",
            "description": "Convert temperature between celsius and fahrenheit.",
            "parameters": {
                "type": "object",
                "properties": {
                    "value":     {"type": "number"},
                    "from_unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
                    "to_unit":   {"type": "string", "enum": ["celsius", "fahrenheit"]},
                },
                "required": ["value", "from_unit", "to_unit"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "finish",
            "description": "Return the final answer to the user and stop. ALWAYS call this when done.",
            "parameters": {
                "type": "object",
                "properties": {"answer": {"type": "string"}},
                "required": ["answer"],
            },
        },
    },
]
 
 
# ---------- 3. The controller loop ----------
def run_agent(user_msg: str) -> str:
    messages = [
        {"role": "system",
         "content": ("You are a concise assistant with tools. "
                     "Reason briefly, call tools when useful, and ALWAYS call "
                     "finish(answer=...) with your final answer. Do not narrate.")},
        {"role": "user", "content": user_msg},
    ]
    for step in range(MAX_STEPS):
        resp = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=TOOLS_SCHEMA,
            tool_choice="auto",
            temperature=0,
        )
        msg = resp.choices[0].message
        messages.append(msg)                # keep assistant turn in history
 
        if not msg.tool_calls:
            print(f"[step {step}] LLM stopped without finish; returning content")
            return msg.content or "(empty)"
 
        for tc in msg.tool_calls:
            name = tc.function.name
            try:
                args = json.loads(tc.function.arguments)
            except json.JSONDecodeError as e:
                result = {"error": f"invalid JSON arguments: {e}"}
            else:
                fn = TOOLS_TABLE.get(name)
                if fn is None:
                    result = {"error": f"unknown tool {name!r}"}
                else:
                    try:
                        result = fn(**args)
                    except TypeError as e:
                        result = {"error": f"bad args: {e}"}
 
            print(f"[step {step}] {name}({args}) -> {result}")
 
            if name == "finish":
                return args["answer"]
 
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": json.dumps(result),
            })
    return "(hit max_steps without finish)"
 
 
if __name__ == "__main__":
    for q in [
        "What's the weather in Tokyo, and what is that in Fahrenheit?",
        "Is it warmer in London or San Francisco right now (report in F)?",
        "What's the weather on Mars?",       # should gracefully say unknown
    ]:
        print(f"\n=== {q}")
        print(">>", run_agent(q))