RAG 生成集成与系统整合

AI 摘要

文章详解了食谱RAG系统的两大核心文件：main.py负责系统初始化、模块协调与用户交互；generation_integration.py实现查询路由、重写、LLM调用及三种回答模式（列表、基础、详细步骤）。整体采用分层架构，结合混合检索、父子文档、流式输出与元数据过滤，实现精准、高效、用户友好的问答体验。

此摘要根据文章内容自动生成，仅用于文章内容的解释与总结

🎯 整体架构概览

┌─────────────────────────────────────────────────────────────┐
│                     main.py (核心编排层)                      │
│                   RecipeRAGSystem 类                         │
│  - 系统初始化                                                 │
│  - 模块协调                                                   │
│  - 用户交互                                                   │
└────────┬────────────────────────────────────────────────────┘
         │ 调用
         │
┌────────▼────────────────────────────────────────────────────┐
│           generation_integration.py (生成层)                 │
│            GenerationIntegrationModule 类                    │
│  - 查询路由                                                   │
│  - 查询重写                                                   │
│  - LLM 调用                                                   │
│  - Prompt 工程                                               │
└─────────────────────────────────────────────────────────────┘

📁 文件 1：generation_integration.py - 生成集成模块

🏗️ 类结构

class GenerationIntegrationModule:
    def __init__(self, model_name, temperature, max_tokens)
    
    # 核心方法
    ├── setup_llm()                          # LLM 初始化
    ├── query_router()                       # 查询路由（分类）
    ├── query_rewrite()                      # 查询重写（优化）
    ├── generate_basic_answer()              # 基础回答
    ├── generate_step_by_step_answer()       # 详细步骤回答
    ├── generate_list_answer()               # 列表式回答
    ├── generate_basic_answer_stream()       # 流式基础回答
    ├── generate_step_by_step_answer_stream() # 流式详细回答
    └── _build_context()                     # 上下文构建（辅助）

① setup_llm() - LLM 初始化

def setup_llm(self):
    api_key = os.getenv("MOONSHOT_API_KEY")
    
    self.llm = MoonshotChat(
        model="kimi-k2-0711-preview",  # Kimi 模型
        temperature=0.1,                # 低温度：输出更稳定
        max_tokens=2048,                # 最大输出长度
        moonshot_api_key=api_key
    )

Kimi 模型特点：

• 国内大模型，API 稳定
• 支持长上下文（200K tokens）
• 中文理解能力强
• 适合食谱场景

Temperature 参数：

temperature = 0.1  # 低温度模式
# 优点：输出稳定、可预测
# 缺点：创造性不足
# 适用：食谱制作（需要准确、标准的步骤）

temperature = 0.7  # 高温度模式
# 优点：输出多样、有创意
# 缺点：可能不稳定
# 适用：文学创作、头脑风暴

② query_router() - 查询路由（智能分类）

功能：根据用户意图将查询分为 3 类

def query_router(self, query: str) -> str:
    prompt = ChatPromptTemplate.from_template("""
根据用户的问题，将其分类为以下三种类型之一：

1. 'list' - 用户想要获取菜品列表或推荐，只需要菜名
   例如：推荐几个素菜、有什么川菜、给我3个简单的菜

2. 'detail' - 用户想要具体的制作方法或详细信息
   例如：宫保鸡丁怎么做、制作步骤、需要什么食材

3. 'general' - 其他一般性问题
   例如：什么是川菜、制作技巧、营养价值

用户问题: {query}

分类结果:""")
    
    chain = (
        {"query": RunnablePassthrough()}
        | prompt
        | self.llm
        | StrOutputParser()
    )
    
    result = chain.invoke(query).strip().lower()
    return result if result in ['list', 'detail', 'general'] else 'general'

LCEL (LangChain Expression Language) 解析：

chain = (
    {"query": RunnablePassthrough()}  # 1. 输入：原始查询
    | prompt                           # 2. 格式化：注入 prompt 模板
    | self.llm                         # 3. 调用：LLM 推理
    | StrOutputParser()                # 4. 解析：提取字符串结果
)

