ERNIE-Speed-Pro-128K结合向量数据库构建知识库案例

大模型开发/技术交流

大模型推理
AI加速器线上加速营
文心大模型

9月20日2529看过

本文使用新发布的ERNIE-Speed-Pro-128K及百度向量数据库VDB构建“骑行”知识库的后端API，它作为“骑行大老”它可以帮助骑行者提升骑行技能，后续可以再丰富其骑行方面能力，如结合AppBuilder的自定义组件封装为应用（或者代码直接）使用百度地图提供的骑行路线规划能力。

技术路线：使用Flask构建API，在Linux服务器上启动3个worker运行，通过Postman本地调用API进行测试后端的VDB和大模型进行回答。

先介绍一下大模型ERNIE-Speed-Pro-128K和向量数据库VectorDB（以下简称VDB）。

ERNIE-Speed-Pro-128K

ERNIE-Speed-Pro-128K是2024年8月30日发布的初始版本，支持128K上下文长度，效果比ERNIE-Speed-128K更优。通用能力优异，适合作为基座模型进行精调，更好地处理特定场景问题，同时具备极佳的推理性能。官方文档：https://cloud.baidu.com/doc/WENXINWORKSHOP/s/Bm0hnziyn，使用前需要到千帆平台开通该大模型，控制台地址：https://console.bce.baidu.com/qianfan/chargemanage/list

向量数据库 VectorDB

百度智能云向量数据库 VectorDB 是一款可支持百亿向量的数据库服务， QPS 性能优于开源 10 倍，提供简洁的访问接口和全面的检索能力，同时可兼容丰富的上下游生态，满足客户在企业级知识库，图片搜索，音乐推荐，文本分类等领域落地的需求。开通/使用数据库请参考，数据库入门：https://cloud.baidu.com/doc/VDB/s/hlrsoazuf

下面代码涉及到的python模块请使用pip install -r requirements.txt进行安装。

  
  
   
   
    Flask==3.0.3
   
   
   
   
    langchain==0.3.0
   
   
   
   
    langchain_community==0.3.0
   
   
   
   
    pymochow==1.3.1
   
   
   
   
    qianfan==0.4.8.1

下面进入正题。

构建知识库

主要步骤：配置文件-->创建数据库-->创建表-->导入数据（我这里准备了docx文件）。

以下文件中的endpoint、ak、sk等信息请修改为自己的值。

配置文件config.py信息

  
  
   
   
    import os
   
   
   
   
    from pymochow.auth.bce_credentials import BceCredentials
   
   
   
   
    
   
   
   
   
    # 定义配置信息
   
   
   
   
    account = 'root'
   
   
   
   
    api_key = 'vdb$20uajhjbyaj'
   
   
   
   
    endpoint = 'http://180.76.55.*:5287'  #形如'http://127.0.0.1:5287'
   
   
   
   
    
   
   
   
   
    # 初始化BceCredentials对象
   
   
   
   
    credentials = BceCredentials(account, api_key)
   
   
   
   
    
   
   
   
   
    # 设置千帆AI平台的安全认证信息（AK/SK），通过环境变量
   
   
   
   
    # 注意替换以下参数为您的Access Key和Secret Key
   
   
   
   
    os.environ["QIANFAN_ACCESS_KEY"] = "ALTAK3zaHaeQ6dMfRwj******"
   
   
   
   
    os.environ["QIANFAN_SECRET_KEY"] = "6b79407e8fd243******0ffdc66ffac"

创建VDB数据库代码

  
  
   
   
    import pymochow
   
   
   
   
    from pymochow.configuration import Configuration
   
   
   
   
    import config  # 导入配置文件
   
   
   
   
    from pymochow.auth.bce_credentials import BceCredentials
   
   
   
   
    
   
   
   
   
    # credentials（凭据）和 endpoint（终端点）信息。这些信息用于数据库。
   
   
   
   
    config_obj = Configuration(credentials=config.credentials, endpoint=config.endpoint)
   
   
   
   
    
   
   
   
   
    # 这个客户端是用来与数据库服务进行交互的。
   
   
   
   
    client = pymochow.MochowClient(config_obj)
   
   
   
   
    
   
   
   
   
    try:
   
   
   
   
        db = client.create_database("riding_vdb")   # 创建名称为 "riding_vdb" 的数据库
   
   
   
   
    
   
   
   
   
    except Exception as e:   # 捕获所有类型的异常
   
   
   
   
        print(f"Error: {e}")   # 打印异常信息
   
   
   
   
    
   
   
   
   
    db_list = client.list_databases()  # 获取当前客户端连接的数据库列表
   
   
   
   
    
   
   
   
   
    for db_name in db_list:
   
   
   
   
        print(db_name.database_name)  # 打印此连接下所有的数据库名称
   
   
   
   
    
   
   
   
   
    client.close() # 关闭客户端

