一、介绍

RAG（检索增强生成） 是一种将信息检索与大型语言模型生成相结合的技术。它通过在生成答案前从知识库中检索相关信息，为 LLM 提供准确的上下文，从而生成更准确、更可靠的回答。

RAG 的核心原理

用户问题 → 检索相关文档 → 结合上下文 → LLM 生成答案

传统 LLM 的局限性：

  • 知识截止：模型训练数据有时间限制

  • 幻觉问题：可能生成不准确的信息

  • 无法访问私有数据：无法获取训练数据之外的信息

RAG 的优势：

  • 事实准确性：基于真实文档生成答案

  • 实时更新：知识库可以随时更新

  • 可追溯性：可以查看答案的来源

  • 成本效益：比微调模型更经济

RAG 的工作流程

1. 文档处理阶段（离线）

原始文档 → 文本分割 → 向量化 → 向量数据库存储

2. 查询阶段（在线）

用户问题 → 向量化 → 相似度检索 → 上下文构建 → LLM 生成答案

RAG 的典型使用场景

1. 企业知识库问答

# 示例：公司内部文档问答
question = "我们公司的请假政策是什么？"
# RAG 会从员工手册、政策文件中检索相关信息

2. 学术研究助手

# 示例：论文研究
question = "关于深度学习在医疗影像中的应用有哪些最新进展？"
# RAG 会从学术论文库中检索相关研究

3. 客服机器人

# 示例：产品支持
question = "如何重置我的账户密码？"
# RAG 会从帮助文档、FAQ 中检索解决方案

4. 法律文档分析

# 示例：法律咨询
question = "劳动合同中关于解约的条款有哪些？"
# RAG 会从法律条文、合同模板中检索相关信息

二、使用

阶段一：准备知识库

1. 收集文档

documents = [
    "公司政策文档.pdf",
    "产品手册.docx",
    "技术文档.md",
    "常见问题解答.txt"
]

2. 文档预处理

def preprocess_documents(docs):
    # 文本提取
    texts = extract_text_from_files(docs)

    # 文本清洗
    cleaned_texts = clean_text(texts)

    # 文本分割
    chunks = split_text_into_chunks(cleaned_texts)

    return chunks

3. 向量化存储

def build_vector_store(chunks):
    # 使用嵌入模型生成向量
    embeddings = embedding_model.encode(chunks)

    # 存储到向量数据库
    vector_store.add_documents(chunks, embeddings)

阶段二：查询使用

1. 基本查询

def rag_query(question):
    # 1. 检索相关文档
    relevant_docs = vector_store.search(question, top_k=3)

    # 2. 构建上下文
    context = build_context(relevant_docs)

    # 3. 生成答案
    prompt = f"""基于以下上下文信息回答问题：

上下文：
{context}

问题：{question}

请根据上下文提供准确的回答。"""

    answer = llm.generate(prompt)
    return answer, relevant_docs

2. 带来源引用的查询

def rag_query_with_sources(question):
    relevant_docs = vector_store.search(question, top_k=5)

    context = ""
    sources = []
    for i, doc in enumerate(relevant_docs):
        context += f"[{i+1}] {doc.content}\n\n"
        sources.append({
            "content": doc.content,
            "metadata": doc.metadata,
            "score": doc.score
        })

    answer = llm.generate(f"上下文：\n{context}\n问题：{question}")

    return {
        "answer": answer,
        "sources": sources,
        "context": context
    }

实际使用示例

示例 1：技术文档问答

# 初始化 RAG 系统
rag_system = RAGSystem(
    embedding_model="text-embedding-ada-002",
    llm="gpt-4",
    vector_store="chromadb"
)

# 构建知识库
rag_system.build_knowledge_base("technical_docs/")

# 查询使用
result = rag_system.query("如何配置数据库连接池？")
print(f"答案: {result['answer']}")
print("参考文档:")
for source in result['sources']:
    print(f"- {source['metadata']['file_name']}")

示例 2：多轮对话 RAG

class ConversationalRAG:
    def __init__(self):
        self.conversation_history = []

    def chat(self, question):
        # 结合对话历史进行检索
        context_query = self.build_context_query(question)

        # 检索相关文档
        relevant_docs = self.vector_store.search(context_query)

        # 构建包含历史的提示
        prompt = self.build_prompt(question, relevant_docs, self.conversation_history)

        # 生成答案
        answer = self.llm.generate(prompt)

        # 更新对话历史
        self.conversation_history.append({"user": question, "assistant": answer})

        return answer, relevant_docs

示例 3：带过滤的 RAG

