add react

config.py

@@ -51,8 +51,8 @@ LOCAL_LLM = "mistral"  # 在Kaggle中可改为 "phi" 或 "tinyllama"
 # 知识库URL配置
 KNOWLEDGE_BASE_URLS = [
     "https://lilianweng.github.io/posts/2023-06-23-agent/",
-    "https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
-    "https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",
+    # "https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
+    # "https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",
 ]
 
 # 文档分块配置

main.py

@@ -3,13 +3,20 @@
 集成所有模块，构建工作流并运行自适应RAG系统
 """
 
+import time
 from langgraph.graph import END, StateGraph, START
 from pprint import pprint
 
-from config import setup_environment, validate_api_keys
+from config import setup_environment, validate_api_keys, ENABLE_GRAPHRAG
 from document_processor import initialize_document_processor
 from routers_and_graders import initialize_graders_and_router
 from workflow_nodes import WorkflowNodes, GraphState
+try:
+    from knowledge_graph import initialize_knowledge_graph, initialize_community_summarizer
+    from graph_retriever import initialize_graph_retriever
+except ImportError:
+    print("⚠️ 无法导入知识图谱模块，GraphRAG功能将不可用")
+    ENABLE_GRAPHRAG = False
 
 
 class AdaptiveRAGSystem:
@@ -54,6 +61,23 @@ class AdaptiveRAGSystem:
         print("初始化评分器和路由器...")
         self.graders = initialize_graders_and_router()
         
+        # 初始化知识图谱 (如果启用)
+        self.graph_retriever = None
+        if ENABLE_GRAPHRAG:
+            print("初始化 GraphRAG...")
+            try:
+                kg = initialize_knowledge_graph()
+                # 尝试加载已有的图谱数据
+                try:
+                    kg.load_from_file("knowledge_graph.json")
+                except FileNotFoundError:
+                    print("   未找到 existing knowledge_graph.json, 将使用空图谱")
+                
+                self.graph_retriever = initialize_graph_retriever(kg)
+                print("✅ GraphRAG 初始化成功")
+            except Exception as e:
+                print(f"⚠️ GraphRAG 初始化失败: {e}")
+        
         # 初始化工作流节点
         print("设置工作流节点...")
         # WorkflowNodes 将在 _build_workflow 中初始化
@@ -91,6 +115,8 @@ class AdaptiveRAGSystem:
         workflow.add_node("grade_documents", self.workflow_nodes.grade_documents)
         workflow.add_node("generate", self.workflow_nodes.generate)
         workflow.add_node("transform_query", self.workflow_nodes.transform_query)
+        workflow.add_node("decompose_query", self.workflow_nodes.decompose_query)
+        workflow.add_node("prepare_next_query", self.workflow_nodes.prepare_next_query)
         
         # 构建图
         workflow.add_conditional_edges(
@@ -98,20 +124,23 @@ class AdaptiveRAGSystem:
             self.workflow_nodes.route_question,
             {
                 "web_search": "web_search",
-                "vectorstore": "retrieve",
+                "vectorstore": "decompose_query", # 向量检索前先进行查询分解
             },
         )
         workflow.add_edge("web_search", "generate")
+        workflow.add_edge("decompose_query", "retrieve")
         workflow.add_edge("retrieve", "grade_documents")
         workflow.add_conditional_edges(
             "grade_documents",
             self.workflow_nodes.decide_to_generate,
             {
                 "transform_query": "transform_query",
+                "prepare_next_query": "prepare_next_query",
                 "generate": "generate",
             },
         )
         workflow.add_edge("transform_query", "retrieve")
+        workflow.add_edge("prepare_next_query", "retrieve")
         workflow.add_conditional_edges(
             "generate",
             self.workflow_nodes.grade_generation_v_documents_and_question,
@@ -151,10 +180,16 @@ class AdaptiveRAGSystem:
         # 设置配置，增加递归限制
         config = {"recursion_limit": 50}  # 增加到 50，默认是 25
         
+        print("\n🤖 思考过程:")
         for output in self.app.stream(inputs, config=config):
             for key, value in output.items():
                 if verbose:
-                    pprint(f"节点 '{key}':")
+                    # 简单的节点执行提示，模拟流式感
+                    print(f"  ↳ 执行节点: {key}...", end="\r")
+                    time.sleep(0.1) # 视觉暂停
+                    print(f"  ✅ 完成节点: {key}      ")
+                    
+                    # pprint(f"节点 '{key}':")
                     # 可选：在每个节点打印完整状态
                     # pprint(value, indent=2, width=80, depth=None)
                 final_generation = value.get("generation", final_generation)
@@ -162,11 +197,25 @@ class AdaptiveRAGSystem:
                 if "retrieval_metrics" in value:
                     retrieval_metrics = value["retrieval_metrics"]
             if verbose:
-                pprint("\n---\n")
+                # pprint("\n---\n")
+                pass
         
+        print("\n" + "=" * 50)
         print("🎯 最终答案:")
         print("-" * 30)
-        print(final_generation)
+        
+        # 模拟流式输出效果 (打字机效果)
+        if final_generation:
+            import sys
+            import time
+            for char in final_generation:
+                sys.stdout.write(char)
+                sys.stdout.flush()
+                time.sleep(0.01) # 控制打字速度
+            print() # 换行
+        else:
+            print("未生成答案")
+            
         print("=" * 50)
         
         # 返回包含答案和评估指标的字典
@@ -220,7 +269,8 @@ def main():
         rag_system: AdaptiveRAGSystem = AdaptiveRAGSystem()
         
         # 测试查询
-        test_question = "AlphaCodium论文讲的是什么？"
+        # test_question = "AlphaCodium论文讲的是什么？"
+        test_question = "LangGraph的作者目前在哪家公司工作？"
         # test_question = "解释embedding嵌入的原理，最好列举实现过程的具体步骤"
         result = rag_system.query(test_question)
         

routers_and_graders.py

@@ -3,6 +3,7 @@
 包含查询路由、文档相关性评分、答案质量评分和幻觉检测
 """
 
