Add interpreter execution instructions

Amp-Thread-ID: https://ampcode.com/threads/T-019fcce1-bc8b-753e-886f-9b77c0ccb376
Co-authored-by: Isaac Miller <isaac@cmpnd.ai>

Author: ampagent

dspy/predict/rlm.py

@@ -658,6 +658,14 @@ def _execute_code(
         except (CodeExecutionError, SyntaxError) as e:
             return f"[Error] {format_error_for_lm(e)}"
 
+    def _action_signature(self, repl: CodeInterpreter) -> Any:
+        instructions = getattr(repl, "execution_instructions", "")
+        if not instructions or type(self.generate_action) is not dspy.Predict:
+            return None
+        return self.generate_action.signature.append_instructions(
+            "Code interpreter execution environment:\n" + instructions
+        )
+
     def _execute_iteration(
         self,
         repl: CodeInterpreter,
@@ -666,13 +674,15 @@ def _execute_iteration(
         iteration: int,
         input_args: dict[str, Any],
         output_field_names: list[str],
+        action_signature: Any = None,
     ) -> Prediction | REPLHistory:
         """Execute one iteration. Returns Prediction if done, else updated REPLHistory."""
         variables_info = [variable.format() for variable in variables]
         action = self.generate_action(
             variables_info=variables_info,
             repl_history=history,
             iteration=f"{iteration + 1}/{self.max_iters}",
+            **({"signature": action_signature} if action_signature else {}),
         )
         if self.verbose:
             logger.info(
@@ -716,12 +726,13 @@ def forward(self, interpreter: CodeInterpreter | None = None, /, **input_args) -
         variables = self._build_variables(**input_args)
 
         with self._interpreter_context(execution_tools, interpreter) as repl:
+            action_signature = self._action_signature(repl)
             regular_args = self._prepare_serializable_vars(input_args, repl)
             history: REPLHistory = REPLHistory(max_output_chars=self.max_output_chars)
 
             for iteration in range(self.max_iters):
                 result: Prediction | REPLHistory = self._execute_iteration(
-                    repl, variables, history, iteration, regular_args, output_field_names
+                    repl, variables, history, iteration, regular_args, output_field_names, action_signature
                 )
                 if isinstance(result, Prediction):
                     return result
@@ -759,13 +770,15 @@ async def _aexecute_iteration(
         iteration: int,
         input_args: dict[str, Any],
         output_field_names: list[str],
+        action_signature: Any = None,
     ) -> Prediction | REPLHistory:
         """Async version: Execute one iteration."""
         variables_info = [variable.format() for variable in variables]
         pred = await self.generate_action.acall(
             variables_info=variables_info,
             repl_history=history,
             iteration=f"{iteration + 1}/{self.max_iters}",
+            **({"signature": action_signature} if action_signature else {}),
         )
         if self.verbose:
             logger.info(
@@ -805,12 +818,13 @@ async def aforward(self, interpreter: CodeInterpreter | None = None, /, **input_
         variables = self._build_variables(**input_args)
 
         with self._interpreter_context(execution_tools, interpreter) as repl:
+            action_signature = self._action_signature(repl)
             regular_args = self._prepare_serializable_vars(input_args, repl)
             history = REPLHistory(max_output_chars=self.max_output_chars)
 
             for iteration in range(self.max_iters):
                 result = await self._aexecute_iteration(
-                    repl, variables, history, iteration, regular_args, output_field_names
+                    repl, variables, history, iteration, regular_args, output_field_names, action_signature
                 )
                 if isinstance(result, Prediction):
                     return result

dspy/primitives/python_interpreter.py

@@ -343,6 +343,14 @@ def __init__(
         self._pending_large_vars = {}
         self._session_ended = False
 
+    @property
+    def execution_instructions(self) -> str:
+        """Describe stable constraints of this execution environment."""
+        return (
+            "Python runs in Pyodide/WebAssembly. State persists across executions, but subprocesses and native "
+            "extensions are unavailable. Filesystem, environment, and network access require explicit permission."
+        )
+
     def _check_session_active(self) -> None:
         if self._session_ended:
             raise CodeInterpreterError(

tests/predict/test_rlm.py

@@ -426,6 +426,40 @@ def make_call():
 
 
 class TestRLMInterpreterLifecycle:
+    def test_execution_instructions_are_invocation_local(self):
+        interpreter = MockInterpreter(responses=[FinalOutput({"answer": "42"})])
+        interpreter.execution_instructions = "Use this runtime."
+        rlm = RLM("query -> answer", max_iters=1)
+        original = rlm.generate_action.signature
+        calls = []
+        rlm.generate_action.forward = lambda **kwargs: calls.append(kwargs) or Prediction(
+            reasoning="done", code="SUBMIT(answer='42')"
+        )
+
+        assert rlm(interpreter, query="test").answer == "42"
+        assert calls[0]["signature"].instructions.endswith(
+            "Code interpreter execution environment:\nUse this runtime."
+        )
+        assert rlm.generate_action.signature is original
+
+    @pytest.mark.asyncio
+    async def test_custom_action_does_not_receive_signature_override(self):
+        class Action:
+            def __call__(self, *, variables_info, repl_history, iteration):
+                return Prediction(reasoning="done", code="SUBMIT(answer='42')")
+
+            async def acall(self, *, variables_info, repl_history, iteration):
+                return self(variables_info=variables_info, repl_history=repl_history, iteration=iteration)
+
+        rlm = RLM("query -> answer", max_iters=1)
+        rlm.generate_action = Action()
+        sync = MockInterpreter(responses=[FinalOutput({"answer": "42"})])
+        async_interpreter = MockInterpreter(responses=[FinalOutput({"answer": "42"})])
+        sync.execution_instructions = async_interpreter.execution_instructions = "runtime"
+
+        assert rlm(sync, query="test").answer == "42"
+        assert (await rlm.acall(async_interpreter, query="test")).answer == "42"
+
     def test_interpreter_remains_available_as_signature_input(self):
         factory = MockInterpreterFactory(responses=[FinalOutput({"answer": "CPython"})])
         rlm = RLM("interpreter -> answer", max_iters=1, interpreter_factory=factory)

tests/primitives/test_python_interpreter.py

@@ -1623,3 +1623,12 @@ def test_base_exceptions_do_not_desync_interpreter():
             with pytest.raises(CodeExecutionError, match=error_type):
                 interpreter.execute(code)
             assert interpreter.execute("print('still alive')") == "still alive\n"
+
+
+def test_execution_instructions_are_stable_and_read_only():
+    interpreter = PythonInterpreter(deno_command=["deno", "run"])
+
+    assert interpreter.execution_instructions == interpreter.execution_instructions
+    assert "Pyodide" in interpreter.execution_instructions
+    with pytest.raises(AttributeError):
+        interpreter.execution_instructions = "replacement"