执行代码：python 1.creat_riding_vdb.py

可以看到成功创建了riding_vdb数据库。

  
  
   
   
    [root@iZbp119o5qmwvZ RAG]# python 1.creat_riding_vdb.py 
   
   
   
   
    AppBuilderDatabase
   
   
   
   
    DocumentInsight
   
   
   
   
    guo_vector_database
   
   
   
   
    qiye_vdb
   
   
   
   
    riding_vdb

3.创建表代码

  
  
   
   
    import time
   
   
   
   
    import pymochow  # 导入pymochow库，用于操作数据库
   
   
   
   
    from pymochow.configuration import Configuration  # 用于配置客户端
   
   
   
   
    import config  # 导入配置文件，包含身份验证和终端信息
   
   
   
   
    
   
   
   
   
    # 导入pymochow模型相关的类和枚举类型
   
   
   
   
    from pymochow.model.schema import Schema, Field, VectorIndex, HNSWParams
   
   
   
   
    from pymochow.model.enum import FieldType, IndexType, MetricType, TableState
   
   
   
   
    from pymochow.model.table import Partition
   
   
   
   
    
   
   
   
   
    # 使用配置文件中的信息初始化客户端
   
   
   
   
    config_obj = Configuration(credentials=config.credentials, endpoint=config.endpoint)
   
   
   
   
    client = pymochow.MochowClient(config_obj)
   
   
   
   
    
   
   
   
   
    # 选择或创建数据库
   
   
   
   
    db = client.database("riding_vdb")
   
   
   
   
    
   
   
   
   
    # 定义数据表的字段列表
   
   
   
   
    fields = [
   
   
   
   
        # 定义 id 字段
   
   
   
   
        Field("id", FieldType.UINT64,   # 类型为 UINT64
   
   
   
   
              primary_key=True,     # 设为主键（primary_key=True）
   
   
   
   
              partition_key=True,   # 设为分区键（partition_key=True）
   
   
   
   
              auto_increment=False, # 不自增（auto_increment=False）
   
   
   
   
              not_null=True),
   
   
   
   
    
   
   
   
   
        # 定义 text 字段，类型为 STRING
   
   
   
   
        Field("text", FieldType.STRING),
   
   
   
   
    
   
   
   
   
        # 定义 metadata 字段，类型为 STRING
   
   
   
   
        Field("metadata", FieldType.STRING),
   
   
   
   
    
   
   
   
   
        # 定义 vector 字段，类型为 FLOAT_VECTOR，不能为空（not_null=True），维度为 384
   
   
   
   
        Field("vector", FieldType.FLOAT_VECTOR,
   
   
   
   
              not_null=True,
   
   
   
   
              dimension=384)
   
   
   
   
    ]
   
   
   
   
    
   
   
   
   
    # 定义数据表的索引列表
   
   
   
   
    indexes = [
   
   
   
   
        # 定义向量索引
   
   
   
   
        VectorIndex(
   
   
   
   
            index_name="vector_idx",  # 索引名称为 "vector_idx"
   
   
   
   
            field="vector",  # 索引的字段为 "vector"
   
   
   
   
            index_type=IndexType.HNSW,  # 索引类型为 HNSW
   
   
   
   
            metric_type=MetricType.L2,  # 度量类型为 L2
   
   
   
   
            params=HNSWParams(m=32, efconstruction=200)  # HNSW 索引的参数，m=32, efconstruction=200
   
   
   
   
        ),
   
   
   
   
    ]
   
   
   
   
    
   
   
   
   
    
   
   
   
   
    # 尝试创建名为 "riding_table" 的数据表
   
   
   
   
    try:
   
   
   
   
        table = db.create_table(
   
   
   
   
            table_name="riding_table",  # 数据表的名称为 "riding_table"
   
   
   
   
            replication=1,  # 若您的vdb实例为免费版,这里的参数需要设置为1; 若vdb实例为标准版,这里的参数不能超过实例数据节点的数量
   
   
   
   
            partition=Partition(partition_num=1),  # 使用单个分区，分区数为1
   
   
   
   
            schema=Schema(fields=fields, indexes=indexes)  # 使用指定的字段和索引结构定义表的模式
   
   
   
   
        )
   
   
   
   
    except Exception as e:  # 捕获所有类型的异常
   
   
   
   
        print(f"Error: {e}")  # 打印异常信息
   
   
   
   
    
   
   
   
   
    # 轮询数据表状态，直到表状态为NORMAL，表示表已准备好
   
   
   
   
    while True:
   
   
   
   
        time.sleep(2)  # 每次检查前暂停2秒，减少对服务器的压力
   
   
   
   
        table = db.describe_table("riding_table")
   
   
   
   
        if table.state == TableState.NORMAL:  # 表状态为NORMAL，跳出循环
   
   
   
   
            break
   
   
   
   
    
   
   
   
   
    # 打印数据表的详细信息
   
   
   
   
    print("table: {}".format(table.to_dict()))
   
   
   
   
    
   
   
   
   
    client.close()  # 关闭客户端连接

