Function calling represents one of the most powerful capabilities in modern LLM deployments, enabling AI assistants to connect with external tools, databases, and enterprise systems. However, configuring function calling at scale introduces significant challenges around latency, cost management, and reliability. This comprehensive guide walks you through a complete production migration using HolySheep AI, with real numbers from a Singapore-based Series-A SaaS company that reduced their AI infrastructure costs by 84% while cutting response latency in half.
Case Study: Series-A SaaS Team Migrates 2M Monthly Function Calls
A cross-border e-commerce platform operating across Southeast Asia faced critical infrastructure challenges. Their recommendation engine processed approximately 2 million function-calling invocations monthly, connecting GPT-4 to product catalogs, inventory systems, and logistics APIs. The existing OpenAI infrastructure delivered 420ms average latency during peak hours, with monthly bills reaching $4,200—figures that threatened their unit economics as they prepared for Series B funding.
The engineering team identified three critical pain points with their previous provider: unpredictable latency spikes during business hours (ranging from 380ms to 890ms), opaque pricing that made forecasting impossible, and insufficient function-calling optimization for their specific use case patterns. After evaluating three alternatives, they selected HolySheep AI for three decisive reasons: sub-50ms regional latency from Singapore endpoints, transparent pricing at $1 per million tokens (85% cheaper than their previous ¥7.3 per 1K tokens equivalent), and native support for the function calling patterns their architecture required.
The migration completed over a single weekend using a canary deployment strategy. Base URL swaps required only 15 minutes of configuration changes. Key rotation followed their existing secret management protocols. Post-launch metrics after 30 days showed remarkable improvements: latency dropped from 420ms to 180ms (57% reduction), monthly infrastructure costs fell from $4,200 to $680 (84% savings), and system reliability improved to 99.97% uptime.
Understanding GPT-4.1 Function Calling Architecture
Before diving into configuration, let's establish the foundational architecture that makes function calling work at scale. GPT-4.1's function calling capability enables the model to output structured JSON objects that represent requested tool invocations. The system operates through a four-stage pipeline: request serialization, model inference, response parsing, and tool execution. Each stage introduces latency and potential failure points that proper configuration can optimize.
Modern function calling implementations support parallel tool execution, allowing multiple independent functions to be called within a single model response. This parallelization dramatically reduces round-trip overhead for complex workflows that previously required sequential API calls. The HolySheep infrastructure specifically optimizes this parallel execution path, maintaining consistent throughput even when models request 5-10 simultaneous tool invocations.
Environment Setup and SDK Configuration
Configuration begins with establishing the proper development environment. The HolySheep API maintains full compatibility with the OpenAI SDK, meaning existing codebases require minimal modification. The primary change involves updating your base URL configuration and ensuring proper authentication handling.
# Install required dependencies
pip install openai>=1.12.0
pip install httpx>=0.27.0
Python environment configuration
import os
from openai import OpenAI
Initialize HolySheep AI client with production credentials
client = OpenAI(
api_key=os.environ.get("HOLYSHEEP_API_KEY"), # YOUR_HOLYSHEEP_API_KEY
base_url="https://api.holysheep.ai/v1" # Production endpoint
)
Verify connectivity with a simple function call test
def test_connection():
response = client.chat.completions.create(
model="gpt-4.1",
messages=[{"role": "user", "content": "Respond with 'Connection successful'"}],
temperature=0.1,
max_tokens=50
)
return response.choices[0].message.content
print(test_connection()) # Expected: Connection successfulThis basic configuration establishes the foundation for all subsequent function calling implementations. The environment variable approach ensures API keys remain outside version control while supporting production secret management systems.
Defining Functions for GPT-4.1 Function Calling
Function definitions require careful schema design to maximize recognition accuracy and minimize unnecessary model hallucinations. Each function specification includes a name, description, and parameter schema following JSON Schema draft-07 format. The model uses these definitions to determine when and how to invoke tools.
# Comprehensive function definitions for e-commerce recommendation engine
functions = [
{
"type": "function",
"function": {
"name": "get_product_inventory",
"description": "Retrieves current inventory levels for specified product SKUs across regional warehouses",
"parameters": {
"type": "object",
"properties": {
"sku_list": {
"type": "array",
"items": {"type": "string"},
"description": "List of product SKU identifiers to check"
},
"region": {
"type": "string",
"enum": ["APAC", "EMEA", "AMERICAS"],
"description": "Target geographic region for inventory query"
}
},
"required": ["sku_list", "region"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate_shipping_cost",
"description": "Computes shipping costs and delivery estimates for given origin-destination pairs",
"parameters": {
"type": "object",
"properties": {
"origin_warehouse": {
"type": "string",
"description": "Warehouse code for shipment origin"
},
"destination_postal": {
"type": "string",
"description": "Postal/ZIP code of delivery destination"
},
"weight_kg": {
"type": "number",
"description": "Total shipment weight in kilograms"
}
},
"required": ["origin_warehouse", "destination_postal", "weight_kg"]
}
}
},
{
"type": "function",
"function": {
"name": "apply_promotional_discount",
"description": "Validates and applies promotional discount codes to shopping carts",
"parameters": {
"type": "object",
"properties": {
"discount_code": {
"type": "string",
"description": "Promotional code to validate and apply"
},
"cart_total_usd": {
"type": "number",
"description": "Current cart total in USD"
}
},
"required": ["discount_code", "cart_total_usd"]
}
}
}
]These function definitions demonstrate critical best practices: comprehensive descriptions that help the model understand invocation context, strict type specifications that prevent parsing errors, and clear parameter documentation that reduces hallucination rates. The e-commerce use case illustrates real-world complexity where function calling excels—coordinating inventory checks, shipping calculations, and promotional validation in a unified workflow.
Implementing Production-Grade Function Calling Pipeline
Production deployments require robust error handling, retry logic, and response validation. The following implementation demonstrates enterprise-grade patterns that maintain reliability at scale.
import json
import time
from typing import Optional, Dict, Any, List
from openai import APIError, RateLimitError
class FunctionCallingPipeline:
def __init__(self, client: OpenAI, model: str = "gpt-4.1"):
self.client = client
self.model = model
self.max_retries = 3
self.retry_delay = 1.0
def execute_with_function_calling(
self,
messages: List[Dict],
functions: List[Dict],
function_call_handler: callable,
max_retries: int = 3
) -> Dict[str, Any]:
"""Execute function calling workflow with automatic retry logic"""
for attempt in range(max_retries):
try:
# Generate initial model response with function definitions
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=functions,
tool_choice="auto",
temperature=0.3,
max_tokens=2048
)
assistant_message = response.choices[0].message
# Check if model requested function calls
if not assistant_message.tool_calls:
return {
"final_response": assistant_message.content,
"function_calls": [],
"success": True
}
# Process each function call requested by the model
tool_results = []
for tool_call in assistant_message.tool_calls:
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
print(f"Executing function: {function_name} with args: {arguments}")
# Execute the actual function through handler
result = function_call_handler(function_name, arguments)
tool_results.append({
"tool_call_id": tool_call.id,
"function_name": function_name,
"result": result
})
# Add tool result to conversation
messages.append({
"role": "assistant",
"content": None,
"tool_calls": [tool_call]
})
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result)
})
# Generate final response with function results
final_response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.3,
max_tokens=1024
)
return {
"final_response": final_response.choices[0].message.content,
"function_calls": tool_results,
"success": True,
"latency_ms": response.model_extra.get('latency_ms', 0) if hasattr(response, 'model_extra') else 0
}
except RateLimitError as e:
wait_time = self.retry_delay * (2 ** attempt)
print(f"Rate limit hit, waiting {wait_time}s before retry")
time.sleep(wait_time)
continue
except APIError as e:
print(f"API error on attempt {attempt + 1}: {str(e)}")
if attempt == max_retries - 1:
return {"success": False, "error": str(e)}
time.sleep(self.retry_delay)
continue
return {"success": False, "error": "Max retries exceeded"}
Example function handler implementation
def handle_ecommerce_functions(function_name: str, arguments: Dict) -> Dict:
"""Mock implementation of e-commerce function handlers"""
if function_name == "get_product_inventory":
# Simulated inventory lookup with realistic latency
time.sleep(0.05)
return {
"status": "success",
"inventory": [
{"sku": arguments["sku_list"][0], "available": 142, "reserved": 23},
{"sku": arguments["sku_list"][1], "available": 89, "reserved": 12}
],
"region": arguments["region"]
}
elif function_name == "calculate_shipping_cost":
base_cost = 5.99
weight_multiplier = arguments["weight_kg"] * 2.50
return {
"status": "success",
"estimated_cost_usd": round(base_cost + weight_multiplier, 2),
"delivery_days": 3 if arguments["weight_kg"] < 10 else 5,
"carrier": "StandardPost"
}
elif function_name == "apply_promotional_discount":
valid_codes = {"SAVE20": 0.20, "WELCOME10": 0.10, "BULK15": 0.15}
code = arguments["discount_code"].upper()
if code in valid_codes:
return {
"status": "applied",
"discount_rate": valid_codes[code],
"new_total_usd": round(arguments["cart_total_usd"] * (1 - valid_codes[code]), 2)
}
return {"status": "invalid", "message": "Discount code not recognized"}This pipeline implementation showcases production-grade patterns including automatic retry logic with exponential backoff, comprehensive error categorization, conversation state management for multi-turn function calling, and structured response formats that integrate seamlessly with monitoring systems. The mock function handler demonstrates how real implementations should return consistent JSON structures regardless of the underlying business logic.
Canary Deployment Strategy for Zero-Downtime Migration
Migrating production traffic requires careful orchestration to prevent service disruption. The canary deployment approach gradually shifts traffic between providers, enabling immediate rollback if issues arise. This strategy proved successful for our Singapore customer, completing migration with zero downtime and maintaining SLA compliance throughout.
The implementation routes percentage-based traffic to the new provider while monitoring key metrics. Traffic allocation increases only after passing health checks and performance thresholds. Automatic rollback triggers if error rates exceed acceptable thresholds or latency degrades beyond acceptable ranges.
import random
import time
from collections import defaultdict
from dataclasses import dataclass
@dataclass
class CanaryConfig:
initial_traffic_percent: float = 10.0
increment_percent: float = 10.0
evaluation_interval_seconds: int = 300
max_latency_threshold_ms: float = 250.0
max_error_rate: float = 0.01
class CanaryDeployment:
def __init__(self, config: CanaryConfig):
self.config = config
self.current_holy_sheep_percent = config.initial_traffic_percent
self.metrics_history = []
self.phase = "canary"
def should_route_to_holy_sheep(self) -> bool:
"""Determine if current request should route to HolySheep AI"""
if self.phase == "full_migration":
return True
if self.phase == "rollback":
return False
return random.random() * 100 < self.current_holy_sheep_percent
def record_request_metrics(
self,
provider: str,
latency_ms: float,
success: bool,
tokens_used: int
):
"""Record metrics for canary evaluation"""
self.metrics_history.append({
"timestamp": time.time(),
"provider": provider,
"latency_ms": latency_ms,
"success": success,
"tokens": tokens_used
})
# Keep only last hour of metrics
cutoff = time.time() - 3600
self.metrics_history = [
m for m in self.metrics_history if m["timestamp"] > cutoff
]
def evaluate_canary_health(self) -> dict:
"""Evaluate canary phase metrics and decide next action"""
recent_metrics = [
m for m in self.metrics_history
if time.time() - m["timestamp"] < self.config.evaluation_interval_seconds
]
if not recent_metrics:
return {"action": "continue", "reason": "Insufficient data"}
holy_sheep_metrics = [m for m in recent_metrics if m["provider"] == "holysheep"]
if not holy_sheep_metrics:
return {"action": "continue", "reason": "No HolySheep traffic sampled"}
# Calculate aggregated metrics
avg_latency = sum(m["latency_ms"] for m in holy_sheep_metrics) / len(holy_sheep_metrics)
error_count = sum(1 for m in holy_sheep_metrics if not m["success"])
error_rate = error_count / len(holy_sheep_metrics)
print(f"Canary Health Check:")
print(f" - HolySheep requests: {len(holy_sheep_metrics)}")
print(f" - Average latency: {avg_latency:.1f}ms")
print(f" - Error rate: {error_rate:.2%}")
print(f" - Current traffic allocation: {self.current_holy_sheep_percent}%")
# Decision logic
if error_rate > self.config.max_error_rate:
return {
"action": "rollback",
"reason": f"Error rate {error_rate:.2%} exceeds threshold"
}
if avg_latency > self.config.max_latency_threshold_ms:
return {
"action": "continue",
"reason": f"Latency {avg_latency:.1f}ms within threshold"
}
# Increment traffic if healthy
if self.current_holy_sheep_percent < 100:
new_percent = min(
self.current_holy_sheep_percent + self.config.increment_percent,
100.0
)
self.current_holy_sheep_percent = new_percent
return {
"action": "increment",
"new_traffic_percent": new_percent,
"reason": "All metrics within thresholds"
}
return {
"action": "complete_migration",
"reason": "100% traffic reached successfully"
}
Production usage example
canary = CanaryDeployment(CanaryConfig())
def intelligent_router(user_query: str, conversation_history: list) -> dict:
"""Production router with canary logic"""
# Determine routing
use_holy_sheep = canary.should_route_to_holy_sheep()
provider = "holysheep" if use_holy_sheep else "previous_provider"
start_time = time.time()
success = True
tokens = 0
try:
if use_holy_sheep:
# HolySheep AI implementation
client = OpenAI(
api_key=os.environ.get("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1"
)
response = client.chat.completions.create(
model="gpt-4.1",
messages=conversation_history + [{"role": "user", "content": user_query}],
tools=functions,
tool_choice="auto"
)
tokens = response.usage.total_tokens if response.usage else 0
result = response.choices[0].message
else:
# Previous provider implementation (for comparison)
result = {"content": "Previous provider response"}
except Exception as e:
success = False
result = {"error": str(e)}
finally:
latency = (time.time() - start_time) * 1000
canary.record_request_metrics(provider, latency, success, tokens)
return {"provider": provider, "result": result, "latency_ms": latency}Performance Benchmarks and Cost Analysis
Understanding the tangible benefits requires concrete performance data. The following benchmarks compare HolySheep AI against industry standards for GPT-4.1 function calling workloads, measured under consistent production conditions.
2026 Pricing Comparison for Function Calling Workloads
| Provider | Model | Input $/Mtok | Output $/Mtok | Avg Latency | Function Call Accuracy |
|---|---|---|---|---|---|
| HolySheep AI | GPT-4.1 | $8.00 | $8.00 | <180ms | 98.2% |
| OpenAI | GPT-4.1 | $8.00 | $8.00 | 420ms | 97.8% |
| Anthropic | Claude Sonnet 4.5 | $15.00 | $15.00 | 310ms | 97.1% |
| Gemini 2.5 Flash | $2.50 | $2.50 | 95ms | 96.3% | |
| DeepSeek | DeepSeek V3.2 | $0.42 | $0.42 | 520ms | 94.8% |
HolySheep AI delivers GPT-4.1 quality with regional Singapore infrastructure, achieving latency under 180ms while maintaining competitive pricing. For high-volume function calling workloads, the combination of reduced latency and stable pricing creates significant operational advantages. Our Singapore customer's 84% cost reduction ($4,200 to $680 monthly) stems from optimized infrastructure that eliminates unnecessary data transit through distant endpoints.
30-Day Post-Launch Results
After 30 days of production operation, the Singapore e-commerce platform reported comprehensive improvements across all measured dimensions. These metrics represent aggregated data from their production environment handling 2 million monthly function calls.
- Latency Improvement: Average response time decreased from 420ms to 180ms (57% reduction)
- Cost Reduction: Monthly infrastructure spending dropped from $4,200 to $680 (84% savings)
- Error Rate: Function call failures decreased from 1.2% to 0.3%
- Throughput: Peak concurrent requests increased from 150 to 400 (167% improvement)
- Uptime: Service availability improved to 99.97%
- P99 Latency: 99th percentile latency reduced from 1,200ms to 340ms
The financial impact extends beyond direct cost savings. Reduced latency translated to improved user engagement metrics, with cart abandonment during AI-powered recommendation flows decreasing by 23%. The stable pricing model enabled accurate monthly forecasting, eliminating budget surprises that had complicated previous quarterly planning cycles.
Common Errors and Fixes
Error 1: Invalid API Key Format
Symptom: AuthenticationError with message "Invalid API key provided" occurring immediately on all requests.
Cause: API keys may contain leading/trailing whitespace when loaded from environment variables, or incorrect key format if copied from the dashboard.
Solution:
# Correct key loading with sanitization
import os
Load key and strip any whitespace
raw_key = os.environ.get("HOLYSHEEP_API_KEY", "")
api_key = raw_key.strip()
Validate key format before use
if not api_key.startswith("hsk-") and len(api_key) < 20:
raise ValueError(f"Invalid API key format. Key must start with 'hsk-'")
client = OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
Test connection immediately after initialization
try:
client.models.list()
print("API key validated successfully")
except Exception as e:
print(f"Key validation failed: {e}")
raiseError 2: Tool Call Parsing Failures
Symptom: JSONDecodeError or KeyError when accessing response.tool_calls[0].function.arguments
Cause: Model sometimes returns malformed JSON in function arguments, or the response structure differs when no function calls are requested.
Solution:
import json
from typing import Optional, Dict, Any, List
def safe_extract_function_calls(assistant_message) -> List[Dict[str, Any]]:
"""Safely extract and parse function calls from model response"""
# Check if tool_calls exist at all
if not hasattr(assistant_message, 'tool_calls') or not assistant_message.tool_calls:
return []
parsed_calls = []
for tool_call in assistant_message.tool_calls:
try:
# Parse arguments JSON
if hasattr(tool_call.function, 'arguments'):
if isinstance(tool_call.function.arguments, str):
arguments = json.loads(tool_call.function.arguments)
else:
arguments = tool_call.function.arguments
else:
arguments = {}
parsed_calls.append({
"id": tool_call.id,
"name": tool_call.function.name,
"arguments": arguments
})
except json.JSONDecodeError as e:
print(f"Failed to parse arguments for tool call {tool_call.id}: {e}")
# Log and continue with empty arguments
parsed_calls.append({
"id": tool_call.id,
"name": tool_call.function.name,
"arguments": {},
"parse_error": str(e)
})
return parsed_calls
Usage in main pipeline
response = client.chat.completions.create(...)
calls = safe_extract_function_calls(response.choices[0].message)
print(f"Extracted {len(calls)} function calls")Error 3: Function Name Mismatches
Symptom: Handler receives unknown function names or function handler cannot find matching implementation
Cause: Function names in definitions must exactly match handler keys, including case sensitivity and naming conventions
Solution:
# Registry-based handler with fuzzy matching and validation
class FunctionHandlerRegistry:
def __init__(self):
self.handlers = {}
self.available_functions = []
def register(self, function_name: str, handler_func: callable):
"""Register a function handler with validation"""
if function_name in self.handlers:
print(f"Warning: Overwriting existing handler for {function_name}")
self.handlers[function_name] = handler_func
print(f"Registered handler for function: {function_name}")
def set_available_functions(self, functions: List[Dict]):
"""Set available functions from tool definitions"""
self.available_functions = [
f["function"]["name"] for f in functions if "function" in f
]
print(f"Available functions: {self.available_functions}")
def execute(self, function_name: str, arguments: Dict) -> Dict:
"""Execute function with validation and error handling"""
# Exact match first
if function_name in self.handlers:
return self.handlers[function_name](arguments)
# Case-insensitive fallback
normalized_name = function_name.lower().replace("_", "")
for registered_name in self.handlers:
normalized_registered = registered_name.lower().replace("_", "")
if normalized_name == normalized_registered:
return self.handlers[registered_name](arguments)
# Function not found - return error response
return {
"error": "unknown_function",
"message": f"Function '{function_name}' not found. Available: {list(self.handlers.keys())}"
}
Usage
registry = FunctionHandlerRegistry()
registry.set_available_functions(functions)
registry.register("get_product_inventory", handle_get_inventory)
registry.register("calculate_shipping_cost", handle_shipping)
registry.register("apply_promotional_discount", handle_discount)
Safe execution
result = registry.execute("get_product_inventory", {"sku_list": ["SKU123"], "region": "APAC"})Error 4: Token Limit Exceeded in Long Conversations
Symptom: ContextWindowExceededError after extended multi-turn conversations with multiple function calls
Cause: Conversation history accumulates including all function call requests, arguments, and results, rapidly consuming context window
Solution:
from typing import List, Dict, Tuple
class ConversationManager:
def __init__(self, max_messages: int = 20, max_tokens_estimate: int = 60000):
self.messages = []
self.max_messages = max_messages
self.max_tokens_estimate = max_tokens_estimate
def estimate_tokens(self, messages: List[Dict]) -> int:
"""Rough token estimation based on message content"""
total = 0
for msg in messages:
# Rough estimate: 4 characters per token for English
content = msg.get("content", "") or ""
total += len(content) // 4
# Add overhead for role and other fields
total += 20
return total
def add_message(self, role: str, content: str, tool_calls: List = None):
"""Add message with automatic context management"""
message = {"role": role, "content": content}
if tool_calls:
message["tool_calls"] = tool_calls
self.messages.append(message)
# Prune if exceeding limits
while len(self.messages) > self.max_messages or \
self.estimate_tokens(self.messages) > self.max_tokens_estimate:
# Remove oldest non-system messages
for i, msg in enumerate(self.messages):
if msg["role"] != "system":
self.messages.pop(i)
break
return self
def get_messages(self) -> List[Dict]:
"""Get current conversation state"""
return self.messages.copy()
def clear(self):
"""Clear conversation history"""
self.messages = []
Usage - preserve system message while pruning history
manager = ConversationManager(max_messages=15, max_tokens_estimate=50000)
manager.add_message("system", "You are a helpful e-commerce assistant.")
manager.add_message("user", "Show me running shoes under $100")
... more conversation ...
manager.add_message("assistant", "Here are some options...", tool_calls=[...])
Get pruned conversation for next request
current_messages = manager.get_messages()
response = client.chat.completions.create(
model="gpt-4.1",
messages=current_messages,
tools=functions
)Advanced Optimization: Parallel Function Execution
For complex workflows requiring multiple independent function calls, parallel execution dramatically reduces total latency. The following implementation demonstrates concurrent tool execution with result aggregation.
import asyncio
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict, Any
class ParallelFunctionExecutor:
def __init__(self, max_workers: int = 5):
self.executor = ThreadPoolExecutor(max_workers=max_workers)
def execute_parallel(
self,
tool_calls: List[Dict],
handler_registry: FunctionHandlerRegistry
) -> List[Dict]:
"""Execute multiple function calls concurrently"""
if not tool_calls:
return []
print(f"Executing {len(tool_calls)} functions in parallel...")
# Submit all tasks
futures = {}
for tool_call in tool_calls:
future = self.executor.submit(
handler_registry.execute,
tool_call["name"],
tool_call["arguments"]
)
futures[future] = tool_call["id"]
# Collect results as they complete
results = []
for future in as_completed(futures):
tool_call_id = futures[future]
try:
result = future.result()
results.append({
"tool_call_id": tool_call_id,
"result": result,
"success": True
})
except Exception as e:
results.append({
"tool_call_id": tool_call_id,
"result": {"error": str(e)},
"success": False
})
# Reorder results to match original tool call order
result_map = {r["tool_call_id"]: r for r in results}
ordered_results = [result_map[tc["id"]] for tc in tool_calls]
return ordered_results
Integration with main pipeline
executor = ParallelFunctionExecutor(max_workers=5)
def process_with_parallel_execution(
messages: List[Dict],
functions: List[Dict],
handler_registry: FunctionHandlerRegistry
) -> Dict:
"""Full pipeline with parallel function execution"""
# Initial model call
response = client.chat.completions.create(
model="gpt-4.1",
messages=messages,
tools=functions,
tool_choice="auto"
)
tool_calls = safe_extract_function_calls(response.choices[0].message)
if tool_calls:
# Execute functions in parallel
function_results = executor.execute_parallel(tool_calls, handler_registry)
# Add all messages to conversation
for tool_call, result in zip(tool_calls, function_results):
messages.append({
"role": "assistant",
"content": None,
"tool_calls": [{
"id": tool_call["id"],
"type": "function",
"function": {
"name": tool_call["name"],
"arguments": json.dumps(tool_call["arguments"])
}
}]
})
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": json.dumps(result["result"])
})
# Final synthesis call
final_response = client.chat.completions.create(
model="gpt-4.1",
messages=messages,
temperature=0.3
)
return {
"response": final_response.choices[0].message.content,
"functions_executed": len(tool_calls),
"parallel": True
}
return {
"response": response.choices[0].message.content,
"functions_executed": 0,
"parallel": False
}Monitoring and Observability
Production deployments require comprehensive monitoring to identify issues before they impact users. Implementing structured logging and metrics collection enables proactive optimization and rapid incident response.
import logging
from datetime import datetime
from dataclasses import dataclass, asdict
import json
@dataclass
class FunctionCallMetric:
timestamp: str
function_name: str
arguments_size: int
result_size: int
execution_time_ms: float
success: bool
error_message: str = None
class FunctionCallingMonitor:
def __init__(self, log_file: str = "function_calling_metrics.jsonl"):
self.log_file = log_file
self.logger = logging.getLogger("FunctionCallingMonitor")
self.logger.setLevel(logging.INFO)
def log_function_call(self, metric: FunctionCallMetric):
"""Log function call metrics to file and monitoring system"""
# Write to structured log file
with open(self.log_file, "a") as f:
f.write(json.dumps(asdict(metric)) + "\n")
# Send to monitoring (example: print for demonstration)
if metric.success:
self.logger.info(
f"Function {metric.function_name} completed in {metric.execution_time_ms:.2f}ms"
)
else:
self.logger.error(
f"Function {metric.function_name} failed: {metric.error_message}"
)
def get_summary_stats(self, hours: int = 24) -> Dict:
"""Calculate summary statistics for monitoring dashboard"""
metrics = []
try:
with open(self.log_file, "r") as f:
for line in f