def filtered_rag_query(question, filters=None):
    # 使用元数据过滤
    relevant_docs = vector_store.search(
        question,
        top_k=5,
        filter={"document_type": "policy"}  # 只检索政策文档
    )

    # 构建上下文和生成答案
    context = build_context(relevant_docs)
    answer = llm.generate(f"基于以下政策文档回答：\n{context}\n问题：{question}")

    return answer, relevant_docs

RAG 的最佳实践

1. 文档预处理优化

  • 分块策略：根据文档类型调整块大小

  • 重叠设置：保持块间重叠以避免信息断裂

  • 元数据丰富：为每个块添加来源、类型等元数据

2. 检索优化

  • 多路检索：结合关键词检索和向量检索

  • 重排序：使用更精细的模型对初步结果重排序

  • 查询扩展：对原始问题进行同义扩展

3. 提示工程优化

def build_optimized_prompt(question, context):
    return f"""你是一个专业的助手。请严格基于提供的上下文信息回答问题。

上下文信息：
{context}

用户问题：{question}

请遵循以下要求：
1. 如果上下文包含答案，请直接基于上下文回答
2. 如果上下文不包含足够信息，请说明信息来源不足
3. 不要编造上下文之外的信息
4. 引用具体的上下文片段

请给出专业、准确的回答："""

4. 评估和监控

  • 答案质量评估：准确性、相关性、完整性

  • 检索质量评估：召回率、准确率

  • 用户反馈收集： thumbs up/down 机制

常用工具和框架

向量数据库

  • ChromaDB：轻量级，易于使用

  • Pinecone：云服务，高性能

  • Weaviate：开源，功能丰富

  • Qdrant：高性能，Rust 开发

嵌入模型

  • OpenAI Embeddings：text-embedding-ada-002

  • Sentence Transformers：all-MiniLM-L6-v2

  • Cohere Embed：多语言支持

  • BGE Models：中文优化

开发框架

  • LangChain：功能全面的 RAG 框架

  • LlamaIndex：专为 RAG 优化

  • Haystack：企业级解决方案

RAG 通过将检索和生成结合，有效解决了 LLM 的知识局限性和幻觉问题。在实际使用中，关键是：

  1. 高质量的知识库构建

  2. 合适的检索策略

  3. 优化的提示工程

  4. 持续的评估和改进

三、创建

基于deepseek api_key和ChromaDB的三体知识库

1.simple_rag

  1. 初始化系统

    • 导入RAG系统模块

    • 创建系统实例

  2. 准备知识库

    • 检查是否已经构建过向量数据库

    • 如果是首次运行，会读取"threebody_problem.txt"文件

    • 将小说内容分割成小块，转换成向量格式存储

  3. 演示问答

    • 准备了6个关于《三体》的问题

    • 比如"叶哲泰是谁？"、"文章开头描述的是什么场景？"等

  4. 处理每个问题

    • 系统在向量数据库中搜索相关内容

    • 生成基于小说内容的答案

    • 显示答案和参考了哪些原文片段

  5. 结果显示

    • 展示生成的答案

    • 列出参考的文档片段及其来源

    • 显示每个片段的前100个字符作为预览

import os
import sys
from typing import List, Dict, Any
import chromadb
import numpy as np
import requests
import json

