Skip to content

RAG 技术

检索增强生成(RAG)是一种将检索与生成相结合的技术,用于增强大语言模型的知识准确性。

检索增强

概述

RAG 通过从外部知识库中检索相关信息,并将其作为上下文输入到语言模型中,从而生成更准确、更可靠的回答。

RAG 流程

用户查询

检索器从知识库中查找相关文档

将检索到的文档作为上下文

语言模型基于上下文生成回答

输出最终回答

实现示例

python
from langchain.document_loaders import TextLoader
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

# 加载文档
loader = TextLoader("knowledge_base.txt")
documents = loader.load()

# 创建向量数据库
embeddings = OpenAIEmbeddings()
db = Chroma.from_documents(documents, embeddings)

# 创建检索问答链
qa_chain = RetrievalQA.from_chain_type(
    llm=OpenAI(),
    chain_type="stuff",
    retriever=db.as_retriever()
)

# 查询
result = qa_chain.run("什么是 RAG 技术?")
print(result)

向量数据库

概述

向量数据库用于存储和检索高维向量数据,是 RAG 系统的核心组件。

常用向量数据库

  • Chroma: 轻量级开源向量数据库
  • Pinecone: 托管式向量数据库服务
  • Weaviate: 开源向量数据库,支持 GraphQL 查询
  • Milvus: 开源向量数据库,支持多种索引类型

使用 Chroma

python
import chromadb
from chromadb.config import Settings

# 初始化客户端
client = chromadb.Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="./chroma_db"
))

# 创建集合
collection = client.create_collection(name="knowledge_base")

# 添加文档
collection.add(
    documents=["RAG 是检索增强生成技术", "向量数据库用于存储嵌入向量"],
    metadatas=[{"source": "doc1"}, {"source": "doc2"}],
    ids=["1", "2"]
)

# 查询
results = collection.query(
    query_texts=["什么是 RAG?"],
    n_results=2
)

print(results)

知识图谱

概述

知识图谱是一种结构化的知识表示方法,用于存储实体之间的关系。

构建知识图谱

python
from pyvis.network import Network
import networkx as nx

# 创建图
G = nx.Graph()

# 添加节点
G.add_node("RAG", label="检索增强生成", color="#3b82f6")
G.add_node("向量数据库", label="向量数据库", color="#8b5cf6")
G.add_node("语言模型", label="语言模型", color="#10b981")

# 添加边
G.add_edge("RAG", "向量数据库", label="使用")
G.add_edge("RAG", "语言模型", label="结合")

# 可视化
nt = Network("500px", "500px")
nt.from_nx(G)
nt.show("knowledge_graph.html")

知识图谱查询

python
# 查询实体关系
def find_related_entities(G, entity):
    neighbors = list(G.neighbors(entity))
    relationships = []
    
    for neighbor in neighbors:
        edge_data = G.get_edge_data(entity, neighbor)
        relationships.append({
            "entity": neighbor,
            "relationship": edge_data.get("label", "相关")
        })
    
    return relationships

# 查询 RAG 相关实体
related = find_related_entities(G, "RAG")
print(related)