踩了无数坑后，我终于搞定了RAG系统的"胡说八道"问题

def create_optimized_prompt(question, context, task_type="default"):    """    创建优化后的提示词模板        Args:        question: 用户问题        context: 检索到的上下文        task_type: 任务类型，用于选择不同模板    """        templates = {        "credit_card_application": """根据以下银行政策文档，详细回答用户关于信用卡申请的问题。严格要求：1. 必须包含完整的申请步骤（按1. 2. 3.格式编号）2. 必须列出所有必需材料3. 如果文档中信息不完整，明确标注"需致电客服确认：400-xxx-xxxx"4. 不要添加文档中没有的任何信息输出格式：## 申请步骤1. [具体步骤1]2. [具体步骤2]...## 必需材料  - [材料1]- [材料2]...用户问题：{question}银行政策文档：{context}""",                "fee_inquiry": """根据以下费用标准文档，用表格形式回答用户的费用咨询。输出要求：- 必须使用markdown表格格式- 必须包含所有卡种的费用信息- 如果某项费用文档中未明确，标注"以实际为准"| 卡片类型 | 年费标准 | 免年费条件 ||---------|---------|-----------|| 普通卡   | X元     | 具体条件   || 金卡     | X元     | 具体条件   |用户问题：{question}费用文档：{context}""",                "default": """根据以下文档内容，准确回答用户问题。要求：1. 答案必须基于提供的文档内容2. 保持信息的完整性和准确性  3. 如果文档信息不足，说明"文档中未提及此信息"4. 使用清晰的格式组织答案用户问题：{question}参考文档：{context}"""    }        template = templates.get(task_type, templates["default"])    return template.format(question=question, context=context)# 使用示例question = "申请你们银行的信用卡需要什么条件？"context = "检索到的相关文档内容..."prompt = create_optimized_prompt(question, context, "credit_card_application")

import reimport jsonfrom typing import Dict, List, Tupleclass RAGResponseValidator:    """RAG响应验证器"""        def __init__(self):        self.rules = self._load_validation_rules()        def _load_validation_rules(self) -> Dict:        """加载验证规则配置"""        return {            "credit_card": {                "required_entities": ["申请", "材料", "步骤"],                "required_sections": ["申请步骤", "必需材料"],                "forbidden_terms": ["100%通过", "无条件批准", "秒批"],                "format_check": "numbered_list",                "max_length": 800            },            "fee_inquiry": {                "required_entities": ["年费", "卡种"],                "format_check": "table",                "forbidden_terms": ["终身免费", "永久优惠"],                "required_columns": ["卡片类型", "年费标准"]            },            "loan": {                "required_entities": ["利率", "期限", "额度"],                "forbidden_terms": ["保证放款", "无需审核", "当天到账"],                "format_check": "structured"            }        }        def validate_response(self, response: str, context: str,                          rule_type: str = "default") -> Tuple[bool, str]:        """        验证生成的响应                Args:            response: 生成的回答            context: 原始上下文            rule_type: 验证规则类型                    Returns:            (是否通过验证, 错误信息)        """                if rule_type not in self.rules:            return True, "无匹配规则，跳过验证"                rule = self.rules[rule_type]                # 1. 检查必需实体        if not self._check_required_entities(response, rule.get("required_entities", [])):            return False, f"缺少必需信息：{rule['required_entities']}"                # 2. 检查格式要求        if not self._check_format(response, rule.get("format_check")):            format_type = rule.get("format_check")            return False, f"格式不符合要求，需要{format_type}格式"                # 3. 检查禁用词汇        forbidden = self._check_forbidden_terms(response, rule.get("forbidden_terms", []))        if forbidden:            return False, f"包含禁用词汇：{forbidden}"                # 4. 检查幻觉内容        if not self._check_hallucination(response, context):            return False, "包含上下文中不存在的信息"                # 5. 检查长度限制        max_len = rule.get("max_length")        if max_len and len(response) > max_len:            return False, f"回答过长，超过{max_len}字符限制"                return True, "验证通过"        def _check_required_entities(self, response: str, entities: List[str]) -> bool:        """检查必需实体是否存在"""        for entity in entities:            if entity not in response:                return False        return True        def _check_format(self, response: str, format_type: str) -> bool:        """检查格式要求"""        if not format_type:            return True                    if format_type == "table":            return "|" in response and "---" in response        elif format_type == "numbered_list":            return bool(re.search(r'\d+\.\s', response))        elif format_type == "json":            try:                json.loads(response)                return True            except:                return False                return True        def _check_forbidden_terms(self, response: str, forbidden: List[str]) -> str:        """检查禁用词汇"""        for term in forbidden:            if term in response:                return term        return ""        def _check_hallucination(self, response: str, context: str) -> bool:        """简单的幻觉检测"""        # 这里可以根据具体业务场景扩展更复杂的检测逻辑        sensitive_terms = ["优惠活动", "限时", "特殊政策"]        for term in sensitive_terms:            if term in response and term not in context:                return False        return True# 使用示例validator = RAGResponseValidator()def generate_with_validation(prompt: str, context: str, rule_type: str = "default",                            max_attempts: int = 3):    """    带验证的生成函数    """        for attempt in range(max_attempts):        # 这里调用你的LLM生成函数        response = your_llm_generate_function(prompt)  # 替换为实际的LLM调用                # 验证响应        is_valid, error_msg = validator.validate_response(response, context, rule_type)                if is_valid:            print(f"生成成功，尝试次数：{attempt + 1}")            return response        else:            print(f"第{attempt + 1}次验证失败：{error_msg}")            # 根据错误类型调整提示词            prompt += f"\n\n[系统修正要求]：{error_msg}，请重新生成。"        return "抱歉，暂时无法生成满足要求的回答，请联系人工客服。"

