Outils et function calling pour agents IA

User: Quelle est la météo à Paris ?
LLM: Je ne peux pas accéder aux données en temps réel. [HALLUCINATION POSSIBLE]

User: Quelle est la météo à Paris ?
LLM: [Appelle get_weather(city="Paris")]
Tool: {"temperature": 18, "condition": "nuageux"}
LLM: Il fait 18°C à Paris avec des nuages.

function_schema = {
    "type": "function",
    "function": {
        "name": "get_weather",  # Nom de la fonction
        "description": "Obtient la météo actuelle pour une ville donnée",  # Description claire
        "parameters": {  # JSON Schema des paramètres
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "Nom de la ville (ex: 'Paris', 'Tokyo')"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Unité de température"
                }
            },
            "required": ["city"]  # Paramètres obligatoires
        }
    }
}

import openai
import json

# ========== 1. DÉFINIR LES FONCTIONS ==========

functions = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Obtient la météo actuelle d'une ville",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "Nom de la ville"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "default": "celsius"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "Obtient le prix d'une action en bourse",
            "parameters": {
                "type": "object",
                "properties": {
                    "ticker": {
                        "type": "string",
                        "description": "Symbole de l'action (ex: 'AAPL', 'TSLA')"
                    },
                    "exchange": {
                        "type": "string",
                        "enum": ["NYSE", "NASDAQ", "EURONEXT"],
                        "description": "Bourse"
                    }
                },
                "required": ["ticker"]
            }
        }
    }
]

# ========== 2. IMPLÉMENTER LES FONCTIONS ==========

def get_weather(city: str, unit: str = "celsius") -> dict:
    """Simule un appel API météo"""
    # En production : appeler une vraie API (OpenWeatherMap, etc.)
    weather_data = {
        "Paris": {"temp": 18, "condition": "nuageux"},
        "Tokyo": {"temp": 25, "condition": "ensoleillé"},
        "New York": {"temp": 22, "condition": "pluvieux"}
    }

    data = weather_data.get(city, {"temp": 20, "condition": "inconnu"})

    if unit == "fahrenheit":
        data["temp"] = data["temp"] * 9/5 + 32

    return {
        "city": city,
        "temperature": data["temp"],
        "condition": data["condition"],
        "unit": unit
    }

def get_stock_price(ticker: str, exchange: str = "NYSE") -> dict:
    """Simule un appel API bourse"""
    prices = {
        "AAPL": 175.50,
        "TSLA": 242.80,
        "GOOGL": 140.25
    }

    return {
        "ticker": ticker,
        "price": prices.get(ticker, 100.0),
        "exchange": exchange,
        "currency": "USD"
    }

# Mapper les noms de fonctions
available_functions = {
    "get_weather": get_weather,
    "get_stock_price": get_stock_price
}

# ========== 3. BOUCLE D'EXÉCUTION ==========

def run_agent_with_functions(user_query: str, max_iterations: int = 5):
    """Exécute l'agent avec function calling"""

    messages = [
        {"role": "system", "content": "Tu es un assistant utile."},
        {"role": "user", "content": user_query}
    ]

    for iteration in range(max_iterations):
        print(f"\n{'='*60}")
        print(f"🔄 Itération {iteration + 1}")

        # Appel OpenAI
        response = openai.chat.completions.create(
            model="gpt-4",
            messages=messages,
            tools=functions,
            tool_choice="auto"  # "auto", "none", ou {"type": "function", "function": {"name": "..."}}
        )

        message = response.choices[0].message
        messages.append(message)

        # Vérifier si le modèle veut appeler des fonctions
        if not message.tool_calls:
            # Réponse finale
            print(f"\n✅ Réponse finale : {message.content}")
            return message.content

        # Exécuter les fonctions demandées
        for tool_call in message.tool_calls:
            function_name = tool_call.function.name
            arguments = json.loads(tool_call.function.arguments)

            print(f"\n🔧 Appel de fonction : {function_name}({arguments})")

            # Exécuter la fonction
            if function_name in available_functions:
                function_result = available_functions[function_name](**arguments)
                print(f"📊 Résultat : {function_result}")

                # Ajouter le résultat aux messages
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "name": function_name,
                    "content": json.dumps(function_result)
                })
            else:
                # Fonction inconnue
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "name": function_name,
                    "content": json.dumps({"error": f"Fonction {function_name} inconnue"})
                })

    return "⚠️ Limite d'itérations atteinte"


# ========== 4. EXEMPLE D'UTILISATION ==========

if __name__ == "__main__":
    result = run_agent_with_functions(
        "Donne-moi la météo à Paris et le prix de l'action Tesla"
    )

============================================================
🔄 Itération 1

