Planner

The ReactPlanner and its supporting configuration: tool context, reflection and tool-selection policies, parallel-call/join primitives, and trajectory objects.

Skill provider and configuration types re-exported by the planner are documented under Skills.

planner

Planner entry points.

AnyContext module-attribute

AnyContext = ToolContext | FlowContext

PlannerEventCallback module-attribute

PlannerEventCallback = Callable[[PlannerEvent], None]

ReactPlanner

ReactPlanner(llm: str | Mapping[str, Any] | None = None, *, nodes: Sequence[Node] | None = None, catalog: Sequence[NodeSpec] | None = None, registry: ModelRegistry | None = None, llm_client: JSONLLMClient | None = None, max_iters: int = 8, temperature: float = 0.0, json_schema_mode: bool = True, system_prompt_extra: str | None = None, token_budget: int | None = None, pause_enabled: bool = True, state_store: Any | None = None, artifact_store: ArtifactStore | None = None, observation_guardrail: ObservationGuardrailConfig | None = None, summarizer_llm: str | Mapping[str, Any] | None = None, planning_hints: Mapping[str, Any] | None = None, repair_attempts: int = 3, max_consecutive_arg_failures: int = 3, arg_fill_enabled: bool = True, deadline_s: float | None = None, hop_budget: int | None = None, time_source: Callable[[], float] | None = None, event_callback: PlannerEventCallback | None = None, llm_timeout_s: float = 360.0, llm_max_retries: int = 3, use_native_reasoning: bool = True, reasoning_effort: str | None = None, reasoning_display: ReasoningDisplay = None, llm_fallback: ModelFallbackConfig | None = None, absolute_max_parallel: int = 50, reflection_config: ReflectionConfig | None = None, reflection_llm: str | Mapping[str, Any] | None = None, tool_policy: ToolPolicy | None = None, stream_final_response: bool = False, final_response_model: type[BaseModel] | None = None, final_response_retries: int = 1, tool_call_mode: str = 'prompted', llm_transport: str | None = None, short_term_memory: ShortTermMemory | ShortTermMemoryConfig | None = None, background_tasks: BackgroundTasksConfig | None = None, error_recovery: ErrorRecoveryConfig | None = None, multi_action_sequential: bool = False, multi_action_read_only_only: bool = True, multi_action_max_tools: int = 2, auto_seq_enabled: bool = False, auto_seq_execute: bool = False, auto_seq_read_only_only: bool = True, use_native_llm: bool = False, guardrail_gateway: Any | None = None, guardrail_conversation_history_turns: int = 1, tool_search: ToolSearchConfig | None = None, tool_examples: ToolExamplesConfig | None = None, skills: SkillsConfig | None = None, skills_provider: Any | None = None, skills_provider_factory: Any | None = None, llm_context_hooks: Sequence[LLMContextHook] | None = None)

JSON-only ReAct planner for autonomous multi-step workflows.

The ReactPlanner orchestrates a loop where an LLM selects and sequences PenguiFlow nodes/tools based on structured JSON contracts. It supports pause/resume for approvals, adaptive re-planning on failures, parallel execution, and trajectory compression for long-running sessions.

Thread Safety

NOT thread-safe. Create separate planner instances per task.

Parameters:

Name	Type	Description	Default
llm	str | Mapping[str, Any] | None	LiteLLM model name (e.g., "gpt-4") or config dict. Required if llm_client is not provided.	None
nodes	Sequence[Node] | None	Sequence of PenguiFlow nodes to make available as tools. Either (nodes + registry) or catalog must be provided.	None
catalog	Sequence[NodeSpec] | None	Pre-built tool catalog. If provided, nodes and registry are ignored.	None
registry	ModelRegistry | None	Model registry for type resolution. Required if nodes is provided.	None
llm_client	JSONLLMClient | None	Custom LLM client implementation. If provided, llm is ignored.	None
max_iters	int	Maximum planning iterations before returning no_path. Default: 8.	8
temperature	float	LLM sampling temperature. Default: 0.0 for deterministic output.	0.0
json_schema_mode	bool	Enable strict JSON schema enforcement via LLM response_format. Default: True.	True
system_prompt_extra	str | None	Optional instructions for interpreting custom context (e.g., memory format). Use this to specify how the planner should use structured data passed via llm_context. The library provides baseline injection; this parameter lets you define format-specific semantics. Examples: "memories contains JSON with user preferences; respect them when planning" "context.knowledge is a flat list of facts; cite relevant ones" "Use context.history to avoid repeating failed approaches"	None
token_budget	int | None	If set, triggers trajectory summarization when history exceeds limit. Token count is estimated by character length (approx).	None
pause_enabled	bool	Allow nodes to trigger pause/resume flow. Default: True.	True
state_store	Any | None	Optional durable state adapter for pause/resume persistence.	None
summarizer_llm	str | Mapping[str, Any] | None	Separate (cheaper) LLM for trajectory compression. Falls back to main LLM if not set.	None
reflection_config	ReflectionConfig | None	Optional configuration enabling automatic answer critique before finishing. Disabled by default.	None
reflection_llm	str | Mapping[str, Any] | None	Optional LiteLLM identifier used for critique when reflection_config.use_separate_llm is True.	None
planning_hints	Mapping[str, Any] | None	Structured constraints and preferences (ordering, disallowed nodes, max_parallel, etc.). See plan.md for schema.	None
tool_policy	ToolPolicy | None	Optional runtime policy that filters the tool catalog (whitelists, blacklists, or tag requirements) for multi-tenant and safety use cases.	None
repair_attempts	int	Max attempts to repair invalid JSON from LLM. Default: 3.	3
max_consecutive_arg_failures	int	Max consecutive tool arg validation failures before forcing a finish with requires_followup=True. Helps small models avoid infinite loops when they repeatedly produce invalid args. Default: 3.	3
arg_fill_enabled	bool	Enable arg-fill mode for missing tool arguments. When True, if a tool call has valid tool selection but missing/invalid args, the planner will make a simplified LLM call asking only for the missing values instead of requiring a full JSON repair. This significantly improves success rates for small models. Default: True.	True
deadline_s	float | None	Wall-clock deadline for planning session (seconds from start).	None
hop_budget	int | None	Maximum tool invocations allowed.	None
time_source	Callable[[], float] | None	Override time.monotonic for testing.	None
event_callback	PlannerEventCallback | None	Optional callback receiving PlannerEvent instances for observability.	None
llm_timeout_s	float	Per-LLM-call timeout in seconds. Default: 360.0.	360.0
llm_max_retries	int	Max retry attempts for transient LLM failures. Default: 3.	3
absolute_max_parallel	int	System-level safety limit on parallel execution regardless of hints. Default: 50.	50

