Complete Guide to LLM Function Calling

from typing import Dict, Any, Optional, List
from pydantic import BaseModel, ValidationError, Field
import json
import logging

logger = logging.getLogger(__name__)

# Define schemas with Pydantic for type safety
class WeatherParams(BaseModel):
    location: str = Field(..., description="City and state")
    unit: str = Field("fahrenheit", pattern="^(celsius|fahrenheit)$")

class DatabaseQueryParams(BaseModel):
    table: str = Field(..., pattern="^[a-zA-Z_][a-zA-Z0-9_]*$")
    filters: Dict[str, Any] = Field(default_factory=dict)
    limit: int = Field(10, ge=1, le=100)

class FunctionCallHandler:
    """Handles function calls with comprehensive error handling"""
    
    def __init__(self):
        self.functions = {
            "get_weather": (self.get_weather, WeatherParams),
            "query_database": (self.query_database, DatabaseQueryParams),
        }
        self.call_history = []
    
    def validate_and_execute(
        self,
        function_name: str,
        arguments: Union[str, dict],
        tool_call_id: Optional[str] = None
    ) -> Dict[str, Any]:
        """Validate arguments and execute function with error handling"""
        
        # Log the call
        call_record = {
            "function": function_name,
            "arguments": arguments,
            "tool_call_id": tool_call_id,
            "timestamp": time.time()
        }
        self.call_history.append(call_record)
        
        try:
            # Check if function exists
            if function_name not in self.functions:
                return self._error_response(
                    f"Unknown function: {function_name}",
                    "FUNCTION_NOT_FOUND"
                )
            
            func, schema = self.functions[function_name]
            
            # Parse arguments if string
            if isinstance(arguments, str):
                try:
                    arguments = json.loads(arguments)
                except json.JSONDecodeError as e:
                    return self._error_response(
                        f"Invalid JSON arguments: {e}",
                        "INVALID_JSON"
                    )
            
            # Validate with Pydantic
            try:
                validated_args = schema(**arguments)
            except ValidationError as e:
                errors = []
                for error in e.errors():
                    field = " -> ".join(str(x) for x in error["loc"])
                    errors.append(f"{field}: {error['msg']}")
                
                return self._error_response(
                    f"Validation failed: {'; '.join(errors)}",
                    "VALIDATION_ERROR"
                )
            
            # Execute function with timeout
            import signal
            
            def timeout_handler(signum, frame):
                raise TimeoutError("Function execution timed out")
            
            # Set timeout (Unix only, use threading.Timer for cross-platform)
            signal.signal(signal.SIGALRM, timeout_handler)
            signal.alarm(30)  # 30 second timeout
            
            try:
                result = func(validated_args)
                signal.alarm(0)  # Cancel timeout
                
                return {
                    "success": True,
                    "result": result,
                    "function": function_name
                }
                
            except TimeoutError:
                return self._error_response(
                    "Function execution timed out after 30 seconds",
                    "TIMEOUT"
                )
            except Exception as e:
                logger.exception(f"Function {function_name} failed")
                return self._error_response(
                    f"Execution failed: {str(e)}",
                    "EXECUTION_ERROR"
                )
                
        except Exception as e:
            logger.exception(f"Unexpected error in function call handler")
            return self._error_response(
                f"Unexpected error: {str(e)}",
                "UNKNOWN_ERROR"
            )
    
    def _error_response(self, message: str, code: str) -> Dict[str, Any]:
        """Create standardized error response"""
        return {
            "success": False,
            "error": {
                "message": message,
                "code": code
            }
        }
    
    def get_weather(self, params: WeatherParams) -> Dict[str, Any]:
        """Mock weather function"""
        # In production, call actual weather API
        return {
            "location": params.location,
            "temperature": 72,
            "unit": params.unit,
            "conditions": "Partly cloudy"
        }
    
    def query_database(self, params: DatabaseQueryParams) -> List[Dict]:
        """Mock database query"""
        # Validate table name against whitelist
        allowed_tables = ["users", "products", "orders"]
        if params.table not in allowed_tables:
            raise ValueError(f"Access denied for table: {params.table}")
        
        # In production, use parameterized queries
        return [
            {"id": 1, "name": "Sample"},
            {"id": 2, "name": "Data"}
        ]

