AlbertAPI-tool-calling.py

"""
Tool Calling avec Albert API
========================================================================
Prérequis : pip install openai
"""

import json
import inspect
from typing import Annotated, Literal, get_type_hints, get_origin, get_args
from functools import wraps
from openai import OpenAI
import sys
sys.stdout.reconfigure(encoding="utf-8")
sys.stderr.reconfigure(encoding="utf-8")


# ─────────────────────────────────────────────
# 1. Le décorateur @tool et le registre
# ─────────────────────────────────────────────

class ToolRegistry:
    """Registre central des tools déclarés avec @tool."""

    def __init__(self):
        self.functions: dict[str, callable] = {}
        self.schemas: list[dict] = []

    def tool(self, func):
        """
        Décorateur qui enregistre une fonction comme tool.
        Le schéma JSON est auto-généré à partir de :
          - La docstring → description de la fonction
          - Les type hints → types des paramètres
          - Annotated[type, "description"] → description des paramètres
          - Literal["a", "b"] → enum
        """
        schema = self._build_schema(func)
        self.functions[func.__name__] = func
        self.schemas.append({"type": "function", "function": schema})

        @wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)

        return wrapper

    def _python_type_to_json(self, annotation) -> dict:
        """Convertit un type Python en type JSON Schema."""
        origin = get_origin(annotation)
        args = get_args(annotation)

        # Literal["a", "b", "c"] → enum
        if origin is Literal:
            # Déduire le type à partir des valeurs
            if all(isinstance(a, str) for a in args):
                return {"type": "string", "enum": list(args)}
            elif all(isinstance(a, int) for a in args):
                return {"type": "integer", "enum": list(args)}
            return {"type": "string", "enum": [str(a) for a in args]}

        # list[str], list[int], etc.
        if origin is list:
            items_type = self._python_type_to_json(args[0]) if args else {"type": "string"}
            return {"type": "array", "items": items_type}

        # Types simples
        type_map = {
            str: {"type": "string"},
            int: {"type": "integer"},
            float: {"type": "number"},
            bool: {"type": "boolean"},
        }
        return type_map.get(annotation, {"type": "string"})

    def _build_schema(self, func) -> dict:
        """Construit le schéma OpenAI function calling à partir de la fonction."""
        hints = get_type_hints(func, include_extras=True)
        sig = inspect.signature(func)
        docstring = inspect.getdoc(func) or ""

        # Extraire la description principale (première ligne de la docstring)
        doc_lines = docstring.strip().split("\n")
        description = doc_lines[0] if doc_lines else func.__name__

        properties = {}
        required = []

        for param_name, param in sig.parameters.items():
            if param_name in ("self", "cls"):
                continue

            annotation = hints.get(param_name, str)
            param_description = None

            # Gérer Annotated[type, "description"]
            if get_origin(annotation) is Annotated:
                args = get_args(annotation)
                real_type = args[0]
                # Chercher une string dans les métadonnées comme description
                for meta in args[1:]:
                    if isinstance(meta, str):
                        param_description = meta
                        break
                annotation = real_type
            
            # Construire la propriété
            prop = self._python_type_to_json(annotation)
            if param_description:
                prop["description"] = param_description

            properties[param_name] = prop

            # Si pas de valeur par défaut → required
            if param.default is inspect.Parameter.empty:
                required.append(param_name)

        schema = {
            "name": func.__name__,
            "description": description,
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required,
            },
        }
        return schema

    def execute(self, function_name: str, arguments: dict):
        """Exécute une fonction enregistrée par son nom."""
        if function_name not in self.functions:
            raise ValueError(f"Fonction '{function_name}' non enregistrée")
        return self.functions[function_name](**arguments)


# ─────────────────────────────────────────────
# 2. Instanciation du registre
# ─────────────────────────────────────────────
registry = ToolRegistry()
tool = registry.tool  # Raccourci pour le décorateur


# ─────────────────────────────────────────────
# 3. Déclaration des tools — C'est ici que la magie opère ✨
# ─────────────────────────────────────────────

@tool
def get_meteo(
    ville: Annotated[str, "Le nom de la ville (ex: Paris, Lyon, Marseille)"],
    unite: Annotated[Literal["celsius", "fahrenheit"], "L'unité de température"] = "celsius",
) -> dict:
    """Obtenir la météo actuelle pour une ville donnée."""
    meteo_data = {
        "Paris": {"temperature": 18, "conditions": "Nuageux", "humidite": 72},
        "Lyon": {"temperature": 22, "conditions": "Ensoleillé", "humidite": 45},
        "Marseille": {"temperature": 26, "conditions": "Ensoleillé", "humidite": 38},
        "Lille": {"temperature": 14, "conditions": "Pluvieux", "humidite": 85},
    }
    data = meteo_data.get(ville, {"temperature": 20, "conditions": "Inconnu", "humidite": 50})
    if unite == "fahrenheit":
        data["temperature"] = round(data["temperature"] * 9 / 5 + 32, 1)
    data["ville"] = ville
    data["unite"] = unite
    return data


