Framework Adapters¶

RLM Code can delegate task execution to external agentic frameworks through a protocol-based adapter system. Each adapter converts framework-native execution into RLM trajectory-compatible step records, enabling unified observability and comparison across execution backends.

Module¶

rlm_code.rlm.frameworks
  +-- base.py                 -- Protocol, data classes
  +-- registry.py             -- Adapter registry
  +-- (runner-native)         -- `framework=native` path in RLMRunner
  +-- dspy_rlm_adapter.py     -- DSPy native RLM adapter
  +-- adk_rlm_adapter.py      -- ADK sample native RLM adapter
  +-- pydantic_ai_adapter.py  -- Pydantic AI adapter
  +-- google_adk_adapter.py   -- Google ADK adapter
  +-- deepagents_adapter.py   -- DeepAgents (LangGraph) adapter

Architecture¶

graph TD
    Runner["RLM Runner"] --> Registry["FrameworkAdapterRegistry"]
    Registry -->|"get('dspy-rlm')"| DSPyRLM["DSPyRLMFrameworkAdapter"]
    Registry -->|"get('adk-rlm')"| ADKRLM["ADKRLMFrameworkAdapter"]
    Registry -->|"get('pydantic-ai')"| PydAI["PydanticAIFrameworkAdapter"]
    Registry -->|"get('google-adk')"| ADK["GoogleADKFrameworkAdapter"]
    Registry -->|"get('deepagents')"| DA["DeepAgentsFrameworkAdapter"]
    DSPyRLM --> Agent0["dspy.RLM"]
    ADKRLM --> Agent00["adk_rlm.completion"]
    PydAI --> Agent1["Pydantic AI Agent"]
    ADK --> Agent2["Google ADK LlmAgent"]
    DA --> Agent3["DeepAgents LangGraph Agent"]
    Agent0 --> Steps0["FrameworkStepRecord[]"]
    Agent00 --> Steps00["FrameworkStepRecord[]"]
    Agent1 --> Steps1["FrameworkStepRecord[]"]
    Agent2 --> Steps2["FrameworkStepRecord[]"]
    Agent3 --> Steps3["FrameworkStepRecord[]"]
    Steps0 --> Result["FrameworkEpisodeResult"]
    Steps00 --> Result
    Steps1 --> Result["FrameworkEpisodeResult"]
    Steps2 --> Result
    Steps3 --> Result

Data Classes¶

FrameworkStepRecord¶

One framework step converted into RLM trajectory-compatible form.

from rlm_code.rlm.frameworks import FrameworkStepRecord

step = FrameworkStepRecord(
    action="tool_call",
    observation={"tool_name": "search", "args": {"query": "RLM"}},
    reward=0.02,
    done=False,
)

FrameworkEpisodeResult¶

Result payload returned by a framework adapter run.

from rlm_code.rlm.frameworks import FrameworkEpisodeResult

result = FrameworkEpisodeResult(
    completed=True,
    final_response="The answer is 42.",
    steps=[step1, step2, step3],
    total_reward=0.35,
    usage_summary={"prompt_tokens": 200, "completion_tokens": 100},
    metadata={"framework": "pydantic-ai", "resolved_model": "openai:gpt-4o"},
)

Adapter Protocol¶

RLMFrameworkAdapter¶

The RLMFrameworkAdapter is a Python Protocol that all adapters must satisfy.

from rlm_code.rlm.frameworks import RLMFrameworkAdapter

class MyAdapter:
    framework_id: str = "my-framework"

    def doctor(self) -> tuple[bool, str]:
        """Check if the framework is installed and ready."""
        ...

    def run_episode(
        self,
        *,
        task: str,
        llm_connector: Any,
        max_steps: int,
        exec_timeout: int,
        workdir: str,
        sub_model: str | None = None,
        sub_provider: str | None = None,
        context: dict[str, Any] | None = None,
    ) -> FrameworkEpisodeResult:
        """Execute one framework-native task run."""
        ...

run_episode Parameters¶

Adapter Registry¶

The FrameworkAdapterRegistry manages adapter registration and lookup.

from rlm_code.rlm.frameworks import FrameworkAdapterRegistry

registry = FrameworkAdapterRegistry.default(workdir="/path/to/project")

Methods¶

Default Registry¶

The default() factory registers all built-in adapters:

