reasoning_state_library#

Predefined body factories for reasoning states — states whose behaviour is driven by an LLM in a tool-calling loop instead of a hand-written body.

The reasoning_body() factory builds a state body that runs a plan→act→observe loop. On each turn the LLM may:

call any registered tool, including the three built-in planning tools — add_tasks, complete_task and skip_task — that read and write a request-scoped TaskList;
emit a final text answer, which is sent back to the user only when every task on the list is either completed or skipped (or no tasks were ever added). If unresolved tasks remain, the orchestrator pushes back with a system message and the loop continues.

This evolves the earlier pure-ReAct loop: simple single-step requests still work with no task list at all (the LLM just answers), while complex multi-step requests are explicitly planned, tracked and either finished or skipped before a final answer is allowed through.

See baf.reasoning.tool.Tool, baf.reasoning.skill.Skill, and baf.reasoning.workspace.Workspace for the user-facing primitives.

class baf.library.state.reasoning_state_library.ReasoningStep(kind, step, summary, details=<factory>)[source]#

Bases: object

One observable event emitted by the reasoning loop.

A ReasoningStep is the canonical shape forwarded to the user via session.platform.reply_reasoning_step and consumed by the UI client. Every kind shares the same envelope (kind + step + summary + details); kind-specific data lives inside details so adding a new kind never breaks existing consumers.

details#: Kind-specific structured payload, e.g. {"tool_calls": [...]} for LLM_TOOL_CALLS or {"task_id": ..., "result": ...} for TASK_COMPLETED.

kind#: One of the constants on ReasoningStepKind.

step#: The loop iteration number this event belongs to (0-indexed). REASONING_STARTED and REASONING_FINISHED carry the step at which they were emitted (0 for started, last step for finished).

summary#: A short human-readable line — suitable for direct rendering in a UI without inspecting details.

to_dict()[source]#: Serialize for transport (e.g., websocket payload message).

class baf.library.state.reasoning_state_library.ReasoningStepKind[source]#

Bases: object

String constants for every kind of ReasoningStep the loop emits.

Two of these — REASONING_STARTED and REASONING_FINISHED — bracket every reasoning_body invocation so a streaming UI knows when to open and close a “live trace” group around the steps in between.

LLM_TEXT = 'llm_text'#

LLM_TOOL_CALLS = 'llm_tool_calls'#

MAX_STEPS = 'max_steps'#

PUSHBACK = 'pushback'#

REASONING_FINISHED = 'reasoning_finished'#

REASONING_STARTED = 'reasoning_started'#

TASK_ADDED = 'task_added'#

TASK_COMPLETED = 'task_completed'#

TASK_SKIPPED = 'task_skipped'#

TOOL_RESULT = 'tool_result'#

class baf.library.state.reasoning_state_library.Task(id, description, status='pending', result='')[source]#

Bases: object

A single planned subtask within a reasoning loop.

description#: A short imperative description.

id#: Integer id, unique within a TaskList. Surfaced to the LLM so it can refer to specific tasks via complete_task / skip_task.

result = ''#: A short summary of the outcome — populated when the task is completed or skipped.

status = 'pending'#: One of pending / in_progress / completed / skipped. New tasks start as pending.

class baf.library.state.reasoning_state_library.TaskList[source]#

Bases: object

A request-scoped list of tasks the LLM is expected to resolve.

A new TaskList is created at the start of every reasoning_body invocation — it does not persist across user messages. The reasoning loop checks all_resolved() before accepting a final answer from the LLM.

add(description)[source]#: Append a new pending task to the list.

all_resolved()[source]#

True when no tasks are pending or in_progress.

Returns True for an empty list — the LLM is allowed to skip planning for simple single-step requests.

get(task_id)[source]#: Return the task with task_id or None.

pending()[source]#: Return the still-unresolved tasks.

to_dict()[source]#: Serialize for the session scratchpad.

to_prompt()[source]#: Render the current state of the list as a markdown block.

update(task_id, status, result='')[source]#: Update the status (and optionally the result) of an existing task.

baf.library.state.reasoning_state_library._assistant_tool_call_message(tool_calls)[source]#

Build the assistant message that records the tool calls the LLM just emitted.

OpenAI’s chat format requires the model’s tool-call announcement to be preserved in the conversation history (with matching tool_call_id values) before the tool-result messages.

baf.library.state.reasoning_state_library._build_pushback_message(pending)[source]#: Compose the system message sent when the LLM tries to finalize too early.

baf.library.state.reasoning_state_library._build_system_prompt(skills, workspaces, base_prompt, task_list)[source]#: Concatenate base prompt + skills + workspace previews + task list.

baf.library.state.reasoning_state_library._execute_tool_calls(tool_calls, tools_by_name, session=None, stream_steps=False, step=0)[source]#

Run each requested tool call and return the tool-result messages.

Each call is logged at DEBUG level with its arguments and a truncated view of the result. When stream_steps is True, a tool_result event is also forwarded to the platform for each non-task tool right after it runs — task tools emit their own task_added / task_completed / task_skipped events from inside the tool function, so they are skipped here to avoid duplicates.

baf.library.state.reasoning_state_library._run_reasoning_loop(session, llm, messages, tool_schemas, tools_by_name, skills, workspaces, task_list, step_cell, scratchpad, max_steps, system_prompt, fallback_message, stream_steps)[source]#: The actual think→act→observe loop, factored out so reasoning_body() can wrap it in a single reasoning_started / reasoning_finished bracket via try/finally.

baf.library.state.reasoning_state_library._send_step(session, step, stream_steps=True)[source]#