🔧 Appel de fonction : get_weather({'city': 'Paris'})
📊 Résultat : {'city': 'Paris', 'temperature': 18, 'condition': 'nuageux', 'unit': 'celsius'}

🔧 Appel de fonction : get_stock_price({'ticker': 'TSLA'})
📊 Résultat : {'ticker': 'TSLA', 'price': 242.8, 'exchange': 'NYSE', 'currency': 'USD'}

============================================================
🔄 Itération 2

✅ Réponse finale : Il fait actuellement 18°C à Paris avec un temps nuageux. Le prix de l'action Tesla (TSLA) est de 242,80 USD sur le NYSE.

response = openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Météo à Paris, Tokyo et New York"}],
    tools=functions,
    tool_choice="auto"
)

# Le modèle retourne 3 tool_calls simultanés
for tool_call in response.choices[0].message.tool_calls:
    print(f"Appel : {tool_call.function.name}({tool_call.function.arguments})")

# Sortie :
# Appel : get_weather({'city': 'Paris'})
# Appel : get_weather({'city': 'Tokyo'})
# Appel : get_weather({'city': 'New York'})

# Forcer l'utilisation d'une fonction spécifique
response = openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Il fait beau aujourd'hui"}],
    tools=functions,
    tool_choice={
        "type": "function",
        "function": {"name": "get_weather"}
    }
)

# Le modèle DOIT appeler get_weather, même si ce n'est pas pertinent

import anthropic

client = anthropic.Anthropic(api_key="your-api-key")

# ========== DÉFINIR LES TOOLS ==========

tools = [
    {
        "name": "get_weather",
        "description": "Obtient la météo actuelle pour une ville",
        "input_schema": {  # Équivalent de "parameters" chez OpenAI
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "Nom de la ville"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Unité de température"
                }
            },
            "required": ["city"]
        }
    }
]

# ========== BOUCLE D'EXÉCUTION ==========

messages = [{"role": "user", "content": "Météo à Paris ?"}]

while True:
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        tools=tools,
        messages=messages
    )

    print(f"\n🤖 Stop reason: {response.stop_reason}")

    # Vérifier si Claude veut utiliser un outil
    if response.stop_reason == "tool_use":
        # Extraire le tool use
        tool_use = next(block for block in response.content if block.type == "tool_use")

        print(f"🔧 Tool: {tool_use.name}")
        print(f"📥 Input: {tool_use.input}")

        # Exécuter la fonction
        tool_result = get_weather(**tool_use.input)

        print(f"📊 Result: {tool_result}")

        # Ajouter la réponse de l'assistant
        messages.append({"role": "assistant", "content": response.content})

        # Ajouter le résultat de l'outil
        messages.append({
            "role": "user",
            "content": [
                {
                    "type": "tool_result",
                    "tool_use_id": tool_use.id,
                    "content": json.dumps(tool_result)
                }
            ]
        })

    else:
        # Réponse finale
        final_text = next(
            (block.text for block in response.content if hasattr(block, "text")),
            None
        )
        print(f"\n✅ Réponse finale : {final_text}")
        break

import google.generativeai as genai

genai.configure(api_key="your-api-key")

# ========== DÉFINIR LES FUNCTIONS ==========

def get_weather(city: str, unit: str = "celsius") -> dict:
    """Obtient la météo d'une ville"""
    # Implémentation...
    pass

# Déclarer la fonction pour Gemini
function_declarations = [
    {
        "name": "get_weather",
        "description": "Obtient la météo actuelle d'une ville",
        "parameters": {
            "type_": "OBJECT",  # Notez le underscore
            "properties": {
                "city": {
                    "type_": "STRING",
                    "description": "Nom de la ville"
                },
                "unit": {
                    "type_": "STRING",
                    "enum": ["celsius", "fahrenheit"]
                }
            },
            "required": ["city"]
        }
    }
]

# ========== CRÉER LE MODÈLE ==========

model = genai.GenerativeModel(
    model_name="gemini-1.5-pro",
    tools=function_declarations
)

chat = model.start_chat()

# ========== BOUCLE ==========

response = chat.send_message("Météo à Paris ?")

# Vérifier les function calls
if response.candidates[0].content.parts[0].function_call:
    function_call = response.candidates[0].content.parts[0].function_call

    print(f"🔧 Function: {function_call.name}")
    print(f"📥 Args: {dict(function_call.args)}")

    # Exécuter
    result = get_weather(**dict(function_call.args))

    # Renvoyer le résultat
    response = chat.send_message(
        genai.protos.Content(
            parts=[
                genai.protos.Part(
                    function_response=genai.protos.FunctionResponse(
                        name=function_call.name,
                        response={"result": result}
                    )
                )
            ]
        )
    )

    print(f"✅ Réponse : {response.text}")