执行代码：python 2.creat_riding_table.py

如果建表成功会返回：

  
  
   
   
    [root@iZbp119o5q408bdvZ RAG]# python 2.creat_riding_table.py 
   
   
   
   
    table: {'database': 'riding_vdb', 'table': 'riding_table', 'description': '', 'replication': 1, 'partition': {'partitionType': <PartitionType.HASH: 'HASH'>, 'partitionNum': 1}, 'enableDynamicField': False, 'schema': {'fields': [{'fieldName': 'id', 'fieldType': 'UINT64', 'notNull': True, 'primaryKey': True, 'partitionKey': True}, {'fieldName': 'text', 'fieldType': 'STRING', 'notNull': False}, {'fieldName': 'metadata', 'fieldType': 'STRING', 'notNull': False}, {'fieldName': 'vector', 'fieldType': 'FLOAT_VECTOR', 'notNull': True, 'dimension': 384}], 'indexes': [{'indexName': 'vector_idx', 'indexType': <IndexType.HNSW: 'HNSW'>, 'field': 'vector', 'metricType': <MetricType.L2: 'L2'>, 'autoBuild': False, 'params': {'M': 32, 'efConstruction': 200}}]}, 'aliases': [], 'createTime': '2024-09-20 16:14:51', 'state': <TableState.NORMAL: 'NORMAL'>}

导入word数据到向量数据库

