Spaces:
Sleeping
Sleeping
File size: 4,552 Bytes
8e3a425 | 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 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | # core/branch_executor.py
"""
Exécuteur de branches – isole chaque hypothèse dans un sandbox.
Si le code est vide, évalue la description sémantiquement.
"""
import logging
import time
import asyncio
from typing import Dict, Any, Optional
from concurrent.futures import ThreadPoolExecutor
from sandbox.secure_executor import execute_safe
from core.hypothesis_tree import HypothesisTree, Branch, BranchStatus
logger = logging.getLogger("vortex.branch_executor")
class BranchExecutor:
"""
Exécute une branche (hypothèse) dans un sandbox isolé.
Retourne le résultat, le score et les éventuelles erreurs.
"""
def __init__(self, timeout: int = 30):
self.timeout = timeout
self.executor = ThreadPoolExecutor(max_workers=4)
async def execute_branch(self, branch: Branch, tree: HypothesisTree) -> Dict:
"""
Exécute une branche et met à jour l'arbre.
Retourne un dict avec le résultat.
"""
branch_id = branch.id
tree.update_branch(branch_id, BranchStatus.RUNNING)
# Si le code est vide, évaluer uniquement la description
if not branch.code.strip():
score = self._evaluate_hypothesis_text(branch.hypothesis, tree.task)
tree.update_branch(
branch_id,
BranchStatus.SUCCESS,
score=score,
result={"evaluation": "descriptive", "score": score},
execution_time=0.0
)
return {"branch_id": branch_id, "success": True, "score": score, "result": "Évalué par description"}
# Sinon, exécuter le code dans le sandbox
t0 = time.time()
try:
result = await asyncio.get_event_loop().run_in_executor(
self.executor,
execute_safe,
branch.code,
self.timeout
)
execution_time = time.time() - t0
# Évaluation du résultat
score = self._evaluate_result(result, branch.hypothesis)
tree.update_branch(
branch_id,
BranchStatus.SUCCESS,
score=score,
result=result,
execution_time=execution_time
)
return {"branch_id": branch_id, "success": True, "score": score, "result": result}
except Exception as e:
execution_time = time.time() - t0
error_msg = str(e)
tree.update_branch(
branch_id,
BranchStatus.FAILED,
score=0.0,
error=error_msg,
execution_time=execution_time
)
return {"branch_id": branch_id, "success": False, "error": error_msg}
def _evaluate_hypothesis_text(self, text: str, task: str) -> float:
"""
Évalue une hypothèse basée sur le texte seul (pas de code exécuté).
Critères : pertinence, couverture des mots-clés, longueur.
"""
score = 0.3 # base
text_lower = text.lower()
task_lower = task.lower()
# Mots-clés importants pour la tâche actuelle (peut être adapté)
keywords = [
"compétences", "skill", "json", "visualisation", "arbre", "cerveau",
"three", "vis", "chromadb", "injection", "prompt", "structure",
"validation", "pydantic", "schema", "plugin", "module"
]
matched = sum(1 for kw in keywords if kw in text_lower)
score += min(0.4, matched * 0.08)
# Longueur (bonus si > 200 caractères)
if len(text) > 200:
score += 0.15
if len(text) > 500:
score += 0.15
# Pertinence par rapport à la tâche (mots communs)
common = set(task_lower.split()) & set(text_lower.split())
if common:
score += min(0.2, len(common) * 0.05)
return min(1.0, round(score, 3))
def _evaluate_result(self, result: Any, hypothesis: str) -> float:
"""
Heuristique d'évaluation d'un résultat (0‑1) lorsque le code a été exécuté.
"""
if result is None:
return 0.0
if isinstance(result, dict):
if "result" in result or "output" in result or "success" in result:
if result.get("success", False):
return 0.8
return 0.3
if isinstance(result, str):
if len(result) > 100:
return 0.7
return 0.4
return 0.5 |