Spaces:
Sleeping
Sleeping
File size: 4,140 Bytes
380a7fc 8a4176f 380a7fc 8a4176f 380a7fc 8a4176f 380a7fc 8a4176f 380a7fc 8a4176f 380a7fc 8a4176f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | """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"
|