e761af6111
- intent_analyzer._parse_json: 解析结果为 list 时取第一个 dict 元素 - llm_utils.parse_json_from_response: 同样处理 list 返回值 - 修复日志错误: 'list' object has no attribute 'get'
650 lines
26 KiB
Python
650 lines
26 KiB
Python
"""
|
||
意图分析器 - 改写 + 双层判断
|
||
|
||
核心功能：
|
||
1. 问题改写：指代消解、省略补全
|
||
2. 双层判断：
|
||
- 历史是否可答
|
||
- 是否需要外部知识
|
||
3. 输出：决策（use_context / need_retrieval）
|
||
4. 语义缓存：相似问题复用结果
|
||
|
||
设计原则：
|
||
- 使用轻量 LLM 完成改写和判断
|
||
- 替代硬编码规则，更灵活可维护
|
||
- 语义缓存减少 LLM 调用
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import hashlib
|
||
import threading
|
||
from dataclasses import dataclass, asdict
|
||
from typing import List, Dict, Optional, Tuple
|
||
|
||
from config import INTENT_TEMPERATURE, INTENT_MAX_TOKENS, INTENT_HISTORY_WINDOW, INTENT_MODEL
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class IntentAnalysis:
|
||
"""意图分析结果"""
|
||
rewritten_query: str # 改写后的完整问题
|
||
use_context: bool # 是否使用上下文回答
|
||
need_retrieval: bool # 是否需要检索
|
||
confidence: float # 置信度
|
||
reason: str # 判断理由
|
||
context_images: List[dict] # 上下文中的图片信息
|
||
sub_queries: List[str] # 用于检索的子查询列表
|
||
intent: str # 意图类型: factual/comparison/reasoning/instruction/other
|
||
|
||
def to_dict(self) -> dict:
|
||
"""转换为字典（用于缓存）"""
|
||
return {
|
||
"rewritten_query": self.rewritten_query,
|
||
"use_context": self.use_context,
|
||
"need_retrieval": self.need_retrieval,
|
||
"confidence": self.confidence,
|
||
"reason": self.reason,
|
||
"context_images": self.context_images,
|
||
"sub_queries": self.sub_queries,
|
||
"intent": self.intent
|
||
}
|
||
|
||
@classmethod
|
||
def from_dict(cls, data: dict) -> "IntentAnalysis":
|
||
"""从字典创建（用于缓存读取）"""
|
||
return cls(
|
||
rewritten_query=data.get("rewritten_query", ""),
|
||
use_context=data.get("use_context", False),
|
||
need_retrieval=data.get("need_retrieval", True),
|
||
confidence=data.get("confidence", 0.5),
|
||
reason=data.get("reason", ""),
|
||
context_images=data.get("context_images", []),
|
||
sub_queries=data.get("sub_queries", []),
|
||
intent=data.get("intent", "factual")
|
||
)
|
||
|
||
|
||
class IntentAnalyzer:
|
||
"""
|
||
意图分析器
|
||
|
||
使用 LLM 一次调用完成：
|
||
1. 问题改写（指代消解）
|
||
2. 意图判断（是否需要检索）
|
||
"""
|
||
|
||
SYSTEM_PROMPT = """你是一个意图分析助手。你的任务是分析用户问题，判断它需要什么类型的回答。
|
||
|
||
## 任务说明
|
||
|
||
根据对话历史和当前用户消息，输出一个 JSON 对象，包含以下字段：
|
||
|
||
1. **rewritten_query**: 改写后的完整问题
|
||
- **指代消解**：如果问题包含指代（如"这两张图片"、"继续说"），将其改写为完整、独立的问题
|
||
- 例如："分析一下这两张图片" → "分析一下对话历史中提到的图片"
|
||
- **追问补全**：如果问题是省略式追问（省略了上一轮讨论的主题实体），必须补全为完整问题
|
||
- 判断方法：当前问题缺少主语/宾语，且对话历史中可以推断出省略的实体
|
||
- 补全方法：从上一轮用户问题中提取主题实体，与追问组合成完整问题
|
||
- 例如：
|
||
- 上一轮问"吸烟点C1类是什么区？"，追问"有完整表格吗？" → "吸烟点C1类有完整表格吗？"
|
||
- 上一轮问"三峡工程的投资情况"，追问"建设地点在哪？" → "三峡工程的建设地点在哪？"
|
||
- 上一轮问"货源投放有哪些原则？"，追问"具体内容是什么？" → "货源投放原则的具体内容是什么？"
|
||
- 如果问题本身已经完整且独立，直接返回原文
|
||
|
||
2. **use_context**: 布尔值
|
||
- true: 问题依赖历史对话中的信息，答案已经在历史回答中
|
||
- false: 问题是新问题，需要新的回答
|
||
|
||
3. **need_retrieval**: 布尔值
|
||
- true: 需要从知识库检索信息才能回答
|
||
- false: 可以直接使用历史上下文或自己的知识回答
|
||
|
||
4. **reason**: 简短说明判断理由
|
||
|
||
5. **sub_queries**: 用于检索的子查询列表
|
||
- 对比类（intent="comparison"）：生成最多3个子查询
|
||
* 关于A的一个最佳检索查询
|
||
* 关于B的一个最佳检索查询
|
||
* 直接对比A和B的检索查询
|
||
- 推理类（intent="reasoning"）：生成最多2个子查询
|
||
* 原问题的检索查询
|
||
* 一个补充角度的检索查询（如原因、背景、影响等），帮助获取更全面的上下文
|
||
- 其他类（factual/instruction/other）：严格只生成1个子查询
|
||
* 子查询应基于 rewritten_query（改写后的完整问题），而非用户原始输入
|
||
* 例如：追问"有完整表格吗？"改写为"吸烟点C1类有完整表格吗？"后，子查询应为"吸烟点C1类的完整表格内容"
|
||
- 不要为同一实体生成语义重叠的查询
|
||
- 子查询应保持原问题的关键词，长度20-60字符为宜
|
||
|
||
6. **intent**: 意图分类（5种类型，请严格按以下定义判断）
|
||
|
||
**"comparison"（对比查询）** —— 明确对比两个或多个事物的差异
|
||
- 关键特征：问题中出现两个以上被比较的对象，且问的是它们之间的差异、异同或优劣
|
||
- 关键词：区别、不同、差异、对比、相比、各自区别、A和B有什么不同、哪个更好
|
||
- ⚠️ 注意：问"如何定义和区分"不是在对比，是在问定义标准 → factual
|
||
- ⚠️ 注意："各自有什么"如果只是分别描述而非对比差异 → factual
|
||
|
||
**"reasoning"（推理查询）** —— 需要因果分析、原因推断或影响评估
|
||
- 关键特征：问题要求分析"为什么"、推导因果关系、或评估某事的影响/后果
|
||
- 关键词：为什么、原因、导致、影响、后果、为何、怎样导致、如何影响、分析原因
|
||
|
||
**"instruction"（操作指导）** —— 询问具体操作步骤、执行方法或实施流程
|
||
- 关键特征：问题要求的是"怎么做某事"——需要具体的步骤、方法或可执行的指导
|
||
- 关键词：怎么做、如何操作、如何实施、步骤是什么、操作流程、配置方法、如何申请、如何办理、升级路径、需要满足哪些条件
|
||
- ⚠️ 注意：问"某机制是怎样的"如果涉及评价标准和执行流程 → instruction
|
||
|
||
**"factual"（事实查询）** —— 查询事实、定义、标准、规定、数据、枚举项
|
||
- 关键特征：问题要求的是文档中明确记载的事实信息
|
||
- 关键词：是什么、有哪些、包括什么、多少、规定、标准、要求、内容、原则、办法、类型、种类、评分标准、占比限制、时限
|
||
- 这是最常见的类型——问"某东西包含哪些内容/原则/办法/标准"都属于事实查询
|
||
|
||
**"other"（其他）** —— 闲聊、问候、不属于以上类型的问题
|
||
|
||
## intent 判断流程（按优先级依次检查）
|
||
|
||
1. 问题是否在对比两个事物的差异？→ comparison
|
||
2. 问题是否在问"为什么"或分析原因/影响？→ reasoning
|
||
3. 问题是否在问具体怎么做、操作步骤、申请流程、升级路径？→ instruction
|
||
4. 以上都不是 → factual（大多数知识库查询属于此类）
|
||
|
||
### 常见易混淆情况：
|
||
- "如何定义和区分" → factual（问的是定义和标准，不是对比差异）
|
||
- "有哪些原则/办法/标准" → factual（枚举事实，不是操作指导）
|
||
- "如何优化/如何实施" → instruction（问的是具体做法和步骤）
|
||
- "评价标准和退出机制是怎样的" → instruction（涉及评价流程和退出流程）
|
||
- "升级路径是什么，需要满足哪些条件" → instruction（涉及升级步骤和条件）
|
||
- "A和B有什么区别/不同" → comparison（明确对比两者差异）
|
||
- "分别包含什么/各自含义" → factual（分别描述，不是对比差异）
|
||
|
||
## 判断原则（两个判断是独立的！）
|
||
|
||
### use_context = true 的情况：
|
||
- 用户引用了历史对话中的内容（"这两张图片"、"上面的数据"、"继续说"）
|
||
- 问题是对历史回答的追问或深入询问
|
||
- 答案可以从历史回答中直接得出
|
||
- **重复提问**：用户再次问相同问题
|
||
|
||
### need_retrieval = true 的情况：
|
||
- 问题询问新的事实、数据、流程
|
||
- 问题提到了具体的图号、表号、章节
|
||
- 问题需要知识库中的文档来回答
|
||
- 用户明确要求查看知识库内容
|
||
|
||
## ⚠️ 重要规则：重复提问必须检索！
|
||
|
||
**当用户重复提问相同或相似问题时：**
|
||
- `use_context` = true（因为确实依赖历史上下文）
|
||
- `need_retrieval` = **true**（必须重新检索！）
|
||
|
||
**原因：**
|
||
1. 用户重复提问说明之前的回答不满意或不正确
|
||
2. 历史回答可能包含错误信息，不能直接复用
|
||
3. 必须重新检索确保回答准确性
|
||
|
||
**判断重复提问的方法：**
|
||
- 当前问题与历史中的用户问题语义相同或高度相似
|
||
- 例如：用户之前问"发电量"，现在又问"发电量统计"
|
||
- 例如：用户之前问过某问题，现在换个方式再问
|
||
|
||
### 两者都可以为 false：
|
||
- 简单的闲聊、问候
|
||
- 基于附件/图片的视觉分析请求（图片信息在上下文中）
|
||
- 常识性问题
|
||
|
||
## 输出格式
|
||
|
||
只输出一个 JSON 对象，不要其他内容：
|
||
```json
|
||
{
|
||
"rewritten_query": "改写后的问题",
|
||
"use_context": true/false,
|
||
"need_retrieval": true/false,
|
||
"reason": "判断理由",
|
||
"sub_queries": ["检索子查询1", "检索子查询2"],
|
||
"intent": "factual"
|
||
}
|
||
```
|
||
|
||
示例：
|
||
- 对比类："新现代终端和普通现代终端有什么区别？" → sub_queries: ["新现代终端的定义和特点", "普通现代终端的定义和特点", "新现代终端与普通现代终端的区别"], intent: "comparison"
|
||
- 推理类："为什么市场状态评估结果与实际投放不一致？" → sub_queries: ["市场状态评估结果与实际投放不一致的原因"], intent: "reasoning"
|
||
- 操作指导："从普通终端升级到加盟终端需要怎么做？" → sub_queries: ["从普通终端升级到加盟终端的流程和条件"], intent: "instruction"
|
||
- 事实类（枚举）："货源投放工作有哪些原则？" → sub_queries: ["货源投放工作的原则"], intent: "factual"
|
||
- 事实类（定义）："市场状态有哪几种类型？" → sub_queries: ["市场状态的类型和评分标准"], intent: "factual"
|
||
"""
|
||
|
||
def __init__(self, model: str = None, use_cache: bool = True):
|
||
"""
|
||
初始化意图分析器
|
||
|
||
Args:
|
||
model: 使用的 LLM 模型（默认从配置读取）
|
||
use_cache: 是否启用语义缓存
|
||
"""
|
||
self.model = model
|
||
self.use_cache = use_cache
|
||
self._client = None
|
||
self._cache = None
|
||
self._embedding_model = None
|
||
# 精确匹配缓存（后备方案，不依赖 embedding）
|
||
self._exact_cache: Dict[str, IntentAnalysis] = {}
|
||
self._exact_cache_max = 500
|
||
|
||
def _get_client(self):
|
||
"""获取 LLM 客户端（百炼快速模型）"""
|
||
if self._client is None:
|
||
from config import get_intent_client
|
||
self._client = get_intent_client()
|
||
return self._client
|
||
|
||
def _get_cache(self):
|
||
"""获取语义缓存"""
|
||
if self._cache is None and self.use_cache:
|
||
try:
|
||
from core.semantic_cache import get_semantic_cache
|
||
self._cache = get_semantic_cache()
|
||
except Exception as e:
|
||
logger.warning(f"语义缓存初始化失败: {e}")
|
||
self._cache = None
|
||
return self._cache
|
||
|
||
def _get_embedding(self, text: str):
|
||
"""获取文本向量，优先复用 engine 已加载的模型"""
|
||
if self._embedding_model is None:
|
||
# 优先从 engine 复用已加载的 embedding 模型
|
||
try:
|
||
from core.engine import get_engine
|
||
engine = get_engine()
|
||
self._embedding_model = engine.embedding_model
|
||
if self._embedding_model is not None:
|
||
logger.info("意图分析缓存: 复用 engine 的 embedding 模型")
|
||
else:
|
||
logger.warning("意图分析缓存: engine.embedding_model 为 None")
|
||
except Exception as e:
|
||
# fallback: 单独加载
|
||
logger.warning(f"意图分析缓存: 无法获取 engine 的 embedding 模型: {e}，尝试单独加载")
|
||
try:
|
||
from sentence_transformers import SentenceTransformer
|
||
self._embedding_model = SentenceTransformer('models/bge-base-zh-v1.5')
|
||
logger.info("意图分析缓存: 单独加载 embedding 模型成功")
|
||
except Exception as e2:
|
||
logger.warning(f"意图分析缓存: 嵌入模型初始化失败: {e2}")
|
||
return None
|
||
|
||
try:
|
||
import numpy as np
|
||
emb = self._embedding_model.encode(text)
|
||
return np.array(emb, dtype='float32')
|
||
except Exception as e:
|
||
logger.warning(f"向量化失败: {e}")
|
||
return None
|
||
|
||
def analyze(
|
||
self,
|
||
query: str,
|
||
history: List[dict],
|
||
context_images: List[dict] = None
|
||
) -> IntentAnalysis:
|
||
"""
|
||
分析用户意图
|
||
|
||
Args:
|
||
query: 用户当前问题
|
||
history: 对话历史 [{"role": "user/assistant", "content": "...", "metadata": {...}}]
|
||
context_images: 上下文中的图片信息
|
||
|
||
Returns:
|
||
IntentAnalysis: 分析结果
|
||
"""
|
||
# 构建历史摘要
|
||
history_summary = self._build_history_summary(history, context_images)
|
||
|
||
# 1. 精确匹配缓存（快速路径，不依赖 embedding）
|
||
exact_key = self._build_cache_key(query, history)
|
||
if exact_key in self._exact_cache:
|
||
logger.info(f"意图分析精确缓存命中: {exact_key[:50]}")
|
||
return self._exact_cache[exact_key]
|
||
|
||
# 2. 尝试从语义缓存获取
|
||
cache = self._get_cache()
|
||
if cache:
|
||
# 使用 query + 历史关键信息作为缓存键
|
||
cache_key = self._build_cache_key(query, history)
|
||
cache_emb = self._get_embedding(cache_key)
|
||
|
||
if cache_emb is not None:
|
||
cached = cache.get(cache_emb)
|
||
# 确保缓存条目是意图分析结果（非 RAG 回答缓存）
|
||
if cached and cached.get("cache_type") != "rag_answer":
|
||
logger.info(f"意图分析缓存命中: {cached.get('reason', '')[:50]}")
|
||
return IntentAnalysis.from_dict(cached)
|
||
else:
|
||
logger.debug(f"意图分析缓存未命中，缓存状态: {cache.get_stats()}")
|
||
else:
|
||
logger.warning("意图分析缓存: _get_embedding 返回 None，跳过缓存")
|
||
else:
|
||
logger.warning("意图分析缓存: _get_cache 返回 None，缓存不可用")
|
||
|
||
# 构建 prompt
|
||
user_prompt = f"""## 对话历史
|
||
|
||
{history_summary}
|
||
|
||
## 当前用户消息
|
||
|
||
{query}
|
||
|
||
请分析用户的意图，输出 JSON 对象。"""
|
||
|
||
try:
|
||
client = self._get_client()
|
||
model = self.model or self._get_default_model()
|
||
|
||
from core.llm_utils import call_llm
|
||
content = call_llm(
|
||
client,
|
||
"",
|
||
model,
|
||
temperature=INTENT_TEMPERATURE,
|
||
max_tokens=INTENT_MAX_TOKENS,
|
||
messages=[
|
||
{"role": "system", "content": self.SYSTEM_PROMPT},
|
||
{"role": "user", "content": user_prompt}
|
||
]
|
||
)
|
||
|
||
# 解析 JSON
|
||
result = self._parse_json(content)
|
||
|
||
if result:
|
||
# 提取子查询，默认使用原始查询
|
||
sub_queries = result.get("sub_queries", [query])
|
||
if not sub_queries:
|
||
sub_queries = [query]
|
||
|
||
# 非对比类查询强制截断子查询（LLM 可能不遵守规则）
|
||
intent_type = result.get("intent", "factual")
|
||
if intent_type == "comparison":
|
||
# 对比类：确保原始查询始终作为第一个子查询（避免拆分后丢失核心信息）
|
||
if query not in sub_queries:
|
||
sub_queries = [query] + sub_queries
|
||
sub_queries = sub_queries[:3] # 最多3个
|
||
elif intent_type == "reasoning":
|
||
sub_queries = sub_queries[:2] # 推理类最多2个子查询
|
||
elif len(sub_queries) > 1:
|
||
sub_queries = sub_queries[:1] # 其他类只保留1个
|
||
|
||
analysis = IntentAnalysis(
|
||
rewritten_query=result.get("rewritten_query", query),
|
||
use_context=result.get("use_context", False),
|
||
need_retrieval=result.get("need_retrieval", True),
|
||
confidence=0.9,
|
||
reason=result.get("reason", ""),
|
||
context_images=context_images or [],
|
||
sub_queries=sub_queries,
|
||
intent=intent_type
|
||
)
|
||
|
||
# 存入语义缓存（标记类型，避免与 RAG 回答缓存混淆）
|
||
if cache and cache_emb is not None:
|
||
cache_data = analysis.to_dict()
|
||
cache_data["cache_type"] = "intent_analysis"
|
||
cache.set(cache_emb, cache_data)
|
||
|
||
# 存入精确匹配缓存
|
||
if len(self._exact_cache) < self._exact_cache_max:
|
||
self._exact_cache[exact_key] = analysis
|
||
|
||
return analysis
|
||
|
||
except Exception as e:
|
||
logger.warning(f"意图分析失败: {e}，使用默认值")
|
||
|
||
# 默认值：需要检索
|
||
return IntentAnalysis(
|
||
rewritten_query=query,
|
||
use_context=False,
|
||
need_retrieval=True,
|
||
confidence=0.5,
|
||
reason="意图分析失败，默认走检索流程",
|
||
context_images=context_images or [],
|
||
sub_queries=[query],
|
||
intent="factual"
|
||
)
|
||
|
||
def _build_cache_key(self, query: str, history: List[dict]) -> str:
|
||
"""
|
||
构建缓存键
|
||
|
||
结合 query 和历史关键信息，确保相同上下文下的相同问题能命中缓存
|
||
"""
|
||
parts = [query]
|
||
|
||
# 添加最近历史的关键信息
|
||
if history:
|
||
for msg in history[-2:]: # 最近1轮
|
||
role = msg.get("role", "")
|
||
content = msg.get("content", "")[:100] # 截断
|
||
parts.append(f"{role[:1]}:{content}")
|
||
|
||
return " | ".join(parts)
|
||
|
||
def _build_history_summary(
|
||
self,
|
||
history: List[dict],
|
||
context_images: List[dict] = None
|
||
) -> str:
|
||
"""
|
||
构建历史摘要
|
||
|
||
Args:
|
||
history: 对话历史
|
||
context_images: 上下文中的图片
|
||
|
||
Returns:
|
||
历史摘要文本
|
||
"""
|
||
if not history:
|
||
return "（无历史对话）"
|
||
|
||
import re
|
||
parts = []
|
||
|
||
# 提取最近 3 轮对话
|
||
recent_history = history[-INTENT_HISTORY_WINDOW:]
|
||
|
||
for msg in recent_history:
|
||
role = "用户" if msg.get("role") == "user" else "助手"
|
||
content = msg.get("content", "")
|
||
original_content = content # 保留原始内容用于结构化提取
|
||
|
||
# 截断过长的内容
|
||
if len(content) > 500:
|
||
content = content[:500] + "..."
|
||
|
||
parts.append(f"【{role}】{content}")
|
||
|
||
# 提取结构化信息（从 assistant 消息中提取章节、表格、来源等）
|
||
metadata = msg.get("metadata", {})
|
||
if isinstance(metadata, dict):
|
||
# 已有的图片提取
|
||
images = metadata.get("images", [])
|
||
if images:
|
||
for img in images[:3]:
|
||
if isinstance(img, dict):
|
||
desc = img.get("description", "")[:100]
|
||
img_type = img.get("type", "图片")
|
||
parts.append(f" └─ {img_type}: {desc}")
|
||
|
||
# 来源文件提取
|
||
sources = metadata.get("sources", [])
|
||
if sources:
|
||
source_names = []
|
||
for s in sources[:3]:
|
||
if isinstance(s, dict):
|
||
name = s.get("source", "") or s.get("name", "")
|
||
if name:
|
||
source_names.append(name)
|
||
elif isinstance(s, str):
|
||
source_names.append(s)
|
||
if source_names:
|
||
parts.append(f" └─ 来源文件: {', '.join(source_names)}")
|
||
|
||
# collections（检索知识库）提取
|
||
colls = metadata.get("collections", [])
|
||
if colls:
|
||
parts.append(f" └─ 检索知识库: {', '.join(colls)}")
|
||
|
||
# 从 assistant 原始内容中提取章节路径和表格结构
|
||
if role == "助手" and original_content:
|
||
# 提取章节路径：━ xxx ━ 格式
|
||
sections = re.findall(r'━\s*(.+?)\s*━', original_content)
|
||
if sections:
|
||
unique_sections = list(dict.fromkeys(sections)) # 去重保序
|
||
parts.append(f" └─ 涉及章节: {'; '.join(unique_sections[:3])}")
|
||
|
||
# 提取表格列名：| A | B | C | 格式的表头行
|
||
table_headers = re.findall(r'^\|\s*(.+?)\s*\|', original_content, re.MULTILINE)
|
||
if table_headers:
|
||
# 取第一个表格的列名
|
||
first_header = table_headers[0]
|
||
cols = [c.strip() for c in first_header.split('|') if c.strip()]
|
||
# 排除分隔符行（--- 格式）
|
||
if cols and not all(re.match(r'^[-:]+$', c) for c in cols):
|
||
parts.append(f" └─ 含表格，列名: {', '.join(cols[:6])}")
|
||
|
||
# 添加图片上下文
|
||
if context_images:
|
||
parts.append("\n【上下文中的图片】")
|
||
for img in context_images[:5]:
|
||
desc = img.get("description", "")[:100]
|
||
img_type = img.get("type", "图片")
|
||
parts.append(f" - {img_type}: {desc}")
|
||
|
||
return "\n".join(parts)
|
||
|
||
def _parse_json(self, content: str) -> Optional[dict]:
|
||
"""解析 JSON 响应"""
|
||
# 尝试直接解析
|
||
try:
|
||
parsed = json.loads(content)
|
||
# mimo 等模型可能返回 JSON 数组而非对象，取第一个元素
|
||
if isinstance(parsed, list) and len(parsed) > 0:
|
||
first = parsed[0]
|
||
if isinstance(first, dict):
|
||
return first
|
||
return None
|
||
if isinstance(parsed, dict):
|
||
return parsed
|
||
return None
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
# 尝试提取 JSON 块
|
||
import re
|
||
json_match = re.search(r'\{[\s\S]*\}', content)
|
||
if json_match:
|
||
try:
|
||
return json.loads(json_match.group())
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
return None
|
||
|
||
def _get_default_model(self) -> str:
|
||
"""获取意图分析专用模型"""
|
||
try:
|
||
from config import INTENT_MODEL
|
||
return INTENT_MODEL
|
||
except (ImportError, NameError):
|
||
return INTENT_MODEL
|
||
|
||
|
||
# ==================== 便捷函数 ====================
|
||
|
||
_analyzer = None
|
||
_analyzer_lock = threading.Lock() # 单例创建锁，防止多线程竞态条件
|
||
|
||
def get_intent_analyzer() -> IntentAnalyzer:
|
||
"""获取意图分析器单例（双重检查锁定，线程安全）"""
|
||
global _analyzer
|
||
if _analyzer is None:
|
||
with _analyzer_lock:
|
||
if _analyzer is None:
|
||
_analyzer = IntentAnalyzer()
|
||
return _analyzer
|
||
|
||
|
||
def analyze_intent(
|
||
query: str,
|
||
history: List[dict],
|
||
context_images: List[dict] = None
|
||
) -> IntentAnalysis:
|
||
"""
|
||
便捷函数：分析用户意图
|
||
|
||
Args:
|
||
query: 用户问题
|
||
history: 对话历史
|
||
context_images: 上下文中的图片
|
||
|
||
Returns:
|
||
IntentAnalysis: 分析结果
|
||
"""
|
||
analyzer = get_intent_analyzer()
|
||
return analyzer.analyze(query, history, context_images)
|
||
|
||
|
||
# ==================== 测试 ====================
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
if sys.platform == 'win32':
|
||
sys.stdout.reconfigure(encoding='utf-8')
|
||
|
||
# 测试用例
|
||
test_cases = [
|
||
{
|
||
"query": "分析一下这两张图片",
|
||
"history": [
|
||
{"role": "user", "content": "三峡工程有什么图片"},
|
||
{"role": "assistant", "content": "我找到了三峡大坝的示意图...", "metadata": {"images": [{"type": "示意图", "description": "三峡大坝全景"}]}},
|
||
],
|
||
"expected": {"use_context": True, "need_retrieval": False}
|
||
},
|
||
{
|
||
"query": "三峡水库补水调度统计",
|
||
"history": [],
|
||
"expected": {"use_context": False, "need_retrieval": True}
|
||
},
|
||
{
|
||
"query": "继续说",
|
||
"history": [
|
||
{"role": "user", "content": "介绍一下三峡工程"},
|
||
{"role": "assistant", "content": "三峡工程是..."},
|
||
],
|
||
"expected": {"use_context": True, "need_retrieval": False}
|
||
},
|
||
]
|
||
|
||
print("=" * 60)
|
||
print("意图分析器测试")
|
||
print("=" * 60)
|
||
|
||
analyzer = IntentAnalyzer()
|
||
|
||
for i, case in enumerate(test_cases, 1):
|
||
print(f"\n--- 测试 {i} ---")
|
||
print(f"问题: {case['query']}")
|
||
|
||
result = analyzer.analyze(case["query"], case["history"])
|
||
|
||
print(f"改写后: {result.rewritten_query}")
|
||
print(f"use_context: {result.use_context}")
|
||
print(f"need_retrieval: {result.need_retrieval}")
|
||
print(f"理由: {result.reason}")
|
||
print(f"子查询: {result.sub_queries}")
|
||
print(f"意图: {result.intent}")
|