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)[source]

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][source]

Template method defining standard execution flow

model_defaults: 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)[source]

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[source]

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[source]

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)[source]

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[source]

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 = '')[source]

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)[source]

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, num_history_messages: int = 10)[source]

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_TRUNCATION_RETRIES = 2

Max times to retry an invoke whose output was truncated (finish_reason == "length"), each time growing the reserved output window.

klea_utils.nodes.base.MIN_OUTPUT_TOKENS = 64

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

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)[source]

Bases: BaseLLMNode

model_defaults: dict[str, Any] = {'max_output_tokens': 1024, '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)[source]

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[source]

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)[source]

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

Provide a fixed answer

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

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_messages: int = 10, fallback_config: FallbackConfig | None = None)[source]

Bases: BaseLLMNode

model_defaults: 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 = '')[source]

Bases: BaseModel

model_config = {}

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

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: int = 10, memory: bool = False)[source]

Bases: BaseLLMNode

model_defaults: 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 aren’t enough recent messages. 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().