Forward a ReasoningStep to session.platform.reply_reasoning_step.

Silently no-ops when stream_steps is False or when the platform does not expose a reply_reasoning_step method (Telegram, A2A, etc.). Any exception from the platform is caught and logged so a UI bug cannot kill the reasoning loop.

baf.library.state.reasoning_state_library._send_task_list(session, task_list, stream_steps=True)[source]#

Forward the current task list snapshot to session.platform.reply_task_list_update.

Same gating as _send_step(). Sent alongside (not in place of) the matching task_added / task_completed / task_skipped step events so a UI can render a live task panel without reconstructing it from the step stream.

baf.library.state.reasoning_state_library._truncate_for_log(text, limit=500)[source]#: Truncate text to limit chars with a clear marker if cut.

baf.library.state.reasoning_state_library._truncate_for_payload(text, limit=4000)[source]#: Truncate text to limit chars with a clear marker if cut.

baf.library.state.reasoning_state_library.new_reasoning_state(agent, llm, name='reasoning_state', initial=True, max_steps=8, system_prompt="You are an autonomous reasoning agent. You have access to tools and to filesystem workspaces. When the user asks about a topic that could plausibly be answered by data in a registered workspace, ALWAYS browse the workspace first — call `list_directory` to see what's there, then `read_file` on relevant files — instead of answering from your training data. Use other tools when their description matches the user's intent. Keep replies concise and grounded in tool outputs.You MUST call `add_tasks` whenever a request requires more than one tool call or more than one piece of information to assemble a complete answer. Only skip the task list for trivial single-step questions like 'what time is it?'.", fallback_message="I couldn't reach a final answer within the step budget. Please rephrase or narrow your question.", enable_task_planning=True, stream_steps=True)[source]#

Create a predefined reasoning state on agent and return it.

Equivalent to:

state = agent.new_state(name, initial=initial)
state.set_body(reasoning_body(llm, ...))
return state

Importable from the library package so callers can register predefined states by import (consistent with how future predefined states will be added):

from baf.library.state.reasoning_state_library import new_reasoning_state

state = new_reasoning_state(agent, llm=gpt)
state.when_event().go_to(state)  # caller wires transitions externally

Transitions are intentionally NOT added here — the developer chooses how to connect this state to the rest of the agent’s state machine (typically a state.when_event().go_to(state) self-loop, but other shapes are valid: gating on a condition, chaining from another state, etc.).

Parameters:

agent (Agent) – the agent the state is created on.
llm (LLM) – the LLM that drives the reasoning loop. Must implement predict_with_tools.
name (str) – the state’s name. Defaults to 'reasoning_state'.
initial (bool) – whether this state is the agent’s initial state. Defaults to True.
max_steps (int) – maximum LLM turns per user message. Defaults to 8.
system_prompt (str) – the base system prompt prepended to the skill / workspace / task-list blocks.
fallback_message (str) – the message sent if max_steps is exhausted.
enable_task_planning (bool) – when False, fall back to a pure ReAct loop (no built-in planning tools, no push-back, first text response wins). Defaults to True.
stream_steps (bool) – when True, forward every intermediate event to the user via session.platform.reply_reasoning_step if implemented. Defaults to True.

Returns:

the newly created reasoning state, ready for transition wiring.

Return type:

State

baf.library.state.reasoning_state_library.reasoning_body(llm, max_steps=8, system_prompt="You are an autonomous reasoning agent. You have access to tools and to filesystem workspaces. When the user asks about a topic that could plausibly be answered by data in a registered workspace, ALWAYS browse the workspace first — call `list_directory` to see what's there, then `read_file` on relevant files — instead of answering from your training data. Use other tools when their description matches the user's intent. Keep replies concise and grounded in tool outputs.You MUST call `add_tasks` whenever a request requires more than one tool call or more than one piece of information to assemble a complete answer. Only skip the task list for trivial single-step questions like 'what time is it?'.", fallback_message="I couldn't reach a final answer within the step budget. Please rephrase or narrow your question.", enable_task_planning=True, stream_steps=True)[source]#

Build a state body that runs an LLM-driven plan→act→observe loop.

The body reads the user message from session.event and, on each iteration, calls llm.predict_with_tools with:

the agent’s registered tools (read at execution time from agent._tools);
the three built-in planning tools (add_tasks, complete_task, skip_task) when enable_task_planning=True;
a system message containing the active skills, workspace previews, task-planning guidance, and the live task list.

The loop terminates when the LLM returns a final text answer and every task on the list is resolved. If unresolved tasks remain, the orchestrator appends a push-back system message and continues. If max_steps is reached without convergence the configured fallback message is sent.

The final task list and a per-step trace (tool calls, results, push-backs) are persisted to session['reasoning_scratchpad'] for inspection.

Parameters:

llm (LLM) – the LLM that drives the loop. Must implement predict_with_tools.
max_steps (int) – maximum number of LLM turns per user message.
system_prompt (str) – the base system prompt prepended to the skill / workspace / task-list blocks.
fallback_message (str) – the message sent if max_steps is exhausted.
enable_task_planning (bool) – when False, fall back to a pure ReAct loop (no built-in planning tools, no push-back, first text response wins). Defaults to True.
stream_steps (bool) – when True, forward every intermediate event (LLM tool calls, tool results, task add/complete/skip, push-back, max-steps fallback) to the user via session.platform.reply_reasoning_step if the platform implements it. The final answer is still sent via session.reply. No-op on platforms without a reply_reasoning_step method. Defaults to True.

Returns:

a state body suitable for baf.core.state.State.set_body().

Return type:

Callable[[Session], None]