Function Calling and Tool Use with LLMs

You're building a customer support chatbot for an e-commerce platform. A customer asks, "What's the status of my order #12345, and when will it arrive?" Your LLM can craft a perfectly polite response, but it can't actually look up order data, check shipping APIs, or access your inventory system. It's stuck generating plausible-sounding but potentially incorrect information.

This is where function calling transforms LLMs from eloquent but isolated text generators into powerful agents that can interact with real systems. By the end of this lesson, you'll understand how to give LLMs access to external tools and data sources, turning them into practical problem-solving systems that can take action in the real world.

What you'll learn:

• How to design and implement function calling systems with OpenAI's API and other providers
• Best practices for defining function schemas that LLMs can reliably use
• Error handling patterns for robust tool integration
• Security considerations when giving LLMs access to external systems
• Performance optimization techniques for multi-step tool workflows
• Real-world patterns for building LLM agents that solve complex business problems

Prerequisites

You should be comfortable with:

• Making API calls and handling JSON responses
• Basic LLM concepts (prompts, tokens, completions)
• Python programming and working with external APIs
• Understanding of RESTful APIs and HTTP methods

Understanding Function Calling Architecture

Function calling, also known as tool use, allows LLMs to invoke external functions with structured parameters. Instead of just generating text, the LLM can decide which tools to use, extract the necessary parameters from user input, and coordinate multiple function calls to solve complex problems.

The basic flow works like this: you provide the LLM with function definitions (schemas), the LLM decides which functions to call based on user input, returns structured function calls instead of text, your application executes those functions, and then you feed the results back to the LLM for final processing.

Let's start with a simple example using OpenAI's function calling API:

import openai
import json
from datetime import datetime

# Define a function the LLM can call
def get_order_status(order_id):
    """Simulate fetching order status from a database."""
    # In reality, this would query your order management system
    mock_orders = {
        "12345": {
            "status": "shipped",
            "tracking_number": "1Z999AA1234567890",
            "estimated_delivery": "2024-01-15",
            "items": ["Wireless Headphones", "Phone Case"]
        },
        "67890": {
            "status": "processing",
            "estimated_ship_date": "2024-01-12",
            "items": ["Laptop Stand"]
        }
    }
    return mock_orders.get(order_id, {"error": "Order not found"})

# Define the function schema for the LLM
order_status_schema = {
    "type": "function",
    "function": {
        "name": "get_order_status",
        "description": "Get the current status and details of a customer order",
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "The order ID to look up"
                }
            },
            "required": ["order_id"]
        }
    }
}

client = openai.OpenAI(api_key="your-api-key")

