Memory Compaction¶

from rlm_code.rlm.memory_compaction import CompactionConfig

# Default configuration
config = CompactionConfig()

# Custom configuration for aggressive compaction
config = CompactionConfig(
    min_entries_for_compaction=3,
    max_entries_before_compaction=6,
    max_chars_before_compaction=4000,
    summary_max_sentences=2,
    preserve_last_n_entries=1,
    include_key_findings=True,
    use_llm_for_summary=True,
    fallback_to_deterministic=True,
)

from rlm_code.rlm.memory_compaction import CompactionResult

# Returned by MemoryCompactor.compact()
result: CompactionResult
print(f"Compressed {result.original_entries} entries to {result.compacted_entries}")
print(f"Character reduction: {result.original_chars} -> {result.compacted_chars}")
print(f"Compression ratio: {result.compression_ratio:.1%}")
print(f"Used LLM: {result.used_llm}")

from rlm_code.rlm.memory_compaction import MemoryCompactor, CompactionConfig

# Default compactor
compactor = MemoryCompactor()

# Compactor with custom config and LLM connector
compactor = MemoryCompactor(
    config=CompactionConfig(
        max_entries_before_compaction=8,
        preserve_last_n_entries=3,
    ),
    llm_connector=my_llm_connector,
)

compactor = MemoryCompactor()
# Later, when LLM is available:
compactor.set_llm_connector(my_llm_connector)

from rlm_code.rlm.repl_types import REPLHistory

history = REPLHistory()
for i in range(12):
    history = history.append(reasoning=f"Step {i}", code=f"x = {i}", output=str(i))

compactor = MemoryCompactor()
if compactor.should_compact(history):
    print("History needs compaction!")

result = compactor.compact(history, task="Analyze the dataset")

print(f"Summary: {result.summary}")
print(f"Preserved {len(result.preserved_entries)} recent entries")
print(f"Compression: {result.compression_ratio:.0%}")

result = compactor.compact(history, task="Analyze data")
compacted_history = compactor.apply_compaction(history, result)

print(f"Original: {len(history)} entries")
print(f"Compacted: {len(compacted_history)} entries")

# The first entry contains the summary
first = compacted_history.entries[0]
print(first.reasoning)   # "[COMPACTED] Working on: Analyze data. Completed 8 steps..."
print(first.code)        # "# Previous steps summarized above"
print(first.output)      # "(Compacted 10 steps)"

# Enable LLM summarization (default)
config = CompactionConfig(use_llm_for_summary=True)
compactor = MemoryCompactor(config=config, llm_connector=my_llm)

# Force deterministic summarization
config = CompactionConfig(use_llm_for_summary=False)
compactor = MemoryCompactor(config=config)

result = compactor.compact(history, task="Parse the log file")
print(result.summary)
# "Working on: Parse the log file. Completed 8 steps (6 successful).
#  Made 3 LLM sub-calls. Key findings: lines=1024; errors=12.
#  Resolved issues: KeyError, ValueError"

from rlm_code.rlm.memory_compaction import ConversationMemory

memory = ConversationMemory(max_turns=20)

# Add conversation turns
memory.add_turn(
    user_message="What's in the dataset?",
    assistant_response="The dataset contains 1000 records with 5 columns...",
    task="Analyze dataset",
)

memory.add_turn(
    user_message="Find the outliers",
    assistant_response="I found 12 outliers in column 'price'...",
)

# Get context for next LLM call
context = memory.get_context()
print(context)

from rlm_code.rlm.memory_compaction import (
    MemoryCompactor,
    CompactionConfig,
    ConversationMemory,
)
from rlm_code.rlm.repl_types import REPLHistory

# 1. Configure compaction
config = CompactionConfig(
    max_entries_before_compaction=8,
    max_chars_before_compaction=5000,
    preserve_last_n_entries=2,
    summary_max_sentences=2,
    use_llm_for_summary=False,  # Deterministic for this example
)

# 2. Create compactor
compactor = MemoryCompactor(config=config)

# 3. Simulate REPL history building
history = REPLHistory()
for i in range(10):
    history = history.append(
        reasoning=f"Step {i}: analyzing chunk {i}",
        code=f"result_{i} = analyze(chunks[{i}])",
        output=f"Processed chunk {i}: found {i * 3} patterns",
    )

# 4. Check and compact
if compactor.should_compact(history):
    result = compactor.compact(history, task="Analyze all data chunks")
    print(f"Compression ratio: {result.compression_ratio:.0%}")

    # 5. Apply compaction
    compacted = compactor.apply_compaction(history, result)
    print(f"Entries: {len(history)} -> {len(compacted)}")
    # Entries: 10 -> 3  (1 summary + 2 preserved)

config = CompactionConfig(
    min_entries_for_compaction=3,
    max_entries_before_compaction=5,
    max_chars_before_compaction=3000,
    use_llm_for_summary=False,  # Avoid extra LLM calls
    preserve_last_n_entries=1,
    summary_max_sentences=2,
)

config = CompactionConfig(
    min_entries_for_compaction=8,
    max_entries_before_compaction=15,
    max_chars_before_compaction=15000,
    use_llm_for_summary=True,
    fallback_to_deterministic=True,
    preserve_last_n_entries=3,
    summary_max_sentences=4,
    include_key_findings=True,
)

config = CompactionConfig(
    min_entries_for_compaction=2,
    max_entries_before_compaction=4,
    max_chars_before_compaction=2000,
    use_llm_for_summary=False,
    preserve_last_n_entries=1,
    summary_max_sentences=1,
)

config = CompactionConfig(
    min_entries_for_compaction=999,  # Effectively never trigger
    max_entries_before_compaction=999,
    max_chars_before_compaction=999999,
)