Classification#

Infrastructure node that handles task classification and capability selection by analyzing user tasks against available capabilities.

ClassificationNode#

class framework.infrastructure.classification_node.ClassificationNode[source]#

Bases: BaseInfrastructureNode

Convention-based classification node with sophisticated capability selection logic.

Analyzes user tasks and selects appropriate capabilities using parallel LLM-based classification with few-shot examples. Handles both initial classification and reclassification scenarios.

Uses LangGraph’s sophisticated state merging with built-in error handling and retry policies optimized for LLM-based classification operations.

name: str = 'classifier'#

description: str = 'Task Classification and Capability Selection'#

static classify_error(exc, context)[source]#

Built-in error classification for classifier operations.

Parameters:

exc (Exception) – Exception that occurred
context (Dict[str, Any]) – Error context information

Returns:

Classification with severity and retry guidance

Return type:

ErrorClassification

static get_retry_policy()[source]#

Custom retry policy for LLM-based classification operations.

Classification uses parallel LLM calls for capability selection and can be flaky due to: - Multiple concurrent LLM requests - Network timeouts to LLM services - LLM provider rate limiting - Classification model variability

Use more attempts with moderate delays for better reliability.

Return type:: Dict[str, Any]

async static execute(state, **kwargs)[source]#

Main classification logic with sophisticated capability selection and reclassification handling.

Analyzes user tasks and selects appropriate capabilities using parallel LLM-based classification. Handles both initial classification and reclassification scenarios with state preservation.

Parameters:

state (AgentState) – Current agent state
kwargs – Additional LangGraph parameters

Returns:

Dictionary of state updates for LangGraph

Return type:

Dict[str, Any]

__repr__()#

Return a string representation of the infrastructure node for debugging.

Provides a concise string representation that includes both the Python class name and the infrastructure node’s registered name. This is useful for debugging, logging, and development workflows where infrastructure nodes need to be identified clearly.

Returns:: String representation including class name and node name
Return type:: str

Example

>>> node = TaskExtractionNode()
>>> repr(node)
'<TaskExtractionNode: task_extraction>'

Note

The format follows the pattern ‘<ClassName: node_name>’ for consistency across all framework components.

async langgraph_node(**kwargs)#

LangGraph-native node function with manual error handling.

This function is called by LangGraph during execution. Infrastructure nodes now use get_stream_writer() and get_config() directly for pure LangGraph integration.

Parameters:

state (AgentState) – Current agent state
kwargs – Additional parameters from LangGraph

Returns:

State updates dictionary

Return type:

Dict[str, Any]

Supporting Functions#

async framework.infrastructure.classification_node.select_capabilities(task, available_capabilities, state, logger, previous_failure=None)[source]#

Select capabilities needed for the task by using classification.

Parameters:

task (str) – Task description for analysis
available_capabilities (List[BaseCapability]) – Available capabilities to choose from
state (AgentState) – Current agent state
logger – Logger instance

Returns:

List of capability names needed for the task

Return type:

List[str]

Core Models#

Classification uses models defined in the core framework:

Registration#

Automatically registered as:

NodeRegistration(
    name="classifier",
    module_path="framework.infrastructure.classification_node",
    function_name="ClassificationNode",
    description="Task classification and capability selection"
)