Function Calling Deep Dive

from openai import OpenAI
from typing import Literal

client = OpenAI()

# Define function schemas
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a location. Use this when the user asks about weather conditions.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City and country, e.g., 'London, UK'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit"
                    }
                },
                "required": ["location"],
                "additionalProperties": False
            },
            "strict": True  # Enables structured outputs
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_products",
            "description": "Search for products in the catalog",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search query"
                    },
                    "category": {
                        "type": "string",
                        "enum": ["electronics", "clothing", "books", "home"]
                    },
                    "max_price": {
                        "type": "number",
                        "description": "Maximum price filter"
                    },
                    "in_stock": {
                        "type": "boolean",
                        "description": "Filter for in-stock items only"
                    }
                },
                "required": ["query"],
                "additionalProperties": False
            },
            "strict": True
        }
    }
]

from pydantic import BaseModel, Field
from typing import Optional, List
import json

class WeatherParams(BaseModel):
    """Parameters for weather lookup"""
    location: str = Field(description="City and country, e.g., 'London, UK'")
    unit: Literal["celsius", "fahrenheit"] = Field(
        default="celsius",
        description="Temperature unit"
    )

class ProductSearchParams(BaseModel):
    """Parameters for product search"""
    query: str = Field(description="Search query")
    category: Optional[str] = Field(
        default=None,
        description="Product category filter"
    )
    max_price: Optional[float] = Field(
        default=None,
        description="Maximum price"
    )
    limit: int = Field(
        default=10,
        ge=1,
        le=100,
        description="Number of results"
    )

def pydantic_to_openai_function(model: type[BaseModel], name: str = None) -> dict:
    """Convert Pydantic model to OpenAI function schema"""
    schema = model.model_json_schema()
    
    return {
        "type": "function",
        "function": {
            "name": name or model.__name__.lower(),
            "description": model.__doc__ or "",
            "parameters": {
                "type": "object",
                "properties": schema.get("properties", {}),
                "required": schema.get("required", []),
                "additionalProperties": False
            },
            "strict": True
        }
    }

# Generate tools from Pydantic models
tools = [
    pydantic_to_openai_function(WeatherParams, "get_weather"),
    pydantic_to_openai_function(ProductSearchParams, "search_products")
]

from dataclasses import dataclass
from typing import Callable, Any, Dict
import json
import inspect

@dataclass
class FunctionResult:
    name: str
    arguments: dict
    result: Any
    success: bool
    error: str = None
    execution_time_ms: float = 0

class FunctionRegistry:
    """Registry for callable functions"""
    
    def __init__(self):
        self.functions: Dict[str, Callable] = {}
        self.schemas: Dict[str, dict] = {}
    
    def register(
        self,
        name: str = None,
        description: str = None,
        param_model: type[BaseModel] = None
    ):
        """Decorator to register a function"""
        def decorator(func: Callable):
            func_name = name or func.__name__
            
            # Generate schema from type hints or param_model
            if param_model:
                schema = pydantic_to_openai_function(param_model, func_name)
            else:
                schema = self._generate_schema_from_hints(func, func_name, description)
            
            self.functions[func_name] = func
            self.schemas[func_name] = schema
            
            return func
        return decorator
    
    def _generate_schema_from_hints(
        self,
        func: Callable,
        name: str,
        description: str = None
    ) -> dict:
        """Generate schema from function type hints"""
        sig = inspect.signature(func)
        hints = func.__annotations__
        
        properties = {}
        required = []
        
        for param_name, param in sig.parameters.items():
            if param_name == "self":
                continue
            
            param_type = hints.get(param_name, str)
            
            # Map Python types to JSON Schema
            type_map = {
                str: "string",
                int: "integer",
                float: "number",
                bool: "boolean",
                list: "array",
                dict: "object"
            }
            
            properties[param_name] = {
                "type": type_map.get(param_type, "string")
            }
            
            if param.default == inspect.Parameter.empty:
                required.append(param_name)
        
        return {
            "type": "function",
            "function": {
                "name": name,
                "description": description or func.__doc__ or "",
                "parameters": {
                    "type": "object",
                    "properties": properties,
                    "required": required
                }
            }
        }
    
    def get_tools(self) -> list:
        """Get all registered tools for OpenAI"""
        return list(self.schemas.values())
    
    async def execute(
        self,
        name: str,
        arguments: dict
    ) -> FunctionResult:
        """Execute a registered function"""
        import time
        
        if name not in self.functions:
            return FunctionResult(
                name=name,
                arguments=arguments,
                result=None,
                success=False,
                error=f"Function '{name}' not found"
            )
        
        start = time.time()
        func = self.functions[name]
        
        try:
            # Check if async
            if inspect.iscoroutinefunction(func):
                result = await func(**arguments)
            else:
                result = func(**arguments)
            
            return FunctionResult(
                name=name,
                arguments=arguments,
                result=result,
                success=True,
                execution_time_ms=(time.time() - start) * 1000
            )
        except Exception as e:
            return FunctionResult(
                name=name,
                arguments=arguments,
                result=None,
                success=False,
                error=str(e),
                execution_time_ms=(time.time() - start) * 1000
            )