# Usage with OpenAI
handler = FunctionCallHandler()

def process_tool_calls(message) -> List[Dict[str, Any]]:
    """Process all tool calls from a message"""
    results = []
    
    if hasattr(message, 'tool_calls') and message.tool_calls:
        for tool_call in message.tool_calls:
            result = handler.validate_and_execute(
                function_name=tool_call.function.name,
                arguments=tool_call.function.arguments,
                tool_call_id=tool_call.id
            )
            results.append(result)
    
    return results

import asyncio
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict, Any
import time

class ParallelFunctionExecutor:
    """Execute multiple function calls in parallel"""
    
    def __init__(self, max_workers: int = 10):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.async_functions = {}
        self.sync_functions = {}
    
    def register_function(
        self,
        name: str,
        func: callable,
        is_async: bool = False
    ):
        """Register a function for parallel execution"""
        if is_async:
            self.async_functions[name] = func
        else:
            self.sync_functions[name] = func
    
    async def execute_parallel_async(
        self,
        tool_calls: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        """Execute multiple async functions in parallel"""
        tasks = []
        
        for call in tool_calls:
            func_name = call["function"]["name"]
            args = call["function"]["arguments"]
            
            if func_name in self.async_functions:
                task = self._execute_async_with_metadata(
                    self.async_functions[func_name],
                    args,
                    call.get("id")
                )
                tasks.append(task)
            elif func_name in self.sync_functions:
                # Run sync function in thread pool
                task = self._execute_sync_in_thread(
                    self.sync_functions[func_name],
                    args,
                    call.get("id")
                )
                tasks.append(task)
        
        # Wait for all tasks to complete
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Process results
        processed_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                processed_results.append({
                    "tool_call_id": tool_calls[i].get("id"),
                    "error": str(result),
                    "success": False
                })
            else:
                processed_results.append(result)
        
        return processed_results
    
    async def _execute_async_with_metadata(
        self,
        func: callable,
        args: Dict[str, Any],
        tool_call_id: str
    ) -> Dict[str, Any]:
        """Execute async function with timing and metadata"""
        start_time = time.time()
        
        try:
            result = await func(**args)
            execution_time = time.time() - start_time
            
            return {
                "tool_call_id": tool_call_id,
                "result": result,
                "success": True,
                "execution_time": execution_time
            }
        except Exception as e:
            return {
                "tool_call_id": tool_call_id,
                "error": str(e),
                "success": False,
                "execution_time": time.time() - start_time
            }
    
    async def _execute_sync_in_thread(
        self,
        func: callable,
        args: Dict[str, Any],
        tool_call_id: str
    ) -> Dict[str, Any]:
        """Execute sync function in thread pool"""
        loop = asyncio.get_event_loop()
        
        def wrapped():
            start_time = time.time()
            try:
                result = func(**args)
                return {
                    "tool_call_id": tool_call_id,
                    "result": result,
                    "success": True,
                    "execution_time": time.time() - start_time
                }
            except Exception as e:
                return {
                    "tool_call_id": tool_call_id,
                    "error": str(e),
                    "success": False,
                    "execution_time": time.time() - start_time
                }
        
        return await loop.run_in_executor(self.executor, wrapped)
    
    def execute_parallel_sync(
        self,
        tool_calls: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        """Execute multiple functions in parallel (sync version)"""
        futures = []
        
        for call in tool_calls:
            func_name = call["function"]["name"]
            args = call["function"]["arguments"]
            
            if func_name in self.sync_functions:
                future = self.executor.submit(
                    self._execute_with_metadata,
                    self.sync_functions[func_name],
                    args,
                    call.get("id")
                )
                futures.append(future)
        
        # Collect results as they complete
        results = []
        for future in as_completed(futures):
            try:
                result = future.result(timeout=30)
                results.append(result)
            except Exception as e:
                results.append({
                    "error": str(e),
                    "success": False
                })
        
        return results
    
    def _execute_with_metadata(
        self,
        func: callable,
        args: Dict[str, Any],
        tool_call_id: str
    ) -> Dict[str, Any]:
        """Execute function with metadata"""
        start_time = time.time()
        
        try:
            result = func(**args)
            return {
                "tool_call_id": tool_call_id,
                "result": result,
                "success": True,
                "execution_time": time.time() - start_time
            }
        except Exception as e:
            return {
                "tool_call_id": tool_call_id,
                "error": str(e),
                "success": False,
                "execution_time": time.time() - start_time
            }

# Example usage
executor = ParallelFunctionExecutor()

# Register functions
executor.register_function("get_weather", get_weather)
executor.register_function("search_products", search_products)
executor.register_function("calculate_shipping", calculate_shipping)

# Simulate multiple tool calls from LLM
tool_calls = [
    {
        "id": "call_123",
        "function": {
            "name": "get_weather",
            "arguments": {"location": "Boston", "unit": "fahrenheit"}
        }
    },
    {
        "id": "call_124",
        "function": {
            "name": "search_products",
            "arguments": {"query": "laptop", "max_price": 1000}
        }
    },
    {
        "id": "call_125",
        "function": {
            "name": "calculate_shipping",
            "arguments": {"weight": 5, "destination": "CA"}
        }
    }
]

# Execute in parallel
results = executor.execute_parallel_sync(tool_calls)

# Process results
for result in results:
    if result["success"]:
        print(f"✓ {result['tool_call_id']}: {result['result']}")
        print(f"  Execution time: {result['execution_time']:.2f}s")
    else:
        print(f"✗ {result['tool_call_id']}: {result['error']}")

from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum
import json

class WorkflowState(Enum):
    STARTED = "started"
    IN_PROGRESS = "in_progress"
    WAITING_FOR_INPUT = "waiting_for_input"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class WorkflowStep:
    name: str
    function: str
    depends_on: List[str] = None
    condition: Optional[str] = None

class WorkflowOrchestrator:
    """Orchestrate complex multi-step function calling workflows"""
    
    def __init__(self, llm_client, function_handler):
        self.llm_client = llm_client
        self.function_handler = function_handler
        self.workflows = {}
        self.context = {}
    
    def define_workflow(self, name: str, steps: List[WorkflowStep]):
        """Define a reusable workflow"""
        self.workflows[name] = steps
    
    async def execute_workflow(
        self,
        workflow_name: str,
        initial_input: str,
        max_steps: int = 10
    ) -> Dict[str, Any]:
        """Execute a predefined workflow"""
        
        if workflow_name not in self.workflows:
            raise ValueError(f"Unknown workflow: {workflow_name}")
        
        workflow = self.workflows[workflow_name]
        state = WorkflowState.STARTED
        context = {
            "initial_input": initial_input,
            "steps_completed": [],
            "results": {},
            "messages": [
                {"role": "user", "content": initial_input}
            ]
        }
        
        for step_count in range(max_steps):
            # Find next step to execute
            next_step = self._find_next_step(workflow, context)
            
            if not next_step:
                state = WorkflowState.COMPLETED
                break
            
            # Check conditions
            if next_step.condition and not self._evaluate_condition(
                next_step.condition, context
            ):
                continue
            
            # Execute step
            result = await self._execute_step(next_step, context)
            
            # Update context
            context["steps_completed"].append(next_step.name)
            context["results"][next_step.name] = result
            
            # Check if workflow needs user input
            if result.get("requires_input"):
                state = WorkflowState.WAITING_FOR_INPUT
                break
        
        return {
            "state": state.value,
            "context": context,
            "final_result": self._prepare_final_result(context)
        }
    
    def _find_next_step(
        self,
        workflow: List[WorkflowStep],
        context: Dict[str, Any]
    ) -> Optional[WorkflowStep]:
        """Find the next step that can be executed"""
        
        completed = set(context["steps_completed"])
        
        for step in workflow:
            # Skip completed steps
            if step.name in completed:
                continue
            
            # Check dependencies
            if step.depends_on:
                deps_met = all(dep in completed for dep in step.depends_on)
                if not deps_met:
                    continue
            
            return step
        
        return None
    
    async def _execute_step(
        self,
        step: WorkflowStep,
        context: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Execute a single workflow step"""
        
        # Prepare prompt with context
        prompt = self._build_step_prompt(step, context)
        
        # Get LLM to decide on function parameters
        response = await self.llm_client.chat.completions.create(
            model="gpt-4-turbo-preview",
            messages=context["messages"] + [
                {"role": "user", "content": prompt}
            ],
            tools=[self._get_tool_definition(step.function)],
            tool_choice={"type": "function", "function": {"name": step.function}}
        )
        
        message = response.choices[0].message
        context["messages"].append(message)
        
        # Execute function calls
        if message.tool_calls:
            results = []
            for tool_call in message.tool_calls:
                result = self.function_handler.validate_and_execute(
                    tool_call.function.name,
                    tool_call.function.arguments,
                    tool_call.id
                )
                results.append(result)
                
                # Add result to context
                context["messages"].append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(result)
                })
            
            return {"step": step.name, "results": results}
        
        return {"step": step.name, "error": "No function call generated"}
    
    def _build_step_prompt(
        self,
        step: WorkflowStep,
        context: Dict[str, Any]
    ) -> str:
        """Build prompt for workflow step"""
        
        prompt = f"Execute step: {step.name}\n"
        
        # Add relevant context from previous steps
        if step.depends_on:
            prompt += "\nContext from previous steps:\n"
            for dep in step.depends_on:
                if dep in context["results"]:
                    prompt += f"- {dep}: {json.dumps(context['results'][dep])}\n"
        
        return prompt

# Example: E-commerce order workflow
orchestrator = WorkflowOrchestrator(llm_client, function_handler)

# Define order processing workflow
order_workflow = [
    WorkflowStep(
        name="validate_customer",
        function="lookup_customer",
        depends_on=[]
    ),
    WorkflowStep(
        name="check_inventory",
        function="check_product_availability",
        depends_on=["validate_customer"]
    ),
    WorkflowStep(
        name="calculate_pricing",
        function="calculate_total_price",
        depends_on=["check_inventory"]
    ),
    WorkflowStep(
        name="process_payment",
        function="charge_payment",
        depends_on=["calculate_pricing"],
        condition="context['results']['calculate_pricing']['total'] > 0"
    ),
    WorkflowStep(
        name="create_order",
        function="create_order_record",
        depends_on=["process_payment"]
    ),
    WorkflowStep(
        name="send_confirmation",
        function="send_email",
        depends_on=["create_order"]
    )
]

orchestrator.define_workflow("process_order", order_workflow)

# Execute workflow
result = await orchestrator.execute_workflow(
    "process_order",
    "I want to order 2 laptops for customer ID 12345"
)

import pytest
from unittest.mock import Mock, patch, AsyncMock
import json

class TestFunctionCalling:
    """Comprehensive test suite for function calling"""
    
    @pytest.fixture
    def mock_llm_client(self):
        """Mock LLM client for testing"""
        client = Mock()
        return client
    
    @pytest.fixture
    def function_handler(self):
        """Create function handler for tests"""
        handler = FunctionCallHandler()
        return handler
    
    def test_function_schema_validation(self):
        """Test that function schemas are valid"""
        schema = {
            "name": "test_function",
            "description": "Test function",
            "parameters": {
                "type": "object",
                "properties": {
                    "param1": {"type": "string"},
                    "param2": {"type": "number"}
                },
                "required": ["param1"]
            }
        }
        
        # Validate schema structure
        assert schema["name"].replace("_", "").isalnum()
        assert len(schema["description"]) > 0
        assert schema["parameters"]["type"] == "object"
    
    def test_argument_validation(self, function_handler):
        """Test argument validation"""
        
        # Valid arguments
        result = function_handler.validate_and_execute(
            "get_weather",
            {"location": "Boston", "unit": "celsius"}
        )
        assert result["success"] is True
        
        # Missing required argument
        result = function_handler.validate_and_execute(
            "get_weather",
            {"unit": "celsius"}  # Missing location
        )
        assert result["success"] is False
        assert "location" in result["error"]["message"]
        
        # Invalid enum value
        result = function_handler.validate_and_execute(
            "get_weather",
            {"location": "Boston", "unit": "kelvin"}  # Invalid unit
        )
        assert result["success"] is False
    
    @pytest.mark.asyncio
    async def test_parallel_execution(self):
        """Test parallel function execution"""
        executor = ParallelFunctionExecutor()
        
        # Mock functions with delays
        async def slow_function(delay: float):
            await asyncio.sleep(delay)
            return f"Completed after {delay}s"
        
        executor.register_function("slow1", lambda: slow_function(0.1), True)
        executor.register_function("slow2", lambda: slow_function(0.2), True)
        executor.register_function("slow3", lambda: slow_function(0.3), True)
        
        tool_calls = [
            {"id": "1", "function": {"name": "slow1", "arguments": {}}},
            {"id": "2", "function": {"name": "slow2", "arguments": {}}},
            {"id": "3", "function": {"name": "slow3", "arguments": {}}}
        ]
        
        # Execute in parallel
        start_time = time.time()
        results = await executor.execute_parallel_async(tool_calls)
        total_time = time.time() - start_time
        
        # Should complete in ~0.3s (not 0.6s sequential)
        assert total_time < 0.4
        assert all(r["success"] for r in results)
    
    def test_error_handling(self, function_handler):
        """Test error handling scenarios"""
        
        # Function that raises exception
        def failing_function():
            raise ValueError("Intentional error")
        
        function_handler.functions["failing"] = (failing_function, BaseModel)
        
        result = function_handler.validate_and_execute("failing", {})
        assert result["success"] is False
        assert "Intentional error" in result["error"]["message"]
    
    def test_security_validation(self):
        """Test security checks"""
        secure_caller = SecureFunctionCaller()
        secure_caller.register_safe_function("query_db", Mock())
        
        # SQL injection attempt
        is_valid, error = secure_caller.validate_function_call(
            "query_db",
            {"query": "SELECT * FROM users; DROP TABLE users;"}
        )
        assert is_valid is False
        assert "dangerous" in error.lower()
        
        # Path traversal attempt
        is_valid, error = secure_caller.validate_function_call(
            "read_file",
            {"path": "../../etc/passwd"}
        )
        assert is_valid is False
        assert "traversal" in error.lower()
    
    @patch('openai.ChatCompletion.create')
    def test_end_to_end_flow(self, mock_create, function_handler):
        """Test complete function calling flow"""
        
        # Mock LLM response with function call
        mock_create.return_value = Mock(
            choices=[
                Mock(
                    message=Mock(
                        tool_calls=[
                            Mock(
                                id="call_123",
                                function=Mock(
                                    name="get_weather",
                                    arguments='{"location": "Boston"}'
                                )
                            )
                        ]
                    )
                )
            ]
        )
        
        # Process the response
        response = mock_create()
        message = response.choices[0].message
        
        # Execute function calls
        results = []
        for tool_call in message.tool_calls:
            result = function_handler.validate_and_execute(
                tool_call.function.name,
                json.loads(tool_call.function.arguments),
                tool_call.id
            )
            results.append(result)
        
        assert len(results) == 1
        assert results[0]["success"] is True
        assert "temperature" in results[0]["result"]

# Run tests
pytest.main([__file__, "-v"])