def handle_customer_query(user_message):
    """Process a customer query that might require function calling."""
    
    # Initial call to see if the LLM wants to use any functions
    response = client.chat.completions.create(
        model="gpt-4-1106-preview",
        messages=[
            {"role": "system", "content": "You are a helpful customer service assistant. Use the available tools to help customers with their orders."},
            {"role": "user", "content": user_message}
        ],
        tools=[order_status_schema],
        tool_choice="auto"
    )
    
    message = response.choices[0].message
    
    # Check if the LLM wants to call a function
    if message.tool_calls:
        # Process each function call
        for tool_call in message.tool_calls:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)
            
            # Call the actual function
            if function_name == "get_order_status":
                result = get_order_status(function_args["order_id"])
                
                # Send the function result back to the LLM
                follow_up_response = client.chat.completions.create(
                    model="gpt-4-1106-preview",
                    messages=[
                        {"role": "system", "content": "You are a helpful customer service assistant."},
                        {"role": "user", "content": user_message},
                        {"role": "assistant", "content": None, "tool_calls": message.tool_calls},
                        {"role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result)}
                    ]
                )
                
                return follow_up_response.choices[0].message.content
    
    # If no function calls, return the direct response
    return message.content

# Test it
query = "Hi, can you check the status of my order #12345?"
response = handle_customer_query(query)
print(response)

This example demonstrates the core pattern: define functions with clear schemas, let the LLM decide when to use them, execute the functions, and feed results back for natural language processing.

Designing Effective Function Schemas

The quality of your function schemas directly impacts how reliably the LLM can use your tools. Poorly defined schemas lead to incorrect parameter extraction and failed function calls.

Here are the key principles for effective schema design:

Be Specific and Descriptive: Function names and descriptions should clearly indicate what the function does and when to use it.

# Poor schema - vague and ambiguous
bad_schema = {
    "name": "get_data",
    "description": "Gets data",
    "parameters": {
        "type": "object",
        "properties": {
            "id": {"type": "string"}
        }
    }
}

# Good schema - specific and clear
good_schema = {
    "name": "get_customer_order_history",
    "description": "Retrieve the complete order history for a specific customer, including order dates, items, amounts, and status",
    "parameters": {
        "type": "object",
        "properties": {
            "customer_email": {
                "type": "string",
                "description": "The customer's email address used to place orders"
            },
            "limit": {
                "type": "integer",
                "description": "Maximum number of recent orders to return (default: 10, max: 100)",
                "minimum": 1,
                "maximum": 100
            }
        },
        "required": ["customer_email"]
    }
}

Use Enums for Constrained Values: When parameters have limited valid values, use enums to guide the LLM:

inventory_check_schema = {
    "type": "function",
    "function": {
        "name": "check_product_inventory",
        "description": "Check current inventory levels for a product across all warehouses",
        "parameters": {
            "type": "object",
            "properties": {
                "product_sku": {
                    "type": "string",
                    "description": "The product SKU code"
                },
                "warehouse_region": {
                    "type": "string",
                    "enum": ["north_america", "europe", "asia_pacific", "all"],
                    "description": "Which warehouse region to check (default: all)"
                },
                "include_reserved": {
                    "type": "boolean",
                    "description": "Whether to include inventory reserved for pending orders"
                }
            },
            "required": ["product_sku"]
        }
    }
}

Provide Examples in Descriptions: For complex parameters, include examples in the description:

date_range_schema = {
    "type": "function",
    "function": {
        "name": "generate_sales_report",
        "description": "Generate a sales report for a specified date range",
        "parameters": {
            "type": "object",
            "properties": {
                "start_date": {
                    "type": "string",
                    "description": "Start date in YYYY-MM-DD format (e.g., '2024-01-01')"
                },
                "end_date": {
                    "type": "string",
                    "description": "End date in YYYY-MM-DD format (e.g., '2024-01-31')"
                },
                "group_by": {
                    "type": "string",
                    "enum": ["day", "week", "month", "product", "category"],
                    "description": "How to group the sales data in the report"
                }
            },
            "required": ["start_date", "end_date"]
        }
    }
}

Building a Multi-Tool System

Real applications typically need multiple tools working together. Let's build a more comprehensive system for our e-commerce customer service bot:

import openai
import json
import requests
from datetime import datetime, timedelta
from typing import Dict, List, Any

class EcommerceTools:
    def __init__(self, api_base_url: str):
        self.api_base_url = api_base_url
        
    def get_order_status(self, order_id: str) -> Dict:
        """Get order status and tracking information."""
        # Simulate API call
        mock_data = {
            "12345": {
                "order_id": "12345",
                "status": "shipped",
                "tracking_number": "1Z999AA1234567890",
                "carrier": "UPS",
                "estimated_delivery": "2024-01-15",
                "items": [
                    {"name": "Wireless Headphones", "quantity": 1, "price": 89.99},
                    {"name": "Phone Case", "quantity": 2, "price": 15.99}
                ],
                "total": 121.97
            }
        }
        return mock_data.get(order_id, {"error": "Order not found"})
    
    def search_products(self, query: str, category: str = None, max_results: int = 10) -> List[Dict]:
        """Search for products in the catalog."""
        mock_products = [
            {
                "id": "PROD001",
                "name": "Wireless Noise-Canceling Headphones",
                "category": "Electronics",
                "price": 199.99,
                "rating": 4.5,
                "in_stock": True,
                "description": "Premium wireless headphones with active noise cancellation"
            },
            {
                "id": "PROD002", 
                "name": "Bluetooth Earbuds",
                "category": "Electronics",
                "price": 79.99,
                "rating": 4.2,
                "in_stock": True,
                "description": "Compact wireless earbuds with charging case"
            },
            {
                "id": "PROD003",
                "name": "Phone Case - Clear",
                "category": "Accessories",
                "price": 12.99,
                "rating": 4.0,
                "in_stock": False,
                "description": "Transparent protective case for smartphones"
            }
        ]
        
        # Simple search simulation
        results = []
        query_lower = query.lower()
        for product in mock_products:
            if (query_lower in product["name"].lower() or 
                query_lower in product["description"].lower()):
                if not category or product["category"].lower() == category.lower():
                    results.append(product)
                    
        return results[:max_results]
    
    def check_return_policy(self, order_id: str, item_name: str = None) -> Dict:
        """Check return policy and eligibility for an order or specific item."""
        order = self.get_order_status(order_id)
        if "error" in order:
            return order
            
        return {
            "eligible_for_return": True,
            "return_window_days": 30,
            "return_methods": ["mail", "store"],
            "restocking_fee": 0,
            "return_shipping_cost": "free",
            "policy_details": "Items can be returned within 30 days of delivery in original condition"
        }
    
    def initiate_return(self, order_id: str, item_names: List[str], reason: str) -> Dict:
        """Initiate a return for specific items from an order."""
        # In a real system, this would create a return request
        return {
            "return_id": f"RET-{order_id}-001",
            "status": "initiated",
            "items": item_names,
            "return_label_url": "https://example.com/return-label.pdf",
            "instructions": "Package items securely and attach the provided return label"
        }

# Define all function schemas
FUNCTION_SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "get_order_status",
            "description": "Get current status, tracking, and details for a customer order",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "The order ID to look up (format: 5-digit number)"
                    }
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function", 
        "function": {
            "name": "search_products",
            "description": "Search the product catalog to help customers find items",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search terms for products (e.g., 'wireless headphones', 'phone case')"
                    },
                    "category": {
                        "type": "string",
                        "enum": ["Electronics", "Accessories", "Clothing", "Home", "Sports"],
                        "description": "Optional category filter"
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Maximum number of products to return (default: 10)",
                        "minimum": 1,
                        "maximum": 20
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "check_return_policy",
            "description": "Check return policy and eligibility for an order",
            "parameters": {
                "type": "object", 
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "The order ID to check return policy for"
                    },
                    "item_name": {
                        "type": "string",
                        "description": "Optional specific item name to check (if not provided, checks whole order)"
                    }
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "initiate_return",
            "description": "Start the return process for specific items from an order",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string", 
                        "description": "The order ID to return items from"
                    },
                    "item_names": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "List of item names to return"
                    },
                    "reason": {
                        "type": "string",
                        "enum": ["defective", "wrong_item", "not_as_described", "changed_mind", "damaged_shipping"],
                        "description": "Reason for the return"
                    }
                },
                "required": ["order_id", "item_names", "reason"]
            }
        }
    }
]

