| """
|
| Abstract base class for communicator components.
|
| """
|
| from typing import Dict, List, Optional, Any, Union, Tuple
|
| import torch
|
| from base_interfaces.common_types import *
|
| from base_interfaces.model_interface import AbstractModel
|
| from abc import ABC, abstractmethod
|
|
|
| class AbstractCommunicator(ABC):
|
| """Minimal communicator interface"""
|
| @abstractmethod
|
| def process_input(self, *args, **kwargs):
|
| pass
|
|
|
| def process_input(self, input_text: str, context: Optional[Dict] = None) -> Dict[str, Any]:
|
| """Process user input through the appropriate model(s) and generate response."""
|
| raise NotImplementedError("Communicator must implement process_input")
|
|
|
| def process_request(self, prompt: str, model: Any) -> str:
|
| """Process a user request through the selected model"""
|
| raise NotImplementedError("Communicator must implement process_request")
|
|
|
| def route_input(self, input_text: str, query: Optional[str] = None) -> List[tuple]:
|
| """Route input to most relevant specializations, returning top-k matches."""
|
| raise NotImplementedError("Communicator must implement route_input")
|
|
|
| def _extract_subject(self, text: str) -> str:
|
| """Extract the primary subject from a text prompt."""
|
| raise NotImplementedError("Communicator must implement _extract_subject")
|
|
|
| def prepare_model_input(self, text: str, model) -> Dict:
|
| """Prepare input text for model processing."""
|
| raise NotImplementedError("Communicator must implement prepare_model_input")
|
|
|
| def _get_fallback_response(self, prompt: str) -> str:
|
| """Get a fallback response when primary model processing fails."""
|
| raise NotImplementedError("Communicator must implement _get_fallback_response")
|
|
|
| def clear_conversation_history(self):
|
| """Clear the conversation history"""
|
| raise NotImplementedError("Communicator must implement clear_conversation_history")
|
|
|
| def get_conversation_history(self) -> List[Dict]:
|
| """Get the current conversation history"""
|
| raise NotImplementedError("Communicator must implement get_conversation_history")
|
|
|
