RAGFlow：2025年新一代RAG框架的技术革新与实践

# 多模态查询理解器核心代码示例
class MultimodalQueryUnderstanding:
    def __init__(self, llm_model, vision_model, audio_model):
        self.llm_model = llm_model
        self.vision_model = vision_model
        self.audio_model = audio_model
    
    def understand_query(self, query):
        # 1. 检测查询类型
        query_type = self._detect_query_type(query)
        
        # 2. 根据查询类型处理
        if query_type == "text":
            return self._process_text_query(query)
        elif query_type == "image":
            return self._process_image_query(query)
        elif query_type == "audio":
            return self._process_audio_query(query)
        elif query_type == "multimodal":
            return self._process_multimodal_query(query)
        else:
            raise ValueError(f"Unsupported query type: {query_type}")
    
    def _detect_query_type(self, query):
        # 检测查询类型
        if isinstance(query, str):
            return "text"
        elif hasattr(query, "image_data"):
            return "image"
        elif hasattr(query, "audio_data"):
            return "audio"
        elif hasattr(query, "multimodal_data"):
            return "multimodal"
        else:
            return "unknown"
    
    def _process_text_query(self, query):
        # 处理文本查询
        # 1. 查询扩展
        expanded_query = self._expand_query(query)
        # 2. 意图识别
        intent = self._recognize_intent(query)
        # 3. 实体提取
        entities = self._extract_entities(query)
        
        return {
            "type": "text",
            "original_query": query,
            "expanded_query": expanded_query,
            "intent": intent,
            "entities": entities
        }
    
    def _process_image_query(self, query):
        # 处理图像查询
        # 1. 图像理解
        image_content = self.vision_model.understand(query.image_data)
        # 2. 生成文本描述
        image_description = self.llm_model.generate(f"Describe this image: {image_content}")
        
        return {
            "type": "image",
            "image_description": image_description,
            "image_features": self.vision_model.extract_features(query.image_data)
        }
    
    def _process_audio_query(self, query):
        # 处理音频查询
        # 1. 语音转文本
        text = self.audio_model.transcribe(query.audio_data)
        # 2. 处理文本查询
        return self._process_text_query(text)
    
    def _process_multimodal_query(self, query):
        # 处理多模态查询
        results = []
        for data in query.multimodal_data:
            result = self.understand_query(data)
            results.append(result)
        return {
            "type": "multimodal",
            "components": results
        }
    
    def _expand_query(self, query):
        # 查询扩展
        pass
    
    def _recognize_intent(self, query):
        # 意图识别
        pass
    
    def _extract_entities(self, query):
        # 实体提取
        pass