# Usage
registry = FunctionRegistry()

@registry.register(description="Get current weather for a location")
async def get_weather(location: str, unit: str = "celsius") -> dict:
    # Actual implementation
    return {
        "location": location,
        "temperature": 22,
        "unit": unit,
        "conditions": "sunny"
    }

@registry.register(description="Search for products")
async def search_products(
    query: str,
    category: str = None,
    max_price: float = None
) -> list:
    # Actual implementation
    return [
        {"name": "Product 1", "price": 29.99},
        {"name": "Product 2", "price": 49.99}
    ]

import asyncio
from typing import List

class FunctionExecutor:
    """Execute function calls from LLM responses"""
    
    def __init__(self, registry: FunctionRegistry):
        self.registry = registry
    
    async def execute_tool_calls(
        self,
        tool_calls: list,
        parallel: bool = True
    ) -> List[FunctionResult]:
        """Execute multiple tool calls"""
        
        if parallel:
            # Execute all calls in parallel
            tasks = [
                self.registry.execute(
                    tc.function.name,
                    json.loads(tc.function.arguments)
                )
                for tc in tool_calls
            ]
            return await asyncio.gather(*tasks)
        else:
            # Execute sequentially
            results = []
            for tc in tool_calls:
                result = await self.registry.execute(
                    tc.function.name,
                    json.loads(tc.function.arguments)
                )
                results.append(result)
            return results
    
    def format_results_for_llm(
        self,
        tool_calls: list,
        results: List[FunctionResult]
    ) -> list:
        """Format results as tool messages for LLM"""
        messages = []
        
        for tc, result in zip(tool_calls, results):
            if result.success:
                content = json.dumps(result.result)
            else:
                content = json.dumps({"error": result.error})
            
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": content
            })
        
        return messages

# Complete conversation loop
async def function_calling_loop(
    client: OpenAI,
    messages: list,
    registry: FunctionRegistry,
    max_iterations: int = 10
) -> str:
    """Run function calling conversation loop"""
    
    executor = FunctionExecutor(registry)
    
    for _ in range(max_iterations):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=registry.get_tools(),
            tool_choice="auto"
        )
        
        message = response.choices[0].message
        messages.append(message)
        
        # Check if done
        if not message.tool_calls:
            return message.content
        
        # Execute tool calls
        results = await executor.execute_tool_calls(message.tool_calls)
        
        # Add results to conversation
        tool_messages = executor.format_results_for_llm(
            message.tool_calls,
            results
        )
        messages.extend(tool_messages)
    
    raise Exception("Max iterations exceeded")

from pydantic import BaseModel, ValidationError, validator
from typing import Optional

class ValidatedWeatherParams(BaseModel):
    location: str
    unit: Literal["celsius", "fahrenheit"] = "celsius"
    
    @validator("location")
    def validate_location(cls, v):
        if len(v) < 2:
            raise ValueError("Location must be at least 2 characters")
        if not any(c.isalpha() for c in v):
            raise ValueError("Location must contain letters")
        return v.strip()

class ArgumentValidator:
    """Validate function arguments before execution"""
    
    def __init__(self, schemas: Dict[str, type[BaseModel]]):
        self.schemas = schemas
    
    def validate(
        self,
        function_name: str,
        arguments: dict
    ) -> tuple[bool, dict, str]:
        """
        Validate arguments against schema.
        Returns: (is_valid, validated_args, error_message)
        """
        schema = self.schemas.get(function_name)
        
        if not schema:
            return True, arguments, None
        
        try:
            validated = schema.model_validate(arguments)
            return True, validated.model_dump(), None
        except ValidationError as e:
            error_messages = []
            for error in e.errors():
                field = ".".join(str(x) for x in error["loc"])
                error_messages.append(f"{field}: {error['msg']}")
            return False, arguments, "; ".join(error_messages)

# Usage with auto-correction
async def validated_function_call(
    client: OpenAI,
    registry: FunctionRegistry,
    validator: ArgumentValidator,
    messages: list,
    max_validation_retries: int = 2
) -> FunctionResult:
    """Execute function call with validation and retry"""
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=registry.get_tools()
    )
    
    tool_call = response.choices[0].message.tool_calls[0]
    func_name = tool_call.function.name
    arguments = json.loads(tool_call.function.arguments)
    
    # Validate
    is_valid, validated_args, error = validator.validate(func_name, arguments)
    
    if not is_valid:
        # Ask LLM to fix the arguments
        messages.append(response.choices[0].message)
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps({
                "error": f"Validation failed: {error}",
                "hint": "Please correct the arguments and try again"
            })
        })
        
        # Retry
        return await validated_function_call(
            client, registry, validator, messages,
            max_validation_retries - 1
        )
    
    # Execute with validated arguments
    return await registry.execute(func_name, validated_args)