from pydantic import BaseModel, Field, validator
from typing import Literal, Optional
from datetime import datetime

# ========== MODÈLE PYDANTIC ==========

class WeatherInput(BaseModel):
    """Validation des inputs pour get_weather"""

    city: str = Field(
        ...,
        min_length=2,
        max_length=50,
        description="Nom de la ville"
    )

    unit: Literal["celsius", "fahrenheit"] = Field(
        default="celsius",
        description="Unité de température"
    )

    date: Optional[str] = Field(
        default=None,
        description="Date au format YYYY-MM-DD (optionnel)"
    )

    @validator("city")
    def validate_city(cls, v):
        """Valide que la ville n'est pas vide"""
        if not v.strip():
            raise ValueError("La ville ne peut pas être vide")
        return v.strip().title()  # Capitalise

    @validator("date")
    def validate_date(cls, v):
        """Valide le format de date"""
        if v is None:
            return None
        try:
            datetime.strptime(v, "%Y-%m-%d")
            return v
        except ValueError:
            raise ValueError("Format de date invalide. Utilisez YYYY-MM-DD")

# ========== GÉNÉRER LE SCHÉMA JSON ==========

def pydantic_to_openai_schema(model: BaseModel) -> dict:
    """Convertit un modèle Pydantic en schéma OpenAI"""
    schema = model.schema()

    return {
        "type": "function",
        "function": {
            "name": model.__name__.replace("Input", "").lower(),
            "description": model.__doc__ or "",
            "parameters": {
                "type": "object",
                "properties": schema["properties"],
                "required": schema.get("required", [])
            }
        }
    }

weather_function = pydantic_to_openai_schema(WeatherInput)

# ========== FONCTION VALIDÉE ==========

def get_weather_validated(city: str, unit: str = "celsius", date: Optional[str] = None) -> dict:
    """Version validée de get_weather"""
    try:
        # Valider les inputs
        validated = WeatherInput(city=city, unit=unit, date=date)

        # Logique métier
        return {
            "city": validated.city,
            "temperature": 18,
            "unit": validated.unit,
            "date": validated.date or "today"
        }

    except ValueError as e:
        return {"error": str(e)}

# ========== UTILISATION ==========

functions = [weather_function]

response = openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Météo à paris demain"}],
    tools=functions
)

# Les paramètres sont maintenant validés automatiquement

def sequential_agent(query: str):
    """Agent avec chaînage de fonctions"""

    # Étape 1 : Identifier l'entreprise
    response1 = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": f"Trouve le symbole boursier de : {query}"}],
        tools=[get_ticker_symbol_function],
        tool_choice={"type": "function", "function": {"name": "get_ticker_symbol"}}
    )

    ticker = json.loads(response1.choices[0].message.tool_calls[0].function.arguments)["ticker"]

    # Étape 2 : Obtenir le prix
    response2 = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": f"Prix de l'action {ticker}"}],
        tools=[get_stock_price_function],
        tool_choice={"type": "function", "function": {"name": "get_stock_price"}}
    )

    price_args = json.loads(response2.choices[0].message.tool_calls[0].function.arguments)
    price = get_stock_price(**price_args)

    return f"Le prix de {query} ({ticker}) est {price['price']} USD"

# Exemple
result = sequential_agent("Apple")
# → "Le prix de Apple (AAPL) est 175.50 USD"

def adaptive_agent(user_query: str, user_subscription: str):
    """Agent avec outils adaptatifs"""

    # Outils de base
    base_tools = [search_tool, calculator_tool]

    # Outils premium
    premium_tools = [
        realtime_data_tool,
        advanced_analytics_tool,
        export_tool
    ]

    # Adapter selon l'abonnement
    if user_subscription == "premium":
        tools = base_tools + premium_tools
        system_msg = "Tu as accès à tous les outils, y compris les données en temps réel."
    else:
        tools = base_tools
        system_msg = "Tu as accès aux outils de base. Pour plus d'outils, suggère un upgrade premium."

    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": system_msg},
            {"role": "user", "content": user_query}
        ],
        tools=tools
    )

    # ...

def tool_with_fallback(tool_function, fallback_function):
    """Wrapper qui essaie un outil, puis un fallback"""

    def wrapper(**kwargs):
        try:
            return tool_function(**kwargs)
        except Exception as e:
            print(f"⚠️ Erreur avec {tool_function.__name__}: {e}")
            print(f"🔄 Utilisation du fallback...")
            return fallback_function(**kwargs)

    return wrapper

# Exemple : recherche web avec fallback
def search_with_tavily(query: str) -> str:
    """Recherche avec Tavily (payant, meilleur)"""
    # Peut échouer si API down
    return tavily_search(query)