+from typing import List
 try:
     from langchain_core.prompts import PromptTemplate
 except ImportError:
@@ -152,23 +153,105 @@ class HallucinationGrader:
             return result.get("score", "no")
 
 
+class QueryDecomposer:
+    """查询分解器，将复杂的多跳问题分解为子问题序列"""
+    
+    def __init__(self):
+        self.llm = ChatOllama(model=LOCAL_LLM, format="json", temperature=0)
+        self.prompt = PromptTemplate(
+            template="""你是一个查询分解专家。你的任务是将一个复杂的多跳问题分解为一系列简单的子问题，这些子问题可以按顺序检索来回答原始问题。
+            
+            分解规则：
+            1. 如果问题很简单，不需要分解，返回只包含原始问题的列表。
+            2. 如果问题需要多步推理（例如"A的作者的大学在哪里"），分解为逻辑步骤：
+               - 步骤1: "谁是A的作者？"
+               - 步骤2: "该作者在哪个大学？"
+            3. 保持子问题简洁明了。
+            4. 即使返回单个问题，也必须包装在JSON的 sub_queries 列表中。
+            
+            输出格式：返回一个包含 'sub_queries' 键的 JSON，其值为字符串列表。
+            不要输出任何前言或解释。
+            
+            复杂问题: {question}""",
+            input_variables=["question"],
+        )
+        self.decomposer = self.prompt | self.llm | JsonOutputParser()
+    
+    def decompose(self, question: str) -> List[str]:
+        """分解问题"""
+        print(f"---分解问题: {question}---")
+        try:
+            result = self.decomposer.invoke({"question": question})
+            sub_queries = result.get("sub_queries", [question])
+            # 确保至少包含原始问题
+            if not sub_queries:
+                sub_queries = [question]
+            print(f"---子问题: {sub_queries}---")
+            return sub_queries
+        except Exception as e:
+            print(f"⚠️ 分解失败: {e}，使用原始问题")
+            return [question]
+
+
+class AnswerabilityGrader:
+    """答案可回答性评分器，用于判断当前检索到的文档是否足够回答原始问题"""
+    
+    def __init__(self):
+        self.llm = ChatOllama(model=LOCAL_LLM, format="json", temperature=0)
+        self.prompt = PromptTemplate(
+            template="""你是一个专家评分员，负责评估检索到的文档是否包含足够的信息来回答用户的问题。
+            
+            原始问题: {question}
+            
+            目前检索到的文档集合:
+            {documents}
+            
+            任务：
+            判断上述文档是否已经包含了回答原始问题所需的全部关键信息。
+            - 如果信息充足，可以终止进一步的检索，返回 'yes'。
+            - 如果信息缺失，需要继续检索更多信息，返回 'no'。
+            
+            输出格式：
+            返回一个只包含 'score' 键的 JSON，值为 'yes' 或 'no'。
+            不要输出任何前言或解释。""",
+            input_variables=["question", "documents"],
+        )
+        self.grader = self.prompt | self.llm | JsonOutputParser()
+    
+    def grade(self, question: str, documents: str) -> str:
+        """评估文档是否足以回答问题"""
+        result = self.grader.invoke({"question": question, "documents": documents})
+        return result.get("score", "no")
+
+
 class QueryRewriter:
     """查询重写器，优化查询以获得更好的检索结果"""
     
     def __init__(self):
         self.llm = ChatOllama(model=LOCAL_LLM, temperature=0)
         self.prompt = PromptTemplate(
-            template="""你是一个问题重写器，将输入问题转换为更适合向量存储检索的更好版本。
-            查看初始问题并制定一个改进的问题。
-            这里是初始问题：\n\n {question}。改进的问题（无前言）：\n """,
-            input_variables=["question"],
+            template="""你是一个问题重写器，负责将输入问题转换为更适合向量存储检索的更好版本。
+            
+            你的目标是根据原始问题和（可选的）之前的检索上下文，生成一个新的查询，以便检索到回答问题所需的缺失信息。
+            如果提供了之前的上下文，请分析其中缺少什么信息，并针对缺失的信息构建查询。
+            
+            初始问题: {question}
+            
+            之前的上下文（如果有）:
+            {context}
+            
+            改进的问题（只输出问题，无前言）:""",
+            input_variables=["question", "context"],
         )
         self.rewriter = self.prompt | self.llm | StrOutputParser()
     