导入本地文档到向量数据库的代码如下：

  
  
   
   
    
   
   
   
   
    from langchain_community.document_loaders import PDFPlumberLoader,Docx2txtLoader,UnstructuredExcelLoader
   
   
   
   
    from langchain.text_splitter import RecursiveCharacterTextSplitter  # 用于文本分割
   
   
   
   
    import qianfan  # 千帆AI平台SDK
   
   
   
   
    import time  # 用于暂停执行，避免请求频率过高
   
   
   
   
    import json
   
   
   
   
    import pymochow
   
   
   
   
    import config  # 导入配置文件
   
   
   
   
    from pymochow.model.table import Row, AnnSearch, HNSWSearchParams  # 用于写入向量数据
   
   
   
   
    from pymochow.configuration import Configuration
   
   
   
   
    
   
   
   
   
    # 从用户输入中获取要导入的文件路径或名称
   
   
   
   
    path_file = input("请输入你要导入的知识库名称:")
   
   
   
   
    
   
   
   
   
    # 如果文件路径中包含 ".docx"，则使用 Docx2txtLoader 加载文档
   
   
   
   
    if ".docx" in path_file:
   
   
   
   
        loader = Docx2txtLoader(path_file)
   
   
   
   
        documents = loader.load()  # 加载文档
   
   
   
   
        print(documents[0])  # 打印加载的第一个文档内容
   
   
   
   
    
   
   
   
   
    # 如果文件路径中包含 ".xlsx"，则使用 UnstructuredExcelLoader 加载文档
   
   
   
   
    elif ".xlsx" in path_file:
   
   
   
   
        loader = UnstructuredExcelLoader(path_file, mode="elements")
   
   
   
   
        documents = loader.load()  # 加载文档
   
   
   
   
        print(documents[0])  # 打印加载的第一个文档内容
   
   
   
   
    
   
   
   
   
    # 如果文件路径中包含 ".pdf"，则使用 PDFPlumberLoader 加载文档
   
   
   
   
    elif ".pdf" in path_file:
   
   
   
   
        loader = PDFPlumberLoader(path_file)
   
   
   
   
        documents = loader.load()  # 加载文档
   
   
   
   
        print(documents[0])  # 打印加载的第一个文档内容
   
   
   
   
    
   
   
   
   
    # 如果文件类型不是支持的格式，输出导入失败信息
   
   
   
   
    else:
   
   
   
   
        print("导入失败,文件只支持 '.docx'、'.pdf'和'.xlsx'文件 ")
   
   
   
   
    
   
   
   
   
    
   
   
   
   
    # 设置文本分割器，指定分割的参数
   
   
   
   
    # chunk_size定义了每个分割块的字符数，chunk_overlap定义了块之间的重叠字符数
   
   
   
   
    # separators列表定义了用于分割的分隔符
   
   
   
   
    text_splitter = RecursiveCharacterTextSplitter(
   
   
   
   
        chunk_size=80,
   
   
   
   
        chunk_overlap=0,
   
   
   
   
        separators=["\n\n", "\n", " ", "", "。", "，"]
   
   
   
   
    )
   
   
   
   
    
   
   
   
   
    all_splits = text_splitter.split_documents(documents)  # 对文档进行分割
   
   
   
   
    
   
   
   
   
    print(all_splits[0])  # 打印分割后的第一个块内容
   
   
   
   
    
   
   
   
   
    emb = qianfan.Embedding()  # 初始化嵌入模型对象
   
   
   
   
    
   
   
   
   
    embeddings = []  # 用于存储每个文本块的嵌入向量
   
   
   
   
    
   
   
   
   
    for chunk in all_splits:  # 遍历所有分割的文本块
   
   
   
   
        # 获取文本块的嵌入向量，使用默认模型Embedding-V1
   
   
   
   
        resp = emb.do(texts=[chunk.page_content])
   
   
   
   
        embeddings.append(resp['data'][0]['embedding'])  # 将嵌入向量添加到列表中
   
   
   
   
        time.sleep(0.3)  # 暂停1秒，避免请求过于频繁
   
   
   
   
    
   
   
   
   
    print(embeddings[0])  # 打印第一个文本块的嵌入向量
   
   
   
   
    
   
   
   
   
    # 初始化一个空列表，用于存储每行的数据对象
   
   
   
   
    rows = []
   
   
   
   
    
   
   
   
   
    # 遍历所有的数据块（all_splits），逐行写入向量化数据
   
   
   
   
    for index, chunk in enumerate(all_splits):
   
   
   
   
    
   
   
   
   
        # 初始化 metadata 字符串
   
   
   
   
        metadata = "{}"
   
   
   
   
    
   
   
   
   
        # 如果数据块的 metadata 不为空，则将其转换为 JSON 格式的字符串表示
   
   
   
   
        if chunk.metadata is not None:
   
   
   
   
            metadata = json.dumps(chunk.metadata)
   
   
   
   
    
   
   
   
   
        # 创建一行数据对象（Row），包括 id、文本内容（text）、metadata 和向量（vector）
   
   
   
   
        row = Row(
   
   
   
   
            id=index,  # 行的唯一标识符为当前索引值
   
   
   
   
            text=chunk.page_content,  # 文本内容为数据块的页面内容
   
   
   
   
            metadata=metadata,  # metadata 字段为数据块的元数据转换后的 JSON 字符串
   
   
   
   
            vector=embeddings[index]  # 向量字段为当前索引处的嵌入向量
   
   
   
   
        )
   
   
   
   
    
   
   
   
   
        # 将创建的行对象添加到行列表中
   
   
   
   
        rows.append(row)
   
   
   
   
    
   
   
   
   
    # 打印第一个Row对象转换成的字典格式，以验证数据结构
   
   
   
   
    print(rows[0].to_dict()["metadata"])
   
   
   
   
    
   
   
   
   
    # 读取数据库配置文件，并且初始化连接
   
   
   
   
    config_obj = Configuration(credentials=config.credentials, endpoint=config.endpoint)
   
   
   
   
    client = pymochow.MochowClient(config_obj)
   
   
   
   
    
   
   
   
   
    # 选择或创建数据库
   
   
   
   
    db = client.database("riding_vdb")
   
   
   
   
    
   
   
   
   
    try:
   
   
   
   
        table = db.describe_table("riding_table")
   
   
   
   
        table.upsert(rows=rows) # 批量写入向量数据，一次最多支持写入1000条
   
   
   
   
        table.rebuild_index("vector_idx") # 创建向量索引，必要步骤
   
   
   
   
    
   
   
   
   
    except Exception as e:  # 捕获所有类型的异常
   
   
   
   
        print(f"Error: {e}")  # 打印异常信息

