Building Reliable LLM Chain Architecture: From Fundamentals to Practice

from typing import Any, Dict, Optional
from abc import ABC, abstractmethod

class BaseProcessor(ABC):
    @abstractmethod
    def process(self, data: Any) -> Any:
        pass

class BaseChain:
    def __init__(self):
        self.preprocessor: Optional[BaseProcessor] = None
        self.prompt_manager: Optional[PromptManager] = None
        self.llm: Optional[BaseLLM] = None
        self.postprocessor: Optional[BaseProcessor] = None
        self.error_handler: Optional[ErrorHandler] = None

    def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        try:
            # 1. Preprocessing
            processed_input = self._preprocess(input_data)

            # 2. Generate prompt
            prompt = self._generate_prompt(processed_input)

            # 3. LLM call
            response = self._call_llm(prompt)

            # 4. Postprocessing
            result = self._postprocess(response)

            return result
        except Exception as e:
            return self.error_handler.handle(e)

class PreProcessor(BaseProcessor):
    def process(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Data preprocessing logic"""
        # 1. Data cleaning
        cleaned_data = self._clean_data(data)

        # 2. Format conversion
        formatted_data = self._format_data(cleaned_data)

        # 3. Validation
        self._validate_data(formatted_data)

        return formatted_data

class PostProcessor(BaseProcessor):
    def process(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Result post-processing logic"""
        # 1. Result parsing
        parsed_result = self._parse_result(data)

        # 2. Format output
        formatted_result = self._format_output(parsed_result)

        # 3. Quality check
        self._quality_check(formatted_result)

        return formatted_result

class PromptTemplate:
    def __init__(self, template: str, input_variables: List[str]):
        self.template = template
        self.input_variables = input_variables

class PromptManager:
    def __init__(self):
        self.templates: Dict[str, PromptTemplate] = {}
        self.version_control = VersionControl()

    def register_template(self, name: str, template: str, 
                         input_variables: List[str]) -> None:
        """Register prompt template"""
        self.templates[name] = PromptTemplate(
            template=template,
            input_variables=input_variables
        )

    def generate_prompt(self, template_name: str, **kwargs) -> str:
        """Generate prompt"""
        template = self.templates.get(template_name)
        if not template:
            raise ValueError(f"Template {template_name} not found")

        # Validate required parameters
        self._validate_inputs(template, kwargs)

        # Generate prompt
        return template.template.format(**kwargs)

class PromptOptimizer:
    def __init__(self):
        self.few_shots: List[Dict[str, str]] = []
        self.context: Dict[str, Any] = {}

    def add_few_shot(self, example: Dict[str, str]) -> None:
        """Add few-shot example"""
        self.few_shots.append(example)

    def set_context(self, context: Dict[str, Any]) -> None:
        """Set context information"""
        self.context.update(context)

    def optimize_prompt(self, base_prompt: str) -> str:
        """Optimize prompt"""
        # 1. Add role setting
        prompt = self._add_role_setting(base_prompt)

        # 2. Inject context
        prompt = self._inject_context(prompt)

        # 3. Add few-shot examples
        prompt = self._add_few_shots(prompt)

        return prompt

class LLMChainError(Exception):
    """Base chain error"""
    pass

class ErrorHandler:
    def __init__(self):
        self.retry_strategy = RetryStrategy()
        self.fallback_handler = FallbackHandler()
        self.monitor = Monitor()

    def handle(self, error: Exception) -> Dict[str, Any]:
        """Unified error handling"""
        try:
            # 1. Log error
            self.monitor.log_error(error)

            # 2. Check if retryable
            if self.is_retryable(error):
                return self.retry_strategy.retry()

            # 3. Fallback handling
            return self.fallback_handler.handle(error)
        finally:
            # 4. Error notification
            self.monitor.notify(error)

class RetryStrategy:
    def __init__(self, max_retries: int = 3, 
                 base_delay: float = 1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.current_retry = 0

    def retry(self) -> bool:
        """Implement exponential backoff retry"""
        if self.current_retry >= self.max_retries:
            return False

        delay = self.base_delay * (2 ** self.current_retry)
        time.sleep(delay)

        self.current_retry += 1
        return True

class QAChain(BaseChain):
    def __init__(self):
        super().__init__()
        self.setup_components()

    def setup_components(self):
        # 1. Set up preprocessor
        self.preprocessor = QAPreProcessor()

        # 2. Configure prompt manager
        self.prompt_manager = self._setup_prompt_manager()

        # 3. Configure LLM
        self.llm = self._setup_llm()

        # 4. Set up postprocessor
        self.postprocessor = QAPostProcessor()

        # 5. Configure error handling
        self.error_handler = QAErrorHandler()

    def _setup_prompt_manager(self):
        manager = PromptManager()
        manager.register_template(
            "qa_template",
            """
            As an intelligent Q&A assistant, please answer the following question:
            Question: {question}
            Requirements:
            1. Answer should be concise and clear
            2. If uncertain, please explicitly state so
            3. If more information is needed, specify what information is required
            """,
            ["question"]
        )
        return manager