""" 文件摘要生成服务 基于 LangChain 和大模型生成文件内容的精准摘要。 参考 server/aaa/jenius_attachment_knowledge_base/jenius_rag.py 的实现。 """ import asyncio from typing import List, Optional try: from langchain_core.documents import Document except ImportError: from langchain.schema import Document from langchain_core.prompts import PromptTemplate from core.llm_catalog import build_chat_model from logger.logging import get_logger logger = get_logger(__name__) # 摘要生成 Prompt - 优化版：强调全面覆盖 GENERATE_SUMMARY_PROMPT = """ 你是一个精准的文件内容总结专家。你的任务是提取并总结用户提供的文件内容或片段的**所有核心内容**。 ## 核心要求 - **总结结果长度为150-300个字**，根据内容复杂度灵活调整 - **必须覆盖文档中的所有主要主题和关键信息点**，不能遗漏任何重要内容 - 完全基于提供的文件内容生成总结，不添加任何未在文件内容中出现的信息 - 如果文档包含多个不同主题（如：多张图片内容、多个段落主题），**必须逐一概括每个主题** - 对于包含数据、事实、人物、事件的内容，**必须保留具体细节**（如：人名、数字、时间等） - 直接输出总结结果，不包含任何引言、前缀或解释 ## 特别说明 - 如果文档包含**图片内容标记**（如 [图片 1 内容]、[图片 2 内容]），**必须总结每张图片的核心内容** - 如果文档包含**多个独立段落或章节**，**必须概括每个段落的要点** - 对于**人名、数字、时间、地点**等关键信息，**必须在摘要中体现** ## 格式与风格 - 使用客观、中立的第三人称陈述语气 - 使用清晰简洁的中文表达 - 保持逻辑连贯性，确保句与句之间有合理过渡 - 多个主题之间使用分号或换行分隔 - 以"此文件"开头，直接输出总结结果 ## 注意事项 - 绝对不输出"无法生成"、"无法总结"、"内容不足"等拒绝回应的词语 - 不要只总结开头或某一部分，**必须通读全文后再生成摘要** - 对于任何文本都尽最大努力提取**所有**重点并总结，无论长度或复杂度 ## 以下是用户给出的文件相关信息： {doc_content} """ class SummaryService: """文件摘要生成服务类""" _llm_cache = None _lock = asyncio.Lock() @classmethod async def _get_llm(cls): """获取或创建 LLM 实例（单例模式）""" if cls._llm_cache is not None: return cls._llm_cache async with cls._lock: if cls._llm_cache is None: cls._llm_cache = build_chat_model( provider="tongyi", api_model="qwen-plus-latest", streaming=False, temperature=0.3, # 适度提高灵活性，更好地总结全文 ) return cls._llm_cache @classmethod async def generate_file_summary( cls, docs: List[Document], max_docs: int = 2 ) -> str: """ 生成文件摘要 Args: docs: 文档列表 max_docs: 最多使用的文档数量（默认2个） Returns: str: 生成的摘要文本 """ if not docs: return "" try: llm = await cls._get_llm() # 限制文档数量，避免超长 docs = docs[:max_docs] # 合并文档内容，去除重叠部分 doc_content = cls._merge_doc_contents(docs) if not doc_content: return "" # 生成摘要 prompt = PromptTemplate( template=GENERATE_SUMMARY_PROMPT, input_variables=["doc_content"] ) chain = prompt | llm response = await chain.ainvoke({"doc_content": doc_content}) summary = response.content.strip() logger.info(f"成功生成文件摘要，长度: {len(summary)}") return summary except Exception as e: logger.error(f"生成文件摘要失败: {e}") return "" @classmethod def _merge_doc_contents(cls, docs: List[Document], overlap_size: int = 50) -> str: """ 合并文档内容，去除重叠部分 Args: docs: 文档列表 overlap_size: 重叠检测大小 Returns: str: 合并后的内容 """ if not docs: return "" # 简单去重策略 contents = [] for doc in docs: content = doc.page_content.strip() if content: contents.append(content) return "\n".join(contents) class ExcelSummaryService: """Excel 文件摘要生成服务""" EXCEL_DESCRIPTION_PROMPT = """ 指令：请根据以下 Excel 文件的内容，为每个工作表生成简洁的描述，然后再生成整个文件的简要描述。 Excel 结构如下（每个sheet提供前5行数据）： {sheet_description_array} sheet_description_array： 对每个sheet表的内容进行描述，不超过20字；工作表的数量为: {sheet_number}个； sheet_summary： 对所有sheet表的描述进行总结，不超过20字； 输出格式:JSON 输出格式示例如下： {{ "sheet_description_array": ["表1的描述","表2的描述"], "sheet_summary": "所有sheet表的简要描述", }} 请直接输出JSON格式的结果，不要输出其他内容。 """ _llm_cache = None _lock = asyncio.Lock() @classmethod async def _get_llm(cls): """获取或创建 LLM 实例""" if cls._llm_cache is not None: return cls._llm_cache async with cls._lock: if cls._llm_cache is None: cls._llm_cache = build_chat_model( provider="tongyi", api_model="qwen-plus-latest", streaming=False, temperature=0.7, model_kwargs={"response_format": {"type": "json_object"}}, ) return cls._llm_cache @classmethod async def generate_excel_description(cls, sheet_description_array: List[dict]) -> dict: """ 生成 Excel 文件的描述 Args: sheet_description_array: Sheet 描述数组，格式为 [{"sheet_name": "xxx", "sheet_data": "xxx"}] Returns: dict: 包含 sheet_summary 和 sheet_description_array 的字典 """ try: llm = await cls._get_llm() prompt = PromptTemplate( template=cls.EXCEL_DESCRIPTION_PROMPT, input_variables=["sheet_description_array", "sheet_number"] ) chain = prompt | llm sheet_number = len(sheet_description_array) response = await chain.ainvoke({ "sheet_description_array": sheet_description_array, "sheet_number": sheet_number }) result_dict = { "sheet_summary": "", "sheet_description_array": [] } # 解析 JSON 响应 import json try: result = json.loads(response.content) if "sheet_summary" in result: result_dict["sheet_summary"] = result["sheet_summary"] if "sheet_description_array" in result and len(result["sheet_description_array"]) == sheet_number: result_dict["sheet_description_array"] = result["sheet_description_array"] except json.JSONDecodeError as e: logger.error(f"解析 Excel 描述 JSON 失败: {e}") logger.info(f"成功生成 Excel 文件描述") return result_dict except Exception as e: logger.error(f"生成 Excel 文件描述失败: {e}") return {"sheet_summary": "", "sheet_description_array": []} class CSVSummaryService: """CSV 文件摘要生成服务""" CSV_DESCRIPTION_PROMPT = """ 指令：请根据以下 csv 文件的内容，生成整个文件的简要描述。 csv文件的文件名和前5行数据(包括表头和样例数据) {csv_description_dict} csv_description: 对csv表格的内容进行描述，不超过20字； 输出格式:JSON 输出格式示例如下： {{ "csv_description": "csv表格的描述" }} 请直接输出JSON格式的结果，不要输出其他内容。 """ _llm_cache = None _lock = asyncio.Lock() @classmethod async def _get_llm(cls): """获取或创建 LLM 实例""" if cls._llm_cache is not None: return cls._llm_cache async with cls._lock: if cls._llm_cache is None: cls._llm_cache = build_chat_model( provider="tongyi", api_model="qwen-plus-latest", streaming=False, temperature=0.7, model_kwargs={"response_format": {"type": "json_object"}}, ) return cls._llm_cache @classmethod async def generate_csv_description(cls, csv_description_dict: dict) -> dict: """ 生成 CSV 文件的描述 Args: csv_description_dict: CSV 描述字典，格式为 {"file_name": "xxx", "csv_data": "xxx"} Returns: dict: 包含 file_description 的字典 """ try: llm = await cls._get_llm() prompt = PromptTemplate( template=cls.CSV_DESCRIPTION_PROMPT, input_variables=["csv_description_dict"] ) chain = prompt | llm response = await chain.ainvoke({ "csv_description_dict": csv_description_dict }) result_dict = {"file_description": ""} # 解析 JSON 响应 import json try: result = json.loads(response.content) if "csv_description" in result: result_dict["file_description"] = result["csv_description"] except json.JSONDecodeError as e: logger.error(f"解析 CSV 描述 JSON 失败: {e}") logger.info(f"成功生成 CSV 文件描述") return result_dict except Exception as e: logger.error(f"生成 CSV 文件描述失败: {e}") return {"file_description": ""}