# 多模态检索器核心代码示例
class MultimodalRetriever:
    def __init__(self, vector_store, keyword_store, semantic_store):
        self.vector_store = vector_store
        self.keyword_store = keyword_store
        self.semantic_store = semantic_store
        self.reranker = Reranker()
    
    def retrieve(self, query_understanding_result, top_k=10):
        # 1. 根据查询类型选择检索策略
        query_type = query_understanding_result["type"]
        
        if query_type == "text":
            results = self._retrieve_text(query_understanding_result, top_k)
        elif query_type == "image":
            results = self._retrieve_image(query_understanding_result, top_k)
        elif query_type == "multimodal":
            results = self._retrieve_multimodal(query_understanding_result, top_k)
        else:
            raise ValueError(f"Unsupported query type: {query_type}")
        
        # 2. 重排序
        reranked_results = self.reranker.rerank(query_understanding_result, results)
        
        return reranked_results[:top_k]
    
    def _retrieve_text(self, query_understanding_result, top_k):
        # 文本检索
        # 1. 向量检索
        vector_results = self.vector_store.search(
            query_understanding_result["expanded_query"], top_k=top_k*2
        )
        
        # 2. 关键词检索
        keyword_results = self.keyword_store.search(
            query_understanding_result["entities"], top_k=top_k*2
        )
        
        # 3. 语义检索
        semantic_results = self.semantic_store.search(
            query_understanding_result["intent"], top_k=top_k*2
        )
        
        # 4. 合并结果
        combined_results = self._merge_results([vector_results, keyword_results, semantic_results])
        
        return combined_results
    
    def _retrieve_image(self, query_understanding_result, top_k):
        # 图像检索
        # 1. 基于图像特征的向量检索
        vector_results = self.vector_store.search(
            query_understanding_result["image_features"], top_k=top_k*2
        )
        
        # 2. 基于图像描述的文本检索
        text_results = self._retrieve_text({
            "type": "text",
            "expanded_query": query_understanding_result["image_description"],
            "intent": "image_content",
            "entities": []
        }, top_k*2)
        
        # 3. 合并结果
        combined_results = self._merge_results([vector_results, text_results])
        
        return combined_results
    
    def _retrieve_multimodal(self, query_understanding_result, top_k):
        # 多模态检索
        all_results = []
        for component in query_understanding_result["components"]:
            component_results = self.retrieve(component, top_k*2)
            all_results.extend(component_results)
        
        # 合并并去重
        combined_results = self._merge_results([all_results])
        
        return combined_results
    
    def _merge_results(self, results_list):
        # 合并检索结果
        merged = {}
        for results in results_list:
            for result in results:
                if result["id"] not in merged:
                    merged[result["id"]] = result
                else:
                    # 合并分数
                    merged[result["id"]]["score"] = max(merged[result["id"]]["score"], result["score"])
        
        return sorted(merged.values(), key=lambda x: x["score"], reverse=True)

