Nodes

Abstract base classes

Abstract node classes for LangGraph processing nodes

See devdocs/system/streams.md (NodeStreamData/_get_inspect) and devdocs/system/prompt-conventions.md (node prompt fields); ADRs 0019, 0040 and 0045.

File: klea_utils/nodes/abstract.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

class klea_utils.nodes.abstract.AbstractLLMNode(logger: logging.Logger, label: str, llm_models: dict[str, Any], output_schema: type[TOutput] | None = None)

Bases: AbstractLangGraphNode[TState, dict[str, Any], LLMNodeContext], Generic

Abstract base class for LangGraph nodes that use LLMs.

Generic over TState (the graph state) and TOutput (the structured output schema; use BaseModel for nodes without one).

Subclasses must set model_role to a key present in the llm_models dict (e.g. "chat", "plan", "guard").

Implements a template execution flow: 1. Pre-execution check (optional skip) 2. Build prompt (system + human) 3. Invoke LLM 4. Process output (structured or raw) 5. Update state

final async execute(state: TState) → dict[str, Any]

Template method defining standard execution flow.

All intermediate values live on the per-run ctx (not on the shared node instance), so concurrent runs cannot clobber each other. The streaming hooks read them from the same ctx.

max_validation_retries: ClassVar[int] = 0

How many times execute re-invokes on a validation failure.

A node overrides _validate_result() to reject structurally invalid output; when it returns an error, execute re-invokes the model with the error exposed as the validation_feedback prompt variable, up to this many retries. Default 0 disables the loop.

model_defaults: ClassVar[dict[str, Any]] = {}

Node-level model configuration defaults.

These are frozen — user context overrides cannot change them. Set this as a class attribute on each subclass to pin model params (temperature, model, num_predict, etc.) that should never be overridden at runtime.

Subclasses that need dynamic initialisation may also set self.model_defaults in __init__.

model_role: str = ''

Key into llm_models dict (e.g. "chat", "plan", "guard").

Determines which LLMModel entry from the graph’s llm_models this node uses. Must match a key set up by the orchestrator in _setup_models().

class klea_utils.nodes.abstract.AbstractLangGraphNode(logger: Logger, label: str)

Bases: ABC, Generic

Abstract base class for all LangGraph nodes.

Generic over TState (the graph state the node operates on) and TReturn (state-updating nodes return dict[str, Any]; router nodes return str).

Provides a consistent interface: all nodes have a logger and an execute(state) method.

abstractmethod async execute(state: TState) → TReturn

Execute this node and return the result.

Parameters:

state – Current graph state

Returns:

State updates (dict) or routing label (str)

stream_tokens: ClassVar[bool] = False

Whether this node’s LLM token deltas are forwarded as token stream events. Off by default: structured-output nodes emit JSON fragments, which are useless to a live-streaming UI. A node that produces user-facing free text sets this True, and the orchestrator adds its label to BaseLangGraph.token_stream_nodes.

write_custom_stream(event: dict) → None

Emit a custom event to the LangGraph v3 stream.

Writes to the custom channel via get_stream_writer(). Requires a StreamTransformer with required_stream_modes = ("custom",) registered so the channel is enabled (done by BaseLangGraph.run_graph_astream_events()).

Call this at the top of execute() to emit progress, or anywhere to emit debug or intermediate data for UI consumers.

Parameters:

event – Dict to emit as a custom protocol event

class klea_utils.nodes.abstract.AbstractRouterNode(logger: Logger, label: str)

Bases: AbstractLangGraphNode[TState, str, NodeContext], Generic

Abstract class for LangGraph router nodes.

Router nodes inspect the state and return a string label that determines which edge to follow next. Used with add_conditional_edges().

abstractmethod async execute(state: TState) → str

Return the routing label (edge name) based on state.

class klea_utils.nodes.abstract.NodeStreamData(*, heading: str = '', summary: str = '', details: dict[str, ~typing.Any]=<factory>, display: str = '', key: str = '', preformatted: bool = False)

