File size: 3,338 Bytes
01d5a5d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
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}")