- Implementation of specialist execution api, including SSE protocol

- 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

.gitignore

@@ -47,3 +47,6 @@ scripts/__pycache__/run_eveai_app.cpython-312.pyc
 /integrations/Wordpress/eveai_sync.zip
 /integrations/Wordpress/eveai-chat.zip
 /db_backups/
+/tests/interactive_client/specialist_client.log
+/.repopackignore
+/patched_packages/crewai/

.idea/misc.xml

@@ -3,5 +3,5 @@
   <component name="Black">
     <option name="sdkName" value="Python 3.12 (eveai_tbd)" />
   </component>
-  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.12 (eveai_tbd)" project-jdk-type="Python SDK" />
+  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.12 (TBD)" project-jdk-type="Python SDK" />
 </project>

.python-version

@@ -1 +1 @@
-eveai_tbd
+3.12.7

.repopackignore_patched_packages

@@ -0,0 +1,11 @@
+docker/
+eveai_api/
+eveai_app/
+eveai_beat/
+eveai_chat/
+eveai_entitlements/
+eveai_workers/
+instance/
+integrations/
+nginx/
+scripts/

common/models/document.py

@@ -56,6 +56,7 @@ class Retriever(db.Model):
     description = db.Column(db.Text, nullable=True)
     catalog_id = db.Column(db.Integer, db.ForeignKey('catalog.id'), nullable=True)
     type = db.Column(db.String(50), nullable=False, default="STANDARD_RAG")
+    type_version = db.Column(db.String(20), nullable=True, default="STANDARD_RAG")
     tuning = db.Column(db.Boolean, nullable=True, default=False)
 
     # Meta Data

common/utils/cache/base.py

@@ -138,11 +138,14 @@ class CacheHandler(Generic[T]):
             Cached or newly created value
         """
         cache_key = self.generate_key(**identifiers)
-        current_app.logger.debug(f"Cache key: {cache_key}")
+        current_app.logger.debug(f"Getting Cache key: {cache_key}")
 
         def creator():
             instance = creator_func(**identifiers)
-            return self._to_cache_data(instance)
+            current_app.logger.debug("Caching object created and received. Now serializing...")
+            serialized_instance = self._to_cache_data(instance)
+            current_app.logger.debug(f"Caching object serialized and received:\n{serialized_instance}")
+            return serialized_instance
 
         cached_data = self.region.get_or_create(
             cache_key,

common/utils/cache/config_cache.py

@@ -6,7 +6,7 @@ import os
 from flask import current_app
 
 from common.utils.cache.base import CacheHandler, CacheKey
-from config.type_defs import agent_types, task_types, tool_types, specialist_types
+from config.type_defs import agent_types, task_types, tool_types, specialist_types, retriever_types, prompt_types
 
 
 def is_major_minor(version: str) -> bool:
@@ -72,6 +72,7 @@ class BaseConfigCacheHandler(CacheHandler[Dict[str, Any]]):
         current_app.logger.debug(f"Loading specific configuration for {type_name}, version: {version_str} - no cache")
         version_tree = self.version_tree_cache.get_versions(type_name)
         versions = version_tree['versions']
+        current_app.logger.debug(f"Loaded specific versions for {type_name}, versions: {versions}")
 
         if version_str == 'latest':
             version_str = version_tree['latest_version']
@@ -80,6 +81,7 @@ class BaseConfigCacheHandler(CacheHandler[Dict[str, Any]]):
             raise ValueError(f"Version {version_str} not found for {type_name}")
 
         file_path = versions[version_str]['file_path']
+        current_app.logger.debug(f'Trying to load configuration from {file_path}')
 
         try:
             with open(file_path) as f:
@@ -418,3 +420,48 @@ SpecialistConfigCacheHandler, SpecialistConfigVersionTreeCacheHandler, Specialis
         config_dir='config/specialists',
         types_module=specialist_types.SPECIALIST_TYPES
     ))
+
+
+RetrieverConfigCacheHandler, RetrieverConfigVersionTreeCacheHandler, RetrieverConfigTypesCacheHandler = (
+    create_config_cache_handlers(
+        config_type='retrievers',
+        config_dir='config/retrievers',
+        types_module=retriever_types.RETRIEVER_TYPES
+    ))
+
+
+PromptConfigCacheHandler, PromptConfigVersionTreeCacheHandler, PromptConfigTypesCacheHandler = (
+    create_config_cache_handlers(
+        config_type='prompts',
+        config_dir='config/prompts',
+        types_module=prompt_types.PROMPT_TYPES
+
+    ))
+
+
+def register_config_cache_handlers(cache_manager) -> None:
+    cache_manager.register_handler(AgentConfigCacheHandler, 'eveai_config')
+    cache_manager.register_handler(AgentConfigTypesCacheHandler, 'eveai_config')
+    cache_manager.register_handler(AgentConfigVersionTreeCacheHandler, 'eveai_config')
+    cache_manager.register_handler(TaskConfigCacheHandler, 'eveai_config')
+    cache_manager.register_handler(TaskConfigTypesCacheHandler, 'eveai_config')
+    cache_manager.register_handler(TaskConfigVersionTreeCacheHandler, 'eveai_config')
+    cache_manager.register_handler(ToolConfigCacheHandler, 'eveai_config')
+    cache_manager.register_handler(ToolConfigTypesCacheHandler, 'eveai_config')
+    cache_manager.register_handler(ToolConfigVersionTreeCacheHandler, 'eveai_config')
+    cache_manager.register_handler(SpecialistConfigCacheHandler, 'eveai_config')
+    cache_manager.register_handler(SpecialistConfigTypesCacheHandler, 'eveai_config')
+    cache_manager.register_handler(SpecialistConfigVersionTreeCacheHandler, 'eveai_config')
+    cache_manager.register_handler(RetrieverConfigCacheHandler, 'eveai_config')
+    cache_manager.register_handler(RetrieverConfigTypesCacheHandler, 'eveai_config')
+    cache_manager.register_handler(RetrieverConfigVersionTreeCacheHandler, 'eveai_config')
+    cache_manager.register_handler(PromptConfigCacheHandler, 'eveai_config')
+    cache_manager.register_handler(PromptConfigVersionTreeCacheHandler, 'eveai_config')
+    cache_manager.register_handler(PromptConfigTypesCacheHandler, 'eveai_config')
+
+    cache_manager.agents_config_cache.set_version_tree_cache(cache_manager.agents_version_tree_cache)
+    cache_manager.tasks_config_cache.set_version_tree_cache(cache_manager.tasks_version_tree_cache)
+    cache_manager.tools_config_cache.set_version_tree_cache(cache_manager.tools_version_tree_cache)
+    cache_manager.specialists_config_cache.set_version_tree_cache(cache_manager.specialists_version_tree_cache)
+    cache_manager.retrievers_config_cache.set_version_tree_cache(cache_manager.retrievers_version_tree_cache)
+    cache_manager.prompts_config_cache.set_version_tree_cache(cache_manager.prompts_version_tree_cache)

common/utils/cache/crewai_config_processor.py

@@ -0,0 +1,220 @@
+from typing import Dict, Any, Type, TypeVar, List
+from abc import ABC, abstractmethod
+from flask import current_app
+
+from common.extensions import cache_manager, db
+from common.models.interaction import EveAIAgent, EveAITask, EveAITool, Specialist
+from common.utils.cache.crewai_configuration import (
+    ProcessedAgentConfig, ProcessedTaskConfig, ProcessedToolConfig,
+    SpecialistProcessedConfig
+)
+
+T = TypeVar('T')  # For generic model types
+
+
+class BaseCrewAIConfigProcessor:
+    """Base processor for specialist configurations"""
+
+    # Standard mapping between model fields and template placeholders
+    AGENT_FIELD_MAPPING = {
+        'role': 'custom_role',
+        'goal': 'custom_goal',
+        'backstory': 'custom_backstory'
+    }
+
+    TASK_FIELD_MAPPING = {
+        'task_description': 'custom_description',
+        'expected_output': 'custom_expected_output'
+    }
+
+    def __init__(self, tenant_id: int, specialist_id: int):
+        self.tenant_id = tenant_id
+        self.specialist_id = specialist_id
+        self.specialist = self._get_specialist()
+        self.verbose = self._get_verbose_setting()
+
+    def _get_specialist(self) -> Specialist:
+        """Get specialist and verify existence"""
+        specialist = Specialist.query.get(self.specialist_id)
+        if not specialist:
+            raise ValueError(f"Specialist {self.specialist_id} not found")
+        return specialist
+
+    def _get_verbose_setting(self) -> bool:
+        """Get verbose setting from specialist"""
+        return bool(self.specialist.tuning)
+
+    def _get_db_items(self, model_class: Type[T], type_list: List[str]) -> Dict[str, T]:
+        """Get database items of specified type"""
+        items = (model_class.query
+                 .filter_by(specialist_id=self.specialist_id)
+                 .filter(model_class.type.in_(type_list))
+                 .all())
+        return {item.type: item for item in items}
+
+    def _apply_replacements(self, text: str, replacements: Dict[str, str]) -> str:
+        """Apply text replacements to a string"""
+        result = text
+        for key, value in replacements.items():
+            if value is not None:  # Only replace if value exists
+                placeholder = "{" + key + "}"
+                result = result.replace(placeholder, str(value))
+        return result
+
+    def _process_agent_configs(self, specialist_config: Dict[str, Any]) -> Dict[str, ProcessedAgentConfig]:
+        """Process all agent configurations"""
+        agent_configs = {}
+
+        if 'agents' not in specialist_config:
+            return agent_configs
+
+        # Get all DB agents at once
+        agent_types = [agent_def['type'] for agent_def in specialist_config['agents']]
+        db_agents = self._get_db_items(EveAIAgent, agent_types)
+
+        for agent_def in specialist_config['agents']:
+            agent_type = agent_def['type']
+            agent_type_lower = agent_type.lower()
+            db_agent = db_agents.get(agent_type)
+
+            # Get full configuration
+            config = cache_manager.agents_config_cache.get_config(
+                agent_type,
+                agent_def.get('version', '1.0')
+            )
+
+            # Start with YAML values
+            role = config['role']
+            goal = config['goal']
+            backstory = config['backstory']
+
+            # Apply DB values if they exist
+            if db_agent:
+                for model_field, placeholder in self.AGENT_FIELD_MAPPING.items():
+                    value = getattr(db_agent, model_field)
+                    if value:
+                        placeholder_text = "{" + placeholder + "}"
+                        role = role.replace(placeholder_text, value)
+                        goal = goal.replace(placeholder_text, value)
+                        backstory = backstory.replace(placeholder_text, value)
+
+            agent_configs[agent_type_lower] = ProcessedAgentConfig(
+                role=role,
+                goal=goal,
+                backstory=backstory,
+                name=agent_def.get('name') or config.get('name', agent_type_lower),
+                type=agent_type,
+                description=agent_def.get('description') or config.get('description'),
+                verbose=self.verbose
+            )
+
+        return agent_configs
+
+    def _process_task_configs(self, specialist_config: Dict[str, Any]) -> Dict[str, ProcessedTaskConfig]:
+        """Process all task configurations"""
+        task_configs = {}
+
+        if 'tasks' not in specialist_config:
+            return task_configs
+
+        # Get all DB tasks at once
+        task_types = [task_def['type'] for task_def in specialist_config['tasks']]
+        db_tasks = self._get_db_items(EveAITask, task_types)
+
+        for task_def in specialist_config['tasks']:
+            task_type = task_def['type']
+            task_type_lower = task_type.lower()
+            db_task = db_tasks.get(task_type)
+
+            # Get full configuration
+            config = cache_manager.tasks_config_cache.get_config(
+                task_type,
+                task_def.get('version', '1.0')
+            )
+
+            # Start with YAML values
+            task_description = config['task_description']
+            expected_output = config['expected_output']
+
+            # Apply DB values if they exist
+            if db_task:
+                for model_field, placeholder in self.TASK_FIELD_MAPPING.items():
+                    value = getattr(db_task, model_field)
+                    if value:
+                        placeholder_text = "{" + placeholder + "}"
+                        task_description = task_description.replace(placeholder_text, value)
+                        expected_output = expected_output.replace(placeholder_text, value)
+
+            task_configs[task_type_lower] = ProcessedTaskConfig(
+                task_description=task_description,
+                expected_output=expected_output,
+                name=task_def.get('name') or config.get('name', task_type_lower),
+                type=task_type,
+                description=task_def.get('description') or config.get('description'),
+                verbose=self.verbose
+            )
+
+        return task_configs
+
+    def _process_tool_configs(self, specialist_config: Dict[str, Any]) -> Dict[str, ProcessedToolConfig]:
+        """Process all tool configurations"""
+        tool_configs = {}
+
+        if 'tools' not in specialist_config:
+            return tool_configs
+
+        # Get all DB tools at once
+        tool_types = [tool_def['type'] for tool_def in specialist_config['tools']]
+        db_tools = self._get_db_items(EveAITool, tool_types)
+
+        for tool_def in specialist_config['tools']:
+            tool_type = tool_def['type']
+            tool_type_lower = tool_type.lower()
+            db_tool = db_tools.get(tool_type)
+
+            # Get full configuration
+            config = cache_manager.tools_config_cache.get_config(
+                tool_type,
+                tool_def.get('version', '1.0')
+            )
+
+            # Combine configuration
+            tool_config = config.get('configuration', {})
+            if db_tool and db_tool.configuration:
+                tool_config.update(db_tool.configuration)
+
+            tool_configs[tool_type_lower] = ProcessedToolConfig(
+                name=tool_def.get('name') or config.get('name', tool_type_lower),
+                type=tool_type,
+                description=tool_def.get('description') or config.get('description'),
+                configuration=tool_config,
+                verbose=self.verbose
+            )
+
+        return tool_configs
+
+    def process_config(self) -> SpecialistProcessedConfig:
+        """Process complete specialist configuration"""
+        try:
+            # Get full specialist configuration
+            specialist_config = cache_manager.specialists_config_cache.get_config(
+                self.specialist.type,
+                self.specialist.type_version
+            )
+
+            if not specialist_config:
+                raise ValueError(f"No configuration found for {self.specialist.type}")
+
+            # Process all configurations
+            processed_config = SpecialistProcessedConfig(
+                agents=self._process_agent_configs(specialist_config),
+                tasks=self._process_task_configs(specialist_config),
+                tools=self._process_tool_configs(specialist_config)
+            )
+            current_app.logger.debug(f"Processed config for tenant {self.tenant_id}, specialist {self.specialist_id}:\n"
+                                     f"{processed_config}")
+            return processed_config
+
+        except Exception as e:
+            current_app.logger.error(f"Error processing specialist configuration: {e}")
+            raise

common/utils/cache/crewai_configuration.py

@@ -0,0 +1,126 @@
+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()
+            }
+        )

common/utils/cache/crewai_processed_config_cache.py

@@ -0,0 +1,75 @@
+from typing import Dict, Any, Type
+from flask import current_app
+
+from common.utils.cache.base import CacheHandler
+from common.utils.cache.crewai_configuration import SpecialistProcessedConfig
+from common.utils.cache.crewai_config_processor import BaseCrewAIConfigProcessor
+
+
+class CrewAIProcessedConfigCacheHandler(CacheHandler[SpecialistProcessedConfig]):
+    """Handles caching of processed specialist configurations"""
+    handler_name = 'crewai_processed_config_cache'
+
+    def __init__(self, region):
+        super().__init__(region, 'crewai_processed_config')
+        self.configure_keys('tenant_id', 'specialist_id')
+
+    def _to_cache_data(self, instance: SpecialistProcessedConfig) -> Dict[str, Any]:
+        """Convert SpecialistProcessedConfig to cache data"""
+        return instance.to_dict()
+
+    def _from_cache_data(self, data: Dict[str, Any], **kwargs) -> SpecialistProcessedConfig:
+        """Create SpecialistProcessedConfig from cache data"""
+        return SpecialistProcessedConfig.from_dict(data)
+
+    def _should_cache(self, value: Dict[str, Any]) -> bool:
+        """Validate cache data"""
+        required_keys = {'agents', 'tasks', 'tools'}
+        if not all(key in value for key in required_keys):
+            current_app.logger.warning(f'CrewAI Processed Config Cache missing required keys: {required_keys}')
+            return False
+        return bool(value['agents'] or value['tasks'])
+
+    def get_specialist_config(self, tenant_id: int, specialist_id: int) -> SpecialistProcessedConfig:
+        """
+        Get or create processed configuration for a specialist
+
+        Args:
+            tenant_id: Tenant ID
+            specialist_id: Specialist ID
+
+        Returns:
+            Processed specialist configuration
+
+        Raises:
+            ValueError: If specialist not found or processor not configured
+        """
+
+        def creator_func(tenant_id: int, specialist_id: int) -> SpecialistProcessedConfig:
+            # Create processor instance and process config
+            processor = BaseCrewAIConfigProcessor(tenant_id, specialist_id)
+            return processor.process_config()
+
+        return self.get(
+            creator_func,
+            tenant_id=tenant_id,
+            specialist_id=specialist_id
+        )
+
+    def invalidate_tenant_specialist(self, tenant_id: int, specialist_id: int):
+        """Invalidate cache for a specific tenant's specialist"""
+        self.invalidate(
+            tenant_id=tenant_id,
+            specialist_id=specialist_id
+        )
+        current_app.logger.info(
+            f"Invalidated cache for tenant {tenant_id} specialist {specialist_id}"
+        )
+
+
+def register_specialist_cache_handlers(cache_manager) -> None:
+    """Register specialist cache handlers with cache manager"""
+    cache_manager.register_handler(
+        CrewAIProcessedConfigCacheHandler,
+        'eveai_chat_workers'
+    )

common/utils/execution_progress.py