class CustomerServiceAgent:
    def __init__(self, openai_client, tools_instance):
        self.client = openai_client
        self.tools = tools_instance
        self.conversation_history = []
        
    def process_message(self, user_message: str) -> str:
        """Process a customer message and return a response."""
        
        # Add user message to conversation history
        self.conversation_history.append({"role": "user", "content": user_message})
        
        # Call LLM with tools available
        response = self.client.chat.completions.create(
            model="gpt-4-1106-preview",
            messages=[
                {"role": "system", "content": """You are a helpful customer service assistant for an e-commerce store. 
                
                Use the available tools to help customers with:
                - Order status and tracking
                - Product searches and recommendations  
                - Return policies and processes
                - General inquiries
                
                Always be polite and professional. If you use tools, explain what you're doing.
                For order IDs, users might say 'order 12345' or just '12345' - extract the numeric ID."""}
            ] + self.conversation_history,
            tools=FUNCTION_SCHEMAS,
            tool_choice="auto"
        )
        
        assistant_message = response.choices[0].message
        
        # Handle function calls
        if assistant_message.tool_calls:
            # Add assistant message with tool calls to history
            self.conversation_history.append({
                "role": "assistant", 
                "content": assistant_message.content,
                "tool_calls": assistant_message.tool_calls
            })
            
            # Execute each tool call
            for tool_call in assistant_message.tool_calls:
                function_name = tool_call.function.name
                function_args = json.loads(tool_call.function.arguments)
                
                # Route to the appropriate tool method
                if hasattr(self.tools, function_name):
                    try:
                        result = getattr(self.tools, function_name)(**function_args)
                        tool_result = json.dumps(result)
                    except Exception as e:
                        tool_result = json.dumps({"error": f"Function call failed: {str(e)}"})
                else:
                    tool_result = json.dumps({"error": f"Function {function_name} not found"})
                
                # Add tool result to conversation
                self.conversation_history.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": tool_result
                })
            
            # Get final response from LLM with tool results
            final_response = self.client.chat.completions.create(
                model="gpt-4-1106-preview",
                messages=[
                    {"role": "system", "content": """You are a helpful customer service assistant. 
                    Use the tool results to provide a comprehensive, friendly response to the customer."""}
                ] + self.conversation_history
            )
            
            final_message = final_response.choices[0].message.content
            self.conversation_history.append({"role": "assistant", "content": final_message})
            
            return final_message
        else:
            # No tool calls needed, use direct response
            self.conversation_history.append({"role": "assistant", "content": assistant_message.content})
            return assistant_message.content

# Usage example
if __name__ == "__main__":
    client = openai.OpenAI(api_key="your-api-key")
    tools = EcommerceTools("https://api.example.com")
    agent = CustomerServiceAgent(client, tools)
    
    # Test conversation
    print("=== Customer Service Chat ===")
    
    queries = [
        "Hi, I'd like to check on my order 12345",
        "Can I return the headphones from that order? They're too big",
        "What other headphones do you have that might be smaller?"
    ]
    
    for query in queries:
        print(f"\nCustomer: {query}")
        response = agent.process_message(query)
        print(f"Agent: {response}")

This multi-tool system demonstrates several important patterns:

1. Tool Organization: Related functions are grouped in a class for better organization and shared state management
2. Conversation History: The agent maintains context across multiple exchanges
3. Error Handling: Function calls are wrapped in try-catch blocks
4. Function Routing: A generic mechanism routes function calls to the appropriate methods

Error Handling and Robustness

Function calling systems need robust error handling since they interact with external systems that can fail. Here's how to build resilience into your tool-calling workflows:

import functools
import time
import logging
from typing import Callable, Any, Optional

class ToolExecutionError(Exception):
    """Custom exception for tool execution failures."""
    pass

def retry_on_failure(max_retries: int = 3, backoff_factor: float = 1.0):
    """Decorator to retry function calls with exponential backoff."""
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt == max_retries:
                        break
                    
                    wait_time = backoff_factor * (2 ** attempt)
                    logging.warning(f"Attempt {attempt + 1} failed: {e}. Retrying in {wait_time}s...")
                    time.sleep(wait_time)
            
            raise ToolExecutionError(f"Function {func.__name__} failed after {max_retries + 1} attempts: {last_exception}")
        
        return wrapper
    return decorator

