Environments¶

class RLMEnvironment(Protocol):
    name: str

    def system_prompt(self) -> str: ...

    def planner_prompt(
        self,
        task: str,
        memory: list[str],
        trajectory: list[dict[str, Any]],
        step_index: int,
    ) -> str: ...

    def execute_action(
        self,
        action: dict[str, Any],
        execution_engine: Any,
        exec_timeout: int,
        llm_connector: Any | None = None,
    ) -> EnvironmentActionResult: ...

    def doctor_checks(self) -> list[EnvironmentDoctorCheck]: ...

from rlm_code.rlm.pure_rlm_environment import PureRLMEnvironment, PureRLMConfig

class PureRLMEnvironment:
    def __init__(
        self,
        workdir: Path | None = None,
        reward_profile: RLMRewardProfile | dict[str, Any] | None = None,
        config: PureRLMConfig | None = None,
    ):

@dataclass
class PureRLMConfig:
    max_llm_calls: int = 50          # Maximum llm_query() calls per run
    max_output_chars: int = 20000    # Maximum stdout characters returned
    preview_length: int = 500        # Character preview length for variables
    max_workers: int = 8             # Thread pool size for batched queries
    sub_model: str | None = None     # Override model for sub-LLM calls
    sub_provider: str | None = None  # Override provider for sub-LLM calls

env = PureRLMEnvironment(config=PureRLMConfig(preview_length=1000))
env.initialize_context(
    context=large_document_text,
    description="Legal contract to analyze",
    additional_vars={"reference_data": lookup_table},
)

# Inside REPL code
summary = llm_query("Summarize this section:\n" + context[0:5000])
print(summary)

# Inside REPL code
chunks = [context[i:i+10000] for i in range(0, len(context), 10000)]
prompts = [f"Summarize:\n{chunk}" for chunk in chunks]
summaries = llm_query_batched(prompts)

# Inside REPL code
SHOW_VARS()
# Output:
# Available variables:
#   context: str
#   summaries: list
#   result: dict

reward = run_python_base                      # 0.1
if success:
    reward += run_python_success_bonus        # +0.7
else:
    reward -= run_python_failure_penalty      # -0.3
if stderr:
    reward -= run_python_stderr_penalty       # -0.1
if llm_calls_made:
    reward += 0.1 * min(len(llm_calls), 5)   # Bonus for using RLM paradigm

from rlm_code.rlm.environments import GenericRLMEnvironment

class GenericRLMEnvironment:
    name = "generic"

    def __init__(
        self,
        workdir: Path | None = None,
        reward_profile: RLMRewardProfile | dict[str, Any] | None = None,
    ):

from rlm_code.rlm.environments import DSPyCodingRLMEnvironment

reward = verifier_base + (dspy_score / 100.0) * verifier_score_weight
if compile_ok:    reward += verifier_compile_bonus
else:             reward -= verifier_compile_penalty
if pytest_ran:
    if pytest_ok: reward += verifier_pytest_bonus
    else:         reward -= verifier_pytest_penalty
if validation_ok: reward += verifier_validation_bonus
else:             reward -= verifier_validation_penalty
reward -= min(warning_penalty_cap, num_warnings * warning_penalty_per_warning)

@dataclass(slots=True)
class EnvironmentActionResult:
    observation: dict[str, Any]       # Execution output visible to the LLM
    reward: float                     # Scalar reward in [-1.0, 1.0]
    done: bool = False                # True if task is complete
    final_response: str | None = None # Final answer (when done=True)
    memory_note: str | None = None    # Short note for rolling memory

@dataclass(slots=True)
class EnvironmentDoctorCheck:
    name: str                          # Check identifier
    status: str                        # "pass" | "warn" | "fail"
    detail: str                        # Human-readable description
    recommendation: str | None = None  # Fix suggestion (if not passing)

@dataclass(slots=True)
class RLMRewardProfile:
    # Global multiplier
    global_scale: float = 1.0

    # run_python scoring
    run_python_base: float = 0.1
    run_python_success_bonus: float = 0.7
    run_python_failure_penalty: float = 0.3
    run_python_stderr_penalty: float = 0.1

    # DSPy heuristic adjustments
    dspy_pattern_match_bonus: float = 0.03
    dspy_pattern_bonus_cap: float = 0.2

    # Verifier scoring
    verifier_base: float = 0.15
    verifier_score_weight: float = 0.5
    verifier_compile_bonus: float = 0.2
    verifier_compile_penalty: float = 0.35
    verifier_pytest_bonus: float = 0.25
    verifier_pytest_penalty: float = 0.25
    verifier_validation_bonus: float = 0.15
    verifier_validation_penalty: float = 0.3
    verifier_warning_penalty_per_warning: float = 0.03
    verifier_warning_penalty_cap: float = 0.15

from rlm_code.rlm.environments import RLMRewardProfile

# Aggressive profile favoring correctness over speed
profile = RLMRewardProfile(
    global_scale=1.2,
    run_python_success_bonus=0.9,
    run_python_failure_penalty=0.5,
    verifier_compile_penalty=0.5,
    verifier_pytest_bonus=0.4,
)

# Or from a dictionary (e.g., loaded from config)
profile = RLMRewardProfile.from_mapping({
    "global_scale": 1.2,
    "run_python_success_bonus": 0.9,
})