registry = FrameworkAdapterRegistry.default(workdir="/my/project")
print(registry.list_ids())
# ["adk-rlm", "deepagents", "dspy-rlm", "google-adk", "pydantic-ai"]

native is intentionally not listed here because it is built directly into RLMRunner (not adapter-registered).

Doctor Output¶

results = registry.doctor()
# [
#   {"framework": "adk-rlm", "ok": False, "mode": "native_rlm", "detail": "adk_rlm not installed..."},
#   {"framework": "dspy-rlm", "ok": True, "mode": "native_rlm", "detail": "dspy RLM available"},
#   {"framework": "google-adk", "ok": False, "detail": "google-adk not installed..."},
#   {"framework": "pydantic-ai", "ok": True, "detail": "pydantic-ai available"},
# ]

TUI / CLI Usage¶

Use the shell command to inspect adapter readiness:

/rlm frameworks

Then select an adapter in any run path:

/rlm run Investigate memory leak framework=dspy-rlm env=generic
/rlm run Build an RLM planner framework=adk-rlm env=generic
/rlm chat Continue analysis framework=deepagents

Compatibility alias: framework=dspy maps to framework=dspy-rlm. Use framework=native env=pure_rlm for paper-style built-in recursive mode.

DSPy RLM Adapter¶

Module: rlm_code.rlm.frameworks.dspy_rlm_adapter

Adapter ID: dspy-rlm

This adapter uses DSPy's native dspy.RLM abstraction (reference: dspy.RLM from the installed DSPy package) and returns one normalized FrameworkEpisodeResult.

Installation¶

pip install dspy

Notes¶

1. Requires a DSPy build that exposes dspy.RLM.
2. Uses dspy.settings.lm if already configured, otherwise resolves from the connected RLM Code model.

ADK Sample RLM Adapter¶

Module: rlm_code.rlm.frameworks.adk_rlm_adapter

Adapter ID: adk-rlm

This adapter runs the Google ADK sample RLM implementation (reference: adk_rlm/main.py from the vendored sample package) through adk_rlm.completion(...).

Installation¶

pip install 'rlm-code[adk]'

Notes¶

1. adk_rlm is vendored inside rlm-code for parity with the ADK reference sample.
2. google-adk + google-genai still need to be installed via extras.
3. Defaults to a Gemini model when no active sub-model is provided.

Pydantic AI Adapter¶

Module: rlm_code.rlm.frameworks.pydantic_ai_adapter

Adapter ID: pydantic-ai

Installation¶

pip install 'rlm-code[pydantic]'
# or directly:
pip install pydantic-ai

How It Works¶

1. Resolves the model name to Pydantic AI's provider:model format
2. Creates a pydantic_ai.Agent with system instructions
3. Calls agent.run_sync(task)
4. Extracts step records from message parts

Model Resolution¶

The adapter normalizes provider names to Pydantic AI's expected format:

Step Extraction¶

Messages from Pydantic AI are parsed by part type:

If no steps are extracted, a single model_response step is created with reward +0.25 (if output is non-empty) or -0.1 (if empty).

Example¶

from rlm_code.rlm.frameworks import PydanticAIFrameworkAdapter

adapter = PydanticAIFrameworkAdapter(workdir="/my/project")

# Check readiness
ok, detail = adapter.doctor()
print(ok, detail)  # True, "pydantic-ai available"

# Run a task
result = adapter.run_episode(
    task="Explain the Fibonacci sequence",
    llm_connector=my_connector,
    max_steps=3,
    exec_timeout=60,
    workdir="/my/project",
)
print(result.final_response)
print(result.total_reward)

Google ADK Adapter¶

Module: rlm_code.rlm.frameworks.google_adk_adapter

Adapter ID: google-adk

Installation¶

pip install 'rlm-code[adk]'
# or directly:
pip install google-adk

How It Works¶

1. Resolves the model name for the ADK runtime
2. Creates a google.adk.agents.LlmAgent with instructions
3. Uses InMemoryRunner with an async event stream
4. Extracts step records from ADK events

Model Resolution¶

Event Serialization¶

ADK events are parsed by content type:

Async Execution¶

The Google ADK adapter uses async for to stream events from the runner. When called from a synchronous context, it spawns a background thread to run the async event loop:

# Internal helper for sync/async bridging
def _run_coro_sync(coro):
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro)
    # If loop is running, use a thread
    thread = threading.Thread(target=lambda: asyncio.run(coro), daemon=True)
    thread.start()
    thread.join()

