rag_server/applications/utils/chat/rag_chat_agent.py

import asyncio
from typing import List

from applications.config import Chunk
from applications.api import fetch_deepseek_completion


class RAGChatAgent:
    @staticmethod
    def generate_summary_prompt(query, search_results):
        """
        生成总结的prompt。交给AI根据搜索结果判断内容是否对回答问题有帮助，
        并结合内容生成总结和判断是否能回答问题。

        :param query: 问题
        :param search_results: 搜索结果列表，每个元素包含 'content', 'contentSummary'
        :return: 生成的总结prompt
        """

        prompt = f"问题: {query}\n\n请结合以下搜索结果，生成一个总结：\n"

        weighted_summaries = []
        weighted_contents = []

        # 将所有搜索结果的摘要和内容按相似度排序
        for result in search_results:
            content = result["content"]
            content_summary = result["contentSummary"]

            weighted_summaries.append(content_summary)
            weighted_contents.append(content)

        # 拼接加权摘要和内容
        prompt += "\n-- 内容摘要 --\n"
        for summary in weighted_summaries:
            prompt += f"摘要: {summary}\n"

        prompt += "\n-- 内容 --\n"
        for content in weighted_contents:
            prompt += f"内容: {content}\n"

        # 约束 AI 输出 JSON
        prompt += """
    请根据上述内容判断能否回答问题，并生成一个总结，返回 JSON 格式，结构如下：

    {
      "query": "<原始问题>",
      "summary": "<简洁总结>",
      "relevance_score": <0到1之间的小数，表示总结与问题的相关度>,
      "status": <判断能否回答这个问题，0代表不能回答，1代表可以回答>
    }

    注意：
    - 只输出 JSON，不要额外解释。
    - relevance_score 数字越大，表示总结和问题越相关。
    - 请根据问题和给定的搜索结果内容，判断是否能回答该问题。返回一个 0 或 1 的 status，表示能否回答问题。
    """

        return prompt

    async def chat_with_deepseek(self, query, search_results):
        prompt = self.generate_summary_prompt(query, search_results)
        response = await fetch_deepseek_completion(
            model="DeepSeek-V3", prompt=prompt, output_type="json"
        )
        return response

    @staticmethod
    def create_query_prompt(question: str) -> str:
        """
        封装生成查询的 prompt，用于发送给 AI 模型。

        参数:
        - question: 需要查询的问题（字符串）

        返回:
        - 返回一个格式化的 prompt 字符串，用于向 AI 提问
        """
        prompt = f"""
    你是一个智能助手，能够帮助我查询任何问题并返回答案。你的任务是接收到我给定的问题后，通过网络查询相关的信息，并以 JSON 格式返回查询结果。

    问题：{question}

    请查询并返回该问题的答案。返回的 JSON 应该包含以下字段：
    - "question": 问题的原始文本。
    - "answer": 问题的答案内容。
    - "source": 如果有来源，提供信息来源链接或描述。
    - "status": 表示查询是否成功，"1代表成功，0代表失败"。

    如果无法找到答案，返回 "status": "failure"，并且 "answer" 字段应为 "No answer found"。

    例子：
    {{
        "question": "什么是量子计算？",
        "answer": "量子计算是一种基于量子力学原理的计算模型。",
        "source": "https://example.com/quantum-computing",
        "status": 1
    }}
    """
        return prompt

    async def llm_search(self, query):
        prompt = self.create_query_prompt(query)
        response = await fetch_deepseek_completion(
            model="DeepSeek-V3", prompt=prompt, output_type="json"
        )
        return response

    @staticmethod
    def make_decision_prompt(chat_res, search_res):
        # 创建一个 prompt 请求给大模型
        prompt = f"""
        这是一个问题的回答任务，以下是两个来源的结果：

        1. **RAG 搜索回答**:
        问题: {chat_res["query"]}
        总结: {chat_res["summary"]}
        相关度评分: {chat_res["relevance_score"]}
        状态: {"可以回答" if chat_res["status"] == 1 else "无法回答"}

        2. **AI 搜索结果**:
        问题: {search_res["question"]}
        答案: {search_res["answer"]}
        来源: {search_res["source"]}
        状态: {"可以回答" if search_res["status"] == 1 else "无法回答"}
        
        基于这两个结果，请你综合判断并生成一个更好的答案，如果可能的话。你可以选择结合 `chat_res` 和 `search_res`，或者选择其中更合适的一个进行回答。如果没有足够的信息可以回答，请用你自己已有的知识回答"。
        基于回答的结果，总结回答的答案中使用的工具，名称以及用途，如果没有涉及到工具的使用，则不需要总结

        请返回以下格式的 JSON 结果：
        {{
            "result": "<最终的答案，输出markdown格式>",
            "relevance_score": <0到1之间的小数，表示总结与问题的相关度>,
            "status": <1代表回答的好，0代表回答的不好>,
            "used_tools": ["工具名1: 用途描述", "工具名2: 用途描述"]
        }}
        """

        return prompt

    async def make_decision(self, chat_res, search_res):
        prompt = self.make_decision_prompt(chat_res, search_res)
        response = await fetch_deepseek_completion(
            model="DeepSeek-R1", prompt=prompt, output_type="json"
        )
        return response

    @staticmethod
    def split_query_prompt(query):
        prompt = f"""
        你是一个信息检索助理，负责把用户的问题拆解为“更宽泛但仍同类、且彼此不重叠”的子问题，用于召回多样证据。

        【目标】
        - 生成 1–3 个“更宽泛”的子问题（broad questions），它们应与原问题同一类别/主题，但从不同角度扩展；避免把原问题切得更细（avoid over-specific）。
        - 子问题之间尽量覆盖不同维度（例如：背景/原理、影响/应用、比较/趋势、方法/评估 等），减少语义重叠（≤20% 相似度）。

        【必须遵守】
        1) 与原问题同类：如果原问题是技术/科普/对比/流程类，子问题也应保持同类语气与目标。
        2) 更宽泛：去掉过细的限制（具体数值/版本/人名/时间点/实现细节），但保留主题核心。
        3) 去重与互补：不要改写成近义句；每个子问题关注的面不同（角度、层面或受众不同）。
        4) 可检索性：避免抽象空话；每个子问题都应是可用于检索/召回的良好查询。
        5) 数量自适应：若无法合理扩展到 3 个，就输出 1–2 个；不要为了凑数而重复。
        6) 语言一致：与原问题同语言输出（中文入→中文出；英文入→英文出）。
        7) 仅输出 JSON，严格符合下述 schema；不要输出额外文字或注释。

        原始问题：
        {query}

        请按照以下JSON格式返回结果：
        {{
            "original_question": "原始问题",
            "split_questions": [
                "第一个宽泛问题", "第二个宽泛问题", "第三个宽泛问题"
            ]
        }}

        请确保返回的内容是纯JSON格式，不要包含其他任何文字。
        """
        return prompt

    async def split_query(self, query):
        prompt = self.split_query_prompt(query)
        response = await fetch_deepseek_completion(
            model="DeepSeek-V3", prompt=prompt, output_type="json"
        )
        return response