LLM Function Calling Explained — The $47k Mistake We Made with Parallel Tool Calls

import json
from openai import OpenAI

client = OpenAI()

# Define a minimal tool schema
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"}
                },
                "required": ["location"]
            }
        }
    }
]

# Send a request and inspect the raw response
response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "What's the weather in Tokyo?"}],
    tools=tools,
    tool_choice="auto"
)

# The raw response object contains the tool call
message = response.choices[0].message
print(f"Finish reason: {response.choices[0].finish_reason}")
print(f"Tool calls: {message.tool_calls}")

# Inspect the raw JSON of the arguments — this is what the model actually output
if message.tool_calls:
    for tc in message.tool_calls:
        print(f"Function: {tc.function.name}")
        print(f"Raw arguments JSON: {tc.function.arguments}")
        # The API parses this into a dict, but the raw string is what the model generated
        parsed = json.loads(tc.function.arguments)
        print(f"Parsed arguments: {parsed}")

import json
import logging
from openai import OpenAI
from typing import List, Dict, Any

logger = logging.getLogger(__name__)

class FunctionCallingAgent:
    def __init__(self, model: str = "gpt-4", max_turns: int = 5):
        self.client = OpenAI()
        self.model = model
        self.max_turns = max_turns
        self.tools = self._define_tools()

    def _define_tools(self) -> List[Dict]:
        return [
            {
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "description": "Get current weather for a city. Format: city name only, no state/country.",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "location": {"type": "string", "description": "City name (e.g., Tokyo, London)"},
                            "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                        },
                        "required": ["location"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "search_web",
                    "description": "Search the web for current information. Use for news, prices, or anything time-sensitive.",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "description": "Search query"}
                        },
                        "required": ["query"]
                    }
                }
            }
        ]

    def _execute_tool(self, tool_call: Any) -> Dict:
        """Execute a single tool call and return the result with the matching ID."""
        try:
            func_name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)
            
            if func_name == "get_weather":
                # Simulate API call — in production, call your weather API
                result = {"temperature": 22, "unit": args.get("unit", "celsius")}
            elif func_name == "search_web":
                result = {"results": [f"Simulated result for {args['query']}"]}
            else:
                raise ValueError(f"Unknown function: {func_name}")
            
            return {
                "role": "tool",
                "tool_call_id": tool_call.id,  # Critical: must match the original call ID
                "content": json.dumps(result)
            }
        except Exception as e:
            logger.error(f"Tool execution failed: {e}")
            return {
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps({"error": str(e)})
            }

    def run(self, user_message: str) -> str:
        messages = [{"role": "user", "content": user_message}]
        
        for turn in range(self.max_turns):
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=self.tools,
                tool_choice="auto"
            )
            
            message = response.choices[0].message
            messages.append(message)
            
            # Check if the model wants to call tools
            if message.tool_calls:
                # Execute all tool calls in parallel (or sequentially — your call)
                tool_results = [self._execute_tool(tc) for tc in message.tool_calls]
                messages.extend(tool_results)
                continue
            
            # No tool calls — this is the final response
            return message.content
        
        return "Max turns reached without final response."

# Usage
agent = FunctionCallingAgent()
print(agent.run("What's the weather in London and search for latest AI news?"))

# BAD: Using function calling for simple extraction
# This costs $0.02 per call and adds 500ms latency
response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Extract the date from: 'Meeting on 2024-01-15'"}],
    tools=[{
        "type": "function",
        "function": {
            "name": "extract_date",
            "parameters": {
                "type": "object",
                "properties": {"date": {"type": "string"}},
                "required": ["date"]
            }
        }
    }]
)

# GOOD: Use a regex or a dedicated parser
import re
match = re.search(r'\d{4}-\d{2}-\d{2}', "Meeting on 2024-01-15")
date = match.group(0) if match else None
print(date)  # 2024-01-15 — free and instant

import time
from collections import deque

class TokenBucket:
    """Rate limiter per tool."""
    def __init__(self, rate: float, capacity: int):
        self.rate = rate  # tokens per second
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill = time.time()

    def consume(self, tokens: int = 1) -> bool:
        now = time.time()
        self.tokens = min(self.capacity, self.tokens + (now - self.last_refill) * self.rate)
        self.last_refill = now
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False

class ContextWindow:
    """Sliding window for conversation history."""
    def __init__(self, max_messages: int = 20):
        self.messages = deque(maxlen=max_messages)

    def add(self, message: dict):
        self.messages.append(message)

    def get_messages(self, system_prompt: str, tools: list) -> list:
        # Always include system prompt and tools at the start
        return [{"role": "system", "content": system_prompt}] + list(self.messages)

# Usage
rate_limiter = TokenBucket(rate=10, capacity=20)  # 10 calls/sec, burst up to 20
if not rate_limiter.consume():
    return {"error": "Rate limit exceeded. Try again in 5 seconds."}

# Mistake 1: Not validating tool output
# BAD: Returning raw error message
tool_output = {"error": "Database connection failed: timeout"}
# The model might then call a 'fix_database' function that doesn't exist

# GOOD: Return a sanitized error
if "timeout" in str(error):
    tool_output = {"error": "Service temporarily unavailable. Please try again later."}

# Mistake 2: Using tool_choice='required' for a greeting
# BAD: Forces tool call even for 'hello'
response = client.chat.completions.create(
    ...,
    tool_choice="required",  # Don't do this unless you have a reason
)

# GOOD: Let the model decide
response = client.chat.completions.create(
    ...,
    tool_choice="auto"
)

# Mistake 3: Assuming tool call order
# If the model calls get_user_id and then get_orders, you must execute them in order
# The model might return them in any order in the list
# Always check dependencies before executing

# OpenAI style (native)
response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Weather in Paris?"}],
    tools=[weather_tool]
)
tool_call = response.choices[0].message.tool_calls[0]

# Anthropic style (XML-like)
# The model outputs: <function_calls><invoke name="get_weather">...
# You must parse this yourself
import re
pattern = r'<invoke name="(\w+)">(.*?)</invoke>'
matches = re.findall(pattern, response.content, re.DOTALL)

# Open-source style (varies by model)
# Llama 3 might output:
# {
#   "function": "get_weather",
#   "arguments": {"location": "Paris"}
# }
# Or inside a markdown code block
# You need a robust parser that handles multiple formats

import logging
import time
from openai import OpenAI

logger = logging.getLogger(__name__)

class MonitoredAgent:
    def __init__(self):
        self.client = OpenAI()
    
    def call_with_logging(self, messages, tools):
        start = time.time()
        
        # Log request
        logger.info(f"Request: {len(messages)} messages, {len(tools)} tools")
        
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=messages,
            tools=tools
        )
        
        elapsed = time.time() - start
        
        # Log response
        choice = response.choices[0]
        logger.info(f"Response time: {elapsed:.2f}s, finish_reason: {choice.finish_reason}")
        
        if choice.finish_reason == "length":
            logger.warning("Context window exceeded! Consider pruning history.")
        
        if choice.message.tool_calls:
            logger.info(f"Tool calls: {len(choice.message.tool_calls)}")
            for tc in choice.message.tool_calls:
                logger.info(f"  - {tc.function.name}({tc.function.arguments})")
        
        return response