Skip to content

Termination Patterns

Module

rlm_code.rlm.termination

The termination module implements the FINAL() and FINAL_VAR() patterns from the RLM paper. These patterns provide a clean, structured mechanism for the LLM to signal task completion and return results from within the REPL execution loop.


Overview

In the RLM paradigm, the LLM operates inside an iterative REPL loop: it reasons, writes code, observes output, and repeats. The termination module answers the critical question: how does the LLM signal that it is done?

Two complementary patterns are provided:

Pattern Purpose Use Case
FINAL(answer) Return a direct answer value Simple answers, strings, dicts
FINAL_VAR("name") Return the value of a REPL variable Large computed results, DataFrames, complex objects

Both patterns work by raising a FinalOutput exception, which the runner catches to cleanly exit the REPL loop without requiring special return-value plumbing.

graph LR
    A[LLM generates code] --> B{Code contains FINAL?}
    B -->|FINAL\(answer\)| C[FinalOutput exception raised]
    B -->|FINAL_VAR\(name\)| D[FinalOutput exception raised]
    B -->|No| E[Continue REPL loop]
    C --> F[Runner catches exception]
    D --> F
    F --> G[Extract answer / resolve variable]
    G --> H[Return result]

Functions

FINAL(answer)

Signal completion with a direct answer value. This immediately terminates the RLM loop.

from rlm_code.rlm.termination import FINAL

# Direct string answer
FINAL("The answer is 42")

# Dictionary answer
FINAL({"sentiment": "positive", "confidence": 0.95})

# List answer
FINAL(["item1", "item2", "item3"])
Parameter Type Description
answer Any The final answer to return. Can be any Python value.

Returns: NoReturn -- this function always raises FinalOutput.

The answer parameter is wrapped in a dict as {"answer": answer, "type": "direct"} inside the FinalOutput exception.

When to use FINAL vs FINAL_VAR

Use FINAL() when the answer is a short, self-contained value that fits naturally as a function argument. Use FINAL_VAR() when the answer is a large computed object already stored in a variable -- this avoids duplicating it in the function call.


FINAL_VAR(variable_name)

Signal completion by referencing a variable in the REPL namespace. The runner resolves the variable value after catching the exception.

from rlm_code.rlm.termination import FINAL_VAR

# In REPL code generated by the LLM:
result = analyze_document(context)
summary = result["summary"]
FINAL_VAR("summary")
Parameter Type Description
variable_name str Name of the variable in the REPL namespace to return.

Returns: NoReturn -- this function always raises FinalOutput.

The variable name is wrapped as {"var": variable_name, "type": "variable"} inside the FinalOutput exception. The runner then calls resolve_final_var() to retrieve the actual value from the REPL namespace.

Variable Must Exist

The referenced variable must exist in the REPL namespace at the time of resolution. If it does not, a KeyError is raised with a helpful message listing the available variables.


detect_final_in_text(text)

Detect FINAL() or FINAL_VAR() patterns in LLM response text (not in executable code). This handles cases where the LLM writes the termination call in its natural-language response rather than inside a code block.

from rlm_code.rlm.termination import detect_final_in_text

# Detect in LLM prose
result = detect_final_in_text('Based on my analysis, FINAL("42")')
assert result.detected is True
assert result.final_type == "direct"
assert result.content == "42"

# Detect variable reference
result = detect_final_in_text('I have stored the answer. FINAL_VAR("result")')
assert result.detected is True
assert result.final_type == "variable"
assert result.content == "result"

# No termination signal
result = detect_final_in_text("Let me continue analyzing...")
assert result.detected is False
Parameter Type Description
text str LLM response text to scan.

Returns: FinalDetection dataclass.

Uses multiple regex patterns to handle various formatting styles:

FINAL patterns:

Pattern Example Match
FINAL\s*\(\s*(.+?)\s*\) FINAL(The answer is 42)
FINAL\s*\(\s*["\'](.+?)["\']\s*\) FINAL("The answer is 42")
FINAL\s*\(\s*"""(.+?)"""\s*\) FINAL("""Multi-line answer""")

FINAL_VAR patterns:

Pattern Example Match
FINAL_VAR\s*\(\s*['\"]?(\w+)['\"]?\s*\) FINAL_VAR(result)
FINAL_VAR\s*\(\s*["\'](\w+)["\']\s*\) FINAL_VAR("result")

Detection Priority

FINAL_VAR patterns are checked before FINAL patterns because FINAL_VAR is the more specific match. This prevents FINAL_VAR("x") from being incorrectly parsed as a FINAL() call with VAR("x") as the argument.


detect_final_in_code(code)

Detect if executable code contains FINAL() or FINAL_VAR() calls. This is used to anticipate termination before actually executing the code.

from rlm_code.rlm.termination import detect_final_in_code

# Detect in code
result = detect_final_in_code('answer = 42\nFINAL(answer)')
assert result.detected is True
assert result.final_type == "direct"

# Detect variable reference in code
result = detect_final_in_code('FINAL_VAR("my_result")')
assert result.detected is True
assert result.final_type == "variable"
assert result.content == "my_result"
Parameter Type Description
code str Python code string to scan.

Returns: FinalDetection dataclass.

This function uses word-boundary-aware patterns (\b) to avoid false positives from variable names like FINALIZE or FINAL_VALUE.

Content for Direct FINAL

When detect_final_in_code() finds a FINAL( call (without _VAR), content is set to None because the actual answer value can only be determined at runtime after execution. For FINAL_VAR() calls, the variable name is extracted statically.


resolve_final_var(variable_name, namespace)

Resolve a FINAL_VAR reference by looking up the variable in the REPL namespace.

from rlm_code.rlm.termination import resolve_final_var

namespace = {"result": 42, "data": [1, 2, 3]}

value = resolve_final_var("result", namespace)
assert value == 42

# Missing variable raises KeyError with helpful message
try:
    resolve_final_var("missing_var", namespace)
except KeyError as e:
    print(e)
    # "FINAL_VAR referenced variable 'missing_var' not found in REPL namespace.
    #  Available variables: ['result', 'data']"
Parameter Type Description
variable_name str Name of the variable to resolve.
namespace dict[str, Any] The REPL's locals/globals namespace.

Returns: The value of the variable.

Raises: KeyError if the variable is not found (includes a list of available variables in the error message).


extract_code_blocks(text, language="repl")

Extract code blocks from LLM response text. Looks for markdown-style fenced code blocks with the specified language tag.

from rlm_code.rlm.termination import extract_code_blocks

response = """
Let me analyze the data.

```python
result = sum(data)
print(result)
```

And then finalize:

```repl
FINAL(result)
```
"""

blocks = extract_code_blocks(response)
# Returns: ['result = sum(data)\nprint(result)', 'FINAL(result)']
Parameter Type Default Description
text str required LLM response text containing code blocks.
language str "repl" Primary language tag to look for.

Returns: list[str] -- list of code strings extracted from code blocks.

Search order:

  1. Fenced blocks with the specified language tag (```repl or ```python)
  2. If none found, untagged code blocks (```) that look like Python code

Heuristic for untagged blocks: Must contain at least one of import, def, class, print(, or =.

Fallback Behavior

If no code blocks with the specified language tag (or python) are found, the function falls back to untagged code blocks and applies the Python keyword heuristic. This handles cases where the LLM omits the language tag from its code blocks.


format_final_answer(answer)

Format a final answer for display or return. Handles various answer types with appropriate formatting.

from rlm_code.rlm.termination import format_final_answer

# String passthrough
assert format_final_answer("hello") == "hello"

# Dict with "answer" key -- extracts the value
assert format_final_answer({"answer": 42}) == "42"

# Dict without "answer" key -- JSON formatted
result = format_final_answer({"key": "value", "count": 10})
# Returns: '{\n  "key": "value",\n  "count": 10\n}'

# List -- joined by newlines
assert format_final_answer(["line1", "line2"]) == "line1\nline2"

# Other types -- str() conversion
assert format_final_answer(42) == "42"
Parameter Type Description
answer Any The final answer to format.

Returns: str -- the formatted answer.

Input Type Formatting Behavior
str Returned as-is
dict with "answer" key Returns str(answer["answer"])
dict without "answer" key JSON-formatted with 2-space indent
list Items joined by newlines
Other str() conversion

Classes

FinalOutput

Control-flow exception raised when FINAL() or FINAL_VAR() is called. This follows the pattern from DSPy's RLM implementation where termination is handled as an exception to cleanly exit the REPL execution loop.

from rlm_code.rlm.termination import FinalOutput, FINAL

# Typically not instantiated directly -- use FINAL() or FINAL_VAR()
try:
    FINAL("The answer is 42")
except FinalOutput as e:
    print(e.output)
    # {"answer": "The answer is 42", "type": "direct"}
Attribute Type Description
output dict[str, Any] Dictionary containing the answer or variable reference.

The output dictionary has two possible shapes:

Key Value (FINAL) Value (FINAL_VAR)
"answer" The direct answer value not present
"var" not present The variable name string
"type" "direct" "variable"

Design Pattern

Using an exception for termination provides a clean way to exit from arbitrarily nested code execution -- even from inside helper functions, loops, or llm_query() callbacks. It means the LLM does not need to structure its code with explicit return paths.


FinalDetection

Dataclass holding the result of detecting FINAL/FINAL_VAR patterns in text or code. Uses @dataclass(slots=True) for memory efficiency.

from rlm_code.rlm.termination import FinalDetection

# Constructed by detect_final_in_text() and detect_final_in_code()
detection = FinalDetection(
    detected=True,
    final_type="direct",
    content="42",
    raw_match='FINAL("42")',
)
Field Type Default Description
detected bool required Whether a termination pattern was found.
final_type str \| None None "direct" for FINAL(), "variable" for FINAL_VAR().
content str \| None None The extracted answer or variable name.
raw_match str \| None None The full matched pattern string from the regex.

Regex Pattern Reference

The module defines two sets of compiled regex patterns for flexible detection of termination calls across different LLM formatting styles.

FINAL_PATTERNS

Three patterns for detecting FINAL(answer):

Pattern Flags Matches
FINAL\s*\(\s*(.+?)\s*\)(?:\s*$\|\n) DOTALL \| MULTILINE Multiline FINAL(...)
FINAL\s*\(\s*["\'](.+?)["\']\s*\) DOTALL FINAL("quoted string")
FINAL\s*\(\s*"""(.+?)"""\s*\) DOTALL FINAL("""triple quoted""")

FINAL_VAR_PATTERNS

Two patterns for detecting FINAL_VAR(name):

Pattern Matches
FINAL_VAR\s*\(\s*['\"]?(\w+)['\"]?\s*\) FINAL_VAR(name) or FINAL_VAR("name")
FINAL_VAR\s*\(\s*["\'](\w+)["\']\s*\) FINAL_VAR("name") or FINAL_VAR('name')

Whitespace Tolerance

All patterns tolerate arbitrary whitespace around parentheses and arguments, so FINAL( answer ), FINAL (answer), and FINAL(answer) all match correctly.


End-to-End Flow

Here is how termination works in a complete RLM execution:

LLM Response
    |
    v
extract_code_blocks() --> ["code with FINAL(...)"]
    |
    v
detect_final_in_code() --> FinalDetection(detected=True)
    |
    v
exec(code, namespace) --> raises FinalOutput
    |
    v
Caught by PureRLMEnvironment._execute_code()
    |
    v
Check final_output["type"]:
    |
    +-- "direct" --> format_final_answer(answer)
    |
    +-- "variable" --> resolve_final_var(var_name, namespace)
                           --> format_final_answer(value)
    |
    v
EnvironmentActionResult(done=True, final_response=answer)

If the LLM does not use code blocks but writes FINAL() directly in text:

LLM Response Text
    |
    v
detect_final_in_text() --> FinalDetection(detected=True, content="answer")
    |
    v
EnvironmentActionResult(done=True, final_response="answer")

Integration with the Runner

The termination module is imported and used by the PureRLMEnvironment to manage the complete execution lifecycle:

from rlm_code.rlm.termination import (
    FINAL, FINAL_VAR, FinalOutput,
    detect_final_in_code, detect_final_in_text,
    extract_code_blocks, format_final_answer,
    resolve_final_var,
)

# Inside the REPL execution loop:
try:
    exec(code, namespace)
except FinalOutput as e:
    if e.output["type"] == "variable":
        # Resolve the variable from namespace
        answer = resolve_final_var(e.output["var"], namespace)
    else:
        # Direct answer
        answer = e.output["answer"]

    formatted = format_final_answer(answer)
    # Return formatted answer as the run result

Exception-Based Control Flow

The FINAL() / FINAL_VAR() mechanism uses exceptions for control flow. This is intentional: it provides a clean way to exit arbitrarily deep call stacks within the REPL without requiring cooperative return-value threading. However, bare except Exception clauses in REPL code will catch FinalOutput and prevent termination. The PureRLMEnvironment handles this by catching FinalOutput at the outermost execution boundary.


Examples

Basic Task Completion

# LLM-generated REPL code:
data = [1, 2, 3, 4, 5]
total = sum(data)
average = total / len(data)
FINAL(f"The average is {average}")

Variable-Based Completion

# LLM-generated REPL code for large output:
import json

results = {}
for key in context.keys():
    results[key] = analyze(context[key])

# Store in variable to avoid token-window bloat
final_report = json.dumps(results, indent=2)
FINAL_VAR("final_report")

Multi-Step Analysis with Sub-LLM

# LLM-generated REPL code with recursive LLM calls:
chunks = [context[i:i+1000] for i in range(0, len(context), 1000)]
summaries = []

for chunk in chunks:
    summary = llm_query(f"Summarize: {chunk}")
    summaries.append(summary)

combined = "\n".join(summaries)
final_answer = llm_query(f"Synthesize these summaries: {combined}")
FINAL(final_answer)

Conditional Termination

# LLM-generated REPL code with conditional FINAL:
word_count = len(context.split())

if word_count < 100:
    FINAL("Document too short for meaningful analysis")

# If we get here, document is long enough
analysis = perform_detailed_analysis(context)
FINAL_VAR("analysis")