Bases: BaseModel

Data payload for node streaming events.

This is the contract between nodes and the frontend.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class klea_utils.nodes.abstract.NodeStreamEvent(*, type: Literal['progress', 'inspect', 'state', 'usage', 'notice'], node: str, data: NodeStreamData)

Bases: BaseModel

Full streaming event emitted by nodes.

This is the contract between the graph infrastructure and the frontend.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Concrete base class

Base node classes for LangGraph processing nodes

See devdocs/system/structured-output-fallback.md and devdocs/system/prompt-conventions.md; ADRs 0017 and 0044.

File: klea_utils/nodes/base.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

class klea_utils.nodes.base.BaseLLMNode(logger: logging.Logger, label: str, llm_models: dict[str, Any], output_schema: type[TOutput] | None, memory: bool = False)

Bases: AbstractLLMNode, Generic

Base class for LangGraph nodes that load prompts from files.

Extends AbstractLLMNode with: - File-based prompt loading via load_prompt() - Optional memory support (appends memory content to system prompt) - Auto-derived prompt registry location from subclass file path

Prompt files are expected to be named {prefix}_system.md and {prefix}_user.md.

Subclasses can override prompt_prefix or prompt_registry_location via the setter if the defaults (lowercase class name / sibling prompts/) are not appropriate.

include_model_identity: ClassVar[bool] = False

Report the agent and effective model identity in the system prompt. Opt in on nodes that compose a user-facing reply, so the model can answer “which agent/model are you running” with the values that actually produced the reply. The constant agent line is prepended (cacheable prefix); the per-run model line is appended after the stable prompt sections. Off by default: most nodes never need it and it costs prompt tokens.

property output_schema: type[TOutput] | None

Return Pydantic schema for structured output if required

property output_schema_json: dict[str, Any]

Return JSON schema string for use in prompts.

property prompt_prefix: str

Return the prompt file prefix.

Falls back to the lowercase class name if not explicitly set.

property prompt_registry_location: Path

Return path to the prompts directory.

Falls back to a sibling prompts/ directory relative to the subclass file if not explicitly set.

klea_utils.nodes.base.DEFAULT_RECENT_MEMORY_CHARS = 10000

Character budget for the verbatim recent-conversation window injected into memory-enabled node prompts. Older history is represented by context_summary instead.

klea_utils.nodes.base.EMPTY_RESULT_FALLBACK = 'Something went wrong. Please retry.'

User-facing fallback when an LLM produced no usable answer and the node has no more specific fallback of its own.

klea_utils.nodes.base.KLEA_AGENT_NAME = 'Klea'

Product name reported to the model by user-facing nodes.

klea_utils.nodes.base.MAX_CONTEXT_OVERFLOW_RETRIES = 3

Max times to retry an invoke that overflowed the context window, each time shrinking the reserved output window to free headroom.

klea_utils.nodes.base.MAX_EMPTY_OUTPUT_RETRIES = 2

Max times to retry an invoke that returned an empty response. Some providers (notably HuggingFace) intermittently return a successful but blank message; the retry is a plain re-invoke with no window change.

klea_utils.nodes.base.MAX_OUTPUT_TOKENS_CEILING = 65536

Output-window ceiling for truncation retries. Reasoning-capable models can spend a large, unpredictable number of tokens thinking before producing a small final answer, so the grow ladder clamps at a generous ceiling (mirrors opencode’s OUTPUT_TOKEN_MAX). This is the fallback when the model’s context is unknown; for a custom endpoint that advertises a larger max_model_len, BaseLLMNode._jump_output_target() raises the ceiling to the endpoint’s remaining context. It is still clamped to the model’s catalog output limit / total budget by resolve_output_token_limit.

klea_utils.nodes.base.MAX_TRANSIENT_RETRIES = 2