class SimpleThreeBodyRAG:
    """简化的三体RAG系统"""
    
    def __init__(self, persist_directory: str = "./chroma_db_simple", api_key: str = None):
        self.persist_directory = persist_directory
        self.client = None
        self.collection = None
        self.embedding_model = "deepseek_api"  # 标记为使用 API
        self.api_key = api_key or "your api_key"
        self._initialize_components()
    
    def _initialize_components(self):
        """初始化组件"""
        print("使用 DeepSeek API 进行问答，ChromaDB 默认嵌入")
        
        # 初始化ChromaDB客户端
        self.client = chromadb.PersistentClient(path=self.persist_directory)
        
        # 创建或获取集合（让 ChromaDB 使用默认嵌入函数）
        try:
            self.collection = self.client.get_collection("threebody_docs")
            print("已加载现有向量数据库")
        except:
            self.collection = self.client.create_collection("threebody_docs")
            print("创建新的向量数据库（使用默认嵌入）")
    
    def load_and_chunk_documents(self, file_path: str) -> List[Dict[str, Any]]:
        """加载文档并分块"""
        text = None  # 先初始化变量
        
        # 尝试多种编码
        for encoding in ['utf-8', 'gbk', 'utf-16']:
            try:
                with open(file_path, 'r', encoding=encoding) as f:
                    text = f.read()
                print(f"成功使用编码: {encoding}")
                break  # 如果成功就跳出循环
            except UnicodeDecodeError:
                continue  # 如果失败就尝试下一个编码
        
        # 检查是否成功读取
        if text is None:
            # 如果所有编码都失败，使用忽略错误的方式
            try:
                with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
                    text = f.read()
                print("使用 utf-8 with ignore 读取文件")
            except Exception as e:
                raise ValueError(f"无法读取文件 {file_path}: {str(e)}")
        
        # 简单的文本分割
        paragraphs = text.split('\n\n')
        chunks = []
        
        for i, para in enumerate(paragraphs):
            cleaned_para = para.strip()
            # 更严格的内容检查
            if len(cleaned_para) > 50 and any(char.isalnum() for char in cleaned_para):
                chunks.append({
                    "id": f"chunk_{i}",
                    "content": cleaned_para,
                    "metadata": {"source": "threebody_problem.txt", "chunk_id": i}
                })
        
        print(f"成功分割为 {len(chunks)} 个文档块")
        
        # 验证所有块都有内容
        for chunk in chunks:
            if not chunk["content"] or chunk["content"] is None:
                print(f"警告: 发现空内容块: {chunk['id']}")
        
        return chunks
    
    
    def build_vector_store(self, file_path: str):
        """构建向量存储"""
        print("正在加载和分割文档...")
        chunks = self.load_and_chunk_documents(file_path)
        print(f"文档已分割为 {len(chunks)} 个片段")
        
        # 过滤有效块
        valid_chunks = [
            chunk for chunk in chunks 
            if chunk["content"] and isinstance(chunk["content"], str) and chunk["content"].strip()
        ]
        
        print(f"有效文档块: {len(valid_chunks)} 个")
        
        if not valid_chunks:
            raise ValueError("没有有效的文档内容可处理")
        
        # 直接添加到向量数据库，ChromaDB 会自动处理嵌入
        print("正在构建向量数据库...")
        self.collection.add(
            documents=[chunk["content"] for chunk in valid_chunks],
            metadatas=[chunk["metadata"] for chunk in valid_chunks],
            ids=[chunk["id"] for chunk in valid_chunks]
        )
        print(f"向量数据库构建完成！共添加 {len(valid_chunks)} 个文档片段")
    
    def _call_deepseek_chat(self, prompt: str) -> str:
        """调用DeepSeek API生成答案"""
        try:
            url = "https://api.deepseek.com/chat/completions"
            headers = {
                "Content-Type": "application/json",
                "Authorization": f"Bearer {self.api_key}"
            }
            
            data = {
                "model": "deepseek-chat",
                "messages": [
                    {
                        "role": "system",
                        "content": "你是一个专业的《三体》系列知识助手。请基于提供的上下文信息，准确、简洁地回答用户的问题。如果上下文信息不足以回答问题，请明确说明。"
                    },
                    {
                        "role": "user",
                        "content": prompt
                    }
                ],
                "temperature": 0.7,
                "max_tokens": 1000
            }
            
            response = requests.post(url, headers=headers, json=data, timeout=30)
            response.raise_for_status()
            
            result = response.json()
            return result["choices"][0]["message"]["content"]
            
        except Exception as e:
            print(f"DeepSeek API调用失败: {e}")
            return f"抱歉，生成答案时出现错误: {str(e)}"
    
    def query(self, question: str, n_results: int = 3) -> Dict[str, Any]:
        """查询系统"""
        try:
            # 直接使用文本查询，ChromaDB 会自动处理嵌入
            print("正在检索相关文档...")
            results = self.collection.query(
                query_texts=[question],
                n_results=n_results
            )
            
            # 构建更详细的参考信息
            source_documents = []
            if results['documents'] and results['metadatas']:
                for i, (doc, metadata) in enumerate(zip(results['documents'][0], results['metadatas'][0])):
                    source_documents.append({
                        "content": doc,
                        "metadata": metadata,
                        "source": metadata.get("source", "unknown"),
                        "chunk_id": metadata.get("chunk_id", i)
                    })
            
            context = "\n".join([doc["content"] for doc in source_documents])
            
            # 使用DeepSeek API生成答案
            if context:
                prompt = f"""请基于以下上下文信息回答用户的问题：

    用户问题：{question}

    相关上下文：
    {context}

    请基于上述上下文信息，给出准确、简洁的回答。"""
                
                print("正在使用DeepSeek API生成答案...")
                answer = self._call_deepseek_chat(prompt)
            else:
                answer = "抱歉，在文档中没有找到与您问题相关的信息。"
            
            return {
                "question": question,
                "answer": answer,
                "context": context,
                "source_documents": source_documents
            }
            
        except Exception as e:
            return {
                "question": question,
                "answer": f"查询过程中出现错误: {str(e)}",
                "context": "",
                "source_documents": []
            }

