A Python framework for building intelligent, recursive task decomposition systems powered by Large Language Models
Project description
RLM - Recursive Language Models
Think Recursively, Solve Intelligently - Emphasizes the intelligent decision-making (EXECUTE vs RECURSE)
A Python framework for building intelligent, recursive task decomposition systems powered by Large Language Models (LLMs).
Overview
RLM enables LLMs to solve complex tasks through intelligent recursive decomposition. Instead of tackling a complex problem all at once, RLM's RecursiveEngine decides whether to:
- EXECUTE: Solve the task atomically
- RECURSE: Break it into manageable sub-tasks
This creates a hierarchical problem-solving approach where complex tasks are automatically decomposed into simpler components, solved independently, and synthesized into a final result.
Features
- ๐ง Intelligent Decomposition: Automatic task breakdown based on complexity
- ๐ค Multi-Agent Support: Route sub-tasks to specialized agents
- โก Async Execution: Parallel sub-task processing for 8-10ร throughput
- ๐ก๏ธ Fault Tolerance: Checkpoint-based recovery from failures
- ๐ Progress Tracking: Real-time execution monitoring
- ๐ง Tool Calling: Native LLM tool/function calling integration
- ๐ฏ Type Safety: Full Python 3.12+ type annotations
- ๐ Recursive Intelligence: Hierarchical problem-solving patterns
Installation
From PyPI (Recommended)
pip install rlm
Optional dependencies:
# With Redis L2 cache (OpenAI embeddings)
pip install rlm[cache-l2]
# With observability (OpenTelemetry)
pip install rlm[observability]
# All extras
pip install rlm[cache-l2,observability]
From Source
Prerequisites:
- Python 3.12+
uvpackage manager
# Clone the repository
git clone https://github.com/Mathews-Tom/rlm.git
cd rlm
# Install dependencies
uv sync
Configure API Keys
Create a .env file in the project root:
OPENAI_API_KEY=sk-your-key-here
Quick Start
Basic Recursive Decomposition
from rlm.engine import RecursiveEngine
# Create LLM caller
def llm_caller(inputs, context):
# Your LLM implementation
pass
# Initialize engine
engine = RecursiveEngine(
llm=llm_caller,
max_depth=15,
max_steps=200
)
# Solve a task
result = engine.solve("Write a 200-word analysis of cloud computing benefits")
print(result["content"])
Multi-Agent System
from rlm.engine import RecursiveEngine
# Create specialized agents
agents = {
"planner": planner_llm,
"researcher": researcher_llm,
"writer": writer_llm,
}
engine = RecursiveEngine(
llm=planner_llm,
agents=agents,
router_model="planner",
max_depth=15,
max_steps=200
)
result = engine.solve("Research and write about AI in healthcare")
Async Execution
from rlm.async_engine import AsyncRecursiveEngine
engine = AsyncRecursiveEngine(
llm=async_llm_caller,
max_depth=15,
max_steps=200
)
# 8-10ร faster for parallel sub-tasks
result = await engine.solve("Complex multi-part task")
Examples
The examples/ directory contains six progressively advanced demonstrations:
Core Recursion (Examples 01-02)
- 01_basic_example.py: Basic recursive decomposition
- 02_multi_agent_example.py: Multi-agent routing and specialization
Feature Demonstrations (Examples 03-06)
- 03_tool_calling_example.py: Tool calling with OpenAI function calling
- 04_streaming_example.py: Real-time progress tracking
- 05_checkpoint_example.py: Fault tolerance and recovery
- 06_advanced_example.py: Combined production-ready features
Run any example:
uv run python examples/01_basic_example.py
See examples/README.md for detailed documentation.
Architecture
Core Components
graph TB
Start([Task Input]) --> Planner{Planner<br/>EXECUTE or RECURSE?}
Planner -->|EXECUTE| Execute[Execute Atomically<br/>Leaf Task]
Planner -->|RECURSE| Decompose[Decompose into Sub-tasks]
Decompose --> Task1[Task 1<br/>Agent 1]
Decompose --> Task2[Task 2<br/>Agent 2]
Decompose --> TaskN[Task N<br/>Agent N]
Task1 --> Synthesis[Synthesize Results]
Task2 --> Synthesis
TaskN --> Synthesis
Execute --> Result([Final Result])
Synthesis --> Result
style Planner fill:#f9f,stroke:#333,stroke-width:2px
style Execute fill:#bfb,stroke:#333,stroke-width:2px
style Decompose fill:#bbf,stroke:#333,stroke-width:2px
style Synthesis fill:#fbb,stroke:#333,stroke-width:2px
style Result fill:#bfb,stroke:#333,stroke-width:3px
Key Concepts
Recursive Decomposition:
- LLM decides: EXECUTE (solve directly) or RECURSE (break down)
- Sub-tasks solved independently and in parallel (async mode)
- Results synthesized into final output
Multi-Agent Routing:
- Router agent makes planning decisions
- Specialized agents handle specific task types
- Automatic agent selection based on task requirements
Context Management:
- Immutable
RLMContexttracks execution state - Breadcrumbs for debugging recursive paths
- Depth and step limits prevent runaway recursion
Configuration
Engine Parameters
| Parameter | Default | Description |
|---|---|---|
max_depth |
5 | Maximum recursion depth (recommend 15 for LLM over-decomposition) |
max_steps |
50 | Maximum total steps across all levels (recommend 200) |
default_model |
"gpt-4" | Model identifier for logging |
verbose |
False | Enable detailed logging |
Best Practices
- High Limits: Use
max_depth=15andmax_steps=200- LLMs tend to over-decompose - Explicit Instructions: Add "execute directly" to task descriptions for simple tasks
- Simple Tasks: Keep individual tasks focused (150-200 words)
- Async for Parallelism: Use
AsyncRecursiveEnginefor 8-10ร throughput on parallel sub-tasks
Error Handling
Common Errors
RecursionDepthError:
from rlm.exceptions import RecursionDepthError
try:
result = engine.solve(task)
except RecursionDepthError as e:
# Increase max_depth or simplify task
print(f"Task too deep: {e}")
MaxStepsError:
from rlm.exceptions import MaxStepsError
try:
result = engine.solve(task)
except MaxStepsError as e:
# Increase max_steps or break into smaller tasks
print(f"Too many steps: {e}")
LLM Integration
RLM works with any LLM by implementing a simple interface:
from rlm.types import Input, Output
def your_llm_caller(inputs: list[Input], context: dict[str, Any]) -> Output:
"""Custom LLM implementation."""
# Convert inputs to your LLM's format
messages = [{"role": msg["role"], "content": msg["content"]} for msg in inputs]
# Call your LLM
response = your_llm_api.complete(
messages=messages,
temperature=context.get("temperature", 0.7),
)
# Return standardized output
return {
"content": response.text,
"metadata": {
"model": "your-model",
"usage": {
"prompt_tokens": response.prompt_tokens,
"completion_tokens": response.completion_tokens,
"total_tokens": response.total_tokens,
},
},
}
Supported LLMs
- โ OpenAI (GPT-4.1, GPT-4, etc.)
- โ Anthropic (Claude)
- โ Any LLM via custom caller implementation
Performance
Async Throughput
AsyncRecursiveEngine provides 8-10ร throughput improvement over synchronous execution for tasks with parallel sub-tasks:
# Synchronous: ~30s for 5 parallel sub-tasks
result = engine.solve(task)
# Async: ~3-4s for same task
result = await async_engine.solve(task)
Optimization Tips
- Use
AsyncRecursiveEnginefor parallel sub-tasks - Enable caching to avoid redundant LLM calls
- Set appropriate limits (
max_depth,max_steps) - Use streaming for long-running tasks
- Enable checkpointing for fault tolerance
Project Structure
rlm/
โโโ src/
โ โโโ rlm/
โ โโโ engine.py # Core RecursiveEngine
โ โโโ async_engine.py # AsyncRecursiveEngine
โ โโโ types.py # Type definitions
โ โโโ exceptions.py # Custom exceptions
โ โโโ prompts.py # System prompts
โ โโโ utils.py # Utilities
โโโ examples/
โ โโโ 01_basic_example.py # Basic recursion
โ โโโ 02_multi_agent_example.py
โ โโโ 03_tool_calling_example.py
โ โโโ 04_streaming_example.py
โ โโโ 05_checkpoint_example.py
โ โโโ 06_advanced_example.py
โ โโโ README.md # Examples documentation
โ โโโ STATUS.md # Technical details
โโโ tests/ # Test suite
โโโ docs/ # Documentation
โโโ pyproject.toml # Project configuration
โโโ README.md # This file
Testing
# Run all tests
uv run pytest
# Run with coverage
uv run pytest --cov=src --cov-report=term-missing
# Run specific test
uv run pytest tests/test_engine.py
Development
Type Checking
# Run mypy
uv run mypy src/
# Run pyright
uv run pyright src/
Code Quality
# Run linter
uv run ruff check .
# Fix auto-fixable issues
uv run ruff check . --fix
Contributing
- Fork the repository
- Create a feature branch (
git checkout -b feat/your-feature) - Make your changes
- Run tests and type checking
- Commit with conventional commit format
- Push and create a pull request
Commit Format
feat(scope): add new feature
fix(scope): fix bug
docs(scope): update documentation
refactor(scope): refactor code
test(scope): add tests
Documentation
- Examples Guide - Comprehensive examples
- System Design - Architecture details
- API Reference - API documentation
- Examples Status - Technical implementation details
Troubleshooting
RecursionDepthError
Problem: Task exceeds max_depth limit
Solutions:
- Increase
max_depthto 15 or higher - Add "execute directly" instruction to task
- Simplify task description
- Use tool calling for data access instead of recursion
MaxStepsError
Problem: Task exceeds max_steps limit
Solutions:
- Increase
max_stepsto 200 or higher - Break task into smaller independent tasks
- Reduce task complexity
Empty or Invalid JSON Errors
Problem: LLM returns invalid JSON during planning
Cause: This shouldn't happen with RecursiveEngine as the planner prompt explicitly requests JSON
Solutions:
- Verify your LLM caller returns complete responses
- Check that
contentfield is not empty - Ensure LLM has sufficient
max_tokens(recommend 2000)
License
MIT License - Copyright (c) 2026 Tom Mathews
See LICENSE for details.
Support
- Issues: GitHub Issues
- Discussions: GitHub Discussions
Citation
If you use RLM in your research, please cite:
@software{rlm2026,
title = {RLM: Recursive Language Models},
author = {Mathews, Tom},
year = {2026},
url = {https://github.com/Mathews-Tom/rlm}
}
Acknowledgments
Built with modern Python tooling:
- Type checking: mypy, pyright
- Package management: uv
- Testing: pytest
- Code quality: ruff
Status: Production Ready โ
All examples work reliably out of the box. See examples/STATUS.md for technical details.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file llm_recursive-0.1.0.tar.gz.
File metadata
- Download URL: llm_recursive-0.1.0.tar.gz
- Upload date:
- Size: 47.8 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.12.8
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
0e3483feb1ec10e78fabe731bda546325fe733df77587b7be8193516d0a70b5e
|
|
| MD5 |
d79ab8b90166ebef4d8d6f27e9f37674
|
|
| BLAKE2b-256 |
35ad3da41ee813ad1875901111130b60f15e42475dbf90443434b6519d2a906f
|
File details
Details for the file llm_recursive-0.1.0-py3-none-any.whl.
File metadata
- Download URL: llm_recursive-0.1.0-py3-none-any.whl
- Upload date:
- Size: 54.9 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.12.8
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
be7e136bb554a0cf9fa76d3fa71a413520c30f694e8ca870d6ecf902e662172b
|
|
| MD5 |
f71161b9a0c2a55b3342a646d21880c0
|
|
| BLAKE2b-256 |
a707e1aa92f9306996245d7e13dd552d66ae2adeceae15b5a7b10c3ae42fdacf
|