将骑行知识文档放到代码同级目录下，文件名：ridingknowledge.docx

该word文档内容：

  
  
   
   
    骑行，作为一项集健康、环保、探索与乐趣于一体的运动方式，近年来在全球范围内越来越受到人们的喜爱。无论是穿梭于繁华都市的街头巷尾，还是驰骋在风景如画的乡村小道，骑行都能带来别样的体验与收获。然而，要想安全、愉快地享受骑行之旅，掌握一些基本的骑行建议和注意事项显得尤为重要。以下是一些资深骑行爱好者分享的宝贵经验：
   
   
   
   
    一、准备工作：安全第一
   
   
   
   
    **1. 选择合适的装备：安全头盔是骑行时必不可少的装备，它能有效减少头部受伤的风险。同时，根据个人需求选择合适的骑行服、手套、骑行鞋以及护目镜等，不仅能提升舒适度，还能在一定程度上保护身体。
   
   
   
   
    **2. 检查车辆状况：每次骑行前，务必对自行车进行全面检查，包括刹车系统、轮胎气压、链条、变速器及车灯等，确保所有部件处于良好状态，避免途中出现故障。
   
   
   
   
    **3. 规划路线：提前规划好骑行路线，了解路况、交通状况及沿途补给点，避免进入未知或危险区域。对于长途骑行，还应考虑住宿和餐饮安排。
   
   
   
   
    二、骑行技巧与策略
   
   
   
   
    **1. 保持正确的骑行姿势：身体稍微前倾，双臂自然弯曲，双手轻握车把，双脚平稳踏动，避免长时间保持同一姿势，以减少疲劳和不适。
   
   
   
   
    **2. 掌握变速技巧：合理利用变速器，根据路况和体力状况调整齿轮比，上坡时减轻阻力，下坡时控制速度，确保骑行更加顺畅。
   
   
   
   
    **3. 注意交通安全：骑行时务必遵守交通规则，佩戴反光背心或装备，提高夜间或光线不佳时的可见性。在路口、转弯处提前减速，注意观察四周，避免与机动车发生碰撞。
   
   
   
   
    **4. 合理分配体力：长途骑行时，要合理分配体力，避免一开始就过度消耗，造成后续体力不支。适时休息，补充能量和水分，保持良好的骑行状态。
   
   
   
   
    三、健康与防护
   
   
   
   
    **1. 防晒与补水：骑行过程中，紫外线强烈，务必做好防晒措施，如涂抹防晒霜、佩戴遮阳帽和太阳镜。同时，要定时补充水分，防止脱水。
   
   
   
   
    **2. 预防运动伤害：骑行前做好热身运动，拉伸肌肉和关节，减少受伤风险。骑行过程中注意姿势正确，避免长时间用力不均导致的肌肉劳损或关节损伤。
   
   
   
   
    **3. 应对突发状况：携带急救包，包括创可贴、消毒液、绷带等基本医疗用品，以备不时之需。了解基本的急救知识，如止血、包扎等，以便在紧急情况下自救或互救。
   
   
   
   
    四、享受骑行乐趣
   
   
   
   
    **1. 探索未知：骑行不仅仅是一种运动方式，更是一种探索世界的途径。不妨尝试不同的骑行路线，发现新的风景和故事，让每一次骑行都成为一次难忘的旅程。
   
   
   
   
    **2. 分享与交流：加入骑行俱乐部或社群，与志同道合的骑友分享骑行经验，交流心得，不仅能增进友谊，还能获得更多实用的骑行建议和技巧。
   
   
   
   
    **3. 保持心态平和：骑行时保持平和的心态，享受沿途的风景和骑行的过程，而不是过分追求速度和距离。让骑行成为一种放松身心、享受生活的方式。
   
   
   
   
    总之，骑行是一项充满乐趣和挑战的运动，只要做好充分的准备，掌握正确的骑行技巧和策略，注重健康与防护，就能安全、愉快地享受骑行的乐趣。希望每一位骑行爱好者都能在骑行的路上，找到属于自己的风景和故事。