class FoRAGGenerator:    """两阶段生成器"""        def __init__(self, llm_function):        self.llm_generate = llm_function        self.validator = RAGResponseValidator()        def generate_outline(self, question: str, context: str, topic_type: str) -> str:        """第一阶段：生成结构化大纲"""                outline_templates = {            "credit_card": """根据文档内容，为信用卡相关问题生成回答大纲：要求大纲包含：1. 主要概念定义（如果涉及）2. 具体步骤或流程（编号列出）  3. 所需材料清单4. 注意事项或特殊说明用户问题：{question}参考文档：{context}请只输出大纲结构，不要展开详细内容：""",                        "fee_inquiry": """根据文档内容，为费用咨询生成表格大纲：表格结构要求：- 列1：产品/服务类型- 列2：费用标准  - 列3：优惠条件（如有）- 列4：备注说明用户问题：{question}参考文档：{context}请输出markdown表格的标题行和分隔行：""",                        "default": """根据文档为用户问题生成结构化回答大纲：1. 核心问题回答2. 详细说明3. 相关注意事项4. 补充信息（如需要）用户问题：{question}参考文档：{context}请只列出大纲要点："""        }                template = outline_templates.get(topic_type, outline_templates["default"])        prompt = template.format(question=question, context=context)                return self.llm_generate(prompt)        def expand_content(self, outline: str, question: str, context: str,                       topic_type: str) -> str:        """第二阶段：基于大纲扩展详细内容"""                expand_prompt = f"""基于以下大纲结构，扩展生成完整的回答内容：大纲：{outline}扩展要求：1. 严格按照大纲结构展开2. 每个要点都要有具体内容，不能空泛3. 所有信息必须来源于提供的文档4. 保持内容简洁，每个部分控制在100字以内5. 使用清晰的格式（如编号、表格等）原始问题：{question}参考文档：{context}请生成最终回答："""                return self.llm_generate(expand_prompt)        def generate_complete_response(self, question: str, context: str,                                  topic_type: str = "default") -> str:        """完整的两阶段生成流程"""                print("第一阶段：生成大纲...")        outline = self.generate_outline(question, context, topic_type)        print(f"大纲生成完成：\n{outline}\n")                print("第二阶段：扩展内容...")        response = self.expand_content(outline, question, context, topic_type)                # 验证最终结果        is_valid, error_msg = self.validator.validate_response(response, context, topic_type)                if not is_valid:            print(f"验证失败：{error_msg}")            # 可以选择重新生成或返回错误信息            return f"生成的回答不符合要求：{error_msg}"                return response# 使用示例def your_llm_generate_function(prompt):    """    这里替换为你实际使用的LLM调用函数    比如调用OpenAI API、本地模型等    """    # 示例代码，实际使用时替换    # response = openai.ChatCompletion.create(...)    # return response.choices[0].message.content    passgenerator = FoRAGGenerator(your_llm_generate_function)# 实际使用question = "申请你们银行信用卡需要准备什么材料？"context = "从知识库检索到的相关文档内容..."response = generator.generate_complete_response(question, context, "credit_card")print("最终回答：", response)