Raises:

Type	Description
ValueError	If neither (nodes + registry) nor catalog is provided, or if neither llm nor llm_client is provided.
RuntimeError	If LiteLLM is not installed and llm_client is not provided.

Example

planner = ReactPlanner( ... llm="gpt-4", ... nodes=[triage_node, retrieve_node, summarize_node], ... registry=my_registry, ... max_iters=10, ... ) result = await planner.run("Explain PenguiFlow's architecture") print(result.reason) # "answer_complete", "no_path", or "budget_exhausted"

DEFAULT_MAX_PARALLEL class-attribute instance-attribute

DEFAULT_MAX_PARALLEL = 50

artifact_store property

artifact_store: ArtifactStore

Return the configured artifact store (NoOp when disabled).

fork

fork(*, catalog_filter: Callable[[NodeSpec], bool] | None = None, tool_policy: ToolPolicy | None = None, inherit_policy: bool = True, background_tasks: BackgroundTasksConfig | None | Literal['inherit'] = 'inherit') -> ReactPlanner

Create a new planner instance with the same configuration.

Background task orchestration requires a fresh ReactPlanner per task because the planner maintains mutable per-run state and is not thread-safe.

Parameters:

Name	Type	Description	Default
catalog_filter	Callable[[NodeSpec], bool] | None	Optional predicate applied to each NodeSpec to restrict which tools the forked planner exposes.	None
tool_policy	ToolPolicy | None	Optional runtime policy applied to the forked planner. When inherit_policy is True it is intersected/merged with the parent's policy.	None
inherit_policy	bool	When True, merge tool_policy with the parent planner's policy (intersecting allowed tools, unioning denied tools and required tags). When False, tool_policy replaces the parent's policy outright. Default: True.	True
background_tasks	BackgroundTasksConfig | None | Literal['inherit']	Background tasks configuration for the forked planner. Defaults to "inherit", which reuses the parent's configuration.	'inherit'

Returns:

Type	Description
ReactPlanner	A new ReactPlanner instance sharing this planner's configuration.

run async

run(query: str, *, input_parts: Sequence[ContentPart] | None = None, llm_context: Mapping[str, Any] | None = None, context_meta: Mapping[str, Any] | None = None, tool_context: Mapping[str, Any] | None = None, memory_key: MemoryKey | None = None, steering: SteeringInbox | None = None, tool_visibility: ToolVisibilityPolicy | None = None) -> PlannerFinish | PlannerPause

Execute planner on a query until completion or pause.

Parameters:

Name	Type	Description	Default
query	str	Natural language task description.	required
input_parts	Sequence[ContentPart] | None	Optional image/audio content parts appended to the initial user message. Omitted by default to preserve text-only behavior.	None
llm_context	Mapping[str, Any] | None	Optional context visible to LLM (memories, status_history, etc.). Should NOT include internal metadata like tenant_id or trace_id.	None
context_meta	Mapping[str, Any] | None	Deprecated: Use llm_context instead. This parameter is kept for backward compatibility but will be removed in a future version.	None
tool_context	Mapping[str, Any] | None	Tool-only context (callbacks, loggers, telemetry objects). Not visible to the LLM. May contain non-serialisable objects.	None
memory_key	MemoryKey | None	Optional explicit short-term memory key. If omitted, the planner may derive a key from tool_context using the configured memory isolation paths. If no key is available and memory is configured to require an explicit key, memory behaves as disabled for this call.	None
steering	SteeringInbox | None	Optional steering inbox for injecting mid-run guidance into the planning loop.	None
tool_visibility	ToolVisibilityPolicy | None	Optional policy restricting which tools are visible for this call.	None

Returns:

Type	Description
PlannerFinish | PlannerPause	PlannerFinish if task completed/failed, PlannerPause if paused for human intervention.

Raises:

Type	Description
RuntimeError	If LLM client fails after all retries.

resume async

resume(token: str, user_input: str | None = None, *, tool_context: Mapping[str, Any] | None = None, memory_key: MemoryKey | None = None, steering: SteeringInbox | None = None, tool_visibility: ToolVisibilityPolicy | None = None) -> PlannerFinish | PlannerPause

Resume a paused planning session.

Parameters:

Name	Type	Description	Default
token	str	Resume token from a previous PlannerPause.	required
user_input	str | None	Optional user response to the pause (e.g., approval decision).	None
tool_context	Mapping[str, Any] | None	Tool-only context (callbacks, loggers, telemetry objects). Not visible to the LLM. May contain non-serialisable objects. Overrides any tool_context captured in the pause record.	None
memory_key	MemoryKey | None	Optional explicit short-term memory key for the resumed session. If omitted, the planner may derive a key from tool_context using the configured memory isolation paths. If no key is available and memory is configured to require an explicit key, memory behaves as disabled for this call.	None
steering	SteeringInbox | None	Optional steering inbox for injecting mid-run guidance into the planning loop.	None
tool_visibility	ToolVisibilityPolicy | None	Optional policy restricting which tools are visible for this call.	None

Returns:

Type	Description
PlannerFinish | PlannerPause	Updated result after resuming execution.

Raises:

Type	Description
KeyError	If resume token is invalid or expired.

step async

step(trajectory: Trajectory) -> PlannerAction

pause async

pause(reason: PlannerPauseReason, payload: Mapping[str, Any] | None = None) -> PlannerPause

ToolContext

Bases: Protocol

Protocol for planner tool execution context.

llm_context property

llm_context: Mapping[str, Any]

Context visible to LLM (read-only mapping).

tool_context property

tool_context: dict[str, Any]

Tool-only context (callbacks, telemetry objects, loggers, etc.).

