dspy.PythonInterpreter

dspy.PythonInterpreter(deno_command: List[str] | None = None, enable_read_paths: List[PathLike | str] | None = None, enable_write_paths: List[PathLike | str] | None = None, enable_env_vars: List[str] | None = None, enable_network_access: List[str] | None = None, sync_files: bool = True)

PythonInterpreter that runs code in a sandboxed environment using Deno and Pyodide.

Prerequisites: - Deno (https://docs.deno.com/runtime/getting_started/installation/).

Example Usage:

code_string = "print('Hello'); 1 + 2"
with PythonInterpreter() as interp:
    output = interp(code_string) # If final statement is non-None, prints the numeric result, else prints captured output

Parameters:

Name	Type	Description	Default
deno_command	List[str] | None	command list to launch Deno.	None
enable_read_paths	List[PathLike | str] | None	Files or directories to allow reading from in the sandbox.	None
enable_write_paths	List[PathLike | str] | None	Files or directories to allow writing to in the sandbox.	None
enable_env_vars	List[str] | None	Environment variable names to allow in the sandbox.	None
enable_network_access	List[str] | None	Domains or IPs to allow network access in the sandbox.	None
sync_files	bool	If set, syncs changes within the sandbox back to original files after execution.	True

Source code in dspy/primitives/python_interpreter.py

def __init__(
    self,
    deno_command: List[str] | None = None,
    enable_read_paths: List[PathLike | str] | None = None,
    enable_write_paths: List[PathLike | str] | None = None,
    enable_env_vars: List[str] | None = None,
    enable_network_access: List[str] | None = None,
    sync_files: bool = True,
) -> None:
    """
    Args:
        deno_command: command list to launch Deno.
        enable_read_paths: Files or directories to allow reading from in the sandbox.
        enable_write_paths: Files or directories to allow writing to in the sandbox.
        enable_env_vars: Environment variable names to allow in the sandbox.
        enable_network_access: Domains or IPs to allow network access in the sandbox.
        sync_files: If set, syncs changes within the sandbox back to original files after execution.
    """
    if isinstance(deno_command, dict):
        deno_command = None  # no-op, just a guard in case someone passes a dict

    self.enable_read_paths = enable_read_paths or []
    self.enable_write_paths = enable_write_paths or []
    self.enable_env_vars = enable_env_vars or []
    self.enable_network_access = enable_network_access or []
    self.sync_files = sync_files
    # TODO later on add enable_run (--allow-run) by proxying subprocess.run through Deno.run() to fix 'emscripten does not support processes' error

    if deno_command:
        self.deno_command = list(deno_command)
    else:
        args = ["deno", "run", "--allow-read"]
        self._env_arg  = ""
        if self.enable_env_vars:
            user_vars = [str(v).strip() for v in self.enable_env_vars]
            args.append("--allow-env=" + ",".join(user_vars))
            self._env_arg = ",".join(user_vars)
        if self.enable_network_access:
            args.append(f"--allow-net={','.join(str(x) for x in self.enable_network_access)}")
        if self.enable_write_paths:
            args.append(f"--allow-write={','.join(str(x) for x in self.enable_write_paths)}")

        args.append(self._get_runner_path())

        # For runner.js to load in env vars
        if self._env_arg:
            args.append(self._env_arg)
        self.deno_command = args

    self.deno_process = None
    self._mounted_files = False

Functions

__call__(code: str, variables: Dict[str, Any] | None = None) -> Any

Source code in dspy/primitives/python_interpreter.py

def __call__(
    self,
    code: str,
    variables: Dict[str, Any] | None = None,
) -> Any:
    return self.execute(code, variables)

execute(code: str, variables: Dict[str, Any] | None = None) -> Any

Source code in dspy/primitives/python_interpreter.py

def execute(
    self,
    code: str,
    variables: Dict[str, Any] | None = None,
) -> Any:
    variables = variables or {}
    code = self._inject_variables(code, variables)
    self._ensure_deno_process()
    self._mount_files()

    # Send the code as JSON
    input_data = json.dumps({"code": code})
    try:
        self.deno_process.stdin.write(input_data + "\n")
        self.deno_process.stdin.flush()
    except BrokenPipeError:
        # If the process died, restart and try again once
        self._ensure_deno_process()
        self.deno_process.stdin.write(input_data + "\n")
        self.deno_process.stdin.flush()

    # Read one JSON line from stdout
    output_line = self.deno_process.stdout.readline().strip()
    if not output_line:
        # Possibly the subprocess died or gave no output
        err_output = self.deno_process.stderr.read()
        raise InterpreterError(f"No output from Deno subprocess. Stderr: {err_output}")

    # Parse that line as JSON
    try:
        result = json.loads(output_line)
    except json.JSONDecodeError:
        # If not valid JSON, just return raw text
        result = {"output": output_line}

    # If we have an error, determine if it's a SyntaxError or other error using error.errorType.
    if "error" in result:
        error_msg = result["error"]
        error_type = result.get("errorType", "Sandbox Error")
        if error_type == "FinalAnswer":
            # The `FinalAnswer` trick to receive output from the sandbox interpreter,
            # just simply replace the output with the arguments.
            result["output"] = result.get("errorArgs", None)
        elif error_type == "SyntaxError":
            raise SyntaxError(f"Invalid Python syntax. message: {error_msg}")
        else:
            raise InterpreterError(f"{error_type}: {result.get('errorArgs') or error_msg}")

    # If there's no error or got `FinalAnswer`, return the "output" field
    self._sync_files()
    return result.get("output", None)

shutdown() -> None

Source code in dspy/primitives/python_interpreter.py

def shutdown(self) -> None:
    if self.deno_process and self.deno_process.poll() is None:
        shutdown_message = json.dumps({"shutdown": True}) + "\n"
        self.deno_process.stdin.write(shutdown_message)
        self.deno_process.stdin.flush()
        self.deno_process.stdin.close()
        self.deno_process.wait()
        self.deno_process = None

:::