Max times to retry a transiently failed invoke (a timeout, e.g. a stalled stream that never produced a chunk). Rate limits, auth and model-not-found failures are deliberately not retried here: the provider SDK already retries 429/5xx and honours Retry-After, so retrying again would multiply the attempts and can amplify a rate limit.

klea_utils.nodes.base.MAX_TRUNCATION_RETRIES = 15

Max times to retry an invoke whose output was truncated (finish_reason == "length"), each time growing the reserved output window. The budget covers the full climb from the smallest node window to the largest advertised context (262144) so heavy-thinking models are not cut short, while still bounding pathological repeated truncations.

klea_utils.nodes.base.MIN_OUTPUT_TOKENS = 64

Floor for the reserved output window when shrinking it on overflow.

klea_utils.nodes.base.TRANSIENT_RETRY_BACKOFF_S = 1.0

Base backoff (seconds) between transient retries; doubles each attempt.

klea_utils.nodes.base.TRUNCATION_LINEAR_CAP = 16384

Ceiling for the linear phase; above it the window grows by TRUNCATION_PHASE2_STEP.

klea_utils.nodes.base.TRUNCATION_LINEAR_STEP = 2048

the window grows by TRUNCATION_LINEAR_STEP while below TRUNCATION_LINEAR_CAP. Covers the common case – a model cut off a few hundred to a couple of thousand tokens short of finishing – with small, predictable reservations.

Type:

Linear-phase step for truncation retries

klea_utils.nodes.base.TRUNCATION_PHASE2_STEP = 32768

once the linear phase is exhausted the window grows by this fixed step up to the output ceiling. Fixed-size steps (not exponential) keep reservations predictable for inference engines that allocate resources against max_tokens (e.g. vLLM KV-cache), avoiding the spikes of a doubling ladder.

Type:

Large-step phase for truncation retries

Per-run node context

Per-run node context.

A node’s intermediate values (prompt, LLM instance, raw output, result, token usage, …) are local to a single execute invocation. They live in a NodeContext built and passed by execute, instead of on the shared node instance - so two concurrent runs cannot clobber each other’s state.

The context deliberately does not carry the graph state: that is already an explicit argument to the hooks that need it, and keeping it out avoids duplicating (and diverging from) the checkpointed state.

File: klea_utils/nodes/context.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

class klea_utils.nodes.context.LLMNodeContext(output_schema: type[BaseModel] | None = None, system_prompt: Any = None, human_prompt: str | None = None, template: ChatPromptTemplate | None = None, variables: dict[str, ~typing.Any]=<factory>, prompt: PromptValue | None = None, llm: Runnable | None = None, config: RunnableConfig | None = None, output: Any = None, result: TOutput | None = None, validation_feedback: str = '', state_updates: dict[str, ~typing.Any] | None=None, final_state: dict[str, ~typing.Any] | None=None, token_usage: TokenUsage | None = None)

Bases: NodeContext, Generic[TOutput]

Per-run scratch for an LLM node.

Mirrors the values the LLM template computes and the streaming/inspection hooks read: the resolved schema, the prompt pipeline, the LLM invocation and its processed output.

output: Any = None

Raw LLM output and its processed/validated result.

output_schema: type[BaseModel] | None = None

Output schema resolved for this run (static or dynamic).

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

State updates produced by _update_state and the final merged dict.

token_usage: TokenUsage | None = None

Token usage for this run.

validation_feedback: str = ''

Validation feedback from the most recent rejected result.

class klea_utils.nodes.context.NodeContext

Bases: object

Base per-run node scratch space (no graph state).

Subclasses add the fields a node family computes during execution. The object is created by execute for a single run and passed to that run’s hooks; it is never stored on the shared node instance.

class klea_utils.nodes.context.ToolCallerContext(tool_results: list[fastmcp.client.client.CallToolResult] = <factory>, display_flags: list[bool] = <factory>, allowed_session_dirs: list[str] = <factory>, allowed_session_files: list[str] = <factory>)

Bases: NodeContext

