龙凤网站建设云聚达,公司做网站哪里好,重庆网站建设设计公司,互联网企业投诉服务平台作者#xff1a; caiyfc 原文来源#xff1a; https://tidb.net/blog/e0035e5e 一、背景 我最近在研究使用向量数据库搭建RAG应用#xff0c;并且已经使用 Milvus、Llama 3、Ollama、LangChain 搭建完成。最近通过活动获取了 TiDB Cloud Serverless 使用配额#xff… 作者 caiyfc 原文来源 https://tidb.net/blog/e0035e5e 一、背景 我最近在研究使用向量数据库搭建RAG应用并且已经使用 Milvus、Llama 3、Ollama、LangChain 搭建完成。最近通过活动获取了 TiDB Cloud Serverless 使用配额于是打算把 Milvus 已完成的向量数据给迁移到 TiDB Cloud Serverless 中。 经过查阅相关资料我发现向量数据迁移的工具还不支持从 Milvus 迁移到 TiDB。那就无法迁移了吗不虽然现有的工具不能迁移但是我可以手动迁移。于是就有了这篇文章。 TiDB Cloud Serverless 活动地址 【TiDB 社区福利】贡献开源代码的开发者看过来最高可获得超 14,000 元的 TiDB Cloud Serverless 云资源额度 搭建RAG应用方法 手把手系列 使用Milvus、Llama 3、Ollama、LangChain本地设置RAG应用 二、迁移方案 要做数据迁移首先需要确定迁移方案。最简单的迁移就两个步骤数据从源库导出、数据导入到目标库这样就完成了数据迁移。 但是这次就不同了。RAG应用使用了 LangChain根据调研LangChain 在 Milvus 和在 TiDB 中创建的结构是不同的。 在 Milvus 中的 collection 名称是LangChainCollection结构是 但是在 TiDB 中的table 名称是langchain_vector结构是 在 LangChain 的文档中也有说明 那么这次数据迁移就需要多增加两个步骤了数据整理、表结构调整。而这两个又是异构数据库所以导出的数据格式选择较为通用的csv。 整体方案如下 三、Milvus 数据导出 根据 Milvus 的官方文档没找到能直接把数据导出成 csv 文件的工具但是我可以用 python 的 SDK 来把数据读取出来然后存成 csv 文件。 import csv
from pymilvus import connections, Collection
# 连接到 Milvus
connections.connect(default, host10.3.xx.xx, port19530)
# 获取 Collection
collection Collection(LangChainCollection)
# 分页查询所有数据
limit 1000
offset 0
all_results []
while True:# 传递 expr 参数使用一个简单的条件查询所有数据results collection.query(expr, output_fields[pk, source, page, text, vector], limitlimit, offsetoffset)if not results:breakall_results.extend(results)offset limit
# 打开 CSV 文件准备写入数据
with open(milvus_data.csv, w, newline, encodingutf-8) as csvfile:# 定义 CSV 列名fieldnames [pk, source, page, text, vector]writer csv.DictWriter(csvfile, fieldnamesfieldnames)
# 写入表头writer.writeheader()
# 写入每一条记录for result in all_results:# 解析 JSON 数据提取字段vector_str ,.join(map(str, result.get(vector, []))) # 将向量数组转换为字符串writer.writerow({pk: result.get(pk), # 获取主键source: result.get(source), # 获取源文件page: result.get(page), # 获取页码text: result.get(text), # 获取文本vector: vector_str # 写入向量数据})
print(fTotal records written to CSV: {len(all_results)})导出的 csv 文件数据的格式为 四、数据整理、表结构整理 我用少量测试数据转换成向量使用 LangChain 加载到 TiDB Cloud 中了。这样就得到了 TiDB Cloud 中的数据结构及数据格式了。 表结构为 CREATE TABLE langchain_vector (
id varchar(36) NOT NULL,
embedding vector(768) NOT NULL COMMENT hnsw(distancecosine),
document text DEFAULT NULL,
meta json DEFAULT NULL,
create_time datetime DEFAULT CURRENT_TIMESTAMP,
update_time datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (id) /*T![clustered_index] CLUSTERED */
) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COLLATEutf8mb4_bin可以在 TiDB Cloud 中直接创建该表。 导出成 csv 的数据格式为省略部分内容 id,embedding,document,meta,create_time,update_time
00a2ad02-eff5-4649-947f-820db0d24afa,[-0.08534411,0.048610855,0.018906716,0.023978366,***********-0.023846595,0.06352842,0.07482053],— 22 — 七移交利用共用部位、共用设施设备经营的相关资料、\n物业服务费用和公共水电分摊费用交纳记录等资料 八法律、法规员会和物业服务企业。,{\page\: 21, \source\: \./湖北省物业服务和管理条例.pdf\},2024-10-15 08:18:16,2024-10-15 08:18:16
已知 Milvus 导出的 csv 的数文件根据对应关系其实就是embedding 对应 vectordocument 对应 textmeta 对应 page加source。这样逻辑就清晰了。根据对应关系编写数据整理的脚本 import pandas as pd
import json
from uuid import uuid4
from datetime import datetime
# 读取CSV文件
input_csv milvus_data.csv # 替换为你的CSV文件名
df pd.read_csv(input_csv)
# 创建新的DataFrame
output_data []
for _, row in df.iterrows():# 提取需要的字段id_value str(uuid4()) # 生成唯一IDembedding f[{,.join(row[vector].split(,))}] # 将vector转换为嵌入格式document row[text]# 生成meta信息meta_dict {page: row[page], source: row[source]}meta json.dumps(meta_dict, ensure_asciiFalse) # 首先生成正常的JSON# meta meta.replace(, \\) # 转义双引号
create_time datetime.now().strftime(%Y-%m-%d %H:%M:%S)update_time create_time # 更新时同样的时间
# 添加到输出数据output_data.append({id: id_value,embedding: embedding,document: document,meta: meta,create_time: create_time,update_time: update_time})
# 转换为DataFrame
output_df pd.DataFrame(output_data)
# 保存为CSV文件
output_csv output.csv # 输出文件名
output_df.to_csv(output_csv, indexFalse, quoting1) # quoting1用于确保字符串加引号
print(f转换完成已保存为 {output_csv})
数据整理完成后就可以导入数据到 TiDB Cloud 中了。 五、导入数据到 TiDB Cloud TiDB Cloud 提供三种导入方式 本次使用本地上传的方式。 小于 50MiB 的 csv文件可以使用第一种上传本地文件的方式如果文件大于 50 MiB可以使用脚本将文件拆分为多个较小的文件再上传 上传文件后选择已经创建好的库和表点击 define table 调整好对应关系点击 start import 即可 更多的导入方式可以查看文档 Migration and Import Overview 六、验证结果 数据成功导入之后就需要开始验证数据了。于是我修改了 RAG应用的代码分别从 Milvus 和 TiDB 中读取向量数据使用同一个问题来让大模型返回答案查看答案是否类似。 from langchain_community.llms import Ollama
from langchain.callbacks.manager import CallbackManager
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain import hub
from langchain.chains import RetrievalQA
from langchain.vectorstores.milvus import Milvus
from langchain_community.embeddings.jina import JinaEmbeddings
from langchain_community.vectorstores import TiDBVectorStore
import os
llm Ollama(
modelllama3,
callback_managerCallbackManager([StreamingStdOutCallbackHandler()]
),
stop[|eot_id|],
)
embeddings JinaEmbeddings(jina_api_keyxxxx, model_namejina-embeddings-v2-base-zh)
vector_store_milvus Milvus(embedding_functionembeddings,connection_args{uri: http://10.3.xx.xx:19530},
)
TIDB_CONN_STRmysqlpymysql://xxxx.root:passwordhost:4000/test?ssl_ca/Downloads/isrgrootx1.pemssl_verify_certtruessl_verify_identitytrue
vector_store_tidb TiDBVectorStore(connection_stringTIDB_CONN_STR,embedding_functionembeddings,table_namelangchain_vector,
)
os.environ[LANGCHAIN_API_KEY] xxxx
query input(\nQuery: )
prompt hub.pull(rlm/rag-prompt)
qa_chain RetrievalQA.from_chain_type(llm, retrievervector_store_milvus.as_retriever(), chain_type_kwargs{prompt: prompt}
)
print(milvus)
result qa_chain({query: query})
print(\n--------------------------------------)
print(tidb)
qa_chain RetrievalQA.from_chain_type(llm, retrievervector_store_tidb.as_retriever(), chain_type_kwargs{prompt: prompt}
)
result qa_chain({query: query})其中 TiDB 的连接串可以直接从 TiDB Cloud 中获取 向 RAG 应用提问之后查看回答发现 Milvus 和 TiDB 的回答基本一致说明向量迁移是成功的。还可以更进一步比对数据条数如果一致那么迁移应该已经成功没有丢失数据。 RAG 应用的执行结果如下图 七、总结 不同数据库之间的数据迁移本质上是将数据转换为所有数据库都能识别的通用格式向量数据也不例外。本次迁移与传统的关系型数据库迁移有所不同尽管 RAG 应用使用了 LangChain但 LangChain 针对不同的数据库创建的表结构和数据格式是不同的因此需要对数据和表结构进行额外的整理才能顺利将数据迁移至目标数据库。值得庆幸的是TiDB Cloud 提供了多种便捷的数据导入方式使迁移过程相对简单。
