Spaces:
Sleeping
Sleeping
| """ | |
| core/rsi_loop.py - Iterative self-improvement loop (RSI) | |
| """ | |
| import logging | |
| from typing import Dict, Any | |
| logger = logging.getLogger("vortex.rsi_loop") | |
| try: | |
| from core.memory import MemoryTier | |
| MEMORY_TIER_OK = True | |
| except ImportError: | |
| MemoryTier = None | |
| MEMORY_TIER_OK = False | |
| class RSILoop: | |
| def __init__(self, llm_engine, sandbox_executor, memory=None, error_tree=None): | |
| self.llm = llm_engine | |
| self.sandbox = sandbox_executor | |
| self.memory = memory | |
| self.error_tree = error_tree | |
| self.max_attempts = 3 | |
| async def run_one_cycle(self, domain: str) -> Dict[str, Any]: | |
| logger.info(f"RSI: Starting cycle on '{domain}'") | |
| prompt = self._build_initial_prompt(domain) | |
| solution = None | |
| feedback = None | |
| for attempt in range(1, self.max_attempts + 1): | |
| logger.info(f"RSI: Attempt {attempt}/{self.max_attempts}") | |
| try: | |
| response = await self.llm.call( | |
| agent="rsi", | |
| system="You are a Python optimization expert. Generate function 'solution()'. Only code, no explanations.", | |
| user=prompt, | |
| max_tokens=400, | |
| temperature=0.4 | |
| ) | |
| code = response.content if hasattr(response, 'content') else str(response) | |
| except Exception as e: | |
| logger.error(f"RSI: LLM call error: {e}") | |
| code = "" | |
| code = self._clean_code(code) | |
| exec_result = self.sandbox(code, timeout=10) | |
| if exec_result.get("success", False): | |
| solution = code | |
| logger.info(f"RSI: Success at attempt {attempt}") | |
| break | |
| else: | |
| error = exec_result.get("error", "Unknown error") | |
| output = exec_result.get("result", "") | |
| logger.warning(f"RSI: Failed attempt {attempt}: {error[:200]}") | |
| prompt = self._build_feedback_prompt(domain, code, error, output, attempt) | |
| feedback = error | |
| if solution is None: | |
| solution = code if code else "" | |
| score = 0.0 | |
| else: | |
| score = 1.0 | |
| if self.memory and MEMORY_TIER_OK and MemoryTier is not None: | |
| try: | |
| self.memory.ingest(f"[RSI] {domain} -> {solution[:200]}", MemoryTier.EPISODIC, "rsi_loop") | |
| except Exception as e: | |
| logger.warning(f"Memory ingestion error: {e}") | |
| return {"hypothesis": solution, "score": score, "iterations": attempt, "feedback": feedback, "domain": domain} | |
| def _build_initial_prompt(self, domain: str) -> str: | |
| return f"Problem: {domain}\n\nWrite a Python function named 'solution()' that solves this. Self-contained, no external deps. Return appropriate type. Only code, no markdown." | |
| def _build_feedback_prompt(self, domain, code, error, output, attempt): | |
| return f"Problem: {domain}\n\nPrevious code failed (attempt {attempt}):\n\n{code[:500]}\n\nError:\n{error}\n\nOutput:\n{output[:500]}\n\nFix the code. Only corrected code, no markdown." | |
| def _clean_code(self, code: str) -> str: | |
| lines = code.split('\n') | |
| cleaned = [] | |
| in_block = False | |
| for line in lines: | |
| if line.strip().startswith('`'): | |
| in_block = not in_block | |
| continue | |
| if not in_block: | |
| cleaned.append(line) | |
| return '\n'.join(cleaned).strip() |