def search_with_duckduckgo(query: str) -> str:
    """Recherche avec DuckDuckGo (gratuit, backup)"""
    return duckduckgo_search(query)

# Créer l'outil avec fallback
search_tool = Tool(
    name="search",
    func=tool_with_fallback(search_with_tavily, search_with_duckduckgo),
    description="Recherche sur le web"
)

from functools import wraps
import time
from collections import defaultdict

class RateLimiter:
    """Rate limiter par outil"""

    def __init__(self):
        self.calls = defaultdict(list)  # {tool_name: [timestamps]}

    def limit(self, calls_per_minute: int):
        """Décorateur pour limiter les appels"""

        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                tool_name = func.__name__
                now = time.time()

                # Nettoyer les appels de plus d'1 minute
                self.calls[tool_name] = [
                    t for t in self.calls[tool_name]
                    if now - t < 60
                ]

                # Vérifier la limite
                if len(self.calls[tool_name]) >= calls_per_minute:
                    wait_time = 60 - (now - self.calls[tool_name][0])
                    print(f"⏳ Rate limit atteint pour {tool_name}. Attente de {wait_time:.1f}s...")
                    time.sleep(wait_time)

                    # Nettoyer à nouveau
                    self.calls[tool_name] = []

                # Enregistrer l'appel
                self.calls[tool_name].append(now)

                # Exécuter
                return func(*args, **kwargs)

            return wrapper
        return decorator

# Utilisation
rate_limiter = RateLimiter()

@rate_limiter.limit(calls_per_minute=10)
def expensive_api_call(query: str) -> str:
    """API coûteuse limitée à 10 appels/min"""
    return call_expensive_api(query)

from functools import lru_cache
import hashlib
import json

def cacheable_tool(ttl_seconds: int = 3600):
    """Décorateur pour cacher les résultats d'outils"""

    cache = {}

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Créer une clé de cache
            cache_key = hashlib.md5(
                json.dumps({"args": args, "kwargs": kwargs}, sort_keys=True).encode()
            ).hexdigest()

            # Vérifier le cache
            if cache_key in cache:
                cached_result, timestamp = cache[cache_key]
                if time.time() - timestamp < ttl_seconds:
                    print(f"⚡ Cache hit pour {func.__name__}")
                    return cached_result

            # Exécuter et cacher
            result = func(*args, **kwargs)
            cache[cache_key] = (result, time.time())

            return result

        return wrapper
    return decorator

# Utilisation
@cacheable_tool(ttl_seconds=300)  # Cache 5 minutes
def get_stock_price(ticker: str) -> dict:
    """Prix d'action (changent souvent mais pas à la seconde)"""
    return fetch_stock_price(ticker)

def safe_function_executor(function_name: str, arguments: dict) -> dict:
    """Exécuteur sûr de fonctions"""

    try:
        # Valider la fonction
        if function_name not in available_functions:
            return {
                "success": False,
                "error": f"Fonction '{function_name}' inconnue",
                "available_functions": list(available_functions.keys())
            }

        # Valider les arguments
        func = available_functions[function_name]
        # Utiliser signature pour vérifier les paramètres
        import inspect
        sig = inspect.signature(func)

        # Vérifier les paramètres requis
        for param_name, param in sig.parameters.items():
            if param.default == inspect.Parameter.empty and param_name not in arguments:
                return {
                    "success": False,
                    "error": f"Paramètre requis manquant : {param_name}"
                }

        # Exécuter
        result = func(**arguments)

        return {
            "success": True,
            "result": result
        }

    except TypeError as e:
        return {
            "success": False,
            "error": f"Arguments invalides : {str(e)}"
        }

    except Exception as e:
        return {
            "success": False,
            "error": f"Erreur d'exécution : {str(e)}",
            "type": type(e).__name__
        }

import signal
from contextlib import contextmanager

@contextmanager
def timeout(seconds: int):
    """Context manager pour timeout"""

    def handler(signum, frame):
        raise TimeoutError(f"Timeout après {seconds}s")

    # Set timeout
    signal.signal(signal.SIGALRM, handler)
    signal.alarm(seconds)

    try:
        yield
    finally:
        signal.alarm(0)  # Cancel alarm

# Utilisation
def get_weather_with_timeout(city: str) -> dict:
    """Météo avec timeout de 5s"""
    try:
        with timeout(5):
            return fetch_weather_from_api(city)
    except TimeoutError:
        return {"error": "L'API météo ne répond pas"}

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def get_weather_with_retry(city: str) -> dict:
    """Météo avec retry automatique"""
    response = requests.get(f"https://api.weather.com/{city}")
    response.raise_for_status()
    return response.json()

# Essaie 3 fois avec délais de 2s, 4s, 8s entre les tentatives