路由决策示例：

查询示例                    → 路由类型
"推荐几个简单的菜"          → list     (需要列表)
"宫保鸡丁怎么做"            → detail   (需要详细步骤)
"川菜的特点是什么"          → general  (一般性问题)
"有什么素菜"                → list
"红烧肉需要什么食材"        → detail

③ query_rewrite() - 查询重写（智能优化）

功能：将模糊查询重写为更精确的检索查询

def query_rewrite(self, query: str) -> str:
    prompt = PromptTemplate(template="""
你是一个智能查询分析助手。请分析用户的查询，判断是否需要重写以提高食谱搜索效果。

原始查询: {query}

分析规则：
1. **具体明确的查询**（直接返回原查询）：
   - 包含具体菜品名称：如"宫保鸡丁怎么做"、"红烧肉的制作方法"
   
2. **模糊不清的查询**（需要重写）：
   - 过于宽泛：如"做菜"、"有什么好吃的"、"推荐个菜"
   - 缺乏具体信息：如"川菜"、"素菜"、"简单的"

重写示例：
- "做菜" → "简单易做的家常菜谱"
- "有饮品推荐吗" → "简单饮品制作方法"
- "宫保鸡丁怎么做" → "宫保鸡丁怎么做"（保持原查询）

请输出最终查询:""", input_variables=["query"])
    
    chain = (
        {"query": RunnablePassthrough()}
        | prompt
        | self.llm
        | StrOutputParser()
    )
    
    response = chain.invoke(query).strip()
    
    if response != query:
        logger.info(f"查询已重写: '{query}' → '{response}'")
    
    return response

重写示例对比：

原始查询              → 重写后查询                → 检索效果
"做菜"               → "简单易做的家常菜谱"      → ✓ 增加上下文
"川菜"               → "经典川菜菜谱"            → ✓ 更具体
"有什么素菜"         → "简单素菜推荐"            → ✓ 明确意图
"宫保鸡丁怎么做"     → "宫保鸡丁怎么做"          → ✓ 已经明确，保持原样

为什么需要查询重写？

# 问题：用户查询往往不够精确
用户输入: "做菜"

# 向量检索会困惑：
embeddings.embed_query("做菜")
# 向量表示模糊，可能匹配到：
# - "做菜技巧"
# - "做菜工具"
# - "做菜视频"（无关内容）

# 重写后：
embeddings.embed_query("简单易做的家常菜谱")
# 向量表示清晰，精准匹配：
# - "番茄炒蛋" ✓
# - "清炒时蔬" ✓
# - "红烧肉"   ✓

④ 生成方法（3种模式）

模式 1：列表式回答 (generate_list_answer)

def generate_list_answer(self, query: str, context_docs: List[Document]) -> str:
    # 提取菜品名称
    dish_names = []
    for doc in context_docs:
        dish_name = doc.metadata.get('dish_name', '未知菜品')
        if dish_name not in dish_names:
            dish_names.append(dish_name)
    
    # 构建简洁列表
    if len(dish_names) == 1:
        return f"为您推荐：{dish_names[0]}"
    elif len(dish_names) <= 3:
        return f"为您推荐以下菜品：\n" + "\n".join([f"{i+1}. {name}" for i, name in enumerate(dish_names)])
    else:
        return f"为您推荐以下菜品：\n" + "\n".join([f"{i+1}. {name}" for i, name in enumerate(dish_names[:3])]) + f"\n\n还有其他 {len(dish_names)-3} 道菜品可供选择。"

特点：

• 不调用 LLM（节省成本和时间）
• 直接从元数据提取菜名
• 简洁明了

输出示例：

为您推荐以下菜品：
1. 番茄炒蛋
2. 清炒时蔬
3. 凉拌黄瓜

模式 2：基础回答 (generate_basic_answer)

