agent-platform/services/knowledge-service/app/application/settings_service.py

"""Knowledge settings read/write sub-service."""

from __future__ import annotations

from typing import TYPE_CHECKING

from core_shared import JSONValue

from app.bootstrap.settings import KnowledgeServiceSettings
from app.schemas.knowledge import (
    KnowledgeSettingsDto,
    KnowledgeSettingsUpdateRequestDto,
)

if TYPE_CHECKING:
    from app.domain.repositories import KnowledgeBaseRepository


class KnowledgeSettingsService:
    def __init__(
        self,
        *,
        settings: KnowledgeServiceSettings,
        base_repository: KnowledgeBaseRepository,
    ) -> None:
        self.settings = settings
        self.base_repository = base_repository

    def read_settings(self, *, knowledge_base_id: str | None = None) -> KnowledgeSettingsDto:
        base_config: dict[str, JSONValue] = {}
        if knowledge_base_id:
            base = self.base_repository.get_by_id(knowledge_base_id=knowledge_base_id)
            if base is not None and isinstance(base.metadata_json, dict):
                value = base.metadata_json.get("retrieval_config")
                if isinstance(value, dict):
                    base_config = value
        defaults = KnowledgeSettingsDto(
            knowledgeBaseId=knowledge_base_id,
            chunkSize=self.settings.default_chunk_size,
            chunkOverlap=self.settings.default_chunk_overlap,
            keywordWeight=self.settings.retrieval_keyword_weight,
            vectorWeight=self.settings.retrieval_vector_weight,
            rerankWeight=self.settings.retrieval_rerank_weight,
            queryRewrite=False,
            requireCitations=True,
        )
        return KnowledgeSettingsDto.model_validate({
            **defaults.model_dump(),
            **base_config,
            "knowledgeBaseId": knowledge_base_id,
        })

    def update_settings(
        self,
        payload: KnowledgeSettingsUpdateRequestDto,
    ) -> KnowledgeSettingsDto:
        settings = KnowledgeSettingsDto.model_validate({
            **payload.model_dump(),
            "knowledgeBaseId": payload.knowledgeBaseId,
        })
        if payload.knowledgeBaseId:
            base = self.base_repository.get_by_id(knowledge_base_id=payload.knowledgeBaseId)
            if base is not None:
                metadata = dict(base.metadata_json or {})
                metadata["retrieval_config"] = settings.model_dump(exclude={"knowledgeBaseId"})
                self.base_repository.update(
                    knowledge_base_id=payload.knowledgeBaseId,
                    metadata_json=metadata,
                )
        return settings