#!/usr/bin/env python3
"""
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>
"""
from __future__ import annotations
import logging
from abc import ABC, abstractmethod
from typing import Any, ClassVar, Literal, final
from langchain.messages import AIMessage
from langchain_core.prompt_values import PromptValue
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable, RunnableConfig
from pydantic import BaseModel, Field
from klea_utils.graph.schemas import TokenUsage
from klea_utils.llm import extract_llm_output_content
from klea_utils.nodes.context import LLMNodeContext, NodeContext
[docs]
class NodeStreamData(BaseModel):
"""Data payload for node streaming events.
This is the contract between nodes and the frontend.
"""
heading: str = Field(
default="",
description=(
"Display heading for the event: the line shown while the node "
"runs for a ``progress`` event, or the section heading for an "
"``inspect`` one."
),
)
summary: str = Field(
default="",
description=(
"Human-readable summary, always rendered by the frontend for an "
"``inspect`` event (left blank for ``progress``)."
),
)
details: dict[str, Any] = Field(
default_factory=dict,
description="Structured data, rendered as collapsible JSON",
)
display: str = Field(
default="",
description="Pre-formatted markdown content for the status pane",
)
key: str = Field(
default="",
description=(
"Status-pane section key. Empty (default) keys the section by the "
"emitting node's label; a shared non-empty key lets several nodes "
"update one section in place (e.g. a live plan)."
),
)
preformatted: bool = Field(
default=False,
description=(
"Render ``display`` as preformatted monospace text (``<pre>``) "
"instead of markdown. Use for aligned, literal content such as a "
"plan with tool names, where markdown would mangle ``_``/``*``."
),
)
[docs]
class NodeStreamEvent(BaseModel):
"""Full streaming event emitted by nodes.
This is the contract between the graph infrastructure and the frontend.
"""
type: Literal["progress", "inspect", "state", "usage", "notice"] = Field(
description="Event type"
)
node: str = Field(description="Node label")
data: NodeStreamData = Field(description="Event payload")
[docs]
class AbstractLangGraphNode[
TState: BaseModel,
TReturn,
TCtx: NodeContext,
](ABC):
"""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.
"""
#: 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``.
stream_tokens: ClassVar[bool] = False
def __init__(self, logger: logging.Logger, label: str):
"""Initialise
Creates a new hierarchical logger.
:param logger: Parent logger instance (used to derive child logger name)
:param label: Human-readable label for this node, used as the
LangGraph node name for UI progress display
"""
# Child logger -- inherits the parent's dual-stream + file handlers
# (set up by BaseLangGraph via plogging.setup_root_logger on the
# root logger) through propagation, so this class does NOT configure
# its own handlers.
self.logger = logging.getLogger(f"{logger.name}.{self.__class__.__name__}")
self.label = label
[docs]
def write_custom_stream(self, 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.
:param event: Dict to emit as a custom protocol event
"""
from langgraph.config import get_stream_writer
get_stream_writer()(event)
[docs]
@abstractmethod
async def execute(self, state: TState) -> TReturn:
"""Execute this node and return the result.
:param state: Current graph state
:returns: State updates (dict) or routing label (str)
"""
...
# ------------------------------------------------------------------
# Execution contract (shared by all nodes)
#
# These hooks form the contract available to every node. ``execute``
# is not a template here (LLM and tool-calling nodes differ too much),
# so each node decides when to call them:
#
# - ``_pre_exec(state)``: whether to run at all. Returning False skips
# execution (``execute`` returns an empty dict). Default: True.
# - ``_pre_exec_stream()``: emit a progress event before doing work.
# - ``_post_exec_stream()``: emit ``inspect``/``state`` events from
# ``_get_inspect``/``_get_status`` after doing work.
#
# ``AbstractLLMNode`` calls all three from its own ``execute`` template
# and additionally emits a token-usage event (LLM-specific).
# ------------------------------------------------------------------
def _pre_exec(self, state: TState, ctx: TCtx) -> bool:
"""Pre-execution check. Override to conditionally skip execution.
Return False to skip execution (``execute`` returns an empty dict).
Return True (default) to proceed with the standard flow.
:param state: Current graph state.
:param ctx: Per-run node context.
"""
return True
def _emit_progress(self, heading: str = "") -> None:
"""Emit a ``progress`` event for this node.
``node`` is always this node's label, so the runner keys timing and
dedup on a stable identity; ``heading`` is the line shown while the
node runs (it defaults to the label, and a retry passes a richer
string). Only ``heading`` is consumed for a ``progress`` event; the
other fields stay empty.
:param heading: Display text; defaults to the node label.
"""
event = NodeStreamEvent(
type="progress",
node=self.label,
data=NodeStreamData(heading=heading or self.label),
)
self.write_custom_stream(event.model_dump())
def _emit_notice(self, content: str, *, kind: str = "progress") -> None:
"""Emit a concise, user-visible ``notice`` for the chat transcript.
Notices are derived progress notes (an evaluator verdict, a reasoning
conclusion), not status snapshots and not part of graph memory. They
are persisted by the API layer as transcript rows. The full content is
emitted; the frontends collapse long notices for display, so the user
can expand them to read everything.
:param content: Note text; blank content emits nothing.
:param kind: Semantic kind for styling/routing (default ``progress``).
"""
text = str(content or "").strip()
if not text:
return
event = NodeStreamEvent(
type="notice",
node=self.label,
data=NodeStreamData(
heading=self.label,
summary=text,
details={"content": text, "kind": kind},
),
)
try:
self.write_custom_stream(event.model_dump())
except RuntimeError:
# Unit tests call ``_update_state`` outside a graph run, where no
# stream writer exists; a notice is optional UI metadata.
pass
def _pre_exec_stream(self, ctx: TCtx) -> None:
"""Emit streaming event before node execution.
Default: emits a ``progress`` event with the node label as heading.
Override to customise pre-execution streaming.
:param ctx: Per-run node context.
"""
self._emit_progress()
def _post_exec_stream(self, state: TState, ctx: TCtx) -> None:
"""Emit streaming events after node execution.
Default: emits ``inspect`` and ``state`` events from
``_get_inspect`` and ``_get_status`` if they return non-None values.
``AbstractLLMNode`` overrides this to also emit the token-usage
event, which is LLM-specific.
Override to customise post-execution streaming.
:param state: Current graph state.
:param ctx: Per-run node context.
"""
inspect = self._get_inspect(state, ctx)
if inspect:
event = NodeStreamEvent(type="inspect", node=self.label, data=inspect)
self.write_custom_stream(event.model_dump())
status = self._get_status(state, ctx)
if status:
event = NodeStreamEvent(type="state", node=self.label, data=status)
self.write_custom_stream(event.model_dump())
def _get_inspect(self, state: TState, ctx: TCtx) -> NodeStreamData | None:
"""Return structured inspection data for an ``inspect`` event.
Override in subclasses to provide node-specific inspection data: a
short summary plus a ``details`` dict (rendered as collapsible JSON
in the inspection pane). The summary is always shown; the details
are the drill-down. Per-run values come from *ctx*.
:param state: Current graph state.
:param ctx: Per-run node context.
:returns: NodeStreamData with summary and details, or None to skip
Example::
return NodeStreamData(
summary="Classified into: neuron, morphology",
details={"classified_domains": ["neuron", "morphology"]}
)
"""
return None
def _get_status(self, state: TState, ctx: TCtx) -> NodeStreamData | None:
"""Return status pane content for this node.
Override to populate the status pane with display-ready
markdown content. The ``display`` field of the returned
``NodeStreamData`` is rendered in the status pane; the frontend
replaces the previous entry for this node label so loops do not
accumulate.
:param state: Current graph state.
:param ctx: Per-run node context.
:returns: NodeStreamData with display content, or None to skip
"""
return None
[docs]
class AbstractLLMNode[TState: BaseModel, TOutput: BaseModel](
AbstractLangGraphNode[TState, dict[str, Any], LLMNodeContext[TOutput]]
):
"""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 :attr:`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
"""
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()``.
"""
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__``.
"""
max_validation_retries: ClassVar[int] = 0
"""How many times ``execute`` re-invokes on a validation failure.
A node overrides :meth:`_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.
"""
def __init__(
self,
logger: logging.Logger,
label: str,
llm_models: dict[str, Any],
output_schema: type[TOutput] | None = None,
):
"""Initialize with logger and model.
:param logger: Logger instance
:param label: Human-readable label for UI progress display
:param llm_models: ``{role: LLMModel}`` dict (from ``BaseLangGraph.llm_models``)
:param output_schema: Pydantic schema for structured output
"""
super().__init__(logger, label)
self.llm_models = llm_models
try:
self._llm_entry = self.llm_models[self.model_role]
except KeyError:
raise KeyError(
f"Node '{type(self).__name__}' has model_role='{self.model_role}', "
f"but llm_models only has keys: {list(self.llm_models)}"
) from None
self._output_schema = output_schema
def _get_output_schema(
self, state: TState, ctx: LLMNodeContext[TOutput]
) -> type[TOutput] | None:
"""Return the structured-output schema for this invocation.
Defaults to the node's static :attr:`output_schema`. Override to
build one per run - for example the tools picker, whose schema is a
per-tool discriminated union derived from the run's disclosed tools.
:param state: Current graph state.
:param ctx: Per-run node context.
:returns: The pydantic output schema, or ``None`` for a plain node.
"""
return self._output_schema
[docs]
@final
async def execute(self, 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``.
"""
ctx: LLMNodeContext[TOutput] = LLMNodeContext()
self.logger.debug(f"{state =}")
if not self._pre_exec(state, ctx):
self.logger.debug("Pre-exec check failed, skipping execution")
return {}
# Resolve the output schema for this run before the prompt is built
# (the prompt embeds it): the static schema by default, or a per-run
# one from a node that overrides ``_get_output_schema``.
ctx.output_schema = self._get_output_schema(state, ctx)
self._pre_exec_stream(ctx)
ctx.human_prompt = self._get_human_prompt(state, ctx)
ctx.system_prompt = self._get_system_prompt(state, ctx)
template = self._create_prompt_template(
ctx.system_prompt, ctx.human_prompt, ctx
)
ctx.template = template
ctx.variables = self._get_prompt_variables(state, ctx)
ctx.llm, ctx.config = self._configure_llm(ctx)
# Validation-retry loop: a node may reject a structurally invalid
# result (``_validate_result``) and re-invoke with the reason exposed
# as ``validation_feedback``, up to ``max_validation_retries`` times.
# On the final attempt the invalid result is kept and handled by
# ``_update_state`` (fail-closed), so the loop always terminates.
attempt = 0
# Accumulate token usage across every LLM invocation of this node.
# A validation retry overwrites ``ctx.output``, so without this the
# tokens burnt by the rejected attempt(s) never reach the totals.
usage_totals: dict[str, int] = {
"input_tokens": 0,
"output_tokens": 0,
"total_tokens": 0,
"reasoning_tokens": 0,
"cached_tokens": 0,
}
usage_calls = 0
while True:
# Re-derive the variables each attempt so the node's own
# ``_get_prompt_variables`` (and any state it reads) stays the
# single source, then inject the validation feedback.
ctx.variables = self._get_prompt_variables(state, ctx)
ctx.variables["validation_feedback"] = ctx.validation_feedback
prompt = self._invoke_prompt(template, ctx.variables, ctx)
ctx.prompt = prompt
chars_sent = len(prompt.to_string())
self.logger.debug(f"{chars_sent = } characters sent to LLM")
output = await self._invoke_llm(ctx)
ctx.output = output
chars_received = len(extract_llm_output_content(output))
self.logger.debug(f"{chars_received = } characters received from LLM")
# Fold this attempt's usage into the accumulator immediately, so a
# rejected attempt (which will be overwritten by the next one) is
# still accounted for exactly once.
attempt_usage = self._extract_usage(ctx)
if attempt_usage is not None:
usage_calls += 1
for field_name in usage_totals:
usage_totals[field_name] += getattr(attempt_usage, field_name)
ctx.result = self._process_output(ctx)
error = self._validate_result(ctx.result, state, ctx)
if error:
if attempt < self.max_validation_retries:
attempt += 1
ctx.validation_feedback = error
self.logger.warning(
"Invalid node result, retrying (%d/%d): %s",
attempt,
self.max_validation_retries,
error,
)
continue
# error, but run out of retries
else:
self.logger.warning(
"Validation failed after %d retries, result: %s",
attempt,
error,
)
# for nodes that want to use this, but others may choose to ignore
ctx.validation_feedback = error
break
else:
ctx.validation_feedback = ""
break
ctx.state_updates = self._update_state(ctx.result, state, ctx)
# token calculations: sum every attempt (retried and final) so the
# discarded-attempt tokens are not lost. Each call was added exactly
# once above, so the final attempt is not double counted.
if usage_calls:
ctx.token_usage = TokenUsage(role=self.model_role, **usage_totals)
else:
ctx.token_usage = None
final_state = self._update_usage_metrics(ctx.token_usage, ctx.state_updates)
ctx.final_state = final_state
self._post_exec_stream(state, ctx)
self.logger.debug(f"{final_state =}")
return final_state
def _post_exec_stream(self, state: TState, ctx: LLMNodeContext[TOutput]) -> None:
"""Emit streaming events after LLM invocation.
Emits the shared ``info``, ``debug``, and ``state`` events (via the
base) plus the ``usage`` event, which is LLM-specific: non-LLM
nodes have no token usage, so this stays in ``AbstractLLMNode``.
"""
super()._post_exec_stream(state, ctx)
usage = self._get_usage(ctx)
if usage:
event = NodeStreamEvent(type="usage", node=self.label, data=usage)
self.write_custom_stream(event.model_dump())
def _extract_usage(self, ctx: LLMNodeContext[TOutput]) -> TokenUsage | None:
"""Extract this node's token usage from the LLM output.
Reads ``usage_metadata`` when the provider populates it, otherwise
falls back to ``response_metadata['token_usage']``: many
OpenAI-compatible endpoints and gateways report counts only in the raw
provider payload, leaving ``usage_metadata`` empty. The reasoning-token
count is logged so per-node budgets can be benchmarked.
:param ctx: Per-run node context (its ``output`` carries the raw LLM
output)
:returns: ``TokenUsage`` for this node, or ``None`` if usage information is
unavailable
"""
output = ctx.output
if isinstance(output, dict) and "raw" in output:
output = output["raw"]
if not isinstance(output, AIMessage):
self.logger.debug("No message to get token usage from")
return None
usage, source = self._normalise_usage(output)
if usage is None:
self.logger.debug("No token usage available in message metadata")
return None
input_tokens = usage["input_tokens"]
output_tokens = usage["output_tokens"]
token_usage = TokenUsage(
input_tokens=input_tokens,
output_tokens=output_tokens,
total_tokens=input_tokens + output_tokens,
reasoning_tokens=usage["reasoning_tokens"],
cached_tokens=usage["cached_tokens"],
role=self.model_role,
)
self.logger.debug(
f"Node token usage ({source}): {token_usage} "
f"reasoning_tokens={usage['reasoning_tokens']}"
)
if usage["cached_tokens"]:
self.logger.info(
f"Cache hit for {self.label}: {usage['cached_tokens']} tokens cached"
)
return token_usage
@staticmethod
def _normalise_usage(output: AIMessage) -> tuple[dict[str, int] | None, str]:
"""Normalise token counts from either metadata shape.
``usage_metadata`` (LangChain-normalised) and
``response_metadata['token_usage']`` (raw provider payload) use
different key names; return a common shape so callers do not care which
the provider populated. ``cached_tokens`` is the cache-read count
only; cache-creation/write tokens remain part of the input count.
:param output: The raw ``AIMessage``.
:returns: ``(counts, source)``, or ``(None, "")`` when neither shape
carries usage.
"""
meta = output.usage_metadata
if meta:
output_details = meta.get("output_token_details") or {}
# langchain-anthropic nests cache counters under
# ``input_token_details``; other providers expose them at the top
# level. ``cached_tokens`` means cache-read (served-from-cache)
# tokens only; cache-creation/write tokens are normal input.
input_details = meta.get("input_token_details") or {}
return (
{
"input_tokens": int(meta.get("input_tokens", 0) or 0),
"output_tokens": int(meta.get("output_tokens", 0) or 0),
"cached_tokens": int(
input_details.get("cache_read")
or meta.get("cache_read_input_tokens")
or meta.get("cached_tokens")
or meta.get("prompt_cache_hit_tokens")
or 0
),
"reasoning_tokens": int(output_details.get("reasoning", 0) or 0),
},
"usage_metadata",
)
raw = (output.response_metadata or {}).get("token_usage") or {}
if not raw:
return None, ""
raw_details = raw.get("prompt_tokens_details") or {}
return (
{
"input_tokens": int(
raw.get("prompt_tokens", raw.get("input_tokens", 0)) or 0
),
"output_tokens": int(
raw.get("completion_tokens", raw.get("output_tokens", 0)) or 0
),
"cached_tokens": int(
raw_details.get("cached_tokens")
or raw.get("cache_read_input_tokens")
or raw.get("cached_tokens")
or raw.get("prompt_cache_hit_tokens")
or 0
),
"reasoning_tokens": int(
(raw.get("completion_tokens_details") or {}).get(
"reasoning_tokens", 0
)
or 0
),
},
"response_metadata",
)
def _get_usage(self, ctx: LLMNodeContext[TOutput]) -> NodeStreamData | None:
"""Build a ``NodeStreamData`` wrapper for the current node's token usage.
:param ctx: Per-run node context.
:returns: ``NodeStreamData`` with numeric token details, or ``None`` if no
usage recorded
"""
if ctx.token_usage is not None:
details = ctx.token_usage.model_dump()
self.logger.debug(f"{details =}")
return NodeStreamData(summary="", details=details)
return None
def _update_usage_metrics(
self,
token_usage: TokenUsage | None = None,
current_state_updates: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Add this node's token usage delta to the state updates dict.
:param token_usage: Usage for this node (returned by
:meth:`_extract_usage`)
:param current_state_updates: The dict returned by :meth:`_update_state`
:returns: ``current_state_updates`` with ``usage_metrics`` injected,
or an empty dict if both arguments are falsy
"""
result = current_state_updates or {}
if token_usage:
result["usage_metrics"] = token_usage
self.logger.debug(
"Injected token usage into state updates: %s", token_usage
)
return result
def _get_inspect(
self, state: TState, ctx: LLMNodeContext[TOutput]
) -> NodeStreamData | None:
"""Return structured inspection data for an ``inspect`` event.
Override in subclasses to provide node-specific inspection data: a
short summary plus a ``details`` dict. The summary is always shown;
the details are the drill-down. Per-run values come from *ctx*.
:returns: NodeStreamData with summary and details, or None to skip
Example::
details = {"classified_domains": ["neuron", "morphology"]}
# ``ctx.system_prompt`` may be a list when the node keeps a
# verbatim memory window; use the formatted prompt instead.
details["input_prompt"] = prompt_value_to_messages(ctx.prompt)
return NodeStreamData(summary="Classified", details=details)
"""
return None
def _get_status(
self, state: TState, ctx: LLMNodeContext[TOutput]
) -> NodeStreamData | None:
"""Return status pane content for this node.
Override to populate the status pane with display-ready
markdown content. The ``display`` field of the returned
``NodeStreamData`` is rendered in the status pane; the frontend
replaces the previous entry for this node label so loops do not
accumulate.
:returns: NodeStreamData with display content, or None to skip
"""
return None
@abstractmethod
def _configure_llm(
self, ctx: LLMNodeContext[TOutput]
) -> tuple[Runnable, RunnableConfig]:
"""Configure LLM with structured output and build per-invoke config.
:param ctx: Per-run node context.
:returns: (llm_with_structured_output, config_dict) where
config_dict is a ``RunnableConfig`` with the ``configurable``
key populated for ``llm.ainvoke()``.
"""
...
@abstractmethod
async def _invoke_llm(
self, ctx: LLMNodeContext[TOutput]
) -> AIMessage | dict[str, Any]:
"""Async invoke LLM --- must use ``await llm.ainvoke()`` so the
event loop can process streaming callbacks (waiter pattern)
during the LLM call rather than blocking until it completes.
Reads the model, prompt and config from *ctx*.
:param ctx: Per-run node context.
"""
...
@abstractmethod
def _process_output(self, ctx: LLMNodeContext[TOutput]) -> Any:
"""Common output processing with error handling.
Reads the raw output and resolved schema from *ctx*.
:param ctx: Per-run node context.
"""
...
@abstractmethod
def _invoke_prompt(
self,
prompt_template: ChatPromptTemplate,
variables: Any | dict[str, Any],
ctx: LLMNodeContext[TOutput],
) -> PromptValue:
"""Format prompt with state-specific parameters.
:param ctx: Per-run node context.
"""
...
@abstractmethod
def _get_human_prompt(self, state: TState, ctx: LLMNodeContext[TOutput]) -> str:
"""Return human prompt for this node.
:param ctx: Per-run node context.
"""
...
@abstractmethod
def _get_system_prompt(
self, state: TState, ctx: LLMNodeContext[TOutput]
) -> str | list[Any]:
"""Return system prompt for this node.
May return a list of ``("system", text)`` plus recent history
message objects (when the node keeps a verbatim memory window),
which ``_create_prompt_template`` places between the system and
human prompts.
:param ctx: Per-run node context.
"""
...
@abstractmethod
def _create_prompt_template(
self,
system_prompt: str | list[Any],
human_prompt: str,
ctx: LLMNodeContext[TOutput],
) -> ChatPromptTemplate:
"""Create ChatPromptTemplate for this node.
:param ctx: Per-run node context.
"""
...
@abstractmethod
def _get_prompt_variables(
self, state: TState, ctx: LLMNodeContext[TOutput]
) -> dict:
"""Format prompt with state-specific parameters.
:param ctx: Per-run node context.
"""
...
@abstractmethod
def _update_state(
self, result: Any, state: TState, ctx: LLMNodeContext[TOutput]
) -> dict[str, Any]:
"""Update and return state dictionary.
:param ctx: Per-run node context.
"""
...
def _validate_result(
self, result: Any, state: TState, ctx: LLMNodeContext[TOutput]
) -> str | None:
"""Return a reason string when *result* is structurally invalid.
Called by :meth:`execute` after :meth:`_process_output`. Returning a
non-``None`` string triggers a bounded re-invoke (see
:attr:`max_validation_retries`) with the reason exposed to the prompt
as ``validation_feedback``. Use this for machine-checkable
consistency (for example a plan whose references do not resolve),
not for subjective quality.
Default: ``None`` (no validation).
:param result: The processed node result.
:param state: Current graph state.
:param ctx: Per-run node context.
:returns: An error description, or ``None`` when valid.
"""
return None
@abstractmethod
def _get_default_error_result(self, ctx: LLMNodeContext[TOutput]) -> Any:
"""Return default result when processing fails.
:param ctx: Per-run node context.
"""
...
[docs]
class AbstractRouterNode[TState: BaseModel](
AbstractLangGraphNode[TState, str, NodeContext]
):
"""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()``.
"""
[docs]
@abstractmethod
async def execute(self, state: TState) -> str:
"""Return the routing label (edge name) based on state."""
...