def generate_basic_answer(self, query: str, context_docs: List[Document]) -> str:
    context = self._build_context(context_docs)
    
    prompt = ChatPromptTemplate.from_template("""
你是一位专业的烹饪助手。请根据以下食谱信息回答用户的问题。

用户问题: {question}

相关食谱信息:
{context}

请提供详细、实用的回答。如果信息不足，请诚实说明。

回答:""")
    
    chain = (
        {"question": RunnablePassthrough(), "context": lambda _: context}
        | prompt
        | self.llm
        | StrOutputParser()
    )
    
    return chain.invoke(query)

特点：

• 简单直接的 QA 模式
• 适合一般性问题
• 灵活性高

模式 3：详细步骤回答 (generate_step_by_step_answer)

def generate_step_by_step_answer(self, query: str, context_docs: List[Document]) -> str:
    context = self._build_context(context_docs)
    
    prompt = ChatPromptTemplate.from_template("""
你是一位专业的烹饪导师。请根据食谱信息，为用户提供详细的分步骤指导。

用户问题: {question}

相关食谱信息:
{context}

请灵活组织回答，建议包含以下部分（可根据实际内容调整）：

## 🥘 菜品介绍
[简要介绍菜品特点和难度]

## 🛒 所需食材
[列出主要食材和用量]

## 👨‍🍳 制作步骤
[详细的分步骤说明，每步包含具体操作和大概所需时间]

## 💡 制作技巧
[仅在有实用技巧时包含。优先使用原文中的实用技巧，如果原文的"附加内容"与烹饪无关或为空，可以基于制作步骤总结关键要点，或者完全省略此部分]

注意：
- 根据实际内容灵活调整结构
- 不要强行填充无关内容或重复制作步骤中的信息
- 重点突出实用性和可操作性

回答:""")
    
    chain = (
        {"question": RunnablePassthrough(), "context": lambda _: context}
        | prompt
        | self.llm
        | StrOutputParser()
    )
    
    return chain.invoke(query)

特点：

• 结构化输出
• 详细的步骤指导
• 适合制作类查询

输出示例：

## 🥘 菜品介绍
番茄炒蛋是一道经典家常菜，难度：非常简单

## 🛒 所需食材
- 番茄 2个（约200g）
- 鸡蛋 3个
- 食盐 适量
- 白糖 少许

## 👨‍🍳 制作步骤
1. 准备工作（5分钟）
   - 番茄洗净切块
   - 鸡蛋打散加少许盐

2. 炒鸡蛋（2分钟）
   - 热锅冷油，倒入蛋液
   - 快速翻炒至凝固

...

## 💡 制作技巧
- 先炒鸡蛋再炒番茄，口感更好
- 加少许白糖可以提鲜

流式输出版本

def generate_step_by_step_answer_stream(self, query: str, context_docs: List[Document]):
    # ... 相同的 prompt 和 chain 构建 ...
    
    for chunk in chain.stream(query):  # stream() 而非 invoke()
        yield chunk

流式 vs 非流式对比：

# 非流式：等待完整输出
answer = module.generate_step_by_step_answer(query, docs)
print(answer)  # 等待 5-10 秒后一次性输出

# 流式：逐字输出
for chunk in module.generate_step_by_step_answer_stream(query, docs):
    print(chunk, end="", flush=True)  # 实时显示，类似 ChatGPT 体验

⑤ _build_context() - 上下文构建

def _build_context(self, docs: List[Document], max_length: int = 2000) -> str:
    context_parts = []
    current_length = 0
    
    for i, doc in enumerate(docs, 1):
        # 构建元数据信息
        metadata_info = f"【食谱 {i}】"
        if 'dish_name' in doc.metadata:
            metadata_info += f" {doc.metadata['dish_name']}"
        if 'category' in doc.metadata:
            metadata_info += f" | 分类: {doc.metadata['category']}"
        if 'difficulty' in doc.metadata:
            metadata_info += f" | 难度: {doc.metadata['difficulty']}"
        
        doc_text = f"{metadata_info}\n{doc.page_content}\n"
        
        # 长度限制（防止超出 LLM 上下文窗口）
        if current_length + len(doc_text) > max_length:
            break
        
        context_parts.append(doc_text)
        current_length += len(doc_text)
    
    return "\n" + "="*50 + "\n".join(context_parts)