class RobustEcommerceTools(EcommerceTools):
    """Enhanced tools class with comprehensive error handling."""
    
    @retry_on_failure(max_retries=3)
    def get_order_status(self, order_id: str) -> Dict:
        """Get order status with retry logic and validation."""
        
        # Validate input
        if not order_id or not order_id.strip():
            raise ValueError("Order ID cannot be empty")
            
        if not order_id.isdigit() or len(order_id) != 5:
            raise ValueError("Order ID must be a 5-digit number")
        
        try:
            # Simulate API call that might fail
            if order_id == "99999":  # Simulate API error
                raise requests.RequestException("API temporarily unavailable")
                
            result = super().get_order_status(order_id)
            
            if "error" in result:
                # This is a business logic error, not a system error
                return result
                
            # Validate response structure
            required_fields = ["order_id", "status", "items", "total"]
            if not all(field in result for field in required_fields):
                raise ValueError("Invalid response format from order API")
                
            return result
            
        except requests.RequestException as e:
            logging.error(f"API error fetching order {order_id}: {e}")
            raise
        except Exception as e:
            logging.error(f"Unexpected error fetching order {order_id}: {e}")
            raise
    
    def search_products(self, query: str, category: str = None, max_results: int = 10) -> List[Dict]:
        """Search products with input validation and error handling."""
        
        # Validate inputs
        if not query or not query.strip():
            return {"error": "Search query cannot be empty"}
        
        if len(query.strip()) < 2:
            return {"error": "Search query must be at least 2 characters"}
        
        if max_results < 1 or max_results > 20:
            max_results = min(20, max(1, max_results))
        
        try:
            results = super().search_products(query, category, max_results)
            
            # Validate each product result
            validated_results = []
            for product in results:
                if all(field in product for field in ["id", "name", "price"]):
                    validated_results.append(product)
                else:
                    logging.warning(f"Skipping invalid product result: {product}")
            
            return validated_results
            
        except Exception as e:
            logging.error(f"Error searching products: {e}")
            return {"error": f"Product search failed: {str(e)}"}

class ResilientCustomerServiceAgent(CustomerServiceAgent):
    """Enhanced agent with better error handling and fallback strategies."""
    
    def __init__(self, openai_client, tools_instance):
        super().__init__(openai_client, tools_instance)
        self.max_function_calls_per_message = 5  # Prevent infinite loops
        
    def execute_function_call(self, tool_call) -> str:
        """Execute a single function call with comprehensive error handling."""
        
        function_name = tool_call.function.name
        
        try:
            function_args = json.loads(tool_call.function.arguments)
        except json.JSONDecodeError as e:
            logging.error(f"Invalid JSON in function arguments: {tool_call.function.arguments}")
            return json.dumps({
                "error": "Invalid function arguments format", 
                "details": str(e)
            })
        
        # Check if function exists
        if not hasattr(self.tools, function_name):
            return json.dumps({
                "error": f"Function '{function_name}' is not available",
                "available_functions": [name for name in dir(self.tools) if not name.startswith('_')]
            })
        
        try:
            # Execute the function
            result = getattr(self.tools, function_name)(**function_args)
            
            # Ensure result is JSON-serializable
            json.dumps(result)  # Test serialization
            return json.dumps(result)
            
        except TypeError as e:
            logging.error(f"Invalid arguments for {function_name}: {e}")
            return json.dumps({
                "error": f"Invalid arguments for {function_name}",
                "details": str(e),
                "provided_args": function_args
            })
        except ToolExecutionError as e:
            logging.error(f"Tool execution failed: {e}")
            return json.dumps({
                "error": "Service temporarily unavailable", 
                "details": "Please try again in a moment",
                "function": function_name
            })
        except Exception as e:
            logging.error(f"Unexpected error in {function_name}: {e}")
            return json.dumps({
                "error": f"An unexpected error occurred",
                "function": function_name
            })
    
    def process_message(self, user_message: str, max_iterations: int = 3) -> str:
        """Process message with limits to prevent infinite function calling loops."""
        
        self.conversation_history.append({"role": "user", "content": user_message})
        
        iteration = 0
        total_function_calls = 0
        
        while iteration < max_iterations:
            iteration += 1
            
            try:
                response = self.client.chat.completions.create(
                    model="gpt-4-1106-preview",
                    messages=[
                        {"role": "system", "content": """You are a helpful customer service assistant.
                        
                        If a tool call fails, acknowledge the issue politely and offer alternative help.
                        Do not repeatedly call the same failing function.
                        If you encounter errors, provide helpful explanations to the customer."""}
                    ] + self.conversation_history,
                    tools=FUNCTION_SCHEMAS,
                    tool_choice="auto"
                )
            except Exception as e:
                logging.error(f"OpenAI API error: {e}")
                return "I'm sorry, I'm experiencing technical difficulties right now. Please try again in a moment."
            
            assistant_message = response.choices[0].message
            
            if not assistant_message.tool_calls:
                # No more function calls needed
                self.conversation_history.append({
                    "role": "assistant", 
                    "content": assistant_message.content
                })
                return assistant_message.content
            
            # Check function call limits
            if total_function_calls + len(assistant_message.tool_calls) > self.max_function_calls_per_message:
                return "I'm sorry, this request is too complex. Please try breaking it into smaller questions."
            
            # Add assistant message to history
            self.conversation_history.append({
                "role": "assistant",
                "content": assistant_message.content,
                "tool_calls": assistant_message.tool_calls
            })
            
            # Execute function calls
            for tool_call in assistant_message.tool_calls:
                total_function_calls += 1
                result = self.execute_function_call(tool_call)
                
                self.conversation_history.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result
                })
        
        return "I'm sorry, I wasn't able to complete your request. Please try rephrasing your question or contact support directly."

Tip: Always validate function inputs and outputs. LLMs can sometimes generate invalid parameters or your external APIs might return unexpected data structures.

Security Considerations

Giving LLMs access to external systems introduces significant security risks. Here are essential security patterns:

Parameter Validation and Sanitization:

import re
from typing import Union

