AI Search 高效构建 RAG 服务
更新时间：2025-02-25
概述
向量数据库目前主要应用于 RAG 场景。在 RAG 的应用中，除了向量数据库，还需要文档管理、文档解析、分档分块、Embedding 服务（将内容向量化）、Rerank，以及 LLM 大模型等多个服务的相互协作，共同实现完整的 RAG 功能。基于这一需求，我们对各模块服务进行了封装，提供了对应的接口，方便用户自定义各模块，并与向量数据库高效结合，快速构建属于自己的 RAG 服务。
准备工作
环境准备
- 运行环境
- Python SDK工具包支持在Python 3.7及以上的环境运行。
源码下载
若您需要Python SDK源码，可从如下两处下载：
- Github地址：https://github.com/baidu/pymochow
- Gitee地址：https://gitee.com/baidu/pymochow
安装和卸载
我们推荐通过pip来安装和卸载Python SDK，方法如下：
安装
您可以在命令行执行如下命令完成Python SDK的安装：
pip install pymochow卸载
您可以在命令行中执行如下命令完成Python SDK的卸载：
pip uninstall pymochow将文档导入向量数据库
import logging
import uuid
from pymochow.ai.dochub import (
LocalDocumentHub,
DocumentHubEnv
)
from pymochow.model.document import Document
from pymochow.ai.processor import QianfanDocProcessor
from pymochow.ai.embedder import QianfanEmbedder
from pymochow.ai.pipeline import DefaultPipeline
env = DocumentHubEnv(root_path="local://your_root_path/")
doc_hub = LocalDocumentHub(env=env)
kb_id = str(uuid.uuid4())
doc = Document(
kb_id=kb_id, # 知识库ID
doc_name="test/test.pdf", # 文档名，指期望文档存储在文档管理中的相对路径，举例：test/test.pdf
file_path="./test.pdf" # 文档的本地路径
)
# 将文件纳入到文档管理中，将文件./test.pdf复制到/root_path/test/test.pdf
doc_hub.add(doc=doc)
os.environ["APPBUILDER_TOKEN"] = "your_ab_token" # 需要填入您AppBuilder的token，会调用AppBuilder的服务进行文档处理和Embedding
doc_processor = QianfanDocProcessor(maximum_page_length=300, page_overlap_length=50)
embedder = QianfanEmbedder(batch=2)
pipeline = DefaultPipeline()
# 需要在向量数据库中创建对应的库和表，用于存储文档的元数据以及文档块内容
db_name = "DocumentInsight" # 向量数据库的库名
meta_table_name = "KnowledgeBase_Meta" # 存储文档元数据的表
chunk_table_name = "KnowledgeBase_Chunk" # 存储文档块的表
# 向量数据库中创建对应表
meta_table = self.create_or_get_meta_table(db_name=db_name, table_name=meta_table_name)
chunk_table = self.create_or_get_chunk_table(db_name=db_name, table_name=chunk_table_name)
# 将文档解析、Embedding后存储到向量数据库
pipeline.ingest_doc(
doc=doc,
doc_processor=doc_processor,
embedder=embedder,
meta_table=meta_table,
chunk_table=chunk_table
)对知识库进行向量检索
from pymochow.ai.embedder import QianfanEmbedder
from pymochow.ai.pipeline import DefaultPipeline
from pymochow.model.table import (
VectorTopkSearchRequest
)
os.environ["APPBUILDER_TOKEN"] = "your ab token"
# embedder 用于对问题进行向量化
embedder = QianfanEmbedder(batch=2)
pipeline = DefaultPipeline()
db_name = "DocumentInsight"
chunk_table_name = "KnowledgeBase_Chunk"
chunk_table = self.create_or_get_chunk_table(db_name=db_name, table_name=chunk_table_name)
search_contents = ["your question"]
search_request = VectorTopkSearchRequest(
vector_field="embedding",
limit=10,
#filter="kb_id='xxx'", 填入标量过滤条件
config=VectorSearchConfig(ef=200)
)
result = pipeline.vector_search(
search_contents=search_contents,
embedder=embedder,
table=chunk_table,
search_request=search_request
)对知识库进行全文检索
from pymochow.ai.pipeline import DefaultPipeline
from pymochow.model.table import (
BM25SearchRequest
)
pipeline = DefaultPipeline()
db_name = "DocumentInsight"
chunk_table_name = "KnowledgeBase_Chunk"
chunk_table = self.create_or_get_chunk_table(db_name=db_name, table_name=chunk_table_name)
search_request = BM25SearchRequest(
index_name="content_inverted_idx",
search_text="bm25 search text",
limit=10,
#filter="kb_id='xxx'"
)
result = pipeline.bm25_search(
table=chunk_table,
search_request=search_request
)对知识库进行混合检索
from pymochow.ai.embedder import QianfanEmbedder
from pymochow.ai.pipeline import DefaultPipeline
from pymochow.model.table import (
VectorTopkSearchRequest,
BM25SearchRequest,
HybridSearchRequest
)
os.environ["APPBUILDER_TOKEN"] = "your ab token"
# embedder 用于对问题进行向量化
embedder = QianfanEmbedder(batch=2)
pipeline = DefaultPipeline()
db_name = "DocumentInsight"
chunk_table_name = "KnowledgeBase_Chunk"
chunk_table = self.create_or_get_chunk_table(db_name=db_name, table_name=chunk_table_name)
search_contents = ["your question"] # 向量检索的内容
vector_search_request = VectorTopkSearchRequest(
vector_field="embedding",
limit=10,
config=VectorSearchConfig(ef=200)
)
bm25_search_request = BM25SearchRequest(
index_name="content_inverted_idx",
search_text="bm25 search text" # 全文检索的内容
)
search_request = HybridSearchRequest(
vector_request=vector_search_request,
vector_weight=0.4, # 向量检索比重
bm25_request=bm25_search_request,
bm25_weight=0.6, # 全文检索比重
#filter="kd_id='xxx'", 填入标量过滤条件
limit=15
)
result = pipeline.hybrid_search(
search_contents=search_contents,
embedder=embedder,
table=chunk_table,
search_request=search_request
)