core/rsi_loop.py

"""
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()