Per-run scratch for the tool-caller node.

allowed_session_dirs: list[str]

Directories the user approved for the session while resolving a permission interrupt (allow for session); persisted to state.allowed_dirs by the node (ADR-0007 update 2026-10-08).

allowed_session_files: list[str]

Sensitive files approved for the session (allow for session on a sensitive request); persisted to state.allowed_files.

display_flags: list[bool]

Per-result display flag, aligned with tool_results.

Guard / safety nodes

Guard node for safety checking

See devdocs/adr/0010-guard-node.md.

File: klea_utils/nodes/guard.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

class klea_utils.nodes.guard.GuardNode(logger: Logger, label: str, llm_models: dict[str, Any], memory: bool = False)

Bases: BaseLLMNode[BaseModel, BaseModel]

model_defaults: ClassVar[dict[str, Any]] = {'max_output_tokens': 2048, 'temperature': 0.3}

Safety guard node that checks if user queries are safe to process.

Evaluates whether a query contains potentially harmful content and returns a routing decision (“safe” or “unsafe”).

Note: to be used with llama-guard, which always returns safe/unsafe.

To skip, do not set a model.

model_role: str = 'guard'

Key into llm_models dict (e.g. "chat", "plan", "guard").

Determines which LLMModel entry from the graph’s llm_models this node uses. Must match a key set up by the orchestrator in _setup_models().

Guard router node for routing based on guard decision

File: klea_utils/nodes/guard_router.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

class klea_utils.nodes.guard_router.GuardRouterNode(logger: Logger, label: str)

Bases: AbstractRouterNode

Router node that routes based on guard decision.

Reads the guard_decision from state and returns routing label: - “safe” -> continue to next node - “unsafe” -> decline to respond node

async execute(state: BaseModel) → str

Route based on guard_decision in state.

Parameters:

state – Current graph state

Returns:

Routing label (“safe” or “unsafe”)

Answer / response nodes

Provide a fixed answer.

File: rag_pkg/klea_rag/nodes/fixed_answer.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

class klea_utils.nodes.fixed_answer.FixedAnswer(logger: Logger, label: str, state_attr: str, message: str)

Bases: AbstractLangGraphNode[BaseModel, dict[str, Any], NodeContext]

Provide a fixed answer

async execute(state: BaseModel) → dict[str, Any]

Return fixed message.

Answer general question node

File: klea_utils/nodes/answer_general.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

class klea_utils.nodes.answer_general.AnswerGeneral(logger: Logger, label: str, llm_models: dict[str, Any], memory: bool = False, num_history_chars: int = 10000, fallback_config: FallbackConfig | None = None)

Bases: BaseLLMNode[BaseModel, BaseModel]

include_model_identity: ClassVar[bool] = True

Report the model that produced the reply, so “which model are you running” is answerable.

model_defaults: ClassVar[dict[str, Any]] = {'temperature': 0.3}

Answer general (non-domain) questions using the LLM’s training data.

Provides a conversational, user-friendly response. Optionally appends conversation history for context and a fallback warning when configured.

model_role: str = 'chat'

Key into llm_models dict (e.g. "chat", "plan", "guard").

Determines which LLMModel entry from the graph’s llm_models this node uses. Must match a key set up by the orchestrator in _setup_models().

stream_tokens: ClassVar[bool] = True

its token deltas are meaningful to stream.

Type:

Free-text answer node

class klea_utils.nodes.answer_general.FallbackConfig(*, enabled: bool = False, warning: str = '')

Bases: BaseModel

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Tool nodes

Shared MCP tools picker node.

See devdocs/system/mcp-permissions.md and devdocs/system/streams.md; ADRs 0020, 0034 and 0044.

File: klea_utils/nodes/tools_picker.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

klea_utils.nodes.tools_picker.MAX_BATCH_CALLS = 16

Maximum tool calls bound in one picker invocation (ADR-0041 batch cap).