meta property

meta: MutableMapping[str, Any]

Combined context. Deprecated: prefer llm_context/tool_context.

artifacts property

artifacts: ScopedArtifacts

Scoped artifact facade for tool developers.

Example

ref = await ctx.artifacts.upload( pdf_bytes, mime_type="application/pdf", filename="report.pdf", ) return {"artifact": ref, "summary": "Downloaded PDF"}

kv property

kv: SessionKV

Durable session key/value facade.

Backed by the configured StateStore's optional memory persistence. Default scope is session-scoped with no TTL. Task scope is opt-in and uses a fixed TTL of 3600 seconds.

pause

pause(reason: PlannerPauseReason, payload: Mapping[str, Any] | None = None) -> Awaitable[Any]

Pause execution for human input or policy decisions.

emit_chunk

emit_chunk(stream_id: str, seq: int, text: str, *, done: bool = False, meta: Mapping[str, Any] | None = None) -> Awaitable[None]

Emit a streaming chunk.

emit_artifact

emit_artifact(stream_id: str, chunk: Any, *, done: bool = False, artifact_type: str | None = None, meta: Mapping[str, Any] | None = None) -> Awaitable[None]

Emit a streaming artifact chunk (e.g., partial chart config).

PlannerAction

Bases: BaseModel

Unified action format (RFC_UNIFIED_ACTION_SCHEMA).

The LLM-facing schema is always: - next_node: non-null string opcode or tool name - args: object payload (defaults to {})

Internally, we keep a best-effort thought field for trajectory logging and repair prompts, but it is excluded from the JSON schema so it is not required (or encouraged) in structured outputs.

Special next_node values: - "final_response": Terminal action, args.answer streams to user - "parallel": Parallel execution, args contains steps and join config - "task.subagent": Background subagent task, args contains query and group config - "task.tool": Background single-tool job, args contains tool/tool_args and group config - Any other value: Tool call, args passed to the tool

next_node class-attribute instance-attribute

next_node: str = Field(description="Non-null opcode or tool name: 'final_response', 'parallel', 'task.subagent', 'task.tool', or any other tool name for a regular tool call.")

args class-attribute instance-attribute

args: dict[str, Any] = Field(default_factory=dict, description='Argument payload for next_node.')

thought class-attribute instance-attribute

thought: SkipJsonSchema[str] = Field(default='', description='Best-effort reasoning trace for trajectory logging and repair prompts.')

raw_llm_response class-attribute instance-attribute

raw_llm_response: SkipJsonSchema[str | None] = Field(default=None, exclude=True, description='Raw LLM response text, kept for debugging only (excluded from serialization).')

alternate_actions class-attribute instance-attribute

alternate_actions: SkipJsonSchema[list[dict[str, Any]] | None] = Field(default=None, exclude=True, description='Additional action candidates parsed from mixed model output, if any.')

is_terminal

is_terminal() -> bool

True if this is a terminal action (final response to user).

is_parallel

is_parallel() -> bool

True if this is a parallel execution plan.

is_background_task

is_background_task() -> bool

True if this is a background task spawn.

is_tool_call

is_tool_call() -> bool

True if this is a regular tool call.

get_answer

get_answer() -> str | None

Extract answer text for terminal actions.

answer_text

answer_text() -> str | None

Extract answer from args.answer or args.raw_answer (backward compatible).

get_plan_steps

get_plan_steps() -> list[PlanStep] | None

Extract parallel plan steps.

get_plan_join

get_plan_join() -> PlanJoin | None

Extract parallel plan join config.

PlannerFinish

Bases: BaseModel

Terminal result of a planner run.

reason class-attribute instance-attribute

reason: Literal['answer_complete', 'no_path', 'budget_exhausted'] = Field(description='Why the planner finished.')

payload class-attribute instance-attribute

payload: Any = Field(default=None, description='Final answer payload, if any.')

metadata class-attribute instance-attribute

metadata: dict[str, Any] = Field(default_factory=dict, description='Additional metadata about the run.')

PlannerPause

Bases: BaseModel

Signals that planner execution has paused and awaits external resumption.

reason class-attribute instance-attribute

reason: PlannerPauseReason = Field(description='Why the planner paused (e.g. HITL confirmation).')

payload class-attribute instance-attribute

payload: dict[str, Any] = Field(default_factory=dict, description='Data associated with the pause reason.')

resume_token class-attribute instance-attribute

resume_token: str = Field(description='Opaque token required to resume this paused run.')

PlannerEvent dataclass

PlannerEvent(event_type: str, ts: float, trajectory_step: int, thought: str | None = None, node_name: str | None = None, latency_ms: float | None = None, token_estimate: int | None = None, error: str | None = None, extra: Mapping[str, Any] = dict())

Structured event emitted during planner execution for observability.

event_type instance-attribute

event_type: str

ts instance-attribute

ts: float

trajectory_step instance-attribute

trajectory_step: int

thought class-attribute instance-attribute

thought: str | None = None

node_name class-attribute instance-attribute

node_name: str | None = None

latency_ms class-attribute instance-attribute

latency_ms: float | None = None

token_estimate class-attribute instance-attribute

token_estimate: int | None = None

error class-attribute instance-attribute

error: str | None = None

extra class-attribute instance-attribute

extra: Mapping[str, Any] = field(default_factory=dict)

to_payload

to_payload() -> dict[str, Any]

Render a dictionary payload suitable for structured logging.

ReflectionConfig

Bases: BaseModel

Configuration controlling the reflection loop behaviour.

enabled class-attribute instance-attribute

enabled: bool = Field(default=False, description='Whether the reflection/critique loop is enabled.')

criteria class-attribute instance-attribute

criteria: ReflectionCriteria = Field(default_factory=ReflectionCriteria, description='Quality criteria used by the reflection LLM to critique answers.')

quality_threshold class-attribute instance-attribute

quality_threshold: float = Field(default=0.8, ge=0.0, le=1.0, description='Minimum critique score required to accept an answer.')

max_revisions class-attribute instance-attribute

max_revisions: int = Field(default=2, ge=1, le=10, description='Maximum number of revision attempts before giving up.')