本代码不仅支持docx，还支持'.pdf'和'.xlsx'文件导入。

执行如下代码，手动输入文件名进行私有知识数据导入。

  
  
   
   
    root@iZbp119o5qt408bdvZ RAG]# python 3.insert_data.py 
   
   
   
   
    请输入你要导入的知识库名称:ridingknowledge.docx

如果看到如下输出证明导入成功：

5.启动API服务

使用了ERNIE-Speed-Pro-128K大模型，该模型在推理速度方面非常有优势，并且支持124k输入、4k输出的超长上下文，价格跌破历史！通用效果优异！更适合精调基座模型！

下面的代码基于Python的Flask框架提供API服务，在Linux服务器上使用gunicorn启动3个worker，如果在windows开发环境不能使用gunicorn，直接python命令启动即可。

  
  
   
   
    import config  # 导入配置模块
   
   
   
   
    from flask import Flask, request, jsonify,Response
   
   
   
   
    
   
   
   
   
    from langchain_community.vectorstores import BaiduVectorDB  # 导入百度向量数据库相关模块
   
   
   
   
    from langchain_community.vectorstores.baiduvectordb import ConnectionParams, TableParams  # 导入数据库连接参数和表参数相关模块
   
   
   
   
    from langchain_community.embeddings import QianfanEmbeddingsEndpoint  # 导入嵌入模型相关模块
   
   
   
   
    from langchain_community.chat_models import QianfanChatEndpoint  # 导入聊天模型相关模块
   
   
   
   
    from langchain.chains import RetrievalQA  # 导入检索问答模块
   
   
   
   
    
   
   
   
   
    app = Flask(__name__)
   
   
   
   
    
   
   
   
   
    # 初始化向量嵌入和连接参数
   
   
   
   
    embeddings = QianfanEmbeddingsEndpoint()  # 初始化Qianfan嵌入模型
   
   
   
   
    
   
   
   
   
    conn_params = ConnectionParams(
   
   
   
   
        endpoint=config.endpoint,  # 设置数据库连接端点
   
   
   
   
        account=config.account,  # 设置数据库连接账号
   
   
   
   
        api_key=config.api_key  # 设置数据库连接API密钥
   
   
   
   
    )
   
   
   
   
    
   
   
   
   
    # 初始化百度云向量数据库
   
   
   
   
    vector_db = BaiduVectorDB(
   
   
   
   
        embedding=embeddings,  # 使用Qianfan嵌入模型进行初始化
   
   
   
   
        connection_params=conn_params,  # 使用设定的连接参数进行初始化
   
   
   
   
        table_params=TableParams(384),  # 设置表参数，表维度为384
   
   
   
   
        database_name="riding_vdb",  # 数据库名称为"riding_vdb"
   
   
   
   
        table_name="riding_table",  # 表名称为"riding_table"
   
   
   
   
        drop_old=False,  # 不删除旧表，保留现有数据
   
   
   
   
    )
   
   
   
   
    
   
   
   
   
    # 初始化检索器和对话模型
   
   
   
   
    retriever = vector_db.as_retriever(search_type="similarity")  # 初始化检索器，设定检索类型为相似度
   
   
   
   
    
   
   
   
   
    qianfan_chat_model = QianfanChatEndpoint(
   
   
   
   
        model="ERNIE-Speed-Pro-128K",  # 使用"ERNIE-Speed-Pro-128K"模型进行初始化
   
   
   
   
        temperature=0.1,  # 设定温度参数为0.1，用于生成回复的多样性
   
   
   
   
        qianfan_ak='oKvwOxYK2Gk*********RNvgx',  # 设置Qianfan的访问密钥
   
   
   
   
        qianfan_sk='ru5ojewn46651oP2qq*******oUdRwt'  # 设置Qianfan的安全密钥
   
   
   
   
    )
   
   
   
   
    
   
   
   
   
    # 初始化问答模块
   
   
   
   
    qa = RetrievalQA.from_chain_type(
   
   
   
   
        llm=qianfan_chat_model,  # 设定语言模型为Qianfan聊天模型
   
   
   
   
        #chain_type="refine",  # 设定问答链类型为"refine"
   
   
   
   
        retriever=retriever,  # 设定检索器为之前初始化的检索器
   
   
   
   
        return_source_documents=True  # 返回相关文档以支持答案生成
   
   
   
   
    )
   
   
   
   
    
   
   
   
   
    @app.route('/chatwithVDB', endpoint='chatwithVDB', methods=['POST'])
   
   
   
   
    def chat_with_VDB():
   
   
   
   
        # 接收用户输入的问题
   
   
   
   
        query = request.args.get('user_input')
   
   
   
   
    
   
   
   
   
        # 处理用户问题并获取答案和相关文档
   
   
   
   
        res = qa(query)  # 使用问答模块处理用户问题
   
   
   
   
        answer, docs = res['result'], res['source_documents']  # 获取问答模块返回的答案和相关文档列表
   
   
   
   
    
   
   
   
   
        # 打印用户提出的问题和系统给出的回答
   
   
   
   
        print("\n\n> Question:")
   
   
   
   
        print(res)  # 打印用户输入的问题
   
   
   
   
        print("\n> Answer:")
   
   
   
   
        print(answer)  # 打印系统生成的回答
   
   
   
   
            
   
   
   
   
        return jsonify({'message': answer}), 200
   
   
   
   
      
   
   
   
   
    import os
   
   
   
   
    import qianfan
   
   
   
   
    
   
   
   
   
    # 通过环境变量初始化认证信息
   
   
   
   
    # 推荐】使用安全认证AK/SK鉴权
   
   
   
   
    # 替换下列示例中参数，安全认证Access Key替换your_iam_ak，Secret Key替换your_iam_sk，如何获取请查看https://cloud.baidu.com/doc/Reference/s/9jwvz2egb
   
   
   
   
    os.environ["QIANFAN_ACCESS_KEY"] = "ALTAKqk85NxpKj******68BcS3"
   
   
   
   
    os.environ["QIANFAN_SECRET_KEY"] = "3416b7f7271b4*******25c61757ae50"
   
   
   
   
    
   
   
   
   
    chat_comp = qianfan.ChatCompletion()
   
   
   
   
    @app.route('/chat_with_speedpro', endpoint='chat_with_speedpro', methods=['POST'])
   
   
   
   
    def chat_with_qianfanSDK():
   
   
   
   
        user_input = request.args.get('user_input')
   
   
   
   
        resp = chat_comp.do(model="ERNIE-Speed-Pro-128K", messages=[{
   
   
   
   
            "role": "user",
   
   
   
   
            "content": user_input
   
   
   
   
        }], stream=True)
   
   
   
   
        
   
   
   
   
        def sse_output():
   
   
   
   
            for r in resp:
   
   
   
   
                yield f"data: {r['body']['result']}\n\n"  # SSE 要求每条消息后有两个换行符  
   
   
   
   
        
   
   
   
   
        return Response(sse_output(), mimetype='text/event-stream')
   
   
   
   
    
   
   
   
   
    if __name__ == '__main__':
   
   
   
   
        app.run(debug=True)