@@ -0,0 +1,112 @@
+# common/utils/execution_progress.py
+from datetime import datetime as dt, timezone as tz
+from typing import Generator
+from redis import Redis, RedisError
+import json
+from flask import current_app
+
+
+class ExecutionProgressTracker:
+    """Tracks progress of specialist executions using Redis"""
+
+    def __init__(self):
+        try:
+            redis_url = current_app.config['SPECIALIST_EXEC_PUBSUB']
+
+            self.redis = Redis.from_url(redis_url, socket_timeout=5)
+            # Test the connection
+            self.redis.ping()
+
+            self.expiry = 3600  # 1 hour expiry
+        except RedisError as e:
+            current_app.logger.error(f"Failed to connect to Redis: {str(e)}")
+            raise
+        except Exception as e:
+            current_app.logger.error(f"Unexpected error during Redis initialization: {str(e)}")
+            raise
+
+    def _get_key(self, execution_id: str) -> str:
+        return f"specialist_execution:{execution_id}"
+
+    def send_update(self, ctask_id: str, processing_type: str, data: dict):
+        """Send an update about execution progress"""
+        try:
+            key = self._get_key(ctask_id)
+
+            # First verify Redis is still connected
+            try:
+                self.redis.ping()
+            except RedisError:
+                current_app.logger.error("Lost Redis connection. Attempting to reconnect...")
+                self.__init__()  # Reinitialize connection
+
+            update = {
+                'processing_type': processing_type,
+                'data': data,
+                'timestamp': dt.now(tz=tz.utc)
+            }
+
+            # Log initial state
+            try:
+                orig_len = self.redis.llen(key)
+
+                # Try to serialize the update and check the result
+                try:
+                    serialized_update = json.dumps(update, default=str)  # Add default handler for datetime
+                except TypeError as e:
+                    current_app.logger.error(f"Failed to serialize update: {str(e)}")
+                    raise
+
+                # Store update in list with pipeline for atomicity
+                with self.redis.pipeline() as pipe:
+                    pipe.rpush(key, serialized_update)
+                    pipe.publish(key, serialized_update)
+                    pipe.expire(key, self.expiry)
+                    results = pipe.execute()
+
+                new_len = self.redis.llen(key)
+
+                if new_len <= orig_len:
+                    current_app.logger.error(
+                        f"List length did not increase as expected. Original: {orig_len}, New: {new_len}")
+
+            except RedisError as e:
+                current_app.logger.error(f"Redis operation failed: {str(e)}")
+                raise
+
+        except Exception as e:
+            current_app.logger.error(f"Unexpected error in send_update: {str(e)}, type: {type(e)}")
+            raise
+
+    def get_updates(self, ctask_id: str) -> Generator[str, None, None]:
+        key = self._get_key(ctask_id)
+        pubsub = self.redis.pubsub()
+        pubsub.subscribe(key)
+
+        try:
+            # First yield any existing updates
+            length = self.redis.llen(key)
+            if length > 0:
+                updates = self.redis.lrange(key, 0, -1)
+                for update in updates:
+                    update_data = json.loads(update.decode('utf-8'))
+                    # Use processing_type for the event
+                    yield f"event: {update_data['processing_type']}\n"
+                    yield f"data: {json.dumps(update_data)}\n\n"
+
+            # Then listen for new updates
+            while True:
+                message = pubsub.get_message(timeout=30)  # message['type'] is Redis pub/sub type
+                if message is None:
+                    yield ": keepalive\n\n"
+                    continue
+
+                if message['type'] == 'message':  # This is Redis pub/sub type
+                    update_data = json.loads(message['data'].decode('utf-8'))
+                    yield f"data: {message['data'].decode('utf-8')}\n\n"
+
+                    # Check processing_type for completion
+                    if update_data['processing_type'] in ['Task Complete', 'Task Error']:
+                        break
+        finally:
+            pubsub.unsubscribe()

common/utils/pydantic_utils.py

@@ -0,0 +1,78 @@
+from pydantic import BaseModel, Field
+from typing import Any, Dict, List, Optional, Type, Union
+
+
+def flatten_pydantic_model(model: BaseModel, merge_strategy: Dict[str, str] = {}) -> Dict[str, Any]:
+    """
+    Flattens a nested Pydantic model by bringing all attributes to the highest level.
+
+    :param model: Pydantic model instance to be flattened.
+    :param merge_strategy: Dictionary defining how to handle duplicate attributes.
+    :return: Flattened dictionary representation of the model.
+    """
+    flat_dict = {}
+
+    def recursive_flatten(obj: BaseModel, parent_key=""):
+        for field_name, value in obj.model_dump(exclude_unset=True, by_alias=True).items():
+            new_key = field_name  # Maintain original field names
+
+            if isinstance(value, BaseModel):
+                # Recursively flatten nested models
+                recursive_flatten(value, new_key)
+            elif isinstance(value, list) and all(isinstance(i, BaseModel) for i in value):
+                # If it's a list of Pydantic models, flatten each element
+                for item in value:
+                    recursive_flatten(item, new_key)
+            else:
+                if new_key in flat_dict and new_key in merge_strategy:
+                    # Apply merge strategy
+                    if merge_strategy[new_key] == "add":
+                        if isinstance(flat_dict[new_key], list) and isinstance(value, list):
+                            flat_dict[new_key] += value  # Concatenate lists
+                        elif isinstance(flat_dict[new_key], (int, float)) and isinstance(value, (int, float)):
+                            flat_dict[new_key] += value  # Sum numbers
+                        elif isinstance(flat_dict[new_key], str) and isinstance(value, str):
+                            flat_dict[new_key] += "\n" + value  # Concatenate strings
+                    elif merge_strategy[new_key] == "first":
+                        pass  # Keep the first occurrence
+                    elif merge_strategy[new_key] == "last":
+                        flat_dict[new_key] = value
+                else:
+                    flat_dict[new_key] = value
+
+    recursive_flatten(model)
+    return flat_dict
+
+
+def merge_dicts(base_dict: Dict[str, Any], new_data: Union[Dict[str, Any], BaseModel], merge_strategy: Dict[str, str]) \
+        -> Dict[str, Any]:
+    """
+    Merges a Pydantic model (or dictionary) into an existing dictionary based on a merge strategy.
+
+    :param base_dict: The base dictionary to merge into.
+    :param new_data: The new Pydantic model or dictionary to merge.
+    :param merge_strategy: Dict defining how to merge duplicate attributes.
+    :return: Updated dictionary after merging.
+    """
+    if isinstance(new_data, BaseModel):
+        new_data = flatten_pydantic_model(new_data)  # Convert Pydantic model to dict
+
+    for key, value in new_data.items():
+        if key in base_dict and key in merge_strategy:
+            strategy = merge_strategy[key]
+
+            if strategy == "add":
+                if isinstance(base_dict[key], list) and isinstance(value, list):
+                    base_dict[key] += value  # Concatenate lists
+                elif isinstance(base_dict[key], (int, float)) and isinstance(value, (int, float)):
+                    base_dict[key] += value  # Sum numbers
+                elif isinstance(base_dict[key], str) and isinstance(value, str):
+                    base_dict[key] += " " + value  # Concatenate strings
+            elif strategy == "first":
+                pass  # Keep the first occurrence (do nothing)
+            elif strategy == "last":
+                base_dict[key] = value  # Always overwrite with latest value
+        else:
+            base_dict[key] = value  # Add new field
+
+    return base_dict

common/utils/startup_eveai.py

@@ -33,6 +33,7 @@ def perform_startup_invalidation(app):
 
                     # Perform invalidation
                     cache_manager.invalidate_region('eveai_config')
+                    cache_manager.invalidate_region('eveai_chat_workers')
 
                     app.logger.debug(f"Cache keys after invalidation: {redis_client.keys('*')}")
 

config/agents/IDENTIFICATION_AGENT/1.0.0.yaml

@@ -3,13 +3,18 @@ name: "Identification Agent"
 role: >
   Identification Administrative force. {custom_role}
 goal: >
-  You are an administrative force that tries to gather identification information of an end-user through conversation. 
+  You are an administrative force that tries to gather identification information to complete the administration of an 
+  end-user, the company he or she works for, through monitoring conversations and advising on questions to help you do
+  your job. You are responsible for completing the company's backend systems (like CRM, ERP, ...) with inputs from the 
+  end user in the conversation. 
   {custom_goal}
 backstory: >
-  You are and administrative force for {company}. Your task is to identify the person in a conversation, so he or she
-  can easily be contacted later on. {custom_backstory}
+  You are and administrative force for {company}, and very proficient in gathering information for the company's backend
+  systems. You do so by monitoring conversations between one of your colleagues (e.g. sales, finance, support, ...) and 
+  an end user. You ask your colleagues to request additional information to complete your task. 
+  {custom_backstory}
 metadata:
   author: "Josako"
   date_added: "2025-01-08"
-  description: "An Agent that gathers identification information"
+  description: "An Agent that gathers administrative information"
   changes: "Initial version"

config/agents/RAG_COMMUNICATION_AGENT/1.0.0.yaml

@@ -0,0 +1,21 @@
+version: "1.0.0"
+name: "Rag Communication Agent"
+role: >
+  {company} Interaction Responsible. {custom_role}
+goal: >
+  Your team has collected answers to a question asked. But it also created some additional questions to be asked. You
+  ensure the necessary answers are returned, and make an informed selection of the additional questions that can be 
+  asked (combining them when appropriate), ensuring the human you're communicating to does not get overwhelmed. 
+  {custom_goal}
+backstory: >
+  You are the online communication expert for {company}. You handled a lot of online communications with both customers
+  and internal employees. You are a master in redacting one coherent reply in a conversation that includes all the 
+  answers, and a selection of additional questions to be asked in a conversation. Although your backoffice team might 
+  want to ask a myriad of questions, you understand that doesn't fit with the way humans communicate. You know how to
+  combine multiple related questions, and understand how to interweave the questions in the answers when related. 
+  {custom_backstory}
+metadata:
+  author: "Josako"
+  date_added: "2025-01-08"
+  description: "An Agent that consolidates both answers and questions in a consistent reply"
+  changes: "Initial version"

config/agents/SPIN_DETECTION_AGENT/1.0.0.yaml

@@ -10,10 +10,10 @@ backstory: >
   trained to understand an analyse ongoing conversations. Your are proficient in detecting SPIN-related information in a
   conversation.
   SPIN stands for:  
-  - Situation questions & information - Understanding the customer's current context
-  - Problem questions & information - Uncovering challenges and pain points
-  - Implication questions & information - Exploring consequences of those problems
-  - Need-payoff questions & information - Helping customers realize value of solutions
+  - Situation information - Understanding the customer's current context
+  - Problem information - Uncovering challenges and pain points
+  - Implication information - Exploring consequences of those problems
+  - Need-payoff information - Helping customers realize value of solutions
   {custom_backstory}
 metadata:
   author: "Josako"

config/agents/SPIN_SALES_SPECIALIST_AGENT/1.0.0.yaml

@@ -11,10 +11,10 @@ backstory: >
   decide on follow-up questions for more in-depth information to ensure we get the required information that may lead to
   selling {products}. 
   SPIN stands for:  
-  - Situation questions & information - Understanding the customer's current context
-  - Problem questions & information - Uncovering challenges and pain points
-  - Implication questions & information - Exploring consequences of those problems
-  - Need-payoff questions & information - Helping customers realize value of solutions
+  - Situation information - Understanding the customer's current context
+  - Problem information - Uncovering challenges and pain points
+  - Implication information - Exploring consequences of those problems
+  - Need-payoff information - Helping customers realize value of solutions
   {custom_backstory}
   You are acquainted with the following product information:
   {product_information}

config/config.py

@@ -197,6 +197,8 @@ class DevConfig(Config):
     CELERY_RESULT_BACKEND_CHAT = f'{REDIS_BASE_URI}/3'
     # eveai_chat_workers cache Redis Settings
     CHAT_WORKER_CACHE_URL = f'{REDIS_BASE_URI}/4'
+    # specialist execution pub/sub Redis Settings
+    SPECIALIST_EXEC_PUBSUB = f'{REDIS_BASE_URI}/5'
 
 
     # Unstructured settings
@@ -290,6 +292,8 @@ class ProdConfig(Config):
     CELERY_RESULT_BACKEND_CHAT = f'{REDIS_BASE_URI}/3'
     # eveai_chat_workers cache Redis Settings
     CHAT_WORKER_CACHE_URL = f'{REDIS_BASE_URI}/4'
+    # specialist execution pub/sub Redis Settings
+    SPECIALIST_EXEC_PUBSUB = f'{REDIS_BASE_URI}/5'
 
     # Session settings
     SESSION_REDIS = redis.from_url(f'{REDIS_BASE_URI}/2')

config/prompts/encyclopedia/1.0.0.yaml

@@ -0,0 +1,12 @@
+version: "1.0.0"
+content: |
+  You have a lot of background knowledge, and as such you are some kind of 
+  'encyclopedia' to explain general terminology. Only answer if you have a clear understanding of the question. 
+  If not, say you do not have sufficient information to answer the question. Use the {language} in your communication.
+  Question:
+  {question}
+metadata:
+  author: "Josako"
+  date_added: "2024-11-10"
+  description: "A background information retriever for Evie"
+  changes: "Initial version migrated from flat file structure"

config/prompts/history/1.0.0.yaml

@@ -0,0 +1,16 @@
+version: "1.0.0"
+content: |
+  You are a helpful assistant that details a question based on a previous context,
+  in such a way that the question is understandable without the previous context. 
+  The context is a conversation history, with the HUMAN asking questions, the AI answering questions.
+  The history is delimited between triple backquotes.
+  You answer by stating the question in {language}.
+  History:
+  ```{history}```
+  Question to be detailed:
+  {question}
+metadata:
+  author: "Josako"
+  date_added: "2024-11-10"
+  description: "Prompt to further detail a question based on the previous conversation"
+  changes: "Initial version migrated from flat file structure"

config/prompts/html_parse/1.0.0.yaml

@@ -0,0 +1,20 @@
+version: "1.0.0"
+content: |
+  You are a top administrative assistant specialized in transforming given HTML into markdown formatted files. The generated files will be used to generate embeddings in a RAG-system.
+  
+  # Best practices are:
+  - Respect wordings and language(s) used in the HTML.
+  - The following items need to be considered: headings, paragraphs, listed items (numbered or not) and tables. Images can be neglected.
+  - Sub-headers can be used as lists. This is true when a header is followed by a series of sub-headers without content (paragraphs or listed items). Present those sub-headers as a list.  
+  - Be careful of encoding of the text. Everything needs to be human readable.
+
+  Process the file carefully, and take a stepped approach. The resulting markdown should be the result of the processing of the complete input html file. Answer with the pure markdown, without any other text.
+
+  HTML is between triple backquotes.
+
+  ```{html}```
+metadata:
+  author: "Josako"
+  date_added: "2024-11-10"
+  description: "An aid in transforming HTML-based inputs to markdown"
+  changes: "Initial version migrated from flat file structure"

config/prompts/pdf_parse/1.0.0.yaml

@@ -0,0 +1,23 @@
+version: "1.0.0"
+content: |
+  You are a top administrative aid specialized in transforming given PDF-files into markdown formatted files. The generated files will be used to generate embeddings in a RAG-system.
+  The content you get is already processed (some markdown already generated), but needs to be corrected. For large files, you may receive only portions of the full file. Consider this when processing the content.
+
+  # Best practices are:
+  - Respect wordings and language(s) used in the provided content.
+  - The following items need to be considered: headings, paragraphs, listed items (numbered or not) and tables. Images can be neglected.
+  - When headings are numbered, show the numbering and define the header level. You may have to correct current header levels, as preprocessing is known to make errors.
+  - A new item is started when a <return> is found before a full line is reached. In order to know the number of characters in a line, please check the document and the context within the document (e.g. an image could limit the number of characters temporarily).
+  - Paragraphs are to be stripped of newlines so they become easily readable.
+  - Be careful of encoding of the text. Everything needs to be human readable.
+
+  Process the file carefully, and take a stepped approach. The resulting markdown should be the result of the processing of the complete input pdf content. Answer with the pure markdown, without any other text.
+
+  PDF content is between triple backquotes.
+
+  ```{pdf_content}```
+metadata:
+  author: "Josako"
+  date_added: "2024-11-10"
+  description: "An assistant to parse PDF-content into markdown"
+  changes: "Initial version migrated from flat file structure"

config/prompts/rag/1.0.0.yaml

@@ -0,0 +1,15 @@
+version: "1.0.0"
+content: |
+  Answer the question based on the following context, delimited between triple backquotes. 
+  {tenant_context}
+  Use the following {language} in your communication, and cite the sources used at the end of the full conversation.
+  If the question cannot be answered using the given context, say "I have insufficient information to answer this question."
+  Context:
+  ```{context}```
+  Question:
+  {question}
+metadata:
+  author: "Josako"
+  date_added: "2024-11-10"
+  description: "The Main RAG retriever"
+  changes: "Initial version migrated from flat file structure"

config/prompts/summary/1.0.0.yaml

@@ -0,0 +1,9 @@
+version: "1.0.0"
+content: |
+  Write a concise summary of the text in {language}. The text is delimited between triple backquotes.
+  ```{text}```
+metadata:
+  author: "Josako"
+  date_added: "2024-11-10"
+  description: "An assistant to create a summary when multiple chunks are required for 1 file"
+  changes: "Initial version migrated from flat file structure"

config/prompts/transcript/1.0.0.yaml

@@ -0,0 +1,25 @@
+version: "1.0.0"
+content: |
+  You are a top administrative assistant specialized in transforming given transcriptions into markdown formatted files. The generated files will be used to generate embeddings in a RAG-system. The transcriptions originate from podcast, videos and similar material.
+  You may receive information in different chunks. If you're not receiving the first chunk, you'll get the last part of the previous chunk, including it's title in between triple $. Consider this last part and the title as the start of the new chunk.
+
+
+  # Best practices and steps are:
+  - Respect wordings and language(s) used in the transcription. Main language is {language}.
+  - Sometimes, the transcript contains speech of several people participating in a conversation. Although these are not obvious from reading the file, try to detect when other people are speaking.    
+  - Divide the transcript into several logical parts. Ensure questions and their answers are in the same logical part. Don't make logical parts too small. They should contain at least 7 or 8 sentences.
+  - annotate the text to identify these logical parts using headings in {language}.
+  - improve errors in the transcript given the context, but do not change the meaning and intentions of the transcription.
+
+  Process the file carefully, and take a stepped approach. The resulting markdown should be the result of processing the complete input transcription. Answer with the pure markdown, without any other text.
+
+  The transcript is between triple backquotes.
+
+  $$${previous_part}$$$
+
+  ```{transcript}```
+metadata:
+  author: "Josako"
+  date_added: "2024-11-10"
+  description: "An assistant to transform a transcript to markdown."
+  changes: "Initial version migrated from flat file structure"

config/retrievers/STANDARD_RAG/1.0.0.yaml

@@ -0,0 +1,26 @@
+version: "1.0.0"
+name: "Standard RAG Retriever"
+configuration:
+  es_k:
+    name: "es_k"
+    type: "int"
+    description: "K-value to retrieve embeddings (max embeddings retrieved)"
+    required: true
+    default: 8
+  es_similarity_threshold:
+    name: "es_similarity_threshold"
+    type: "float"
+    description: "Similarity threshold for retrieving embeddings"
+    required: true
+    default: 0.3
+arguments:
+  query:
+    name: "query"
+    type: "str"
+    description: "Query to retrieve embeddings"
+    required: True
+metadata:
+  author: "Josako"
+  date_added: "2025-01-24"
+  changes: "Initial version"
+  description: "Retrieving all embeddings conform the query"

config/specialists/SPIN_SPECIALIST/1.0.0.yaml