use_separate_llm class-attribute instance-attribute

use_separate_llm: bool = Field(default=False, description="Whether to use a separate LLM client for reflection instead of the planner's main client.")

ReflectionCriteria

Bases: BaseModel

Quality criteria used when critiquing an answer.

completeness class-attribute instance-attribute

completeness: str = Field(default='Addresses all parts of the query', description='Criterion describing what counts as a complete answer.')

accuracy class-attribute instance-attribute

accuracy: str = Field(default='Factually correct based on observations', description='Criterion describing what counts as an accurate answer.')

clarity class-attribute instance-attribute

clarity: str = Field(default='Well-explained and coherent', description='Criterion describing what counts as a clear answer.')

ReflectionCritique

Bases: BaseModel

Structured critique returned by the reflection LLM.

score class-attribute instance-attribute

score: float = Field(ge=0.0, le=1.0)

passed instance-attribute

passed: bool

feedback instance-attribute

feedback: str

issues class-attribute instance-attribute

issues: list[str] = Field(default_factory=list)

suggestions class-attribute instance-attribute

suggestions: list[str] = Field(default_factory=list)

ErrorRecoveryConfig dataclass

ErrorRecoveryConfig(enabled: bool = True, max_compress_retries: int = 1, compression_threshold_chars: int = 2000, summarize_on_compress: bool = True)

Configuration for LLM error recovery.

enabled class-attribute instance-attribute

enabled: bool = True

max_compress_retries class-attribute instance-attribute

max_compress_retries: int = 1

compression_threshold_chars class-attribute instance-attribute

compression_threshold_chars: int = 2000

summarize_on_compress class-attribute instance-attribute

summarize_on_compress: bool = True

LLMContextHook

Bases: Protocol

Hook that can patch llm_context before the first LLM call.

Hooks are opt-in and best-effort; failures should not block a run.

Optional attributes (read via getattr): - name: str: Used for events/logging. Defaults to class name. - overwrite: bool: When True, the hook may overwrite existing keys. Default: False.

before_run async

before_run(inp: LLMContextHookInput) -> Mapping[str, Any] | None

LLMContextHookInput dataclass

LLMContextHookInput(query: str, llm_context: Mapping[str, Any], tool_context: Mapping[str, Any], memory_key: MemoryKey | None)

Inputs to LLM context hooks.

llm_context and tool_context are the normalized values that will be used for the run. Hooks should treat them as read-only.

query instance-attribute

query: str

llm_context instance-attribute

llm_context: Mapping[str, Any]

tool_context instance-attribute

tool_context: Mapping[str, Any]

memory_key instance-attribute

memory_key: MemoryKey | None

LLMErrorType

Bases: Enum

Classification of LLM errors for recovery strategy selection.

CONTEXT_LENGTH_EXCEEDED class-attribute instance-attribute

CONTEXT_LENGTH_EXCEEDED = 'context_length_exceeded'

RATE_LIMIT class-attribute instance-attribute

RATE_LIMIT = 'rate_limit'

SERVICE_UNAVAILABLE class-attribute instance-attribute

SERVICE_UNAVAILABLE = 'service_unavailable'

TIMEOUT class-attribute instance-attribute

TIMEOUT = 'timeout'

BAD_REQUEST_OTHER class-attribute instance-attribute

BAD_REQUEST_OTHER = 'bad_request_other'

UNKNOWN class-attribute instance-attribute

UNKNOWN = 'unknown'

JoinInjection

Bases: BaseModel

Mapping of join args to parallel execution data sources.

mapping class-attribute instance-attribute

mapping: dict[str, str] = Field(default_factory=dict, description="Mapping of join argument names to result references (e.g. '$results').")

ParallelCall

Bases: BaseModel

Single tool invocation within a parallel execution plan.

node class-attribute instance-attribute

node: str = Field(description='Name of the tool/node to invoke.')

args class-attribute instance-attribute

args: dict[str, Any] = Field(default_factory=dict, description='Arguments passed to the tool call.')

ParallelJoin

Bases: BaseModel

Aggregation step run after all parallel calls in a plan complete.

node class-attribute instance-attribute

node: str = Field(description='Name of the tool/node to invoke for aggregation.')

args class-attribute instance-attribute

args: dict[str, Any] = Field(default_factory=dict, description='Arguments passed to the join call.')

inject class-attribute instance-attribute

inject: JoinInjection | None = Field(default=None, description="Optional mapping of parallel results into the join call's args.")

ToolPolicy

Bases: BaseModel

Runtime policy for tool availability and permissions.

allowed_tools class-attribute instance-attribute

allowed_tools: set[str] | None = Field(default=None, description='If set, only tool names in this set may be used; all others are denied.')

denied_tools class-attribute instance-attribute

denied_tools: set[str] = Field(default_factory=set, description='Tool names that are always denied.')

require_tags class-attribute instance-attribute

require_tags: set[str] = Field(default_factory=set, description="Tags a tool's node must all carry to be allowed.")

is_allowed

is_allowed(node_name: str, node_tags: Mapping[str, Any] | Sequence[str]) -> bool

Check whether a tool is permitted under this policy.

Parameters:

Name	Type	Description	Default
node_name	str	Name of the tool/node being checked.	required
node_tags	Mapping[str, Any] | Sequence[str]	Tags associated with the node, either as a mapping (keys used as tags) or a sequence of tag strings.	required

Returns:

Type	Description
bool	True if the tool is allowed, False if it is denied, not in the allow
bool	list, or missing one or more required tags.

ToolVisibilityPolicy

Bases: Protocol

Dynamic, per-run filtering for which tools are shown to the LLM.

This is opt-in and intended for per-tenant/per-user tool visibility without constructing a brand new planner instance. Implementations must only return specs from the provided specs sequence.

visible_tools

visible_tools(specs: Sequence[NodeSpec], tool_context: Mapping[str, Any]) -> Sequence[NodeSpec]

ToolSearchConfig

Bases: BaseModel

Configuration for on-demand tool search and lazy tool loading.

enabled class-attribute instance-attribute

enabled: bool = Field(default=False, description='Whether tool search/lazy loading is enabled.')

cache_dir class-attribute instance-attribute

