|
|
"""
|
|
|
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")
|
|
|
|
