Skip to content

ThinkingManager

File: brain/agent/thinking.py

Purpose

ThinkingManager controls two aspects of agent reasoning:

  1. Thinking depth — how much reasoning the LLM is asked to do before each response (via system prompt augmentation and token budgets)
  2. Reflection scheduling — when to trigger mid-turn self-assessment steps

ThinkLevel Enum

class ThinkLevel(Enum):
    OFF    = "off"     # No thinking — fastest responses
    LOW    = "low"     # Brief (~500 token budget)
    MEDIUM = "medium"  # Moderate reasoning (~2000 token budget)
    HIGH   = "high"    # Deep analysis (~5000 token budget)

Properties

ThinkLevel.MEDIUM.token_budget   # → 2000
ThinkLevel.MEDIUM.prompt_prefix  # → "Wrap your reasoning in <think>...</think> before your answer."

Construction from string

ThinkLevel.from_string("medium")  # → ThinkLevel.MEDIUM
ThinkLevel.from_string("invalid") # → ThinkLevel.OFF (safe default)

ThinkingConfig

@dataclass
class ThinkingConfig:
    level: ThinkLevel = ThinkLevel.OFF
    reflect_every: int = 0        # 0 = never; N = reflect every N turns
    reflect_on_tool_error: bool = True
    reflect_on_uncertainty: bool = True

ThinkingManager

manager = ThinkingManager(
    config=ThinkingConfig(
        level=ThinkLevel.MEDIUM,
        reflect_every=3,
        reflect_on_tool_error=True,
    )
)

State

manager._turn_count: int            # Increments on each on_turn_start()
manager._tool_errors: list[str]     # Cleared after each reflection
manager._last_reflection: ReflectionResult | None

Turn Lifecycle

manager.on_turn_start()        # Called at the start of each run_turn(); increments turn_count
manager.on_tool_error(error)   # Called by BoundedToolExecutor on error/exception
manager.record_reflection(r)   # Called by ReflectionEngine; clears tool_errors

Thinking Prompt

system_prompt_override = manager.get_thinking_prompt(base_prompt)

Builds the system prompt by appending thinking instructions:

[base_prompt]

## Thinking Instructions
Wrap your reasoning in <think>...</think> before your answer.

For ThinkLevel.OFF, the base prompt is returned unchanged.

Reflection Scheduling

manager.should_reflect()
# → True if:
#     level != OFF
#     AND (tool errors exist OR periodic interval hit)

Reflection Prompt

prompt = manager.get_reflection_prompt(messages, tool_results)

Builds a reflection prompt asking the LLM to assess: - What has been accomplished - Whether the approach is on track - What information is still missing - The last 3 tool results (with status) - Any recent tool error (as a note)

Harness Integration

# In AgentHarness.__init__():
self._thinking = ThinkingManager(
    ThinkingConfig(level=think_level, reflect_every=reflect_every)
)

# In run_turn():
self._thinking.on_turn_start()

# In prepare():
system_prompt_override = self._thinking.get_thinking_prompt(base_prompt)

# In _execute_turn_loop_v2() via ReflectionEngine:
if self._reflection_engine.should_reflect():   # delegates to manager.should_reflect()
    self._reflection_engine.reflect(...)        # uses manager.get_reflection_prompt()
                                                # then calls manager.record_reflection()

ReflectionResult

@dataclass
class ReflectionResult:
    reflection: str              # LLM's self-assessment text
    should_continue: bool = True # Always True — reflection never stops the loop
    suggested_action: str | None = None
    confidence: float = 0.7

    def to_dict(self) -> dict: ...

Properties

manager.last_reflection   # → ReflectionResult | None (most recent)
manager.turn_count        # → int (how many turns this session)