cache_dir: str = Field(default='.penguiflow', description='Directory used to persist the tool search index.')

default_loading_mode class-attribute instance-attribute

default_loading_mode: ToolLoadingMode = Field(default=ToolLoadingMode.ALWAYS, description='Default loading mode applied to tools without an explicit mode.')

always_loaded_patterns class-attribute instance-attribute

always_loaded_patterns: list[str] = Field(default=['tasks.*', 'tool_search', 'tool_get', 'finish'], description='Name patterns for tools that are always loaded regardless of loading mode.')

activation_scope class-attribute instance-attribute

activation_scope: Literal['run', 'session'] = Field(default='run', description='Scope at which a discovered tool remains activated: per-run or per-session.')

preferred_namespaces class-attribute instance-attribute

preferred_namespaces: list[str] = Field(default=[], description='Namespaces prioritized when ranking tool search results.')

fts_fallback_to_regex class-attribute instance-attribute

fts_fallback_to_regex: bool = Field(default=True, description='Whether to fall back to regex search when full-text search finds no matches.')

enable_incremental_index class-attribute instance-attribute

enable_incremental_index: bool = Field(default=True, description='Whether to update the search index incrementally instead of rebuilding fully.')

rebuild_cache_on_init class-attribute instance-attribute

rebuild_cache_on_init: bool = Field(default=False, description='Whether to force a full rebuild of the cached index on planner initialization.')

max_search_results class-attribute instance-attribute

max_search_results: int = Field(default=10, description='Maximum number of results returned per search.')

hints class-attribute instance-attribute

hints: ToolHintsConfig = Field(default_factory=ToolHintsConfig, description='Configuration for optional tool-search hints in the prompt.')

directory class-attribute instance-attribute

directory: ToolDirectoryConfig = Field(default_factory=ToolDirectoryConfig, description='Configuration for the optional tool directory summary in the prompt.')

ToolExamplesConfig

Bases: BaseModel

Configuration for including tool usage examples in the planner prompt.

enabled class-attribute instance-attribute

enabled: bool = Field(default=True, description='Whether to include tool usage examples in the prompt.')

max_examples_per_tool class-attribute instance-attribute

max_examples_per_tool: int = Field(default=3, ge=1, le=10, description='Maximum number of examples shown per tool.')

include_descriptions class-attribute instance-attribute

include_descriptions: bool = Field(default=True, description='Whether to include a short description alongside each example.')

ToolDirectoryConfig

Bases: BaseModel

Configuration for the tool directory summary shown to the planner LLM.

enabled class-attribute instance-attribute

enabled: bool = Field(default=False, description='Whether to include the tool directory in the prompt.')

max_groups class-attribute instance-attribute

max_groups: int = Field(default=20, ge=1, le=100, description='Maximum number of groups to list.')

max_tools_per_group class-attribute instance-attribute

max_tools_per_group: int = Field(default=6, ge=0, le=50, description='Maximum number of tools listed per group.')

include_tool_counts class-attribute instance-attribute

include_tool_counts: bool = Field(default=True, description='Whether to show the number of tools in each group.')

include_default_groups class-attribute instance-attribute

include_default_groups: bool = Field(default=True, description='Whether to auto-generate default groups in addition to configured ones.')

groups class-attribute instance-attribute

groups: list[ToolGroupConfig] = Field(default_factory=list, description='Explicitly configured tool groups.')

ToolGroupConfig

Bases: BaseModel

Definition of a named tool group used to organize the tool directory.

name class-attribute instance-attribute

name: str = Field(description='Unique identifier for the group.')

title class-attribute instance-attribute

title: str | None = Field(default=None, description='Human-readable title shown in the directory.')

trigger class-attribute instance-attribute

trigger: str | None = Field(default=None, description='Description of when to use tools in this group.')

task_type class-attribute instance-attribute

task_type: Literal['browser', 'api', 'code', 'domain', 'unknown'] | None = Field(default=None, description="Category of task this group's tools address.")

match_namespaces class-attribute instance-attribute

match_namespaces: list[str] = Field(default_factory=list, description='Tool namespaces whose members belong to this group.')

match_tags class-attribute instance-attribute

match_tags: list[str] = Field(default_factory=list, description='Tags used to match tools into this group.')

match_name_patterns class-attribute instance-attribute

match_name_patterns: list[str] = Field(default_factory=list, description='Glob/regex-like name patterns used to match tools into this group.')

tool_names class-attribute instance-attribute

tool_names: list[str] = Field(default_factory=list, description='Explicit tool names assigned to this group.')

ToolHintsConfig

Bases: BaseModel

Configuration for surfacing tool-search hints in the planner prompt.

enabled class-attribute instance-attribute

enabled: bool = Field(default=False, description='Whether to include tool-search hints in the prompt.')

top_k class-attribute instance-attribute

top_k: int = Field(default=5, ge=1, le=20, description='Maximum number of hinted tools to include.')

include_always_loaded class-attribute instance-attribute

include_always_loaded: bool = Field(default=False, description='Whether always-loaded tools are also included among the hints.')

search_type class-attribute instance-attribute

search_type: Literal['fts', 'regex', 'exact'] = Field(default='fts', description='Search strategy used to generate hints: full-text, regex, or exact match.')

BackgroundTasksConfig

Bases: BaseModel

Configuration for background tasks/subagent orchestration.

This is the single source of truth for background task settings, consumed by: - ReactPlanner (prompt guidance, tool validation) - SessionManager/TaskService (runtime enforcement) - Spec generation engine (agent.yaml generation) - Template engine (scaffolding new agents)

Downstream teams configure these values in their agent's Config class, which builds this model before passing to ReactPlanner.

enabled class-attribute instance-attribute

enabled: bool = False

Master switch for background task capabilities.

include_prompt_guidance class-attribute instance-attribute

include_prompt_guidance: bool = True

Whether to inject background task guidance into the system prompt.

allow_tool_background class-attribute instance-attribute

allow_tool_background: bool = False

Whether tools marked with background=True can spawn async tasks.

default_mode class-attribute instance-attribute

default_mode: str = 'subagent'

Default execution mode: 'subagent' (full reasoning) or 'job' (single tool).