上下文格式示例：

==================================================
【食谱 1】 番茄炒蛋 | 分类: 家常菜 | 难度: 非常简单
# 番茄炒蛋
## 必备原料
番茄 2个，鸡蛋 3个...
## 操作步骤
1. 准备工作...
==================================================
【食谱 2】 红烧肉 | 分类: 荤菜 | 难度: 中等
# 红烧肉
## 必备原料
五花肉 500g...

为什么限制长度？

# 问题：LLM 有上下文窗口限制
Kimi-k2: 200K tokens 输入 + 4K tokens 输出

# 如果上下文过长：
context_length = 150K tokens
# 后果：
# 1. API 调用变慢
# 2. 成本大幅增加
# 3. 可能触发限制

# 解决：限制为 2000 字符（约 1000 tokens）
max_length = 2000
# 好处：
# 1. 响应快速
# 2. 成本可控
# 3. 信息足够（3-5个完整食谱）

📁 文件 2：main.py - 系统主程序

🏗️ 类结构

class RecipeRAGSystem:
    def __init__(self, config)
    
    # 系统管理
    ├── initialize_system()           # 初始化所有模块
    ├── build_knowledge_base()        # 构建知识库
    
    # 核心查询
    ├── ask_question()                # 主查询入口
    ├── _extract_filters_from_query() # 提取过滤条件
    
    # 辅助功能
    ├── search_by_category()          # 分类搜索
    ├── get_ingredients_list()        # 获取食材
    └── run_interactive()             # 交互式运行

🔄 完整执行流程

阶段 1：系统启动

def main():
    # 1. 创建 RAG 系统
    rag_system = RecipeRAGSystem()
    
    # 2. 运行交互式问答
    rag_system.run_interactive()

阶段 2：系统初始化 (initialize_system)

def initialize_system(self):
    print("🚀 正在初始化RAG系统...")
    
    # 1. 数据准备模块
    self.data_module = DataPreparationModule(self.config.data_path)
    
    # 2. 索引构建模块
    self.index_module = IndexConstructionModule(
        model_name=self.config.embedding_model,
        index_save_path=self.config.index_save_path
    )
    
    # 3. 生成集成模块
    self.generation_module = GenerationIntegrationModule(
        model_name=self.config.llm_model,
        temperature=self.config.temperature,
        max_tokens=self.config.max_tokens
    )
    
    print("✅ 系统初始化完成！")

模块依赖关系：

RecipeRAGSystem
    ├── data_module (DataPreparationModule)
    │   └── 负责：文档加载、分块、父子关系
    │
    ├── index_module (IndexConstructionModule)
    │   └── 负责：向量化、FAISS 索引
    │
    ├── retrieval_module (RetrievalOptimizationModule)
    │   └── 负责：混合检索、RRF 重排
    │
    └── generation_module (GenerationIntegrationModule)
        └── 负责：LLM 调用、回答生成

阶段 3：知识库构建 (build_knowledge_base)

def build_knowledge_base(self):
    print("\n正在构建知识库...")
    
    # 1. 尝试加载已保存的索引
    vectorstore = self.index_module.load_index()
    
    if vectorstore is not None:
        print("✅ 成功加载已保存的向量索引！")
        # 仍需加载文档（用于父子检索）
        self.data_module.load_documents()
        chunks = self.data_module.chunk_documents()
    else:
        print("未找到已保存的索引，开始构建新索引...")
        
        # 2. 加载文档
        self.data_module.load_documents()
        
        # 3. 文本分块
        chunks = self.data_module.chunk_documents()
        
        # 4. 构建向量索引
        vectorstore = self.index_module.build_vector_index(chunks)
        
        # 5. 保存索引
        self.index_module.save_index()
    
    # 6. 初始化检索优化模块
    self.retrieval_module = RetrievalOptimizationModule(vectorstore, chunks)
    
    # 7. 显示统计信息
    stats = self.data_module.get_statistics()
    print(f"\n📊 知识库统计:")
    print(f"   文档总数: {stats['total_documents']}")
    print(f"   文本块数: {stats['total_chunks']}")
    
    print("✅ 知识库构建完成！")