def main():
    """主函数"""
    rag_system = SimpleThreeBodyRAG()
    
    # 检查是否已构建向量数据库
    if rag_system.collection.count() == 0:
        print("首次运行，正在构建向量数据库...")
        rag_system.build_vector_store("threebody_problem.txt")
    else:
        print(f"向量数据库已存在，包含 {rag_system.collection.count()} 个文档片段")
    
    print("\n=== 简化版三体RAG系统已就绪 ===")
    print("你可以用自然语言询问关于《三体》系列的任何问题")
    print("输入 'quit' 或 '退出' 来结束程序\n")
    
    while True:
        try:
            question = input("请输入你的问题: ").strip()
            
            if question.lower() in ['quit', '退出', 'exit']:
                print("感谢使用三体RAG系统！")
                break
            
            if not question:
                continue
            
            print("\n正在处理问题...")
            result = rag_system.query(question)
            
            print(f"\n答案: {result['answer']}")
            print(f"\n参考了 {len(result['source_documents'])} 个相关文档片段")
            print("-" * 80)
            
        except KeyboardInterrupt:
            print("\n\n程序被用户中断")
            break
        except Exception as e:
            print(f"发生错误: {e}")

if __name__ == "__main__":
    main()

2.demo

#!/usr/bin/env python3
"""
三体RAG系统演示脚本
"""

import os
import sys

def demo_simple_rag():
    """演示简化版RAG系统"""
    print("=== 三体RAG系统演示 ===\n")
    
    try:
        # 尝试导入简化版RAG系统
        from simple_rag import SimpleThreeBodyRAG
        
        # 初始化系统
        print("正在初始化RAG系统...")
        rag_system = SimpleThreeBodyRAG()
        
        # 检查是否需要构建向量数据库
        if rag_system.collection.count() == 0:
            print("首次运行，正在构建向量数据库...")
            rag_system.build_vector_store("threebody_problem.txt")
        else:
            print(f"向量数据库已存在，包含 {rag_system.collection.count()} 个文档片段")
        
        print("\n=== 演示开始 ===\n")
        
        # 演示问题列表
        demo_questions = [
        "叶哲泰是谁？发生了什么事情？",
        "文章开头描述的是什么场景？", 
        "文化大革命在故事中如何体现？",
        "红卫兵在故事中扮演什么角色？",
        "物理学教授在故事中遇到了什么？",
        "故事开头的时间背景是什么？"
    ]
        
        for i, question in enumerate(demo_questions, 1):
            print(f"问题 {i}: {question}")
            print("-" * 50)
            
            result = rag_system.query(question)
            
            print(f"答案: {result['answer']}")
            print(f"\n参考了 {len(result['source_documents'])} 个相关文档片段：")
            
            # 显示具体的参考片段
            for j, doc in enumerate(result['source_documents'], 1):
                print(f"\n参考片段 {j} (来源: {doc['metadata'].get('source', 'unknown')}, 块ID: {doc['metadata'].get('chunk_id', 'N/A')}):")
                # 显示片段的前100个字符作为预览
                preview = doc['content'][:100] + "..." if len(doc['content']) > 100 else doc['content']
                print(f"   {preview}")
            
            print("\n" + "="*80 + "\n")

        print("演示完成！")
        
    except ImportError as e:
        print(f"导入错误: {e}")
        print("请先安装依赖：pip install chromadb sentence-transformers numpy")
    except Exception as e:
        print(f"演示过程中出现错误: {e}")

if __name__ == "__main__":
    demo_simple_rag()

# python # 人工智能
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

AI Agent的隐私保护与安全机制

随着人工智能技术的飞速发展，AI Agent在各个领域得到了广泛应用。然而，AI Agent在处理和使用数据的过程中，涉及到大量用户的隐私信息，如个人身份、健康数据、财务信息等。因此，如何保护这些隐私信息的安全，防止数据泄露和滥用，成为了当前人工智能领域面临的重要挑战。本文的目的在于深入探讨AI Agent的隐私保护与安全机制，介绍相关的技术和方法，为开发人员、研究人员和决策者提供参考。

avatar 2048 AI社区
cover

005-Spring AI Alibaba Structured Output 功能完整案例

avatar 2048 AI社区
cover

制造业转型：一线工人技能升级的机遇与挑战。

avatar 2048 AI社区