class SecureEcommerceTools(RobustEcommerceTools):
    """Tools class with security-focused validation."""
    
    def __init__(self, api_base_url: str, user_context: Dict):
        super().__init__(api_base_url)
        self.user_context = user_context  # Contains user ID, permissions, etc.
        
    def validate_order_access(self, order_id: str) -> bool:
        """Check if the current user has access to this order."""
        # In a real system, verify the order belongs to the authenticated user
        user_orders = self.user_context.get("accessible_orders", [])
        return order_id in user_orders
    
    def sanitize_search_query(self, query: str) -> str:
        """Sanitize search input to prevent injection attacks."""
        # Remove potentially dangerous characters
        sanitized = re.sub(r'[<>"\';\\]', '', query)
        # Limit length
        return sanitized[:100]
    
    def get_order_status(self, order_id: str) -> Dict:
        """Secure order status lookup with access control."""
        
        # Validate format
        if not re.match(r'^\d{5}$', order_id):
            return {"error": "Invalid order ID format"}
        
        # Check access permissions
        if not self.validate_order_access(order_id):
            return {"error": "Order not found"}  # Don't reveal existence
        
        return super().get_order_status(order_id)
    
    def search_products(self, query: str, category: str = None, max_results: int = 10) -> List[Dict]:
        """Secure product search with input sanitization."""
        
        # Sanitize inputs
        clean_query = self.sanitize_search_query(query)
        if category:
            category = self.sanitize_search_query(category)
        
        # Enforce limits
        max_results = min(20, max(1, max_results))
        
        return super().search_products(clean_query, category, max_results)

class SecureAgent:
    """Agent with security controls and audit logging."""
    
    def __init__(self, openai_client, tools_instance, user_id: str):
        self.client = openai_client
        self.tools = tools_instance
        self.user_id = user_id
        self.conversation_history = []
        self.audit_log = []
        
    def log_action(self, action: str, details: Dict):
        """Log all actions for audit purposes."""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "user_id": self.user_id,
            "action": action,
            "details": details
        }
        self.audit_log.append(log_entry)
        
        # In production, send to centralized logging
        logging.info(f"User action: {json.dumps(log_entry)}")
    
    def execute_function_call(self, tool_call) -> str:
        """Execute function with security logging."""
        
        function_name = tool_call.function.name
        
        try:
            function_args = json.loads(tool_call.function.arguments)
        except json.JSONDecodeError:
            self.log_action("function_call_failed", {
                "function": function_name,
                "error": "invalid_json",
                "raw_args": tool_call.function.arguments
            })
            return json.dumps({"error": "Invalid function arguments"})
        
        # Log the function call attempt
        self.log_action("function_call_attempted", {
            "function": function_name,
            "args": function_args
        })
        
        # Check rate limits (simplified)
        recent_calls = [log for log in self.audit_log[-10:] 
                       if log["action"] == "function_call_attempted"]
        if len(recent_calls) > 5:
            self.log_action("rate_limit_exceeded", {"function": function_name})
            return json.dumps({"error": "Rate limit exceeded"})
        
        try:
            result = getattr(self.tools, function_name)(**function_args)
            
            self.log_action("function_call_succeeded", {
                "function": function_name,
                "args": function_args
            })
            
            return json.dumps(result)
            
        except Exception as e:
            self.log_action("function_call_failed", {
                "function": function_name,
                "args": function_args,
                "error": str(e)
            })
            return json.dumps({"error": "Function execution failed"})

Access Control Patterns:

• Always validate user permissions before executing functions
• Use allowlists for function parameters where possible
• Implement rate limiting to prevent abuse
• Log all function calls for audit purposes
• Never expose internal system details in error messages

Performance Optimization

Function calling can be expensive in terms of API calls and latency. Here are optimization strategies:

Batching and Caching:

import hashlib
from functools import lru_cache
import asyncio
import aiohttp

class OptimizedEcommerceTools:
    """Performance-optimized tools with caching and batching."""
    
    def __init__(self, api_base_url: str):
        self.api_base_url = api_base_url
        self.cache = {}
        self.cache_ttl = 300  # 5 minutes
        
    def get_cache_key(self, function_name: str, **kwargs) -> str:
        """Generate a cache key for function results."""
        params_str = json.dumps(sorted(kwargs.items()))
        return hashlib.md5(f"{function_name}:{params_str}".encode()).hexdigest()
    
    @lru_cache(maxsize=1000)
    def get_order_status(self, order_id: str) -> Dict:
        """Cached order status lookup."""
        # This would normally hit an external API
        return self._fetch_order_data(order_id)
    
    def batch_get_orders(self, order_ids: List[str]) -> Dict[str, Dict]:
        """Fetch multiple orders in a single API call."""
        # In a real system, this would make one API call for all orders
        results = {}
        for order_id in order_ids:
            results[order_id] = self.get_order_status(order_id)
        return results
    
    async def async_search_products(self, query: str, category: str = None) -> List[Dict]:
        """Async product search for better concurrency."""
        # Simulate async API call
        await asyncio.sleep(0.1)  # Simulated network delay
        return self.search_products(query, category)
    
    def smart_function_selection(self, user_intent: str, available_functions: List[str]) -> List[str]:
        """Pre-filter functions likely to be useful for better performance."""
        
        intent_keywords = {
            "order": ["get_order_status", "check_return_policy"],
            "search": ["search_products"],
            "return": ["check_return_policy", "initiate_return"],
            "product": ["search_products"]
        }
        
        relevant_functions = []
        for keyword, functions in intent_keywords.items():
            if keyword in user_intent.lower():
                relevant_functions.extend(functions)
        
        # If no specific intent detected, return all functions
        return relevant_functions if relevant_functions else available_functions