from enum import Enum
from dataclasses import dataclass

class FunctionErrorType(str, Enum):
    NOT_FOUND = "function_not_found"
    VALIDATION = "validation_error"
    EXECUTION = "execution_error"
    TIMEOUT = "timeout"
    PERMISSION = "permission_denied"

@dataclass
class FunctionError:
    type: FunctionErrorType
    message: str
    function_name: str
    recoverable: bool = True
    suggestion: str = None

class RobustFunctionExecutor:
    """Execute functions with comprehensive error handling"""
    
    def __init__(
        self,
        registry: FunctionRegistry,
        timeout: float = 30.0,
        max_retries: int = 2
    ):
        self.registry = registry
        self.timeout = timeout
        self.max_retries = max_retries
    
    async def execute_with_retry(
        self,
        name: str,
        arguments: dict
    ) -> tuple[Any, FunctionError]:
        """Execute with retry on transient errors"""
        
        last_error = None
        
        for attempt in range(self.max_retries + 1):
            try:
                result = await asyncio.wait_for(
                    self.registry.execute(name, arguments),
                    timeout=self.timeout
                )
                
                if result.success:
                    return result.result, None
                else:
                    last_error = FunctionError(
                        type=FunctionErrorType.EXECUTION,
                        message=result.error,
                        function_name=name
                    )
            
            except asyncio.TimeoutError:
                last_error = FunctionError(
                    type=FunctionErrorType.TIMEOUT,
                    message=f"Function timed out after {self.timeout}s",
                    function_name=name,
                    recoverable=True,
                    suggestion="Try with simpler parameters or split the request"
                )
            
            except Exception as e:
                last_error = FunctionError(
                    type=FunctionErrorType.EXECUTION,
                    message=str(e),
                    function_name=name,
                    recoverable=False
                )
                break
            
            # Exponential backoff for retries
            if attempt < self.max_retries:
                await asyncio.sleep(2 ** attempt)
        
        return None, last_error
    
    def format_error_for_llm(self, error: FunctionError) -> str:
        """Format error message for LLM understanding"""
        response = {
            "error": True,
            "type": error.type.value,
            "message": error.message
        }
        
        if error.suggestion:
            response["suggestion"] = error.suggestion
        
        if error.recoverable:
            response["hint"] = "You may retry with different parameters"
        
        return json.dumps(response)

class ToolChoiceController:
    """Control when and how tools are used"""
    
    @staticmethod
    def force_tool(tool_name: str) -> dict:
        """Force the model to use a specific tool"""
        return {
            "type": "function",
            "function": {"name": tool_name}
        }
    
    @staticmethod
    def no_tools() -> str:
        """Prevent tool usage"""
        return "none"
    
    @staticmethod
    def auto() -> str:
        """Let model decide"""
        return "auto"
    
    @staticmethod
    def required() -> str:
        """Model must use at least one tool"""
        return "required"

# Usage patterns
def query_with_tool_control(
    client: OpenAI,
    messages: list,
    tools: list,
    intent: str
) -> str:
    """Query with appropriate tool control based on intent"""
    
    # Determine tool choice based on intent
    if intent == "lookup":
        # Force tool usage for lookups
        tool_choice = ToolChoiceController.required()
    elif intent == "conversation":
        # Allow but don't require tools
        tool_choice = ToolChoiceController.auto()
    elif intent == "specific_action":
        # Force specific tool
        tool_choice = ToolChoiceController.force_tool("execute_action")
    else:
        tool_choice = ToolChoiceController.auto()
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        tool_choice=tool_choice
    )
    
    return response

async def stream_with_functions(
    client: OpenAI,
    messages: list,
    tools: list
):
    """Handle streaming responses with function calls"""
    
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        stream=True
    )
    
    # Accumulate function call data
    tool_calls = {}
    current_content = ""
    
    for chunk in stream:
        delta = chunk.choices[0].delta
        
        # Handle content
        if delta.content:
            current_content += delta.content
            yield {"type": "content", "data": delta.content}
        
        # Handle tool calls
        if delta.tool_calls:
            for tc in delta.tool_calls:
                idx = tc.index
                
                if idx not in tool_calls:
                    tool_calls[idx] = {
                        "id": tc.id,
                        "function": {"name": "", "arguments": ""}
                    }
                
                if tc.function.name:
                    tool_calls[idx]["function"]["name"] += tc.function.name
                
                if tc.function.arguments:
                    tool_calls[idx]["function"]["arguments"] += tc.function.arguments
        
        # Check finish reason
        if chunk.choices[0].finish_reason == "tool_calls":
            yield {
                "type": "tool_calls",
                "data": list(tool_calls.values())
            }