RAG-5-案例1

          本文通过整合前面4篇博文，包括文本读取、切片、向量化及向量数据库等技术，实现了一个完整的简单的RAG应用案例，回顾加强前面的知识。

        本文对话模型使用智谱GLM-4, embedding模型使用智谱embedding-3

一、数据加载

案例使用网页数据百度百科黑神话悟空，地址：https://baike.baidu.com/item/%E9%BB%91%E7%A5%9E%E8%AF%9D%EF%BC%9A%E6%82%9F%E7%A9%BA/53303078

import jsonimport bs4
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.embeddings import ZhipuAIEmbeddings
from langchain_openai import ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from pymilvus import MilvusClient, FieldSchema, DataType, Function, FunctionType, CollectionSchema
from langchain import hub
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI# 加载文档
loader = WebBaseLoader(web_paths=("https://baike.baidu.com/item/%E9%BB%91%E7%A5%9E%E8%AF%9D%EF%BC%9A%E6%82%9F%E7%A9%BA/53303078",),bs_kwargs=dict(parse_only=bs4.SoupStrainer(class_=("mainContent_XnadV",)))
)docs = loader.load()
print(f"文档数量:{len(docs)}")
print(f"文档内容:{docs[0].page_content}")
print(f"文档元数据:{docs[0].metadata}")

二、数据切片

# 文档切片
text_splitter = RecursiveCharacterTextSplitter(chunk_size=800,chunk_overlap=200
)
splits_doc = text_splitter.split_documents(docs)
print(f"文档切片数量:{len(splits_doc)}")
print(f"文档切片内容:{splits_doc[0].page_content}")
print(f"文档切片元数据:{splits_doc[0].metadata}")

chunk_size=800：
这个参数定义了每个文本块的最大长度为800个字符
文本分割器会尝试将文档分割成不超过800个字符的块
注意这是字符数而不是字节数，对于中文文本，一个汉字算作一个字符
chunk_overlap=200：
这个参数定义了相邻文本块之间的重叠部分为200个字符
重叠的目的是避免在分割点处丢失上下文信息
例如，如果第一个块是字符0-800，第二个块会从字符600开始（800-200=600）到140

三、embedding

#embedding模型
embedding_generator = ZhipuAIEmbeddings(api_key=ZHIPU_API_KEY,model="embedding-3",dimensions=1536
)texts = [document.page_content for document in splits_doc]
# 生成向量
embeddings = embedding_generator.embed_documents(texts)

四、导入到向量数据库

#向量数据库
client = MilvusClient(uri='http://127.0.0.1:19530')# 删除已存在的集合（如果存在）
collection_name = "rag_collection"
if client.has_collection(collection_name):client.drop_collection(collection_name)fields = [FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),# 这里使用结巴分词FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=2500, enable_analyzer=True,analyzer_params={"tokenizer": "jieba", "filter": ["cnalphanumonly"]}),FieldSchema(name="metadata", dtype=DataType.VARCHAR, max_length=500),# 向量维度 我这里使用openai的embedding生成的向量默认维度是1536# embedding_llm1.embed_documents(titles)FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=1536),# 稀疏向量字段FieldSchema(name="sparse_vector", dtype=DataType.SPARSE_FLOAT_VECTOR)
]bm25_function = Function(name="bm25",input_field_names=["text"],output_field_names=["sparse_vector"],function_type=FunctionType.BM25,
)
schema = CollectionSchema(fields=fields, functions=[bm25_function], description="文档集合")client.create_collection(collection_name=collection_name,schema=schema,primary_key="id",consistency_level="Strong",  # 一致性级别enable_dynamic_field=True,  # 动态字段 如果设为 True，Milvus 将自动接受未在 schema 中定义的新字段数据，并将其存储为 JSON 格式。auto_id=True  # 自动生成ID。
)insert_data = []for i in range(len(splits_doc)):insert_data.append({"text": splits_doc[i].page_content,"metadata": json.dumps(splits_doc[i].metadata, ensure_ascii=False),#元数据从dict转成json"vector": embeddings[i],})
res = client.insert(collection_name=collection_name, data=insert_data)
print(res)# 创建索引参数对象
index_params = client.prepare_index_params()# 添加一个索引，指定字段名称、索引类型、度量类型和参数
index_params.add_index(field_name="vector",index_type="HNSW",metric_type="IP",params={"M": 16, "efConstruction": 64}
)# 为 sparse_vector 字段添加 SPARSE_INVERTED_INDEX 索引
index_params.add_index(field_name="sparse_vector",index_type="SPARSE_INVERTED_INDEX",metric_type="BM25",params={"inverted_index_algo": "DAAT_MAXSCORE","bm25_k1": 1.2,"bm25_b": 0.75})# 创建索引
client.create_index(collection_name, index_params)

五、检索向量数据库

这里我直接使用向量检索也可以混合检索，实现可参考：https://blog.csdn.net/sunqingzhong44/article/details/148929075?spm=1001.2014.3001.5502

client.load_collection(COLLECTION_NAME)def search_milvus(query):"""在Milvus中搜索相关文档"""# 生成查询向量query_vector = embedding_generator.embed_query(query)# 执行搜索results = client.search(collection_name=COLLECTION_NAME,data=[query_vector],anns_field="vector",output_fields=["text", "metadata"],limit=3)# 提取文本内容texts = []for result in results[0]:  # 第一个查询的结果texts.append(result['entity']['text'])return textsprint(search_milvus("黑神话悟空的社会影响"))

六、整合大模型

#大模型
chat = ChatOpenAI(api_key=ZHIPU_API_KEY,base_url=ZHIPU_BASE_URL,model_name="GLM-4",temperature=0.5
)
# 提示模板
prompt = hub.pull("rlm/rag-prompt")#检索结果整合到一起
def format_docs(docs):"""将文档列表格式化为字符串"""return "\n\n".join(docs)# 构建RAG链
rag_chain = ({"context": lambda x: format_docs(search_milvus(x)), "question": RunnablePassthrough()}| prompt| chat| StrOutputParser()
)

七、测试验证

# 使用RAG链回答问题
response = rag_chain.invoke("黑神话悟空的社会影响")
print(response)

输出如下：

《黑神话：悟空》在2024年12月入选中国国际传播热词和十大网络用语，2025年1月写入浙江省政府工作报告，显示出其对中国文化及游戏产业的社会影响。此外，该游戏在国内外市场取得巨大成功，销量和收入均创下纪录。

和百度百科基本一致：

# 使用RAG链回答问题
response = rag_chain.invoke("黑神话悟空有哪些角色")
print(response)

输出如下：

《黑神话：悟空》的主要角色包括主角“天命人”和若干其他角色。

这个问题回答的不是很好，后面接入知识图谱，我们可以再试一下