OSSS.ai.agents.metadata¶
OSSS.ai.agents.metadata
¶
Unified Agent Metadata with Multi-Axis Classification.
This module consolidates agent metadata from registry.py and dynamic_composition.py while adding the multi-axis classification system needed for event-driven architecture and future utility agent integration.
AgentMetadata
¶
Bases: BaseModel
Unified agent metadata with multi-axis classification.
Combines functionality from registry-based agent management with dynamic discovery capabilities and adds multi-axis classification for intelligent event routing and service extraction.
validate_string_lists(v)
classmethod
¶
Validate string list fields.
validate_agent_class(v)
classmethod
¶
Handle agent_class validation and conversion from string.
model_post_init(__context)
¶
Initialize derived fields and defaults.
can_replace(other)
¶
Check if this agent can replace another agent.
Uses both capability matching and version compatibility.
is_compatible_with_task(task_type, domain=None)
¶
Check if this agent is compatible with a specific task type.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
task_type
|
str
|
Type of task ("transform", "evaluate", "retrieve", etc.) |
required |
domain
|
Optional[str]
|
Optional domain specialization |
None
|
Returns:
| Type | Description |
|---|---|
bool
|
True if agent can handle this task type |
get_performance_tier()
¶
Get performance tier based on cognitive characteristics.
Returns:
| Type | Description |
|---|---|
str
|
Performance tier: "fast", "balanced", "thorough" |
to_dict()
¶
Convert to dictionary representation for serialization.
create_for_registry(name, agent_class, requires_llm=False, constructor_pattern=None, description='', dependencies=None, is_critical=True, failure_strategy=FailurePropagationStrategy.FAIL_FAST, fallback_agents=None, health_checks=None, cognitive_speed='adaptive', cognitive_depth='variable', processing_pattern='atomic', primary_capability='', secondary_capabilities=None, pipeline_role='standalone', bounded_context='reflection')
classmethod
¶
Create AgentMetadata for registry registration.
This factory method properly handles None values for list fields, letting Pydantic handle default_factory creation instead of manually constructing empty lists.
create_default(name='default_agent', agent_class=None, description='Default agent metadata')
classmethod
¶
Create default agent metadata.
from_dict(data)
classmethod
¶
Create AgentMetadata from dictionary representation.
TaskClassification
¶
Bases: BaseModel
Classification of work being performed for semantic event routing.
This accepts the raw classifier payload from agent_output_meta["_classifier"]
and derives a concrete task_type for NodeExecutionContext / DecisionNode.
Expected upstream payload keys (all optional except task_type when provided): - intent: high-level label, e.g. "action", "question", "query" - confidence: overall intent confidence - domain, domain_confidence - topic, topic_confidence, topics - sub_intent, sub_intent_confidence - labels, raw - model_version - original_text, normalized_text, query_terms
derive_task_type_from_classifier(data)
classmethod
¶
Accept raw classifier payloads (like the _classifier block) and derive a concrete task_type if it's missing.
This is what prevents NodeExecutionContext construction from failing when only classifier fields are present.
to_dict()
¶
Convert to dictionary representation - backward compatibility.
from_dict(data)
classmethod
¶
Create TaskClassification from dictionary - backward compatibility.
classify_query_task(query)
¶
Classify a user query into task type and characteristics.
This is a simplified rule-based implementation. In production, this could use LLM-powered classification or ML models.