# 完整的RAG优化系统示例class OptimizedRAGSystem:    def __init__(self, llm_function, embedding_function, vector_db):        self.llm_generate = llm_function        self.embed = embedding_function        self.vector_db = vector_db        self.generator = FoRAGGenerator(llm_function)        self.validator = RAGResponseValidator()        def classify_question(self, question: str) -> str:        """问题分类，决定使用哪种处理策略"""                keywords_map = {            "credit_card": ["信用卡", "申请", "办卡", "开卡"],            "fee_inquiry": ["年费", "手续费", "费用", "收费标准"],            "loan": ["贷款", "借款", "贷", "额度"],            "deposit": ["存款", "定期", "活期", "利息"]        }                question_lower = question.lower()        for category, keywords in keywords_map.items():            if any(keyword in question_lower for keyword in keywords):                return category                return "default"        def retrieve_context(self, question: str, top_k: int = 5) -> str:        """检索相关文档"""        # 这里是文档检索逻辑，根据你的实际情况调整        query_embedding = self.embed(question)        docs = self.vector_db.similarity_search(query_embedding, k=top_k)        return "\n".join([doc.content for doc in docs])        def process_query(self, question: str) -> Dict:        """处理用户查询的完整流程"""                print(f"处理问题：{question}")                # 1. 问题分类        question_type = self.classify_question(question)        print(f"问题类型：{question_type}")                # 2. 检索相关文档        context = self.retrieve_context(question)        print(f"检索到{len(context)}字符的相关内容")                # 3. 使用两阶段生成        try:            response = self.generator.generate_complete_response(                question, context, question_type            )                        return {                "success": True,                "answer": response,                "question_type": question_type,                "context_length": len(context)            }                    except Exception as e:            print(f"生成过程出错：{str(e)}")            return {                "success": False,                "error": str(e),                "fallback": "抱歉，系统暂时无法回答您的问题，请联系人工客服。"            }# 使用示例def demo_llm_function(prompt):    """演示用的LLM函数，实际使用时替换为真实的API调用"""    print(f"调用LLM生成，prompt长度：{len(prompt)}")    # 这里应该是实际的模型调用    return "模拟生成的回答内容"def demo_embedding_function(text):    """演示用的嵌入函数"""    return [0.1, 0.2, 0.3]  # 实际应该返回真实的向量class MockVectorDB:    """模拟向量数据库"""    def similarity_search(self, embedding, k=5):        # 模拟返回检索结果        class MockDoc:            def __init__(self, content):                self.content = content                return [            MockDoc("信用卡申请需要提供身份证、收入证明等材料"),            MockDoc("普通卡年费200元，金卡年费600元"),        ]# 初始化系统rag_system = OptimizedRAGSystem(    llm_function=demo_llm_function,    embedding_function=demo_embedding_function,    vector_db=MockVectorDB())# 测试result = rag_system.process_query("申请你们的信用卡需要什么条件？")print("处理结果：", result)

# 先替换现有的提示词模板prompt = create_optimized_prompt(question, context, question_type)

# 添加基本验证逻辑is_valid, error = validator.validate_response(response, context, "basic")

# 选择性使用两阶段生成if question_type in ["credit_card", "fee_inquiry"]:    response = generator.generate_complete_response(...)

class RAGMonitor:    def __init__(self):        self.metrics = {            "validation_pass_rate": [],            "generation_attempts": [],            "response_length": [],            "user_satisfaction": []        }        def log_generation(self, attempts, passed_validation, response_length):        """记录生成指标"""        self.metrics["generation_attempts"].append(attempts)        self.metrics["validation_pass_rate"].append(1 if passed_validation else 0)        self.metrics["response_length"].append(response_length)        def get_daily_report(self):        """生成日报"""        return {            "avg_attempts": sum(self.metrics["generation_attempts"]) / len(self.metrics["generation_attempts"]),            "pass_rate": sum(self.metrics["validation_pass_rate"]) / len(self.metrics["validation_pass_rate"]),            "avg_length": sum(self.metrics["response_length"]) / len(self.metrics["response_length"])        }monitor = RAGMonitor()