@@ -1,4 +1,4 @@
-version: 1.0.0
+version: "1.0.0"
 name: "Spin Sales Specialist"
 framework: "crewai"
 configuration:
@@ -31,6 +31,12 @@ configuration:
     type: "str"
     description: "The language code used for internal information. If not provided, the tenant's default language will be used"
     required: false
+  nr_of_questions:
+    name: "nr_of_questions"
+    type: "int"
+    description: "The maximum number of questions to formulate extra questions"
+    required: true
+    default: 3
 arguments:
   language:
     name: "Language"
@@ -48,11 +54,7 @@ arguments:
     description: "Initial identification information when available"
     required: false
 results:
-  detailed_query:
-    name: "detailed_query"
-    type: "str"
-    description: "The query detailed with the Chat Session History."
-    required: true
+  rag_output:
     answer:
       name: "answer"
       type: "str"
@@ -69,40 +71,96 @@ results:
       description: "Whether or not the query is insufficient info"
       required: true
   spin:
-    situation_information:
-      name: "situation_information"
-      type: "List[str]"
-      description: "A list of situation descriptions"
+    situation:
+      name: "situation"
+      type: "str"
+      description: "A description of the customer's current situation / context"
       required: false
-    problem_information:
-      name: "problem_information"
-      type: "List[str]"
-      description: "A list of problems"
+    problem:
+      name: "problem"
+      type: "str"
+      description: "The current problems the customer is facing, for which he/she seeks a solution"
       required: false
-    implication_information:
-      name: "implication_information"
-      type: "List[str]"
+    implication:
+      name: "implication"
+      type: "str"
       description: "A list of implications"
       required: false
-    needs_information:
-      name: "needs_information"
-      type: "List[str]"
+    needs:
+      name: "needs"
+      type: "str"
       description: "A list of needs"
       required: false
+    additional_info:
+      name: "additional_info"
+      type: "str"
+      description: "Additional information that may be commercially interesting"
+      required: false
+  lead_info:
+    lead_personal_info:
+      name:
+        name: "name"
+        type: "str"
+        description: "name of the lead"
+        required: "true"
+      job_title:
+        name: "job_title"
+        type: "str"
+        description: "job title"
+        required: false
+      email:
+        name: "email"
+        type: "str"
+        description: "lead email"
+        required: "false"
+      phone:
+        name: "phone"
+        type: "str"
+        description: "lead phone"
+        required: false
+      additional_info:
+        name: "additional_info"
+        type: "str"
+        description: "additional info on the lead"
+        required: false
+    lead_company_info:
+      company_name:
+        name: "company_name"
+        type: "str"
+        description: "Name of the lead company"
+        required: false
+      industry:
+        name: "industry"
+        type: "str"
+        description: "The industry of the company"
+        required: false
+      company_size:
+        name: "company_size"
+        type: "int"
+        description: "The size of the company"
+        required: false
+      company_website:
+        name: "company_website"
+        type: "str"
+        description: "The main website for the company"
+        required: false
+      additional_info:
+        name: "additional_info"
+        type: "str"
+        description: "Additional information that may be commercially interesting"
+        required: false
 agents:
   - type: "RAG_AGENT"
     version: "1.0"
-    name: "Default RAG Agent"   # Just added as an example. Overwrites the default agent name.
-    description: "An Agent that does RAG based on a user's question, RAG content & history"   # Just added as an example. Overwrites the default agent description.
+  - type: "RAG_COMMUNICATION_AGENT"
+    version: "1.0"
   - type: "SPIN_DETECTION_AGENT"
     version: "1.0"
   - type: "SPIN_SALES_SPECIALIST_AGENT"
     version: "1.0"
   - type: "IDENTIFICATION_AGENT"
     version: "1.0"
-  - type: "EMAIL_CONTENT_AGENT"
-    version: "1.0"
-  - type: "EMAIL_ENGAGEMENT_AGENT"
+  - type: "RAG_COMMUNICATION_AGENT"
     version: "1.0"
 tasks:
   - type: "RAG_TASK"
@@ -115,9 +173,7 @@ tasks:
     version: "1.0"
   - type: "IDENTIFICATION_QUESTIONS_TASK"
     version: "1.0"
-  - type: "EMAIL_LEAD_DRAFTING_TASK"
-    version: "1.0"
-  - type: "EMAIL_LEAD_ENGAGEMENT_TASK"
+  - type: "RAG_CONSOLIDATION_TASK"
     version: "1.0"
 metadata:
   author: "Josako"

config/tasks/EMAIL_LEAD_DRAFTING_TASK/1.0.0.yaml

@@ -27,6 +27,7 @@ expected_output: >
     - Addresses the lead by name
     - Acknowledges their role and company
     - Highlights how {company} can meet their specific needs or interests
+  {customer_expected_output}
 metadata:
   author: "Josako"
   date_added: "2025-01-08"

config/tasks/IDENTIFICATION_DETECTION_TASK/1.0.0.yaml

@@ -1,13 +1,22 @@
 version: "1.0.0"
 name: "Identification Gathering"
 task_description: >
-  Detect and pass on identification information in the ongoing conversation, from within the following information:
-    {question}
-  Add to or refine the following already gathered identification information (between triple $)
-    $$${Identification}$$$
+  You are asked to gather lead information in a conversation with a new prospect. This is information about the person
+  participating in the conversation, and information on the company he or she is working for. Try to be as precise as
+  possible.
+  Take into account information already gathered in the history (between triple backquotes) and add information found in
+  the latest reply.
+  
+  history:
+    ```{history}```
+  latest reply:
+    {query}
+  
   {custom_description}
 expected_output: >
-  Identification information such as name, email, phone number, company, role, company website, ...
+  - Personal Identification information such as name, email, phone number, job title, and any additional information that
+  may prove to be interesting in the current or future conversations.
+  - Company information such as company name, industry, size, company website, ...
   {custom_expected_output}
 metadata:
   author: "Josako"

config/tasks/IDENTIFICATION_QUESTIONS_TASK/1.0.0.yaml

@@ -1,12 +1,21 @@
 version: "1.0.0"
 name: "Define Identification Questions"
 task_description: >
-  Ask questions to complete or confirm the identification information gathered.
-  Current Identification Information:
-    $$${Identification}$$$
+  Gather the identification information gathered by your team mates , take into account the history (in between triple 
+  backquotes) of the conversation, and the latest reply of the user. 
+  Define questions to be asked to complete the personal and company information for the end user in the conversation.
+  history:
+    ```{history}```
+  latest reply:
+    {query}
+  
   {custom_description}
 expected_output: >
-  Top 2 questions to ask in order to complete identification.
+  - Personal Identification information such as name, email, phone number, job title, and any additional information that
+    may prove to be interesting in the current or future conversations.
+  - Company information such as company name, industry, size, company website, ...
+    {custom_expected_output}
+  - Top {nr_of_questions} questions to ask in order to complete identification.
   {custom_expected_output}
 metadata:
   author: "Josako"

config/tasks/RAG_CONSOLIDATION_TASK/1.0.0.yaml

@@ -0,0 +1,23 @@
+version: "1.0.0"
+name: "Rag Consolidation"
+task_description: >
+  Your teams have collected answers to a user's query (in between triple backquotes), and collected additional follow-up
+  questions (in between triple %) to reach their goals. Ensure the answers are provided, and select the additional 
+  questions to be asked in order not to overwhelm the user. Make a selection of maximum {nr_of_questions} questions to
+  be returned to the user. You ensure both answers and additional questions are bundled into 1 clear communication back 
+  to the user. Use {language} for your consolidated communication.
+  {custom_description}
+  
+  Anwers:
+  ```{prepared_answers}```
+  
+  Additional Questions:
+  %%%{additional_questions}%%%
+expected_output: >
+  One consolidated communication towards the end user with both answers and maximum {nr_of_questions} questions.
+  {custom_expected_output}
+metadata:
+  author: "Josako"
+  date_added: "2025-01-08"
+  description: "A Task to consolidate questions and answers"
+  changes: "Initial version"

config/tasks/RAG_TASK/1.0.0.yaml

@@ -1,21 +1,21 @@
 version: "1.0.0"
 name: "RAG Task"
 task_description: >
-  Answer the question based on the following context, delimited between triple backquotes, and taking into account
+  Answer the query based on the following context, delimited between triple backquotes, and taking into account
   the history of the discussion, in between triple %
   {custom_description}
   Use the following {language} in your communication, and cite the sources used at the end of the full conversation.
-  If the question cannot be answered using the given context, say "I have insufficient information to answer this question."
+  If the question cannot be answered using the given context, answer "I have insufficient information to answer this question."
   Context:
     ```{context}```
   History:
     %%%{history}%%%
-  Question:
-    {question}
+  Query:
+    {query}
 expected_output: >
-  An answer to the question asked formatted in markdown, without '```'.
-  A list of sources used in generating the answer.
-  An indication (True or False) of your ability to provide an answer.
+  - Answer
+  - A list of sources used in generating the answer, citations
+  - An indication (True or False) if there's insufficient information to give an answer.
 metadata:
   author: "Josako"
   date_added: "2025-01-08"

config/tasks/SPIN_DETECT_TASK/1.0.0.yaml

@@ -1,25 +1,22 @@
 version: "1.0.0"
 name: "SPIN Information Detection"
 task_description: >
-  Detect the SPIN-context, taking into account the history of the discussion (in between triple %) with main focus on 
-  the latest reply (which can contain answers on previously asked questions by the user). Do not remove elements from
-  the known SPIN (in between triple $) analysis unless explicitly stated by the end user in the latest reply. In all other cases, refine the
-  current SPIN analysis or add elements to it.
+  Detect the SPIN-context, taking into account the history of the discussion (in between triple backquotes) with main focus on 
+  the latest reply (which can contain answers on previously asked questions by the user). Spin elements may already be 
+  provided in the history. Add or refine these with the new input provided in the latest reply of the end user.
   {custom_description}
-  Use the following {tenant_language} to define the SPIN-elements. If no additional information can be added, just
-  return the already known SPIN.
+  Use the following {tenant_language} to define the SPIN-elements. 
   History:
-    %%%{history}%%%
-  Known SPIN:
-    $$${SPIN}$$$
+    ```{history}```
   Latest reply:
-    {question}
+    {query}
 expected_output: >
   The SPIN analysis, comprised of:
-  - Situation information - Information to understanding the customer's current context, as a markdown list without '```'.
-  - Problem information - Information on uncovering the customer's challenges and pain points, as a markdown list without '```'.
-  - Implication information - Exploration of the consequences of those problems, as a markdown list without '```'.
-  - Need-payoff information - Helping customers realize value of solutions and defining their direct needs, as a markdown list without '```'.
+  - Situation information: a description of the customer's current context / situation.
+  - Problem information: a description of the customer's problems uncovering it's challenges and pain points.
+  - Implication information: implications of situation / identified problems, i.e. of the consequences of those problems.
+  - Need-payoff information: Customer's needs, helping customers realize value of solutions.
+  - Additional info: Information that does not fit in the above SPIN-categories, but that can be commercially interesting.
   {custom_expected_output}
 metadata:
   author: "Josako"

config/tasks/SPIN_QUESTIONS_TASK/1.0.0.yaml

@@ -1,21 +1,25 @@
 version: "1.0.0"
 name: "SPIN Question Identification"
 task_description: >
-  Define, taking into account the history of the discussion (in between triple %), the latest reply and the currently 
-  known SPIN-elements (in between triple $), the top questions that need to be asked to understand the full SPIN context
+  Define, taking into account the history of the discussion (in between triple backquotes) and the latest reply and the 
+  currently known SPIN-elements, the top questions that need to be asked to understand the full SPIN context
   of the customer. If you think this user could be a potential customer, please indicate so.
   {custom_description}
   Use the following {tenant_language} to define the SPIN-elements. If you have a full SPIN context, just skip and don't
   ask for more information or confirmation.
   History:
-      %%%{history}%%%
-  Known SPIN:
-    $$${SPIN}$$$
+      ```{history}```
   Latest reply:
-    {question}
+    {query}
 expected_output: >
+  The SPIN analysis, comprised of:
+   - Situation information: a description of the customer's current context / situation.
+   - Problem information: a description of the customer's problems uncovering it's challenges and pain points.
+   - Implication information: implications of situation / identified problems, i.e. of the consequences of those problems.
+   - Need-payoff information: Customer's needs, helping customers realize value of solutions.
+   - Additional info: Information that does not fit in the above SPIN-categories, but that can be commercially interesting.
   The SPIN questions:
-  - At max {nr_of_spin_questions} questions to complete the SPIN-context of the customer, as a markdown list without '```'.
+  - At max {nr_of_questions} questions to complete the SPIN-context of the customer, as a markdown list without '```'.
   Potential Customer Indication:
   - An indication if - given the current SPIN - this could be a good customer (True) or not (False).
   {custom_expected_output}

config/type_defs/agent_types.py

@@ -16,6 +16,10 @@ AGENT_TYPES = {
         "name": "Rag Agent",
         "description": "An Agent that does RAG based on a user's question, RAG content & history",
     },
+    "RAG_COMMUNICATION_AGENT": {
+        "name": "Rag Communication Agent",
+        "description": "An Agent that consolidates both answers and questions in a consistent reply",
+    },
     "SPIN_DETECTION_AGENT": {
         "name": "SPIN Sales Assistant",
         "description": "An Agent that detects SPIN information in an ongoing conversation",

config/type_defs/prompt_types.py

@@ -0,0 +1,31 @@
+# Agent Types
+PROMPT_TYPES = {
+    "encyclopedia": {
+        "name": "encyclopedia",
+        "description": "A background information retriever for Evie",
+    },
+    "history": {
+        "name": "history",
+        "description": "Prompt to further detail a question based on the previous conversation",
+    },
+    "html_parse": {
+        "name": "html_parse",
+        "description": "An aid in transforming HTML-based inputs to markdown",
+    },
+    "pdf_parse": {
+        "name": "pdf_parse",
+        "description": "An assistant to parse PDF-content into markdown",
+    },
+    "rag": {
+        "name": "rag",
+        "description": "The Main RAG retriever",
+    },
+    "summary": {
+        "name": "summary",
+        "description": "An assistant to create a summary when multiple chunks are required for 1 file",
+    },
+    "transcript": {
+        "name": "transcript",
+        "description": "An assistant to transform a transcript to markdown.",
+    },
+}

config/type_defs/retriever_types.py

@@ -2,30 +2,6 @@
 RETRIEVER_TYPES = {
     "STANDARD_RAG": {
         "name": "Standard RAG Retriever",
-        "description": "Retrieving all embeddings conform the query",
-        "configuration": {
-            "es_k": {
-                "name": "es_k",
-                "type": "int",
-                "description": "K-value to retrieve embeddings (max embeddings retrieved)",
-                "required": True,
-                "default": 8,
-            },
-            "es_similarity_threshold": {
-                "name": "es_similarity_threshold",
-                "type": "float",
-                "description": "Similarity threshold for retrieving embeddings",
-                "required": True,
-                "default": 0.3,
-            },
-        },
-        "arguments": {
-            "query": {
-                "name": "query",
-                "type": "str",
-                "description": "Query to retrieve embeddings",
-                "required": True,
-            },
-        }
+        "description": "Retrieving all embeddings from the catalog conform the query",
     }
 }

config/type_defs/service_types.py

@@ -11,5 +11,9 @@ SERVICE_TYPES = {
     "DEPLOY_API": {
         "name": "DEPLOY_API",
         "description": "Service allows to use deployment API functionality.",
+    },
+    "SPECIALIST_API": {
+        "name": "SPECIALIST_API",
+        "description": "Service allows to use specialist execution API functionality.",
     }
 }

config/type_defs/specialist_types.py

@@ -1,6 +1,6 @@
 # Specialist Types
 SPECIALIST_TYPES = {
-    "STANDARD_RAG": {
+    "STANDARD_RAG_SPECIALIST": {
         "name": "Q&A RAG Specialist",
         "description": "Standard Q&A through RAG Specialist",
     },

config/type_defs/task_types.py

@@ -28,4 +28,8 @@ TASK_TYPES = {
         "name": "SPIN Question Identification",
         "description": "A Task that identifies questions to complete the SPIN context in a conversation",
     },
+    "RAG_CONSOLIDATION_TASK": {
+        "name": "RAG Consolidation",
+        "description": "A Task to consolidate questions and answers",
+    }
 }

docker/compose_dev.yaml

@@ -37,6 +37,7 @@ x-common-variables: &common-variables
   NGINX_SERVER_NAME: 'localhost http://macstudio.ask-eve-ai-local.com/'
   LANGCHAIN_API_KEY: "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"
   SERPER_API_KEY: "e4c553856d0e6b5a171ec5e6b69d874285b9badf"
+  CREWAI_STORAGE_DIR: "/app/crewai_storage"
 
 services:
   nginx:

docker/compose_stackhero.yaml

@@ -41,6 +41,7 @@ x-common-variables: &common-variables
   NGINX_SERVER_NAME: 'evie.askeveai.com mxz536.stackhero-network.com'
   LANGCHAIN_API_KEY: "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"
   SERPER_API_KEY: "e4c553856d0e6b5a171ec5e6b69d874285b9badf"
+  CREWAI_STORAGE_DIR: "/app/crewai_storage"
 
 networks:
   eveai-network:

docker/eveai_chat_workers/Dockerfile

@@ -39,6 +39,7 @@ RUN apt-get update && apt-get install -y \
 
 # Create logs directory and set permissions
 RUN mkdir -p /app/logs && chown -R appuser:appuser /app/logs
+RUN mkdir -p /app/crewai_storage && chown -R appuser:appuser /app/crewai_storage
 
 # Download dependencies as a separate step to take advantage of Docker's caching.
 # Leverage a cache mount to /root/.cache/pip to speed up subsequent builds.

eveai_api/__init__.py

@@ -15,6 +15,7 @@ from common.utils.database import Database
 from config.logging_config import LOGGING
 from .api.document_api import document_ns
 from .api.auth import auth_ns
+from .api.specialist_execution_api import specialist_execution_ns
 from config.config import get_config
 from common.utils.celery_utils import make_celery, init_celery
 from common.utils.eveai_exceptions import EveAIException
@@ -127,7 +128,7 @@ def register_extensions(app):
             "expose_headers": ["Content-Length", "Content-Range"],
             "supports_credentials": True,
             "max_age": 1728000,  # 20 days
-            "allow_credentials": True
+            # "allow_credentials": True
         }
     })
 
@@ -135,6 +136,7 @@ def register_extensions(app):
 def register_namespaces(app):
     api_rest.add_namespace(document_ns, path='/api/v1/documents')
     api_rest.add_namespace(auth_ns, path='/api/v1/auth')
+    api_rest.add_namespace(specialist_execution_ns, path='/api/v1/specialist-execution')
 
 
 def register_blueprints(app):

