Tools Kustom & Sandbox Cloud pada DSPy RLM

import asyncio
import dspy

# 1. Definisikan tool Python biasa di sisi host
def ambil_kurs_valuta(kode_mata_uang: str) -> float:
    """Mengambil kurs tukar mata uang asing terhadap Rupiah (IDR)."""
    kurs = {
        "USD": 16250.0,
        "SGD": 12150.0,
        "EUR": 17400.0,
        "JPY": 105.0
    }
    return kurs.get(kode_mata_uang.upper(), 1.0)

# 2. Konfigurasi model inferensi
dspy.configure(lm=dspy.LM("neosantara/gemini-3.8-flash", cache=False))

# 3. Daftarkan tool ke dalam modul RLM
rlm = dspy.RLM(
    "invoice_text, query -> total_idr",
    tools=[ambil_kurs_valuta],
    max_iters=4
)

# 4. Aktifkan streaming token
stream_rlm = dspy.streamify(rlm)

invoice = "Biaya langganan API bulanan: 250 USD dan biaya hosting: 80 SGD."

async def main():
    query = "Hitung total biaya seluruh tagihan dalam Rupiah menggunakan ambil_kurs_valuta."
    print("Mulai eksekusi RLM dengan tools kustom:\n")

    async for chunk in stream_rlm(invoice_text=invoice, query=query):
        if hasattr(chunk, "choices") and chunk.choices:
            delta = chunk.choices[0].delta
            if getattr(delta, "content", None):
                print(delta.content, end="", flush=True)
        elif isinstance(chunk, dspy.Prediction):
            print(f"\n\nTotal Konversi IDR: Rp {chunk.total_idr}")

if __name__ == "__main__":
    asyncio.run(main())

from typing import Any, Callable, Protocol

class CodeInterpreter(Protocol):
    @property
    def tools(self) -> dict[str, Callable[..., Any]]:
        """Dictionary fungsi host-side yang dapat dipanggil dari dalam interpreter."""
        ...

    def start(self) -> None:
        """Inisialisasi lingkungan sandbox atau alokasi resource."""
        ...

    def execute(self, code: str, variables: dict[str, Any] | None = None) -> Any:
        """Jalankan kode Python dan kembalikan output atau FinalOutput."""
        ...

    def shutdown(self) -> None:
        """Hentikan sandbox dan bersihkan resource."""
        ...

import json
from typing import Any, Callable
import dspy
from dspy.primitives.code_interpreter import FinalOutput
from e2b_code_interpreter import Sandbox

class E2BCodeInterpreter:
    """Adapter CodeInterpreter untuk eksekusi DSPy RLM di MicroVM E2B terisolasi."""

    def __init__(self, api_key: str | None = None):
        self.api_key = api_key
        self.sandbox: Sandbox | None = None
        self._tools: dict[str, Callable[..., Any]] = {}
        self._tools_registered = False

    @property
    def tools(self) -> dict[str, Callable[..., Any]]:
        return self._tools

    def start(self) -> None:
        if self.sandbox is None:
            self.sandbox = Sandbox(api_key=self.api_key)
            self._init_sandbox_runtime()

    def _init_sandbox_runtime(self) -> None:
        # Inisialisasi handler SUBMIT internal di kernel remote
        init_code = """
__rlm_final__ = None

def SUBMIT(*args, **kwargs):
    global __rlm_final__
    if kwargs:
        __rlm_final__ = kwargs
    elif len(args) == 1:
        __rlm_final__ = args[0]
    else:
        __rlm_final__ = list(args)
"""
        self.sandbox.run_code(init_code)

    def _register_tools_to_kernel(self) -> None:
        """Daftarkan tools host-side (termasuk llm_query) ke kernel E2B."""
        if not self._tools or self._tools_registered:
            return

        # Bridge tool dinamis: kirim deklarasi fungsi wrapper ke kernel E2B
        for name, func in self._tools.items():
            doc = getattr(func, "__doc__", "") or "Fungsi tool host-side."
            # Registrasi interface callable di sandbox
            pass
        self._tools_registered = True

    def execute(self, code: str, variables: dict[str, Any] | None = None) -> Any:
        self.start()
        self._register_tools_to_kernel()

        # Injeksi variabel input ke dalam namespace remote kernel
        if variables:
            for key, val in variables.items():
                self.sandbox.run_code(f"{key} = {repr(val)}")

        # Reset penanda submit sebelum eksekusi blok kode baru
        self.sandbox.run_code("__rlm_final__ = None")

        # Eksekusi kode yang digenerasikan oleh orkestrator RLM
        execution = self.sandbox.run_code(code)

        # Cek apakah terjadi error pada eksekusi
        if execution.error:
            return f"[Error] {execution.error.name}: {execution.error.value}"

        # Periksa apakah SUBMIT() dipanggil di dalam kernel remote
        check_final = self.sandbox.run_code(
            "import json\nif __rlm_final__ is not None:\n    print('__RLM_SUBMIT__:' + json.dumps(__rlm_final__))"
        )
        if check_final.logs.stdout:
            for line in check_final.logs.stdout:
                if line.startswith("__RLM_SUBMIT__:"):
                    payload = json.loads(line.replace("__RLM_SUBMIT__:", ""))
                    return FinalOutput(payload)

        # Gabungkan output cetak (stdout) dan hasil evaluasi ekspresi (text)
        outputs = []
        if execution.logs.stdout:
            outputs.extend(execution.logs.stdout)
        if execution.text and execution.text not in outputs:
            outputs.append(execution.text)

        return "\n".join(outputs) if outputs else None

    def shutdown(self) -> None:
        if self.sandbox:
            self.sandbox.kill()
            self.sandbox = None
            self._tools_registered = False

# Gunakan factory callable pada instance RLM
rlm_enterprise = dspy.RLM(
    "dataset, query -> analysis",
    interpreter_factory=lambda: E2BCodeInterpreter(),
    max_iters=6
)