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:
- Fenced blocks with the specified language tag (
```replor```python) - 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)