dspy-rlm-hooks 0.1.3

Lifecycle instrumentation for DSPy's RLM (Recursive Language Model).

DSPy RLM Hooks

Lifecycle instrumentation for DSPy's RLM (Recursive Language Model).
Explore the Documentation »
Report Bug · Request Feature

Table of Contents

1. About
2. Quick Start
3. Usage
4. Development
5. Contributing
6. License

About

DSPy RLM Hooks injects lifecycle hooks into DSPy's internal RLM iteration loop, giving you full control over every stage of code generation, execution, and history tracking.

• Code Rewriting — Fix or augment LLM-generated code before it runs
• Variable Injection — Seed the interpreter with persistent variables and imports
• Result Auditing — Transform, validate, or retry on errors
• History Management — Inspect and modify the REPL history between iterations
• Sync & Async — Hooks work in either mode; coroutines are auto-detected

Requires DSPy 3.1+ and Pydantic 2+.

(back to top)

Architecture

RLM Hook Lifecycle

flowchart LR
    subgraph Iteration["RLM Iteration"]
        PreIter["pre_iteration_hook"] --> Gen["Generate Code"]
        Gen --> PreExec["pre_execution_hook"]
        PreExec --> Exec["Execute Code"]
        Exec --> PostExec["post_execution_hook"]
        PostExec --> PostIter["post_iteration_hook"]
    end

    PreIter -.->|inject vars, prepend code| Gen
    PreExec -.->|rewrite code| Exec
    PostExec -.->|transform result| PostIter

Hooks fire at each stage of an RLM iteration, allowing inspection and modification of behaviour.

(back to top)

Quick Start

Install

Install dspy-rlm-hooks with uv (recommended):

uv add dspy-rlm-hooks

Or with pip:

pip install dspy-rlm-hooks

Basic Usage

import dspy
from dspy_rlm_hooks import enable_rlm_hooks, PreIterationOutput

rlm = dspy.RLM(...)

def inject_math(iteration, variables, history, input_args):
    return PreIterationOutput(
        extra_vars={"tool": "calculator"},
        python_code="import math",
    )

enable_rlm_hooks(rlm, pre_iteration_hook=inject_math)

result = rlm(question="What is the square root of 1764?")

(back to top)

Usage

All Four Hooks

A realistic example showing how each hook can be used to build a safe, instrumented agent:

from dspy_rlm_hooks import (
    enable_rlm_hooks,
    PreIterationOutput,
    PreExecutionOutput,
    PostExecutionOutput,
    PostIterationOutput,
)
from dspy.primitives.repl_types import REPLHistory
import re

# ── Pre-iteration: seed interpreter with a regex toolkit ──

def pre_iteration(iteration, variables, history, input_args):
    """Inject a regex helper and seed variables before every iteration."""
    return PreIterationOutput(
        extra_vars={"search_pattern": r"TODO|FIXME|HACK"},
        python_code="""
import re

def grep(pattern, text):
    return re.findall(pattern, text)
""",
    )

# ── Pre-execution: block dangerous code ──

FORBIDDEN = re.compile(r"\b(eval|exec|compile|__import__)\b")

def pre_execution(iteration, code, variables, history, input_args):
    """Sanitise generated code before it reaches the interpreter."""
    if FORBIDDEN.search(code):
        safe_code = FORBIDDEN.sub("# BLOCKED", code)
        return PreExecutionOutput(code=safe_code)
    return PreExecutionOutput(code=code)

# ── Post-execution: retry on error ──

def post_execution(iteration, code, result, variables, history, input_args):
    """If execution raised an error, wrap a hint so the LLM retries next round."""
    if isinstance(result, str) and result.startswith("[Error]"):
        return PostExecutionOutput(
            result=f"{result}\n# Hint: the variable 'search_pattern' is already in scope."
        )
    return PostExecutionOutput(result=result)

# ── Post-iteration: enforce iteration budget ──

MAX_ITERATIONS = 5
current_iter_count = 0

def post_iteration(iteration, pred, code, result, history: REPLHistory):
    """Track iterations and stop early if the budget is exhausted."""
    global current_iter_count
    current_iter_count += 1
    if current_iter_count >= MAX_ITERATIONS:
        # Return empty history to signal stop
        return PostIterationOutput(history=REPLHistory(entries=[]))
    return PostIterationOutput(history=history)

# ── Wire everything up ──

enable_rlm_hooks(
    rlm,
    pre_iteration_hook=pre_iteration,
    pre_execution_hook=pre_execution,
    post_execution_hook=post_execution,
    post_iteration_hook=post_iteration,
)

result = rlm(question="Find all TODO comments in the codebase")

Async Hooks

Return a coroutine and the system handles it automatically:

async def fetch_context(iteration, variables, history, input_args):
    context = await remote_cache.get(input_args["question"])
    return PreIterationOutput(extra_vars={"cached_context": context})

enable_rlm_hooks(rlm, pre_iteration_hook=fetch_context)

Disabling Hooks

from dspy_rlm_hooks import disable_rlm_hooks

disable_rlm_hooks(rlm)

Removes all monkey-patched overrides and reverts to original behaviour.

(back to top)

Hook Reference

Hook	When it fires	What it can do
PreIteration	Before action generation	Inject variables (extra_vars) and persistent code (python_code)
PreExecution	After code generation, before running	Rewrite or sanitise the generated code string
PostExecution	After code runs, before history processing	Transform, audit, or replace the raw result
PostIteration	After result is folded into history	Save learnings, trigger side effects, or modify history

(back to top)

Development

Code Quality

This project uses several tools to maintain code quality:

• Ruff: Linting and formatting
• isort: Import sorting
• pytest: Testing framework
• ty: Type checking
• deptry: Dependency analysis

Available commands:

# Run all quality checks
uv run poe clean-full

# Individual checks
uv run poe lint          # Ruff linting
uv run poe format        # Ruff formatting
uv run poe sort          # Import sorting
uv run poe typecheck     # Type checking
uv run poe deptry        # Dependency analysis

Testing

Run tests using pytest:

# Run all tests
uv run pytest

# Run specific test
uv run pytest path/to/test.py::test_name

(back to top)

Contributing

Quick workflow:

1. Fork and branch: git checkout -b feature/name
2. Make changes
3. Run checks: uv run poe clean-full
4. Commit and push
5. Open a Pull Request

(back to top)

License

MIT (as declared in pyproject.toml).

---

_{Built by thememium}