default_merge_strategy class-attribute instance-attribute

default_merge_strategy: str = 'HUMAN_GATED'

How task results merge into context: HUMAN_GATED, APPEND, or REPLACE.

context_depth class-attribute instance-attribute

context_depth: str = 'full'

Context snapshot depth for spawned tasks: 'full', 'summary', or 'minimal'.

propagate_on_cancel class-attribute instance-attribute

propagate_on_cancel: str = 'cascade'

Cancel propagation: 'cascade' (cancel children), 'orphan' (leave running).

spawn_requires_confirmation class-attribute instance-attribute

spawn_requires_confirmation: bool = False

Whether spawning a task requires explicit user confirmation.

max_concurrent_tasks class-attribute instance-attribute

max_concurrent_tasks: int = 5

Maximum number of tasks running concurrently per session.

max_tasks_per_session class-attribute instance-attribute

max_tasks_per_session: int = 50

Maximum total tasks (active + completed) per session.

task_timeout_s class-attribute instance-attribute

task_timeout_s: int = 3600

Task timeout in seconds (default: 1 hour).

max_pending_steering class-attribute instance-attribute

max_pending_steering: int = 2

Maximum steering messages queued per task before backpressure.

proactive_report_enabled class-attribute instance-attribute

proactive_report_enabled: bool = False

Master switch for proactive messages on auto-merge completion.

proactive_report_strategies class-attribute instance-attribute

proactive_report_strategies: list[str] = ['APPEND', 'REPLACE']

Merge strategies that trigger proactive reports (not HUMAN_GATED).

proactive_report_max_queued class-attribute instance-attribute

proactive_report_max_queued: int = 5

Maximum queued reports before dropping oldest.

proactive_report_timeout_s class-attribute instance-attribute

proactive_report_timeout_s: float = 30.0

Timeout for proactive message generation.

proactive_report_max_hops class-attribute instance-attribute

proactive_report_max_hops: int = 2

Maximum proactive recursion hops before disabling background spawning.

proactive_report_fallback_notification class-attribute instance-attribute

proactive_report_fallback_notification: bool = True

Fall back to notification panel if generation fails.

default_group_merge_strategy class-attribute instance-attribute

default_group_merge_strategy: str = 'APPEND'

Default merge strategy for task groups.

default_group_report class-attribute instance-attribute

default_group_report: str = 'all'

Default report strategy for groups: 'all', 'any', or 'none'.

group_timeout_s class-attribute instance-attribute

group_timeout_s: float = 600.0

Timeout for group completion (seal to complete).

group_partial_on_failure class-attribute instance-attribute

group_partial_on_failure: bool = True

If True, report partial results when some tasks in a group fail.

max_tasks_per_group class-attribute instance-attribute

max_tasks_per_group: int = 10

Maximum tasks allowed in a single group.

auto_seal_groups_on_foreground_yield class-attribute instance-attribute

auto_seal_groups_on_foreground_yield: bool = True

Auto-seal OPEN groups when foreground yields to user.

retain_turn_timeout_s class-attribute instance-attribute

retain_turn_timeout_s: float = 60.0

Max time foreground waits for retained tasks/groups before force-yield.

background_continuation_max_hops class-attribute instance-attribute

background_continuation_max_hops: int = 2

Maximum background continuation cycles after retain-timeout.

background_continuation_cooldown_s class-attribute instance-attribute

background_continuation_cooldown_s: float = 0.0

Delay between background continuation cycles.

BackgroundTaskHandle

Bases: BaseModel

Return type for tools that run asynchronously in the background.

task_id class-attribute instance-attribute

task_id: str = Field(description='Unique identifier of the spawned background task.')

status class-attribute instance-attribute

status: str = Field(default='PENDING', description='Current status of the task (e.g. PENDING, RUNNING).')

message class-attribute instance-attribute

message: str | None = Field(default=None, description='Optional human-readable status message.')

BackgroundTaskResult dataclass

BackgroundTaskResult(task_id: str, group_id: str | None = None, status: Literal['completed', 'failed'] = 'completed', summary: str | None = None, payload: Any = None, facts: dict[str, Any] = dict(), artifacts: list[dict[str, Any]] = list(), consumed: bool = False, completed_at: float = time.time())

Result from a completed background task, stored in trajectory.

Attributes:

Name	Type	Description
task_id	str	Unique identifier of the background task this result belongs to.
group_id	str | None	Optional identifier grouping related background tasks together.
status	Literal['completed', 'failed']	Terminal status of the task, either "completed" or "failed".
summary	str | None	Optional human-readable summary of the task's outcome.
payload	Any	Arbitrary result payload produced by the task.
facts	dict[str, Any]	Structured facts extracted from the task's execution.
artifacts	list[dict[str, Any]]	List of artifact records (as dicts) produced by the task.
consumed	bool	Whether this result has already been consumed by the planner.
completed_at	float	Unix timestamp (seconds) when the task completed.

task_id instance-attribute

task_id: str

group_id class-attribute instance-attribute

group_id: str | None = None

status class-attribute instance-attribute

status: Literal['completed', 'failed'] = 'completed'

summary class-attribute instance-attribute

summary: str | None = None

payload class-attribute instance-attribute

payload: Any = None

facts class-attribute instance-attribute

facts: dict[str, Any] = field(default_factory=dict)

artifacts class-attribute instance-attribute

artifacts: list[dict[str, Any]] = field(default_factory=list)

consumed class-attribute instance-attribute

consumed: bool = False

completed_at class-attribute instance-attribute

completed_at: float = field(default_factory=time.time)

from_payload classmethod

from_payload(payload: Mapping[str, Any]) -> BackgroundTaskResult | None

Build a :class:BackgroundTaskResult from a raw mapping payload.

Parameters:

Name	Type	Description	Default
payload	Mapping[str, Any]	Mapping containing at least a task_id key, plus optional status, summary, digest, facts, artifacts, group_id, payload, consumed, and completed_at keys.	required

Returns:

Type	Description
BackgroundTaskResult | None	BackgroundTaskResult | None: The constructed result, or None if
BackgroundTaskResult | None	payload does not contain a valid non-empty string task_id.

to_payload

to_payload() -> dict[str, Any]