class OptimizedAgent:
    """Performance-optimized agent with smart function calling."""
    
    def __init__(self, openai_client, tools_instance):
        self.client = openai_client
        self.tools = tools_instance
        self.conversation_history = []
        
    def select_relevant_tools(self, user_message: str) -> List[Dict]:
        """Dynamically select which tools to make available."""
        
        # Analyze user intent to reduce tool options
        relevant_functions = self.tools.smart_function_selection(
            user_message, 
            [schema["function"]["name"] for schema in FUNCTION_SCHEMAS]
        )
        
        # Return only relevant schemas
        return [schema for schema in FUNCTION_SCHEMAS 
                if schema["function"]["name"] in relevant_functions]
    
    async def process_message_async(self, user_message: str) -> str:
        """Async message processing for better performance."""
        
        self.conversation_history.append({"role": "user", "content": user_message})
        
        # Select relevant tools
        relevant_tools = self.select_relevant_tools(user_message)
        
        try:
            response = self.client.chat.completions.create(
                model="gpt-4-1106-preview",
                messages=[
                    {"role": "system", "content": "You are a helpful customer service assistant."}
                ] + self.conversation_history,
                tools=relevant_tools,  # Only pass relevant tools
                tool_choice="auto"
            )
        except Exception as e:
            return f"I apologize, but I'm experiencing technical difficulties: {str(e)}"
        
        assistant_message = response.choices[0].message
        
        if assistant_message.tool_calls:
            # Process function calls (potentially in parallel)
            tasks = []
            for tool_call in assistant_message.tool_calls:
                task = self.execute_function_call_async(tool_call)
                tasks.append(task)
            
            # Execute all function calls concurrently
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # Add results to conversation history
            self.conversation_history.append({
                "role": "assistant",
                "content": assistant_message.content,
                "tool_calls": assistant_message.tool_calls
            })
            
            for tool_call, result in zip(assistant_message.tool_calls, results):
                if isinstance(result, Exception):
                    result = json.dumps({"error": str(result)})
                    
                self.conversation_history.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result
                })
            
            # Get final response
            final_response = self.client.chat.completions.create(
                model="gpt-4-1106-preview",
                messages=self.conversation_history
            )
            
            return final_response.choices[0].message.content
        
        return assistant_message.content
    
    async def execute_function_call_async(self, tool_call) -> str:
        """Async function execution."""
        function_name = tool_call.function.name
        function_args = json.loads(tool_call.function.arguments)
        
        # Check if async version exists
        if hasattr(self.tools, f"async_{function_name}"):
            result = await getattr(self.tools, f"async_{function_name}")(**function_args)
        else:
            # Fall back to sync version
            result = getattr(self.tools, function_name)(**function_args)
        
        return json.dumps(result)

Hands-On Exercise

Let's build a complete real-world system: a financial advisor chatbot that can analyze portfolios, get market data, and make investment recommendations.

import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta
import numpy as np