eveai_api/api/specialist_execution_api.py

@@ -0,0 +1,89 @@
+# eveai_api/api/specialist_execution_api.py
+import uuid
+
+from flask import Response, stream_with_context, current_app
+from flask_restx import Namespace, Resource, fields
+from flask_jwt_extended import jwt_required, get_jwt_identity
+
+from common.utils.celery_utils import current_celery
+from common.utils.execution_progress import ExecutionProgressTracker
+from eveai_api.api.auth import requires_service
+
+specialist_execution_ns = Namespace('specialist-execution', description='Specialist execution operations')
+
+specialist_start_session_response = specialist_execution_ns.model('StartSessionResponse', {
+    'session_id': fields.String(required=True, description='A new Chat session ID'),
+})
+
+
+@specialist_execution_ns.route('/start_session', methods=['GET'])
+class StartSession(Resource):
+    @jwt_required()
+    @requires_service("SPECIALIST_API")
+    @specialist_execution_ns.response(201, 'New Session ID created Successfully', specialist_start_session_response)
+    def get(self):
+        new_session_id = f"{uuid.uuid4()}"
+        return {
+            'session_id': new_session_id,
+        }, 201
+
+
+specialist_execution_input = specialist_execution_ns.model('SpecialistExecutionInput', {
+    'specialist_id': fields.Integer(required=True, description='ID of the specialist to use'),
+    'arguments': fields.Raw(required=True, description='Dynamic arguments for specialist and retrievers'),
+    'session_id': fields.String(required=True, description='Chat session ID'),
+    'user_timezone': fields.String(required=True, description='User timezone')
+})
+
+specialist_execution_response = specialist_execution_ns.model('SpecialistExecutionResponse', {
+    'task_id': fields.String(description='ID of specialist execution task, to be used to retrieve execution stream'),
+    'status': fields.String(description='Status of the execution'),
+    'stream_url': fields.String(description='Stream URL'),
+})
+
+
+@specialist_execution_ns.route('')
+class StartExecution(Resource):
+    @jwt_required()
+    @requires_service('SPECIALIST_API')
+    @specialist_execution_ns.expect(specialist_execution_input)
+    @specialist_execution_ns.response(201, 'Specialist execution successfully queued.', specialist_execution_response)
+    def post(self):
+        """Start execution of a specialist"""
+        tenant_id = get_jwt_identity()
+        data = specialist_execution_ns.payload
+
+        # Send task to queue
+        task = current_celery.send_task(
+            'execute_specialist',
+            args=[tenant_id,
+                  data['specialist_id'],
+                  data['arguments'],
+                  data['session_id'],
+                  data['user_timezone'],
+                  ],
+            queue='llm_interactions'
+        )
+
+        return {
+            'task_id': task.id,
+            'status': 'queued',
+            'stream_url': f'/api/v1/specialist-execution/{task.id}/stream'
+        }, 201
+
+
+@specialist_execution_ns.route('/<string:task_id>/stream')
+class ExecutionStream(Resource):
+    @jwt_required()
+    @requires_service('SPECIALIST_API')
+    def get(self, task_id: str):
+        """Get streaming updates for a specialist execution"""
+        progress_tracker = ExecutionProgressTracker()
+        return Response(
+            stream_with_context(progress_tracker.get_updates(task_id)),
+            mimetype='text/event-stream',
+            headers={
+                'Cache-Control': 'no-cache',
+                'Connection': 'keep-alive'
+            }
+        )

eveai_app/__init__.py

@@ -160,30 +160,10 @@ def register_blueprints(app):
 
 
 def register_cache_handlers(app):
-    from common.utils.cache.config_cache import (
-        AgentConfigCacheHandler, AgentConfigTypesCacheHandler, AgentConfigVersionTreeCacheHandler,
-        TaskConfigCacheHandler, TaskConfigTypesCacheHandler, TaskConfigVersionTreeCacheHandler,
-        ToolConfigCacheHandler, ToolConfigTypesCacheHandler, ToolConfigVersionTreeCacheHandler,
-        SpecialistConfigCacheHandler, SpecialistConfigTypesCacheHandler, SpecialistConfigVersionTreeCacheHandler,)
-
-    cache_manager.register_handler(AgentConfigCacheHandler, 'eveai_config')
-    cache_manager.register_handler(AgentConfigTypesCacheHandler, 'eveai_config')
-    cache_manager.register_handler(AgentConfigVersionTreeCacheHandler, 'eveai_config')
-    cache_manager.register_handler(TaskConfigCacheHandler, 'eveai_config')
-    cache_manager.register_handler(TaskConfigTypesCacheHandler, 'eveai_config')
-    cache_manager.register_handler(TaskConfigVersionTreeCacheHandler, 'eveai_config')
-    cache_manager.register_handler(ToolConfigCacheHandler, 'eveai_config')
-    cache_manager.register_handler(ToolConfigTypesCacheHandler, 'eveai_config')
-    cache_manager.register_handler(ToolConfigVersionTreeCacheHandler, 'eveai_config')
-    cache_manager.register_handler(SpecialistConfigCacheHandler, 'eveai_config')
-    cache_manager.register_handler(SpecialistConfigTypesCacheHandler, 'eveai_config')
-    cache_manager.register_handler(SpecialistConfigVersionTreeCacheHandler, 'eveai_config')
-
-    cache_manager.agents_config_cache.set_version_tree_cache(cache_manager.agents_version_tree_cache)
-    cache_manager.tasks_config_cache.set_version_tree_cache(cache_manager.tasks_version_tree_cache)
-    cache_manager.tools_config_cache.set_version_tree_cache(cache_manager.tools_version_tree_cache)
-    cache_manager.specialists_config_cache.set_version_tree_cache(cache_manager.specialists_version_tree_cache)
-
+    from common.utils.cache.config_cache import register_config_cache_handlers
+    register_config_cache_handlers(cache_manager)
+    from common.utils.cache.crewai_processed_config_cache import register_specialist_cache_handlers
+    register_specialist_cache_handlers(cache_manager)
 
 
 

eveai_app/views/interaction_forms.py

@@ -55,7 +55,7 @@ class EditSpecialistForm(DynamicFormBase):
 
     type = StringField('Specialist Type', validators=[DataRequired()], render_kw={'readonly': True})
     type_version = StringField('Type Version', validators=[DataRequired()], render_kw={'readonly': True})
-    tuning = BooleanField('Enable Retrieval Tuning', default=False)
+    tuning = BooleanField('Enable Specialist Tuning', default=False)
 
 
 class BaseComponentForm(DynamicFormBase):

eveai_app/views/interaction_views.py

@@ -141,7 +141,6 @@ def specialist():
 
             # Populate fields individually instead of using populate_obj (gives problem with QueryMultipleSelectField)
             new_specialist.name = form.name.data
-            new_specialist.description = form.description.data
             new_specialist.type = form.type.data
             new_specialist.type_version = cache_manager.specialists_version_tree_cache.get_latest_version(new_specialist.type)
             new_specialist.tuning = form.tuning.data

eveai_chat_workers/__init__.py

@@ -33,6 +33,8 @@ def create_app(config_file=None):
     celery = make_celery(app.name, app.config)
     init_celery(celery, app)
 
+    register_cache_handlers(app)
+
     from eveai_chat_workers import tasks
     print(tasks.tasks_ping())
 
@@ -45,5 +47,14 @@ def register_extensions(app):
     template_manager.init_app(app)
 
 
+def register_cache_handlers(app):
+    from common.utils.cache.config_cache import register_config_cache_handlers
+    register_config_cache_handlers(cache_manager)
+    from common.utils.cache.crewai_processed_config_cache import register_specialist_cache_handlers
+    register_specialist_cache_handlers(cache_manager)
+    from eveai_chat_workers.chat_session_cache import register_chat_session_cache_handlers
+    register_chat_session_cache_handlers(cache_manager)
+    
+
 app, celery = create_app()
 

eveai_chat_workers/chat_session_cache.py

@@ -60,7 +60,6 @@ class ChatSessionCacheHandler(CacheHandler[CachedSession]):
                 .filter_by(session_id=session_id)
                 .first()
             )
-
             if not session:
                 if not create_params:
                     raise ValueError(f"Chat session {session_id} not found and no creation parameters provided")
@@ -90,13 +89,13 @@ class ChatSessionCacheHandler(CacheHandler[CachedSession]):
                 for interaction in session.interactions
                 if interaction.specialist_results is not None  # Only include completed interactions
             ]
-
-            return CachedSession(
+            cached_session = CachedSession(
                 id=session.id,
                 session_id=session_id,
                 interactions=cached_interactions,
                 timezone=session.timezone
             )
+            return cached_session
 
         return self.get(creator_func, session_id=session_id)
 
@@ -126,16 +125,17 @@ class ChatSessionCacheHandler(CacheHandler[CachedSession]):
                 )
             )
 
-            # Force cache update
-            self.invalidate(session_id=session_id)
+            # Update cache directly with modified session using region's set()
+            key = self.generate_key(session_id=session_id)
+            self.region.set(key, self._to_cache_data(cached_session))
 
         except ValueError:
             # If session not in cache yet, load it fresh from DB
             self.get_cached_session(session_id)
 
-    def to_cache_data(self, instance: CachedSession) -> Dict[str, Any]:
+    def _to_cache_data(self, instance: CachedSession) -> Dict[str, Any]:
         """Convert CachedSession to cache data"""
-        return {
+        cached_data = {
             'id': instance.id,
             'session_id': instance.session_id,
             'timezone': instance.timezone,
@@ -148,8 +148,9 @@ class ChatSessionCacheHandler(CacheHandler[CachedSession]):
             ],
             'last_updated': dt.now(tz=tz.utc).isoformat()
         }
+        return cached_data
 
