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的安装：

Shell

1pip install pymochow

卸载

您可以在命令行中执行如下命令完成Python SDK的卸载：

Shell

1pip uninstall pymochow

将文档导入向量数据库

                Python
                
            

                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
)
            

对知识库进行向量检索

                Python
                
            

                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
)
            

对知识库进行全文检索

                Python
                
            

                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
)
            

对知识库进行混合检索

                Python
                
            

                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
)
            

VectorDB CLI + VectorDB Lite 本地高效调试向量数据库

生态方案

百度智能云

向量数据库 VectorDB