class FinancialAdvisorTools:
    """Tools for financial analysis and portfolio management."""
    
    def __init__(self):
        self.cache = {}
        
    def get_stock_price(self, symbol: str, period: str = "1d") -> Dict:
        """Get current and historical stock price data."""
        try:
            stock = yf.Ticker(symbol.upper())
            hist = stock.history(period=period)
            
            if hist.empty:
                return {"error": f"No data found for symbol {symbol}"}
            
            current_price = float(hist['Close'].iloc[-1])
            previous_close = float(hist['Close'].iloc[-2]) if len(hist) > 1 else current_price
            change = current_price - previous_close
            change_percent = (change / previous_close) * 100 if previous_close != 0 else 0
            
            return {
                "symbol": symbol.upper(),
                "current_price": round(current_price, 2),
                "previous_close": round(previous_close, 2),
                "change": round(change, 2),
                "change_percent": round(change_percent, 2),
                "period": period,
                "last_updated": datetime.now().isoformat()
            }
        except Exception as e:
            return {"error": f"Failed to fetch data for {symbol}: {str(e)}"}
    
    def analyze_portfolio(self, holdings: List[Dict]) -> Dict:
        """Analyze a portfolio of stock holdings."""
        try:
            total_value = 0
            total_cost_basis = 0
            portfolio_data = []
            
            for holding in holdings:
                symbol = holding["symbol"]
                shares = holding["shares"] 
                cost_basis = holding.get("cost_basis", 0)
                
                price_data = self.get_stock_price(symbol)
                if "error" in price_data:
                    continue
                
                current_value = shares * price_data["current_price"]
                total_invested = shares * cost_basis if cost_basis > 0 else 0
                gain_loss = current_value - total_invested if total_invested > 0 else 0
                gain_loss_percent = (gain_loss / total_invested * 100) if total_invested > 0 else 0
                
                portfolio_data.append({
                    "symbol": symbol,
                    "shares": shares,
                    "current_price": price_data["current_price"],
                    "current_value": round(current_value, 2),
                    "cost_basis": cost_basis,
                    "total_invested": round(total_invested, 2),
                    "gain_loss": round(gain_loss, 2),
                    "gain_loss_percent": round(gain_loss_percent, 2),
                    "portfolio_weight": 0  # Will calculate after getting total
                })
                
                total_value += current_value
                total_cost_basis += total_invested
            
            # Calculate portfolio weights
            for holding in portfolio_data:
                holding["portfolio_weight"] = round(
                    (holding["current_value"] / total_value) * 100, 2
                ) if total_value > 0 else 0
            
            total_gain_loss = total_value - total_cost_basis
            total_gain_loss_percent = (total_gain_loss / total_cost_basis * 100) if total_cost_basis > 0 else 0
            
            return {
                "total_portfolio_value": round(total_value, 2),
                "total_cost_basis": round(total_cost_basis, 2),
                "total_gain_loss": round(total_gain_loss, 2),
                "total_gain_loss_percent": round(total_gain_loss_percent, 2),
                "holdings": portfolio_data,
                "analysis_date": datetime.now().isoformat()
            }
            
        except Exception as e:
            return {"error": f"Portfolio analysis failed: {str(e)}"}
    
    def get_diversification_advice(self, holdings: List[Dict]) -> Dict:
        """Provide diversification recommendations based on portfolio."""
        try:
            analysis = self.analyze_portfolio(holdings)
            if "error" in analysis:
                return analysis
            
            recommendations = []
            
            # Check for concentration risk
            for holding in analysis["holdings"]:
                if holding["portfolio_weight"] > 20:
                    recommendations.append({
                        "type": "concentration_risk",
                        "symbol": holding["symbol"],
                        "message": f"{holding['symbol']} represents {holding['portfolio_weight']}% of your portfolio. Consider reducing this position to below 20%."
                    })
            
            # Check for sector diversification (simplified)
            sectors = {
                "AAPL": "Technology", "MSFT": "Technology", "GOOGL": "Technology",
                "JPM": "Financial", "BAC": "Financial", "WFC": "Financial",
                "JNJ": "Healthcare", "PFE": "Healthcare", "UNH": "Healthcare",
                "XOM": "Energy", "CVX": "Energy"
            }
            
            sector_weights = {}
            for holding in analysis["holdings"]:
                sector = sectors.get(holding["symbol"], "Other")
                sector_weights[sector] = sector_weights.get(sector, 0) + holding["portfolio_weight"]
            
            for sector, weight in sector_weights.items():
                if weight > 30 and sector != "Other":
                    recommendations.append({
                        "type": "sector_concentration",
                        "sector": sector,
                        "message": f"Your {sector} allocation is {weight:.1f}%. Consider diversifying into other sectors."
                    })
            
            # Suggest additions for small portfolios
            if len(analysis["holdings"]) < 5:
                recommendations.append({
                    "type": "diversification",
                    "message": "Consider adding more holdings to improve diversification. A portfolio of 10-20 stocks across different sectors provides good diversification."
                })
            
            return {
                "current_diversification": sector_weights,
                "recommendations": recommendations,
                "diversification_score": min(100, len(analysis["holdings"]) * 10),  # Simplified score
                "analysis_date": datetime.now().isoformat()
            }
            
        except Exception as e:
            return {"error": f"Diversification analysis failed: {str(e)}"}

# Define function schemas for financial tools
FINANCIAL_SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "Get current stock price and recent performance data for a given ticker symbol",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "Stock ticker symbol (e.g., AAPL, MSFT, GOOGL)"
                    },
                    "period": {
                        "type": "string",
                        "enum": ["1d", "5d", "1mo", "3mo", "6mo", "1y"],
                        "description": "Time period for historical data (default: 1d)"
                    }
                },
                "required": ["symbol"]
            }
        }
    },
    {
        "type": "function", 
        "function": {
            "name": "analyze_portfolio",
            "description": "Analyze a portfolio of stock holdings including current values, gains/losses, and allocation weights",
            "parameters": {
                "type": "object",
                "properties": {
                    "holdings": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "symbol": {"type": "string", "description": "Stock ticker symbol"},
                                "shares": {"type": "number", "description": "Number of shares owned"},
                                "cost_basis": {"type": "number", "description": "Average cost per share (optional)"}
                            },
                            "required": ["symbol", "shares"]
                        },
                        "description": "List of stock holdings in the portfolio"
                    }
                },
                "required": ["holdings"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_diversification_advice", 
            "description": "Get personalized diversification recommendations based on current portfolio holdings",
            "parameters": {
                "type": "object",
                "properties": {
                    "holdings": {
                        "type": "array",
                        "items": {
                            "type": "object", 
                            "properties": {
                                "symbol": {"type": "string"},
                                "shares": {"type": "number"},
                                "cost_basis": {"type": "number"}
                            },
                            "required": ["symbol", "shares"]
                        }
                    }
                },
                "required": ["holdings"]
            }
        }
    }
]

# Your task: Complete this financial advisor agent
class FinancialAdvisorAgent:
    def __init__(self, openai_client):
        self.client = openai_client
        self.tools = FinancialAdvisorTools()
        self.conversation_history = []
    
    def process_message(self, user_message: str) -> str:
        """
        Complete this method to:
        1. Handle the user message
        2. Make appropriate function calls
        3. Provide comprehensive financial advice
        4. Maintain conversation context
        """
        # TODO: Implement the agent logic
        pass

# Test your implementation
if __name__ == "__main__":
    # Example portfolio for testing
    test_portfolio = [
        {"symbol": "AAPL", "shares": 10, "cost_basis": 150},
        {"symbol": "MSFT", "shares": 5, "cost_basis": 300},
        {"symbol": "GOOGL", "shares": 2, "cost_basis": 2500},
        {"symbol": "JPM", "shares": 8, "cost_basis": 140}
    ]
    
    client = openai.OpenAI(api_key="your-api-key")  # Replace with your key
    advisor = FinancialAdvisorAgent(client)
    
    # Test queries
    queries = [
        f"Can you analyze my portfolio: {test_portfolio}",
        "What's the current price of Tesla stock?",
        "Should I diversify my holdings more?"
    ]
    
    for query in queries:
        print(f"\nUser: {query}")
        response = advisor.process_message(query)
        print(f"Advisor: {response}")

