File size: 2,222 Bytes
c17543d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 |
"""
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")
|