冷启动 vs 热启动：

# 首次运行（冷启动）：
load_index() → None
    ↓
load_documents() → 读取 .md 文件
    ↓
chunk_documents() → Markdown 分块
    ↓
build_vector_index() → 向量化（耗时 2-3 分钟）
    ↓
save_index() → 保存到磁盘
    ↓
总耗时：3-5 分钟

# 后续运行（热启动）：
load_index() → 加载现有索引
    ↓
load_documents() → 读取 .md 文件（仅用于父子检索）
    ↓
chunk_documents() → Markdown 分块
    ↓
总耗时：5-10 秒

阶段 4：处理用户查询 (ask_question)

这是最核心的方法，展示了整个 RAG 流程：

def ask_question(self, question: str, stream: bool = False):
    print(f"\n❓ 用户问题: {question}")
    
    # ========== 步骤 1：查询路由 ==========
    route_type = self.generation_module.query_router(question)
    print(f"🎯 查询类型: {route_type}")
    # 输出：'list' | 'detail' | 'general'
    
    
    # ========== 步骤 2：智能查询重写 ==========
    if route_type == 'list':
        # 列表查询保持原样
        rewritten_query = question
        print(f"📝 列表查询保持原样: {question}")
    else:
        # 详细查询和一般查询使用智能重写
        print("🤖 智能分析查询...")
        rewritten_query = self.generation_module.query_rewrite(question)
    
    
    # ========== 步骤 3：检索相关文档 ==========
    print("🔍 检索相关文档...")
    
    # 3.1 提取元数据过滤条件
    filters = self._extract_filters_from_query(question)
    
    # 3.2 执行检索（带过滤或混合检索）
    if filters:
        print(f"应用过滤条件: {filters}")
        relevant_chunks = self.retrieval_module.metadata_filtered_search(
            rewritten_query, filters, top_k=self.config.top_k
        )
    else:
        relevant_chunks = self.retrieval_module.hybrid_search(
            rewritten_query, top_k=self.config.top_k
        )
    
    # 3.3 显示检索到的子块信息
    if relevant_chunks:
        chunk_info = []
        for chunk in relevant_chunks:
            dish_name = chunk.metadata.get('dish_name', '未知菜品')
            content_preview = chunk.page_content[:50].replace('\n', ' ').strip()
            # ... 提取章节信息 ...
            chunk_info.append(f"{dish_name}(内容片段)")
        print(f"找到 {len(relevant_chunks)} 个相关文档块: {', '.join(chunk_info)}")
    
    
    # ========== 步骤 4：检查结果 ==========
    if not relevant_chunks:
        return "抱歉，没有找到相关的食谱信息。请尝试其他菜品名称或关键词。"
    
    
    # ========== 步骤 5：根据路由类型生成回答 ==========
    if route_type == 'list':
        # 列表查询：直接返回菜品名称列表
        print("📋 生成菜品列表...")
        relevant_docs = self.data_module.get_parent_documents(relevant_chunks)
        return self.generation_module.generate_list_answer(question, relevant_docs)
    
    else:
        # 详细查询：获取完整文档并生成详细回答
        print("获取完整文档...")
        relevant_docs = self.data_module.get_parent_documents(relevant_chunks)
        
        print("✍️ 生成详细回答...")
        
        if route_type == "detail":
            # 详细查询使用分步指导模式
            if stream:
                return self.generation_module.generate_step_by_step_answer_stream(question, relevant_docs)
            else:
                return self.generation_module.generate_step_by_step_answer(question, relevant_docs)
        else:
            # 一般查询使用基础回答模式
            if stream:
                return self.generation_module.generate_basic_answer_stream(question, relevant_docs)
            else:
                return self.generation_module.generate_basic_answer(question, relevant_docs)

📊 完整查询流程图（带实际数据流）

