""" Agentic RAG - 查询重写 Mixin 包含查询改写、实体补全、专业术语映射等方法 """ import re import logging from .agentic_base import logger, MODEL logger = logging.getLogger(__name__) class QueryRewriteMixin: """查询重写方法""" def _rewrite_query(self, query: str, history: list = None, strategy: str = "professional") -> str: """ 增强版查询重写：将口语化表达转为专业术语 Args: query: 原始查询 history: 对话历史（用于实体补全） strategy: 重写策略 - professional: 口语化→专业术语 - expand: 扩展关键词 - clarify: 消歧义 - entity: 实体补全 Returns: str: 重写后的查询 """ # 尝试多种策略组合 rewritten = query # 策略1: 口语化→专业术语映射 if strategy in ["professional", "all"]: rewritten = self._apply_professional_mapping(rewritten) # 策略2: 实体补全（利用对话历史） if strategy in ["entity", "all"] and history: rewritten = self._complete_entities(rewritten, history) # 策略3: LLM 深度重写（仅在需要时调用） if strategy in ["professional", "all"]: llm_rewritten = self._llm_rewrite(rewritten) if llm_rewritten and len(llm_rewritten) > len(rewritten) * 0.5: rewritten = llm_rewritten return rewritten def _apply_professional_mapping(self, query: str) -> str: """应用口语化→专业术语映射""" TERM_MAPPING = { "报销": "差旅报销 费用报销 报销审批", "请假": "休假申请 请假审批 考勤管理", "加班": "加班申请 工时管理 加班审批", "工资": "薪酬管理 工资发放 薪资结构", "合同": "合同管理 合同签署 合同审批", "流程": "审批流程 业务流程 工作流", "制度": "管理制度 规章制度 企业规范", "规定": "管理规定 制度规定 政策要求", "几天": "时限 审批时限 办理时限", "多久": "处理时效 审批周期 办理周期", "多少": "标准 额度 限额 标准", "能不能": "是否允许 是否可以 权限", "人事": "人力资源 HR 人力部门", "财务": "财务部 财务部门 财务管理", "技术": "技术部 研发部 IT部门", } result = query for colloquial, professional in TERM_MAPPING.items(): if colloquial in query: result = result.replace(colloquial, f"{colloquial} {professional.split()[0]}") return result def _complete_entities(self, query: str, history: list) -> str: """实体补全：利用对话历史补充缺失的实体""" if not history: return query # 图片指代识别 image_reference = self._detect_image_reference(query, history) if image_reference: return image_reference # 获取最近用户消息 last_user_msg = None for msg in reversed(history): if msg.get("role") == "user": last_user_msg = msg.get("content", "") break if not last_user_msg: return query # 检查当前查询是否缺少主语 BUSINESS_KEYWORDS = ["报销", "出差", "请假", "工资", "合同", "审批", "流程", "制度", "规定", "标准", "金额", "时间"] has_subject = any(kw in query for kw in BUSINESS_KEYWORDS) if not has_subject: try: import jieba entities = [] for word in jieba.cut(last_user_msg): word = word.strip() if len(word) >= 2 and any(kw in word for kw in BUSINESS_KEYWORDS): entities.append(word) if entities: return f"{entities[0]} {query}" except ImportError: pass return query def _detect_image_reference(self, query: str, history: list) -> str: """检测图片指代查询并重写""" IMAGE_REFERENCE_PATTERNS = [ r'这[张些]图片', r'那[张些]图片', r'上面的图片', r'刚才的图片', r'这[张些]图', r'那[张些]图', r'上面的图', r'刚才的图', r'解释一下这[张些]图', r'说明一下这[张些]图', r'这[张些]是什么图', r'图[里内]是什么', r'图片[里内]是什么', ] is_image_reference = False for pattern in IMAGE_REFERENCE_PATTERNS: if re.search(pattern, query): is_image_reference = True break if not is_image_reference: return "" last_images = [] for msg in reversed(history): if msg.get("role") == "assistant": metadata = msg.get("metadata", {}) if isinstance(metadata, dict): images = metadata.get("images", []) if images: for img in images[:5]: if isinstance(img, dict): desc = img.get("description", "") img_type = img.get("type", "图片") if desc: last_images.append(f"{img_type}：{desc}") elif isinstance(img, str): last_images.append(f"图片：{img}") if not last_images: content = msg.get("content", "") if "图片" in content or "图表" in content or "图" in content: sentences = content.split("。") for sentence in sentences: if "图片" in sentence or "图表" in sentence: last_images.append(sentence.strip()) if last_images: break if last_images: image_context = " ".join(last_images[:3]) question_intent = re.sub( r'这[张些]图片?|那[张些]图片?|上面的图片?|刚才的图片?|解释一下|说明一下', '', query ).strip() if question_intent: return f"{image_context} {question_intent}" else: return f"详细解释：{image_context}" return query def _extract_image_context_from_history(self, history: list) -> str: """从对话历史中提取图片上下文""" if not history: return "" for msg in reversed(history): if msg.get("role") == "assistant": metadata = msg.get("metadata", {}) images = metadata.get("images", []) content = msg.get("content", "") image_descriptions = [] if images: for i, img in enumerate(images[:5], 1): if isinstance(img, dict): desc = img.get("description", "") img_type = img.get("type", "图片") source = img.get("source", "") page = img.get("page", "") img_info = f"图片{i}：{img_type}" if desc: img_info += f"，描述：{desc}" if source: img_info += f"，来源：{source}" if page: img_info += f"，第{page}页" image_descriptions.append(img_info) if not image_descriptions: if "图片" in content or "图表" in content: sentences = content.split("。") for sentence in sentences: if "图片" in sentence or "图表" in sentence: image_descriptions.append(sentence.strip()) if len(image_descriptions) >= 3: break if image_descriptions: return "\n".join(image_descriptions) return "" def _answer_image_reference(self, enhanced_query: str, history: list) -> str: """回答图片引用问题""" from core.llm_utils import call_llm messages = [ {"role": "system", "content": "你是一个专业的助手，请根据提供的图片信息回答用户的问题。"} ] for h in history[-4:]: if h.get("role") in ["user", "assistant"]: messages.append({"role": h["role"], "content": h.get("content", "")}) messages.append({"role": "user", "content": enhanced_query}) try: result = call_llm( self.client, "", MODEL, temperature=0.3, max_tokens=1000, messages=messages ) return result or "" except Exception as e: logger.error(f"图片引用回答失败: {e}") return f"抱歉，回答图片问题时出现错误：{str(e)}" def _llm_rewrite(self, query: str) -> str: """LLM 深度重写查询""" from core.llm_utils import call_llm prompt = f"""请将以下用户问题改写为更专业、更清晰的表达，保持原意不变。 原问题：{query} 改写后的问题：""" try: rewritten = call_llm( self.client, prompt, MODEL, temperature=0.3, max_tokens=100 ) return rewritten.strip() if rewritten else query except Exception as e: logger.warning(f"LLM 重写失败: {e}") return query