在Linux服务器上启动的命令如下:

  
  
   
   
    # 服务工作进程数
   
   
   
   
    SERVER_WORKER_AMOUNT=3
   
   
   
   
    # 服务工作进程启动方式
   
   
   
   
    SERVER_WORKER_CLASS=gevent
   
   
   
   
    # 服务超时时间
   
   
   
   
    GUNICORN_TIMEOUT=60
   
   
   
   
    
   
   
   
   
    gunicorn \
   
   
   
   
      --bind "0.0.0.0:5000" \
   
   
   
   
      --workers ${SERVER_WORKER_AMOUNT} \
   
   
   
   
      --worker-class ${SERVER_WORKER_CLASS} \
   
   
   
   
      --timeout ${GUNICORN_TIMEOUT} \
   
   
   
   
      "RAG_API:app"

启动后的控制台输出3个worker并且没有输出异常证明启动正常。

  
  
   
   
    [2024-09-20 16:26:46 +0800] [1391165] [INFO] Starting gunicorn 23.0.0
   
   
   
   
    [2024-09-20 16:26:46 +0800] [1391165] [INFO] Listening at: http://0.0.0.0:5000 (1391165)
   
   
   
   
    [2024-09-20 16:26:46 +0800] [1391165] [INFO] Using worker: gevent
   
   
   
   
    [2024-09-20 16:26:46 +0800] [1391166] [INFO] Booting worker with pid: 1391166
   
   
   
   
    [2024-09-20 16:26:46 +0800] [1391167] [INFO] Booting worker with pid: 1391167
   
   
   
   
    [2024-09-20 16:26:46 +0800] [1391168] [INFO] Booting worker with pid: 1391168