用户输入: "推荐几个简单的素菜"
        │
        ├─────────────────────────────────────┐
        │  步骤1：查询路由                      │
        │  generation_module.query_router()    │
        │  └─> LLM 分析意图                    │
        │      输出: route_type = "list"       │
        └─────────────────────────────────────┘
        │
        ├─────────────────────────────────────┐
        │  步骤2：查询重写                      │
        │  if route_type == 'list':            │
        │      保持原查询                       │
        │  else:                               │
        │      generation_module.query_rewrite()│
        │  输出: "推荐几个简单的素菜"（保持）   │
        └─────────────────────────────────────┘
        │
        ├─────────────────────────────────────┐
        │  步骤3：提取过滤条件                  │
        │  _extract_filters_from_query()       │
        │  └─> 检测关键词：素菜、简单           │
        │  输出: filters = {                   │
        │      'category': '素菜',              │
        │      'difficulty': '简单'             │
        │  }                                   │
        └─────────────────────────────────────┘
        │
        ├─────────────────────────────────────┐
        │  步骤4：混合检索（带过滤）             │
        │  retrieval_module.metadata_filtered_search()│
        │      │                                │
        │      ├─> hybrid_search()              │
        │      │   ├─> 向量检索 Top-9           │
        │      │   │   [番茄炒蛋, 清炒时蔬, ...] │
        │      │   └─> BM25检索 Top-9           │
        │      │       [凉拌黄瓜, 素炒豆芽, ...]  │
        │      │                                │
        │      └─> RRF融合                     │
        │          [番茄炒蛋, 清炒时蔬, 凉拌黄瓜, ...]│
        │      │                                │
        │      └─> 元数据过滤                   │
        │          过滤掉不满足条件的文档         │
        │          [清炒时蔬, 凉拌黄瓜, 素炒豆芽]  │
        │                                       │
        │  输出: relevant_chunks (3个子块)      │
        └─────────────────────────────────────┘
        │
        ├─────────────────────────────────────┐
        │  步骤5：获取完整父文档                 │
        │  data_module.get_parent_documents()  │
        │  └─> 根据 parent_id 找完整食谱       │
        │  输出: relevant_docs (3个完整食谱)    │
        └─────────────────────────────────────┘
        │
        ├─────────────────────────────────────┐
        │  步骤6：根据路由生成回答               │
        │  if route_type == 'list':            │
        │      generate_list_answer()          │
        │      └─> 不调用LLM，直接提取菜名      │
        │                                       │
        │  输出:                                │
        │  为您推荐以下菜品：                    │
        │  1. 清炒时蔬                          │
        │  2. 凉拌黄瓜                          │
        │  3. 素炒豆芽                          │
        └─────────────────────────────────────┘

🎯 不同查询类型的处理示例

示例 1：列表查询

用户: "推荐几个简单的素菜"

执行流程:
1. query_router() → "list"
2. query_rewrite() → 跳过（列表查询保持原样）
3. _extract_filters() → {'category': '素菜', 'difficulty': '简单'}
4. metadata_filtered_search() → [doc1, doc2, doc3]
5. get_parent_documents() → [完整食谱1, 完整食谱2, 完整食谱3]
6. generate_list_answer() → "为您推荐：1. 清炒时蔬 2. 凉拌黄瓜 3. 素炒豆芽"

特点：
- 不调用 LLM 生成（快速）
- 直接提取菜名
- 简洁明了

示例 2：详细查询

用户: "番茄炒蛋怎么做？"

执行流程:
1. query_router() → "detail"
2. query_rewrite() → "番茄炒蛋制作方法步骤"（可能重写）
3. _extract_filters() → {}（无特定过滤条件）
4. hybrid_search() → [番茄炒蛋-原料块, 番茄炒蛋-步骤块, ...]
5. get_parent_documents() → [番茄炒蛋完整食谱]
6. generate_step_by_step_answer() → 调用 LLM 生成详细步骤

输出:
## 🥘 菜品介绍
番茄炒蛋是一道经典家常菜...

## 🛒 所需食材
- 番茄 2个
- 鸡蛋 3个
...

