320 lines
11 KiB
Python
320 lines
11 KiB
Python
"""
|
||
文件摘要生成服务
|
||
|
||
基于 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": ""}
|