klea_utils.nodes.tools_picker.MAX_BATCH_STEPS = 8

Maximum plan steps bound in one picker invocation (ADR-0041 batch cap).

klea_utils.nodes.tools_picker.MAX_BIND_GROUP = 4

Maximum steps bound in one binder invocation when steps share a tool set.

class klea_utils.nodes.tools_picker.ToolsPicker(logger: Logger, label: str, llm_models: dict[str, Any], tools_info: dict[str, dict[str, ToolInfo]] | None = None, model_role: str = 'chat', prompt_prefix: str = 'ToolsPicker', prompt_registry_location: str | Path | None = None, on_unusable: Callable[[Any, str], dict[str, Any]] | None = None, bind_planned_steps: bool = False)

Bases: BaseLLMNode[BaseModel, ToolCallsSchema]

Node that selects MCP tools for the current step or query.

Shared by Klea Agent and Klea RAG. The two applications differ only in the prompt files, the model role, and which context fields exist in the state, so all of that is configuration:

• prompt_registry_location points at the application’s prompts/ directory. RAG uses ToolsPicker_*; the agent’s ToolCallBinder uses ToolCallBinder_*.

• model_role selects the llm_models role ("tool_picker" for the agent binder, "chat" for RAG).

• tools_info is the per-domain BaseLangGraph.tools_info; when the state carries query_domains the descriptions are filtered to those domains (RAG), otherwise all tools are offered (agent).

_get_prompt_variables returns a superset of variables; each prompt file uses only the ones it declares (ChatPromptTemplate ignores the rest), so one class serves both prompts.

With bind_planned_steps=True (the agent’s ToolCallBinder) the node binds the Planner’s chosen tools instead of choosing tools: it processes one contiguous group of plan steps that share the same suggested-tool set, exposes only those tools, and accumulates calls in binding_calls until the batch is fully bound. RAG keeps the default picker behavior.

feedback_key: ClassVar[str] = 'picker_feedback'

Prompt variable and heading for the conditional feedback section.

include_model_identity: ClassVar[bool] = True

Report the agent and effective model identity in the system prompt. Opt in on nodes that compose a user-facing reply, so the model can answer “which agent/model are you running” with the values that actually produced the reply. The constant agent line is prepended (cacheable prefix); the per-run model line is appended after the stable prompt sections. Off by default: most nodes never need it and it costs prompt tokens.

max_validation_retries: ClassVar[int] = 2

Binder mode retries an off-contract selection before failing closed. Picker mode does not validate selections, so this is inert for RAG.

model_defaults: ClassVar[dict[str, Any]] = {'temperature': 0.01}

Node-level model configuration defaults.

These are frozen — user context overrides cannot change them. Set this as a class attribute on each subclass to pin model params (temperature, model, num_predict, etc.) that should never be overridden at runtime.

Subclasses that need dynamic initialisation may also set self.model_defaults in __init__.

model_role: str = 'chat'

Key into llm_models dict (e.g. "chat", "plan", "guard").

Determines which LLMModel entry from the graph’s llm_models this node uses. Must match a key set up by the orchestrator in _setup_models().

Shared MCP tools caller node.

See devdocs/system/streams.md (chat-renderable tool events) and devdocs/system/mcp-permissions.md; ADRs 0020, 0034 and 0040.

File: klea_utils/nodes/tools_caller.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

class klea_utils.nodes.tools_caller.ToolsCallerNode(logger: Logger, label: str, mcp_client: Any | None, tool_infos: dict[str, ToolInfo] | None = None, post_dispatch: Callable[[Any, list[fastmcp.client.client.CallToolResult], list[bool]], dict[str, Any]] | None = None, permission_policy: Literal['deny', 'ask'] = 'deny', mcp_session_pool: Any | None = None)

Bases: AbstractLangGraphNode[BaseModel, dict[str, Any], ToolCallerContext]

Node that gates and dispatches the selected MCP tool calls.