## 👨‍🍳 制作步骤
1. 准备工作...
2. 炒鸡蛋...
...

特点：
- 调用 LLM 生成（详细）
- 结构化输出
- 分步骤指导

示例 3：一般查询

用户: "什么是川菜？"

执行流程:
1. query_router() → "general"
2. query_rewrite() → "川菜的特点和代表菜品"（可能重写）
3. _extract_filters() → {'category': '川菜'}（如果有）
4. hybrid_search() → [相关文档]
5. get_parent_documents() → [相关完整食谱]
6. generate_basic_answer() → 调用 LLM 生成一般性回答

输出:
川菜是中国四大菜系之一，以麻辣鲜香著称。代表菜品包括...

特点：
- 调用 LLM 生成
- 灵活回答
- 适应性强

🔧 辅助方法详解

_extract_filters_from_query() - 提取元数据过滤条件

def _extract_filters_from_query(self, query: str) -> dict:
    filters = {}
    
    # 提取分类
    category_keywords = DataPreparationModule.get_supported_categories()
    # ['荤菜', '素菜', '汤品', '甜品', '早餐', '主食', '水产', '调料', '饮品']
    
    for cat in category_keywords:
        if cat in query:
            filters['category'] = cat
            break
    
    # 提取难度
    difficulty_keywords = DataPreparationModule.get_supported_difficulties()
    # ['非常简单', '简单', '中等', '困难', '非常困难']
    
    for diff in sorted(difficulty_keywords, key=len, reverse=True):
        if diff in query:
            filters['difficulty'] = diff
            break
    
    return filters

关键词匹配示例：

查询                          → 提取的过滤条件
"推荐几个简单的素菜"          → {'category': '素菜', 'difficulty': '简单'}
"有什么荤菜"                  → {'category': '荤菜'}
"非常简单的菜"                → {'difficulty': '非常简单'}
"宫保鸡丁怎么做"              → {}（无特定过滤条件）

为什么按长度排序？

# 问题：关键词可能重叠
difficulty_keywords = ['非常简单', '简单', '中等', '困难', '非常困难']

查询 = "非常简单的菜"

# 错误顺序（短到长）：
for diff in difficulty_keywords:
    if diff in query:  # 首先匹配到 "简单"
        filters['difficulty'] = '简单'  # ✗ 错误！应该是"非常简单"
        break

# 正确顺序（长到短）：
for diff in sorted(difficulty_keywords, key=len, reverse=True):
    if diff in query:  # 首先匹配到 "非常简单"
        filters['difficulty'] = '非常简单'  # ✓ 正确！
        break

🎓 完整交互流程总结

用户交互示例

========================================
🍽️  尝尝咸淡RAG系统 - 交互式问答  🍽️
========================================
💡 解决您的选择困难症，告别'今天吃什么'的世纪难题！

🚀 正在初始化RAG系统...
初始化数据准备模块...
初始化索引构建模块...
🤖 初始化生成集成模块...
✅ 系统初始化完成！

正在构建知识库...
✅ 成功加载已保存的向量索引！
加载食谱文档...
进行文本分块...
初始化检索优化...

📊 知识库统计:
   文档总数: 50
   文本块数: 150
   菜品分类: ['荤菜', '素菜', '汤品', '甜品']
   难度分布: {'简单': 25, '中等': 15, '困难': 10}

✅ 知识库构建完成！

交互式问答 (输入'退出'结束):

您的问题: 推荐几个简单的素菜

是否使用流式输出? (y/n, 默认y): n

❓ 用户问题: 推荐几个简单的素菜
🎯 查询类型: list
📝 列表查询保持原样: 推荐几个简单的素菜
🔍 检索相关文档...
应用过滤条件: {'category': '素菜', 'difficulty': '简单'}
找到 3 个相关文档块: 清炒时蔬(内容片段), 凉拌黄瓜(内容片段), 素炒豆芽(内容片段)
📋 生成菜品列表...
找到文档: 清炒时蔬, 凉拌黄瓜, 素炒豆芽