-    def rewrite(self, question: str) -> str:
+    def rewrite(self, question: str, context: str = "") -> str:
         """重写查询以获得更好的检索效果"""
         print(f"---原始查询: {question}---")
-        rewritten_query = self.rewriter.invoke({"question": question})
+        if context:
+            print(f"---参考上下文长度: {len(context)} 字符---")
+            
+        rewritten_query = self.rewriter.invoke({"question": question, "context": context})
         print(f"---重写查询: {rewritten_query}---")
         return rewritten_query
 
@@ -187,11 +270,15 @@ def initialize_graders_and_router():
     answer_grader = AnswerGrader()
     hallucination_grader = HallucinationGrader(method=detection_method)
     query_rewriter = QueryRewriter()
+    query_decomposer = QueryDecomposer()
+    answerability_grader = AnswerabilityGrader()
     
     return {
         "query_router": query_router,
         "document_grader": document_grader,
         "answer_grader": answer_grader,
         "hallucination_grader": hallucination_grader,
-        "query_rewriter": query_rewriter
+        "query_rewriter": query_rewriter,
+        "query_decomposer": query_decomposer,
+        "answerability_grader": answerability_grader
     }

workflow_nodes.py

@@ -49,6 +49,9 @@ class GraphState(TypedDict):
     documents: List[str]
     retry_count: int
     retrieval_metrics: dict  # 添加检索评估指标
+    sub_queries: List[str]  # 分解后的子问题列表
+    current_query_index: int  # 当前处理的子问题索引
+    original_question: str # 原始问题，用于早期终止检查
 
 
 class WorkflowNodes:
@@ -64,14 +67,18 @@ class WorkflowNodes:
         
         # 设置RAG链 - 使用本地提示模板
         rag_prompt_template = PromptTemplate(
-            template="""你是一个问答助手。使用以下检索到的上下文来回答问题。
-如果你不知道答案，就说你不知道。最多使用三句话并保持答案简洁。
-
-问题: {question}
-
-上下文: {context}
-
-答案:""",
+            template="""你是一个智能问答助手。使用以下检索到的上下文来回答问题。
+            
+            规则：
+            1. 如果你不知道答案，就说你不知道。
+            2. 如果用户请求特定格式（如Markdown、列表、代码块等），请严格遵守。
+            3. 如果没有特定格式要求，保持答案简洁。
+            
+            问题: {question}
+            
+            上下文: {context}
+            
+            答案:""",
             input_variables=["question", "context"]
         )
         llm = ChatOllama(model=LOCAL_LLM, temperature=0)
