Nodes

Abstract base classes

Abstract node classes for LangGraph processing nodes

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[TSchema] | None = None)

Bases: AbstractLangGraphNode[TSchema, dict[str, Any]], Generic

Abstract base class for LangGraph nodes that use LLMs.

Subclasses must set model_type 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: BaseModel) → dict[str, Any]

Template method defining standard execution flow

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_type: 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 TReturn to support both state-updating nodes (Dict[str, Any]) and other nodes, e.g., router nodes (str) and tool caller nodes.

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

abstractmethod async execute(state: TSchema) → TReturn

Execute this node and return the result.

Parameters:

state – Current graph state

Returns:

State updates (dict) or routing label (str)

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[TSchema, str], 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: TSchema) → 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 = '')

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['info', 'debug', 'state', 'usage'], 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

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[TSchema] | 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.

property output_schema: type[TSchema] | 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.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_OUTPUT_TOKENS_CEILING = 32768

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_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.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

Guard / safety nodes

Guard node for safety checking

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

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_type: 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]]

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

model_defaults: ClassVar[dict[str, Any]] = {'max_output_tokens': 2048, '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_type: 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().

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.

File: klea_utils/nodes/tools_picker.py

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

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_type: str = 'chat', prompt_prefix: str = 'ToolsPicker', prompt_registry_location: str | Path | None = None)

Bases: BaseLLMNode[BaseModel]

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 file, 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 (both apps name their picker prompt ToolsPicker_system.md).

• model_type selects the llm_models role ("plan" for the agent, "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.

model_defaults: ClassVar[dict[str, Any]] = {'max_output_tokens': 2048, '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_type: 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.

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, tools_meta: dict[str, dict[str, Any]] | None = None, project_root: str | None = None, post_dispatch: Callable[[Any, list[fastmcp.client.client.CallToolResult]], dict[str, Any]] | None = None)

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

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 and the results and returns additional state updates.

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

Gate and dispatch the tool calls in state.tool_calls.

Parameters:

state – Current graph state (must carry tool_calls).

Returns:

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

Memory

Summarise conversation history node

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)

Bases: BaseLLMNode

model_defaults: ClassVar[dict[str, Any]] = {'max_output_tokens': 4096, '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 only summarises history up to that window – the summary and the verbatim window never overlap. Does NOT append the summary to messages – it’s metadata, not a turn.

Expects state to have the following fields:

• messages: list of messages

• summarised_till: index of messages that have been summarised already

• context_summary: previous memory/context summary

model_type: 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().