Your Challenge: Complete the FinancialAdvisorAgent.process_message() method using the patterns we've covered. Your implementation should:

• Properly handle function calling flow
• Provide contextual financial advice
• Handle errors gracefully
• Maintain conversation history

Common Mistakes & Troubleshooting

Function Schema Issues:

The most common problems stem from poorly designed schemas:

# ❌ Common mistakes
bad_schema = {
    "name": "update_data",  # Too vague
    "description": "Updates some data",  # No context
    "parameters": {
        "type": "object",
        "properties": {
            "data": {"type": "string"}  # No format specification
        }
    }
}

# ✅ Better approach
good_schema = {
    "name": "update_customer_shipping_address", 
    "description": "Update the shipping address for a specific customer order before it ships",
    "parameters": {
        "type": "object",
        "properties": {
            "order_id": {
                "type": "string",
                "pattern": "^[0-9]{5}$",  # Specific format
                "description": "5-digit order ID"
            },
            "address": {
                "type": "object",
                "properties": {
                    "street": {"type": "string", "maxLength": 100},
                    "city": {"type": "string", "maxLength": 50},
                    "state": {"type": "string", "pattern": "^[A-Z]{2}$"},
                    "zip_code": {"type": "string", "pattern": "^[0-9]{5}(-[0-9]{4})?$"}
                },
                "required": ["street", "city", "state", "zip_code"]
            }
        },
        "required": ["order_id", "address"]
    }
}

Parameter Extraction Issues:

LLMs sometimes struggle with parameter extraction. Add examples and constraints:

# ❌ Problematic - LLM might extract wrong date format
{
    "start_date": {
        "type": "string",
        "description": "Start date"
    }
}

# ✅ Clear expectations
{
    "start_date": {
        "type": "string",
        "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}$",
        "description": "Start date in YYYY-MM-DD format (e.g., '2024-01-15'). Use today's date if user says 'today' or 'now'."
    }
}

Infinite Loop Prevention:

def safe_process_message(self, user_message: str, max_iterations: int = 3) -> str:
    """Prevent infinite function calling loops."""
    
    for iteration in range(max_iterations):
        response = self.client.chat.completions.create(
            model="gpt-4-1106-preview",
            messages=self.conversation_history + [{"role": "user", "content": user_message}],
            tools=FUNCTION_SCHEMAS
        )
        
        if not response.choices[0].message.tool_calls:
            return response.choices[0].message.content
            
        # Execute function calls and continue...
        # (Add your function execution logic here)
        
    return "I need to research this further. Can you please rephrase your question?"

Debugging Function Calls:

Add comprehensive logging to troubleshoot issues:

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def debug_function_call(self, tool_call):
    """Debug function calling issues."""
    
    logger.info(f"Function call attempted: {tool_call.function.name}")
    logger.info(f"Raw arguments: {tool_call.function.arguments}")
    
    try:
        args = json.loads(tool_call.function.arguments)
        logger.info(f"Parsed arguments: {args}")
    except json.JSONDecodeError as e:
        logger.error(f"JSON parsing failed: {e}")
        return {"error": "Invalid JSON in function arguments"}
    
    # Function execution with detailed logging
    try:
        result = getattr(self.tools, tool_call.function.name)(**args)
        logger.info(f"Function result: {result}")
        return result
    except Exception as e:
        logger.error(f"Function execution failed: {e}")
        logger.error(f"Function: {tool_call.function.name}, Args: {args}")
        raise

Warning: Always test your function schemas with various input formats. LLMs can be creative in how they interpret parameter requirements.

Summary & Next Steps

You've now learned how to transform LLMs from simple text generators into powerful agents that can interact with external systems. Function calling enables LLMs to:

• Access real-time data from APIs and databases
• Perform complex multi-step workflows
• Take actions in external systems
• Provide personalized responses based on user data

Key takeaways:

1. Schema Design is Critical: Well-designed function schemas are the foundation of reliable tool use
2. Error Handling is Essential: External systems fail, so build robust error handling and fallback strategies
3. Security Cannot be Overlooked: Validate inputs, implement access controls, and audit all actions
4. Performance Matters: Use caching, batching, and async patterns for production systems

Next Steps for Your Learning Journey:

• Explore Multi-Modal Tools: Learn how to give LLMs access to image processing, document analysis, and web browsing capabilities
• Build Agent Workflows: Study frameworks like LangChain Agents and AutoGPT for complex multi-step reasoning
• Production Deployment: Learn about scaling function calling systems with proper monitoring, rate limiting, and error tracking
• Advanced Patterns: Investigate dynamic tool generation, tool composition, and self-improving agent systems

Practice Project Ideas:

1. Personal Finance Manager: Build an agent that can analyze bank statements, track spending, and provide budget recommendations
2. Code Review Assistant: Create a tool that can access Git repositories, analyze code quality, and suggest improvements
3. Travel Planning Agent: Develop a system that can check flight prices, book hotels, and create detailed itineraries
4. Business Intelligence Bot: Build an agent that can query databases, generate reports, and answer complex business questions

The combination of LLMs and function calling opens up endless possibilities for automation and intelligent assistance. Start with simple tools and gradually build more sophisticated systems as you gain experience with the patterns and challenges involved.