213 lines
6.6 KiB
Python
213 lines
6.6 KiB
Python
"""
|
||
RAG 意图判断服务
|
||
基于 server 实现的智能路由策略
|
||
"""
|
||
import json
|
||
from typing import List, Dict, Optional
|
||
from pydantic import BaseModel, Field
|
||
from langchain_core.prompts import PromptTemplate
|
||
|
||
from core.llm_catalog import build_chat_model
|
||
from logger.logging import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
class FileIntent(BaseModel):
|
||
"""单个文件的意图判断结果"""
|
||
file_name: str = Field(description="文件名")
|
||
file_id: int = Field(description="文件ID")
|
||
question_type: str = Field(
|
||
description="问题类型: summary(需要全文), search(向量检索), excel_analysis(表格分析)",
|
||
default="search"
|
||
)
|
||
|
||
|
||
class RagIntentResult(BaseModel):
|
||
"""RAG 意图判断结果"""
|
||
result: List[FileIntent] = Field(description="涉及的文件及其处理方式", default=[])
|
||
|
||
|
||
# 意图判断的 Prompt（参考 server 实现）
|
||
INTENT_JUDGE_PROMPT = """
|
||
你是一个 RAG 问答系统的意图分类器。请根据用户的问题和文件摘要，判断：
|
||
1. 哪些文件与问题相关
|
||
2. 每个文件需要什么类型的处理
|
||
|
||
## 任务 1：过滤文件列表
|
||
- 从候选文件中选出与用户问题相关的文件
|
||
- 按关联度从高到低排序
|
||
- 若问题中有"本文"、"这个文件"等指代词，结合上下文判断
|
||
- 若无法判断相关文件，返回空数组
|
||
|
||
## 任务 2：问题类型判断
|
||
对每个文件，判断用户需要什么类型的处理：
|
||
|
||
### "summary" - 需要完整文件内容
|
||
适用于以下情况：
|
||
- 需要文件的全部内容才能回答（如：总结、概括、归纳、分析）
|
||
- 基于文件的整体内容问答
|
||
- **简单的事实查询**（如"XX是多少"、"XX排名第几"、"XX是什么"、"谁夺冠"等）
|
||
- 文件内容的改写、润色、翻译
|
||
- 图片内容的具体描述
|
||
- 问题中提到"文件"、"文档"、"文章"、"图片"等词语
|
||
- **🔑 重要：当不确定时，优先选择 summary！**
|
||
|
||
**示例**：
|
||
- "总结一下这个文档的主要内容"
|
||
- "詹姆斯得了多少分"（简单事实查询 → summary）
|
||
- "南京在苏超的排名是第几"（简单事实查询 → summary）
|
||
- "成年人的修养是什么"（需要完整文档内容 → summary）
|
||
- "翻译这篇文章"
|
||
|
||
### "search" - 只需部分内容（向量检索）
|
||
适用于以下情况：
|
||
- 只需要在文件中定位、查找或提取特定的、局部的内容
|
||
- 基于关键词的搜索
|
||
- 问题明确指向某个具体片段
|
||
|
||
**示例**：
|
||
- "文件中哪里提到了xxx"
|
||
- "找出关于xxx的段落"
|
||
- "第三章讲了什么"
|
||
|
||
### "excel_analysis" - 表格数据分析
|
||
适用于以下情况：
|
||
- 文件类型必须为 xlsx、xls、csv
|
||
- 基于表格的数据问答、筛选、排序、汇总、统计分析
|
||
- 查询单元格、行、列数据
|
||
|
||
**示例**：
|
||
- "A1单元格是什么"
|
||
- "第二行第三列的值是多少"
|
||
- "计算平均值"
|
||
|
||
## 输入信息
|
||
候选文件列表（按上传时间排序，最后一个为最新）:
|
||
{{ file_list }}
|
||
|
||
文件摘要信息:
|
||
{{ file_summaries }}
|
||
|
||
用户问题:
|
||
{{ query }}
|
||
|
||
## 输出格式
|
||
严格按照以下 JSON 格式输出，不要输出其他内容：
|
||
```json
|
||
{
|
||
"result": [
|
||
{
|
||
"file_name": "文件名",
|
||
"file_id": 123,
|
||
"question_type": "summary"
|
||
}
|
||
]
|
||
}
|
||
```
|
||
|
||
如果没有相关文件，返回：
|
||
```json
|
||
{
|
||
"result": []
|
||
}
|
||
```
|
||
"""
|
||
|
||
|
||
class RagIntentService:
|
||
"""RAG 意图判断服务"""
|
||
|
||
def __init__(self):
|
||
self.model = build_chat_model(
|
||
provider="tongyi",
|
||
api_model="qwen-plus-latest",
|
||
streaming=False,
|
||
temperature=0.1, # 降低温度，让判断更稳定
|
||
)
|
||
|
||
async def judge_intent(
|
||
self,
|
||
query: str,
|
||
file_list: List[Dict[str, any]],
|
||
chat_history: Optional[List[str]] = None
|
||
) -> List[FileIntent]:
|
||
"""
|
||
判断用户问题的 RAG 意图
|
||
|
||
Args:
|
||
query: 用户问题
|
||
file_list: 文件列表 [{"file_id": 1, "file_name": "test.docx", "summary": "..."}]
|
||
chat_history: 聊天历史（可选）
|
||
|
||
Returns:
|
||
List[FileIntent]: 涉及的文件及其处理方式
|
||
"""
|
||
try:
|
||
# 构建文件列表字符串
|
||
file_names = [f["file_name"] for f in file_list]
|
||
file_list_str = ", ".join(file_names)
|
||
|
||
# 构建文件摘要字符串
|
||
file_summaries_str = ""
|
||
for f in file_list:
|
||
file_summaries_str += f"【{f['file_name']}】（ID: {f['file_id']}）:\n"
|
||
summary = f.get('summary', '无摘要')
|
||
# 截取摘要前 500 字符（避免过长）
|
||
if len(summary) > 500:
|
||
summary = summary[:500] + "..."
|
||
file_summaries_str += f"{summary}\n\n"
|
||
|
||
# 构建完整输入
|
||
full_query = query
|
||
if chat_history:
|
||
history_str = "\n".join(chat_history[-3:]) # 最近3轮对话
|
||
full_query = f"【聊天历史】\n{history_str}\n\n【当前问题】\n{query}"
|
||
|
||
# 创建 Prompt
|
||
prompt_template = PromptTemplate(
|
||
template=INTENT_JUDGE_PROMPT,
|
||
input_variables=["file_list", "file_summaries", "query"],
|
||
template_format="jinja2"
|
||
)
|
||
|
||
# 调用 LLM 判断意图（使用 Pydantic schema）
|
||
chain = prompt_template | self.model.with_structured_output(
|
||
schema=RagIntentResult
|
||
)
|
||
|
||
result = await chain.ainvoke({
|
||
"file_list": file_list_str,
|
||
"file_summaries": file_summaries_str,
|
||
"query": full_query
|
||
})
|
||
|
||
# 解析结果（with_structured_output 直接返回 Pydantic 对象）
|
||
if isinstance(result, RagIntentResult):
|
||
intents = result.result
|
||
|
||
logger.info(f"意图判断完成: {len(intents)} 个文件")
|
||
for intent in intents:
|
||
logger.info(f" - {intent.file_name} ({intent.file_id}): {intent.question_type}")
|
||
|
||
return intents
|
||
else:
|
||
logger.warning(f"意图判断返回格式异常: {type(result)}")
|
||
return []
|
||
|
||
except Exception as e:
|
||
logger.error(f"意图判断失败: {e}")
|
||
return []
|
||
|
||
|
||
# 全局实例（单例模式）
|
||
_intent_service = None
|
||
|
||
|
||
async def get_rag_intent_service() -> RagIntentService:
|
||
"""获取 RAG 意图服务实例（单例）"""
|
||
global _intent_service
|
||
if _intent_service is None:
|
||
_intent_service = RagIntentService()
|
||
return _intent_service
|