Shared by Klea Agent and Klea RAG. Reads state.tool_calls (a list of ToolCallSchema), gates each call client-side through klea_utils.mcp.dispatch.dispatch_tool_calls() (permission layer), emits info/debug stream events, and writes state.tool_results.

Applications that need extra post-dispatch state updates (e.g. the agent’s per-plan-step status) pass a post_dispatch callback that receives the state, the results, and a per-result “displayed” flag, and returns additional state updates.

async execute(state: BaseModel) → dict[str, Any]

Gate and dispatch the tool calls in state.tool_calls.

Skipped entirely – no dispatch, no streaming, no state update – when there is nothing to dispatch (no tool calls, or no MCP client). An empty round is routed around upstream (the agent’s picker router sends it to replan; RAG resets tool_results at graph entry), so a dispatched round always overwrites tool_results and no stale batch survives.

Parameters:

state – Current graph state (must carry tool_calls).

Returns:

{"tool_results": [...]} plus any callback extras.

Memory

Summarise conversation history node

See devdocs/adr/0018-message-memory.md.

File: klea_utils/nodes/summarise_memory.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

class klea_utils.nodes.summarise_memory.SummariseMemoryNode(logger: Logger, label: str, llm_models: dict[str, Any], summarisation_threshold_chars: int = 10000, num_history_chars: int = 10000, memory: bool = False, announce: str = '')

Bases: BaseLLMNode[BaseModel, BaseModel]

model_defaults: ClassVar[dict[str, Any]] = {'temperature': 0.3}

Node that summarises conversation history into a context summary.

Uses _pre_exec() to skip execution if there isn’t enough old conversation to summarise. The most recent messages (within num_history_chars) form the verbatim window that the prompt assembly injects as real messages, so this node summarises user-visible history up to that window. Internal run history (messages_internal) is not injected verbatim into prompts; it is summarised from its own cursor to the end. Does NOT append the summary to messages – it is metadata, not a turn.

Expects state to have the following fields:

• messages: user-visible conversation messages

• messages_internal: internal plan/evaluation/reasoning run history

• summarised_till / summarised_internal_till: per-stream cursors

• context_summary: previous memory/context summary

model_role: str = 'chat'

Key into llm_models dict (e.g. "chat", "plan", "guard").

Determines which LLMModel entry from the graph’s llm_models this node uses. Must match a key set up by the orchestrator in _setup_models().

Command

Command node for the session-command framework (ADR-0047).

Commands are user-invoked tool calls; this node is the user-facing counterpart of the tools picker/caller. It parses a leading / command from state.query, dispatches it to an app-provided graph handler, emits an inspect event for provenance, and ends the run with the handler’s message. A handler may set a route to hand off to another node first (the app’s conditional edge on command_route maps it; "" ends the run). A command that is not persists: message is kept out of the LLM’s message history (the graph-state change still persists in the checkpoint).

See devdocs/system/session-commands.md; ADR-0047.

File: klea_utils/nodes/command.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

class klea_utils.nodes.command.CommandNode(logger: Logger, label: str, *, registry: CommandRegistry, handlers: Mapping[str, Callable[[BaseModel, ParsedCommand], GraphCommandResult | Awaitable[GraphCommandResult]]])

Bases: AbstractLangGraphNode[BaseModel, dict[str, Any], NodeContext]

Execute a graph (server-side) command, optionally handing off, then end.

Constructed with the app’s CommandRegistry (metadata) and its graph handler map. A command with no handler, or one marked implemented=False, replies that it is not implemented yet. A handler may be synchronous or async (the node awaits an awaitable result), so a workflow command can await an LLM/tool call, and may set a route to hand off to another node before the run ends.

async execute(state: BaseModel) → dict[str, Any]

Dispatch the command in state.query and end the run.

An executable command runs (its inspect event records the effect). An unknown, client-side, not-yet-implemented, or malformed command is rejected with a direct user message; the inspect event is reserved for a command’s actual effect, not for rejections.