rag_server/applications/utils/chat/rag_chat_agent.py

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 search_with_deepseek(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 select_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": "<最终的答案>",
            "relevance_score": <0到1之间的小数，表示总结与问题的相关度>,
            "status": <1代表回答的好，0代表回答的不好>
        }}
        """

        return prompt

    async def select_with_deepseek(self, chat_res, search_res):
        prompt = self.select_prompt(chat_res, search_res)
        response = await fetch_deepseek_completion(
            model="DeepSeek-V3", prompt=prompt, output_type="json"
        )
        return response