RAG Solution Development Guide¶

from sqlalchemy.orm import Session
from typing import Optional
from uuid import UUID

class BaseService:
    """Base class for all services."""

    def __init__(self, db: Session):
        """Initialize service with database session."""
        self.db = db
        self._initialize_dependencies()

    def _initialize_dependencies(self) -> None:
        """Initialize service dependencies."""
        pass

class ExampleService(BaseService):
    """Example service demonstrating architecture."""

    def __init__(self, db: Session):
        """Initialize service with dependencies."""
        super().__init__(db)
        self._repository: Optional[ExampleRepository] = None

    @property
    def repository(self) -> ExampleRepository:
        """Lazy initialization of repository."""
        if self._repository is None:
            self._repository = ExampleRepository(self.db)
        return self._repository

# Service initialization
db = SessionLocal()
provider_service = LLMProviderService(db)
parameters_service = LLMParametersService(db)
template_service = PromptTemplateService(db)

# Get provider and parameters
provider = provider_service.get_provider_by_name("watsonx")
parameters = parameters_service.get_parameters(parameters_id)

# Create template
template = template_service.create_template(
    PromptTemplateInput(
        name="example-template",
        provider="watsonx",
        template_type=PromptTemplateType.RAG_QUERY,
        template_format="Context:\n{context}\nQuestion:{question}"
    )
)

from core.custom_exceptions import (
    ValidationError,
    NotFoundError,
    ConfigurationError
)

try:
    # Service operations
    provider = provider_service.get_provider_by_name("watsonx")
    parameters = parameters_service.get_parameters(parameters_id)
    template = template_service.get_by_type(PromptTemplateType.RAG_QUERY)

except ValidationError as e:
    logger.error(f"Validation error: {str(e)}")
except NotFoundError as e:
    logger.error(f"Not found error: {str(e)}")
except ConfigurationError as e:
    logger.error(f"Configuration error: {str(e)}")

class PipelineService(BaseService):
    """Service for managing RAG pipelines."""

    def __init__(self, db: Session):
        """Initialize pipeline service."""
        super().__init__(db)
        self._pipeline_repository: Optional[PipelineRepository] = None
        self._provider_service: Optional[LLMProviderService] = None
        self._template_service: Optional[PromptTemplateService] = None
        self._parameters_service: Optional[LLMParametersService] = None
        self._evaluator: Optional[RAGEvaluator] = None

    async def initialize(self, collection_id: UUID) -> None:
        """Initialize pipeline components."""
        # Initialize services
        self._initialize_services()

        # Get configuration
        config = await self._get_pipeline_config(collection_id)

        # Initialize components
        self.provider = self.provider_service.get_provider_by_name(config.provider)
        self.parameters = self.parameters_service.get_parameters(config.parameters_id)
        self.template = self.template_service.get_by_type(
            PromptTemplateType.RAG_QUERY,
            collection_id
        )

        # Initialize evaluator
        self._evaluator = RAGEvaluator(
            provider=self.provider,
            parameters=self.parameters
        )

    async def execute_pipeline(
        self,
        search_input: SearchInput,
        user_id: Optional[UUID] = None
    ) -> PipelineResult:
        """Execute RAG pipeline with error handling and evaluation."""
        try:
            # Rewrite query
            rewritten_query = await self.query_rewriter.rewrite(
                search_input.question
            )

            # Retrieve relevant documents
            query_results = await self.retriever.search(
                rewritten_query,
                search_input.collection_id,
                search_input.metadata
            )

            # Generate answer
            generated_answer = await self._generate_answer(
                rewritten_query,
                query_results,
                search_input.metadata
            )

            # Evaluate result
            evaluation = await self._evaluator.evaluate(
                question=search_input.question,
                answer=generated_answer,
                context=self._get_context(query_results)
            )

            return PipelineResult(
                rewritten_query=rewritten_query,
                query_results=query_results,
                generated_answer=generated_answer,
                evaluation=evaluation
            )

        except Exception as e:
            logger.error(f"Pipeline error: {str(e)}")
            return PipelineResult(
                rewritten_query="",
                query_results=[],
                generated_answer="",
                evaluation={"error": str(e)}
            )

class QuestionService(BaseService):
    """Service for managing question suggestions."""

    def __init__(
        self,
        db: Session,
        provider: LLMProvider,
        config: Optional[Dict[str, Any]] = None
    ):
        """Initialize question service."""
        super().__init__(db)
        self.provider = provider
        self.config = config or {}
        self._question_repository: Optional[QuestionRepository] = None
        self._template_service: Optional[PromptTemplateService] = None

    async def suggest_questions(
        self,
        texts: List[str],
        collection_id: UUID,
        num_questions: Optional[int] = None
    ) -> List[str]:
        """Generate questions from texts."""
        # Get template
        template = self.template_service.get_by_type(
            PromptTemplateType.QUESTION_GENERATION,
            collection_id
        )

        # Format prompt
        formatted_prompt = self.template_service.format_prompt(
            template.id,
            {
                "context": "\n\n".join(texts),
                "num_questions": str(num_questions or self.num_questions)
            }
        )

        # Generate questions
        questions = await self._generate_questions(
            formatted_prompt,
            template
        )

        return questions

class PromptTemplateService(BaseService):
    """Service for managing prompt templates."""

    def __init__(self, db: Session):
        """Initialize template service."""
        super().__init__(db)
        self._template_repository: Optional[PromptTemplateRepository] = None

    def get_by_type(
        self,
        template_type: PromptTemplateType,
        collection_id: Optional[UUID] = None
    ) -> Optional[PromptTemplate]:
        """Get template by type."""
        return self.template_repository.get_by_type(
            template_type,
            collection_id
        )

    def format_prompt(
        self,
        template_id: UUID,
        variables: Dict[str, Any]
    ) -> str:
        """Format prompt with variables."""
        template = self.template_repository.get_by_id(template_id)
        if not template:
            raise NotFoundError(f"Template {template_id} not found")

        return template.format_prompt(variables)

# Provider Configuration
provider_config = ProviderConfigInput(
    provider="watsonx",
    api_key="${WATSONX_APIKEY}",
    project_id="${WATSONX_INSTANCE_ID}",
    active=True
)
config_service.create_provider_config(provider_config)

# Template Configuration
template = PromptTemplateInput(
    name="rag-query",
    provider="watsonx",
    template_type=PromptTemplateType.RAG_QUERY,
    template_format="Context:\n{context}\nQuestion:{question}"
)
template_service.create_template(template)