En tant qu'ingénieur senior spécialisé dans l'intégration d'API IA depuis plus de sept ans, j'ai migré des dizaines de projets critiques vers des providers alternatifs. Aujourd'hui, je partage mon retour d'expérience complet sur la gestion des erreurs Claude Function Calling, une compétence devenue essentielle avec l'adoption massive des modèles conversationnels avancés.
Étude de Cas : Migration d'une Scale-Up E-commerce Lyonnaise
Contexte Métier Initial
Mon dernier projet concernait une scale-up SaaS e-commerce basée à Lyon, employant 45 personnes et générant 12 millions d'euros de chiffre d'affaires annuel via sa plateforme de vente en ligne. Leur système de chatbot client exploitait intensivement les Function Calling de Claude pour automatiser les réponses aux interrogations sur les commandes, le suivi des expéditions et les recommandations personnalisées.
Douleurs du Fournisseur Précédent
Les équipes techniques de cette entreprise avaient souscrit un plan Enterprise chez leur précédent fournisseur, mais rencontraient des problèmes récurrents qui impactaient directement leur指标de satisfaction client :
- Latence moyenne de 420 millisecondes sur les appels Function Calling, provoquant des timeouts lors des pics de trafic
- Facture mensuelle de 4 200 dollars malgré une optimisation constante des prompts
- Gestion des erreurs inconsistante avec des codes d'erreur obscure et une documentation insuffisante
- Support technique réactif uniquement en anglais, posant des problèmes de communication avec les équipes francophones
Pourquoi HolySheep AI
Après une évaluation comparative rigoureuse, j'ai recommandé HolySheep AI pour plusieurs raisons déterminantes. Premièrement, leur latence moyenne inférieure à 50 millisecondes représentait une amélioration de plus de 87% par rapport à leur setup actuel. Deuxièmement, le modèle DeepSeek V3.2 à seulement 0,42 dollar par million de tokens offrait un rapport coût-performances imbattable, soit une économie potentielle de 85% sur leur facture mensuelle. Enfin, l'intégration des méthodes de paiement WeChat et Alipay avec un taux de change avantageux de ¥1 pour 1 dollar facilitait la gestion financière pour une entreprise avec des partenaires chinois.
Migration Technique Étape par Étape
Étape 1 : Configuration Initiale et Rotation des Clés API
La migration commença par une préparation minutieuse de l'environnement. Je dus générer une nouvelle clé API HolySheep tout en conservant l'ancienne pendant une période de transition de deux semaines, permettant un rollback rapide en cas de problème.
import anthropic
from anthropic import AsyncAnthropic
import os
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from enum import Enum
import asyncio
import logging
from datetime import datetime
Configuration HolySheep - NOUVELLE CONFIGURATION
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Ancienne configuration (à supprimer après validation)
OPENAI_COMPATIBLE_BASE = "https://api.anthropic.com/v1"
class FunctionCallingError(Exception):
"""Exception personnalisée pour les erreurs Function Calling"""
def __init__(self, message: str, code: str, retry_after: Optional[int] = None):
self.message = message
self.code = code
self.retry_after = retry_after
super().__init__(self.message)
class ErrorCode(Enum):
RATE_LIMIT = "rate_limit_exceeded"
TIMEOUT = "request_timeout"
INVALID_PARAM = "invalid_parameter"
AUTH_FAILED = "authentication_failed"
SERVER_ERROR = "internal_server_error"
CONTEXT_OVERFLOW = "context_length_exceeded"
FUNCTION_ERROR = "function_execution_error"
@dataclass
class RetryConfig:
max_retries: int = 3
base_delay: float = 1.0
max_delay: float = 60.0
exponential_base: float = 2.0
jitter: bool = True
class HolySheepClaudeClient:
"""Client optimisé pour HolySheep AI avec gestion avancée des erreurs"""
def __init__(self, api_key: str, base_url: str = HOLYSHEEP_BASE_URL):
self.client = AsyncAnthropic(
api_key=api_key,
base_url=base_url,
timeout=30.0,
max_retries=0 # Gestion manuelle des retries
)
self.logger = logging.getLogger(__name__)
self.retry_config = RetryConfig()
async def call_with_function_calling(
self,
messages: List[Dict[str, Any]],
tools: List[Dict[str, Any]],
system_prompt: Optional[str] = None
) -> Dict[str, Any]:
"""Appel principal avec gestion complète des erreurs"""
try:
response = await self.client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
messages=messages,
tools=tools,
system=system_prompt
)
return self._process_response(response)
except Exception as e:
return await self._handle_error(e)
def _process_response(self, response) -> Dict[str, Any]:
"""Traitement de la réponse avec extraction des Function Calls"""
result = {
"content": [],
"usage": {
"input_tokens": response.usage.input_tokens,
"output_tokens": response.usage.output_tokens
},
"model": response.model,
"stop_reason": response.stop_reason
}
for content in response.content:
if content.type == "text":
result["content"].append({
"type": "text",
"text": content.text
})
elif content.type == "tool_use":
result["content"].append({
"type": "tool_use",
"id": content.id,
"name": content.name,
"input": content.input
})
return result
async def _handle_error(self, error: Exception) -> Dict[str, Any]:
"""Gestion centralisée des erreurs avec retry automatique"""
error_mapping = {
"RateLimitError": (ErrorCode.RATE_LIMIT, 429),
"TimeoutError": (ErrorCode.TIMEOUT, 408),
"AuthenticationError": (ErrorCode.AUTH_FAILED, 401),
"BadRequestError": (ErrorCode.INVALID_PARAM, 400),
"InternalServerError": (ErrorCode.SERVER_ERROR, 500),
}
error_type = type(error).__name__
error_info = error_mapping.get(error_type, (ErrorCode.FUNCTION_ERROR, 500))
self.logger.error(
f"Erreur détectée: {error_type} - {str(error)}",
extra={"timestamp": datetime.now().isoformat()}
)
# Levée d'exception structurée
raise FunctionCallingError(
message=str(error),
code=error_info[0].value,
retry_after=getattr(error, 'retry_after', None)
)
Étape 2 : Déploiement Canari et Validation
Le déploiement canari constitua une phase critique. Je configurai un système de routing qui dirigeant progressivement 5%, puis 25%, puis 100% du trafic vers HolySheep. Cette approche permit d'identifier les problèmes de compatibilité avant qu'ils n'impactent l'ensemble des utilisateurs.
import hashlib
from typing import Callable, Any
import random
class CanaryRouter:
"""Router pour déploiement canari avec fallback intelligent"""
def __init__(self, holy_sheep_client, legacy_client, canary_percentage: float = 5.0):
self.holy_sheep = holy_sheep_client
self.legacy = legacy_client
self.canary_percentage = canary_percentage
self.metrics = {
"holy_sheep_calls": 0,
"legacy_calls": 0,
"holy_sheep_errors": 0,
"legacy_errors": 0,
"fallbacks": 0
}
def _should_use_canary(self, user_id: str) -> bool:
"""Décision de routage basée sur l'ID utilisateur pour cohérence"""
user_hash = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
threshold = int((self.canary_percentage / 100) * (2**32))
return user_hash < threshold
async def process_request(
self,
user_id: str,
request_data: Dict[str, Any],
fallback_handler: Callable
) -> Dict[str, Any]:
"""Traitement de requête avec décision canari"""
use_canary = self._should_use_canary(user_id)
if use_canary:
self.metrics["holy_sheep_calls"] += 1
try:
result = await self.holy_sheep.call_with_function_calling(
messages=request_data["messages"],
tools=request_data["tools"],
system_prompt=request_data.get("system")
)
return {
"source": "holy_sheep",
"data": result,
"latency_ms": 0 # À mesurer réellement
}
except FunctionCallingError as e:
self.metrics["holy_sheep_errors"] += 1
self.logger.warning(f"Erreur HolySheep, fallback vers legacy: {e.code}")
if self._should_retry_with_legacy(e):
self.metrics["fallbacks"] += 1
return await self._fallback_to_legacy(request_data)
raise
else:
self.metrics["legacy_calls"] += 1
return await self._fallback_to_legacy(request_data)
def _should_retry_with_legacy(self, error: FunctionCallingError) -> bool:
"""Détermine si un retry vers le legacy est pertinent"""
# Retry uniquement pour les erreurs temporaires
retryable_codes = [
ErrorCode.RATE_LIMIT.value,
ErrorCode.TIMEOUT.value,
ErrorCode.SERVER_ERROR.value
]
return error.code in retryable_codes
async def _fallback_to_legacy(self, request_data: Dict[str, Any]) -> Dict[str, Any]:
"""Fallback vers l'ancien provider"""
self.metrics["legacy_calls"] += 1
try:
result = await self.legacy.call_with_function_calling(
messages=request_data["messages"],
tools=request_data["tools"]
)
return {
"source": "legacy",
"data": result
}
except Exception as e:
self.metrics["legacy_errors"] += 1
raise FunctionCallingError(
message=f"Échec des deux providers: {str(e)}",
code=ErrorCode.FUNCTION_ERROR.value
)
def get_metrics_report(self) -> Dict[str, Any]:
"""Génération du rapport de métriques canari"""
total_calls = self.metrics["holy_sheep_calls"] + self.metrics["legacy_calls"]
return {
"total_calls": total_calls,
"holy_sheep_success_rate": (
(self.metrics["holy_sheep_calls"] - self.metrics["holy_sheep_errors"])
/ max(self.metrics["holy_sheep_calls"], 1)
) * 100,
"legacy_success_rate": (
(self.metrics["legacy_calls"] - self.metrics["legacy_errors"])
/ max(self.metrics["legacy_calls"], 1)
) * 100,
"fallback_rate": (
self.metrics["fallbacks"] / max(self.metrics["holy_sheep_calls"], 1)
) * 100,
"metrics": self.metrics
}
Exemple d'utilisation du système canari
async def main():
router = CanaryRouter(
holy_sheep_client=HolySheepClaudeClient(HOLYSHEEP_API_KEY),
legacy_client=HolySheepClaudeClient(LEGACY_API_KEY),
canary_percentage=25.0 # 25% du trafic vers HolySheep
)
test_request = {
"messages": [
{"role": "user", "content": "Où est ma commande #12345?"}
],
"tools": [
{
"name": "check_order_status",
"description": "Vérifie le statut d'une commande",
"input_schema": {
"type": "object",
"properties": {
"order_id": {"type": "string"}
},
"required": ["order_id"]
}
}
],
"system": "Tu es un assistant e-commerce helpful."
}
try:
result = await router.process_request(
user_id="user_12345",
request_data=test_request,
fallback_handler=None
)
print(f"Résultat depuis {result['source']}: {result['data']}")
except FunctionCallingError as e:
print(f"Échec final: {e.message} (code: {e.code})")
if __name__ == "__main__":
asyncio.run(main())
Gestion Avancée des Erreurs Function Calling
Pattern de Retry Exponentiel avec Jitter
La gestion des erreurs temporaires nécessite une stratégie de retry sophistiquée. J'ai implémenté un système de retry exponentiel avec jitter aléatoire pour éviter le thundering herd problem tout en maximisant les chances de succès.
import asyncio
import random
from typing import Callable, Any, Optional, TypeVar, Union
from functools import wraps
import time
T = TypeVar('T')
class RetryStrategy:
"""Stratégie de retry configurable avec backoff exponentiel"""
def __init__(
self,
max_attempts: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2.0,
jitter_range: float = 0.5,
retryable_errors: Optional[list] = None
):
self.max_attempts = max_attempts
self.base_delay = base_delay
self.max_delay = max_delay
self.exponential_base = exponential_base
self.jitter_range = jitter_range
self.retryable_errors = retryable_errors or [
"RateLimitError",
"TimeoutError",
"InternalServerError",
"ServiceUnavailableError"
]
def calculate_delay(self, attempt: int) -> float:
"""Calcule le délai avec backoff exponentiel et jitter"""
delay = min(
self.base_delay * (self.exponential_base ** attempt),
self.max_delay
)
# Ajout du jitter pour éviter la synchronisation
jitter = delay * self.jitter_range * (2 * random.random() - 1)
return max(0, delay + jitter)
def is_retryable(self, error: Exception) -> bool:
"""Détermine si l'erreur est éligible au retry"""
error_type = type(error).__name__
return error_type in self.retryable_errors or (
hasattr(error, 'code') and error.code in [
ErrorCode.RATE_LIMIT.value,
ErrorCode.TIMEOUT.value,
ErrorCode.SERVER_ERROR.value
]
)
def with_retry(strategy: RetryStrategy):
"""Décorateur pour retry automatique avec stratégie personnalisée"""
def decorator(func: Callable[..., T]) -> Callable[..., T]:
@wraps(func)
async def async_wrapper(*args, **kwargs) -> T:
last_exception = None
for attempt in range(strategy.max_attempts):
try:
return await func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt == strategy.max_attempts - 1:
break
if not strategy.is_retryable(e):
raise
delay = strategy.calculate_delay(attempt)
print(f"Tentative {attempt + 1} échouée: {type(e).__name__}. "
f"Retry dans {delay:.2f}s...")
await asyncio.sleep(delay)
raise last_exception
@wraps(func)
def sync_wrapper(*args, **kwargs) -> T:
last_exception = None
for attempt in range(strategy.max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt == strategy.max_attempts - 1:
break
if not strategy.is_retryable(e):
raise
delay = strategy.calculate_delay(attempt)
print(f"Tentative {attempt + 1} échouée: {type(e).__name__}. "
f"Retry dans {delay:.2f}s...")
time.sleep(delay)
raise last_exception
if asyncio.iscoroutinefunction(func):
return async_wrapper
return sync_wrapper
return decorator
Configuration HolySheep recommandée
HOLYSHEEP_RETRY_STRATEGY = RetryStrategy(
max_attempts=5,
base_delay=2.0,
max_delay=30.0,
exponential_base=2.0,
jitter_range=0.3
)
class EnhancedHolySheepClient(HolySheepClaudeClient):
"""Client HolySheep avec retry automatique optimisé"""
def __init__(self, api_key: str, retry_strategy: Optional[RetryStrategy] = None):
super().__init__(api_key)
self.retry_strategy = retry_strategy or HOLYSHEEP_RETRY_STRATEGY
@with_retry(HOLYSHEEP_RETRY_STRATEGY)
async def call_with_function_calling(
self,
messages: List[Dict[str, Any]],
tools: List[Dict[str, Any]],
system_prompt: Optional[str] = None
) -> Dict[str, Any]:
"""Appel avec retry automatique selon la stratégie configurée"""
return await super().call_with_function_calling(
messages=messages,
tools=tools,
system_prompt=system_prompt
)
async def batch_process(
self,
requests: List[Dict[str, Any]],
concurrency_limit: int = 5
) -> List[Dict[str, Any]]:
"""Traitement par lot avec contrôle de concurrence"""
semaphore = asyncio.Semaphore(concurrency_limit)
async def limited_call(request):
async with semaphore:
return await self.call_with_function_calling(
messages=request["messages"],
tools=request["tools"],
system_prompt=request.get("system")
)
return await asyncio.gather(
*[limited_call(req) for req in requests],
return_exceptions=True
)
Erreurs Courantes et Solutions
Durant mes années d'expérience avec les Function Calling, j'ai identifié les erreurs les plus fréquentes et leurs solutions éprouvées. Voici mon guide de dépannage complet basé sur des centaines de cas de support.
Cas 1 : Erreur Rate Limit avec Retry Infructueux
Symptôme : L'erreur "rate_limit_exceeded" persiste malgré plusieurs tentatives de retry, parfois pendant plusieurs minutes.
Cause racine : La stratégie de retry ne respecte pas correctement l'en-tête Retry-After ou tente trop rapidement des requêtes supplémentaires, aggravant la situation.
Solution implémentée :
# Solution pour le Cas 1 : Rate Limit persistant
class SmartRateLimitHandler:
"""Gestionnaire intelligent des rate limits avec respect des quotas"""
def __init__(self, client: EnhancedHolySheepClient):
self.client = client
self.token_bucket = TokenBucket(capacity=100, refill_rate=10)
self.request_history = []
async def smart_call(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""Appel intelligent avec gestion proactive des rate limits"""
# Vérification proactive du quota
if not self.token_bucket.try_consume(1):
wait_time = self.token_bucket.time_until_refill()
print(f"Quota épuisé, attente de {wait_time:.2f}s")
await asyncio.sleep(wait_time)
try:
result = await self.client.call_with_function_calling(
messages=request["messages"],
tools=request["tools"],
system_prompt=request.get("system")
)
# Mise à jour du tracking
self.request_history.append({
"timestamp": datetime.now(),
"success": True
})
return result
except FunctionCallingError as e:
if e.code == ErrorCode.RATE_LIMIT.value:
return await self._handle_rate_limit_error(e, request)
raise
async def _handle_rate_limit_error(
self,
error: FunctionCallingError,
request: Dict[str, Any]
) -> Dict[str, Any]:
"""Traitement spécifique du rate limit avec backoff intelligent"""
# Utilisation du retry_after si disponible
if error.retry_after:
wait_time = error.retry_after
else:
# Calcul d'un délai raisonnable basé sur l'historique
recent_errors = len([
h for h in self.request_history[-10:]
if not h.get("success", True)
])
wait_time = min(60, 2 ** recent_errors) # Max 60 secondes
print(f"Rate limit atteint. Pause de {wait_time}s avant retry...")
await asyncio.sleep(wait_time)
# Retry avec quota rafraîchi
self.token_bucket.refill()
return await self.client.call_with_function_calling(
messages=request["messages"],
tools=request["tools"],
system_prompt=request.get("system")
)
class TokenBucket:
"""Implémentation du token bucket pour rate limiting local"""
def __init__(self, capacity: int, refill_rate: float):
self.capacity = capacity
self.tokens = float(capacity)
self.refill_rate = refill_rate
self.last_refill = time.time()
def try_consume(self, tokens: int = 1) -> bool:
"""Tente de consommer des tokens"""
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def _refill(self):
"""Rafraîchit les tokens selon le taux de refill"""
now = time.time()
elapsed = now - self.last_refill
new_tokens = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now
def time_until_refill(self) -> float:
"""Calcule le temps avant d'avoir assez de tokens"""
if self.tokens >= 1:
return 0
tokens_needed = 1 - self.tokens
return tokens_needed / self.refill_rate
Cas 2 : Context Length Exceeded sur Prompts Longs
Symptôme : L'erreur "context_length_exceeded" survient lors du traitement de conversations longues ou de documents volumineux, bloquant le fonctionnel.
Cause racine : L'historique de conversation n'est pas correctement géré, et le contexte accumulé dépasse la fenêtre maximale du modèle.
Solution implémentée :
# Solution pour le Cas 2 : Context Length Overflow
class ConversationManager:
"""Gestionnaire de conversation avec troncature intelligente"""
def __init__(
self,
max_context_tokens: int = 160000, # Limit HolySheep
reserved_tokens: int = 4000, # Pour la réponse
summarization_threshold: float = 0.8
):
self.max_context_tokens = max_context_tokens
self.reserved_tokens = reserved_tokens
self.summarization_threshold = summarization_threshold
self.messages = []
self.conversation_history_summary = ""
def estimate_tokens(self, text: str) -> int:
"""Estimation grossière du nombre de tokens"""
# Approximation: ~4 caractères par token en français
return len(text) // 4
async def add_message(self, role: str, content: str) -> None:
"""Ajoute un message avec gestion automatique du contexte"""
self.messages.append({
"role": role,
"content": content,
"timestamp": datetime.now()
})
await self._optimize_context()
async def _optimize_context(self) -> None:
"""Optimise le contexte par troncature ou résumé"""
current_tokens = sum(
self.estimate_tokens(m["content"]) for m in self.messages
)
available_tokens = self.max_context_tokens - self.reserved_tokens
if current_tokens > available_tokens:
usage_ratio = current_tokens / available_tokens
if usage_ratio > self.summarization_threshold:
await self._summarize_old_messages()
else:
self._truncate_oldest_messages(
available_tokens - current_tokens
)
async def _summarize_old_messages(self) -> None:
"""Résumé les messages anciens via HolySheep"""
if len(self.messages) <= 2:
return
# Garde les derniers messages intacts
recent_messages = self.messages[-2:]
old_messages = self.messages[:-2]
# Construction du prompt de résumé
summary_request = {
"messages": [
{
"role": "user",
"content": f"Résume cette conversation en conservant les informations importantes:\n\n" +
"\n".join([
f"{m['role']}: {m['content'][:200]}..."
for m in old_messages[:5]
])
}
],
"tools": [],
"system": "Tu es un assistant qui résume des conversations. Sois concis mais exhaustif."
}
try:
client = EnhancedHolySheepClient(HOLYSHEEP_API_KEY)
response = await client.call_with_function_calling(**summary_request)
summary_text = response["content"][0]["text"]
self.conversation_history_summary = summary_text
# Remplace les anciens messages par le résumé
self.messages = [
{"role": "system", "content": f"Résumé de la conversation précédente:\n{summary_text}"}
] + recent_messages
except Exception as e:
# Fallback vers la troncature simple
self._truncate_oldest_messages(
self.estimate_tokens("\n".join([m["content"] for m in old_messages])) // 2
)
def _truncate_oldest_messages(self, tokens_to_remove: int) -> None:
"""Tronque les messages les plus anciens"""
removed_tokens = 0
truncated_messages = []
for message in self.messages:
message_tokens = self.estimate_tokens(message["content"])
if removed_tokens + message_tokens <= tokens_to_remove:
removed_tokens += message_tokens
continue
# Tronque le message si nécessaire
if removed_tokens < tokens_to_remove:
chars_to_keep = (tokens_to_remove - removed_tokens) * 4
message["content"] = message["content"][:chars_to_keep] + "... [tronqué]"
removed_tokens += message_tokens
truncated_messages.append(message)
self.messages = truncated_messages
def get_messages_for_api(self) -> List[Dict[str, str]]:
"""Retourne les messages formatés pour l'API HolySheep"""
result = []
if self.conversation_history_summary:
result.append({
"role": "system",
"content": self.conversation_history_summary
})
for msg in self.messages:
result.append({
"role": msg["role"],
"content": msg["content"]
})
return result
Cas 3 : Function Execution Error avec Outils Mal Configurés
Symptôme : Les Function Calls sont générés mais l'exécution échoue avec des erreurs de type "function_execution_error" ou "invalid_tool_input".
Cause racine : Mauvaise correspondance entre le schéma de l'outil défini dans le prompt et les paramètres réellement attendus par la fonction Python sous-jacente.
Solution implémentée :
# Solution pour le Cas 3 : Function Execution Error
from typing import get_type_hints, get_origin, get_args
import inspect
class FunctionCallingExecutor:
"""Exécuteur sécurisé pour les Function Calls avec validation"""
def __init__(self):
self.tools_registry: Dict[str, Callable] = {}
self.tools_schemas: Dict[str, Dict] = {}
def register_tool(
self,
name: str,
func: Callable,
description: str = "",
schema_override: Optional[Dict] = None
) -> None:
"""Enregistre un outil avec son schéma auto-généré"""
self.tools_registry[name] = func
if schema_override:
self.tools_schemas[name] = schema_override
else:
self.tools_schemas[name] = self._generate_schema(name, func, description)
def _generate_schema(self, name: str, func: Callable, description: str) -> Dict:
"""Génère automatiquement le schéma JSON du Function Call"""
hints = get_type_hints(func)
sig = inspect.signature(func)
properties = {}
required = []
for param_name, param in sig.parameters.items():
if param_name in ('self', 'cls'):
continue
param_type = hints.get(param_name, str)
json_type = self._python_type_to_json(param_type)
properties[param_name] = {
"type": json_type,
"description": f"Parameter {param_name}"
}
if param.default is inspect.Parameter.empty:
required.append(param_name)
return {
"name": name,
"description": description or func.__doc__ or f"Execute {name}",
"parameters": {
"type": "object",
"properties": properties,
"required": required
}
}
def _python_type_to_json(self, python_type) -> str:
"""Convertit un type Python en type JSON Schema"""
origin = get_origin(python_type)
if origin is Union:
args = get_args(python_type)
non_none = [a for a in args if a is not type(None)]
if len(non_none) == 1:
return self._python_type_to_json(non_none[0])
type_map = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
list: "array",
dict: "object"
}
return type_map.get(python_type, "string")
async def execute_function_call(
self,
function_name: str,
function_input: Dict[str, Any]
) -> Dict[str, Any]:
"""Exécute un Function Call avec validation complète"""
if function_name not in self.tools_registry:
raise FunctionCallingError(
message=f"Fonction {function_name} non trouvée",
code=ErrorCode.FUNCTION_ERROR.value
)
func = self.tools_registry[function_name]
# Validation des paramètres
validated_input = self._validate_input(
function_name,
function_input
)
try:
# Exécution synchrone ou asynchrone selon la fonction
if asyncio.iscoroutinefunction(func):
result = await func(**validated_input)
else:
result = func(**validated_input)
return {
"success": True,
"result": result,
"function": function_name
}
except TypeError as e:
# Erreur de paramètres
raise FunctionCallingError(
message=f"Paramètres invalides pour {function_name}: {str(e)}",
code=ErrorCode.INVALID_PARAM.value
)
except Exception as e:
raise FunctionCallingError(
message=f"Erreur d'exécution de {function_name}: {str(e)}",
code=ErrorCode.FUNCTION_ERROR.value
)
def _validate_input(
self,
function_name: str,
function_input: Dict[str, Any]
) -> Dict[str, Any]:
"""Valide les entrées contre le schéma de l'outil"""
schema = self.tools_schemas.get(function_name, {})
required = schema.get("parameters", {}).get("required", [])
properties = schema.get("parameters", {}).get("properties", {})
# Vérification des paramètres requis
missing = [p for p in required if p not in function_input]
if missing:
raise FunctionCallingError(
message=f"Paramètres manquants pour {function_name}: {missing}",
code=ErrorCode.INVALID_PARAM.value
)
# Validation des types
validated = {}
for key, value in function_input.items():
if key in properties:
expected_type = properties[key].get("type")
if not self._validate_type(value, expected_type):
validated[key] = self._coerce_type(value, expected_type)
else:
validated[key] = value
else:
validated[key] = value
return validated
def _validate_type(self, value: Any, expected_type: str) -> bool:
"""Valide le type d'une valeur"""
validators = {
"string": lambda v: isinstance(v, str),
"integer": lambda v: isinstance(v, int) and not isinstance(v, bool),
"number": lambda v: isinstance(v, (int, float)) and not isinstance(v, bool),
"boolean": lambda v: isinstance(v, bool),
"array": lambda v: isinstance(v, list),
"object": lambda v: isinstance(v, dict)
}
return validators.get(expected_type, lambda v: True)(value)
def _coerce_type(self, value: Any, expected_type: str) -> Any:
"""Convertit le type si possible"""
if expected_type == "string":
return str(value)
elif expected_type == "integer":
return int(value)
elif expected_type == "number":
return float(value)
elif expected_type == "boolean":
return bool(value)
return value
def get_tools_for_api(self) -> List[Dict]:
"""Retourne les schémas pour l'API HolySheep"""
return list(self.tools_schemas.values())
Exemple d'utilisation avec les outils e-commerce
executor = FunctionCallingExecutor()
@executor.register_tool(
name="check_order_status",
description="Vérifie le statut d'une commande e-commerce"
)
def check_order_status(order_id: str) -> dict:
"""Vérifie le statut d'une commande"""
# Logique métier réelle
return {
"order_id": order_id,
"status": "shipped",
"estimated_delivery": "2026-02-15",
"tracking_number": "TRK123456789"
}
@executor.register_tool(
name="get_product_info",
description="Récupère les informations d'un