lpm_kernel/kernel/l0_base.py

from typing import Dict, Optional
import logging
import time
from lpm_kernel.file_data.document import Document
from lpm_kernel.L0.l0_generator import L0Generator
from lpm_kernel.L0.models import (
    InsighterInput,
    SummarizerInput,
    FileInfo,
    BioInfo,
    DocumentType,
)
from lpm_kernel.configs.config import Config
from lpm_kernel.file_data.document_dto import DocumentDTO


logger = logging.getLogger(__name__)


class InsightKernel:
    def __init__(self):
        self.generator = L0Generator()
        config = Config.from_env()
        self.preferred_language = config.get("PREFER_LANGUAGE", "en")

    def analyze(self, doc: DocumentDTO) -> Dict:
        """Generate document insight"""
        try:
            self.generator.preferred_language = self.preferred_language
            document_type = DocumentType.from_mime_type(doc.mime_type)

            if document_type is DocumentType.TEXT:
                return {
                    "title": "",
                    "insight": doc.raw_content,
                }

            # Prepare input data
            file_info = FileInfo(
                data_type=document_type.value,
                filename=doc.name,
                content="",
                file_content={"content": doc.raw_content},
            )

            bio_info = BioInfo(global_bio="", status_bio="", about_me="")

            insighter_input = InsighterInput(file_info=file_info, bio_info=bio_info)

            insight_result = self.generator.insighter(insighter_input)

            return {
                "title": insight_result.get("title"),
                "insight": insight_result.get("insight"),
            }

        except Exception as e:
            logger.error(f"Failed to generate insight: {str(e)}", exc_info=True)
            raise Exception(f"Error generating insight: {e}")


class SummaryKernel:
    def __init__(self):
        self.generator = L0Generator()
        config = Config.from_env()
        self.preferred_language = config.get("PREFER_LANGUAGE", "en")

    def analyze(self, doc: DocumentDTO, insight: str = "") -> Dict:
        """Generate document summary"""

        try:
            self.generator.preferred_language = self.preferred_language
            document_type = DocumentType.from_mime_type(doc.mime_type)

            if document_type is DocumentType.TEXT:
                return {
                    "title": "",
                    "summary": doc.raw_content,
                    "keywords": [],
                }

            # Prepare input data
            file_info = FileInfo(
                data_type=document_type.value,
                filename=doc.name,
                content="",
                file_content={"content": doc.raw_content},
            )

            summarizer_input = SummarizerInput(file_info=file_info, insight=insight)

            # Call LLM
            summary_result = self.generator.summarizer(summarizer_input)

            return {
                "title": summary_result.get("title"),
                "summary": summary_result.get("summary"),
                "keywords": summary_result.get("keywords", []),
            }

        except Exception as e:
            logger.error(f"Failed to generate summary: {str(e)}", exc_info=True)
            raise Exception(f"Error generating summary: {e}")