Serialise this result to a JSON-safe dict.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: A dict with task_id, group_id, status,
dict[str, Any]	summary, payload (passed through JSON-safe coercion), facts,
dict[str, Any]	artifacts, consumed, and completed_at keys.

DSPyLLMClient

DSPyLLMClient(llm: str | dict[str, Any], *, output_schema: type[BaseModel] | None = None, temperature: float = 0.0, max_retries: int = 3, timeout_s: float = 360.0, max_tokens: int = 4096)

LLM client using DSPy for structured outputs.

Deprecated

DSPyLLMClient is deprecated and unmaintained as of 3.11.0a6. It is no longer part of PenguiFlow's supported LLM surface and does not participate in built-in rate-limit fallback (llm_fallback). Use the native LLM layer (default) or transport="pydantic-ai" instead; this class will be removed in a future release.

This client implements the JSONLLMClient protocol and uses DSPy's signature system to generate structured outputs. DSPy handles the prompt engineering and parsing internally, providing more reliable structured outputs across different LLM providers.

Benefits over direct LiteLLM: - Better structured output reliability across providers - Automatic prompt optimization for structure extraction - Works with models that don't support native JSON schema mode - Graceful degradation with retry logic

Parameters:

Name	Type	Description	Default
llm	str | dict[str, Any]	Model identifier (e.g., "gpt-4o-mini", "databricks/databricks-gpt-oss-120b")	required
temperature	float	Sampling temperature (0.0 = deterministic)	0.0
max_retries	int	Number of retry attempts for transient failures	3
timeout_s	float	Timeout per LLM call in seconds	360.0

Example

client = DSPyLLMClient( ... llm="databricks/databricks-gpt-oss-120b", ... temperature=0.0, ... ) response = await client.complete( messages=[{"role": "user", "content": "..."}], response_format={"type": "json_schema", "json_schema": {...}}, )

expects_json_schema class-attribute instance-attribute

expects_json_schema = True

from_base_client classmethod

from_base_client(base_client: DSPyLLMClient, output_schema: type[BaseModel]) -> DSPyLLMClient

Create a new DSPy client with a different output schema.

This factory method allows creating separate clients for different tasks (e.g., reflection, summarization) while reusing the same LLM configuration.

Parameters:

Name	Type	Description	Default
base_client	DSPyLLMClient	Existing DSPy client to clone configuration from	required
output_schema	type[BaseModel]	New output schema for the cloned client	required

Returns:

Type	Description
DSPyLLMClient	New DSPyLLMClient instance with same config but different output schema

Example

planner_client = DSPyLLMClient(llm="gpt-4o") reflection_client = DSPyLLMClient.from_base_client( ... planner_client, ReflectionCritique ... )

complete async

complete(*, messages: Sequence[Mapping[str, Any]], response_format: Mapping[str, Any] | None = None, stream: bool = False, on_stream_chunk: Any = None, on_reasoning_chunk: Any = None) -> tuple[str, float]

Generate completion with structured output via DSPy.

Parameters:

Name	Type	Description	Default
messages	Sequence[Mapping[str, Any]]	OpenAI-style message list	required
response_format	Mapping[str, Any] | None	Optional JSON schema for structured output	None

Returns:

Type	Description
tuple[str, float]	Tuple of JSON string response and cost in USD (DSPy cost is 0.0)

Raises:

Type	Description
RuntimeError	If all retry attempts fail
TimeoutError	If the call exceeds timeout_s

Trajectory dataclass

Trajectory(query: str, llm_context: Mapping[str, Any] | None = None, tool_context: dict[str, Any] | None = None, input_parts: tuple[ContentPart, ...] = (), artifacts: dict[str, Any] = dict(), sources: list[Mapping[str, Any]] = list(), metadata: dict[str, Any] = dict(), steps: list[TrajectoryStep] = list(), summary: TrajectorySummary | None = None, hint_state: dict[str, Any] = dict(), resume_user_input: str | None = None, steering_inputs: list[str] = list(), background_results: dict[str, BackgroundTaskResult] = dict())

Mutable record of a planner run: inputs, executed steps, and derived state.

Tracks the original query and context alongside every :class:TrajectoryStep executed so far, plus auxiliary state (artifacts, sources, background task results, hints) needed to resume or summarise the run.

Attributes:

Name	Type	Description
query	str	The original user query that started this trajectory.
llm_context	Mapping[str, Any] | None	Optional read-only context mapping shared with the LLM.
tool_context	dict[str, Any] | None	Optional mutable context mapping available to tools.
input_parts	tuple[ContentPart, ...]	Tuple of non-text content parts (images/audio) supplied with the query.
artifacts	dict[str, Any]	Named artifacts accumulated over the run.
sources	list[Mapping[str, Any]]	List of source records (as mappings) collected during the run.
metadata	dict[str, Any]	Arbitrary metadata associated with the run.
steps	list[TrajectoryStep]	Ordered list of executed :class:TrajectoryStep instances.
summary	TrajectorySummary | None	Optional compact :class:TrajectorySummary of progress so far.
hint_state	dict[str, Any]	Mutable state used to track planner hints across steps.
resume_user_input	str | None	Optional user input supplied when resuming a paused run.
steering_inputs	list[str]	List of user steering messages injected during the run.
background_results	dict[str, BackgroundTaskResult]	Mapping of task id to :class:BackgroundTaskResult for completed background tasks.

query instance-attribute

query: str

llm_context class-attribute instance-attribute

llm_context: Mapping[str, Any] | None = None

tool_context class-attribute instance-attribute

tool_context: dict[str, Any] | None = None

input_parts class-attribute instance-attribute

input_parts: tuple[ContentPart, ...] = ()

artifacts class-attribute instance-attribute

artifacts: dict[str, Any] = field(default_factory=dict)

sources class-attribute instance-attribute

sources: list[Mapping[str, Any]] = field(default_factory=list)

metadata class-attribute instance-attribute

metadata: dict[str, Any] = field(default_factory=dict)

steps class-attribute instance-attribute

steps: list[TrajectoryStep] = field(default_factory=list)

summary class-attribute instance-attribute

summary: TrajectorySummary | None = None