# 生成优化器核心代码示例
class GenerationOptimizer:
    def __init__(self, llm_model):
        self.llm_model = llm_model
    
    def generate(self, query, retrieved_knowledge):
        # 1. 知识融合
        fused_knowledge = self._fuse_knowledge(query, retrieved_knowledge)
        
        # 2. 生成提示词
        prompt = self._build_prompt(query, fused_knowledge)
        
        # 3. 多阶段生成
        preliminary_result = self.llm_model.generate(prompt)
        
        # 4. 结果验证
        validation_result = self._validate_result(preliminary_result, retrieved_knowledge)
        
        # 5. 结果修正
        if not validation_result["valid"]:
            refined_prompt = self._build_refined_prompt(query, fused_knowledge, preliminary_result, validation_result["issues"])
            final_result = self.llm_model.generate(refined_prompt)
        else:
            final_result = preliminary_result
        
        # 6. 添加引用
        final_result_with_citations = self._add_citations(final_result, retrieved_knowledge)
        
        return final_result_with_citations
    
    def _fuse_knowledge(self, query, retrieved_knowledge):
        # 知识融合
        # 1. 去重
        unique_knowledge = self._deduplicate_knowledge(retrieved_knowledge)
        
        # 2. 按相关性排序
        sorted_knowledge = sorted(unique_knowledge, key=lambda x: x["score"], reverse=True)
        
        # 3. 提取关键信息
        key_information = []
        for knowledge in sorted_knowledge[:5]:  # 只保留Top 5相关知识
            key_points = self.llm_model.generate(f"Extract key points from this knowledge that are relevant to the query '{query}': {knowledge['content']}")
            key_information.append({
                "id": knowledge["id"],
                "key_points": key_points,
                "content": knowledge["content"]
            })
        
        return key_information
    
    def _build_prompt(self, query, fused_knowledge):
        # 构建提示词
        knowledge_text = "\n".join([f"[{i+1}] {k['key_points']}" for i, k in enumerate(fused_knowledge)])
        
        prompt = f"""
        You are a helpful assistant. Answer the following query based on the provided knowledge. Make sure to:
        1. Directly address the query
        2. Use the provided knowledge to support your answer
        3. Maintain a natural and coherent flow
        4. Avoid hallucinations
        
        Query: {query}
        
        Knowledge:
        {knowledge_text}
        
        Answer:
        """
        
        return prompt
    
    def _validate_result(self, result, retrieved_knowledge):
        # 验证结果
        validation_prompt = f"""
        Validate if the following answer is consistent with the provided knowledge. Check for:
        1. Factual accuracy
        2. Consistency with the knowledge
        3. Absence of hallucinations
        
        Answer: {result}
        
        Knowledge: {retrieved_knowledge}
        
        Return a JSON object with:
        - valid: boolean indicating if the answer is valid
        - issues: list of issues found, if any
        """
        
        validation_result = self.llm_model.generate(validation_prompt)
        return json.loads(validation_result)
    
    def _build_refined_prompt(self, query, fused_knowledge, preliminary_result, issues):
        # 构建修正提示词
        knowledge_text = "\n".join([f"[{i+1}] {k['key_points']}" for i, k in enumerate(fused_knowledge)])
        
        prompt = f"""
        Refine the following answer based on the provided knowledge and issues. Make sure to:
        1. Fix all the identified issues
        2. Maintain consistency with the provided knowledge
        3. Keep the answer natural and coherent
        
        Query: {query}
        
        Knowledge:
        {knowledge_text}
        
        Preliminary Answer: {preliminary_result}
        
        Issues: {issues}
        
        Refined Answer:
        """
        
        return prompt
    
    def _add_citations(self, result, retrieved_knowledge):
        # 添加引用
        # 1. 识别结果中使用的知识
        used_knowledge = []
        for knowledge in retrieved_knowledge:
            if any(keyword in result for keyword in self._extract_keywords(knowledge["content"])):
                used_knowledge.append(knowledge)
        
        # 2. 添加引用标记
        cited_result = result
        for i, knowledge in enumerate(used_knowledge):
            # 简单的引用添加逻辑，实际实现会更复杂
            cited_result += f" [^{i+1}]"
        
        # 3. 添加引用列表
        cited_result += "\n\n**References:**\n"
        for i, knowledge in enumerate(used_knowledge):
            cited_result += f"[^{i+1}] {knowledge['source']}\n"
        
        return cited_result
    
    def _deduplicate_knowledge(self, knowledge_list):
        # 知识去重
        seen = set()
        unique = []
        for knowledge in knowledge_list:
            content_hash = hash(knowledge["content"])
            if content_hash not in seen:
                seen.add(content_hash)
                unique.append(knowledge)
        return unique
    
    def _extract_keywords(self, text):
        # 提取关键词
        keywords = self.llm_model.generate(f"Extract 5-10 keywords from this text: {text}")
        return keywords.split(", ")

# 克隆仓库
git clone https://github.com/RAGFlow/RAGFlow.git
cd RAGFlow

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/macOS
venv\Scripts\activate  # Windows

# 安装依赖
pip install -r requirements.txt

# 配置环境变量
export RAGFLOW_CONFIG="./config.yaml"
export RAGFLOW_API_KEY="your-api-key"
export RAGFLOW_LOG_LEVEL="INFO"

# 启动RAGFlow服务
python -m ragflow serve

# 导入RAGFlow
from ragflow import RAGFlow

# 初始化RAGFlow
ragflow = RAGFlow()

# 添加知识库
ragflow.add_knowledge_source("https://example.com/blog")
ragflow.add_knowledge_source("data/my_documents")

# 更新知识库
ragflow.update_knowledge_base()

# 文本查询示例
result = ragflow.query("什么是RAG技术？")
print("文本查询结果：")
print(result)

# 图像查询示例
# result = ragflow.query(ImageQuery(image_path="data/sample_image.jpg"))
# print("图像查询结果：")
# print(result)

# 多模态查询示例
# result = ragflow.query(MultimodalQuery([
#     TextQuery("这个产品的主要功能是什么？"),
#     ImageQuery(image_path="data/product_image.jpg")
# ]))
# print("多模态查询结果：")
# print(result)