-    def from_cache_data(self, data: Dict[str, Any], session_id: str, **kwargs) -> CachedSession:
+    def _from_cache_data(self, data: Dict[str, Any], session_id: str, **kwargs) -> CachedSession:
         """Create CachedSession from cache data"""
         interactions = [
             CachedInteraction(
@@ -166,14 +167,14 @@ class ChatSessionCacheHandler(CacheHandler[CachedSession]):
             timezone=data['timezone']
         )
 
-    def should_cache(self, value: Dict[str, Any]) -> bool:
+    def _should_cache(self, value: Dict[str, Any]) -> bool:
         """Validate cache data"""
-        required_fields = {'id','session_id', 'timezone', 'interactions'}
+        required_fields = {'id', 'session_id', 'timezone', 'interactions'}
         return all(field in value for field in required_fields)
 
 
-# Register the handler with the cache manager
-cache_manager.register_handler(ChatSessionCacheHandler, 'eveai_chat_workers')
+def register_chat_session_cache_handlers(cache_manager):
+    cache_manager.register_handler(ChatSessionCacheHandler, 'eveai_chat_workers')
 
 
 # Helper function similar to get_model_variables

eveai_chat_workers/outputs/identification/identification_v1_0.py

@@ -0,0 +1,45 @@
+from typing import Optional, List
+
+from pydantic import BaseModel, Field
+
+
+class LeadPersonalInfo(BaseModel):
+    name: Optional[str] = Field(None, description="The full name of the lead.")
+    job_title: Optional[str] = Field(None, description="The job title of the lead.")
+    email: Optional[str] = Field(None, description="The email address of the lead.")
+    phone: Optional[str] = Field(None, description="The phone number of the lead.")
+    additional_info: Optional[str] = Field(None, description="Additional information about the lead.")
+
+
+class LeadCompanyInfo(BaseModel):
+    company_name: Optional[str] = Field(..., description="The name of the company the lead works for.")
+    company_website: Optional[str] = Field(None, description="The website of the company the lead works for.")
+    industry: Optional[str] = Field(..., description="The industry in which the company operates.")
+    company_size: Optional[int] = Field(..., description="The size of the company in terms of employee count.")
+    additional_info: Optional[str] = Field(None, description="Additional information about the lead's company.")
+
+
+class LeadInfoOutput(BaseModel):
+    personal_info: Optional[LeadPersonalInfo] = Field(None, description="Personal information of the lead.")
+    company_info: Optional[LeadCompanyInfo] = Field(None, description="Company information related to the lead.")
+    questions: Optional[str] = Field(None, description="Additional questions to further clarify Identification")
+
+    def __str__(self):
+        output = ""
+        if self.personal_info:
+            output += (f"PERSONAL INFO:\n\n"
+                       f"Name: {self.personal_info.name or 'N/A'}\n"
+                       f"Job Title: {self.personal_info.job_title or 'N/A'}\n"
+                       f"Email: {self.personal_info.email or 'N/A'}\n"
+                       f"Phone: {self.personal_info.phone or 'N/A'}\n"
+                       f"Additional Info: {self.personal_info.additional_info or 'N/A'}\n\n")
+
+        if self.company_info:
+            output += (f"COMPANY INFO:\n\n"
+                       f"Company Name: {self.company_info.company_name or 'N/A'}\n"
+                       f"Industry: {self.company_info.industry or 'N/A'}\n"
+                       f"Company Size: {self.company_info.company_size or 'N/A'}\n"
+                       f"Additional Info: {self.company_info.additional_info or 'N/A'}\n\n")
+
+        if self.questions:
+            output += f"QUESTIONS:\n\n{self.questions}\n\n"

eveai_chat_workers/outputs/rag/rag_v1_0.py

@@ -0,0 +1,9 @@
+from typing import List, Optional
+
+from pydantic import BaseModel, Field
+
+
+class RAGOutput(BaseModel):
+    answer: Optional[str] = Field(None, description="Answer to the questions asked")
+    citations: Optional[List[str]] = Field(None, description="A list of sources used in generating the answer")
+    insufficient_info: Optional[bool] = Field(None, description="An indication if there's insufficient information to answer")

eveai_chat_workers/outputs/spin/spin_v1_0.py

@@ -0,0 +1,24 @@
+from typing import List, Optional
+
+from pydantic import BaseModel, Field
+
+
+class SPINOutput(BaseModel):
+    situation: Optional[str] = Field(None, description="Situation information")
+    problem: Optional[str] = Field(None, description="Problem information")
+    implication: Optional[str] = Field(None, description="Implication information")
+    need: Optional[str] = Field(None, description="Need-payoff information")
+    additional_info: Optional[str] = Field(None, description="Additional sales-related information.")
+    questions: Optional[str] = Field(None, description="Additional questions to further clarify SPIN")
+    potential_customer: Optional[bool] = Field(False, description="Indication if this could be a good customer")
+
+    def __str__(self):
+        """Custom string output for usage in agents and tasks"""
+        return (f"Situation: {self.situation or 'N/A'}\n"
+                f"Problem: {self.problem or 'N/A'}\n"
+                f"Implication: {self.implication or 'N/A'}\n"
+                f"Need: {self.need or 'N/A'}\n"
+                f"Additional Info: {self.additional_info or 'N/A'}\n"
+                f"Questions: {self.questions or 'N/A'}\n"
+                f"Potential Customer: {self.potential_customer or 'N/A'}\n")
+

eveai_chat_workers/retrievers/base.py

@@ -36,7 +36,7 @@ class BaseRetriever(ABC):
             current_app.logger.error(f"Failed to setup tuning logger: {str(e)}")
             raise
 
-    def _log_tuning(self, message: str, data: Dict[str, Any] = None) -> None:
+    def log_tuning(self, message: str, data: Dict[str, Any] = None) -> None:
         if self.tuning and self.tuning_logger:
             try:
                 self.tuning_logger.log_tuning('retriever', message, data)

eveai_chat_workers/retrievers/retriever_typing.py

@@ -1,6 +1,9 @@
 from typing import Dict, Any
+
+from flask import current_app
 from pydantic import BaseModel, Field, model_validator
-from config.type_defs.retriever_types import RETRIEVER_TYPES
+from common.extensions import cache_manager
+
 
 
 class RetrieverMetadata(BaseModel):
@@ -28,6 +31,7 @@ class RetrieverArguments(BaseModel):
     based on RETRIEVER_TYPES configuration.
     """
     type: str = Field(..., description="Type of retriever (e.g. STANDARD_RAG)")
+    type_version: str = Field(..., description="Version of retriever type (e.g. 1.0)")
 
     # Allow any additional fields
     model_config = {
@@ -37,7 +41,7 @@ class RetrieverArguments(BaseModel):
     @model_validator(mode='after')
     def validate_required_arguments(self) -> 'RetrieverArguments':
         """Validate that all required arguments for this retriever type are present"""
-        retriever_config = RETRIEVER_TYPES.get(self.type)
+        retriever_config = cache_manager.retrievers_config_cache.get_config(self.type, self.type_version)
         if not retriever_config:
             raise ValueError(f"Unknown retriever type: {self.type}")
 

eveai_chat_workers/retrievers/standard_rag.py

@@ -30,7 +30,7 @@ class StandardRAGRetriever(BaseRetriever):
         self.tuning = retriever.tuning
         self.model_variables = get_model_variables(self.tenant_id)
 
-        self._log_tuning("Standard RAG retriever initialized")
+        self.log_tuning("Standard RAG retriever initialized")
 
     @property
     def type(self) -> str:
@@ -140,7 +140,7 @@ class StandardRAGRetriever(BaseRetriever):
                 compiled_query = str(query_obj.statement.compile(
                     compile_kwargs={"literal_binds": True}  # This will include the actual values in the SQL
                 ))
-                self._log_tuning('retrieve', {
+                self.log_tuning('retrieve', {
                     "arguments": arguments.model_dump(),
                     "similarity_threshold": self.similarity_threshold,
                     "k": self.k,

eveai_chat_workers/specialists/SPIN_SPECIALIST/1_0.py

@@ -0,0 +1,296 @@
+import json
+from os import wait
+from typing import Optional, List
+
+from crewai.flow.flow import start, listen, and_
+from flask import current_app
+from gevent import sleep
+from pydantic import BaseModel, Field
+
+from common.extensions import cache_manager
+from common.models.user import Tenant
+from common.utils.business_event_context import current_event
+from eveai_chat_workers.retrievers.retriever_typing import RetrieverArguments
+from eveai_chat_workers.specialists.crewai_base_specialist import CrewAIBaseSpecialistExecutor
+from eveai_chat_workers.specialists.specialist_typing import SpecialistResult, SpecialistArguments
+from eveai_chat_workers.outputs.identification.identification_v1_0 import LeadInfoOutput
+from eveai_chat_workers.outputs.spin.spin_v1_0 import SPINOutput
+from eveai_chat_workers.outputs.rag.rag_v1_0 import RAGOutput
+from eveai_chat_workers.specialists.crewai_base_classes import EveAICrewAICrew, EveAICrewAIFlow, EveAIFlowState
+from common.utils.pydantic_utils import flatten_pydantic_model
+
+
+class SpecialistExecutor(CrewAIBaseSpecialistExecutor):
+    """
+    type: SPIN_SPECIALIST
+    type_version: 1.0
+    SPIN Specialist Executor class
+    """
+
+    def __init__(self, tenant_id, specialist_id, session_id, task_id, **kwargs):
+        self.rag_crew = None
+        self.spin_crew = None
+        self.identification_crew = None
+        self.rag_consolidation_crew = None
+
+        super().__init__(tenant_id, specialist_id, session_id, task_id)
+
+        # Load the Tenant & set language
+        self.tenant = Tenant.query.get_or_404(tenant_id)
+        if self.specialist.configuration['tenant_language'] is None:
+            self.specialist.configuration['tenant_language'] = self.tenant.language
+
+    @property
+    def type(self) -> str:
+        return "SPIN_SPECIALIST"
+
+    @property
+    def type_version(self) -> str:
+        return "1.0"
+
+    def _config_task_agents(self):
+        self._add_task_agent("rag_task", "rag_agent")
+        self._add_task_agent("spin_detect_task", "spin_detection_agent")
+        self._add_task_agent("spin_questions_task", "spin_sales_specialist_agent")
+        self._add_task_agent("identification_detection_task", "identification_agent")
+        self._add_task_agent("identification_questions_task", "identification_agent")
+        self._add_task_agent("email_lead_drafting_task", "email_content_agent")
+        self._add_task_agent("email_lead_engagement_task", "email_engagement_agent")
+        self._add_task_agent("email_lead_retrieval_task", "email_engagement_agent")
+        self._add_task_agent("rag_consolidation_task", "rag_communication_agent")
+
+    def _config_pydantic_outputs(self):
+        self._add_pydantic_output("rag_task", RAGOutput, "rag_output")
+        self._add_pydantic_output("spin_questions_task", SPINOutput, "spin_questions")
+        self._add_pydantic_output("identification_questions_task", LeadInfoOutput, "lead_identification_questions")
+        self._add_pydantic_output("rag_consolidation_task", RAGOutput, "rag_output")
+
+    def _instantiate_specialist(self):
+        verbose = self.tuning
+
+        rag_agents = [self.rag_agent]
+        rag_tasks = [self.rag_task]
+        self.rag_crew = EveAICrewAICrew(
+            self,
+            "Rag Crew",
+            agents=rag_agents,
+            tasks=rag_tasks,
+            verbose=verbose,
+        )
+
+        spin_agents = [self.spin_detection_agent, self.spin_sales_specialist_agent]
+        spin_tasks = [self.spin_detect_task, self.spin_questions_task]
+        self.spin_crew = EveAICrewAICrew(
+            self,
+            "SPIN Crew",
+            agents=spin_agents,
+            tasks=spin_tasks,
+            verbose=verbose,
+        )
+
+        identification_agents = [self.identification_agent]
+        identification_tasks = [self.identification_detection_task, self.identification_questions_task]
+        self.identification_crew = EveAICrewAICrew(
+            self,
+            "Identification Crew",
+            agents=identification_agents,
+            tasks=identification_tasks,
+            verbose=verbose,
+        )
+
+        consolidation_agents = [self.rag_communication_agent]
+        consolidation_tasks = [self.rag_consolidation_task]
+        self.rag_consolidation_crew = EveAICrewAICrew(
+            self,
+            "Rag Consolidation Crew",
+            agents=consolidation_agents,
+            tasks=consolidation_tasks,
+            verbose=verbose,
+        )
+
+        self.flow = SPINFlow(
+            self,
+            self.rag_crew,
+            self.spin_crew,
+            self.identification_crew,
+            self.rag_consolidation_crew
+        )
+
+    def execute(self, arguments: SpecialistArguments) -> SpecialistResult:
+        formatted_context, citations = self.retrieve_context(arguments)
+
+        self.log_tuning("SPIN Specialist execution started", {})
+
+        flow_inputs = {
+            "language": arguments.language,
+            "query": arguments.query,
+            "context": formatted_context,
+            "citations": citations,
+            "history": self._formatted_history,
+            "name": self.specialist.configuration.get('name', ''),
+            "company": self.specialist.configuration.get('company', ''),
+            "products": self.specialist.configuration.get('products', ''),
+            "product_information": self.specialist.configuration.get('product_information', ''),
+            "engagement_options": self.specialist.configuration.get('engagement_options', ''),
+            "tenant_language": self.specialist.configuration.get('tenant_language', ''),
+            "nr_of_questions": self.specialist.configuration.get('nr_of_questions', ''),
+        }
+        # crew_results = self.rag_crew.kickoff(inputs=flow_inputs)
+        # current_app.logger.debug(f"Test Crew Output received: {crew_results}")
+        flow_results = self.flow.kickoff(inputs=flow_inputs)
+
+        flow_state = self.flow.state
+
+        results = SPINSpecialistResult.create_for_type(self.type, self.type_version)
+        update_data = {}
+        if flow_state.final_output:
+            update_data["rag_output"] = flow_state.final_output
+        elif flow_state.rag_output:     # Fallback
+            update_data["rag_output"] = flow_state.rag_output
+        if flow_state.spin:
+            update_data["spin"] = flow_state.spin
+        if flow_state.lead_info:
+            update_data["lead_info"] = flow_state.lead_info
+
+        results = results.model_copy(update=update_data)
+
+        self.log_tuning(f"SPIN Specialist execution ended", {"Results": results.model_dump()})
+
+        return results
+
+        # TODO: metrics
+
+
+class SPINSpecialistInput(BaseModel):
+    language: Optional[str] = Field(None, alias="language")
+    query: Optional[str] = Field(None, alias="query")
+    context: Optional[str] = Field(None, alias="context")
+    citations: Optional[List[int]] = Field(None, alias="citations")
+    history: Optional[str] = Field(None, alias="history")
+    name: Optional[str] = Field(None, alias="name")
+    company: Optional[str] = Field(None, alias="company")
+    products: Optional[str] = Field(None, alias="products")
+    product_information: Optional[str] = Field(None, alias="product_information")
+    engagement_options: Optional[str] = Field(None, alias="engagement_options")
+    tenant_language: Optional[str] = Field(None, alias="tenant_language")
+    nr_of_questions: Optional[int] = Field(None, alias="nr_of_questions")
+
+
+class SPINSpecialistResult(SpecialistResult):
+    rag_output: Optional[RAGOutput] = Field(None, alias="Rag Output")
+    spin: Optional[SPINOutput] = Field(None, alias="Spin Output")
+    lead_info: Optional[LeadInfoOutput] = Field(None, alias="Lead Info Output")
+
+
+class SPINFlowState(EveAIFlowState):
+    """Flow state for SPIN specialist that automatically updates from task outputs"""
+    input: Optional[SPINSpecialistInput] = None
+    rag_output: Optional[RAGOutput] = None
+    lead_info: Optional[LeadInfoOutput] = None
+    spin: Optional[SPINOutput] = None
+    final_output: Optional[RAGOutput] = None
+
+
+class SPINFlow(EveAICrewAIFlow[SPINFlowState]):
+    def __init__(self, specialist_executor, rag_crew, spin_crew, identification_crew, rag_consolidation_crew, **kwargs):
+        super().__init__(specialist_executor, "SPIN Specialist Flow", **kwargs)
+        self.specialist_executor = specialist_executor
+        self.rag_crew = rag_crew
+        self.spin_crew = spin_crew
+        self.identification_crew = identification_crew
+        self.rag_consolidation_crew = rag_consolidation_crew
+        self.exception_raised = False
+
+    @start()
+    def process_inputs(self):
+        return ""
+
+    @listen(process_inputs)
+    def execute_rag(self):
+        inputs = self.state.input.model_dump()
+        try:
+            crew_output = self.rag_crew.kickoff(inputs=inputs)
+            self.specialist_executor.log_tuning("RAG Crew Output", crew_output.model_dump())
+            output_pydantic = crew_output.pydantic
+            if not output_pydantic:
+                raw_json = json.loads(crew_output.raw)
+                output_pydantic = RAGOutput.model_validate(raw_json)
+            self.state.rag_output = output_pydantic
+            return crew_output
+        except Exception as e:
+            current_app.logger.error(f"CREW rag_crew Kickoff Error: {str(e)}")
+            self.exception_raised = True
+            raise e
+
+    @listen(process_inputs)
+    def execute_spin(self):
+        inputs = self.state.input.model_dump()
+        try:
+            crew_output = self.spin_crew.kickoff(inputs=inputs)
+            self.specialist_executor.log_tuning("Spin Crew Output", crew_output.model_dump())
+            output_pydantic = crew_output.pydantic
+            if not output_pydantic:
+                raw_json = json.loads(crew_output.raw)
+                output_pydantic = SPINOutput.model_validate(raw_json)
+            self.state.spin = output_pydantic
+            return crew_output
+        except Exception as e:
+            current_app.logger.error(f"CREW spin_crew Kickoff Error: {str(e)}")
+            self.exception_raised = True
+            raise e
+
+    @listen(process_inputs)
+    def execute_identification(self):
+        inputs = self.state.input.model_dump()
+        try:
+            crew_output = self.identification_crew.kickoff(inputs=inputs)
+            self.specialist_executor.log_tuning("Identification Crew Output", crew_output.model_dump())
+            output_pydantic = crew_output.pydantic
+            if not output_pydantic:
+                raw_json = json.loads(crew_output.raw)
+                output_pydantic = LeadInfoOutput.model_validate(raw_json)
+            self.state.lead_info = output_pydantic
+            return crew_output
+        except Exception as e:
+            current_app.logger.error(f"CREW identification_crew Kickoff Error: {str(e)}")
+            self.exception_raised = True
+            raise e
+
+    @listen(and_(execute_rag, execute_spin, execute_identification))
+    def consolidate(self):
+        inputs = self.state.input.model_dump()
+        if self.state.rag_output:
+            inputs["prepared_answers"] = self.state.rag_output.answer
+        additional_questions = ""
+        if self.state.lead_info:
+            additional_questions = self.state.lead_info.questions + "\n"
+        if self.state.spin:
+            additional_questions = additional_questions + self.state.spin.questions
+        inputs["additional_questions"] = additional_questions
+        try:
+            crew_output = self.rag_consolidation_crew.kickoff(inputs=inputs)
+            self.specialist_executor.log_tuning("RAG Consolidation Crew Output", crew_output.model_dump())
+            output_pydantic = crew_output.pydantic
+            if not output_pydantic:
+                raw_json = json.loads(crew_output.raw)
+                output_pydantic = LeadInfoOutput.model_validate(raw_json)
+            self.state.final_output = output_pydantic
+            return crew_output
+        except Exception as e:
+            current_app.logger.error(f"CREW rag_consolidation_crew Kickoff Error: {str(e)}")
+            self.exception_raised = True
+            raise e
+
+    def kickoff(self, inputs=None):
+        with current_event.create_span("SPIN Specialist Execution"):
+            self.specialist_executor.log_tuning("Inputs retrieved", inputs)
+            self.state.input = SPINSpecialistInput.model_validate(inputs)
+            self.specialist.update_progress("EveAI Flow Start", {"name": "SPIN"})
+            try:
+                result = super().kickoff()
+            except Exception as e:
+                current_app.logger.error(f"Error kicking of Flow: {str(e)}")
+
+            self.specialist.update_progress("EveAI Flow End", {"name": "SPIN"})
+
+            return self.state

eveai_chat_workers/specialists/STANDARD_RAG_SPECIALIST/1_0.py

@@ -9,24 +9,24 @@ from langchain_core.runnables import RunnableParallel, RunnablePassthrough
 from common.langchain.outputs.base import OutputRegistry
 from common.langchain.outputs.rag import RAGOutput
 from common.utils.business_event_context import current_event
-from .specialist_typing import SpecialistArguments, SpecialistResult
-from ..chat_session_cache import CachedInteraction, get_chat_history
-from ..retrievers.registry import RetrieverRegistry
-from ..retrievers.base import BaseRetriever
-from common.models.interaction import SpecialistRetriever, Specialist
+from eveai_chat_workers.specialists.specialist_typing import SpecialistArguments, SpecialistResult
+from eveai_chat_workers.chat_session_cache import get_chat_history
+from common.models.interaction import Specialist
 from common.utils.model_utils import get_model_variables, create_language_template, replace_variable_in_template
-from .base import BaseSpecialist
-from .registry import SpecialistRegistry
-from eveai_chat_workers.retrievers.retriever_typing import RetrieverArguments, RetrieverResult
+from eveai_chat_workers.specialists.base_specialist import BaseSpecialistExecutor
+from eveai_chat_workers.retrievers.retriever_typing import RetrieverArguments
 
 
-class RAGSpecialist(BaseSpecialist):
+class SpecialistExecutor(BaseSpecialistExecutor):
     """
+    type: STANDARD_RAG
+    type_version: 1.0
     Standard Q&A RAG Specialist implementation that combines retriever results
     with LLM processing to generate answers.
     """
-    def __init__(self, tenant_id: int, specialist_id: int, session_id: str):
-        super().__init__(tenant_id, specialist_id, session_id)
+
+    def __init__(self, tenant_id: int, specialist_id: int, session_id: str, task_id: str):
+        super().__init__(tenant_id, specialist_id, session_id, task_id)
 
         # Check and load the specialist
         specialist = Specialist.query.get_or_404(specialist_id)
@@ -43,66 +43,17 @@ class RAGSpecialist(BaseSpecialist):
 
     @property
     def type(self) -> str:
-        return "STANDARD_RAG"
+        return "STANDARD_RAG_SPECIALIST"
 
-    def _initialize_retrievers(self) -> List[BaseRetriever]:
-        """Initialize all retrievers associated with this specialist"""
-        retrievers = []
-
-        # Get retriever associations from database
-        specialist_retrievers = (
-            SpecialistRetriever.query
-            .filter_by(specialist_id=self.specialist_id)
-            .all()
-        )
-
-        self._log_tuning("_initialize_retrievers", {"Nr of retrievers": len(specialist_retrievers)})
-
-        for spec_retriever in specialist_retrievers:
-            # Get retriever configuration from database
-            retriever = spec_retriever.retriever
-            retriever_class = RetrieverRegistry.get_retriever_class(retriever.type)
-            self._log_tuning("_initialize_retrievers", {
-                "Retriever id": spec_retriever.retriever_id,
-                "Retriever Type": retriever.type,
-                "Retriever Class": str(retriever_class),
-            })
-
-            # Initialize retriever with its configuration
-            retrievers.append(
-                retriever_class(
-                    tenant_id=self.tenant_id,
-                    retriever_id=retriever.id,
-                )
-            )
-
-        return retrievers
+    @property
+    def type_version(self) -> str:
+        return "1.0"
 
     @property
     def required_templates(self) -> List[str]:
         """List of required templates for this specialist"""
         return ['rag', 'history']
 
-    # def _detail_question(question, language, model_variables, session_id):
-    #     retriever = EveAIHistoryRetriever(model_variables=model_variables, session_id=session_id)
-    #     llm = model_variables['llm']
-    #     template = model_variables['history_template']
-    #     language_template = create_language_template(template, language)
-    #     full_template = replace_variable_in_template(language_template, "{tenant_context}",
-    #                                                  model_variables['rag_context'])
-    #     history_prompt = ChatPromptTemplate.from_template(full_template)
-    #     setup_and_retrieval = RunnableParallel({"history": retriever, "question": RunnablePassthrough()})
-    #     output_parser = StrOutputParser()
-    #
-    #     chain = setup_and_retrieval | history_prompt | llm | output_parser
-    #
-    #     try:
-    #         answer = chain.invoke(question)
-    #         return answer
-    #     except LangChainException as e:
-    #         current_app.logger.error(f'Error detailing question: {e}')
-    #         raise
-
     def _detail_question(self, language: str, question: str) -> str:
         """Detail question based on conversation history"""
         try:
@@ -138,7 +89,7 @@ class RAGSpecialist(BaseSpecialist):
             })
 
             if self.tuning:
-                self._log_tuning("_detail_question", {
+                self.log_tuning("_detail_question", {
                     "cached_session_id": cached_session.session_id,
                     "cached_session.interactions": str(cached_session.interactions),
                     "original_question": question,
@@ -160,17 +111,20 @@ class RAGSpecialist(BaseSpecialist):
 
         try:
             with current_event.create_span("Specialist Detail Question"):
+                self.update_progress("Detail Question Start", {})
                 # Get required arguments
                 language = arguments.language
                 query = arguments.query
                 detailed_question = self._detail_question(language, query)
+                self.update_progress("Detail Question End", {})
 
             # Log the start of retrieval process if tuning is enabled
             with current_event.create_span("Specialist Retrieval"):
-                self._log_tuning("Starting context retrieval", {
+                self.log_tuning("Starting context retrieval", {
                     "num_retrievers": len(self.retrievers),
                     "all arguments": arguments.model_dump(),
                 })
+                self.update_progress("EveAI Retriever Start", {})
 
                 # Get retriever-specific arguments
                 retriever_arguments = arguments.retriever_arguments
@@ -208,12 +162,13 @@ class RAGSpecialist(BaseSpecialist):
                         unique_contexts.append(ctx)
                         seen_chunks.add(ctx.chunk)
 
-                self._log_tuning("Context retrieval completed", {
+                self.log_tuning("Context retrieval completed", {
                     "total_contexts": len(all_context),
                     "unique_contexts": len(unique_contexts),
                     "average_similarity": sum(ctx.similarity for ctx in unique_contexts) / len(
                         unique_contexts) if unique_contexts else 0
                 })
+                self.update_progress("EveAI Retriever Complete", {})
 
             # Prepare context for LLM
             formatted_context = "\n\n".join([
@@ -223,6 +178,7 @@ class RAGSpecialist(BaseSpecialist):
 
             with current_event.create_span("Specialist RAG invocation"):
                 try:
+                    self.update_progress(self.task_id, "EveAI Chain Start", {})
                     # Get LLM with specified temperature
                     llm = self.model_variables.get_llm(temperature=self.temperature)
 
@@ -236,7 +192,7 @@ class RAGSpecialist(BaseSpecialist):
                     )
 
                     if self.tuning:
-                        self._log_tuning("Template preparation completed", {
+                        self.log_tuning("Template preparation completed", {
                             "template": full_template,
                             "context": formatted_context,
                             "tenant_context": self.specialist_context,
@@ -258,7 +214,8 @@ class RAGSpecialist(BaseSpecialist):
 
                     raw_result = chain.invoke(detailed_question)
                     result = SpecialistResult.create_for_type(
-                        "STANDARD_RAG",
+                        self.type,
+                        self.type_version,
                         detailed_query=detailed_question,
                         answer=raw_result.answer,
                         citations=[ctx.metadata.document_id for ctx in unique_contexts
@@ -267,14 +224,15 @@ class RAGSpecialist(BaseSpecialist):
                     )
 
                     if self.tuning:
-                        self._log_tuning("LLM chain execution completed", {
+                        self.log_tuning("LLM chain execution completed", {
                             "Result": result.model_dump()
                         })
+                    self.update_progress("EveAI Chain Complete", {})
 
                 except Exception as e:
                     current_app.logger.error(f"Error in LLM processing: {e}")
                     if self.tuning:
-                        self._log_tuning("LLM processing error", {"error": str(e)})
+                        self.log_tuning("LLM processing error", {"error": str(e)})
                     raise
 
             return result
@@ -285,5 +243,4 @@ class RAGSpecialist(BaseSpecialist):
 
 
 # Register the specialist type
-SpecialistRegistry.register("STANDARD_RAG", RAGSpecialist)
-OutputRegistry.register("STANDARD_RAG", RAGOutput)
+OutputRegistry.register("STANDARD_RAG_SPECIALIST", RAGOutput)

eveai_chat_workers/specialists/__init__.py

@@ -1,5 +0,0 @@
-# Import all specialist implementations here to ensure registration
-from . import rag_specialist
-
-# List of all available specialist implementations
-__all__ = ['rag_specialist']

eveai_chat_workers/specialists/base.py

@@ -1,50 +0,0 @@
-from abc import ABC, abstractmethod
-from typing import Dict, Any
-from flask import current_app
-
-from config.logging_config import TuningLogger
-from eveai_chat_workers.specialists.specialist_typing import SpecialistArguments, SpecialistResult
-
-
-class BaseSpecialist(ABC):
-    """Base class for all specialists"""
-
-    def __init__(self, tenant_id: int, specialist_id: int, session_id: str):
-        self.tenant_id = tenant_id
-        self.specialist_id = specialist_id
-        self.session_id = session_id
-        self.tuning = False
-        self.tuning_logger = None
-        self._setup_tuning_logger()
-
-    @property
-    @abstractmethod
-    def type(self) -> str:
-        """The type of the specialist"""
-        pass
-
-    def _setup_tuning_logger(self):
-        try:
-            self.tuning_logger = TuningLogger(
-                'tuning',
-                tenant_id=self.tenant_id,
-                specialist_id=self.specialist_id,
-            )
-            # Verify logger is working with a test message
-            if self.tuning:
-                self.tuning_logger.log_tuning('specialist', "Tuning logger initialized")
-        except Exception as e:
-            current_app.logger.error(f"Failed to setup tuning logger: {str(e)}")
-            raise
-
-    def _log_tuning(self, message: str, data: Dict[str, Any] = None) -> None:
-        if self.tuning and self.tuning_logger:
-            try:
-                self.tuning_logger.log_tuning('specialist', message, data)
-            except Exception as e:
-                current_app.logger.error(f"Processor: Error in tuning logging: {e}")
-
-    @abstractmethod
-    def execute(self, arguments: SpecialistArguments) -> SpecialistResult:
-        """Execute the specialist's logic"""
-        pass

eveai_chat_workers/specialists/base_specialist.py

@@ -0,0 +1,106 @@
+import importlib
+from abc import ABC, abstractmethod
+from typing import Dict, Any, List
+from flask import current_app
+
+from common.models.interaction import SpecialistRetriever
+from common.utils.execution_progress import ExecutionProgressTracker
+from config.logging_config import TuningLogger
+from eveai_chat_workers.retrievers.base import BaseRetriever
+from eveai_chat_workers.retrievers.registry import RetrieverRegistry
+from eveai_chat_workers.specialists.specialist_typing import SpecialistArguments, SpecialistResult
+
+
+class BaseSpecialistExecutor(ABC):
+    """Base class for all specialists"""
+
+    def __init__(self, tenant_id: int, specialist_id: int, session_id: str, task_id: str):
+        self.tenant_id = tenant_id
+        self.specialist_id = specialist_id
+        self.session_id = session_id
+        self.task_id = task_id
+        self.tuning = False
+        self.tuning_logger = None
+        self._setup_tuning_logger()
+        self.ept = ExecutionProgressTracker()
+
+    @property
+    @abstractmethod
+    def type(self) -> str:
+        """The type of the specialist"""
+        pass
+
+    @property
+    @abstractmethod
+    def type_version(self) -> str:
+        """The type version of the specialist"""
+        pass
+
+    def _initialize_retrievers(self) -> List[BaseRetriever]:
+        """Initialize all retrievers associated with this specialist"""
+        retrievers = []
+
+        # Get retriever associations from database
+        specialist_retrievers = (
+            SpecialistRetriever.query
+            .filter_by(specialist_id=self.specialist_id)
+            .all()
+        )
+
+        self.log_tuning("_initialize_retrievers", {"Nr of retrievers": len(specialist_retrievers)})
+
+        for spec_retriever in specialist_retrievers:
+            # Get retriever configuration from database
+            retriever = spec_retriever.retriever
+            retriever_class = RetrieverRegistry.get_retriever_class(retriever.type)
+            self.log_tuning("_initialize_retrievers", {
+                "Retriever id": spec_retriever.retriever_id,
+                "Retriever Type": retriever.type,
+                "Retriever Class": str(retriever_class),
+            })
+
+            # Initialize retriever with its configuration
+            retrievers.append(
+                retriever_class(
+                    tenant_id=self.tenant_id,
+                    retriever_id=retriever.id,
+                )
+            )
+
+        return retrievers
+
+    def _setup_tuning_logger(self):
+        try:
+            self.tuning_logger = TuningLogger(
+                'tuning',
+                tenant_id=self.tenant_id,
+                specialist_id=self.specialist_id,
+            )
+            # Verify logger is working with a test message
+            if self.tuning:
+                self.tuning_logger.log_tuning('specialist', "Tuning logger initialized")
+        except Exception as e:
+            current_app.logger.error(f"Failed to setup tuning logger: {str(e)}")
+            raise
+
+    def log_tuning(self, message: str, data: Dict[str, Any] = None) -> None:
+        if self.tuning and self.tuning_logger:
+            try:
+                self.tuning_logger.log_tuning('specialist', message, data)
+            except Exception as e:
+                current_app.logger.error(f"Processor: Error in tuning logging: {e}")
+
+    def update_progress(self, processing_type, data) -> None:
+        self.ept.send_update(self.task_id, processing_type, data)
+
+    @abstractmethod
+    def execute(self, arguments: SpecialistArguments) -> SpecialistResult:
+        """Execute the specialist's logic"""
+        pass
+
+
+def get_specialist_class(specialist_type: str, type_version: str):
+    major_minor = '_'.join(type_version.split('.')[:2])
+    module_path = f"eveai_chat_workers.specialists.{specialist_type}.{major_minor}"
+    module = importlib.import_module(module_path)
+    return module.SpecialistExecutor

eveai_chat_workers/specialists/crewai_base_classes.py

@@ -0,0 +1,129 @@
+import json
+
+from crewai import Agent, Task, Crew, Flow
+from crewai.agents.parser import AgentAction, AgentFinish
+from crewai.tools import BaseTool
+from flask import current_app
+from pydantic import BaseModel, create_model, Field, ConfigDict
+from typing import Dict, Type, get_type_hints, Optional, List, Any, Callable
+
+
+class EveAICrewAIAgent(Agent):
+    specialist: Any = Field(default=None, exclude=True)
+    name: str = Field(default=None, exclude=True)
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    def __init__(self, specialist, name: str, **kwargs):
+        super().__init__(**kwargs)
+        self.specialist = specialist
+        self.name = name
+        self.specialist.log_tuning("Initializing EveAICrewAIAgent", {"name": name})
+        self.specialist.update_progress("EveAI Agent Initialisation", {"name": self.name})
+
+    def execute_task(
+            self,
+            task: Task,
+            context: Optional[str] = None,
+            tools: Optional[List[BaseTool]] = None,
+    ) -> str:
+        """Execute a task with the agent. Performs AskEveAI specific  fuctionality on top of task execution
+
+        Args:
+            task: Task to execute.
+            context: Context to execute the task in.
+            tools: Tools to use for the task.
+
+        Returns:
+            Output of the agent
+        """
+        self.specialist.log_tuning("EveAI Agent Task Start",
+                                   {"name": self.name,
+                                    'task': task.name,
+                                    })
+        self.specialist.update_progress("EveAI Agent Task Start",
+                                        {"name": self.name,
+                                         'task': task.name,
+                                         })
+
+        result = super().execute_task(task, context, tools)
+
+        self.specialist.log_tuning("EveAI Agent Task Complete",
+                                   {"name": self.name,
+                                    'task': task.name,
+                                    'result': result,
+                                    })
+        self.specialist.update_progress("EveAI Agent Task Complete",
+                                        {"name": self.name,
+                                         'task': task.name,
+                                         })
+
+        return result
+
+
+class EveAICrewAITask(Task):
+    specialist: Any = Field(default=None, exclude=True)
+    name: str = Field(default=None, exclude=True)
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    def __init__(self, specialist, name: str, **kwargs):
+        # kwargs.update({"callback": create_task_callback(self)})
+        super().__init__(**kwargs)
+        # current_app.logger.debug(f"Task pydantic class for {name}: {"class", self.output_pydantic}")
+        self.specialist = specialist
+        self.name = name
+        self.specialist.log_tuning("Initializing EveAICrewAITask", {"name": name})
+        self.specialist.update_progress("EveAI Task Initialisation", {"name": name})
+
+
+# def create_task_callback(task: EveAICrewAITask):
+#     def task_callback(output):
+#         # Todo Check if required with new version of crewai
+#         if isinstance(output, BaseModel):
+#             task.specialist.log_tuning(f"TASK CALLBACK: EveAICrewAITask {task.name} Output:",
+#                                        {'output': output.model_dump()})
+#             if output.output_format == "pydantic" and not output.pydantic:
+#                 try:
+#                     raw_json = json.loads(output.raw)
+#                     output_pydantic = task.output_pydantic(**raw_json)
+#                     output.pydantic = output_pydantic
+#                     task.specialist.log_tuning(f"TASK CALLBACK: EveAICrewAITask {task.name} Converted Output",
+#                                                {'output': output_pydantic.model_dump()})
+#                 except Exception as e:
+#                     task.specialist.log_tuning(f"TASK CALLBACK: EveAICrewAITask {task.name} Output Conversion Error: "
+#                                                f"{str(e)}", {})
+#
+#     return task_callback
+
+
+class EveAICrewAICrew(Crew):
+    specialist: Any = Field(default=None, exclude=True)
+    name: str = Field(default=None, exclude=True)
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    def __init__(self, specialist, name: str, **kwargs):
+        super().__init__(**kwargs)
+        self.specialist = specialist
+        self.name = name
+        self.specialist.log_tuning("Initializing EveAICrewAICrew", {"name": self.name})
+        self.specialist.update_progress("EveAI Crew Initialisation", {"name": self.name})
+
+
+class EveAICrewAIFlow(Flow):
+    specialist: Any = Field(default=None, exclude=True)
+    name: str = Field(default=None, exclude=True)
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    def __init__(self, specialist, name: str, **kwargs):
+        super().__init__(**kwargs)
+        self.specialist = specialist
+        self.name = name
+        self.specialist.log_tuning("Initializing EveAICrewAIFlow", {"name": self.name})
+        self.specialist.update_progress("EveAI Flow Initialisation", {"name": self.name})
+
+
+class EveAIFlowState(BaseModel):
+    """Base class for all EveAI flow states"""
+    pass
+
+
+

eveai_chat_workers/specialists/crewai_base_specialist.py

@@ -0,0 +1,243 @@
+import json
+from typing import Dict, Any, Optional, Type, TypeVar, List, Tuple
+
+from crewai.flow.flow import FlowState
+from flask import current_app
+
+from common.models.interaction import Specialist
+from common.utils.business_event_context import current_event
+from common.utils.model_utils import get_model_variables
+from eveai_chat_workers.retrievers.retriever_typing import RetrieverArguments
+from eveai_chat_workers.specialists.crewai_base_classes import EveAICrewAIAgent, EveAICrewAITask
+from crewai.tools import BaseTool
+from abc import ABC, abstractmethod
+
+from pydantic import BaseModel
+
+from common.extensions import cache_manager
+from eveai_chat_workers.specialists.base_specialist import BaseSpecialistExecutor
+from common.utils.cache.crewai_configuration import (
+    ProcessedAgentConfig, ProcessedTaskConfig, ProcessedToolConfig,
+    SpecialistProcessedConfig
+)
+from eveai_chat_workers.specialists.specialist_typing import SpecialistArguments
+
+T = TypeVar('T')  # For generic type hints
+
+
+class CrewAIBaseSpecialistExecutor(BaseSpecialistExecutor):
+    """Base class for all CrewAI-based specialists"""
+
+    def __init__(self, tenant_id: int, specialist_id: int, session_id: str, task_id):
+        super().__init__(tenant_id, specialist_id, session_id, task_id)
+
+        # Check and load the specialist
+        self.specialist = Specialist.query.get_or_404(specialist_id)
+        # Set the specific configuration for the SPIN Specialist
+        # self.specialist_configuration = json.loads(self.specialist.configuration)
+        self.tuning = self.specialist.tuning
+        # Initialize retrievers
+        self.retrievers = self._initialize_retrievers()
+
+        # Initialize model variables
+        self.model_variables = get_model_variables(tenant_id)
+
+        # initialize the Flow
+        self.flow = None
+
+        # Runtime instances
+        self._agents: Dict[str, EveAICrewAIAgent] = {}
+        self._tasks: Dict[str, EveAICrewAITask] = {}
+        self._tools: Dict[str, BaseTool] = {}
+
+        # Crew configuration
+        self._task_agents: Dict[str, str] = {}
+        self._task_pydantic_outputs: Dict[str, Type[BaseModel]] = {}
+        self._task_state_names: Dict[str, str] = {}
+
+        # Processed configurations
+        self._config = cache_manager.crewai_processed_config_cache.get_specialist_config(tenant_id, specialist_id)
+        self._config_task_agents()
+        self._config_pydantic_outputs()
+        self._instantiate_crew_assets()
+        self._instantiate_specialist()
+
+        # Retrieve history
+        self._cached_session = cache_manager.chat_session_cache.get_cached_session(self.session_id)
+        # Format history for the prompt
+        self._formatted_history = "\n\n".join([
+            f"HUMAN:\n{interaction.specialist_results.get('detailed_query')}\n\n"
+            f"AI:\n{interaction.specialist_results.get('answer')}"
+            for interaction in self._cached_session.interactions
+        ])
+
+    def _add_task_agent(self, task_name: str, agent_name: str):
+        self._task_agents[task_name.lower()] = agent_name
+
+    @abstractmethod
+    def _config_task_agents(self):
+        """Configure the task agents by adding task-agent combinations. Use _add_task_agent()
+        """
+
+    @property
+    def task_agents(self) -> Dict[str, str]:
+        return self._task_agents
+
+    def _add_pydantic_output(self, task_name: str, output: Type[BaseModel], state_name: str is None):
+        self._task_pydantic_outputs[task_name.lower()] = output
+        if state_name is not None:
+            self._task_state_names[task_name.lower()] = state_name
+
+    @abstractmethod
+    def _config_pydantic_outputs(self):
+        """Configure the task pydantic outputs by adding task-output combinations. Use _add_pydantic_output()"""
+
+    @property
+    def task_pydantic_outputs(self):
+        return self._task_pydantic_outputs
+
+    @property
+    def task_state_names(self):
+        return self._task_state_names
+
+    def _instantiate_crew_assets(self):
+        self._instantiate_crew_agents()
+        self._instantiate_tasks()
+        self._instantiate_tools()
+
+    def _instantiate_crew_agents(self):
+        for agent in self.specialist.agents:
+            agent_config = cache_manager.agents_config_cache.get_config(agent.type, agent.type_version)
+            agent_role = agent_config.get('role', '').replace('{custom_role}', agent.role or '')
+            agent_goal = agent_config.get('goal', '').replace('{custom_goal}', agent.goal or '')
+            agent_backstory = agent_config.get('backstory', '').replace('{custom_backstory}', agent.backstory or '')
+            new_agent = EveAICrewAIAgent(
+                self,
+                agent.type.lower(),
+                role=agent_role,
+                goal=agent_goal,
+                backstory=agent_backstory,
+                verbose=agent.tuning,
+            )
+            agent_name = agent.type.lower()
+            self.log_tuning(f"CrewAI Agent {agent_name} initialized", agent_config)
+            self._agents[agent_name] = new_agent
+
+    def _instantiate_tasks(self):
+        for task in self.specialist.tasks:
+            task_config = cache_manager.tasks_config_cache.get_config(task.type, task.type_version)
+            task_description = (task_config.get('task_description', '')
+                                .replace('{custom_description}', task.task_description or ''))
+            task_expected_output = (task_config.get('expected_output', '')
+                                    .replace('{custom_expected_output}', task.expected_output or ''))
+            # dynamically build the arguments
+            task_kwargs = {
+                "description": task_description,
+                "expected_output": task_expected_output,
+                "verbose": task.tuning
+            }
+            task_name = task.type.lower()
+            if task_name in self._task_pydantic_outputs:
+                task_kwargs["output_pydantic"] = self._task_pydantic_outputs[task_name]
+            if task_name in self._task_agents:
+                task_kwargs["agent"] = self._agents[self._task_agents[task_name]]
+
+            # Instantiate the task with dynamic arguments
+            new_task = EveAICrewAITask(self, task_name, **task_kwargs)
+
+            # Logging and storing the task
+            self.log_tuning(f"CrewAI Task {task_name} initialized", task_config)
+            self._tasks[task_name] = new_task
+
+    def _instantiate_tools(self):
+        # This currently is not implemented
+        # TODO: complete Tool instantiation
+        pass
+
+    def __getattr__(self, name: str) -> Any:
+        """Enable dynamic access to agents as attributes"""
+        try:
+            if name.endswith('_agent'):
+                return self._agents[name]
+
+            if name.endswith('_task'):
+                return self._tasks[name]
+
+            if name.endswith('_tool'):
+                return self._tools[name]
+
+            # Not a known component request
+            raise AttributeError(f"'{self.__class__.__name__}' has no attribute '{name}'")
+        except KeyError:
+            raise AttributeError(f"'{self.__class__.__name__}' has no attribute '{name}'")
+
+    @abstractmethod
+    def _instantiate_specialist(self):
+        """Instantiate a crew (or flow) to set up the complete specialist, using the assets (agents, tasks, tools).
+         The assets can be retrieved using their type name in lower case, e.g. rag_agent"""
+
+    def retrieve_context(self, arguments: SpecialistArguments) -> Tuple[str, List[int]]:
+        with current_event.create_span("Specialist Retrieval"):
+            self.log_tuning("Starting context retrieval", {
+                "num_retrievers": len(self.retrievers),
+                "all arguments": arguments.model_dump(),
+            })
+
+            # Get retriever-specific arguments
+            retriever_arguments = arguments.retriever_arguments
+
+            # Collect context from all retrievers
+            all_context = []
+            for retriever in self.retrievers:
+                # Get arguments for this specific retriever
+                retriever_id = str(retriever.retriever_id)
+                if retriever_id not in retriever_arguments:
+                    current_app.logger.error(f"Missing arguments for retriever {retriever_id}")
+                    continue
+
+                # Get the retriever's arguments and update the query
+                current_retriever_args = retriever_arguments[retriever_id]
+                if isinstance(retriever_arguments[retriever_id], RetrieverArguments):
+                    updated_args = current_retriever_args.model_dump()
+                    updated_args['query'] = arguments.query
+                    updated_args['language'] = arguments.language
+                    retriever_args = RetrieverArguments(**updated_args)
+                else:
+                    # Create a new RetrieverArguments instance from the dictionary
+                    current_retriever_args['query'] = arguments.query
+                    retriever_args = RetrieverArguments(**current_retriever_args)
+
+                # Each retriever gets its own specific arguments
+                retriever_result = retriever.retrieve(retriever_args)
+                all_context.extend(retriever_result)
+
+            # Sort by similarity if available and get unique contexts
+            all_context.sort(key=lambda x: x.similarity, reverse=True)
+            unique_contexts = []
+            seen_chunks = set()
+            for ctx in all_context:
+                if ctx.chunk not in seen_chunks:
+                    unique_contexts.append(ctx)
+                    seen_chunks.add(ctx.chunk)
+
+            self.log_tuning("Context retrieval completed", {
+                "total_contexts": len(all_context),
+                "unique_contexts": len(unique_contexts),
+                "average_similarity": sum(ctx.similarity for ctx in unique_contexts) / len(
+                    unique_contexts) if unique_contexts else 0
+            })
+
+        # Prepare context for LLM
+        formatted_context = "\n\n".join([
+            f"SOURCE: {ctx.metadata.document_id}\n{ctx.chunk}\n\n"
+            for ctx in unique_contexts
+        ])
+
+        # Return document_ids for citations
+        citations = [ctx.metadata.document_id for ctx in unique_contexts]
+
+        self.log_tuning("Context Retrieval Results",
+                        {"Formatted Context": formatted_context,
+                         "Citations": citations})
+
+        return formatted_context, citations

eveai_chat_workers/specialists/registry.py

@@ -1,21 +0,0 @@
-from typing import Dict, Type
-from .base import BaseSpecialist
-
-
-class SpecialistRegistry:
-    """Registry for specialist types"""
-
-    _registry: Dict[str, Type[BaseSpecialist]] = {}
-
-    @classmethod
-    def register(cls, specialist_type: str, specialist_class: Type[BaseSpecialist]):
-        """Register a new specialist type"""
-        cls._registry[specialist_type] = specialist_class
-
-    @classmethod
-    def get_specialist_class(cls, specialist_type: str) -> Type[BaseSpecialist]:
-        """Get the specialist class for a given type"""
-        if specialist_type not in cls._registry:
-            raise ValueError(f"Unknown specialist type: {specialist_type}")
-        return cls._registry[specialist_type]
-

eveai_chat_workers/specialists/specialist_typing.py

@@ -1,7 +1,7 @@
 from typing import Dict, Any
 from pydantic import BaseModel, Field, model_validator
-from config.type_defs.specialist_types import SPECIALIST_TYPES
 from eveai_chat_workers.retrievers.retriever_typing import RetrieverArguments
+from common.extensions import cache_manager
 
 
 class SpecialistArguments(BaseModel):
@@ -10,6 +10,7 @@ class SpecialistArguments(BaseModel):
     based on SPECIALIST_TYPES configuration.
     """
     type: str = Field(..., description="Type of specialist (e.g. STANDARD_RAG)")
+    type_version: str = Field(..., description="Type version (e.g. 1.0)")
     retriever_arguments: Dict[str, Any] = Field(
         default_factory=dict,
         description="Arguments for each retriever, keyed by retriever ID"
@@ -23,7 +24,7 @@ class SpecialistArguments(BaseModel):
     @model_validator(mode='after')
     def validate_required_arguments(self) -> 'SpecialistArguments':
         """Validate that all required arguments for this specialist type are present"""
-        specialist_config = SPECIALIST_TYPES.get(self.type)
+        specialist_config = cache_manager.specialists_config_cache.get_config(self.type, self.type_version)
         if not specialist_config:
             raise ValueError(f"Unknown specialist type: {self.type}")
 
@@ -44,7 +45,7 @@ class SpecialistArguments(BaseModel):
         return self
 
     @classmethod
-    def create(cls, type_name: str, specialist_args: Dict[str, Any],
+    def create(cls, type_name: str, type_version: str, specialist_args: Dict[str, Any],
                retriever_args: Dict[str, Dict[str, Any]]) -> 'SpecialistArguments':
         """
         Factory method to create SpecialistArguments with validated retriever arguments
@@ -63,12 +64,15 @@ class SpecialistArguments(BaseModel):
             # Ensure type is included in retriever arguments
             if 'type' not in args:
                 raise ValueError(f"Retriever arguments for {retriever_id} must include 'type'")
+            if 'type_version' not in args:
+                raise ValueError(f"Retriever arguments for {retriever_id} must include 'type_version'")
 
             validated_retriever_args[retriever_id] = RetrieverArguments(**args)
 
         # Combine everything into the specialist arguments
         return cls(
             type=type_name,
+            type_version=type_version,
             **specialist_args,
             retriever_arguments=validated_retriever_args
         )
@@ -80,6 +84,7 @@ class SpecialistResult(BaseModel):
     SPECIALIST_TYPES configuration.
     """
     type: str = Field(..., description="Type of specialist (e.g. STANDARD_RAG)")
+    type_version: str = Field(..., description="Type version (e.g. 1.0)")
 
     # Allow any additional fields
     model_config = {
@@ -89,9 +94,9 @@ class SpecialistResult(BaseModel):
     @model_validator(mode='after')
     def validate_required_results(self) -> 'SpecialistResult':
         """Validate that all required result fields for this specialist type are present"""
-        specialist_config = SPECIALIST_TYPES.get(self.type)
+        specialist_config = cache_manager.specialists_config_cache.get_config(self.type, self.type_version)
         if not specialist_config:
-            raise ValueError(f"Unknown specialist type: {self.type}")
+            raise ValueError(f"Unknown specialist type: {self.type}, {self.type_version}")
 
         # Check required results from configuration
         required_results = specialist_config.get('results', {})
@@ -117,12 +122,13 @@ class SpecialistResult(BaseModel):
         return self
 
     @classmethod
-    def create_for_type(cls, specialist_type: str, **results) -> 'SpecialistResult':
+    def create_for_type(cls, specialist_type: str, specialist_type_version: str, **results) -> 'SpecialistResult':
         """
         Factory method to create a type-specific result
 
         Args:
             specialist_type: The type of specialist (e.g., 'STANDARD_RAG')
+            specialist_type_version: The type of specialist (e.g., '1.0')
             **results: The result values to include
 
         Returns:
@@ -132,6 +138,7 @@ class SpecialistResult(BaseModel):
             For STANDARD_RAG:
             result = SpecialistResult.create_for_type(
                 'STANDARD_RAG',
+                '1.0',
                 answer="The answer text",
                 citations=["doc1", "doc2"],
                 insufficient_info=False
@@ -139,6 +146,7 @@ class SpecialistResult(BaseModel):
         """
         # Add the type to the results
         results['type'] = specialist_type
+        results['type_version'] = specialist_type_version
 
         # Create and validate the result
         return cls(**results)

eveai_chat_workers/tasks.py

@@ -13,10 +13,9 @@ from common.extensions import db, cache_manager
 from common.utils.celery_utils import current_celery
 from common.utils.business_event import BusinessEvent
 from common.utils.business_event_context import current_event
-from config.type_defs.specialist_types import SPECIALIST_TYPES
-from eveai_chat_workers.specialists.registry import SpecialistRegistry
-from config.type_defs.retriever_types import RETRIEVER_TYPES
 from eveai_chat_workers.specialists.specialist_typing import SpecialistArguments
+from eveai_chat_workers.specialists.base_specialist import get_specialist_class
+from common.utils.execution_progress import ExecutionProgressTracker
 
 
 # Healthcheck task
@@ -30,18 +29,19 @@ class ArgumentPreparationError(Exception):
     pass
 
 
-def validate_specialist_arguments(specialist_type: str, arguments: Dict[str, Any]) -> None:
+def validate_specialist_arguments(specialist_type: str, specialist_type_version:str, arguments: Dict[str, Any]) -> None:
     """
     Validate specialist-specific arguments
 
     Args:
         specialist_type: Type of specialist
+        specialist_type_version: Version of specialist type
         arguments: Arguments to validate (excluding retriever-specific arguments)
 
     Raises:
         ArgumentPreparationError: If validation fails
     """
-    specialist_config = SPECIALIST_TYPES.get(specialist_type)
+    specialist_config = cache_manager.specialists_config_cache.get_config(specialist_type, specialist_type_version)
     if not specialist_config:
         raise ArgumentPreparationError(f"Unknown specialist type: {specialist_type}")
 
@@ -61,20 +61,21 @@ def validate_specialist_arguments(specialist_type: str, arguments: Dict[str, Any
                 raise ArgumentPreparationError(f"Argument '{arg_name}' must be an integer")
 
 
-def validate_retriever_arguments(retriever_type: str, arguments: Dict[str, Any],
+def validate_retriever_arguments(retriever_type: str, retriever_type_version: str, arguments: Dict[str, Any],
                                  catalog_config: Optional[Dict[str, Any]] = None) -> None:
     """
     Validate retriever-specific arguments
 
     Args:
         retriever_type: Type of retriever
+        retriever_type_version: Version of retriever type
         arguments: Arguments to validate
         catalog_config: Optional catalog configuration for metadata validation
 
     Raises:
         ArgumentPreparationError: If validation fails
     """
-    retriever_config = RETRIEVER_TYPES.get(retriever_type)
+    retriever_config = cache_manager.retrievers_config_cache.get_config(retriever_type, retriever_type_version)
     if not retriever_config:
         raise ArgumentPreparationError(f"Unknown retriever type: {retriever_type}")
 
@@ -141,7 +142,7 @@ def prepare_arguments(specialist: Any, arguments: Dict[str, Any]) -> Dict[str, A
                 specialist_args[key] = value
 
         # Validate specialist arguments
-        validate_specialist_arguments(specialist.type, specialist_args)
+        validate_specialist_arguments(specialist.type, specialist.type_version, specialist_args)
 
         # Get all retrievers associated with this specialist
         specialist_retrievers = (
@@ -177,10 +178,11 @@ def prepare_arguments(specialist: Any, arguments: Dict[str, Any]) -> Dict[str, A
 
             # Always include the retriever type
             inherited_args['type'] = retriever.type
+            inherited_args['type_version'] = retriever.type_version
 
             # Validate the combined arguments
             validate_retriever_arguments(
-                retriever.type,
+                retriever.type, retriever.type_version,
                 inherited_args,
                 catalog_config
             )
@@ -202,9 +204,9 @@ def prepare_arguments(specialist: Any, arguments: Dict[str, Any]) -> Dict[str, A
         raise ArgumentPreparationError(str(e))
 
 
-@current_celery.task(name='execute_specialist', queue='llm_interactions')
-def execute_specialist(tenant_id: int, specialist_id: int, arguments: Dict[str, Any],
-                       session_id: str, user_timezone: str, room: str) -> dict:
+@current_celery.task(name='execute_specialist', queue='llm_interactions', bind=True)
+def execute_specialist(self, tenant_id: int, specialist_id: int, arguments: Dict[str, Any],
+                       session_id: str, user_timezone: str) -> dict:
     """
     Execute a specialist with given arguments
 
@@ -214,15 +216,16 @@ def execute_specialist(tenant_id: int, specialist_id: int, arguments: Dict[str,
         arguments: Dictionary containing all required arguments for specialist and retrievers
         session_id: Chat session ID
         user_timezone: User's timezone
-        room: Socket.IO room for the response
 
     Returns:
         dict: {
             'result': Dict - Specialist execution result
             'interaction_id': int - Created interaction ID
-            'room': str - Socket.IO room
         }
     """
+    task_id = self.request.id
+    ept = ExecutionProgressTracker()
+    ept.send_update(task_id, "EveAI Specialist Started", {})
     with BusinessEvent("Execute Specialist", tenant_id=tenant_id, chat_session_id=session_id) as event:
         current_app.logger.info(
             f'execute_specialist: Processing request for tenant {tenant_id} using specialist {specialist_id}')
@@ -241,6 +244,10 @@ def execute_specialist(tenant_id: int, specialist_id: int, arguments: Dict[str,
                 session_id,
                 create_params={'timezone': user_timezone}
             )
+            if cached_session:
+                current_app.logger.debug(f"Cached Session successfully retrieved for {session_id}: {cached_session.id}")
+            else:
+                current_app.logger.debug(f"No Cached Session retrieved for {session_id}")
 
             # Get specialist from database
             specialist = Specialist.query.get_or_404(specialist_id)
@@ -251,6 +258,7 @@ def execute_specialist(tenant_id: int, specialist_id: int, arguments: Dict[str,
                 # Convert the prepared arguments into a SpecialistArguments instance
                 complete_arguments = SpecialistArguments.create(
                     type_name=specialist.type,
+                    type_version=specialist.type_version,
                     specialist_args={k: v for k, v in raw_arguments.items() if k != 'retriever_arguments'},
                     retriever_args=raw_arguments.get('retriever_arguments', {})
                 )
@@ -276,12 +284,14 @@ def execute_specialist(tenant_id: int, specialist_id: int, arguments: Dict[str,
                 raise
 
             with current_event.create_span("Specialist invocation"):
+                ept.send_update(task_id, "EveAI Specialist Start", {})
                 # Initialize specialist instance
-                specialist_class = SpecialistRegistry.get_specialist_class(specialist.type)
+                specialist_class = get_specialist_class(specialist.type, specialist.type_version)
                 specialist_instance = specialist_class(
                     tenant_id=tenant_id,
                     specialist_id=specialist_id,
                     session_id=session_id,
+                    task_id=task_id,
                 )
 
                 # Execute specialist
@@ -304,13 +314,14 @@ def execute_specialist(tenant_id: int, specialist_id: int, arguments: Dict[str,
             # Prepare response
             response = {
                 'result': result.model_dump(),
-                'interaction_id': new_interaction.id,
-                'room': room
+                'interaction_id': new_interaction.id
             }
+            ept.send_update(task_id, "EveAI Specialist Complete", response)
 
             return response
 
         except Exception as e:
+            ept.send_update(task_id, "EveAI Specialist Error", {'Error': str(e)})
             current_app.logger.error(f'execute_specialist: Error executing specialist: {e}')
             raise
 

migrations/tenant/versions/209ae2db55f0_specialist_standard_rag_renamed.py

@@ -0,0 +1,41 @@
+"""Specialist STANDARD_RAG renamed
+
+Revision ID: 209ae2db55f0
+Revises: b9cc547a0512
+Create Date: 2025-02-08 14:58:29.960295
+
+"""
+from alembic import op
+import sqlalchemy as sa
+import pgvector
+from sqlalchemy import table, column, String
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = '209ae2db55f0'
+down_revision = 'b9cc547a0512'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    # Define specialist table structure needed for the update
+    specialist = table('specialist',
+                       column('id', sa.Integer),
+                       column('type', String)
+                       )
+
+    # Update all specialists with type STANDARD_RAG to STANDARD_RAG_SPECIALIST
+    op.execute(
+        specialist.update().
+        where(specialist.c.type == 'STANDARD_RAG').
+        values(type='STANDARD_RAG_SPECIALIST')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    pass
+    # ### end Alembic commands ###

migrations/tenant/versions/6857672e8164_add_type_version_default_for_specialist.py

@@ -0,0 +1,44 @@
+"""Add type_version default for specialist
+
+Revision ID: 6857672e8164
+Revises: 209ae2db55f0
+Create Date: 2025-02-10 04:45:46.336174
+
+"""
+from alembic import op
+import sqlalchemy as sa
+import pgvector
+from sqlalchemy import table, column, String, text
+from sqlalchemy.dialects import postgresql
+from sqlalchemy.exc import SQLAlchemyError
+
+# revision identifiers, used by Alembic.
+revision = '6857672e8164'
+down_revision = '209ae2db55f0'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    connection = op.get_bind()
+    table_name = 'specialist'
+    try:
+        result = connection.execute(
+            text(f"""
+                        UPDATE {table_name}
+                        SET type_version = '1.0.0'
+                        WHERE type_version IS NULL OR type_version = ''
+                    """)
+        )
+        print(f"Updated {result.rowcount} rows for type_version in {table_name}")
+    except SQLAlchemyError as e:
+        print(f"Error updating type_version in {table_name}: {str(e)}")
+        raise
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    pass
+    # ### end Alembic commands ###

migrations/tenant/versions/b9cc547a0512_add_version_type_to_retriever_model.py

@@ -0,0 +1,29 @@
+"""Add version_type to Retriever model
+
+Revision ID: b9cc547a0512
+Revises: efcd6a0d2989
+Create Date: 2025-01-24 06:56:33.459264
+
+"""
+from alembic import op
+import sqlalchemy as sa
+import pgvector
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = 'b9cc547a0512'
+down_revision = 'efcd6a0d2989'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('retriever', sa.Column('type_version', sa.String(length=20), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('retriever', 'type_version')
+    # ### end Alembic commands ###

migrations/tenant/versions/e58835fadd96_update_retriever_type_version_to_1_0_0.py

@@ -0,0 +1,44 @@
+"""Update Retriever type_version to 1.0.0
+
+Revision ID: e58835fadd96
+Revises: 6857672e8164
+Create Date: 2025-02-10 12:20:22.748172
+
+"""
+from alembic import op
+import sqlalchemy as sa
+import pgvector
+from sqlalchemy import text
+from sqlalchemy.dialects import postgresql
+from sqlalchemy.exc import SQLAlchemyError
+
+# revision identifiers, used by Alembic.
+revision = 'e58835fadd96'
+down_revision = '6857672e8164'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    connection = op.get_bind()
+    table_name = 'retriever'
+    try:
+        result = connection.execute(
+            text(f"""
+                            UPDATE {table_name}
+                            SET type_version = '1.0.0'
+                            WHERE type_version IS NULL OR type_version = ''
+                        """)
+        )
+        print(f"Updated {result.rowcount} rows for type_version in {table_name}")
+    except SQLAlchemyError as e:
+        print(f"Error updating type_version in {table_name}: {str(e)}")
+        raise
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    pass
+    # ### end Alembic commands ###

requirements.txt

@@ -3,7 +3,7 @@ annotated-types~=0.7.0
 bcrypt~=4.1.3
 beautifulsoup4~=4.12.3
 celery~=5.4.0
-certifi~=2024.6.2
+certifi~=2024.7.4
 chardet~=5.2.0
 cors~=1.0.1
 Flask~=3.0.3
@@ -15,7 +15,7 @@ Flask-Login~=0.6.3
 flask-mailman~=1.1.1
 Flask-Migrate~=4.0.7
 Flask-Principal~=0.4.0
-Flask-Security-Too~=5.4.3
+Flask-Security-Too~=5.5.2
 Flask-Session~=0.8.0
 Flask-SocketIO~=5.3.6
 Flask-SQLAlchemy~=3.1.1
@@ -31,13 +31,13 @@ langchain-anthropic~=0.2.0
 langchain-community~=0.3.0
 langchain-core~=0.3.0
 langchain-mistralai~=0.2.0
-langchain-openai~=0.2.10
+langchain-openai~=0.3.5
 langchain-postgres~=0.0.12
 langchain-text-splitters~=0.3.0
 langcodes~=3.4.0
 langdetect~=1.0.9
 langsmith~=0.1.81
-openai~=1.55.3
+openai~=1.62.0
 pg8000~=1.31.2
 pgvector~=0.2.5
 pycryptodome~=3.20.0
@@ -83,10 +83,12 @@ psutil~=6.0.0
 celery-redbeat~=2.2.0
 WTForms-SQLAlchemy~=0.4.1
 packaging~=24.1
-
 typing_extensions~=4.12.2
 prometheus_flask_exporter~=0.23.1
 prometheus_client~=0.20.0
 babel~=2.16.0
 dogpile.cache~=1.3.3
 python-docx~=1.1.2
+crewai~=0.102.0
+sseclient~=0.0.27
+termcolor~=2.5.0

scripts/start_eveai_api.sh

@@ -12,4 +12,4 @@ export FLASK_APP=${PROJECT_DIR}/scripts/run_eveai_app.py  # Adjust the path to y
 chown -R appuser:appuser /app/logs
 
 # Start Flask app
-gunicorn -w 1 -k gevent -b 0.0.0.0:5001 --worker-connections 100 scripts.run_eveai_api:app
+gunicorn -w 1 -k gevent -b 0.0.0.0:5003 --worker-connections 100 scripts.run_eveai_api:app

scripts/start_eveai_chat.sh

@@ -13,4 +13,4 @@ chown -R appuser:appuser /app/logs
 echo "Starting EveAI Chat"
 
 # Start Flask app
-gunicorn -w 1 -k gevent -b 0.0.0.0:5001 --worker-connections 100 scripts.run_eveai_chat:app
+gunicorn -w 1 -k gevent -b 0.0.0.0:5002 --worker-connections 100 scripts.run_eveai_chat:app

tests/interactive_client/specialist_client.py

@@ -0,0 +1,247 @@
+#!/usr/bin/env python3
+import json
+import logging
+import sys
+import time
+import requests  # Used for calling the auth API
+from datetime import datetime
+import yaml      # For loading the YAML configuration
+from urllib.parse import urlparse
+
+import socketio  # Official python-socketio client
+
+# ----------------------------
+# Constants for authentication and specialist selection
+# ----------------------------
+API_KEY = "EveAI-8342-2966-4731-6578-1010-8903-4230-4378"
+TENANT_ID = 2
+SPECIALIST_ID = 2
+BASE_API_URL = "http://macstudio.ask-eve-ai-local.com:8080/api/api/v1"
+BASE_SOCKET_URL = "http://macstudio.ask-eve-ai-local.com:8080"
+CONFIG_FILE = "config/specialists/SPIN_SPECIALIST/1.0.0.yaml"  # Path to specialist configuration
+
+# ----------------------------
+# Logging Configuration
+# ----------------------------
+LOG_FILENAME = "specialist_client.log"
+logging.basicConfig(
+    filename=LOG_FILENAME,
+    level=logging.DEBUG,
+    format="%(asctime)s %(levelname)s: %(message)s"
+)
+console_handler = logging.StreamHandler(sys.stdout)
+console_handler.setLevel(logging.INFO)
+logging.getLogger('').addHandler(console_handler)
+
+# ----------------------------
+# Create the Socket.IO client using the official python-socketio client
+# ----------------------------
+sio = socketio.Client(logger=True, engineio_logger=True)
+room = None  # Global variable to store the assigned room
+
+# ----------------------------
+# Event Handlers
+# ----------------------------
+@sio.event
+def connect():
+    logging.info("Connected to Socket.IO server.")
+    print("Connected to server.")
+
+@sio.event
+def disconnect():
+    logging.info("Disconnected from Socket.IO server.")
+    print("Disconnected from server.")
+
+@sio.on("connect_error")
+def on_connect_error(data):
+    logging.error("Connect error: %s", data)
+    print("Connect error:", data)
+
+@sio.on("authenticated")
+def on_authenticated(data):
+    global room
+    room = data.get("room")
+    logging.info("Authenticated. Room: %s", room)
+    print("Authenticated. Room:", room)
+
+@sio.on("room_join")
+def on_room_join(data):
+    global room
+    room = data.get("room")
+    logging.info("Room join event received. Room: %s", room)
+    print("Joined room:", room)
+
+@sio.on("token_expired")
+def on_token_expired(data):
+    logging.warning("Token expired.")
+    print("Token expired. Please refresh your session.")
+
+@sio.on("reconnect_attempt")
+def on_reconnect_attempt(attempt):
+    logging.info("Reconnect attempt #%s", attempt)
+    print(f"Reconnect attempt #{attempt}")
+
+@sio.on("reconnect")
+def on_reconnect():
+    logging.info("Reconnected successfully.")
+    print("Reconnected to server.")
+
+@sio.on("reconnect_failed")
+def on_reconnect_failed():
+    logging.error("Reconnection failed.")
+    print("Reconnection failed. Please refresh.")
+
+@sio.on("room_rejoin_result")
+def on_room_rejoin_result(data):
+    if data.get("success"):
+        global room
+        room = data.get("room")
+        logging.info("Successfully rejoined room: %s", room)
+        print("Rejoined room:", room)
+    else:
+        logging.error("Failed to rejoin room.")
+        print("Failed to rejoin room.")
+
+@sio.on("bot_response")
+def on_bot_response(data):
+    logging.info("Received bot response: %s", data)
+    print("Bot response received:")
+    print(json.dumps(data, indent=2))
+
+@sio.on("task_status")
+def on_task_status(data):
+    logging.info("Received task status: %s", data)
+    print("Task status:")
+    print(json.dumps(data, indent=2))
+
+# ----------------------------
+# Helper: Retrieve token from REST API
+# ----------------------------
+def retrieve_token(api_url: str) -> str:
+    payload = {
+        "tenant_id": TENANT_ID,
+        "api_key": API_KEY
+    }
+    try:
+        logging.info("Requesting token from %s with payload: %s", api_url, payload)
+        response = requests.post(api_url, json=payload)
+        response.raise_for_status()
+        token = response.json()["access_token"]
+        logging.info("Token retrieved successfully.")
+        return token
+    except Exception as e:
+        logging.error("Failed to retrieve token: %s", e)
+        raise e
+
+# ----------------------------
+# Main Interactive UI Function
+# ----------------------------
+def main():
+    global room
+
+    # Retrieve the token
+    auth_url = f"{BASE_API_URL}/auth/token"
+    try:
+        token = retrieve_token(auth_url)
+        print("Token retrieved successfully.")
+    except Exception as e:
+        print("Error retrieving token. Check logs for details.")
+        sys.exit(1)
+
+    # Parse the BASE_SOCKET_URL
+    parsed_url = urlparse(BASE_SOCKET_URL)
+    host_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
+
+    # Connect to the Socket.IO server.
+    # Note: Use `auth` instead of `query_string` (the official client uses the `auth` parameter)
+    try:
+        sio.connect(
+            host_url,
+            socketio_path='/chat/socket.io',
+            auth={"token": token},
+        )
+    except Exception as e:
+        logging.error("Failed to connect to Socket.IO server: %s", e)
+        print("Failed to connect to Socket.IO server:", e)
+        sys.exit(1)
+
+    # Allow time for authentication and room assignment.
+    time.sleep(2)
+    if not room:
+        logging.warning("No room assigned. Exiting.")
+        print("No room assigned by the server. Exiting.")
+        sio.disconnect()
+        sys.exit(1)
+
+    # Load specialist configuration from YAML.
+    try:
+        with open(CONFIG_FILE, "r") as f:
+            specialist_config = yaml.safe_load(f)
+        arg_config = specialist_config.get("arguments", {})
+        logging.info("Loaded specialist argument configuration: %s", arg_config)
+    except Exception as e:
+        logging.error("Failed to load specialist configuration: %s", e)
+        print("Failed to load specialist configuration. Exiting.")
+        sys.exit(1)
+
+    # Dictionary to store default values for static arguments (except "query")
+    static_defaults = {}
+
+    print("\nInteractive Specialist Client")
+    print("For each iteration, you will be prompted for the following arguments:")
+    for key, details in arg_config.items():
+        print(f" - {details.get('name', key)}: {details.get('description', '')}")
+    print("Type 'quit' or 'exit' as the query to end the session.\n")
+
+    # Interactive loop: prompt for arguments and send user message.
+    while True:
+        current_arguments = {}
+        for arg_key, arg_details in arg_config.items():
+            prompt_msg = f"Enter {arg_details.get('name', arg_key)}"
+            desc = arg_details.get("description", "")
+            if desc:
+                prompt_msg += f" ({desc})"
+            if arg_key != "query":
+                default_value = static_defaults.get(arg_key, "")
+                if default_value:
+                    prompt_msg += f" [default: {default_value}]"
+                prompt_msg += ": "
+                value = input(prompt_msg).strip()
+                if not value:
+                    value = default_value
+                static_defaults[arg_key] = value
+            else:
+                prompt_msg += " (required): "
+                value = input(prompt_msg).strip()
+                while not value:
+                    print("Query is required. Please enter a value.")
+                    value = input(prompt_msg).strip()
+            current_arguments[arg_key] = value
+
+        if current_arguments.get("query", "").lower() in ["quit", "exit"]:
+            break
+
+        try:
+            timezone = datetime.now().astimezone().tzname()
+        except Exception:
+            timezone = "UTC"
+
+        payload = {
+            "token": token,
+            "tenant_id": TENANT_ID,
+            "specialist_id": SPECIALIST_ID,
+            "arguments": current_arguments,
+            "timezone": timezone,
+            "room": room
+        }
+
+        logging.info("Sending user_message with payload: %s", payload)
+        print("Sending message to specialist...")
+        sio.emit("user_message", payload)
+        time.sleep(1)
+
+    print("Exiting interactive session.")
+    sio.disconnect()
+
+if __name__ == "__main__":
+    main()

tests/specialist_execution/test_specialist_client.py

@@ -0,0 +1,225 @@
+# test_specialist_client.py
+from pathlib import Path
+
+import requests
+import json
+from datetime import datetime
+import sseclient
+from typing import Dict, Any
+import yaml
+import os
+from termcolor import colored
+import sys
+
+project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))
+sys.path.append(project_root)
+
+# Configuration Constants
+API_BASE_URL = "http://macstudio.ask-eve-ai-local.com:8080/api/api/v1"
+TENANT_ID = 2  # Replace with your tenant ID
+API_KEY = "EveAI-5096-5466-6143-1487-8085-4174-2080-7208"  # Replace with your API key
+SPECIALIST_TYPE = "SPIN_SPECIALIST"  # Replace with your specialist type
+SPECIALIST_ID = 5  # Replace with your specialist ID
+ROOT_FOLDER = "../.."
+
+
+def get_auth_token() -> str:
+    """Get authentication token from API"""
+    response = requests.post(
+        f"{API_BASE_URL}/auth/token",
+        json={
+            "tenant_id": TENANT_ID,
+            "api_key": API_KEY
+        }
+    )
+    print(colored(f"Status Code: {response.status_code}", "cyan"))
+    print(colored(f"Response Headers: {response.headers}", "cyan"))
+    print(colored(f"Response Content: {response.text}", "cyan"))
+    if response.status_code == 200:
+        return response.json()['access_token']
+    else:
+        raise Exception(f"Authentication failed: {response.text}")
+
+
+def get_session_id(auth_token: str) -> str:
+    """Get a new session ID from the API"""
+    headers = {'Authorization': f'Bearer {auth_token}'}
+    response = requests.get(
+        f"{API_BASE_URL}/specialist-execution/start_session",
+        headers=headers
+    )
+    response.raise_for_status()
+    return response.json()["session_id"]
+
+
+def load_specialist_config() -> Dict[str, Any]:
+    """Load specialist configuration from YAML file"""
+    config_path = f"{ROOT_FOLDER}/config/specialists/{SPECIALIST_TYPE}/1.0.0.yaml"
+    if not os.path.exists(config_path):
+        print(colored(f"Error: Configuration file not found: {config_path}", "red"))
+        sys.exit(1)
+
+    with open(config_path, 'r') as f:
+        return yaml.safe_load(f)
+
+
+def get_argument_value(arg_name: str, arg_config: Dict[str, Any], previous_value: Any = None) -> Any:
+    """Get argument value from user input"""
+    arg_type = arg_config.get('type', 'str')
+    description = arg_config.get('description', '')
+
+    # Show previous value if it exists
+    previous_str = f" (previous: {previous_value})" if previous_value is not None else ""
+
+    while True:
+        print(colored(f"\n{arg_name}: {description}{previous_str}", "cyan"))
+        value = input(colored("Enter value (or press Enter for previous): ", "yellow"))
+
+        if not value and previous_value is not None:
+            return previous_value
+
+        try:
+            if arg_type == 'int':
+                return int(value)
+            elif arg_type == 'float':
+                return float(value)
+            elif arg_type == 'bool':
+                return value.lower() in ('true', 'yes', '1', 't')
+            else:
+                return value
+        except ValueError:
+            print(colored(f"Invalid input for type {arg_type}. Please try again.", "red"))
+
+
+def get_specialist_arguments(config: Dict[str, Any], previous_args: Dict[str, Any] = None) -> Dict[str, Any]:
+    """Get all required arguments for specialist execution"""
+    arguments = {}
+    previous_args = previous_args or {}
+
+    for arg_name, arg_config in config.get('arguments', {}).items():
+        previous_value = previous_args.get(arg_name)
+        arguments[arg_name] = get_argument_value(arg_name, arg_config, previous_value)
+
+    return arguments
+
+
+def process_specialist_updates(task_id: str, auth_token: str):
+    """Process SSE updates from specialist execution"""
+    headers = {'Authorization': f'Bearer {auth_token}'}
+    url = f"{API_BASE_URL}/specialist-execution/{task_id}/stream"
+
+    print(colored("\nConnecting to execution stream...", "cyan"))
+
+    with requests.get(url, headers=headers, stream=True) as response:
+        response.raise_for_status()
+
+        for line in response.iter_lines():
+            if not line:
+                continue
+
+            line = line.decode('utf-8')
+            if not line.startswith('data: '):
+                continue
+
+            # Extract the data part
+            data = line[6:]  # Skip 'data: '
+
+            try:
+                update = json.loads(data)
+                update_type = update['processing_type']
+                data = update['data']
+                timestamp = update.get('timestamp', datetime.now().isoformat())
+
+                # Print updates in different colors based on type
+                if update_type.endswith('Start'):
+                    print(colored(f"\n[{timestamp}] {update_type}: {data}", "blue"))
+                elif update_type == 'EveAI Specialist Error':
+                    print(colored(f"\n[{timestamp}] Error: {data}", "red"))
+                    break
+                elif update_type == 'EveAI Specialist Complete':
+                    print(colored(f"\n[{timestamp}] {update_type}: {data}", "green"))
+                    print(colored(f"\n[{timestamp}] {type(data)}", "green"))
+                    print(colored("Full Results:\n", "grey"))
+                    formatted_data = json.dumps(data, indent=4)
+                    print(colored(formatted_data, "grey"))
+                    print(colored("Answer:\n", "cyan"))
+                    answer = data.get('result', {}).get('rag_output', {}).get('answer', "")
+                    print(colored(answer, "cyan"))
+                    break
+                elif update_type.endswith('Complete'):
+                    print(colored(f"\n[{timestamp}] {update_type}: {data}", "green"))
+                else:
+                    print(colored(f"\n[{timestamp}] {update_type}: {data.get('message', '')}", "white"))
+            except json.JSONDecodeError:
+                print(colored(f"Error decoding message: {data}", "red"))
+            except Exception as e:
+                print(colored(f"Error processing message: {str(e)}", "red"))
+
+
+def main():
+    try:
+        # Get authentication token
+        print(colored("Getting authentication token...", "cyan"))
+        auth_token = get_auth_token()
+
+        # Load specialist configuration
+        print(colored(f"Loading specialist configuration {SPECIALIST_TYPE}", "cyan"))
+        config = load_specialist_config()
+        previous_args = None
+
+        while True:
+            try:
+                # Get new session ID
+                print(colored("Getting session ID...", "cyan"))
+                session_id = get_session_id(auth_token)
+                print(colored(f"New session ID: {session_id}", "cyan"))
+
+                # Get arguments
+                arguments = get_specialist_arguments(config, previous_args)
+                previous_args = arguments
+
+                # Start specialist execution
+                print(colored("\nStarting specialist execution...", "cyan"))
+                headers = {
+                    'Authorization': f'Bearer {auth_token}',
+                    'Content-Type': 'application/json'
+                }
+                response = requests.post(
+                    f"{API_BASE_URL}/specialist-execution",
+                    headers=headers,
+                    json={
+                        'specialist_id': SPECIALIST_ID,
+                        'arguments': arguments,
+                        'session_id': session_id,
+                        'user_timezone': 'UTC'
+                    }
+                )
+                response.raise_for_status()
+
+                execution_data = response.json()
+                task_id = execution_data['task_id']
+                print(colored(f"Execution queued with Task ID: {task_id}", "cyan"))
+
+                # Process updates
+                process_specialist_updates(task_id, auth_token)
+
+                # Ask if user wants to continue
+                if input(colored("\nRun another execution? (y/n): ", "yellow")).lower() != 'y':
+                    break
+
+            except KeyboardInterrupt:
+                print(colored("\nExecution cancelled by user", "yellow"))
+                if input(colored("Run another execution? (y/n): ", "yellow")).lower() != 'y':
+                    break
+            except requests.exceptions.HTTPError as e:
+                print(colored(f"\nHTTP Error: {e.response.status_code} - {e.response.text}", "red"))
+                if input(colored("Try again? (y/n): ", "yellow")).lower() != 'y':
+                    break
+
+    except Exception as e:
+        print(colored(f"\nError: {str(e)}", "red"))
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()