Skip to content

rlm_repl

rlm_repl

Sandboxed REPL environment for the RLM agent.

Provides a persistent Python namespace with injected helper functions (llm_query, llm_batch, FINAL, FINAL_VAR) that the RLM agent's generated code uses to decompose context and make recursive sub-LM calls.

Classes

RLMRepl 

RLMRepl(llm_query_fn: Optional[Callable[[str], str]] = None, llm_batch_fn: Optional[Callable[[List[str]], List[str]]] = None, *, max_output_chars: int = 10000)

Sandboxed Python REPL with persistent namespace for the RLM agent.

PARAMETER DESCRIPTION
llm_query_fn

Callback invoked when REPL code calls llm_query(prompt).

TYPE: Optional[Callable[[str], str]] DEFAULT: None

llm_batch_fn

Callback invoked when REPL code calls llm_batch(prompts).

TYPE: Optional[Callable[[List[str]], List[str]]] DEFAULT: None

max_output_chars

Maximum characters captured from stdout per execution.

TYPE: int DEFAULT: 10000

Source code in src/openjarvis/agents/rlm_repl.py 
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
def __init__(
    self,
    llm_query_fn: Optional[Callable[[str], str]] = None,
    llm_batch_fn: Optional[Callable[[List[str]], List[str]]] = None,
    *,
    max_output_chars: int = 10000,
) -> None:
    self._max_output_chars = max_output_chars
    self._terminated = False
    self._final_value: Any = None

    # Build namespace
    self._namespace: Dict[str, Any] = {}

    # Inject safe stdlib modules
    for mod_name in _SAFE_MODULES:
        try:
            import importlib

            self._namespace[mod_name] = importlib.import_module(mod_name)
        except ImportError:
            pass

    # answer dict — code can set answer["ready"] = True, answer["value"] = ...
    self._namespace["answer"] = {"ready": False, "value": None}

    # Inject FINAL / FINAL_VAR
    self._namespace["FINAL"] = self._final
    self._namespace["FINAL_VAR"] = self._final_var

    # Inject llm_query / llm_batch
    if llm_query_fn is not None:
        self._namespace["llm_query"] = llm_query_fn
    if llm_batch_fn is not None:
        self._namespace["llm_batch"] = llm_batch_fn
Attributes
is_terminated property 
is_terminated: bool

Check if FINAL/FINAL_VAR was called or answer["ready"] is True.

final_answer property 
final_answer: Any

Return the termination value.

Functions
security_check 
security_check(code: str) -> Optional[str]

Check code for dangerous patterns. Returns error message or None.

Source code in src/openjarvis/agents/rlm_repl.py 
132
133
134
135
136
137
def security_check(self, code: str) -> Optional[str]:
    """Check code for dangerous patterns. Returns error message or None."""
    for pattern in _BLOCKED_PATTERNS:
        if pattern in code:
            return f"Blocked: code contains prohibited pattern '{pattern}'"
    return None
execute 
execute(code: str) -> str

Execute code in the persistent namespace and return captured stdout.

Raises are caught and returned as error strings.

Source code in src/openjarvis/agents/rlm_repl.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def execute(self, code: str) -> str:
    """Execute *code* in the persistent namespace and return captured stdout.

    Raises are caught and returned as error strings.
    """
    # Security check
    violation = self.security_check(code)
    if violation is not None:
        return f"Error: {violation}"

    stdout_buf = io.StringIO()
    stderr_buf = io.StringIO()

    try:
        with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
            exec(code, self._namespace)  # noqa: S102
    except Exception as exc:
        error_msg = f"{type(exc).__name__}: {exc}"
        return error_msg

    output = stdout_buf.getvalue()
    err_output = stderr_buf.getvalue()
    if err_output:
        output += ("\n" if output else "") + err_output

    # Truncate if needed
    if len(output) > self._max_output_chars:
        output = output[: self._max_output_chars] + "\n... (output truncated)"

    return output
set_variable 
set_variable(name: str, value: Any) -> None

Set a variable in the REPL namespace.

Source code in src/openjarvis/agents/rlm_repl.py 
174
175
176
def set_variable(self, name: str, value: Any) -> None:
    """Set a variable in the REPL namespace."""
    self._namespace[name] = value
get_variable 
get_variable(name: str) -> Any

Get a variable from the REPL namespace.

Source code in src/openjarvis/agents/rlm_repl.py 
178
179
180
def get_variable(self, name: str) -> Any:
    """Get a variable from the REPL namespace."""
    return self._namespace.get(name)