Example¶

from rlm_code.rlm.frameworks import GoogleADKFrameworkAdapter

adapter = GoogleADKFrameworkAdapter(workdir="/my/project")

# Check readiness
ok, detail = adapter.doctor()
print(ok, detail)  # True, "google-adk available"

# Run a task
result = adapter.run_episode(
    task="Analyze this dataset",
    llm_connector=my_connector,
    max_steps=5,
    exec_timeout=120,
    workdir="/my/project",
    sub_provider="gemini",
    sub_model="gemini-2.5-pro",
)
print(result.completed)
print(len(result.steps))

DeepAgents Adapter¶

Module: rlm_code.rlm.frameworks.deepagents_adapter

Adapter ID: deepagents

The DeepAgents adapter integrates DeepAgents, a LangGraph-based agentic framework, into the RLM execution pipeline. It converts LangGraph message histories into RLM trajectory-compatible step records, enabling unified observability across all framework backends.

Installation¶

pip install 'rlm-code[deepagents]'
# or directly:
pip install deepagents langchain-core

How It Works¶

1. Resolves the model name to DeepAgents' provider:model format
2. Selects a backend (StateBackend, FilesystemBackend, or LocalShellBackend)
3. Creates a deep agent via create_deep_agent() with system instructions
4. Invokes the agent with a HumanMessage
5. Extracts step records from LangChain message history

Model Resolution¶

The adapter normalizes RLM provider names to DeepAgents' expected format:

If the model name already contains a colon (e.g., anthropic:claude-opus-4-6), it is passed through unchanged.

Backend Selection¶

DeepAgents supports multiple execution backends, selectable via the context parameter:

# Use filesystem backend
result = adapter.run_episode(
    task="Organize these files",
    llm_connector=connector,
    max_steps=10,
    exec_timeout=60,
    workdir="/my/project",
    context={"deepagents_backend": "filesystem"},
)

Step Extraction¶

LangChain messages are parsed into FrameworkStepRecord entries:

Steps are capped at 80 per episode. The total_reward is clamped to [-1.0, 1.0].

Example¶

from rlm_code.rlm.frameworks import DeepAgentsFrameworkAdapter

adapter = DeepAgentsFrameworkAdapter(workdir="/my/project")

# Check readiness
ok, detail = adapter.doctor()
print(ok, detail)  # True, "deepagents available"

# Run a task
result = adapter.run_episode(
    task="Analyze this codebase and suggest improvements",
    llm_connector=my_connector,
    max_steps=10,
    exec_timeout=120,
    workdir="/my/project",
    sub_provider="anthropic",
    sub_model="claude-sonnet-4-20250514",
)
print(result.completed)
print(result.final_response)
print(f"Steps: {len(result.steps)}, Reward: {result.total_reward:.2f}")

Writing a Custom Adapter¶

To add support for a new framework:

1. Implement the protocol:

from dataclasses import dataclass
from rlm_code.rlm.frameworks.base import (
    FrameworkEpisodeResult,
    FrameworkStepRecord,
    RLMFrameworkAdapter,
)

@dataclass(slots=True)
class MyFrameworkAdapter:
    workdir: str
    framework_id: str = "my-framework"

    def doctor(self) -> tuple[bool, str]:
        try:
            import my_framework
            return (True, "my-framework available")
        except ImportError:
            return (False, "my-framework not installed")

    def run_episode(self, *, task, llm_connector, max_steps, exec_timeout,
                    workdir, sub_model=None, sub_provider=None, context=None):
        # ... run the framework ...
        return FrameworkEpisodeResult(
            completed=True,
            final_response="...",
            steps=[FrameworkStepRecord(action="run", reward=0.5)],
            total_reward=0.5,
        )

1. Register the adapter:

from rlm_code.rlm.frameworks import FrameworkAdapterRegistry

registry = FrameworkAdapterRegistry()
registry.register(MyFrameworkAdapter(workdir="/path"))

1. Use it:

adapter = registry.get("my-framework")
result = adapter.run_episode(task="...", ...)

Class Reference¶

DSPyRLMFrameworkAdapter¶

ADKRLMFrameworkAdapter¶

PydanticAIFrameworkAdapter¶

GoogleADKFrameworkAdapter¶

DeepAgentsFrameworkAdapter¶