25213f2004
- eveai_chat becomes deprecated and should be replaced with SSE - Adaptation of STANDARD_RAG specialist - Base class definition allowing to realise specialists with crewai framework - Implementation of SPIN_SPECIALIST - Implementation of test app for testing specialists (test_specialist_client). Also serves as an example for future SSE-based client - Improvements to startup scripts to better handle and scale multiple connections - Small improvements to the interaction forms and views - Caching implementation improved and augmented with additional caches
126 lines
3.7 KiB
Python
126 lines
3.7 KiB
Python
from dataclasses import dataclass
|
|
from typing import Dict, Any, Optional
|
|
|
|
|
|
@dataclass
|
|
class ProcessedAgentConfig:
|
|
"""Processed and ready-to-use agent configuration"""
|
|
role: str
|
|
goal: str
|
|
backstory: str
|
|
name: str
|
|
type: str
|
|
description: Optional[str] = None
|
|
verbose: bool = False
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert to dictionary for serialization"""
|
|
return {
|
|
'role': self.role,
|
|
'goal': self.goal,
|
|
'backstory': self.backstory,
|
|
'name': self.name,
|
|
'type': self.type,
|
|
'description': self.description,
|
|
'verbose': self.verbose
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'ProcessedAgentConfig':
|
|
"""Create from dictionary"""
|
|
return cls(**data)
|
|
|
|
|
|
@dataclass
|
|
class ProcessedTaskConfig:
|
|
"""Processed and ready-to-use task configuration"""
|
|
task_description: str
|
|
expected_output: str
|
|
name: str
|
|
type: str
|
|
description: Optional[str] = None
|
|
verbose: bool = False
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert to dictionary for serialization"""
|
|
return {
|
|
'task_description': self.task_description,
|
|
'expected_output': self.expected_output,
|
|
'name': self.name,
|
|
'type': self.type,
|
|
'description': self.description,
|
|
'verbose': self.verbose
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'ProcessedTaskConfig':
|
|
"""Create from dictionary"""
|
|
return cls(**data)
|
|
|
|
|
|
@dataclass
|
|
class ProcessedToolConfig:
|
|
"""Processed and ready-to-use tool configuration"""
|
|
name: str
|
|
type: str
|
|
description: Optional[str] = None
|
|
configuration: Optional[Dict[str, Any]] = None
|
|
verbose: bool = False
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert to dictionary for serialization"""
|
|
return {
|
|
'name': self.name,
|
|
'type': self.type,
|
|
'description': self.description,
|
|
'configuration': self.configuration,
|
|
'verbose': self.verbose
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'ProcessedToolConfig':
|
|
"""Create from dictionary"""
|
|
return cls(**data)
|
|
|
|
|
|
@dataclass
|
|
class SpecialistProcessedConfig:
|
|
"""Complete processed configuration for a specialist"""
|
|
agents: Dict[str, ProcessedAgentConfig]
|
|
tasks: Dict[str, ProcessedTaskConfig]
|
|
tools: Dict[str, ProcessedToolConfig]
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert entire configuration to dictionary"""
|
|
return {
|
|
'agents': {
|
|
agent_type: config.to_dict()
|
|
for agent_type, config in self.agents.items()
|
|
},
|
|
'tasks': {
|
|
task_type: config.to_dict()
|
|
for task_type, config in self.tasks.items()
|
|
},
|
|
'tools': {
|
|
tool_type: config.to_dict()
|
|
for tool_type, config in self.tools.items()
|
|
}
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'SpecialistProcessedConfig':
|
|
"""Create from dictionary"""
|
|
return cls(
|
|
agents={
|
|
agent_type: ProcessedAgentConfig.from_dict(config)
|
|
for agent_type, config in data['agents'].items()
|
|
},
|
|
tasks={
|
|
task_type: ProcessedTaskConfig.from_dict(config)
|
|
for task_type, config in data['tasks'].items()
|
|
},
|
|
tools={
|
|
tool_type: ProcessedToolConfig.from_dict(config)
|
|
for tool_type, config in data['tools'].items()
|
|
}
|
|
) |