# Author: Simon-Pierre Boucher # Contact: contact@spboucher.ai """Contrôleur agentique rigoureux : Claude Haiku 4.5 pilote la cartographie. Points clés : - Méthodologie imposée (plan -> découverte -> collecte -> vérification). - Garde anti-boucle : on rejette les actions déjà faites (mêmes URL/recherches). - Journal de progression persistant (outil note_progress). - AUTO-COMPACTION du contexte : quand l'historique dépasse un seuil de tokens, les anciens échanges sont résumés par le modèle et remplacés par un condensé, ce qui permet des missions très longues sans dépasser la fenêtre de contexte. """ from __future__ import annotations import json from typing import Any, Callable, Optional from anthropic import Anthropic TOOLS = [ { "name": "update_plan", "description": "Établis ou révise ton plan de mission : étapes, secteurs/régions à couvrir, " "et prochaines actions. À appeler au début, puis quand la stratégie change.", "input_schema": { "type": "object", "properties": { "plan": {"type": "array", "items": {"type": "string"}}, "rationale": {"type": "string"}, }, "required": ["plan"], }, }, { "name": "search_web", "description": "Recherche web (Firecrawl). Retourne titres, URLs et extraits.", "input_schema": { "type": "object", "properties": { "query": {"type": "string"}, "limit": {"type": "integer", "default": 8}, }, "required": ["query"], }, }, { "name": "map_site", "description": "Liste rapidement les URLs d'un site (découverte). 'search' filtre par mot-clé.", "input_schema": { "type": "object", "properties": {"url": {"type": "string"}, "search": {"type": "string"}}, "required": ["url"], }, }, { "name": "scrape_page", "description": "Récupère le contenu d'une page ET en extrait/enregistre le graphe " "(entités + relations typées avec rôles). Retourne un résumé.", "input_schema": { "type": "object", "properties": {"url": {"type": "string"}}, "required": ["url"], }, }, { "name": "deep_dive", "description": "APPROFONDIT une organisation : mappe son site et scrape automatiquement ses " "pages clés (à propos, équipe, direction, contact) pour extraire un profil complet ET les " "relations (personnes ↔ organisation avec rôles). À privilégier sur toute org prometteuse " "pour bâtir un vrai graphe plutôt que des entités isolées.", "input_schema": { "type": "object", "properties": { "url": {"type": "string", "description": "URL ou domaine de l'organisation"}, "max_pages": {"type": "integer", "default": 5}, }, "required": ["url"], }, }, { "name": "crawl_queue", "description": "Traite EN PARALLÈLE un lot d'URLs déjà en file (crawling concurrent). " "La frontière se remplit automatiquement à chaque page ; sers-t'en pour avancer vite.", "input_schema": { "type": "object", "properties": {"count": {"type": "integer", "default": 6}}, }, }, { "name": "enqueue_urls", "description": "Ajoute des URLs prometteuses à la file d'exploration long terme.", "input_schema": { "type": "object", "properties": {"urls": {"type": "array", "items": {"type": "string"}}}, "required": ["urls"], }, }, { "name": "note_progress", "description": "Consigne une note de progression (couverture, angles morts, pistes). " "Sers-t'en pour rester rigoureux et éviter les redites.", "input_schema": { "type": "object", "properties": {"note": {"type": "string"}}, "required": ["note"], }, }, { "name": "finish", "description": "Termine la mission avec un bilan (couverture atteinte, entités clés, reste à faire).", "input_schema": { "type": "object", "properties": {"summary": {"type": "string"}}, "required": ["summary"], }, }, ] SYSTEM = ( "Tu es ka6, l'agent FLAGSHIP du Groupe KA. Objectif initial : cartographier à grande échelle " "l'écosystème des CRÉATEURS DE CONTENU et INFLUENCEURS québécois (et l'industrie autour : agences " "de talents/influence, marques, plateformes, collectifs).\n\n" "Pour chaque créateur : nom, handle, plateformes + nombre d'abonnés, niche, langues, région, et " "les RELATIONS (représenté par une agence, collabore avec / sponsorisé par des marques, membre d'un " "collectif, publie sur telles plateformes).\n\n" "MÉTHODE (agressive, flagship) :\n" "1. PLANIFIE : update_plan (niches × plateformes, agences, palmarès, marques).\n" "2. DÉCOUVRE en volume : search_web (ex. « top influenceurs québécois mode TikTok », " "« agences d'influence Montréal », « créateurs YouTube Québec gaming », palmarès, répertoires).\n" "3. APPROFONDIS : deep_dive sur les annuaires, rosters d'agences et pages de créateurs (scrape " "parallèle) — c'est là que se trouvent handles, abonnés et relations. Outil PRINCIPAL.\n" "4. AVANCE VITE : crawl_queue traite en parallèle la frontière qui se remplit automatiquement.\n" "5. COMPLÈTE : scrape_page pour une page précise; enqueue_urls pour de nouvelles pistes.\n" "6. VÉRIFIE : note_progress (couverture par niche/plateforme, qualité des relations).\n" "7. finish quand la limite est atteinte.\n\n" "RÈGLES : données PUBLIQUES uniquement; priorise le Québec et les LIENS créateur↔agence↔marque; " "ne répète pas une action; une action à la fois; raisonne brièvement. robots.txt/débit gérés par les outils." ) ToolFn = Callable[[dict[str, Any]], dict[str, Any]] EventFn = Callable[[str, str], None] class Controller: def __init__( self, client: Anthropic, model: str, tool_impls: dict[str, ToolFn], on_event: Optional[EventFn] = None, max_context_tokens: int = 120_000, keep_last_messages: int = 6, ): self.client = client self.model = model self.tools = tool_impls self.on_event = on_event or (lambda kind, msg: None) self.max_context_tokens = max_context_tokens self.keep_last_messages = keep_last_messages # garde anti-boucle self._done_actions: set[str] = set() # -- boucle principale ------------------------------------------------- def run(self, goal: str, seed: str | None = None, max_steps: int = 40, should_abort: Optional[Callable[[], bool]] = None) -> str: user = f"Objectif de mission : {goal}" if seed: user += f"\nPoint de départ suggéré : {seed}" messages: list[dict[str, Any]] = [{"role": "user", "content": user}] for step in range(max_steps): if should_abort and should_abort(): self.on_event("agent", "Mission interrompue par l'utilisateur.") return "Interrompu par l'utilisateur." messages = self._maybe_compact(messages) resp = self.client.messages.create( model=self.model, max_tokens=1600, system=SYSTEM, tools=TOOLS, messages=messages, ) messages.append({"role": "assistant", "content": resp.content}) tool_uses = [b for b in resp.content if b.type == "tool_use"] for b in resp.content: if b.type == "text" and b.text.strip(): self.on_event("agent", b.text.strip()[:500]) if not tool_uses: return self._final(resp) results = [] for tu in tool_uses: if tu.name == "finish": summary = tu.input.get("summary", "Mission terminée.") self.on_event("agent", f"FIN : {summary[:300]}") return summary out = self._dispatch(tu.name, tu.input) results.append( {"type": "tool_result", "tool_use_id": tu.id, "content": _stringify(out)} ) messages.append({"role": "user", "content": results}) return "Budget d'étapes épuisé (mission non conclue explicitement)." # -- exécution d'un outil (avec garde anti-boucle) -------------------- def _dispatch(self, name: str, inp: dict[str, Any]) -> Any: key = name + ":" + json.dumps(inp, ensure_ascii=False, sort_keys=True) if name in ("search_web", "map_site", "scrape_page", "deep_dive") and key in self._done_actions: return {"skipped": "action déjà effectuée — change d'angle ou d'URL."} self._done_actions.add(key) self.on_event("agent", f"→ {name}({_short(inp)})") fn = self.tools.get(name) if not fn: return {"error": f"outil inconnu: {name}"} try: return fn(inp) except Exception as e: # noqa: BLE001 self.on_event("error", f"{name}: {e}") return {"error": str(e)} # -- auto-compaction --------------------------------------------------- def _maybe_compact(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]: if self._estimate_tokens(messages) < self.max_context_tokens: return messages if len(messages) <= self.keep_last_messages + 2: return messages cut = self._safe_cut(messages) if cut <= 1: return messages head, tail = messages[:cut], messages[cut:] summary = self._summarize(head) self.on_event("info", f"Contexte compacté : {len(head)} messages résumés.") return [{"role": "user", "content": f"[MÉMOIRE COMPACTÉE — progrès à ce jour]\n{summary}"}] + tail def _safe_cut(self, messages: list[dict[str, Any]]) -> int: """Coupe à une frontière valide : juste après un message user (tool_result), pour que le premier message conservé soit un assistant et que toute paire tool_use/tool_result reste intacte.""" target = len(messages) - self.keep_last_messages cut = 0 for i in range(min(target, len(messages) - 1)): if messages[i]["role"] == "user": cut = i + 1 return cut def _summarize(self, head: list[dict[str, Any]]) -> str: transcript = _stringify(head, limit=40000) try: resp = self.client.messages.create( model=self.model, max_tokens=1200, system=( "Résume l'avancement d'une mission de cartographie web pour poursuivre sans perdre " "le contexte. Conserve : plan courant, secteurs/régions couverts, URLs et domaines " "déjà traités, entités clés trouvées, pistes en attente, angles morts. Sois dense et factuel." ), messages=[{"role": "user", "content": f"Historique à résumer :\n{transcript}"}], ) return "".join(b.text for b in resp.content if b.type == "text").strip() or "(résumé vide)" except Exception as e: # noqa: BLE001 return f"(échec du résumé: {e})" def _estimate_tokens(self, messages: list[dict[str, Any]]) -> int: return len(_stringify(messages, limit=10**9)) // 3 def _final(self, resp) -> str: return "".join(b.text for b in resp.content if b.type == "text").strip() or "(fin sans appel d'outil)" def _short(d: dict[str, Any]) -> str: s = json.dumps(d, ensure_ascii=False) return s if len(s) <= 120 else s[:117] + "..." def _stringify(obj: Any, limit: int = 6000) -> str: try: s = json.dumps(obj, ensure_ascii=False, default=_json_default) except Exception: s = str(obj) return s[:limit] def _json_default(o: Any) -> Any: # Les blocs de contenu Anthropic ne sont pas JSON-sérialisables nativement. for attr in ("model_dump", "dict", "to_dict"): if hasattr(o, attr): try: return getattr(o, attr)() except Exception: pass return str(o)