spb/ka2 Public

1# Author: Simon-Pierre Boucher2# Contact: contact@spboucher.ai3"""Orchestrateur ka4 — version plus large / agressive / capable.45Nouveautés vs ka2 :6- Crawling CONCURRENT : fetch (Firecrawl/Scrapfly) + extraction (Haiku) en parallèle7  via un pool de threads ; les écritures SQLite restent sérialisées sur le thread8  principal (une seule connexion) pour rester sûres.9- Frontière AGRESSIVE : chaque page auto-alimente la file avec les liens découverts10  (nouveaux domaines priorisés) pour élargir la couverture en continu.11- Contrôleur agentique sur un modèle PLUS PUISSANT (Sonnet), extraction sur Haiku.12"""1314from __future__ import annotations1516from concurrent.futures import ThreadPoolExecutor, as_completed17from typing import Any18from urllib.parse import urlparse1920from anthropic import Anthropic2122from .config import Config23from .extractor import Extractor24from .llm_controller import Controller25from .scraper import Scraper26from .storage import Store2728_SKIP_HOSTS = ("facebook.", "instagram.", "twitter.", "x.com", "linkedin.", "youtube.",29               "tiktok.", "pinterest.", "google.", "maps.", "apple.", "amazon.")30_SKIP_EXT = (".jpg", ".jpeg", ".png", ".pdf", ".zip", ".mp4", ".svg", ".gif", ".webp",31             ".css", ".js", ".xml", ".ico", ".woff", ".woff2")323334class Orchestrator:35    def __init__(self, cfg: Config):36        cfg.validate()37        self.cfg = cfg38        self.anthropic = Anthropic(api_key=cfg.anthropic_api_key)39        self.fc = Scraper(cfg)40        self.store = Store(cfg.db_path)41        # extraction sur le modèle rapide ; contrôleur sur cfg.model (plus puissant)42        self.extractor = Extractor(self.anthropic, cfg.extract_model)43        self._pages_scraped = 044        self._domain_fails: dict[str, int] = {}  # coupe-circuit par domaine (robustesse)4546    # -- fetch pur (THREAD-SAFE : réseau + LLM, aucun accès DB) -------------47    def _fetch_only(self, url: str) -> dict[str, Any]:48        data = self.fc.scrape(url)49        if data.get("blocked"):50            return {"url": url, "blocked": True}51        if data.get("error"):52            return {"url": url, "error": data["error"]}53        md = data.get("markdown", "") or ""54        meta = data.get("metadata", {}) or {}55        title = meta.get("title", "") or ""56        graph = (self.extractor.extract_graph(url, title, md)57                 if md.strip() else {"entities": [], "relations": []})58        return {"url": url, "title": title, "backend": data.get("backend"),59                "markdown": md, "links": data.get("links", []) or [], "graph": graph}6061    # -- persistance (THREAD PRINCIPAL uniquement) -------------------------62    def _save_result(self, r: dict[str, Any]) -> dict[str, Any]:63        url = r["url"]64        if r.get("blocked"):65            self.store.log_event("info", "robots.txt interdit", url)66            return {"entities_found": 0, "relations_found": 0, "blocked": True}67        if r.get("error"):68            self._domain_fails[urlparse(url).netloc.lower()] = \69                self._domain_fails.get(urlparse(url).netloc.lower(), 0) + 170            self.store.log_event("error", r["error"], url)71            return {"entities_found": 0, "relations_found": 0, "error": r["error"]}7273        self.store.record_source(url, r.get("title", ""), r.get("markdown", ""))74        self.store.log_event("scrape", f"{r.get('title') or url} [{r.get('backend')}]", url,75                             {"chars": len(r.get("markdown", ""))})76        graph = r.get("graph", {}) or {}77        # robustesse : le LLM peut occasionnellement renvoyer un item non-dict -> on filtre78        entities = [e for e in (graph.get("entities") or []) if isinstance(e, dict)]79        relations = [x for x in (graph.get("relations") or []) if isinstance(x, dict)]80        idmap: dict[str, int] = {}81        for e in entities:82            eid = self.store.upsert_entity(e, source_url=url)83            if e.get("temp_id"):84                idmap[e["temp_id"]] = eid85        rel = 086        for rr in relations:87            fid, tid = idmap.get(rr.get("from")), idmap.get(rr.get("to"))88            if fid and tid:89                self.store.add_relation(fid, tid, rr.get("type", ""), rr.get("role", ""),90                                        source_url=url, confidence=float(rr.get("confidence", 0.6) or 0.6))91                rel += 192        self._enqueue_frontier(url, r.get("links", []))93        self.store.log_event("extract", f"{len(idmap)} entité(s), {rel} relation(s)", url,94                             {"relations": rel})95        return {"entities_found": len(idmap), "relations_found": rel,96                "links_on_page": (r.get("links", []) or [])[:25]}9798    def _enqueue_frontier(self, url: str, links: list) -> None:99        if self.cfg.frontier_per_page <= 0 or not links:100            return101        base = urlparse(url).netloc.lower()102        picked = []103        for l in links:104            u = l["url"] if isinstance(l, dict) else l105            if not isinstance(u, str) or not u.startswith("http"):106                continue107            h = urlparse(u).netloc.lower()108            if not h or any(s in h for s in _SKIP_HOSTS) or u.lower().endswith(_SKIP_EXT):109                continue110            picked.append(u)111        # priorise les NOUVEAUX domaines (élargit la couverture)112        picked.sort(key=lambda u: urlparse(u).netloc.lower() == base)113        self.store.enqueue(picked[: self.cfg.frontier_per_page])114115    # -- crawling CONCURRENT ----------------------------------------------116    def _tripped(self, url: str) -> bool:117        """Coupe-circuit : True si le domaine a trop échoué dans ce run."""118        return self._domain_fails.get(urlparse(url).netloc.lower(), 0) >= self.cfg.domain_fail_threshold119120    def crawl_batch(self, urls: list[str]) -> dict[str, Any]:121        seen, todo = set(), []122        for u in urls:123            if not u or u in seen:124                continue125            seen.add(u)126            if self._tripped(u):127                continue  # domaine coupé (trop d'échecs)128            if not self.store.source_seen(u):129                todo.append(u)130        remaining = max(0, self.cfg.max_pages_per_run - self._pages_scraped)131        todo = todo[:remaining]132        if not todo:133            return {"pages": 0, "entities_found": 0, "relations_found": 0}134        results = []135        with ThreadPoolExecutor(max_workers=max(1, self.cfg.concurrency)) as ex:136            futs = [ex.submit(self._fetch_only, u) for u in todo]137            for f in as_completed(futs):138                try:139                    results.append(f.result())140                except Exception as e:  # noqa: BLE001141                    self.store.log_event("error", f"fetch: {e}", "")142        tot_e = tot_r = pages = 0143        for r in results:144            out = self._save_result(r)145            if not (r.get("blocked") or r.get("error")):146                self._pages_scraped += 1147                pages += 1148            tot_e += out.get("entities_found", 0)149            tot_r += out.get("relations_found", 0)150        return {"pages": pages, "entities_found": tot_e, "relations_found": tot_r}151152    # -- scrape unitaire (utilisé par l'agent) ----------------------------153    def _scrape_and_extract(self, url: str) -> dict[str, Any]:154        if self._pages_scraped >= self.cfg.max_pages_per_run:155            return {"stopped": "limite de pages atteinte", "entities_found": 0}156        if self.store.source_seen(url):157            return {"skipped": "déjà scrapé", "entities_found": 0}158        r = self._fetch_only(url)159        out = self._save_result(r)160        if not (r.get("blocked") or r.get("error")):161            self._pages_scraped += 1162        return {**out, "url": url, "title": r.get("title", ""), "backend": r.get("backend")}163164    # -- crawling centré-entité (concurrent) ------------------------------165    _DEEP_KEYWORDS = ("about", "a-propos", "apropos", "propos", "equipe", "team", "notre-equipe",166                      "leadership", "direction", "gouvernance", "membres", "conseil", "contact",167                      "coordonnees", "nous-joindre", "qui-sommes", "notre-histoire", "carrieres")168169    def deep_dive(self, url: str, max_pages: int = 8) -> dict[str, Any]:170        parts = urlparse(url if "://" in url else "http://" + url)171        root = f"{parts.scheme or 'https'}://{parts.netloc or parts.path}"172        try:173            links = self.fc.map(root)174        except Exception as e:  # noqa: BLE001175            links = [root]176            self.store.log_event("error", f"map deep_dive: {e}", root)177178        def score(u: str) -> int:179            lu = u.lower()180            return sum(2 if k in lu else 0 for k in self._DEEP_KEYWORDS)181        targets = [t for t in sorted(set(links), key=score, reverse=True) if score(t) > 0][: max_pages - 1]182        targets = [root] + targets183        self.store.log_event("agent", f"deep_dive {parts.netloc}: {len(targets)} page(s), concurrent", root)184        res = self.crawl_batch(targets)185        return {"domain": parts.netloc, "pages": res["pages"],186                "entities_found": res["entities_found"], "relations_found": res["relations_found"]}187188    # -- mode agent (contrôleur = modèle plus puissant) -------------------189    def run_agent(self, goal: str, seed: str | None, max_steps: int, should_abort=None) -> str:190        self.store.log_event("agent", f"Mission: {goal}", "")191192        def t_search(inp: dict[str, Any]) -> dict[str, Any]:193            res = self.fc.search(inp["query"], int(inp.get("limit", 10)))194            self.store.log_event("search", inp["query"], "", {"results": len(res)})195            return {"results": [{"url": r.get("url"), "title": r.get("title"),196                                 "desc": r.get("description")} for r in res]}197198        def t_map(inp: dict[str, Any]) -> dict[str, Any]:199            links = self.fc.map(inp["url"], inp.get("search"))200            self.store.enqueue(links[:150])201            self.store.log_event("map", inp["url"], inp["url"], {"links": len(links)})202            return {"count": len(links), "sample": links[:40]}203204        def t_scrape(inp: dict[str, Any]) -> dict[str, Any]:205            return self._scrape_and_extract(inp["url"])206207        def t_deep(inp: dict[str, Any]) -> dict[str, Any]:208            return self.deep_dive(inp["url"], int(inp.get("max_pages", 8)))209210        def t_enqueue(inp: dict[str, Any]) -> dict[str, Any]:211            return {"enqueued": self.store.enqueue(inp.get("urls", []))}212213        def t_drain(inp: dict[str, Any]) -> dict[str, Any]:214            return self.crawl_batch([r["url"] for r in self.store.pending_batch(215                int(inp.get("count", self.cfg.concurrency)))])216217        def t_plan(inp: dict[str, Any]) -> dict[str, Any]:218            self.store.log_event("agent", "PLAN: " + " | ".join(inp.get("plan", [])), "",219                                 {"rationale": inp.get("rationale", "")})220            return {"ack": True}221222        def t_note(inp: dict[str, Any]) -> dict[str, Any]:223            self.store.log_event("agent", "NOTE: " + inp.get("note", ""), "")224            return {"ack": True}225226        controller = Controller(227            self.anthropic, self.cfg.model,228            {"search_web": t_search, "map_site": t_map, "scrape_page": t_scrape,229             "deep_dive": t_deep, "crawl_queue": t_drain, "enqueue_urls": t_enqueue,230             "update_plan": t_plan, "note_progress": t_note},231            on_event=lambda kind, msg: self.store.log_event(kind, msg, ""),232            max_context_tokens=self.cfg.max_context_tokens,233        )234        summary = controller.run(goal=goal, seed=seed, max_steps=max_steps, should_abort=should_abort)235        self.store.log_event("agent", "Mission conclue", "", {"summary": summary})236        return summary237238    # -- pipeline (concurrent) --------------------------------------------239    def run_pipeline(self, query: str, limit: int = 10, per_site_pages: int = 3) -> dict[str, Any]:240        results = self.fc.search(query, limit)241        urls = [r.get("url") for r in results if r.get("url")]242        res = self.crawl_batch(urls)243        # approfondir les domaines découverts244        for u in urls[:per_site_pages]:245            if self._pages_scraped >= self.cfg.max_pages_per_run:246                break247            self.deep_dive(u, max_pages=per_site_pages + 2)248        return {"query": query, "pages_scraped": self._pages_scraped,249                "entities_found": res["entities_found"], "stats": self.store.stats()}250251    # -- file d'attente (drainée en concurrence) --------------------------252    def drain_queue(self, max_pages: int | None = None, should_abort=None) -> dict[str, Any]:253        budget = max_pages or self.cfg.max_pages_per_run254        processed = 0255        while processed < budget and self._pages_scraped < self.cfg.max_pages_per_run:256            if should_abort and should_abort():257                break258            rows = self.store.pending_batch(min(self.cfg.concurrency, budget - processed),259                                            max_attempts=self.cfg.max_attempts)260            if not rows:261                break262            self.crawl_batch([r["url"] for r in rows])263            # robustesse : succès -> done ; échec -> retry (attempts++) puis dead-letter264            for r in rows:265                if self.store.source_seen(r["url"]):266                    self.store.mark(r["id"], "done")267                else:268                    self.store.bump_attempt(r["id"], self.cfg.max_attempts)269            processed += len(rows)270        return {"processed": processed, "stats": self.store.stats()}271272    def close(self) -> None:273        self.store.close()274