freemo/cleveragents-core
0
0
Fork 0
forked from cleveragents/cleveragents-core
Code Pull Requests Activity

fix: remove cast with runtime type checking; replace all Dict[..] with dict[..]

Browse Source
This commit is contained in:
Aditya Chhabra
2025-10-30 18:30:47 +05:30
parent 6be9073aa0
commit 170657113f
36 changed files with 441 additions and 417 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cleveragents/agents/base.py
+14 -15
View File
@@ -12,7 +12,6 @@ from abc import ABC
from abc import abstractmethod
from typing import Any
from typing import Callable
from typing import Dict
from typing import List
from typing import Optional
@@ -36,14 +35,14 @@ class Agent(ABC):
Attributes:
name (str): The name of the agent.
config (Dict[str, Any]): The agent's configuration.
config (dict[str, Any]): The agent's configuration.
template_renderer (TemplateRenderer): Renderer for processing templates.
input_stream (Subject): Input stream for receiving messages.
output_stream (Subject): Output stream for emitting processed messages.
"""
def __init__(
self, name: str, config: Dict[str, Any], template_renderer: TemplateRenderer
self, name: str, config: dict[str, Any], template_renderer: TemplateRenderer
):
"""
Initialize a reactive agent.
@@ -76,7 +75,7 @@ class Agent(ABC):
on_next=self.output_stream.on_next, on_error=self.output_stream.on_error
)
async def _process_wrapper(self, message_data: tuple[str, Dict[str, Any]]) -> str:
async def _process_wrapper(self, message_data: tuple[str, dict[str, Any]]) -> str:
"""Wrapper for async processing."""
try:
if isinstance(message_data, tuple):
@@ -93,7 +92,7 @@ class Agent(ABC):
@abstractmethod
async def process_message(
self, message: str, context: Optional[Dict[str, Any]] = None
self, message: str, context: Optional[dict[str, Any]] = None
) -> str:
"""
Process a message asynchronously.
@@ -111,7 +110,7 @@ class Agent(ABC):
# Legacy method for backward compatibility
async def process(
self, message: str, context: Optional[Dict[str, Any]] = None
self, message: str, context: Optional[dict[str, Any]] = None
) -> str:
"""Legacy synchronous interface adapter."""
return await self.process_message(message, context)
@@ -125,14 +124,14 @@ class Agent(ABC):
A list of capability identifiers.
"""
def get_metadata(self) -> Dict[str, Any]:
def get_metadata(self) -> dict[str, Any]:
"""
Get metadata about the agent.
Returns:
A dictionary of agent metadata.
"""
metadata: Dict[str, Any] = {
metadata: dict[str, Any] = {
"name": self.name,
"type": self.__class__.__name__,
"capabilities": self.get_capabilities(),
@@ -147,7 +146,7 @@ class Agent(ABC):
return metadata
def send_message(self, message: str, context: Optional[Dict[str, Any]] = None) -> None:
def send_message(self, message: str, context: Optional[dict[str, Any]] = None) -> None:
"""Send a message to the agent's input stream."""
self.input_stream.on_next((message, context or {}))
@@ -175,14 +174,14 @@ class AgentWithMemory(Agent):
capabilities for maintaining context between message processing calls.
Attributes:
memory (Dict[str, Any]): The agent's memory/state.
memory (dict[str, Any]): The agent's memory/state.
"""
# Class-level threading lock to protect asyncio.Lock initialization
_memory_lock_init_lock = threading.Lock()
def __init__(
self, name: str, config: Dict[str, Any], template_renderer: TemplateRenderer
self, name: str, config: dict[str, Any], template_renderer: TemplateRenderer
):
"""
Initialize a reactive agent with memory.
@@ -193,7 +192,7 @@ class AgentWithMemory(Agent):
template_renderer: Renderer for processing templates.
"""
super().__init__(name, config, template_renderer)
self.memory: Dict[str, Any] = {}
self.memory: dict[str, Any] = {}
self._memory_lock_instance: Optional[asyncio.Lock] = None
@property
@@ -221,12 +220,12 @@ class AgentWithMemory(Agent):
self._memory_lock_instance = asyncio.Lock()
return self._memory_lock_instance
async def _process_wrapper(self, message_data: tuple[str, Dict[str, Any]]) -> str:
async def _process_wrapper(self, message_data: tuple[str, dict[str, Any]]) -> str:
"""Wrapper that manages memory access."""
async with self._memory_lock:
return await super()._process_wrapper(message_data)
def save_memory(self) -> Dict[str, Any]:
def save_memory(self) -> dict[str, Any]:
"""
Save the agent's memory to a serializable format.
@@ -235,7 +234,7 @@ class AgentWithMemory(Agent):
"""
return copy.deepcopy(self.memory)
def load_memory(self, memory: Dict[str, Any]) -> None:
def load_memory(self, memory: dict[str, Any]) -> None:
"""
Load the agent's memory from a serialized format.
src/cleveragents/agents/chain.py
+4 -5
View File
@@ -5,7 +5,6 @@ This module defines the ChainAgent class, which represents an agent that chains
"""
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
@@ -20,7 +19,7 @@ class ChainAgent(Agent):
Attributes:
name (str): The name of the agent.
config (Dict[str, Any]): The agent's configuration.
config (dict[str, Any]): The agent's configuration.
template_renderer (TemplateRenderer): Renderer for processing templates.
steps (List[str]): A list of processing steps.
"""
@@ -28,7 +27,7 @@ class ChainAgent(Agent):
prompt_template: Optional[str]
def __init__(
self, name: str, config: Dict[str, Any], template_renderer: TemplateRenderer
self, name: str, config: dict[str, Any], template_renderer: TemplateRenderer
):
super().__init__(name, config, template_renderer)
self.steps = config.get("steps", [])
@@ -47,7 +46,7 @@ class ChainAgent(Agent):
self.prompt_template = prompt
async def process_message(
self, message: str, context: Optional[Dict[str, Any]] = None
self, message: str, context: Optional[dict[str, Any]] = None
) -> str:
"""
Process a message by sequentially applying a series of steps.
@@ -73,7 +72,7 @@ class ChainAgent(Agent):
return f"ChainAgent processed: {result}"
async def process(
self, message: str, context: Optional[Dict[str, Any]] = None
self, message: str, context: Optional[dict[str, Any]] = None
) -> str:
"""
Process a message (legacy method for compatibility).
src/cleveragents/agents/composite.py
+21 -22
View File
@@ -9,7 +9,6 @@ import asyncio
import logging
from typing import TYPE_CHECKING
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
@@ -41,29 +40,29 @@ class CompositeAgent(Agent): # pylint: disable=too-many-instance-attributes
complexity of its internal components.
Configuration:
components (Dict[str, Any]):
components (dict[str, Any]):
Defines the internal components:
- agents: Agent instances or templates
- graphs: LangGraph definitions or templates
- streams: RxPy stream definitions or templates
routing (Dict[str, Any]):
routing (dict[str, Any]):
Defines how data flows through components:
- input: Where incoming messages are sent
- output: Where final results are collected
- connections: How components are connected
expose_params (Dict[str, Any]):
expose_params (dict[str, Any]):
Parameters exposed by this composite that can be
overridden when instantiated.
Attributes:
name (str): The name of the agent.
config (Dict[str, Any]): The agent's configuration.
config (dict[str, Any]): The agent's configuration.
template_renderer (TemplateRenderer): Renderer for processing templates.
components (Dict[str, Dict[str, Any]]): Internal components.
routing (Dict[str, Any]): Routing configuration.
expose_params (Dict[str, Any]): Exposed parameters.
components (dict[str, dict[str, Any]]): Internal components.
routing (dict[str, Any]): Routing configuration.
expose_params (dict[str, Any]): Exposed parameters.
stream_router (Optional[ReactiveStreamRouter]): Stream router for RxPy.
langgraph_bridge (Optional[RxPyLangGraphBridge]): Bridge for LangGraphs.
"""
@@ -71,7 +70,7 @@ class CompositeAgent(Agent): # pylint: disable=too-many-instance-attributes
def __init__( # pylint: disable=too-many-arguments,too-many-positional-arguments
self,
name: str,
config: Dict[str, Any],
config: dict[str, Any],
template_renderer: TemplateRenderer,
stream_router: Optional[ReactiveStreamRouter] = None,
langgraph_bridge: Optional[RxPyLangGraphBridge] = None,
@@ -81,7 +80,7 @@ class CompositeAgent(Agent): # pylint: disable=too-many-instance-attributes
Args:
name (str): The name of the agent.
config (Dict[str, Any]): The agent's configuration.
config (dict[str, Any]): The agent's configuration.
template_renderer (TemplateRenderer): Renderer for processing templates.
stream_router (Optional[ReactiveStreamRouter]): Stream router for RxPy integration.
langgraph_bridge (Optional[RxPyLangGraphBridge]): Bridge for LangGraph integration.
@@ -100,9 +99,9 @@ class CompositeAgent(Agent): # pylint: disable=too-many-instance-attributes
self.expose_params = self.config.get("expose_params", {})
# Internal component storage
self.agents: Dict[str, Agent] = {}
self.graphs: Dict[str, Any] = {}
self.streams: Dict[str, Any] = {}
self.agents: dict[str, Agent] = {}
self.graphs: dict[str, Any] = {}
self.streams: dict[str, Any] = {}
# Reject legacy strategy-based configuration
if "strategy" in self.config:
@@ -162,7 +161,7 @@ class CompositeAgent(Agent): # pylint: disable=too-many-instance-attributes
# Parameter propagation to components is not yet implemented
async def process_message(
self, message: str, context: Optional[Dict[str, Any]] = None
self, message: str, context: Optional[dict[str, Any]] = None
) -> str:
"""
Process a message through the composite's internal workflow.
@@ -172,14 +171,14 @@ class CompositeAgent(Agent): # pylint: disable=too-many-instance-attributes
return await self.process(message, context)
async def process(
self, message: str, context: Optional[Dict[str, Any]] = None
self, message: str, context: Optional[dict[str, Any]] = None
) -> str:
"""
Process a message through the composite's internal workflow.
Args:
message (str): The message to process.
context (Dict[str, Any], optional): Additional context for processing.
context (dict[str, Any], optional): Additional context for processing.
Returns:
str: The processed response.
@@ -227,7 +226,7 @@ class CompositeAgent(Agent): # pylint: disable=too-many-instance-attributes
)
async def _process_via_agent(
self, agent_name: str, message: str, context: Dict[str, Any]
self, agent_name: str, message: str, context: dict[str, Any]
) -> str:
"""
Process message through a specific agent.
@@ -235,7 +234,7 @@ class CompositeAgent(Agent): # pylint: disable=too-many-instance-attributes
Args:
agent_name (str): Name of the agent to use.
message (str): The message to process.
context (Dict[str, Any]): Processing context.
context (dict[str, Any]): Processing context.
Returns:
str: Processed response.
@@ -249,7 +248,7 @@ class CompositeAgent(Agent): # pylint: disable=too-many-instance-attributes
return await agent.process(message, context)
async def _process_via_graph(
self, graph_name: str, message: str, context: Dict[str, Any]
self, graph_name: str, message: str, context: dict[str, Any]
) -> str:
"""
Process message through a LangGraph.
@@ -257,7 +256,7 @@ class CompositeAgent(Agent): # pylint: disable=too-many-instance-attributes
Args:
graph_name (str): Name of the graph to use.
message (str): The message to process.
context (Dict[str, Any]): Processing context.
context (dict[str, Any]): Processing context.
Returns:
str: Processed response.
@@ -293,7 +292,7 @@ class CompositeAgent(Agent): # pylint: disable=too-many-instance-attributes
return str(result)
async def _process_via_stream(
self, stream_name: str, message: str, context: Dict[str, Any]
self, stream_name: str, message: str, context: dict[str, Any]
) -> str:
"""
Process message through an RxPy stream.
@@ -301,7 +300,7 @@ class CompositeAgent(Agent): # pylint: disable=too-many-instance-attributes
Args:
stream_name (str): Name of the stream to use.
message (str): The message to process.
context (Dict[str, Any]): Processing context.
context (dict[str, Any]): Processing context.
Returns:
str: Processed response.
src/cleveragents/agents/factory.py
+8 -9
View File
@@ -8,7 +8,6 @@ configuration, supporting the RxPy-based reactive architecture.
from __future__ import annotations
from typing import Any
from typing import Dict
from typing import Optional
from typing import Type
@@ -28,14 +27,14 @@ class AgentFactory:
reactive architecture, providing stream processing capabilities.
Attributes:
agent_types (Dict[str, Type[Agent]]): Registry of agent types.
agent_types (dict[str, Type[Agent]]): Registry of agent types.
template_renderer (TemplateRenderer): Renderer for processing templates.
config (Dict[str, Any]): Configuration for agent creation.
config (dict[str, Any]): Configuration for agent creation.
"""
def __init__(
self,
config: Dict[str, Any],
config: dict[str, Any],
template_renderer: TemplateRenderer,
stream_router: Optional[Any] = None,
langgraph_bridge: Optional[Any] = None,
@@ -57,7 +56,7 @@ class AgentFactory:
self.config = config
self.stream_router = stream_router
self.langgraph_bridge = langgraph_bridge
self.agents: Dict[str, Agent] = {} # Cache of created agents
self.agents: dict[str, Agent] = {} # Cache of created agents
def register_agent_type(self, type_name: str, agent_class: Type[Agent]) -> None:
"""
@@ -77,7 +76,7 @@ class AgentFactory:
self.agent_types[type_name] = agent_class
def get_agent_types(self) -> Dict[str, Type[Agent]]:
def get_agent_types(self) -> dict[str, Type[Agent]]:
"""
Get registered agent types.
@@ -122,7 +121,7 @@ class AgentFactory:
return agent
def _create_agent_instance(
self, agent_name: str, agent_type: str, agent_config: Dict[str, Any]
self, agent_name: str, agent_type: str, agent_config: dict[str, Any]
) -> Agent:
"""
Create an agent instance of the specified type.
@@ -205,7 +204,7 @@ class AgentFactory:
f"Failed to create agent '{agent_name}' of type '{agent_type}': {str(e)}"
) from e
def create_agents_from_config(self) -> Dict[str, Agent]:
def create_agents_from_config(self) -> dict[str, Agent]:
"""
Create all agents from the configuration.
@@ -257,7 +256,7 @@ class AgentFactory:
f"'config' for agent '{agent_name}' must be a dictionary"
)
def get_agent_metadata(self, agent_name: str) -> Dict[str, Any]:
def get_agent_metadata(self, agent_name: str) -> dict[str, Any]:
"""
Get metadata for an agent without creating it.
src/cleveragents/agents/llm.py
+6 -7
View File
@@ -7,7 +7,6 @@ capabilities to work within the RxPy-based reactive architecture using LangChain
import logging
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Type
@@ -41,7 +40,7 @@ class LLMAgent(AgentWithMemory):
"""
def __init__(
self, name: str, config: Dict[str, Any], template_renderer: TemplateRenderer
self, name: str, config: dict[str, Any], template_renderer: TemplateRenderer
):
"""
Initialize a reactive LLM agent using LangChain.
@@ -81,7 +80,7 @@ class LLMAgent(AgentWithMemory):
def _create_chat_model(self) -> BaseChatModel:
"""Create LangChain chat model based on provider configuration."""
# Provider configuration mapping with explicit types
provider_config: Dict[str, Dict[str, Any]] = {
provider_config: dict[str, dict[str, Any]] = {
"openai": {
"class": ChatOpenAI,
"param_renames": {
@@ -115,13 +114,13 @@ class LLMAgent(AgentWithMemory):
config = provider_config[provider_lower]
# Build common kwargs
common_kwargs: Dict[str, Any] = {
common_kwargs: dict[str, Any] = {
"model": self.model,
"temperature": self.temperature,
}
# Apply parameter renames
param_renames: Dict[str, str] = config["param_renames"]
param_renames: dict[str, str] = config["param_renames"]
# Add max_tokens with provider-specific name
max_tokens_param = param_renames["max_tokens"]
@@ -143,7 +142,7 @@ class LLMAgent(AgentWithMemory):
) from e
async def process_message( # pylint: disable=too-many-branches
self, message: str, context: Optional[Dict[str, Any]] = None
self, message: str, context: Optional[dict[str, Any]] = None
) -> str:
"""
Process a message through the LLM using LangChain.
@@ -244,7 +243,7 @@ class LLMAgent(AgentWithMemory):
return capabilities
def get_metadata(self) -> Dict[str, Any]:
def get_metadata(self) -> dict[str, Any]:
"""Get metadata about the LLM agent."""
metadata = super().get_metadata()
metadata.update(
src/cleveragents/agents/states.py
+1 -2
View File
@@ -3,10 +3,9 @@ This module contains state management for agents.
"""
from typing import Any
from typing import Dict
def initial_state() -> Dict[str, Any]:
def initial_state() -> dict[str, Any]:
"""
Returns the initial state for an agent.
"""
src/cleveragents/agents/tool.py
+14 -15
View File
@@ -12,7 +12,6 @@ import os
import re
import subprocess
from typing import Any
from typing import Dict
from typing import List
from typing import Optional, Union, Literal
@@ -35,7 +34,7 @@ class ToolAgent(Agent):
"""
def __init__(
self, name: str, config: Dict[str, Any], template_renderer: TemplateRenderer
self, name: str, config: dict[str, Any], template_renderer: TemplateRenderer
):
"""
Initialize a reactive tool agent.
@@ -84,7 +83,7 @@ class ToolAgent(Agent):
else:
raise AgentCreationError(f"Invalid tool configuration: {tool}")
def _extract_json_from_message(self, message: str) -> Optional[Dict[str, Any]]:
def _extract_json_from_message(self, message: str) -> Optional[dict[str, Any]]:
"""
Extract JSON from a message that might contain markdown code blocks or extra text.
@@ -134,7 +133,7 @@ class ToolAgent(Agent):
return None
async def process_message(
self, message: str, context: Optional[Dict[str, Any]] = None
self, message: str, context: Optional[dict[str, Any]] = None
) -> str:
"""
Process a message by executing the appropriate tool.
@@ -185,8 +184,8 @@ class ToolAgent(Agent):
async def _execute_tool(
self,
tool_name: str,
tool_args: Dict[str, Any],
context: Optional[Dict[str, Any]],
tool_args: dict[str, Any],
context: Optional[dict[str, Any]],
) -> Any:
"""Execute a specific tool."""
# Check if tool is in the allowed list first
@@ -213,7 +212,7 @@ class ToolAgent(Agent):
f"Shell execution disabled, cannot execute '{tool_name}'"
)
async def _execute_shell_command(self, command: str, args: Dict[str, Any]) -> str:
async def _execute_shell_command(self, command: str, args: dict[str, Any]) -> str:
"""Execute a shell command safely."""
if self.safe_mode:
# Basic safety checks
@@ -253,7 +252,7 @@ class ToolAgent(Agent):
# Built-in tools
async def _echo_tool(
self, args: Dict[str, Any], context: Optional[Dict[str, Any]] # pylint: disable=unused-argument
self, args: dict[str, Any], context: Optional[dict[str, Any]] # pylint: disable=unused-argument
) -> str:
"""Simple echo tool."""
text = args.get("text", "")
@@ -262,7 +261,7 @@ class ToolAgent(Agent):
return str(text)
async def _math_tool(
self, args: Dict[str, Any], context: Optional[Dict[str, Any]] # pylint: disable=unused-argument
self, args: dict[str, Any], context: Optional[dict[str, Any]] # pylint: disable=unused-argument
) -> str:
"""Basic math evaluation tool."""
expression = args.get("expression", "")
@@ -291,7 +290,7 @@ class ToolAgent(Agent):
raise ExecutionError(f"Math evaluation failed: {e}") from e
async def _json_parse_tool(
self, args: Dict[str, Any], context: Optional[Dict[str, Any]] # pylint: disable=unused-argument
self, args: dict[str, Any], context: Optional[dict[str, Any]] # pylint: disable=unused-argument
) -> str:
"""JSON parsing tool."""
json_str = args.get("json", "")
@@ -305,7 +304,7 @@ class ToolAgent(Agent):
raise ExecutionError(f"JSON parsing failed: {e}") from e
async def _http_request_tool(
self, args: Dict[str, Any], context: Optional[Dict[str, Any]] # pylint: disable=unused-argument
self, args: dict[str, Any], context: Optional[dict[str, Any]] # pylint: disable=unused-argument
) -> str:
"""HTTP request tool."""
url = args.get("url", "")
@@ -327,7 +326,7 @@ class ToolAgent(Agent):
raise ExecutionError(f"HTTP request failed: {e}") from e
async def _file_read_tool(
self, args: Dict[str, Any], context: Optional[Dict[str, Any]]
self, args: dict[str, Any], context: Optional[dict[str, Any]]
) -> str:
"""File reading tool."""
filepath = args.get("file", "")
@@ -499,7 +498,7 @@ class ToolAgent(Agent):
return existing_lines, insert_location
async def _file_write_tool(
self, args: Dict[str, Any], context: Optional[Dict[str, Any]]
self, args: dict[str, Any], context: Optional[dict[str, Any]]
) -> str:
"""File writing tool with support for write, append, and insert modes."""
filepath = args.get("file", "")
@@ -574,9 +573,9 @@ class ToolAgent(Agent):
return capabilities
def get_metadata(self) -> Dict[str, Any]:
def get_metadata(self) -> dict[str, Any]:
"""Get metadata about the tool agent."""
metadata: Dict[str, Any] = super().get_metadata()
metadata: dict[str, Any] = super().get_metadata()
metadata.update(
{
"tools": self.tools,
src/cleveragents/core/application.py
+6 -7
View File
@@ -13,7 +13,6 @@ import re
import sys
from pathlib import Path
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Type
@@ -123,7 +122,7 @@ class ReactiveCleverAgentsApp: # pylint: disable=too-many-instance-attributes
# Configuration storage
self.config: Optional[ReactiveConfig] = None
self.agents: Dict[str, Agent] = {}
self.agents: dict[str, Agent] = {}
self.template_renderer: Optional[TemplateRenderer] = None
# Result tracking
@@ -360,7 +359,7 @@ class ReactiveCleverAgentsApp: # pylint: disable=too-many-instance-attributes
continue
# Send user input to the stream network
metadata: Dict[str, Any] = {"context": self.config.global_context}
metadata: dict[str, Any] = {"context": self.config.global_context}
metadata["_unsafe_mode"] = self.unsafe
# Create a new future for this message
@@ -389,7 +388,7 @@ class ReactiveCleverAgentsApp: # pylint: disable=too-many-instance-attributes
f"Failed to start interactive session: {str(e)}"
) from e
def _config_to_dict(self) -> Dict[str, Any]:
def _config_to_dict(self) -> dict[str, Any]:
"""Convert reactive config to dictionary format for agent factory."""
if not self.config:
return {}
@@ -498,7 +497,7 @@ class ReactiveCleverAgentsApp: # pylint: disable=too-many-instance-attributes
# Register built-in agent types
registered_types = self.agent_factory.get_agent_types()
type_map: Dict[str, Type[Agent]] = {
type_map: dict[str, Type[Agent]] = {
"llm": LLMAgent,
"tool": ToolAgent,
}
@@ -859,7 +858,7 @@ class ReactiveCleverAgentsApp: # pylint: disable=too-many-instance-attributes
print(f"Stream '{stream_name}' not found")
return
metadata: Dict[str, Any] = {
metadata: dict[str, Any] = {
"context": self.config.global_context if self.config else {}
}
metadata["_unsafe_mode"] = self.unsafe
@@ -905,7 +904,7 @@ class ReactiveCleverAgentsApp: # pylint: disable=too-many-instance-attributes
if not self.config:
raise CleverAgentsException("Configuration not loaded")
metadata: Dict[str, Any] = {"context": self.config.global_context}
metadata: dict[str, Any] = {"context": self.config.global_context}
metadata["_unsafe_mode"] = self.unsafe
result = await graph.execute(
src/cleveragents/core/config.py
+7 -8
View File
@@ -13,7 +13,6 @@ import re
from pathlib import Path
from re import Match
from typing import Any
from typing import Dict
from typing import List
import yaml
@@ -31,7 +30,7 @@ class ConfigurationManager:
configuration files.
Attributes:
config (Dict[str, Any]): The complete configuration dictionary.
config (dict[str, Any]): The complete configuration dictionary.
schema_validator (SchemaValidator): Validator for configuration schema.
"""
@@ -41,7 +40,7 @@ class ConfigurationManager:
Sets up an empty configuration and initializes the schema validator.
"""
self.config: Dict[str, Any] = {}
self.config: dict[str, Any] = {}
self.schema_validator = SchemaValidator()
def load_files(self, config_files: List[Path]) -> None:
@@ -58,7 +57,7 @@ class ConfigurationManager:
Raises:
ConfigurationError: If a file cannot be loaded or parsed.
"""
merged_config: Dict[str, Any] = {}
merged_config: dict[str, Any] = {}
for file_path in config_files:
try:
@@ -89,8 +88,8 @@ class ConfigurationManager:
self.config = merged_config
def _deep_merge(
self, dict1: Dict[str, Any], dict2: Dict[str, Any]
) -> Dict[str, Any]:
self, dict1: dict[str, Any], dict2: dict[str, Any]
) -> dict[str, Any]:
"""
Deep merge two dictionaries.
@@ -249,7 +248,7 @@ class ConfigurationManager:
return float(config)
return config
def to_dict(self) -> Dict[str, Any]:
def to_dict(self) -> dict[str, Any]:
"""
Convert the configuration to a dictionary.
@@ -284,7 +283,7 @@ class SchemaValidator: # pylint: disable=too-few-public-methods
"""
# Schema will be defined here
def validate(self, config: Dict[str, Any]) -> None:
def validate(self, config: dict[str, Any]) -> None:
"""
Validate a configuration dictionary against the schema.
src/cleveragents/langgraph/bridge.py
+9 -10
View File
@@ -6,7 +6,6 @@ import asyncio
import logging
from typing import Any
from typing import Callable
from typing import Dict
from typing import List
from typing import Optional
@@ -38,7 +37,7 @@ class RxPyLangGraphBridge:
"""Initialize the bridge."""
self.stream_router = stream_router
self.logger = logging.getLogger(__name__)
self.graphs: Dict[str, LangGraph] = {}
self.graphs: dict[str, LangGraph] = {}
# Register custom operators
self._register_langgraph_operators()
@@ -68,7 +67,7 @@ class RxPyLangGraphBridge:
# Store operator factory function
setattr(self, f"_operator_{name}", factory_func)
def create_graph_from_config(self, config: Dict[str, Any]) -> LangGraph:
def create_graph_from_config(self, config: dict[str, Any]) -> LangGraph:
"""Create a LangGraph from configuration."""
# Parse graph configuration
graph_config = GraphConfig(
@@ -141,7 +140,7 @@ class RxPyLangGraphBridge:
return stream_config
def _create_graph_executor(self, params: Dict[str, Any]) -> Any:
def _create_graph_executor(self, params: dict[str, Any]) -> Any:
"""Create an operator that executes a LangGraph."""
graph_name = params.get("graph")
if not graph_name or graph_name not in self.graphs:
@@ -190,7 +189,7 @@ class RxPyLangGraphBridge:
return ops.flat_map(lambda msg: rx.from_future(create_future_task(msg)))
def _create_state_updater(self, params: Dict[str, Any]) -> Any:
def _create_state_updater(self, params: dict[str, Any]) -> Any:
"""Create an operator that updates graph state."""
graph_name = params.get("graph")
@@ -216,7 +215,7 @@ class RxPyLangGraphBridge:
return ops.map(update_state)
def _create_state_checkpointer(self, params: Dict[str, Any]) -> Any:
def _create_state_checkpointer(self, params: dict[str, Any]) -> Any:
"""Create an operator that checkpoints graph state."""
graph_name = params.get("graph")
@@ -239,7 +238,7 @@ class RxPyLangGraphBridge:
return ops.map(checkpoint_state)
def _create_node_operator(self, params: Dict[str, Any]) -> Any:
def _create_node_operator(self, params: dict[str, Any]) -> Any:
"""Create an operator that executes a specific LangGraph node."""
graph_name = params.get("graph")
node_name = params.get("node")
@@ -293,7 +292,7 @@ class RxPyLangGraphBridge:
return ops.flat_map(lambda msg: rx.from_future(create_future_task(msg)))
def _create_conditional_router(
self, params: Dict[str, Any]
self, params: dict[str, Any]
) -> Callable[[Observable], Observable]:
"""Create an operator that routes based on LangGraph conditions."""
routes = params.get("routes", {})
@@ -325,7 +324,7 @@ class RxPyLangGraphBridge:
return router
def _evaluate_route_condition(
self, msg: StreamMessage, condition: Dict[str, Any]
self, msg: StreamMessage, condition: dict[str, Any]
) -> bool:
"""Evaluate a routing condition."""
condition_type = condition.get("type", "always")
@@ -397,7 +396,7 @@ class RxPyLangGraphBridge:
observer = Observer(on_next=on_state_update)
graph.state_manager.get_state_observable().subscribe(observer)
def create_hybrid_pipeline(self, config: Dict[str, Any]) -> None:
def create_hybrid_pipeline(self, config: dict[str, Any]) -> None:
"""
Create a hybrid pipeline that combines RxPy streams and LangGraph nodes.
src/cleveragents/langgraph/graph.py
+9 -10
View File
@@ -10,7 +10,6 @@ from dataclasses import dataclass
from dataclasses import field
from pathlib import Path
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Set
@@ -38,7 +37,7 @@ class GraphConfig: # pylint: disable=too-many-instance-attributes
"""Configuration for a LangGraph."""
name: str
nodes: Dict[str, NodeConfig] = field(default_factory=dict)
nodes: dict[str, NodeConfig] = field(default_factory=dict)
edges: List[Edge] = field(default_factory=list)
entry_point: str = "start"
state_class: Optional[type] = None
@@ -46,7 +45,7 @@ class GraphConfig: # pylint: disable=too-many-instance-attributes
checkpoint_dir: Optional[Path] = None
enable_time_travel: bool = False
parallel_execution: bool = True
metadata: Dict[str, Any] = field(default_factory=dict)
metadata: dict[str, Any] = field(default_factory=dict)
class LangGraph: # pylint: disable=too-many-instance-attributes
@@ -60,7 +59,7 @@ class LangGraph: # pylint: disable=too-many-instance-attributes
def __init__(
self,
config: GraphConfig,
agents: Optional[Dict[str, Agent]] = None,
agents: Optional[dict[str, Agent]] = None,
stream_router: Optional[ReactiveStreamRouter] = None,
scheduler: Optional[AsyncIOScheduler] = None,
):
@@ -85,7 +84,7 @@ class LangGraph: # pylint: disable=too-many-instance-attributes
self.stream_router = stream_router or ReactiveStreamRouter(self.scheduler)
# Initialize nodes
self.nodes: Dict[str, Node] = {}
self.nodes: dict[str, Node] = {}
self._initialize_nodes()
# State management
@@ -132,7 +131,7 @@ class LangGraph: # pylint: disable=too-many-instance-attributes
stream_name = f"__{self.name}_node_{node_name}__"
# Create stream with node execution operator
operators: List[Dict[str, Any]] = [
operators: List[dict[str, Any]] = [
{"type": "map", "params": {"function": f"execute_node_{node_name}"}}
]
@@ -232,8 +231,8 @@ class LangGraph: # pylint: disable=too-many-instance-attributes
def _analyze_graph(self) -> None:
"""Analyze graph structure for validation and optimization."""
# Build adjacency lists
self.adjacency_list: Dict[str, List[str]] = defaultdict(list)
self.reverse_adjacency_list: Dict[str, List[str]] = defaultdict(list)
self.adjacency_list: dict[str, List[str]] = defaultdict(list)
self.reverse_adjacency_list: dict[str, List[str]] = defaultdict(list)
for edge in self.config.edges:
self.adjacency_list[edge.source].append(edge.target)
@@ -293,7 +292,7 @@ class LangGraph: # pylint: disable=too-many-instance-attributes
return parallel_groups
def _topological_levels(self) -> Dict[int, Set[str]]:
def _topological_levels(self) -> dict[int, Set[str]]:
"""Compute topological levels for the graph."""
in_degree = {node: 0 for node in self.nodes}
@@ -359,7 +358,7 @@ class LangGraph: # pylint: disable=too-many-instance-attributes
return visited
async def execute(self, input_data: Optional[Dict[str, Any]] = None) -> GraphState:
async def execute(self, input_data: Optional[dict[str, Any]] = None) -> GraphState:
"""Execute the graph with optional input data."""
if self.is_running:
raise RuntimeError("Graph is already running")
src/cleveragents/langgraph/nodes.py
+12 -13
View File
@@ -8,7 +8,6 @@ from dataclasses import dataclass
from dataclasses import field
from enum import Enum
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
@@ -38,12 +37,12 @@ class NodeConfig: # pylint: disable=too-many-instance-attributes
agent: Optional[str] = None
function: Optional[str] = None
tools: List[str] = field(default_factory=list)
retry_policy: Optional[Dict[str, Any]] = None
retry_policy: Optional[dict[str, Any]] = None
timeout: Optional[float] = None
parallel: bool = False
condition: Optional[Dict[str, Any]] = None
condition: Optional[dict[str, Any]] = None
subgraph: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass
@@ -52,8 +51,8 @@ class Edge:
source: str
target: str
condition: Optional[Dict[str, Any]] = None
metadata: Dict[str, Any] = field(default_factory=dict)
condition: Optional[dict[str, Any]] = None
metadata: dict[str, Any] = field(default_factory=dict)
class Node: # pylint: disable=too-many-instance-attributes
@@ -63,7 +62,7 @@ class Node: # pylint: disable=too-many-instance-attributes
Nodes can be agents, functions, tools, conditionals, or subgraphs.
"""
def __init__(self, config: NodeConfig, agents: Optional[Dict[str, Agent]] = None):
def __init__(self, config: NodeConfig, agents: Optional[dict[str, Agent]] = None):
"""Initialize node."""
self.config = config
self.name = config.name
@@ -78,7 +77,7 @@ class Node: # pylint: disable=too-many-instance-attributes
async def execute( # pylint: disable=too-many-branches
self, state: GraphState
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Execute the node and return state updates."""
self.execution_count += 1
start_time = asyncio.get_event_loop().time()
@@ -138,7 +137,7 @@ class Node: # pylint: disable=too-many-instance-attributes
finally:
self.last_execution_time = asyncio.get_event_loop().time() - start_time
async def _execute_agent(self, state: GraphState) -> Dict[str, Any]:
async def _execute_agent(self, state: GraphState) -> dict[str, Any]:
"""Execute an agent node."""
if not self.config.agent:
raise ValueError(f"Agent node {self.name} has no agent specified")
@@ -197,7 +196,7 @@ class Node: # pylint: disable=too-many-instance-attributes
]
}
async def _execute_function(self, state: GraphState) -> Dict[str, Any]:
async def _execute_function(self, state: GraphState) -> dict[str, Any]:
"""Execute a function node."""
if not self.config.function:
raise ValueError(f"Function node {self.name} has no function specified")
@@ -227,7 +226,7 @@ class Node: # pylint: disable=too-many-instance-attributes
return {}
async def _execute_tool(self) -> Dict[str, Any]:
async def _execute_tool(self) -> dict[str, Any]:
"""Execute a tool node."""
if not self.config.tools:
return {}
@@ -245,7 +244,7 @@ class Node: # pylint: disable=too-many-instance-attributes
async def _execute_conditional( # pylint: disable=too-many-branches,too-many-statements
self, state: GraphState
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Execute a conditional node."""
if not self.config.condition:
return {"metadata": {"condition_result": True}}
@@ -316,7 +315,7 @@ class Node: # pylint: disable=too-many-instance-attributes
return {"metadata": {"condition_result": result}}
async def _execute_subgraph(self, _state: GraphState) -> Dict[str, Any]:
async def _execute_subgraph(self, _state: GraphState) -> dict[str, Any]:
"""Execute a subgraph node."""
if not self.config.subgraph:
raise ValueError(f"Subgraph node {self.name} has no subgraph specified")
src/cleveragents/langgraph/state.py
+8 -9
View File
@@ -10,7 +10,6 @@ from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Type
@@ -34,10 +33,10 @@ class StateUpdateMode(Enum):
class StateSnapshot:
"""Snapshot of graph state at a point in time."""
state: Dict[str, Any]
state: dict[str, Any]
timestamp: datetime
node_id: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass
@@ -49,14 +48,14 @@ class GraphState:
Users can extend this class to define custom state schemas.
"""
messages: List[Dict[str, Any]] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
messages: List[dict[str, Any]] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
current_node: Optional[str] = None
execution_count: int = 0
error: Optional[str] = None
def update( # pylint: disable=too-many-branches
self, updates: Dict[str, Any], mode: StateUpdateMode = StateUpdateMode.MERGE
self, updates: dict[str, Any], mode: StateUpdateMode = StateUpdateMode.MERGE
) -> None:
"""Update state based on mode."""
if mode == StateUpdateMode.REPLACE:
@@ -83,7 +82,7 @@ class GraphState:
else:
current.append(value)
def to_dict(self) -> Dict[str, Any]:
def to_dict(self) -> dict[str, Any]:
"""Convert state to dictionary."""
return {
"messages": self.messages,
@@ -94,7 +93,7 @@ class GraphState:
}
@classmethod
def from_dict(cls: Type[T], data: Dict[str, Any]) -> T:
def from_dict(cls: Type[T], data: dict[str, Any]) -> T:
"""Create state from dictionary."""
return cls(**data)
@@ -139,7 +138,7 @@ class StateManager: # pylint: disable=too-many-instance-attributes
def update_state(
self,
updates: Dict[str, Any],
updates: dict[str, Any],
mode: StateUpdateMode = StateUpdateMode.MERGE,
node_id: Optional[str] = None,
) -> GraphState:
src/cleveragents/network.py
+1 -2
View File
@@ -7,7 +7,6 @@ management, and execution of an agent network.
from pathlib import Path
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
@@ -40,7 +39,7 @@ class AgentNetwork: # pylint: disable=too-few-public-methods
self.agent_factory: Optional[AgentFactory] = None
async def process(
self, message: str, context: Optional[Dict[Any, Any]] = None
self, message: str, context: Optional[dict[Any, Any]] = None
) -> str:
"""
Process a single message through the agent network.
src/cleveragents/reactive/config_parser.py
+23 -24
View File
@@ -14,7 +14,6 @@ from dataclasses import field
from pathlib import Path
from re import Match
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
@@ -34,7 +33,7 @@ class AgentConfig:
name: str
type: str
config: Dict[str, Any] = field(default_factory=dict)
config: dict[str, Any] = field(default_factory=dict)
@dataclass
@@ -42,16 +41,16 @@ class LangGraphConfig: # pylint: disable=too-many-instance-attributes
"""Configuration for a LangGraph."""
name: str
nodes: Dict[str, Dict[str, Any]] = field(default_factory=dict)
edges: List[Dict[str, Any]] = field(default_factory=list)
nodes: dict[str, dict[str, Any]] = field(default_factory=dict)
edges: List[dict[str, Any]] = field(default_factory=list)
entry_point: str = "start"
checkpointing: bool = False
checkpoint_dir: Optional[str] = None
enable_time_travel: bool = False
parallel_execution: bool = True
state_class: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
template_config: Optional[Dict[str, Any]] = None # Template configuration
metadata: dict[str, Any] = field(default_factory=dict)
template_config: Optional[dict[str, Any]] = None # Template configuration
@dataclass
@@ -59,28 +58,28 @@ class HybridPipelineConfig:
"""Configuration for hybrid RxPy-LangGraph pipelines."""
name: str
stages: List[Dict[str, Any]] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
stages: List[dict[str, Any]] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass
class ReactiveConfig: # pylint: disable=too-many-instance-attributes
"""Complete reactive configuration."""
agents: Dict[str, AgentConfig] = field(default_factory=dict)
routes: Dict[str, RouteConfig] = field(default_factory=dict) # Unified routes
merges: List[Dict[str, Any]] = field(default_factory=list)
splits: List[Dict[str, Any]] = field(default_factory=list)
pipelines: Dict[str, HybridPipelineConfig] = field(default_factory=dict)
templates: Dict[str, Dict[str, Any]] = field(
agents: dict[str, AgentConfig] = field(default_factory=dict)
routes: dict[str, RouteConfig] = field(default_factory=dict) # Unified routes
merges: List[dict[str, Any]] = field(default_factory=list)
splits: List[dict[str, Any]] = field(default_factory=list)
pipelines: dict[str, HybridPipelineConfig] = field(default_factory=dict)
templates: dict[str, dict[str, Any]] = field(
default_factory=dict
) # New: template definitions
instances: Dict[str, Dict[str, Any]] = field(
instances: dict[str, dict[str, Any]] = field(
default_factory=dict
) # New: template instances
global_context: Dict[str, Any] = field(default_factory=dict)
global_context: dict[str, Any] = field(default_factory=dict)
template_engine: str = "JINJA2"
prompts: Dict[str, Any] = field(default_factory=dict)
prompts: dict[str, Any] = field(default_factory=dict)
class ReactiveConfigParser: # pylint: disable=too-few-public-methods
@@ -91,8 +90,8 @@ class ReactiveConfigParser: # pylint: disable=too-few-public-methods
def parse_files(self, config_files: List[Path]) -> ReactiveConfig:
"""Parse configuration from multiple files."""
combined_config: Dict[str, Any] = {}
_all_template_sections: Dict[str, Any] = {} # Reserved for future use
combined_config: dict[str, Any] = {}
_all_template_sections: dict[str, Any] = {} # Reserved for future use
for file_path in config_files:
self.logger.info("Loading configuration from %s", file_path)
@@ -118,7 +117,7 @@ class ReactiveConfigParser: # pylint: disable=too-few-public-methods
return self._build_reactive_config(combined_config)
def _merge_configs(self, base: Dict[str, Any], new: Dict[str, Any]) -> None:
def _merge_configs(self, base: dict[str, Any], new: dict[str, Any]) -> None:
"""Merge configuration dictionaries."""
if new is None:
return
@@ -175,7 +174,7 @@ class ReactiveConfigParser: # pylint: disable=too-few-public-methods
return config
def _build_reactive_config( # pylint: disable=too-many-locals,too-many-branches
self, config_dict: Dict[str, Any]
self, config_dict: dict[str, Any]
) -> ReactiveConfig:
"""Build reactive configuration from dictionary."""
reactive_config = ReactiveConfig()
@@ -289,7 +288,7 @@ class ReactiveConfigParser: # pylint: disable=too-few-public-methods
self._validate_config(reactive_config)
return reactive_config
def _parse_stream_route(self, name: str, route_data: Dict[str, Any]) -> RouteConfig:
def _parse_stream_route(self, name: str, route_data: dict[str, Any]) -> RouteConfig:
"""Parse a stream-type route."""
stream_type = StreamType(route_data.get("stream_type", "cold"))
@@ -320,7 +319,7 @@ class ReactiveConfigParser: # pylint: disable=too-few-public-methods
metadata=route_data.get("metadata", {}),
)
def _parse_graph_route(self, name: str, route_data: Dict[str, Any]) -> RouteConfig:
def _parse_graph_route(self, name: str, route_data: dict[str, Any]) -> RouteConfig:
"""Parse a graph-type route."""
# Parse bridge config if present
bridge_config = None
@@ -350,7 +349,7 @@ class ReactiveConfigParser: # pylint: disable=too-few-public-methods
metadata=route_data.get("metadata", {}),
)
def _parse_bridge_route(self, name: str, route_data: Dict[str, Any]) -> RouteConfig:
def _parse_bridge_route(self, name: str, route_data: dict[str, Any]) -> RouteConfig:
"""Parse a bridge-type route."""
# Bridge routes are special - they define conversion logic
bridge_config = BridgeConfig(
src/cleveragents/reactive/route.py
+13 -14
View File
@@ -11,7 +11,6 @@ from dataclasses import dataclass
from dataclasses import field
from enum import Enum
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
@@ -37,9 +36,9 @@ class BridgeConfig:
"""Configuration for type conversion/bridging between routes."""
# When to upgrade from stream to graph
upgrade_conditions: Dict[str, Any] = field(default_factory=dict)
upgrade_conditions: dict[str, Any] = field(default_factory=dict)
# When to downgrade from graph to stream
downgrade_conditions: Dict[str, Any] = field(default_factory=dict)
downgrade_conditions: dict[str, Any] = field(default_factory=dict)
# State extraction for stream->graph conversion
state_extractor: Optional[str] = None
# State flattener for graph->stream conversion
@@ -59,7 +58,7 @@ class RouteConfig: # pylint: disable=too-many-instance-attributes
# Stream-specific config (used when type=STREAM)
stream_type: Optional[StreamType] = None
operators: List[Dict[str, Any]] = field(default_factory=list)
operators: List[dict[str, Any]] = field(default_factory=list)
subscriptions: List[str] = field(default_factory=list)
publications: List[str] = field(default_factory=list)
agents: List[str] = field(default_factory=list)
@@ -67,8 +66,8 @@ class RouteConfig: # pylint: disable=too-many-instance-attributes
buffer_size: int = 1
# Graph-specific config (used when type=GRAPH)
nodes: Dict[str, Dict[str, Any]] = field(default_factory=dict)
edges: List[Dict[str, Any]] = field(default_factory=list)
nodes: dict[str, dict[str, Any]] = field(default_factory=dict)
edges: List[dict[str, Any]] = field(default_factory=list)
entry_point: str = "start"
checkpointing: bool = False
checkpoint_dir: Optional[str] = None
@@ -80,8 +79,8 @@ class RouteConfig: # pylint: disable=too-many-instance-attributes
bridge: Optional[BridgeConfig] = None
# Common metadata
metadata: Dict[str, Any] = field(default_factory=dict)
template_config: Optional[Dict[str, Any]] = None
metadata: dict[str, Any] = field(default_factory=dict)
template_config: Optional[dict[str, Any]] = None
def __post_init__(self) -> None:
"""Validate configuration based on route type."""
@@ -185,7 +184,7 @@ class RouteConfig: # pylint: disable=too-many-instance-attributes
# Convert nodes from NodeConfig to dict
nodes_dict = {}
for name, node_config in graph_config.nodes.items():
node_dict: Dict[str, Any] = {
node_dict: dict[str, Any] = {
"type": node_config.type.value,
"parallel": node_config.parallel,
}
@@ -204,7 +203,7 @@ class RouteConfig: # pylint: disable=too-many-instance-attributes
# Convert edges from Edge to dict
edges_list = []
for edge in graph_config.edges:
edge_dict: Dict[str, Any] = {
edge_dict: dict[str, Any] = {
"source": edge.source,
"target": edge.target,
}
@@ -245,7 +244,7 @@ class RouteComplexityAnalyzer:
"""
@staticmethod
def analyze_route(config: RouteConfig) -> Dict[str, Any]:
def analyze_route(config: RouteConfig) -> dict[str, Any]:
"""Analyze a route and return complexity metrics."""
if config.type == RouteType.STREAM:
return RouteComplexityAnalyzer._analyze_stream(config)
@@ -254,7 +253,7 @@ class RouteComplexityAnalyzer:
return {"complexity": "bridge", "score": 0}
@staticmethod
def _analyze_stream(config: RouteConfig) -> Dict[str, Any]:
def _analyze_stream(config: RouteConfig) -> dict[str, Any]:
"""Analyze stream complexity."""
score = 1 # Base score for streams
features = []
@@ -282,7 +281,7 @@ class RouteComplexityAnalyzer:
}
@staticmethod
def _analyze_graph(config: RouteConfig) -> Dict[str, Any]:
def _analyze_graph(config: RouteConfig) -> dict[str, Any]:
"""Analyze graph complexity."""
score = 5 # Base score for graphs
features = []
@@ -336,7 +335,7 @@ class RouteComplexityAnalyzer:
return "Advanced setup - ensure you need all features"
@staticmethod
def suggest_route_type(requirements: Dict[str, Any]) -> RouteType:
def suggest_route_type(requirements: dict[str, Any]) -> RouteType:
"""Suggest the best route type based on requirements."""
needs_state = requirements.get("needs_state", False)
needs_conditionals = requirements.get("needs_conditionals", False)
src/cleveragents/reactive/route_bridge.py
+3 -4
View File
@@ -8,7 +8,6 @@ stream and graph routes based on runtime conditions.
import asyncio
import logging
from typing import Any
from typing import Dict
from typing import Optional
from rx.scheduler.eventloop import AsyncIOScheduler # type: ignore[attr-defined]
@@ -40,7 +39,7 @@ class RouteBridge:
def __init__(
self,
stream_router: ReactiveStreamRouter,
agents: Dict[str, Agent],
agents: dict[str, Agent],
scheduler: Optional[AsyncIOScheduler] = None,
):
"""Initialize the route bridge."""
@@ -48,7 +47,7 @@ class RouteBridge:
self.agents = agents
self.scheduler = scheduler
self.logger = logging.getLogger(__name__)
self._active_conversions: Dict[str, Any] = {}
self._active_conversions: dict[str, Any] = {}
async def check_upgrade_conditions(
self,
@@ -302,6 +301,6 @@ class RouteBridge:
# For now, return a placeholder
return 0
def get_active_conversion(self, route_name: str) -> Optional[Dict[str, Any]]:
def get_active_conversion(self, route_name: str) -> Optional[dict[str, Any]]:
"""Get active conversion info for a route."""
return self._active_conversions.get(route_name)
src/cleveragents/reactive/stream_router.py
+15 -16
View File
@@ -14,7 +14,6 @@ from dataclasses import field
from enum import Enum
from typing import Any
from typing import Callable
from typing import Dict
from typing import List
from typing import Optional
@@ -43,7 +42,7 @@ class StreamMessage:
"""Message container for reactive streams."""
content: Any
metadata: Dict[str, Any] = field(default_factory=dict)
metadata: dict[str, Any] = field(default_factory=dict)
source_stream: Optional[str] = None
timestamp: Optional[float] = None
@@ -65,13 +64,13 @@ class StreamConfig: # pylint: disable=too-many-instance-attributes
name: str
type: StreamType = StreamType.COLD
operators: List[Dict[str, Any]] = field(default_factory=list)
operators: List[dict[str, Any]] = field(default_factory=list)
subscriptions: List[str] = field(default_factory=list)
publications: List[str] = field(default_factory=list)
agents: List[str] = field(default_factory=list)
initial_value: Optional[Any] = None # For hot streams
buffer_size: int = 1 # For replay streams
template_config: Optional[Dict[str, Any]] = None # Template configuration
template_config: Optional[dict[str, Any]] = None # Template configuration
class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
@@ -95,14 +94,14 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
self.scheduler = scheduler or AsyncIOScheduler(loop)
# Stream storage
self.streams: Dict[str, Any] = (
self.streams: dict[str, Any] = (
{}
) # Can be Subject, BehaviorSubject, or ReplaySubject
self.stream_configs: Dict[str, StreamConfig] = {}
self.observables: Dict[str, ObservableType] = {}
self.stream_configs: dict[str, StreamConfig] = {}
self.observables: dict[str, ObservableType] = {}
# Agent registry
self.agents: Dict[str, Agent] = {}
self.agents: dict[str, Agent] = {}
# Subscription tracking
self.subscriptions: List[Any] = []
@@ -176,7 +175,7 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
return stream
def _build_observable(
self, stream_name: str, operator_configs: List[Dict[str, Any]]
self, stream_name: str, operator_configs: List[dict[str, Any]]
) -> ObservableType:
"""Build an observable with configured operators."""
base_stream = self.streams[stream_name]
@@ -188,7 +187,7 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
return observable
def _create_operator(self, config: Dict[str, Any]) -> Any: # pylint: disable=too-many-return-statements,too-many-branches,too-many-statements,too-many-locals
def _create_operator(self, config: dict[str, Any]) -> Any: # pylint: disable=too-many-return-statements,too-many-branches,too-many-statements,too-many-locals
"""Create an RxPy operator from configuration."""
op_type = config["type"]
params = config.get("params", {})
@@ -399,7 +398,7 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
return mapper
def _apply_transform(
self, msg: StreamMessage, transform: Dict[str, Any]
self, msg: StreamMessage, transform: dict[str, Any]
) -> StreamMessage: # pylint: disable=too-many-return-statements
"""Apply a transformation to a message."""
transform_type = transform.get("type", "identity")
@@ -423,7 +422,7 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
return msg
def _evaluate_condition( # pylint: disable=too-many-return-statements
self, msg: StreamMessage, condition: Dict[str, Any]
self, msg: StreamMessage, condition: dict[str, Any]
) -> bool:
"""Evaluate a filter condition."""
condition_type = condition.get("type", "always")
@@ -450,7 +449,7 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
return True
def _apply_accumulator(
self, acc: Any, msg: StreamMessage, accumulator: Dict[str, Any]
self, acc: Any, msg: StreamMessage, accumulator: dict[str, Any]
) -> Any:
"""Apply an accumulator function."""
acc_type = accumulator.get("type", "collect")
@@ -541,7 +540,7 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
self.logger.info("Merged streams %s into %s", stream_names, output_stream_name)
def split_stream(
self, source_stream_name: str, conditions: Dict[str, Dict[str, Any]]
self, source_stream_name: str, conditions: dict[str, dict[str, Any]]
) -> None:
"""Split a stream into multiple streams based on conditions."""
if source_stream_name not in self.observables:
@@ -560,7 +559,7 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
self.streams[output_name] = output_stream
# Create filtered observable with closure to avoid cell-var-from-loop
def make_filter(cond: Dict[str, Any]) -> Any:
def make_filter(cond: dict[str, Any]) -> Any:
return ops.filter(lambda x: self._evaluate_condition(x, cond))
filter_op = make_filter(condition)
@@ -576,7 +575,7 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
)
def send_message(
self, stream_name: str, content: Any, metadata: Optional[Dict[str, Any]] = None
self, stream_name: str, content: Any, metadata: Optional[dict[str, Any]] = None
) -> None:
"""Send a message to a stream."""
if stream_name not in self.streams:
src/cleveragents/templates/agent_templates.py
+8 -9
View File
@@ -6,7 +6,6 @@ import copy
import logging
from typing import TYPE_CHECKING
from typing import Any
from typing import Dict
from cleveragents.templates.base import BaseTemplate
@@ -25,10 +24,10 @@ class AgentTemplate(BaseTemplate):
def instantiate(
self,
params: Dict[str, Any],
params: dict[str, Any],
registry: "TemplateRegistryProtocol",
context: InstantiationContext,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Create agent configuration from template."""
# Validate and fill defaults
filled_params = self.validate_params(params)
@@ -57,10 +56,10 @@ class CompositeAgentTemplate(BaseTemplate):
def instantiate( # pylint: disable=too-many-locals,too-many-branches
self,
params: Dict[str, Any],
params: dict[str, Any],
registry: "TemplateRegistryProtocol",
context: InstantiationContext,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Create composite agent configuration with instantiated sub-components."""
filled_params = self.validate_params(params)
@@ -76,7 +75,7 @@ class CompositeAgentTemplate(BaseTemplate):
# Process components section
components_config = definition.get("components", {})
instantiated_components: Dict[str, Dict[str, Any]] = {
instantiated_components: dict[str, dict[str, Any]] = {
"agents": {},
"graphs": {},
"streams": {},
@@ -179,10 +178,10 @@ class CompositeAgentTemplate(BaseTemplate):
def _process_graph_definition(
self,
graph_def: Dict[str, Any],
params: Dict[str, Any],
graph_def: dict[str, Any],
params: dict[str, Any],
context: InstantiationContext,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Process a graph definition, resolving agent references."""
# Apply template vars
graph_def = self._apply_template_vars(graph_def, params)
src/cleveragents/templates/base.py
+13 -14
View File
@@ -12,7 +12,6 @@ from dataclasses import field
from enum import Enum
from typing import TYPE_CHECKING
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
@@ -35,7 +34,7 @@ if TYPE_CHECKING:
self,
template_type: "TemplateType",
name: str,
params: Dict[str, Any],
params: dict[str, Any],
context: Optional["InstantiationContext"] = None,
) -> Any:
"""Instantiate a template with parameters."""
@@ -103,7 +102,7 @@ class ComponentReference:
ref_type: str # 'agent', 'graph', 'stream'
ref_name: str
ref_params: Dict[str, Any] = field(default_factory=dict)
ref_params: dict[str, Any] = field(default_factory=dict)
def resolve(self, context: "InstantiationContext") -> Any:
"""Resolve this reference in the given context."""
@@ -115,7 +114,7 @@ class InstantiationContext:
def __init__(self, parent: Optional["InstantiationContext"] = None):
self.parent = parent
self.components: Dict[str, Dict[str, Any]] = {
self.components: dict[str, dict[str, Any]] = {
"agents": {},
"graphs": {},
"streams": {},
@@ -156,7 +155,7 @@ class InstantiationContext:
refs = [f"{ref.ref_type}/{ref.ref_name}" for ref, _ in unresolved]
raise ValueError(f"Cannot resolve references: {refs}")
def get_all_components(self) -> Dict[str, Dict[str, Any]]:
def get_all_components(self) -> dict[str, dict[str, Any]]:
"""Get all components in this context."""
return copy.deepcopy(self.components)
@@ -165,7 +164,7 @@ class BaseTemplate(ABC):
"""Base class for all templates."""
def __init__(
self, name: str, template_type: TemplateType, definition: Dict[str, Any]
self, name: str, template_type: TemplateType, definition: dict[str, Any]
):
self.name = name
self.template_type = template_type
@@ -173,7 +172,7 @@ class BaseTemplate(ABC):
self.parameters = self._parse_parameters()
logger.debug("Created %s template '%s'", template_type.value, name)
def _parse_parameters(self) -> Dict[str, TemplateParameter]:
def _parse_parameters(self) -> dict[str, TemplateParameter]:
"""Parse parameter definitions from template."""
params = {}
params_def = self.definition.get("parameters", {})
@@ -200,7 +199,7 @@ class BaseTemplate(ABC):
return params
def validate_params(self, params: Dict[str, Any]) -> Dict[str, Any]:
def validate_params(self, params: dict[str, Any]) -> dict[str, Any]:
"""Validate and fill in default parameters."""
result = {}
@@ -219,15 +218,15 @@ class BaseTemplate(ABC):
@abstractmethod
def instantiate(
self,
params: Dict[str, Any],
params: dict[str, Any],
registry: "TemplateRegistryProtocol",
context: InstantiationContext,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Create concrete instance from template."""
raise NotImplementedError("Subclasses must implement instantiate()")
def _apply_template_vars( # pylint: disable=too-many-locals,too-many-return-statements,too-many-branches
self, config: Any, params: Dict[str, Any]
self, config: Any, params: dict[str, Any]
) -> Any:
"""Recursively apply Jinja2 template variables."""
if isinstance(config, str):
@@ -290,7 +289,7 @@ class BaseTemplate(ABC):
# Condition was false, exclude this section
return None
# Normal dictionary
result: Dict[Any, Any] = {}
result: dict[Any, Any] = {}
for key, value in config.items():
processed_value = self._apply_template_vars(value, params)
if (
@@ -311,8 +310,8 @@ class BaseTemplate(ABC):
return config
def _merge_params(
self, base_params: Dict[str, Any], override_params: Dict[str, Any]
) -> Dict[str, Any]:
self, base_params: dict[str, Any], override_params: dict[str, Any]
) -> dict[str, Any]:
"""Merge parameter dictionaries with override semantics."""
result = copy.deepcopy(base_params)
src/cleveragents/templates/deferred_template.py
+5 -6
View File
@@ -7,7 +7,6 @@ and process them only during instantiation.
import logging
from typing import Any
from typing import Dict
from typing import Optional
import yaml
@@ -29,7 +28,7 @@ class DeferredTemplate:
self.template_str = template_str
self.processor = YAMLTemplateProcessor()
def render(self, context: Dict[str, Any]) -> Any:
def render(self, context: dict[str, Any]) -> Any:
"""
Render the template with the given context.
@@ -45,7 +44,7 @@ class DeferredTemplate:
@classmethod
def from_yaml_section(
cls, yaml_dict: Dict[str, Any], key: str
cls, yaml_dict: dict[str, Any], key: str
) -> Optional["DeferredTemplate"]:
"""
Create a deferred template from a YAML section if it contains template syntax.
@@ -70,7 +69,7 @@ class DeferredTemplate:
return None
def process_template_definition(definition: Dict[str, Any]) -> Dict[str, Any]:
def process_template_definition(definition: dict[str, Any]) -> dict[str, Any]:
"""
Process a template definition to handle deferred template sections.
@@ -102,8 +101,8 @@ def process_template_definition(definition: Dict[str, Any]) -> Dict[str, Any]:
def apply_deferred_templates(
config: Dict[str, Any], context: Dict[str, Any]
) -> Dict[str, Any]:
config: dict[str, Any], context: dict[str, Any]
) -> dict[str, Any]:
"""
Apply any deferred templates in a configuration.
src/cleveragents/templates/graph_templates.py
+7 -8
View File
@@ -6,7 +6,6 @@ import copy
import logging
from typing import TYPE_CHECKING
from typing import Any
from typing import Dict
from typing import List
from cleveragents.templates.base import BaseTemplate
@@ -24,15 +23,15 @@ class GraphTemplate(BaseTemplate):
def instantiate(
self,
params: Dict[str, Any],
params: dict[str, Any],
registry: "TemplateRegistryProtocol",
context: InstantiationContext,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Create graph configuration from template."""
filled_params = self.validate_params(params)
# Deep copy the definition
graph_def: Dict[str, Any] = copy.deepcopy(self.definition)
graph_def: dict[str, Any] = copy.deepcopy(self.definition)
# Remove parameters section
if "parameters" in graph_def:
@@ -59,10 +58,10 @@ class GraphTemplate(BaseTemplate):
def _process_nodes(
self,
nodes: Dict[str, Any],
params: Dict[str, Any],
nodes: dict[str, Any],
params: dict[str, Any],
context: InstantiationContext,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Process graph nodes, resolving agent references."""
processed_nodes = {}
@@ -100,7 +99,7 @@ class GraphTemplate(BaseTemplate):
return processed_nodes
def _process_edges(self, edges: List[Any], params: Dict[str, Any]) -> List[Any]:
def _process_edges(self, edges: List[Any], params: dict[str, Any]) -> List[Any]:
"""Process edges, handling conditional inclusions."""
processed_edges = []
src/cleveragents/templates/inline_jinja_handler.py
+8 -9
View File
@@ -10,7 +10,6 @@ import re
import uuid
from pathlib import Path
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
@@ -66,7 +65,7 @@ class InlineJinjaHandler:
self,
yaml_content: str,
defer_rendering: bool = True,
context: Optional[Dict[str, Any]] = None,
context: Optional[dict[str, Any]] = None,
) -> Any:
"""
Process YAML content containing inline Jinja2 templates.
@@ -232,8 +231,8 @@ class InlineJinjaHandler:
return parsed
def _render_templates(
self, yaml_content: str, context: Dict[str, Any]
) -> Dict[str, Any]:
self, yaml_content: str, context: dict[str, Any]
) -> dict[str, Any]:
"""
Render all templates immediately with the given context.
"""
@@ -267,8 +266,8 @@ class InlineJinjaHandler:
return result
def apply_templates(
self, config: Dict[str, Any], context: Dict[str, Any]
) -> Dict[str, Any]:
self, config: dict[str, Any], context: dict[str, Any]
) -> dict[str, Any]:
"""
Apply stored templates to a configuration.
@@ -288,14 +287,14 @@ class InlineJinjaHandler:
result = {k: v for k, v in config.items() if k != "__templates__"}
# Apply templates recursively
result_with_templates: Dict[str, Any] = self._apply_templates_recursive(
result_with_templates: dict[str, Any] = self._apply_templates_recursive(
result, templates, context
)
return result_with_templates
def _apply_templates_recursive( # pylint: disable=too-many-nested-blocks
self, data: Any, templates: Dict[str, Any], context: Dict[str, Any]
self, data: Any, templates: dict[str, Any], context: dict[str, Any]
) -> Any:
"""
Recursively apply templates in a data structure.
@@ -353,7 +352,7 @@ class InlineJinjaHandler:
return data
def _render_template_string(
self, template_str: str, context: Dict[str, Any]
self, template_str: str, context: dict[str, Any]
) -> str:
"""Render a single template string."""
full_context = {
src/cleveragents/templates/inline_yaml_jinja.py
+34 -21
View File
@@ -10,10 +10,8 @@ import logging
import re
from pathlib import Path
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import cast
import yaml
from jinja2 import Environment
@@ -64,16 +62,16 @@ class InlineYAMLJinja:
)
def process_file(
self, file_path: Path, context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, file_path: Path, context: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
"""Process a YAML file with inline Jinja2."""
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
return self.process_string(content, context)
def process_string(
self, yaml_content: str, context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, yaml_content: str, context: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
"""
Process YAML content with inline Jinja2 templates.
@@ -86,7 +84,12 @@ class InlineYAMLJinja:
"""
# No templates? Just parse
if not self._has_templates(yaml_content):
return cast(Dict[str, Any], yaml.safe_load(yaml_content))
result = yaml.safe_load(yaml_content)
if not isinstance(result, dict):
raise ValueError(
f"Expected YAML to parse as dict, got {type(result).__name__}"
)
return result
if context is not None:
# Render immediately
@@ -99,8 +102,8 @@ class InlineYAMLJinja:
return "{%" in content or "{{" in content
def _render_and_parse(
self, content: str, context: Dict[str, Any]
) -> Dict[str, Any]:
self, content: str, context: dict[str, Any]
) -> dict[str, Any]:
"""Render templates and parse YAML."""
# First, analyze the content to understand its structure
structure = self._analyze_structure(content)
@@ -119,17 +122,27 @@ class InlineYAMLJinja:
# Parse the rendered YAML
try:
return cast(Dict[str, Any], yaml.safe_load(rendered))
except yaml.YAMLError:
result = yaml.safe_load(rendered)
if not isinstance(result, dict):
raise ValueError(
f"Expected rendered YAML to parse as dict, got {type(result).__name__}"
)
return result
except yaml.YAMLError as exc:
logger.debug("Rendered YAML:\n%s", rendered)
# Try to fix common issues
fixed = self._fix_yaml_issues(rendered)
return cast(Dict[str, Any], yaml.safe_load(fixed))
result = yaml.safe_load(fixed)
if not isinstance(result, dict):
raise ValueError(
f"Expected fixed YAML to parse as dict, got {type(result).__name__}"
) from exc
return result
def _analyze_structure(self, content: str) -> Dict[str, Any]:
def _analyze_structure(self, content: str) -> dict[str, Any]:
"""Analyze content structure for smarter rendering."""
lines = content.split("\n")
structure: Dict[str, Any] = {
structure: dict[str, Any] = {
"has_block_templates": False,
"has_inline_templates": False,
"template_blocks": [],
@@ -153,7 +166,7 @@ class InlineYAMLJinja:
return structure
def _render_structured( # pylint: disable=too-many-locals,too-many-nested-blocks
self, content: str, context: Dict[str, Any]
self, content: str, context: dict[str, Any]
) -> str:
"""
Render content with awareness of YAML structure.
@@ -204,7 +217,7 @@ class InlineYAMLJinja:
def _split_into_sections( # pylint: disable=too-many-nested-blocks
self, content: str
) -> List[Dict[str, Any]]:
) -> List[dict[str, Any]]:
"""Split content into template blocks and regular sections."""
lines = content.split("\n")
sections = []
@@ -275,7 +288,7 @@ class InlineYAMLJinja:
return sections
def _render_block(
self, block_content: str, _base_indent: int, context: Dict[str, Any]
self, block_content: str, _base_indent: int, context: dict[str, Any]
) -> str:
"""
Render a template block ensuring proper YAML structure.
@@ -377,7 +390,7 @@ class InlineYAMLJinja:
return "\n".join(fixed_lines)
def _prepare_context(self, context: Dict[str, Any]) -> Dict[str, Any]:
def _prepare_context(self, context: dict[str, Any]) -> dict[str, Any]:
"""Prepare rendering context with utilities."""
full_context = {
# Built-ins
@@ -407,7 +420,7 @@ class InlineYAMLJinja:
full_context.update(context or {})
return full_context
def _store_for_deferred(self, content: str) -> Dict[str, Any]:
def _store_for_deferred(self, content: str) -> dict[str, Any]:
"""Store content for deferred rendering."""
# For deferred rendering, we store the template as is
# This is the simplest approach that preserves everything
@@ -420,8 +433,8 @@ class InlineYAMLJinja:
}
def render_deferred(
self, config: Dict[str, Any], context: Dict[str, Any]
) -> Dict[str, Any]:
self, config: dict[str, Any], context: dict[str, Any]
) -> dict[str, Any]:
"""Render a deferred template configuration."""
if "__yaml_template__" in config:
template_info = config["__yaml_template__"]
src/cleveragents/templates/jinja_yaml_preprocessor.py
+32 -15
View File
@@ -9,10 +9,8 @@ import logging
import re
from pathlib import Path
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import cast
import yaml
from jinja2 import Environment
@@ -44,16 +42,16 @@ class JinjaYAMLPreprocessor:
)
def load_file(
self, file_path: Path, context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, file_path: Path, context: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
"""Load a YAML file with inline Jinja2 templates."""
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
return self.load_string(content, context)
def load_string(
self, yaml_content: str, context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, yaml_content: str, context: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
"""
Load YAML content with inline Jinja2 templates.
@@ -66,7 +64,12 @@ class JinjaYAMLPreprocessor:
"""
# If no templates, just parse
if "{%" not in yaml_content and "{{" not in yaml_content:
return cast(Dict[str, Any], yaml.safe_load(yaml_content))
result = yaml.safe_load(yaml_content)
if not isinstance(result, dict):
raise ValueError(
f"Expected YAML to parse as dict, got {type(result).__name__}"
)
return result
if context:
# Render immediately
@@ -75,8 +78,8 @@ class JinjaYAMLPreprocessor:
return self._preprocess_for_storage(yaml_content)
def _render_and_parse(
self, yaml_content: str, context: Dict[str, Any]
) -> Dict[str, Any]:
self, yaml_content: str, context: dict[str, Any]
) -> dict[str, Any]:
"""Render templates and parse YAML."""
# Add utilities to context
full_context = {
@@ -94,11 +97,16 @@ class JinjaYAMLPreprocessor:
rendered = template.render(**full_context)
# Parse result
return cast(Dict[str, Any], yaml.safe_load(rendered))
result = yaml.safe_load(rendered)
if not isinstance(result, dict):
raise ValueError(
f"Expected rendered YAML to parse as dict, got {type(result).__name__}"
)
return result
def _preprocess_for_storage( # pylint: disable=too-many-locals,too-many-branches,too-many-statements,too-many-nested-blocks
self, yaml_content: str
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""
Preprocess YAML for storage with templates preserved.
@@ -202,11 +210,15 @@ class JinjaYAMLPreprocessor:
logger.debug("Preprocessed:\n%s", preprocessed)
raise
return cast(Dict[str, Any], parsed)
if not isinstance(parsed, dict):
raise ValueError(
f"Expected preprocessed YAML to parse as dict, got {type(parsed).__name__}"
)
return parsed
def render_deferred(
self, config: Dict[str, Any], context: Dict[str, Any]
) -> Dict[str, Any]:
self, config: dict[str, Any], context: dict[str, Any]
) -> dict[str, Any]:
"""
Render a configuration that contains deferred templates.
@@ -233,7 +245,12 @@ class JinjaYAMLPreprocessor:
# Check if this is a template value
if "__is_template__" in data and "__value__" in data:
# Inline template
return cast(str, data["__value__"])
value = data["__value__"]
if not isinstance(value, str):
raise ValueError(
f"Expected template value to be str, got {type(value).__name__}"
)
return value
# Check if this is a template block
if "__jinja_template__" in data:
src/cleveragents/templates/loaders.py
+2 -3
View File
@@ -8,7 +8,6 @@ directories, or direct strings.
import os
from pathlib import Path
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
@@ -193,10 +192,10 @@ class ConfigTemplateLoader(TemplateLoader): # pylint: disable=too-few-public-me
Attributes:
renderer (TemplateRenderer): The template renderer to register templates with.
config (Dict[str, Any]): The configuration dictionary containing templates.
config (dict[str, Any]): The configuration dictionary containing templates.
"""
def __init__(self, renderer: TemplateRenderer, config: Dict[str, Any]):
def __init__(self, renderer: TemplateRenderer, config: dict[str, Any]):
"""
Initialize the configuration template loader.
src/cleveragents/templates/registry.py
+5 -6
View File
@@ -5,7 +5,6 @@ Template registry for managing all template types.
import logging
from typing import TYPE_CHECKING
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
@@ -23,7 +22,7 @@ class TemplateRegistry:
"""Registry for all template types."""
def __init__(self) -> None:
self.templates: Dict[TemplateType, Dict[str, BaseTemplate]] = {
self.templates: dict[TemplateType, dict[str, BaseTemplate]] = {
TemplateType.AGENT: {},
TemplateType.GRAPH: {},
TemplateType.STREAM: {},
@@ -31,7 +30,7 @@ class TemplateRegistry:
logger.debug("Initialized template registry")
def register_template(
self, template_type: TemplateType, name: str, definition: Dict[str, Any]
self, template_type: TemplateType, name: str, definition: dict[str, Any]
) -> None:
"""Register a new template."""
# Import template classes here to avoid circular imports
@@ -74,7 +73,7 @@ class TemplateRegistry:
self,
template_type: TemplateType,
name: str,
params: Dict[str, Any],
params: dict[str, Any],
context: Optional[InstantiationContext] = None,
) -> Any:
"""Instantiate a template by type and name (protocol compliance)."""
@@ -85,7 +84,7 @@ class TemplateRegistry:
return template.instantiate(params, self, context)
def instantiate_from_config( # pylint: disable=too-many-return-statements
self, config: Dict[str, Any], context: Optional[InstantiationContext] = None
self, config: dict[str, Any], context: Optional[InstantiationContext] = None
) -> Any:
"""Create an instance from configuration."""
# Handle None config
@@ -157,7 +156,7 @@ class TemplateRegistry:
def list_templates(
self, template_type: Optional[TemplateType] = None
) -> Dict[str, List[str]]:
) -> dict[str, List[str]]:
"""List all registered templates."""
if template_type:
return {template_type.value: list(self.templates[template_type].keys())}
src/cleveragents/templates/renderer.py
+7 -8
View File
@@ -11,7 +11,6 @@ import re
from enum import Enum
from typing import Any
from typing import Callable
from typing import Dict
from typing import List
from typing import Mapping
from typing import Optional
@@ -40,7 +39,7 @@ class Jinja2Environment(Protocol): # pylint: disable=too-few-public-methods
class PystacheRenderer(Protocol): # pylint: disable=too-few-public-methods
"""Protocol for Pystache renderer objects."""
def render(self, template: str, context: Dict[str, Any]) -> str:
def render(self, template: str, context: dict[str, Any]) -> str:
"""Render the template with the given context."""
@@ -52,7 +51,7 @@ class TemplateEngine(Enum):
MUSTACHE = "mustache" # Requires pystache package
def _resolve_path(path: str, ctx: Dict[str, Any]) -> str:
def _resolve_path(path: str, ctx: dict[str, Any]) -> str:
"""Return ctx value addressed by a dotted *path* (e.g. 'user.name')."""
cur: Any = ctx
for part in path.split("."):
@@ -69,7 +68,7 @@ class TemplateRenderer:
Attributes:
engine_type (TemplateEngine): The template engine to use.
templates (Dict[str, str]): Dictionary of registered templates.
templates (dict[str, str]): Dictionary of registered templates.
engine: The actual template engine instance.
"""
@@ -84,7 +83,7 @@ class TemplateRenderer:
TemplateError: If the specified engine is not available.
"""
self.engine_type = engine_type
self.templates: Dict[str, Any] = {}
self.templates: dict[str, Any] = {}
self.engine: Union[Callable[..., Any], Jinja2Environment, PystacheRenderer, None] = None
self._initialize_engine()
@@ -125,7 +124,7 @@ class TemplateRenderer:
raise TemplateError(f"Unsupported template engine: {self.engine_type}")
@staticmethod
def _render_simple_with_jinja_like(template: str, context: Dict[str, Any]) -> str:
def _render_simple_with_jinja_like(template: str, context: dict[str, Any]) -> str:
"""
Render the template string replacing {{ ... }} placeholders using a
very small Jinja-like subset.
@@ -171,7 +170,7 @@ class TemplateRenderer:
# For other engines, just store the template string
self.templates[name] = template_str
def render(self, template_name: str, context: Dict[str, Any]) -> str:
def render(self, template_name: str, context: dict[str, Any]) -> str:
"""
Render a template with the given context.
@@ -219,7 +218,7 @@ class TemplateRenderer:
def render_string(
self,
template_str: str,
context: Dict[str, Any],
context: dict[str, Any],
source_description: Optional[str] = None,
) -> str:
"""
src/cleveragents/templates/smart_yaml_loader.py
+12 -13
View File
@@ -9,7 +9,6 @@ import logging
import re
from pathlib import Path
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Tuple
@@ -30,7 +29,7 @@ class SmartYAMLLoader:
def __init__(self) -> None:
self.template_pattern = re.compile(r"{[{%].*?[%}]}")
def load_file(self, file_path: Path) -> Tuple[Dict[str, Any], Dict[str, str]]:
def load_file(self, file_path: Path) -> Tuple[dict[str, Any], dict[str, str]]:
"""
Load a YAML file, separating regular content from template sections.
@@ -47,7 +46,7 @@ class SmartYAMLLoader:
def load_string( # pylint: disable=too-many-locals,too-many-branches,too-many-statements,too-many-nested-blocks
self, yaml_content: str
) -> Tuple[Dict[str, Any], Dict[str, str]]:
) -> Tuple[dict[str, Any], dict[str, str]]:
"""
Load YAML content, separating regular content from template sections.
@@ -63,7 +62,7 @@ class SmartYAMLLoader:
# Process the content to extract template sections
lines = yaml_content.split("\n")
template_sections: Dict[str, Any] = {}
template_sections: dict[str, Any] = {}
processed_lines: List[str] = []
current_template_block = []
@@ -177,10 +176,10 @@ class TemplateDefinitionStore:
def __init__(self) -> None:
self.loader = SmartYAMLLoader()
self.definitions: Dict[str, Any] = {}
self.template_sections: Dict[str, str] = {}
self.definitions: dict[str, Any] = {}
self.template_sections: dict[str, str] = {}
def load_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
def load_config(self, config: dict[str, Any]) -> dict[str, Any]:
"""
Process a configuration to extract and store template definitions.
@@ -198,7 +197,7 @@ class TemplateDefinitionStore:
return result
def _process_templates_section(self, templates: Dict[str, Any]) -> Dict[str, Any]:
def _process_templates_section(self, templates: dict[str, Any]) -> dict[str, Any]:
"""
Process the templates section to identify definitions with Jinja2.
@@ -208,7 +207,7 @@ class TemplateDefinitionStore:
Returns:
Processed templates section
"""
result: Dict[str, Dict[str, Any]] = {}
result: dict[str, dict[str, Any]] = {}
for template_type, type_templates in templates.items():
result[template_type] = {}
@@ -243,13 +242,13 @@ class TemplateDefinitionStore:
return any(self._contains_template_markers(item) for item in data)
return False
def get_definition(self, def_id: str) -> Optional[Dict[str, Any]]:
def get_definition(self, def_id: str) -> Optional[dict[str, Any]]:
"""Get a stored template definition."""
return self.definitions.get(def_id)
def render_definition(
self, def_id: str, template_sections: Dict[str, str], context: Dict[str, Any]
) -> Dict[str, Any]:
self, def_id: str, template_sections: dict[str, str], context: dict[str, Any]
) -> dict[str, Any]:
"""
Render a template definition with context.
@@ -276,7 +275,7 @@ class TemplateDefinitionStore:
return processor.process_string(yaml_str, context)
def _reconstruct_yaml( # pylint: disable=too-many-branches,too-many-nested-blocks
self, data: Any, template_sections: Dict[str, str], indent: int = 0
self, data: Any, template_sections: dict[str, str], indent: int = 0
) -> str:
"""
Reconstruct YAML with original template syntax.
src/cleveragents/templates/stream_templates.py
+4 -5
View File
@@ -6,7 +6,6 @@ import copy
import logging
from typing import TYPE_CHECKING
from typing import Any
from typing import Dict
from typing import List
from cleveragents.templates.base import BaseTemplate
@@ -24,15 +23,15 @@ class StreamTemplate(BaseTemplate):
def instantiate(
self,
params: Dict[str, Any],
params: dict[str, Any],
registry: "TemplateRegistryProtocol",
context: InstantiationContext,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Create stream configuration from template."""
filled_params = self.validate_params(params)
# Deep copy the definition
stream_def: Dict[str, Any] = copy.deepcopy(self.definition)
stream_def: dict[str, Any] = copy.deepcopy(self.definition)
# Remove parameters section
if "parameters" in stream_def:
@@ -54,7 +53,7 @@ class StreamTemplate(BaseTemplate):
return stream_def
def _process_operators( # pylint: disable=too-many-branches
self, operators: List[Any], params: Dict[str, Any], context: InstantiationContext
self, operators: List[Any], params: dict[str, Any], context: InstantiationContext
) -> List[Any]:
"""Process stream operators, resolving component references."""
processed_operators = []
src/cleveragents/templates/template_store.py
+25 -15
View File
@@ -7,10 +7,8 @@ Jinja2 syntax intact, processing them only during instantiation.
import logging
from typing import Any
from typing import Dict
from typing import Optional
from typing import Union
from typing import cast
import yaml
@@ -27,21 +25,21 @@ class TemplateStore:
def __init__(self) -> None:
# Store raw template strings by type and name
self.raw_templates: Dict[str, Dict[str, str]] = {
self.raw_templates: dict[str, dict[str, str]] = {
"agents": {},
"graphs": {},
"streams": {},
}
# Store parsed metadata (parameters, type, etc.)
self.metadata: Dict[str, Dict[str, Dict[str, Any]]] = {
self.metadata: dict[str, dict[str, dict[str, Any]]] = {
"agents": {},
"graphs": {},
"streams": {},
}
def add_template(
self, template_type: str, name: str, definition: Union[str, Dict[str, Any]]
self, template_type: str, name: str, definition: Union[str, dict[str, Any]]
) -> None:
"""
Add a template definition.
@@ -67,7 +65,7 @@ class TemplateStore:
self._extract_metadata(template_type, name, definition)
def _extract_metadata(
self, template_type: str, name: str, definition: Dict[str, Any]
self, template_type: str, name: str, definition: dict[str, Any]
) -> None:
"""Extract metadata from template definition."""
metadata = {
@@ -80,13 +78,13 @@ class TemplateStore:
"""Get raw template string."""
return self.raw_templates.get(template_type, {}).get(name)
def get_metadata(self, template_type: str, name: str) -> Optional[Dict[str, Any]]:
def get_metadata(self, template_type: str, name: str) -> Optional[dict[str, Any]]:
"""Get template metadata."""
return self.metadata.get(template_type, {}).get(name)
def instantiate_template(
self, template_type: str, name: str, params: Dict[str, Any]
) -> Dict[str, Any]:
self, template_type: str, name: str, params: dict[str, Any]
) -> dict[str, Any]:
"""
Instantiate a template with parameters.
@@ -134,7 +132,7 @@ class TemplateDefinition:
"""
def __init__(
self, definition: Union[str, Dict[str, Any]], contains_templates: bool = False
self, definition: Union[str, dict[str, Any]], contains_templates: bool = False
):
"""
Initialize template definition.
@@ -163,15 +161,23 @@ class TemplateDefinition:
"""Check if the definition contains Jinja2 syntax."""
return "{%" in self.raw_yaml or "{{" in self.raw_yaml
def get_parameters(self) -> Dict[str, Any]:
def get_parameters(self) -> dict[str, Any]:
"""Get template parameters."""
return cast(Dict[str, Any], self.parsed.get("parameters", {}))
params = self.parsed.get("parameters", {})
if not isinstance(params, dict):
raise ValueError(
f"Expected parameters to be dict, got {type(params).__name__}"
)
return params
def get_type(self) -> str:
"""Get template type."""
return cast(str, self.parsed.get("type", "unknown"))
type_val = self.parsed.get("type", "unknown")
if not isinstance(type_val, str):
return "unknown"
return type_val
def instantiate(self, params: Dict[str, Any]) -> Dict[str, Any]:
def instantiate(self, params: dict[str, Any]) -> dict[str, Any]:
"""
Instantiate the template with parameters.
@@ -183,7 +189,11 @@ class TemplateDefinition:
"""
if not self.contains_templates:
# No templates, return parsed version
return cast(Dict[str, Any], self.parsed)
if not isinstance(self.parsed, dict):
raise ValueError(
f"Expected parsed to be dict, got {type(self.parsed).__name__}"
)
return self.parsed
from cleveragents.templates.yaml_preprocessor import ( # pylint: disable=import-outside-toplevel
YAMLTemplateProcessor,
src/cleveragents/templates/yaml_jinja_loader.py
+35 -23
View File
@@ -9,10 +9,8 @@ import logging
import re
from pathlib import Path
from typing import Any
from typing import Dict
from typing import Optional
from typing import Tuple
from typing import cast
import yaml
from jinja2 import Environment
@@ -41,8 +39,8 @@ class YAMLJinjaLoader:
)
def load_file(
self, file_path: Path, context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, file_path: Path, context: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
"""
Load a YAML file that may contain Jinja2 templates.
@@ -59,8 +57,8 @@ class YAMLJinjaLoader:
return self.load_string(content, context)
def load_string(
self, yaml_content: str, context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, yaml_content: str, context: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
"""
Load YAML content that may contain Jinja2 templates.
@@ -103,11 +101,16 @@ class YAMLJinjaLoader:
# No context, need to defer rendering
return self._defer_template_rendering(yaml_content)
# No templates, parse normally
return cast(Dict[str, Any], yaml.safe_load(yaml_content))
result = yaml.safe_load(yaml_content)
if not isinstance(result, dict):
raise ValueError(
f"Expected YAML to parse as dict, got {type(result).__name__}"
)
return result
def _render_and_parse(
self, yaml_content: str, context: Dict[str, Any]
) -> Dict[str, Any]:
self, yaml_content: str, context: dict[str, Any]
) -> dict[str, Any]:
"""
Render Jinja2 templates and parse the result as YAML.
@@ -124,12 +127,17 @@ class YAMLJinjaLoader:
rendered = template.render(**context)
# Parse the rendered YAML
return cast(Dict[str, Any], yaml.safe_load(rendered))
result = yaml.safe_load(rendered)
if not isinstance(result, dict):
raise ValueError(
f"Expected rendered YAML to parse as dict, got {type(result).__name__}"
)
return result
except Exception as e:
logger.error("Failed to render template: %s", e)
raise
def _defer_template_rendering(self, yaml_content: str) -> Dict[str, Any]:
def _defer_template_rendering(self, yaml_content: str) -> dict[str, Any]:
"""
Parse YAML while preserving template sections for later rendering.
@@ -158,11 +166,15 @@ class YAMLJinjaLoader:
# Restore template sections with markers
result = self._restore_template_sections(parsed, template_sections)
return cast(Dict[str, Any], result)
if not isinstance(result, dict):
raise ValueError(
f"Expected restored result to be dict, got {type(result).__name__}"
)
return result
def _protect_template_sections( # pylint: disable=too-many-locals,too-many-branches,too-many-statements
self, content: str
) -> Tuple[str, Dict[str, str]]:
) -> Tuple[str, dict[str, str]]:
"""
Protect template sections by replacing them with placeholders.
@@ -177,7 +189,7 @@ class YAMLJinjaLoader:
# Pattern for block templates with their content
# This handles nested blocks by matching the complete structure
def find_block_templates(content: str) -> Tuple[str, Dict[str, str]]:
def find_block_templates(content: str) -> Tuple[str, dict[str, str]]:
"""Find and replace block templates with proper nesting support."""
result = content
local_template_sections = {}
@@ -290,7 +302,7 @@ class YAMLJinjaLoader:
return protected, template_sections
def _restore_template_sections(
self, data: Any, template_sections: Dict[str, str]
self, data: Any, template_sections: dict[str, str]
) -> Any:
"""
Restore template sections in the parsed data.
@@ -303,7 +315,7 @@ class YAMLJinjaLoader:
Data with template markers
"""
if isinstance(data, dict):
result: Dict[str, Any] = {}
result: dict[str, Any] = {}
for key, value in data.items():
if (
key == "__template__"
@@ -358,8 +370,8 @@ class TemplateAwareYAMLParser:
self.logger = logging.getLogger(__name__)
def parse_file(
self, file_path: Path, context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, file_path: Path, context: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
"""
Parse a YAML file that may contain Jinja2 templates.
@@ -373,8 +385,8 @@ class TemplateAwareYAMLParser:
return self.loader.load_file(file_path, context)
def parse_string(
self, yaml_content: str, context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, yaml_content: str, context: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
"""
Parse YAML content that may contain Jinja2 templates.
@@ -388,8 +400,8 @@ class TemplateAwareYAMLParser:
return self.loader.load_string(yaml_content, context)
def extract_raw_templates(
self, config: Dict[str, Any]
) -> Dict[str, Dict[str, str]]:
self, config: dict[str, Any]
) -> dict[str, dict[str, str]]:
"""
Extract raw template definitions from a configuration.
@@ -402,7 +414,7 @@ class TemplateAwareYAMLParser:
Returns:
Dictionary of template types to template definitions
"""
raw_templates: Dict[str, Dict[str, Any]] = {
raw_templates: dict[str, dict[str, Any]] = {
"agents": {},
"graphs": {},
"streams": {},
src/cleveragents/templates/yaml_preprocessor.py
+15 -12
View File
@@ -9,9 +9,7 @@ import logging
import re
from pathlib import Path
from typing import Any
from typing import Dict
from typing import Optional
from typing import cast
import yaml
from jinja2 import Environment
@@ -40,7 +38,7 @@ class YAMLTemplateProcessor:
lstrip_blocks=True,
)
def process_file(self, file_path: Path, context: Dict[str, Any]) -> Dict[str, Any]:
def process_file(self, file_path: Path, context: dict[str, Any]) -> dict[str, Any]:
"""
Process a YAML file containing Jinja2 templates.
@@ -57,8 +55,8 @@ class YAMLTemplateProcessor:
return self.process_string(content, context)
def process_string(
self, yaml_content: str, context: Dict[str, Any]
) -> Dict[str, Any]:
self, yaml_content: str, context: dict[str, Any]
) -> dict[str, Any]:
"""
Process a YAML string containing Jinja2 templates.
@@ -75,7 +73,12 @@ class YAMLTemplateProcessor:
processed_content = template.render(**context)
# Parse the final YAML
return cast(Dict[str, Any], yaml.safe_load(processed_content))
result = yaml.safe_load(processed_content)
if not isinstance(result, dict):
raise ValueError(
f"Expected YAML to parse as dict, got {type(result).__name__}"
)
return result
except yaml.YAMLError as e:
logger.error("Failed to parse processed YAML: %s", e)
logger.debug("Processed content:\n%s", processed_content)
@@ -84,7 +87,7 @@ class YAMLTemplateProcessor:
logger.error("Failed to process template: %s", e)
raise
def _process_template_blocks(self, content: str, context: Dict[str, Any]) -> str:
def _process_template_blocks(self, content: str, context: dict[str, Any]) -> str:
"""
Process Jinja2 block templates (loops, conditionals) in YAML.
@@ -135,7 +138,7 @@ class YAMLTemplateProcessor:
return processed
def _process_inline_templates(self, content: str, context: Dict[str, Any]) -> str:
def _process_inline_templates(self, content: str, context: dict[str, Any]) -> str:
"""
Process inline Jinja2 templates (variable substitutions).
@@ -190,8 +193,8 @@ class TemplateAwareConfigParser:
self.logger = logging.getLogger(__name__)
def parse_template_file(
self, file_path: Path, template_context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, file_path: Path, template_context: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
"""
Parse a YAML file that may contain Jinja2 templates.
@@ -226,8 +229,8 @@ class TemplateAwareConfigParser:
raise
def parse_template_string(
self, yaml_content: str, template_context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, yaml_content: str, template_context: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
"""
Parse a YAML string that may contain Jinja2 templates.
src/cleveragents/templates/yaml_template_engine.py
+17 -19
View File
@@ -9,7 +9,6 @@ import logging
import re
from pathlib import Path
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Tuple
@@ -52,16 +51,16 @@ class YAMLTemplateEngine:
self.env.filters["selectattr"] = self._selectattr_filter
def load_file(
self, file_path: Path, context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, file_path: Path, context: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
"""Load a YAML file with inline Jinja2 templates."""
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
return self.load_string(content, context)
def load_string(
self, yaml_content: str, context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
self, yaml_content: str, context: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
"""
Load YAML content with inline Jinja2 templates.
@@ -88,8 +87,8 @@ class YAMLTemplateEngine:
return self._prepare_for_deferred_rendering(yaml_content)
def _render_and_parse(
self, yaml_content: str, context: Dict[str, Any]
) -> Dict[str, Any]:
self, yaml_content: str, context: dict[str, Any]
) -> dict[str, Any]:
"""Render templates and parse YAML with proper structure preservation."""
# Pre-process to handle special cases
processed_content = self._preprocess_for_rendering(yaml_content)
@@ -233,7 +232,7 @@ class YAMLTemplateEngine:
return "\n".join(fixed_lines)
def _prepare_for_deferred_rendering(self, yaml_content: str) -> Dict[str, Any]:
def _prepare_for_deferred_rendering(self, yaml_content: str) -> dict[str, Any]:
"""
Prepare YAML with templates for deferred rendering.
@@ -244,12 +243,12 @@ class YAMLTemplateEngine:
# The complex extraction was causing issues
return self._simple_template_extraction(yaml_content)
def _analyze_yaml_structure(self, content: str) -> Dict[str, Any]:
def _analyze_yaml_structure(self, content: str) -> dict[str, Any]:
"""
Analyze YAML structure to understand where templates are.
"""
lines = content.split("\n")
structure: Dict[str, Any] = {
structure: dict[str, Any] = {
"template_blocks": [],
"inline_templates": [],
"hierarchy": [],
@@ -318,8 +317,8 @@ class YAMLTemplateEngine:
return len(lines) - 1
def _extract_template_sections( # pylint: disable=too-many-locals
self, content: str, structure: Dict[str, Any]
) -> Dict[str, Any]:
self, content: str, structure: dict[str, Any]
) -> dict[str, Any]:
"""
Extract template sections and create a parseable structure.
"""
@@ -396,14 +395,14 @@ class YAMLTemplateEngine:
# Fall back to simpler approach
return self._simple_template_extraction(content)
def _simple_template_extraction(self, content: str) -> Dict[str, Any]:
def _simple_template_extraction(self, content: str) -> dict[str, Any]:
"""
Simpler extraction method as fallback.
"""
# Just mark the whole content as a template if it contains Jinja2
return {"_raw_template": content, "_is_template": True}
def _create_render_context(self, context: Dict[str, Any]) -> Dict[str, Any]:
def _create_render_context(self, context: dict[str, Any]) -> dict[str, Any]:
"""Create a complete rendering context with utilities."""
full_context = {
# Python built-ins
@@ -463,11 +462,10 @@ class YAMLTemplateEngine:
) -> List[Any]:
"""Custom selectattr filter."""
import operator # pylint: disable=import-outside-toplevel
from typing import cast # pylint: disable=import-outside-toplevel
def default_eq(x: Any, y: Any) -> bool:
"""Default equality comparison."""
return cast(bool, x == y)
return bool(x == y)
if func == ">":
op = operator.gt
@@ -497,8 +495,8 @@ class YAMLTemplateEngine:
return result
def render_template(
self, template_config: Dict[str, Any], context: Dict[str, Any]
) -> Dict[str, Any]:
self, template_config: dict[str, Any], context: dict[str, Any]
) -> dict[str, Any]:
"""
Render a stored template configuration.
"""
@@ -510,7 +508,7 @@ class YAMLTemplateEngine:
yaml_content = self._reconstruct_from_structure(template_config)
return self._render_and_parse(yaml_content, context)
def _reconstruct_from_structure(self, config: Dict[str, Any]) -> str:
def _reconstruct_from_structure(self, config: dict[str, Any]) -> str:
"""Reconstruct YAML with templates from stored structure."""
def process_value(value: Any, indent: int = 0) -> str: # pylint: disable=too-many-branches
tests/unit/core/test_json_sanitization.py
+37 -37
View File
@@ -6,53 +6,54 @@ Tests the _sanitize_json_string method that handles malformed JSON from LLMs.
import pytest
import json
import re
class TestJSONSanitization:
"""Test JSON sanitization functionality."""
@pytest.fixture
def app_with_sanitizer(self):
"""Create a minimal app instance with sanitization method."""
from cleveragents.core.application import ReactiveCleverAgentsApp
# Create app with minimal config
app = ReactiveCleverAgentsApp.__new__(ReactiveCleverAgentsApp)
return app
def test_valid_json_unchanged(self, app_with_sanitizer):
"""Test that already valid JSON is not modified."""
valid_json = '{"file": "test.txt", "content": "Hello World"}'
result = app_with_sanitizer._sanitize_json_string(valid_json)
assert result == valid_json
# Should parse successfully
parsed = json.loads(result)
assert parsed["file"] == "test.txt"
assert parsed["content"] == "Hello World"
def test_newline_escaping(self, app_with_sanitizer):
"""Test that literal newlines are escaped."""
malformed_json = '{"file": "test.txt", "content": "Line 1\nLine 2\nLine 3"}'
result = app_with_sanitizer._sanitize_json_string(malformed_json)
# Should parse successfully now
parsed = json.loads(result)
assert parsed["file"] == "test.txt"
assert "Line 1" in parsed["content"]
assert "Line 2" in parsed["content"]
assert "Line 3" in parsed["content"]
def test_multiple_newlines(self, app_with_sanitizer):
"""Test multiple consecutive newlines."""
malformed_json = '{"file": "paper.txt", "content": "Title\n\nIntroduction\n\nConclusion"}'
result = app_with_sanitizer._sanitize_json_string(malformed_json)
parsed = json.loads(result)
assert "Title" in parsed["content"]
assert "Introduction" in parsed["content"]
assert "Conclusion" in parsed["content"]
def test_complex_content_with_newlines(self, app_with_sanitizer):
"""Test complex multi-section content like the blockchain example."""
malformed_json = '''{"file": "blockchain.txt", "content": "Title: Blockchain Technology
@@ -65,9 +66,9 @@ The blockchain works by...
III. Conclusion
In conclusion, blockchain represents..."}'''
result = app_with_sanitizer._sanitize_json_string(malformed_json)
# Should parse successfully
parsed = json.loads(result)
assert parsed["file"] == "blockchain.txt"
@@ -75,83 +76,83 @@ In conclusion, blockchain represents..."}'''
assert "I. Introduction" in parsed["content"]
assert "II. Technical Details" in parsed["content"]
assert "III. Conclusion" in parsed["content"]
def test_tab_escaping(self, app_with_sanitizer):
"""Test that literal tabs are escaped."""
malformed_json = '{"file": "test.txt", "content": "Item 1:\tValue 1\nItem 2:\tValue 2"}'
result = app_with_sanitizer._sanitize_json_string(malformed_json)
parsed = json.loads(result)
assert "Item 1:" in parsed["content"]
assert "Value 1" in parsed["content"]
def test_carriage_return_escaping(self, app_with_sanitizer):
"""Test that carriage returns are escaped."""
malformed_json = '{"file": "test.txt", "content": "Line 1\r\nLine 2\r\nLine 3"}'
result = app_with_sanitizer._sanitize_json_string(malformed_json)
parsed = json.loads(result)
assert "Line 1" in parsed["content"]
assert "Line 2" in parsed["content"]
def test_mixed_control_characters(self, app_with_sanitizer):
"""Test multiple types of control characters."""
malformed_json = '{"file": "test.txt", "content": "Title\n\tSection 1\r\n\tSection 2"}'
result = app_with_sanitizer._sanitize_json_string(malformed_json)
parsed = json.loads(result)
assert "Title" in parsed["content"]
assert "Section 1" in parsed["content"]
assert "Section 2" in parsed["content"]
def test_empty_content(self, app_with_sanitizer):
"""Test with empty content."""
valid_json = '{"file": "empty.txt", "content": ""}'
result = app_with_sanitizer._sanitize_json_string(valid_json)
parsed = json.loads(result)
assert parsed["file"] == "empty.txt"
assert parsed["content"] == ""
def test_content_with_quotes(self, app_with_sanitizer):
"""Test content that contains quotes (edge case)."""
# This is a tricky case - content has escaped quotes
valid_json = '{"file": "test.txt", "content": "He said \\"Hello\\""}'
result = app_with_sanitizer._sanitize_json_string(valid_json)
parsed = json.loads(result)
assert parsed["file"] == "test.txt"
# The content should preserve the quotes
assert "Hello" in parsed["content"]
def test_long_content(self, app_with_sanitizer):
"""Test with very long content (realistic paper length)."""
long_content = "Section 1\n" + ("This is a long paragraph. " * 100) + "\n\nSection 2\n" + ("Another paragraph. " * 100)
malformed_json = f'{{"file": "long.txt", "content": "{long_content}"}}'
result = app_with_sanitizer._sanitize_json_string(malformed_json)
# Should parse successfully
parsed = json.loads(result)
assert parsed["file"] == "long.txt"
assert "Section 1" in parsed["content"]
assert "Section 2" in parsed["content"]
assert len(parsed["content"]) > 1000 # Should be long
def test_special_characters_preserved(self, app_with_sanitizer):
"""Test that special characters are preserved."""
valid_json = '{"file": "test.txt", "content": "Math: x + y = z, Cost: $100"}'
result = app_with_sanitizer._sanitize_json_string(valid_json)
parsed = json.loads(result)
assert "x + y = z" in parsed["content"]
assert "$100" in parsed["content"]
def test_unicode_characters(self, app_with_sanitizer):
"""Test that unicode characters are preserved."""
valid_json = '{"file": "test.txt", "content": "Hello 世界 🌍"}'
result = app_with_sanitizer._sanitize_json_string(valid_json)
parsed = json.loads(result)
assert "世界" in parsed["content"]
assert "🌍" in parsed["content"]
@@ -159,29 +160,28 @@ In conclusion, blockchain represents..."}'''
class TestToolCommandProcessing:
"""Test full tool command processing with sanitization."""
def test_tool_command_pattern_matching(self):
"""Test that the regex pattern matches tool commands correctly."""
import re
pattern = r'\[TOOL_EXECUTE:(\w+)\]\s*(\{[^}]+\})\s*\[/TOOL_EXECUTE\]'
# Simple case
content = '[TOOL_EXECUTE:file_write]{"file": "test.txt", "content": "Hello"}[/TOOL_EXECUTE]'
match = re.search(pattern, content, re.DOTALL)
assert match is not None
assert match.group(1) == "file_write"
assert '{"file": "test.txt"' in match.group(2)
def test_tool_command_with_newlines(self):
"""Test pattern matching with newlines in content."""
import re
# Note: The current pattern has a limitation - it uses [^}]+ which stops at first }
# This is actually OK for our use case since we sanitize after extraction
pattern = r'\[TOOL_EXECUTE:(\w+)\]\s*(\{[^}]+\})\s*\[/TOOL_EXECUTE\]'
content = '''Here is your paper:
[TOOL_EXECUTE:file_write]
@@ -190,7 +190,7 @@ World"}
[/TOOL_EXECUTE]
Done!'''
match = re.search(pattern, content, re.DOTALL)
assert match is not None
tests/unit/core/test_tool_command_processing.py
+1 -2
View File
@@ -96,7 +96,7 @@ class TestToolCommandProcessing:
app = ReactiveCleverAgentsApp(config_files=None, unsafe=True)
import re
pattern = r'\[TOOL_EXECUTE:(\w+)\]\s*(.*?)\s*\[/TOOL_EXECUTE\]'
matches = list(re.finditer(pattern, content, re.DOTALL))
@@ -117,7 +117,6 @@ class TestToolCommandProcessing:
app = ReactiveCleverAgentsApp(config_files=None, unsafe=True)
import re
pattern = r'\[TOOL_EXECUTE:(\w+)\]\s*(.*?)\s*\[/TOOL_EXECUTE\]'
matches = list(re.finditer(pattern, content, re.DOTALL))