File size: 4,290 Bytes
92c4ae6 | 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 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | """
Base Learning Engine
Abstract interface for self-improving agent modules. Both MementoEngine
(skill generation) and AlphaEvolverEngine (skill optimization) implement
this interface, enabling a unified lifecycle:
analyze_episode → propose_code_change → validate_change
"""
import logging
from abc import ABC, abstractmethod
from typing import Any, Protocol, runtime_checkable
from sqlalchemy.orm import Session
logger = logging.getLogger(__name__)
@runtime_checkable
class SandboxProtocol(Protocol):
"""
Abstract sandbox interface for executing untrusted code.
Upstream uses ContainerSandbox (Docker).
SaaS implementations can inject SandboxExecutionService (Fly.io).
"""
async def execute_raw_python(
self,
tenant_id: str,
code: str,
input_params: dict[str, Any],
timeout: int = 60,
safety_level: str = "MEDIUM_RISK",
**kwargs,
) -> dict[str, Any]:
"""
Execute raw Python code in an isolated sandbox.
Returns:
{
"status": "success" | "failed",
"output": str,
"execution_seconds": float,
"execution_id": str,
}
"""
...
class BaseLearningEngine(ABC):
"""
Unified interface for self-improving agent modules.
Subclasses must implement three core lifecycle methods:
1. analyze_episode — read and interpret execution data
2. propose_code_change — generate a code modification
3. validate_change — execute in sandbox and assess fitness
"""
def __init__(
self,
db: Session,
llm_service: Any | None = None,
sandbox: SandboxProtocol | None = None,
):
self.db = db
self.llm = llm_service
self.sandbox = sandbox
@abstractmethod
async def analyze_episode(self, episode_id: str, **kwargs) -> dict[str, Any]:
"""
Read and interpret an episode's execution data.
Returns a structured analysis dict containing:
- task_description, error_trace, tool_calls (for failures)
- latency, token_usage, edge_case_signals (for successes)
"""
@abstractmethod
async def propose_code_change(
self, context: dict[str, Any], **kwargs
) -> str:
"""
Generate a code modification proposal via LLM.
Args:
context: Analysis output from analyze_episode()
Returns:
Generated Python code string
"""
@abstractmethod
async def validate_change(
self, code: str, test_inputs: list[dict[str, Any]], tenant_id: str, **kwargs
) -> dict[str, Any]:
"""
Execute proposed code in sandbox and assess fitness.
Returns:
{
"passed": bool,
"proxy_signals": dict,
"execution_result": dict,
}
"""
def _get_llm_service(self):
"""Get LLM service with graceful fallback."""
if self.llm is not None:
return self.llm
try:
from core.llm_service import get_llm_service
self.llm = get_llm_service()
return self.llm
except Exception as e:
logger.warning(
f"LLM service unavailable — Auto-Dev features requiring LLM will be skipped: {e}"
)
return None
def _get_sandbox(self):
"""Get sandbox with graceful fallback to ContainerSandbox."""
if self.sandbox is not None:
return self.sandbox
try:
from core.auto_dev.container_sandbox import ContainerSandbox
self.sandbox = ContainerSandbox()
return self.sandbox
except Exception as e:
logger.warning(f"Sandbox unavailable — validation will be skipped: {e}")
return None
def _strip_markdown_fences(self, code: str) -> str:
"""Strip markdown code fences from LLM output."""
code = code.strip()
if code.startswith("```python"):
code = code[len("```python") :]
elif code.startswith("```"):
code = code[3:]
if code.endswith("```"):
code = code[:-3]
return code.strip()
|