| """ | | | 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.") |