@@ -80,6 +87,37 @@ class WorkflowNodes:
         # 设置网络搜索
         self.web_search_tool = TavilySearchResults(k=WEB_SEARCH_RESULTS_COUNT)
     
+    def decompose_query(self, state):
+        """
+        将初始查询分解为子查询
+        
+        Args:
+            state (dict): 当前图状态
+            
+        Returns:
+            state (dict): 更新sub_queries和current_query_index
+        """
+        print("---查询分解---")
+        question = state["question"]
+        
+        # 使用分解器
+        sub_queries = self.graders["query_decomposer"].decompose(question)
+        
+        # 如果分解器返回空或只有一个问题，我们仍然将其视为列表
+        if not sub_queries:
+            sub_queries = [question]
+            
+        print(f"   生成了 {len(sub_queries)} 个子查询")
+        
+        return {
+            "sub_queries": sub_queries, 
+            "current_query_index": 0,
+            "question": sub_queries[0], # 将当前问题设置为第一个子查询
+            "original_question": question, # 保存原始问题
+            "documents": [], # 清空文档，准备开始新的检索
+            "retry_count": 0
+        }
+
     def retrieve(self, state):
         """
         检索文档
@@ -111,7 +149,7 @@ class WorkflowNodes:
             
             # 记录使用的检索方法
             if ENABLE_HYBRID_SEARCH:
-                print("---使用混合检索---")
+                print("---使用混合检索(向量+关键词)---")
             if ENABLE_QUERY_EXPANSION:
                 print("---使用查询扩展---")
             if image_paths and ENABLE_MULTIMODAL:
@@ -136,6 +174,26 @@ class WorkflowNodes:
                 print(f"❌ 回退检索也失败: {fallback_e}")
                 documents = []
         
+        # === 向量多跳检索支持：合并上下文 ===
+        # 如果这不是第一次检索（即重试次数 > 0 或 正在处理后续子查询），说明之前的检索结果可能不完整或问题被重写了
+        # 我们应该保留之前的有价值文档，实现 "累积式上下文" (Accumulated Context)
+        current_query_index = state.get("current_query_index", 0)
+        if (retry_count > 0 or current_query_index > 0) and "documents" in state and state["documents"]:
+            print(f"---多跳上下文合并 (轮次 {retry_count}, 子查询 {current_query_index})---")
+            previous_docs = state["documents"]
+            if previous_docs:
+                # 简单的去重合并（基于内容）
+                current_content = {d.page_content for d in documents}
+                merged_count = 0
+                for prev_doc in previous_docs:
+                    # 只有当内容不重复时才添加
+                    if prev_doc.page_content not in current_content:
+                        documents.append(prev_doc)
+                        current_content.add(prev_doc.page_content)
+                        merged_count += 1
+                print(f"   合并了 {merged_count} 个来自上一轮/上一跳的文档，当前总文档数: {len(documents)}")
+        # =================================
+        
         # 计算检索时间
         retrieval_time = time.time() - retrieval_start_time
         
@@ -161,10 +219,12 @@ class WorkflowNodes:
         """
         print("---生成---")
         question = state["question"]
+        original_question = state.get("original_question", question) # 优先使用原始问题
         documents = state["documents"]
         
-        # RAG生成
-        generation = self.rag_chain.invoke({"context": documents, "question": question})
+        # RAG生成 - 使用原始问题以确保回答用户的初始意图
+        # 如果用户有特定的格式要求（如Markdown），通常包含在original_question中
+        generation = self.rag_chain.invoke({"context": documents, "question": original_question})
         return {"documents": documents, "question": question, "generation": generation}
     
     def grade_documents(self, state):
@@ -211,8 +271,18 @@ class WorkflowNodes:
         
         print(f"   重试次数: {retry_count}")
         
-        # 重写问题
-        better_question = self.graders["query_rewriter"].rewrite(question)
+        # 提取当前上下文摘要，帮助重写器理解缺失信息
+        context_summary = ""
+        if documents:
+            # 只提取前两个文档的摘要，避免上下文过长
+            docs_content = [d.page_content for d in documents[:2]]
+            context_summary = "\n---\n".join(docs_content)
+            # 截断以防止过长
+            if len(context_summary) > 2000:
+                context_summary = context_summary[:2000] + "...(截断)"
+        
+        # 重写问题，传入上下文
+        better_question = self.graders["query_rewriter"].rewrite(question, context=context_summary)
         return {"documents": documents, "question": better_question, "retry_count": retry_count}
     
     def web_search(self, state):
