kg_agent/knowledge_v2/multi_search_knowledge.py

'''
多渠道获取知识，当前有两个渠道 llm_search_knowledge.py 和 xhs_search_knowledge.py
1. 输入：问题
2. 判断选择哪些渠道获取知识，目录默认返回 llm_search 和 xhs_search 两个渠道
3. 根据选择的结果调用对应的渠道获取知识
4. 合并多个渠道返回知识文本，返回知识文本，使用大模型合并，prompt在 prompt/multi_search_merge_knowledge_prompt.md 中
补充：暂时将xhs_search_knowledge渠道的调用注释掉，后续完成xhs_search_knowledge的实现
'''

import os
import sys
import json
from typing import List, Dict
from loguru import logger
import re

# 设置路径以便导入工具类
current_dir = os.path.dirname(os.path.abspath(__file__))
root_dir = os.path.dirname(current_dir)
sys.path.insert(0, root_dir)

from utils.gemini_client import generate_text
from knowledge_v2.llm_search_knowledge import get_knowledge as get_llm_knowledge
from knowledge_v2.cache_manager import CacheManager
# from knowledge_v2.xhs_search_knowledge import get_knowledge as get_xhs_knowledge

class MultiSearchKnowledge:
    """多渠道知识获取类"""
    
    def __init__(self, use_cache: bool = True):
        """
        初始化
        
        Args:
            use_cache: 是否启用缓存，默认启用
        """
        logger.info("=" * 60)
        logger.info("初始化 MultiSearchKnowledge")
        self.prompt_dir = os.path.join(current_dir, "prompt")
        self.use_cache = use_cache
        self.cache = CacheManager() if use_cache else None
        

        
        logger.info(f"缓存状态: {'启用' if use_cache else '禁用'}")
        logger.info("=" * 60)
        
    def _load_prompt(self, filename: str) -> str:
        """
        加载prompt文件内容
        
        Args:
            filename: prompt文件名
            
        Returns:
            str: prompt内容
        """
        prompt_path = os.path.join(self.prompt_dir, filename)
        
        if not os.path.exists(prompt_path):
            error_msg = f"Prompt文件不存在: {prompt_path}"
            logger.error(error_msg)
            raise FileNotFoundError(error_msg)
        
        try:
            with open(prompt_path, 'r', encoding='utf-8') as f:
                content = f.read().strip()
                if not content:
                    error_msg = f"Prompt文件内容为空: {prompt_path}"
                    logger.error(error_msg)
                    raise ValueError(error_msg)
                return content
        except Exception as e:
            logger.error(f"读取prompt文件 {filename} 失败: {e}")
            raise

    def merge_knowledge(self, question: str, knowledge_map: Dict[str, str]) -> str:
        """
        合并多个渠道的知识文本
        
        Args:
            question: 用户问题  
            knowledge_map: 渠道名到知识文本的映射
            
        Returns:
            str: 合并后的知识文本
        """
        logger.info(f"[Multi-Search] 合并多渠道知识 - {len(knowledge_map)} 个渠道")
        
        # 尝试从缓存读取
        if self.use_cache:
            cached_data = self.cache.get(question, 'multi_search', 'merged_knowledge_detail.json')
            if cached_data:
                # Support reading from detail json
                merged_text = cached_data.get('response', '') or cached_data.get('merged_text', '')
                logger.info(f"✓ 使用缓存的合并知识 (长度: {len(merged_text)})")
                return merged_text
            
            # Legacy txt file fallback
            cached_merged = self.cache.get(question, 'multi_search', 'merged_knowledge.txt')
            if cached_merged:
                logger.info(f"✓ 使用缓存的合并知识 (长度: {len(cached_merged)})")
                return cached_merged
        
        try:
            # 过滤空文本
            valid_knowledge = {k: v for k, v in knowledge_map.items() if v and v.strip()}
            logger.info(f"  有效渠道: {list(valid_knowledge.keys())}")
            
            if not valid_knowledge:
                logger.warning("  ⚠ 所有渠道的知识文本都为空")
                return ""
            
            # 如果只有一个渠道有内容，也经过LLM整理以保证输出风格一致
            
            # 加载prompt
            prompt_template = self._load_prompt("multi_search_merge_knowledge_prompt.md")
            
            # 构建知识文本部分
            knowledge_texts_str = ""
            for source, text in valid_knowledge.items():
                knowledge_texts_str += f"【来源：{source}】\n{text}\n\n"
            
            # 填充prompt
            prompt = prompt_template.format(question=question, knowledge_texts=knowledge_texts_str)
            
            # 调用大模型
            logger.info("  → 调用Gemini合并多渠道知识...")
            merged_text = generate_text(prompt=prompt)
            
            logger.info(f"✓ 多渠道知识合并完成 (长度: {len(merged_text)})")
            
            # 写入缓存
            if self.use_cache:
                self.cache.set(question, 'multi_search', 'merged_knowledge.txt', merged_text.strip())
                merge_data = {
                    "prompt": prompt,
                    "response": merged_text,
                    "sources_count": len(knowledge_map),
                    "valid_sources_count": len(valid_knowledge)
                }
                self.cache.set(question, 'multi_search', 'merged_knowledge_detail.json', merge_data)
            
            return merged_text.strip()
            
        except Exception as e:
            logger.error(f"✗ 合并知识失败: {e}")
            raise

    def extract_and_validate_json(self, text: str):
        """
        从字符串中提取 JSON 部分，并返回标准的 JSON 字符串。
        如果无法提取或解析失败，返回 None (或者你可以改为抛出异常)。
        """
        # 1. 使用正则表达式寻找最大的 JSON 块
        # r"(\{[\s\S]*\}|\[[\s\S]*\])" 的含义：
        # - \{[\s\S]*\} : 匹配以 { 开头，} 结尾的最长字符串（[\s\S] 包含换行符）
        # - | : 或者
        # - \[[\s\S]*\] : 匹配以 [ 开头，] 结尾的最长字符串（处理 JSON 数组）
        match = re.search(r"(\{[\s\S]*\}|\[[\s\S]*\])", text)

        if match:
            json_str = match.group(0)
            try:
                # 2. 尝试解析提取出的字符串，验证是否为合法 JSON
                parsed_json = json.loads(json_str)

                # 3. 重新转储为标准字符串 (去除原本可能存在的缩进、多余空格等)
                # ensure_ascii=False 保证中文不会变成 \uXXXX
                return json.dumps(parsed_json, ensure_ascii=False)

            except json.JSONDecodeError as e:
                print(f"提取到了类似JSON的片段，但解析失败: {e}")
                return None
        else:
            print("未在文本中发现 JSON 结构")
            return None

    def filter_tools(self, knowledge: str, question: str, actual_cache_key: str) -> str:
        """
        筛选出有用的工具

        Args:
            knowledge: 合并后的知识文本
            question: 用户问题

        Returns:
            str: 筛选后的工具文本
        """
        logger.info(f"[Multi-Search] 筛选工具 - 输入长度: {len(knowledge)}")

        cached_data = self.cache.get(actual_cache_key, 'multi_search', 'match_tools.json')
        if cached_data:
            # Support reading from detail json
            return cached_data

        con_prompt_template = self._load_prompt("function_knowledge_result_extract_tool_prompt.md")
        # 填充prompt
        con_prompt = con_prompt_template.replace("{query}", question).replace("{search_result}", knowledge)

        # 调用大模型
        logger.info("  → 调用Gemini筛选工具...")
        con_response = generate_text(prompt=con_prompt)

        logger.info(f"✓ 工具筛选完成 (长度: {len(con_response)})")

        match_prompt_template = self._load_prompt("function_knowledge_match_new_tool_prompt.md")
        # 填充prompt
        match_prompt = match_prompt_template.replace("{input_data}", con_response)

        # 调用大模型
        logger.info("  → 调用Gemini筛选工具...")
        match_response = generate_text(prompt=match_prompt)

        match_data = {
            "extract_tool_prompt": con_prompt,
            "extract_tool_response": json.loads(self.extract_and_validate_json(con_response)),
            "match_tool_prompt": match_prompt,
            "match_tool_response": json.loads(self.extract_and_validate_json(match_response))
        }

        self.cache.set(actual_cache_key, 'multi_search', 'match_tools.json', match_data)

        return match_response.strip()



    def get_knowledge(self, question: str, cache_key: str = None) -> str:
        """
        获取知识的主方法
        
        Args:
            question: 问题字符串
            cache_key: 可选的缓存键，用于与主流程共享同一缓存目录
            
        Returns:
            str: 最终的知识文本
        """
        #使用cache_key或question作为缓存键
        actual_cache_key = cache_key if cache_key is not None else question
        
        import time
        start_time = time.time()
        
        logger.info(f"{'='*60}")
        logger.info(f"Multi-Search - 开始处理问题: {question[:50]}...")
        logger.info(f"{'='*60}")
        
        knowledge_map = {}
        
        # 1. 获取 LLM Search 知识
        try:
            logger.info("[渠道1] 调用 LLM Search...")
            llm_knowledge = get_llm_knowledge(question, cache_key=actual_cache_key, need_generate_query = False)
            knowledge_map["LLM Search"] = llm_knowledge
            logger.info(f"✓ LLM Search 完成 (长度: {len(llm_knowledge)})")
            logger.info(f"✓ LLM Search 完成 (长度: {len(llm_knowledge)})")
        except Exception as e:
            logger.error(f"✗ LLM Search 失败: {e}")
            knowledge_map["LLM Search"] = ""
            
        # 2. 获取 XHS Search 知识 (暂时注释)
        # try:
        #     logger.info("[渠道2] 调用 XHS Search...")
        #     xhs_knowledge = get_xhs_knowledge(question)
        #     knowledge_map["XHS Search"] = xhs_knowledge
        # except Exception as e:
        #     logger.error(f"✗ XHS Search 失败: {e}")
        #     knowledge_map["XHS Search"] = ""
        
        # 3. 合并知识
        final_knowledge = self.merge_knowledge(actual_cache_key, knowledge_map)

        # 4. 筛选工具
        filter_tools_result = self.filter_tools(final_knowledge, question, actual_cache_key)
        
        logger.info(f"{'='*60}")
        logger.info(f"✓ Multi-Search 完成 (最终长度: {len(final_knowledge)})")
        logger.info(f"{'='*60}\n")
        
        # 计算执行时间并保存详情
        execution_time = time.time() - start_time
        
        return final_knowledge

def get_knowledge(question: str, cache_key: str = None) -> str:
    """
    便捷调用函数
    
    Args:
        question: 问题
        cache_key: 可选的缓存键
    """
    agent = MultiSearchKnowledge()
    return agent.get_knowledge(question, cache_key=cache_key)

if __name__ == "__main__":
    # 测试代码
    test_question = "如何评价最近的国产3A游戏黑神话悟空？"
    try:
        result = get_knowledge(test_question)
        print("=" * 50)
        print("最终整合知识:")
        print("=" * 50)
        print(result)
    except Exception as e:
        logger.error(f"测试失败: {e}")