hint_state class-attribute instance-attribute

hint_state: dict[str, Any] = field(default_factory=dict)

resume_user_input class-attribute instance-attribute

resume_user_input: str | None = None

steering_inputs class-attribute instance-attribute

steering_inputs: list[str] = field(default_factory=list)

background_results class-attribute instance-attribute

background_results: dict[str, BackgroundTaskResult] = field(default_factory=dict)

to_history

to_history() -> list[dict[str, Any]]

Return the dumped history of all steps.

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: One serialised dict (via :meth:TrajectoryStep.dump)
list[dict[str, Any]]	per step, in execution order.

add_background_result

add_background_result(result: BackgroundTaskResult) -> None

Record a completed background task result on this trajectory.

Parameters:

Name	Type	Description	Default
result	BackgroundTaskResult	The background task result to store, keyed by its task_id.	required

mark_background_consumed

mark_background_consumed(task_id: str) -> bool

Mark a stored background task result as consumed.

Parameters:

Name	Type	Description	Default
task_id	str	Identifier of the background task result to mark.	required

Returns:

Name	Type	Description
bool	bool	True if a result with task_id was found and marked consumed,
	bool	False if no such result exists.

clear_consumed_background

clear_consumed_background() -> int

Remove all background task results already marked as consumed.

Returns:

Name	Type	Description
int	int	The number of background task results that were removed.

get_unconsumed_background

get_unconsumed_background() -> dict[str, BackgroundTaskResult]

Return background task results that have not yet been consumed.

Returns:

Type	Description
dict[str, BackgroundTaskResult]	dict[str, BackgroundTaskResult]: Mapping of task id to result, for
dict[str, BackgroundTaskResult]	entries whose consumed flag is False.

serialise

serialise() -> dict[str, Any]

Serialise the full trajectory to a plain, JSON-safe dict.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: A dict capturing query, llm_context,
dict[str, Any]	tool_context, input_parts (stubbed for binary content),
dict[str, Any]	artifacts, sources, metadata, steps (via
dict[str, Any]	meth:to_history), summary, hint_state, resume_user_input,
dict[str, Any]	steering_inputs, and background_results.

from_serialised classmethod

from_serialised(payload: Mapping[str, Any]) -> Trajectory

Reconstruct a :class:Trajectory from a previously serialised payload.

Parameters:

Name	Type	Description	Default
payload	Mapping[str, Any]	Mapping produced by :meth:serialise (or a compatible legacy shape), containing at minimum a query key.	required

Returns:

Name	Type	Description
Trajectory	Trajectory	A new trajectory populated from payload, including
	Trajectory	restored steps, summary, background results, and auxiliary state.

compress

compress() -> TrajectorySummary

Compute and store a compact :class:TrajectorySummary for this trajectory.

Derives goals, gathered facts, pending retry actions, and a truncated digest of the last observation from the recorded steps, using simple rule-based heuristics (no LLM call).

Returns:

Name	Type	Description
TrajectorySummary	TrajectorySummary	The computed summary, which is also assigned to
	TrajectorySummary	self.summary.

TrajectoryStep dataclass

TrajectoryStep(action: PlannerAction, observation: Any | None = None, llm_observation: Any | None = None, error: str | None = None, failure: Mapping[str, Any] | None = None, streams: Mapping[str, Sequence[Mapping[str, Any]]] | None = None)

A single recorded step (action + observation) within a :class:Trajectory.

Attributes:

Name	Type	Description
action	PlannerAction	The planner action that was executed for this step.
observation	Any | None	Raw observation/result returned by executing action, if any.
llm_observation	Any | None	Optional pre-serialised observation tailored for LLM consumption, used in place of observation when present.
error	str | None	Optional error message if the step failed.
failure	Mapping[str, Any] | None	Optional mapping with structured failure details.
streams	Mapping[str, Sequence[Mapping[str, Any]]] | None	Optional mapping of stream id to the sequence of chunks emitted on that stream during this step.

action instance-attribute

action: PlannerAction

observation class-attribute instance-attribute

observation: Any | None = None

llm_observation class-attribute instance-attribute

llm_observation: Any | None = None

error class-attribute instance-attribute

error: str | None = None

failure class-attribute instance-attribute

failure: Mapping[str, Any] | None = None

streams class-attribute instance-attribute

streams: Mapping[str, Sequence[Mapping[str, Any]]] | None = None

dump

dump() -> dict[str, Any]

Serialise this step to a plain, JSON-safe dict.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: A dict with action (legacy-format action dump),
dict[str, Any]	observation, error, and failure keys, plus llm_observation
dict[str, Any]	and streams when set.

serialise_for_llm

serialise_for_llm() -> Any

Return the observation representation to send to the LLM.

Prefers llm_observation when set, otherwise falls back to the serialised observation.

Returns:

Name	Type	Description
Any	Any	The LLM-facing observation value for this step.

TrajectorySummary

Bases: BaseModel

Compact, LLM-facing summary of a trajectory's progress.

Produced by :meth:Trajectory.compress (or restored via model_validate) to give callers/LLMs a bounded, JSON-serialisable snapshot of what a run has accomplished so far.

Attributes:

Name	Type	Description
goals	list[str]	List of goal statements the trajectory is working toward.
facts	dict[str, Any]	Arbitrary key/value facts gathered while executing the trajectory.
pending	list[str]	List of pending follow-up actions (e.g., retries) still outstanding.
last_output_digest	str | None	Truncated string digest of the most recent observation, if any.
note	str | None	Optional free-form annotation describing how the summary was produced.

goals class-attribute instance-attribute

goals: list[str] = []

facts class-attribute instance-attribute

facts: dict[str, Any] = {}

pending class-attribute instance-attribute

pending: list[str] = []

last_output_digest class-attribute instance-attribute

last_output_digest: str | None = None

note class-attribute instance-attribute

note: str | None = None

compact

compact() -> dict[str, Any]

Return a compact JSON-serialisable dict view of this summary.

Omits the note field entirely when it is falsy, keeping the payload minimal for contexts (e.g. LLM prompts) where every token counts.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: A dict with goals, facts, pending, and
dict[str, Any]	last_output_digest keys, plus note when set.