Introduction
Large language models with function calling capabilities are essential for building intelligent applications that interact with external systems. Moving from prototype to production introduces new challenges around reliability, error handling, and predictability. Unlike simple text generation, function calling pipelines must handle partial successes, rate limits, authentication failures, and unexpected tool responses while maintaining audit trails and providing meaningful feedback to users. This article presents an original engineering framework for building production-ready LLM function calling pipelines with practical patterns that can be implemented immediately.
The patterns discussed here build on the workflow design principles outlined in "Choose a Workflow Before You Build an Autonomous Agent," which emphasizes defining clear output contracts and bounded decisions before adding autonomy. Just as that guide warns against treating autonomy as the default, we should treat function calling as a complex workflow requiring its own reliability engineering practices rather than a simple proxy for model input.
Input Validation and Schema Design
Before calling any LLM, validate the requested function parameters against predefined JSON schemas. This prevents malformed requests from reaching the model and provides immediate feedback to calling applications. A well-designed schema establishes clear contracts for expected inputs, reduces unnecessary model calls, and creates a foundation for robust error handling.
import jsonschema
from typing import Dict, Any
FUNCTION_SCHEMA = {
"type": "object",
"required": ["tool_name", "parameters"],
"properties": {
"tool_name": {"type": "string"},
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string"},
"limit": {"type": "number", "minimum": 1, "maximum": 100}
}
}
}
}
def validate_tool_request(request: Dict[str, Any]) -> bool:
try:
jsonschema.validate(request, FUNCTION_SCHEMA)
return True
except jsonschema.ValidationError as e:
raise ToolError(
f"Invalid request: {e.message}",
ErrorCategory.VALIDATION,
{"validation_error": e.message}
)
Validation should occur at the application boundary before LLM invocation. This ensures only well-formed requests are sent to the model, reducing downstream processing errors and optimizing resource usage. Schemas should be versioned and maintained alongside the application code. Schema evolution requires careful consideration: when adding new optional fields, ensure backward compatibility with existing clients. When removing fields, provide a deprecation period and migration path. Consider using tools like jsonschema with draft-07 or later for comprehensive validation including format checks, pattern matching, and conditional validation rules.
Beyond basic schema validation, implement semantic validation that checks business logic constraints. For example, a date range parameter should validate that start dates precede end dates, and numerical parameters should respect domain-specific limits beyond what JSON schema can express. This layer catches errors that schema validation misses and provides more actionable error messages to clients.
Schema Versioning Strategy
Maintain schema versions alongside application code. When deploying schema changes:
- Deprecate old schemas with a version migration period
- Support both old and new schemas during transition
- Add deprecation warnings for clients using outdated schemas
- Remove old schemas only after all clients have migrated
Retry Logic and Error Handling
Network issues, rate limits, and temporary service unavailability require intelligent retry strategies. Exponential backoff with jitter prevents thundering herd problems while respecting service constraints. Different error categories demand distinct handling approaches.
import time
import random
from enum import Enum
class ErrorCategory(Enum):
VALIDATION = "validation"
AUTHENTICATION = "authentication"
RATE_LIMIT = "rate_limit"
TOOL_ERROR = "tool_error"
NETWORK = "network"
TIMEOUT = "timeout"
UNKNOWN = "unknown"
class ToolError(Exception):
def __init__(self, message: str, category: ErrorCategory,
details: Dict[str, Any], recoverable: bool = False):
super().__init__(message)
self.category = category
self.details = details
self.recoverable = recoverable
class RetryManager:
def __init__(self, max_retries: int = 3, base_delay: float = 1.0,
jitter: float = 0.1):
self.max_retries = max_retries
self.base_delay = base_delay
self.jitter = jitter
self.failure_count = 0
self.last_failure_time = None
self.state = "CLOSED"
def execute_with_retry(self, func, *args, **kwargs):
for attempt in range(self.max_retries + 1):
if self.state == "OPEN":
if self.last_failure_time and \
time.time() - self.last_failure_time > 60:
self.state = "HALF_OPEN"
else:
raise ToolError(
"Circuit breaker is OPEN",
ErrorCategory.TOOL_ERROR,
{"state": "open"},
recoverable=False
)
try:
result = func(*args, **kwargs)
if self.state == "HALF_OPEN":
self.state = "CLOSED"
self.failure_count = 0
return result
except ToolError as e:
self.failure_count += 1
self.last_failure_time = time.time()
if e.recoverable and attempt < self.max_retries:
delay = self.base_delay * (2 ** attempt) + random.uniform(0, self.jitter)
time.sleep(delay)
continue
raise
except (ConnectionError, TimeoutError) as e:
self.failure_count += 1
self.last_failure_time = time.time()
if attempt < self.max_retries:
delay = self.base_delay * (2 ** attempt) + random.uniform(0, self.jitter)
time.sleep(delay)
continue
raise
raise ToolError(
"Max retries exceeded",
ErrorCategory.TOOL_ERROR,
{"max_retries": self.max_retries}
)
Implement circuit breakers to prevent cascading failures when tools become unavailable. Rate limiting protects both the LLM service and downstream systems from overload. Error categorization enables appropriate responses: validation errors trigger immediate feedback, authentication failures require credential review, rate limits necessitate backoff, and tool-specific errors demand context-aware handling.
Error Classification Taxonomy
Create a comprehensive error classification system that distinguishes between recoverable and non-recoverable errors. Recoverable errors include temporary network issues, rate limit responses, and service unavailability. Non-recoverable errors include invalid authentication, malformed requests, and business logic violations. This classification determines whether to retry, escalate, or fail fast.
The recoverable flag enables intelligent retry logic: only retry operations marked as recoverable, preventing infinite loops on persistent failures. Circuit breakers add another layer of protection by opening after consecutive failures, preventing further requests during service outages.
def classify_tool_error(error: ToolError) -> str:
messages = {
ErrorCategory.VALIDATION: (
"Invalid parameters provided. Please check your request."
),
ErrorCategory.AUTHENTICATION: (
"Authentication failed. Check your credentials and permissions."
),
ErrorCategory.RATE_LIMIT: (
"Rate limit exceeded. Wait before making another request."
),
ErrorCategory.TOOL_ERROR: (
f"Tool error: {error.details.get('message', 'Unknown error')}."
),
ErrorCategory.NETWORK: (
"Network connectivity issue. Check your connection."
),
ErrorCategory.TIMEOUT: (
"The request timed out. Try again with a simpler query."
),
ErrorCategory.UNKNOWN: (
"An unexpected error occurred. Try again or contact support."
)
}
return messages.get(error.category, messages[ErrorCategory.UNKNOWN])
Structured Output and Monitoring
LLM outputs must be parsed and validated against expected schemas using libraries like Pydantic for robust data validation. Track all function calls, responses, errors, and retry attempts for observability through comprehensive logging and monitoring.
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any
import logging
from datetime import datetime
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class ToolCall(BaseModel):
tool_name: str = Field(..., description="Name of the tool to call")
parameters: Dict[str, Any] = Field(default_factory=dict)
class FunctionCallResult(BaseModel):
tool_calls: List[ToolCall] = Field(
..., description="List of tool calls extracted from LLM response"
)
reasoning: Optional[str] = Field(None, description="Optional reasoning")
def parse_llm_response(response: str) -> FunctionCallResult:
try:
import json
parsed_data = json.loads(response)
return FunctionCallResult(**parsed_data)
except json.JSONDecodeError as e:
raise ToolError(
f"Failed to parse LLM response: {e}",
ErrorCategory.TOOL_ERROR,
{"parse_error": str(e)}
)
class FunctionCallLogger:
def __init__(self, correlation_id: str):
self.correlation_id = correlation_id
def log_call(self, tool_name: str, parameters: Dict[str, Any],
attempt: int, success: bool, error: Optional[str] = None,
latency: Optional[float] = None):
log_data = {
"correlation_id": self.correlation_id,
"timestamp": datetime.utcnow().isoformat() + "Z",
"tool_name": tool_name,
"parameters": self._sanitize_parameters(parameters),
"attempt": attempt,
"success": success,
"error": error,
"latency_ms": int(latency * 1000) if latency else None
}
logger.info(f"FUNCTION_CALL: {log_data}")
def _sanitize_parameters(self, params: Dict[str, Any]) -> Dict[str, Any]:
sensitive_keys = {"api_key", "password", "token", "secret",
"authorization", "credential"}
sanitized = {}
for key, value in params.items():
if key.lower() in sensitive_keys:
sanitized[key] = "[REDACTED]"
elif isinstance(value, str) and len(value) > 100:
sanitized[key] = value[:100] + "..."
else:
sanitized[key] = value
return sanitized
Monitoring should include metrics for success rates, error frequencies, latency measurements, and retry counts. Consider implementing distributed tracing using OpenTelemetry or similar frameworks to track requests across service boundaries, which is essential for debugging complex multi-tool workflows.
Distributed Tracing with OpenTelemetry
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)
def execute_tool_with_tracing(tool_name: str, parameters: Dict[str, Any]):
with tracer.start_as_current_span("tool_execution") as span:
span.set_attribute("tool.name", tool_name)
span.set_attribute("tool.parameters", str(parameters))
try:
result = execute_tool(tool_name, parameters)
span.set_attribute("tool.success", True)
return result
except Exception as e:
span.set_attribute("tool.success", False)
span.set_attribute("tool.error", str(e))
raise
Distributed tracing provides end-to-end visibility into function calling workflows, enabling teams to identify bottlenecks, trace failures across service boundaries, and understand the full request lifecycle.
Production Considerations
Circuit Breaker Pattern
Implement circuit breakers to prevent cascading failures when tools become unavailable. A circuit breaker monitors failure counts and opens the circuit after a threshold is reached, allowing time for recovery before attempting reconnection.
class CircuitBreaker:
def __init__(self, failure_threshold: int = 5, timeout: int = 60):
self.failure_threshold = failure_threshold
self.timeout = timeout
self.failure_count = 0
self.last_failure_time = None
self.state = "CLOSED"
def call(self, func, *args, **kwargs):
if self.state == "OPEN":
if time.time() - self.last_failure_time > self.timeout:
self.state = "HALF_OPEN"
else:
raise Exception("Circuit breaker is OPEN")
try:
result = func(*args, **kwargs)
self.failure_count = 0
self.state = "CLOSED"
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
raise e
Rate Limiting and Throttling
Protect both the LLM service and downstream systems from overload using rate limiting and throttling strategies. Implement token bucket algorithms or sliding window techniques to maintain service stability during traffic spikes.
import time
from collections import deque
from threading import Lock
class RateLimiter:
def __init__(self, capacity: int, refill_rate: float):
self.capacity = capacity
self.refill_rate = refill_rate
self.tokens = capacity
self.last_refill = time.time()
self.lock = Lock()
def acquire(self) -> bool:
with self.lock:
now = time.time()
elapsed = now - self.last_refill
self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
self.last_refill = now
if self.tokens >= 1:
self.tokens -= 1
return True
return False
Comprehensive Logging and Audit Trails
Track all function calls with timestamps, parameters, attempt numbers, success status, and error details. Store logs in a centralized system with appropriate retention policies. Include correlation IDs to trace requests across service boundaries, and ensure sensitive data is redacted before logging.
import uuid
import json
import hashlib
class AuditLogger:
def __init__(self, storage_backend):
self.storage = storage_backend
def log_function_call(self, tool_name: str, parameters: Dict[str, Any],
result: Any):
record = {
"correlation_id": str(uuid.uuid4()),
"timestamp": datetime.utcnow().isoformat() + "Z",
"tool_name": tool_name,
"parameters": self._sanitize(parameters),
"result_hash": hashlib.sha256(str(result).encode()).hexdigest()[:16]
}
self.storage.write(record)
def _sanitize(self, params: Dict[str, Any]) -> Dict[str, Any]:
sensitive_keys = {"api_key", "password", "token", "secret",
"authorization", "credential"}
return {k: "[REDACTED]" if k.lower() in sensitive_keys else v
for k, v in params.items()}
Conclusion
Building reliable LLM function calling pipelines requires a comprehensive approach combining input validation, retry logic, error categorization, structured output parsing, and comprehensive monitoring. By implementing these patterns, you can create production systems that handle real-world failures gracefully while maintaining predictable behavior for users. Treat function calling not as a simple proxy for model inference, but as a complex workflow requiring its own reliability engineering practices. With validation, retry strategies, error handling, and monitoring, you can build systems that scale reliably and provide a smooth user experience even when individual tools fail or services become temporarily unavailable.
The patterns presented here draw from established reliability engineering practices including the circuit breaker pattern and retry pattern. Just as the "Choose a Workflow Before You Build an Autonomous Agent" guide emphasizes defining output contracts and bounded decisions, we should approach function calling with the same discipline: define clear boundaries, implement robust error handling, and keep final approval with reliable system design rather than relying on model behavior alone.
Keywords: LLM function calling, production pipelines, error handling, retry strategies, input validation, structured output, circuit breaker, rate limiting, observability, OpenTelemetry
Sources
- OpenAI Function Calling Documentation - Official guide to function calling patterns and considerations.
- Circuit Breaker Pattern - Design pattern for handling service failures.
- Retry Pattern - Essential pattern for building resilient distributed systems.
- Structured Output Guide - OpenAI's guide to parsing and validating LLM outputs.
- Rate Limiting Best Practices - AWS API Gateway rate limiting strategies.
- OpenTelemetry Documentation - Distributed tracing for microservice architectures.
- Choose a Workflow Before You Build an Autonomous Agent - Guide to defining clear output contracts and bounded decisions.