Spaces:
Running
Running
| """Composition côté Space : routeur -> stratégie (backend + vérificateur) -> contrôles d'ancrage. | |
| ExtSession a la forme de UniqueSession (`generate` et `info`) : `stream_session` l'utilise sans modification, | |
| et son `info` apparaît tout seul dans le rapport, comme la réflexion de la v3.8. La boucle M.O.U.H. certifie | |
| toujours le résultat final : l'extension propose, elle ne certifie jamais. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import time | |
| from typing import Optional | |
| from mouh.learning.prompts import chat_messages | |
| from .. import enabled | |
| from ..contracts import Candidate, Files | |
| from ..grounding.api_check import ApiCheck | |
| from ..grounding.citations import CitationCheck | |
| from .router import ModeRouter | |
| from .tree import AdaptiveStrategy | |
| from .verify import PytestVerifier | |
| log = logging.getLogger(__name__) | |
| class ExtSession: | |
| def __init__(self, fallback, backend, strategy=None, verifier=None, grounders=None, router=None, | |
| memory=None) -> None: | |
| self.fallback, self.backend = fallback, backend | |
| self.strategy = strategy or AdaptiveStrategy() | |
| self.verifier = verifier or PytestVerifier() | |
| self.grounders = tuple(grounders) if grounders is not None else (ApiCheck(), CitationCheck()) | |
| self.router = router or ModeRouter() | |
| if memory is None and enabled("MEMORY"): | |
| from ..memory.gated import shared_memory | |
| memory = shared_memory() | |
| self.memory = memory | |
| self.info: dict = {"extension": getattr(self.strategy, "name", "?"), "backend": backend.name, "essais": []} | |
| def generate(self, request) -> Optional[Files]: | |
| try: | |
| return self._generate(request) | |
| except Exception as exc: # noqa: BLE001 — une extension ne casse jamais une session | |
| log.warning("extension en échec, repli : %s", exc) | |
| self.info["repli"] = f"{type(exc).__name__}: {str(exc)[:200]}" | |
| return self.fallback.generate(request) if self.fallback is not None else None | |
| def _generate(self, request) -> Optional[Files]: | |
| t0 = time.time() | |
| tests = dict(request.task.test_files) | |
| instruction = request.task.instruction | |
| plan = self.router.plan(instruction, tests, request.attempt) | |
| examples = None | |
| if self.memory is not None: | |
| try: | |
| examples = self.memory.examples(instruction, tests, plan.level) or None | |
| except Exception as exc: # noqa: BLE001 — sans mémoire plutôt que pas de réponse | |
| log.warning("mémoire indisponible : %s", exc) | |
| def build(previous=None, diagnostics=None): | |
| prev = previous if previous is not None else request.previous_files | |
| if diagnostics is None: | |
| diagnostics = [d.message for d in getattr(request, "diagnostics", ())] | |
| return chat_messages(instruction, tests, dict(prev) if prev else None, diagnostics if prev else None, | |
| examples=examples) | |
| cands = self.strategy.solve(request, plan, self.backend, self.verifier, build) | |
| for c in cands: | |
| c.issues = [i for g in self.grounders for i in g.check(c.files, None)] # sans notes : aucune URL n'est sourcée | |
| cands.sort(key=Candidate.key, reverse=True) | |
| best: Optional[Candidate] = cands[0] if cands else None | |
| self.info["essais"].append({ | |
| "tentative": request.attempt, "niveau": plan.level, "mode": plan.mode, "budget": plan.budget, | |
| "propositions": len(cands), "tests_ok": bool(best and best.ok), | |
| "tests_reussis": best.passed if best else 0, "remarques": best.issues[:3] if best else [], | |
| "exemples_memoire": len(examples or []), "secondes": round(time.time() - t0, 2)}) | |
| return best.files if best else None | |
| def tree_session(load_unique): | |
| """Fabrique UNIQUE_EXT_TREE : modèle transformers déjà chargé par le Space + raisonnement adaptatif.""" | |
| from ..backends.hf import HFBackend | |
| base, note = load_unique() | |
| return ExtSession(base, HFBackend(base)), f"{note} · recherche arborescente adaptative" | |