回答:
为您推荐以下菜品：
1. 清炒时蔬
2. 凉拌黄瓜
3. 素炒豆芽


您的问题: 番茄炒蛋怎么做？

是否使用流式输出? (y/n, 默认y): y

❓ 用户问题: 番茄炒蛋怎么做？
🎯 查询类型: detail
🤖 智能分析查询...
查询已重写: '番茄炒蛋怎么做？' → '番茄炒蛋制作方法详细步骤'
🔍 检索相关文档...
找到 5 个相关文档块: 番茄炒蛋(必备原料), 番茄炒蛋(操作步骤), ...
获取完整文档...
找到文档: 番茄炒蛋
✍️ 生成详细回答...

回答:
## 🥘 菜品介绍
番茄炒蛋是一道经典的家常菜...（流式输出，逐字显示）

## 🛒 所需食材
- 番茄 2个...

## 👨‍🍳 制作步骤
1. 准备工作（5分钟）...

## 💡 制作技巧
- 先炒鸡蛋再炒番茄...


您的问题: 退出

感谢使用尝尝咸淡RAG系统！

🎨 架构设计亮点

1. 分层设计

┌──────────────────────────────────────┐
│  应用层 (main.py)                     │
│  - 用户交互                           │
│  - 流程编排                           │
└─────────────┬────────────────────────┘
              │
┌─────────────▼────────────────────────┐
│  生成层 (generation_integration.py)  │
│  - 查询路由                           │
│  - 查询重写                           │
│  - LLM 调用                          │
└─────────────┬────────────────────────┘
              │
┌─────────────▼────────────────────────┐
│  检索层 (retrieval_optimization.py)  │
│  - 混合检索                           │
│  - RRF 重排                          │
└─────────────┬────────────────────────┘
              │
┌─────────────▼────────────────────────┐
│  索引层 (index_construction.py)      │
│  - 向量检索                           │
│  - FAISS 管理                        │
└─────────────┬────────────────────────┘
              │
┌─────────────▼────────────────────────┐
│  数据层 (data_preparation.py)        │
│  - 文档加载                           │
│  - 父子关系                           │
└──────────────────────────────────────┘

2. 智能路由机制

query_router() → route_type
    │
    ├─> "list"    → generate_list_answer()      (快速)
    ├─> "detail"  → generate_step_by_step...()  (详细)
    └─> "general" → generate_basic_answer()     (灵活)

3. 查询优化管道

原始查询
    │
    ├─> query_router()   (分类意图)
    ├─> query_rewrite()  (优化表达)
    └─> extract_filters() (提取条件)
    │
    ▼
优化后的检索请求

4. 父子文档策略

检索阶段：使用子块（小块，精准）
    │
    ▼
get_parent_documents()
    │
    ▼
生成阶段：使用父文档（大块，完整）

5. 流式输出支持

# 用户体验优化
if stream:
    for chunk in generate_xxx_stream():
        print(chunk, end="")  # 实时显示
else:
    answer = generate_xxx()
    print(answer)  # 一次性显示

🎓 总结

这两个文件构成了一个完整的生产级 RAG 系统：

main.py 的职责：

1. 系统编排：初始化所有模块，协调执行流程
2. 用户交互：提供友好的交互界面
3. 流程控制：根据查询类型选择合适的处理策略
4. 结果整合：将各模块的输出组合成最终答案

generation_integration.py 的职责：

1. 查询理解：路由分类、查询重写
2. Prompt 工程：设计不同场景的提示词
3. LLM 调用：封装大模型 API
4. 格式化输出：生成结构化、用户友好的回答

核心优势：

• ✅ 智能路由：根据意图选择最合适的回答策略
• ✅ 查询优化：自动重写模糊查询，提高检索精度
• ✅ 混合检索：融合向量和关键词，提高召回率
• ✅ 父子文档：平衡检索精度和上下文完整性
• ✅ 流式输出：提升用户体验
• ✅ 元数据过滤：支持精准的结构化筛选

这是一个模块化、可扩展、生产就绪的 RAG 系统实现！