@tool
def rechercher_restaurants(
    ville: Annotated[str, "La ville où chercher des restaurants"],
    cuisine: Annotated[str, "Le type de cuisine (française, italienne, japonaise, etc.)"] = "française",
    budget: Annotated[Literal["économique", "moyen", "haut de gamme"], "Le niveau de budget"] = "moyen",
) -> dict:
    """Rechercher des restaurants dans une ville selon le type de cuisine et le budget."""
    restaurants = {
        ("Paris", "française"): [
            {"nom": "Le Petit Cler", "note": 4.5, "prix": "€€"},
            {"nom": "Chez Janou", "note": 4.3, "prix": "€€"},
        ],
        ("Paris", "italienne"): [
            {"nom": "Pink Mamma", "note": 4.4, "prix": "€€"},
            {"nom": "Ober Mamma", "note": 4.2, "prix": "€€"},
        ],
        ("Lyon", "française"): [
            {"nom": "Le Bouchon des Filles", "note": 4.6, "prix": "€€"},
            {"nom": "Daniel et Denise", "note": 4.7, "prix": "€€€"},
        ],
    }
    result = restaurants.get(
        (ville, cuisine),
        [{"nom": f"Restaurant {cuisine} à {ville}", "note": 4.0, "prix": "€€"}],
    )
    return {"ville": ville, "cuisine": cuisine, "budget": budget, "restaurants": result}


@tool
def calculer_itineraire(
    depart: Annotated[str, "La ville de départ"],
    arrivee: Annotated[str, "La ville d'arrivée"],
    mode: Annotated[Literal["voiture", "train", "avion"], "Le mode de transport"] = "voiture",
) -> dict:
    """Calculer un itinéraire entre deux villes avec un mode de transport."""
    itineraires = {
        ("Paris", "Lyon", "voiture"): {"distance_km": 465, "duree": "4h30", "peages": "35€"},
        ("Paris", "Lyon", "train"): {"distance_km": 427, "duree": "1h58", "prix": "49-89€"},
        ("Paris", "Marseille", "train"): {"distance_km": 775, "duree": "3h15", "prix": "59-109€"},
        ("Paris", "Marseille", "voiture"): {"distance_km": 775, "duree": "7h15", "peages": "65€"},
    }
    data = itineraires.get(
        (depart, arrivee, mode),
        {"distance_km": 300, "duree": "3h00", "info": "Estimation approximative"},
    )
    data["depart"] = depart
    data["arrivee"] = arrivee
    data["mode"] = mode
    return data


# ─────────────────────────────────────────────
# 4. Client Albert API
# ─────────────────────────────────────────────
client = OpenAI(
    base_url="https://albert.api.etalab.gouv.fr/v1",
    api_key="sk-",
)

MODEL = "openweight-large"


# ─────────────────────────────────────────────
# 5. Boucle de conversation avec tool calling
# ─────────────────────────────────────────────
def chat(user_message: str, messages: list | None = None) -> str:
    """Conversation avec gestion automatique du tool calling."""
    if messages is None:
        messages = [
            {
                "role": "system",
                "content": (
                    "Tu es un assistant intelligent. Utilise les outils à disposition "
                    "quand c'est pertinent. Réponds en français."
                ),
            }
        ]

    messages.append({"role": "user", "content": user_message})
    print(f"\n{'─'*50}")
    print(f"👤 {user_message}")

    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=registry.schemas,
        tool_choice="auto",
    )

    msg = response.choices[0].message

    # Boucle de résolution des tool calls
    while msg.tool_calls:
        messages.append(msg)

        for tc in msg.tool_calls:
            args = json.loads(tc.function.arguments)
            print(f"   🔧 {tc.function.name}({', '.join(f'{k}={v!r}' for k, v in args.items())})")

            result = registry.execute(tc.function.name, args)
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": json.dumps(result, ensure_ascii=False),
            })

        response = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=registry.schemas,
            tool_choice="auto",
        )
        msg = response.choices[0].message

    messages.append(msg)
    print(f"🤖 {msg.content}")
    return msg.content


# ─────────────────────────────────────────────
# 6. Debug : voir les schémas générés
# ─────────────────────────────────────────────
def show_schemas():
    """Affiche les schémas JSON générés automatiquement."""
    print("\n📋 Schémas auto-générés :")
    print(json.dumps(registry.schemas, indent=2, ensure_ascii=False))


# ─────────────────────────────────────────────
# 7. Utilisation
# ─────────────────────────────────────────────
if __name__ == "__main__":
    # Voir ce que le décorateur a généré automatiquement
    show_schemas()

    # Conversation
    chat("Quel temps fait-il à Lyon ?")
    chat("Je veux aller de Paris à Marseille en train, et manger sur place.")