Orchestrator framework

class klea_utils.graph.base.BaseLangGraph(logging_level: int = 10, checkpoint: str = 'inmemory', log_file: bool = True)

Bases: ABC

Abstract base class for LangGraph-based orchestrators.

Provides common infrastructure for: - Configuration loading from env files - MCP client creation from JSON config - LLM model setup (delegated to subclasses) - LangGraph compilation and execution - Session checkpointing - Dual-stream logging

Subclasses must implement: - _setup_models(): Create LLM model instances - _create_graph(): Build and compile the LangGraph - Set env_class to the appropriate Pydantic settings class

config_class: type[BaseModel]

Pydantic BaseModel class for configuration loading. Subclasses must set this to their AppConfig class.

env_class: type[BaseModel]

Pydantic BaseSettings class for env loading. Subclasses must set this to their AppEnv class.

env_file_default: str = 'config.env'

Default config file name if the environment variable is not set.

env_var: str = 'ENV_FILE'

Name of the environment variable that controls the env file path.

graph_name: str = 'BaseLangGraph'

Logger name for this orchestrator, also used as the app name for platformdirs data/cache directories.

async graph_stream(query: str, thread_id: str = 'default_thread') → Any

Run the graph and return the raw astream result.

Parameters:

• query – User query string

• thread_id – Session/thread identifier for checkpointing

Returns:

Raw async generator from graph.astream()

async run_graph_astream_events(query: str, thread_id: str = 'default_thread')

Run the graph and yield structured streaming events.

Yields dicts with:

{"type": "progress", "node": "<label>"}

When the graph enters a new node (via write_custom_stream)

{"type": "info", "node": "<label>", "data": {...}}

Structured summary data from a node after execution

{"type": "debug", "node": "<label>", "data": {...}}

Full data dump from a node after execution

{"type": "token", "content": "<chunk>", "node": "<label>"}

LLM token chunk from the current node

{"type": "usage", "node": "<label>", "data": {...}}

Per-node token usage (input / output / total tokens)

{"type": "complete", "message_for_user": "<answer>"}

Final answer from the completed graph

Uses LangGraph’s astream_events v3 protocol. Progress events from all nodes (LLM and non-LLM) arrive via the custom channel. LLM token output is read from the messages channel. A StreamTransformer enables the custom channel so those events flow through.

Reference: https://docs.langchain.com/oss/python/langgraph/event-streaming

Parameters:

• query – User query string

• thread_id – Session/thread identifier for checkpointing

Yields:

Structured event dicts

async run_graph_invoke(query: str, thread_id: str = 'default_thread') → str

Run the graph with a simple string query.

Parameters:

• query – User query string

• thread_id – Session/thread identifier for checkpointing

Returns:

The message_for_user field from the final state

async run_graph_invoke_state(state: dict, thread_id: str = 'default_thread') → dict

Run the graph, accepting and returning full state dicts.

Parameters:

• state – Initial graph state (must contain query key)

• thread_id – Session/thread identifier for checkpointing

Returns:

Final graph state

async run_graph_stream(query: str, thread_id: str = 'default_thread')

Run the graph and yield intermediate message_for_user values.

Parameters:

• query – User query string

• thread_id – Session/thread identifier for checkpointing

Yields:

message_for_user strings from each node

final async setup() → None

Set up the orchestrator.

Calls hooks and template methods in this order: 1. _pre_setup() 2. _setup_checkpointer() 3. _load_env() 4. _setup_models() 5. _create_mcp_client() 6. _pre_graph() 7. _create_graph() 8. _post_setup()

Shared state schemas

Schemas shared by LangGraph orchestrators.

File: klea_utils/graph/schemas.py

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

class klea_utils.graph.schemas.TokenUsage(*, input_tokens: int = 0, output_tokens: int = 0, total_tokens: int = 0)

Bases: BaseModel

Token usage accumulated across the nodes in a graph run.

model_config = {}

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

Reducers

Reducers shared by LangGraph orchestrators.

File: klea_utils/graph/reducers.py

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

klea_utils.graph.reducers.add_token_usage(left: TokenUsage | dict[str, int], right: TokenUsage | dict[str, int]) → TokenUsage

Add token usage updates from sequential or concurrent graph nodes.