"""Communication avec l'API de Claude (Anthropic).""" from __future__ import annotations import base64 import logging from dataclasses import dataclass from pathlib import Path import anthropic from .config import Settings logger = logging.getLogger("chatbot.claude") SUPPORTED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"} DEFAULT_PERSONALITY = ( "Tu es un membre sympathique d'un serveur Discord francophone. " "Tu réponds toujours en français, avec un ton naturel, amical et décontracté, " "et des réponses plutôt courtes, adaptées à une discussion de chat. " "Chaque message que tu reçois indique la date, l'heure et le pseudo de son auteur ; " "plusieurs personnes peuvent parler dans la même conversation. " "Quand on te montre une image, décris ce que tu vois puis réponds à la question s'il y en a une." ) # Ajouté automatiquement à la fin de la personnalité, quel que soit le contenu du fichier. MEMORY_NOTE = ( "\n\nNote technique sur ta mémoire : les images ne sont pas conservées d'un message à l'autre. " "Dans les messages précédents, la mention « [image jointe : …] » signifie qu'une image a bien été envoyée " "et que tu l'as réellement vue et analysée à ce moment-là ; ta réponse d'alors en décrit le contenu. " "Ne dis donc jamais que tu ne l'as pas vue, et ne t'excuse pas à ce sujet. " "Si on te demande un détail de cette image que ta réponse précédente ne mentionnait pas, " "explique simplement que tu ne gardes pas les images en mémoire et propose qu'on te la renvoie." ) class ClaudeError(Exception): """Erreur lors de l'appel à Claude, avec un message affichable sur Discord.""" def __init__(self, user_message: str): super().__init__(user_message) self.user_message = user_message @dataclass class ImageData: media_type: str data: bytes class ClaudeClient: def __init__(self, settings: Settings): self.settings = settings # Le SDK réessaie tout seul les erreurs temporaires (surcharge, limite de requêtes…). self._client = anthropic.AsyncAnthropic( api_key=settings.anthropic_api_key, timeout=settings.api_timeout, max_retries=3, ) self._personality_cache: tuple[float, str] | None = None async def close(self) -> None: await self._client.close() # ------------------------------------------------------------ personnalité def personality(self) -> str: """Lit personality_prompt.txt (relu automatiquement s'il a été modifié).""" path: Path = self.settings.personality_file try: mtime = path.stat().st_mtime if self._personality_cache is None or self._personality_cache[0] != mtime: text = path.read_text(encoding="utf-8").strip() self._personality_cache = (mtime, text or DEFAULT_PERSONALITY) logger.info("Personnalité chargée depuis %s", path) return self._personality_cache[1] except FileNotFoundError: if self._personality_cache != (0.0, DEFAULT_PERSONALITY): logger.warning("%s introuvable : utilisation de la personnalité par défaut.", path) self._personality_cache = (0.0, DEFAULT_PERSONALITY) return DEFAULT_PERSONALITY # ------------------------------------------------------------------ appel @staticmethod def build_user_content(text: str, image: ImageData | None = None) -> str | list[dict]: if image is None: return text return [ { "type": "image", "source": { "type": "base64", "media_type": image.media_type, "data": base64.standard_b64encode(image.data).decode("ascii"), }, }, {"type": "text", "text": text}, ] async def ask(self, history: list[dict], user_text: str, image: ImageData | None = None) -> str: """Envoie l'historique + le nouveau message, et renvoie la réponse de Claude.""" messages = list(history) + [{"role": "user", "content": self.build_user_content(user_text, image)}] params: dict = { "model": self.settings.model, "max_tokens": self.settings.max_tokens, "system": self.personality() + MEMORY_NOTE, "messages": messages, } # Par défaut, Sonnet 5.5 réfléchit avant de répondre, et cette réflexion consomme # le budget MAX_TOKENS. "between_tools" désactive la réflexion préalable # (le bot n'utilisant pas d'outils, il n'y a alors aucune réflexion). params["thinking"] = {"type": "adaptive"} if self.settings.thinking else {"type": "between_tools"} if self.settings.effort: params["extra_body"] = {"output_config": {"effort": self.settings.effort}} try: response = await self._client.messages.create(**params) except anthropic.RateLimitError: logger.warning("Limite de requêtes Anthropic atteinte.") raise ClaudeError("Je reçois trop de messages en ce moment, réessaie dans une minute.") except anthropic.APITimeoutError: logger.warning("Délai dépassé lors de l'appel à Claude.") raise ClaudeError("Je mets trop de temps à répondre, réessaie dans un instant.") except anthropic.APIConnectionError as exc: logger.error("Connexion à l'API impossible : %s", exc) raise ClaudeError("Je n'arrive pas à joindre mon cerveau pour l'instant, réessaie plus tard.") except anthropic.BadRequestError as exc: logger.error("Requête refusée par l'API : %s", exc) if image is not None: raise ClaudeError("Je n'ai pas réussi à lire cette image, essaie avec une autre.") raise ClaudeError("Ta demande a été refusée par l'API, essaie de la reformuler.") except anthropic.AuthenticationError: logger.critical("Clé API Anthropic invalide : vérifie ANTHROPIC_API_KEY dans le .env.") raise ClaudeError("Je suis mal configuré, préviens l'administrateur du bot.") except anthropic.APIStatusError as exc: logger.error("Erreur de l'API (%s) : %s", exc.status_code, exc) raise ClaudeError("Une erreur est survenue de mon côté, réessaie dans un instant.") text = "".join(block.text for block in response.content if block.type == "text").strip() self._log_usage(response) if response.stop_reason == "max_tokens": logger.warning("Réponse coupée par MAX_TOKENS (%s) : augmente MAX_TOKENS dans le .env " "si cela arrive souvent.", self.settings.max_tokens) text += " […]" if not text: logger.warning("Réponse vide de Claude (stop_reason=%s).", response.stop_reason) raise ClaudeError("Je n'ai rien trouvé à répondre à ça.") return text def _log_usage(self, response) -> None: usage = response.usage cost = (usage.input_tokens * self.settings.price_input_per_mtok + usage.output_tokens * self.settings.price_output_per_mtok) / 1_000_000 logger.info( "Appel Claude (%s) — entrée : %s tokens, sortie : %s tokens, coût estimé : %.5f $", response.model, usage.input_tokens, usage.output_tokens, cost, )