@@ -257,18 +327,65 @@ class WorkflowNodes:
             print("---将问题路由到RAG---")
             return "vectorstore"
     
+    def prepare_next_query(self, state):
+        """
+        准备下一个子查询：提取桥接实体并重写查询
+        
+        Args:
+            state (dict): 当前图状态
+            
+        Returns:
+            state (dict): 更新question, current_query_index, retry_count
+        """
+        print("---准备下一个子查询---")
+        current_query_index = state.get("current_query_index", 0)
+        sub_queries = state.get("sub_queries", [])
+        documents = state["documents"]
+        
+        # 移动到下一个索引
+        next_index = current_query_index + 1
+        next_query_raw = sub_queries[next_index]
+        
+        print(f"   原始下一个子查询: {next_query_raw}")
+        
+        # 提取上下文摘要用于重写（桥接实体提取）
+        context_summary = ""
+        if documents:
+            # 使用所有相关文档作为上下文
+            docs_content = [d.page_content for d in documents]
+            context_summary = "\n---\n".join(docs_content)
+            # 截断
+            if len(context_summary) > 3000:
+                context_summary = context_summary[:3000] + "...(截断)"
+                
+        # 使用重写器将上下文（包含桥接实体）注入到下一个查询中
+        # 例如：Q1结果是"作者是J.K. Rowling"，Q2是"她出生在哪里？" -> "J.K. Rowling出生在哪里？"
+        better_next_query = self.graders["query_rewriter"].rewrite(next_query_raw, context=context_summary)
+        
+        print(f"   优化后的下一个子查询: {better_next_query}")
+        
+        return {
+            "question": better_next_query,
+            "current_query_index": next_index,
+            "retry_count": 0, # 重置重试计数
+            "documents": documents # 保留文档作为上下文
+        }
+
     def decide_to_generate(self, state):
         """
-        确定是生成答案还是重新生成问题
+        确定是生成答案、继续下一个子查询还是重新生成问题
         
         Args:
             state (dict): 当前图状态
             
         Returns:
-            str: 要调用的下一个节点的二进制决策
+            str: 要调用的下一个节点的决策
         """
         print("---评估已评分的文档---")
         filtered_documents = state["documents"]
+        current_query_index = state.get("current_query_index", 0)
+        sub_queries = state.get("sub_queries", [])
+        original_question = state.get("original_question", "")
         
         if not filtered_documents:
             # 所有文档都被过滤掉了
@@ -276,9 +393,34 @@ class WorkflowNodes:
             print("---决策：所有文档都与问题不相关，转换查询---")
             return "transform_query"
         else:
-            # 我们有相关文档，所以生成答案
-            print("---决策：生成---")
-            return "generate"
+            # 我们有相关文档
+            # 检查是否有更多子查询
+            if sub_queries and current_query_index < len(sub_queries) - 1:
+                # === 早期终止检查 ===
+                # 检查当前累积的文档是否已经足以回答原始问题
+                if original_question:
+                    print("---检查是否已获取足够信息 (早期终止)---")
+                    
+                    # 准备文档上下文
+                    docs_content = [d.page_content for d in filtered_documents]
+                    context_summary = "\n---\n".join(docs_content)
+                    if len(context_summary) > 5000: # 限制上下文长度
+                        context_summary = context_summary[:5000]
+                        
+                    score = self.graders["answerability_grader"].grade(original_question, context_summary)
+                    
+                    if score == "yes":
+                        print(f"---决策：当前信息已足够回答原始问题，跳过剩余 {len(sub_queries) - 1 - current_query_index} 个子查询---")
+                        return "generate"
+                    else:
+                        print("---决策：信息尚不完整，继续下一个子查询---")
+                
+                print(f"---决策：当前子查询 ({current_query_index + 1}/{len(sub_queries)}) 完成，准备下一个---")
+                return "prepare_next_query"
+            else:
+                # 所有子查询都完成（或没有子查询），生成答案
+                print("---决策：所有子查询完成，生成---")
+                return "generate"
     
     def grade_generation_v_documents_and_question(self, state):
         """