基于PaddleNLP使用DeepSeek-R1搭建智能体

最近在学习DeepSeek，找到了PaddleNLP星河社区大模型，跟着敲写了一遍。内容来源：DeepSeek实战训练营：从云端模型部署到应用开发 - 飞桨AI Studio星河社区-人工智能学习与实训社区

本项目基于 Langchain 框架，结合星河社区大模型 API 提供的词嵌入模型和 Milvus 向量服务器搭建向量知识库，并使用 PaddleNLP 的 DeepSeek-R1 模型搭建杭创赛解说智能体，旨在测试模型性能并帮助参赛者更好地了解比赛。

来探索测试PaddleNLP中DeepSeek-R1所搭建的智能体使用RAG向量知识库的效果，测试效果很不错

运行环境Tesla V00 32G，飞桨星河社区每日运行项目就送8算力点，所以大家可以放心运行哈

构建流程

环境配置

PaddleNLP环境配置（DeepSeek-R1）

飞桨官方已经为大家配置好在V100和A100环境下的依赖环境，大家仅需使用解压即可使用。同时飞桨官方提供了环境安装命令，方便复现安装，本部分操作主要完成了以下任务：

自动卸载已安装的 PaddleNLP 库，确保环境干净，避免版本冲突。

解压并安装了特定版本（2025年1月24日的 develop 分支）的 PaddleNLP，同时创建了外部库目录，用于存放安装的外部依赖，以防止重启后库被清空。

安装 paddlenlp_ops 的 CUDA 算子，以提高 GPU 的利用率和运行速度。

In [1]

# 自动确认并卸载已安装的 paddlenlp 库
!pip uninstall paddlenlp -y
# 解压安装PaddleNLP，文件来自PaddleNLP GitHub仓库 develop分支（2025年1月24日）
!tar -xf PaddleNLP-20250124-1801.tgz
# 创建一个名为 'external-libraries' 的目录，用于存放安装的外部库（星河平台默认安装库重启后会被清空，所以需要将安装的库放到外部目录中）
# 创建环境依赖库，针对不同的GPU环境，请使用者自行切换
# !mkdir -p /home/aistudio/external-libraries
!tar zxf external-libraries.tgz

# !!! 依赖文件已经安装在external-libraries中，无需再次安装, 选择性安装paddlenlp_ops算子即可 !!!
# 安装PaddleNLP
# !pip install --pre --upgrade paddlenlp==3.0.0b3.post20250123 -f https://www.paddlepaddle.org.cn/whl/paddlenlp.html -t /home/aistudio/external-libraries
# 【选择性安装】安装paddlenlp_ops cuda算子, 提高GPU利用率，加快运行速度
!cd /home/aistudio/PaddleNLP/csrc && python /home/aistudio/PaddleNLP/csrc/setup_cuda.py install

知识库构建环境配置（Langchian、milvus、ernie等）

本部分操作安装了多个Python库，包括langchain-community、langchain、tiktoken、langchain_openai、unstructured、erniebot-agent、openai、milvus客户端和docx2txt，

用于支持自然语言处理、文档处理和向量数据库操作等功能。

In [2]

!pip install langchain-community --user
!pip install langchain --user
!pip install tiktoken --user
!pip install langchain_openai --user
!pip install unstructured --user
!pip install erniebot-agent langchain --user
!pip install openai
!pip install "milvus[client]" --user
!pip install docx2txt --user

In [3]

# 将自定义的外部库目录添加到系统路径中，以便Python能够导入这些库中的模块
import sys 
# 添加包含已安装外部库的目录
sys.path.insert(0, '/home/aistudio/external-libraries')
# 添加PaddleNLP项目的根目录到系统路径，以便可以导入其模块和包
sys.path.insert(0, '/home/aistudio/PaddleNLP')
import paddle
from llm.predict.predictor import PredictorArgument, ModelArgument, create_predictor
from paddlenlp.utils import is_paddlenlp_ops_available

predictor_args = PredictorArgument(
    model_name_or_path="deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B",
    src_length=1024,
    min_length=32,
    max_length=512,
    top_k=0,
    top_p=0.7,
    temperature=0.95,
    repetition_penalty=1.0,
    dtype="float16",
    inference_model= True if is_paddlenlp_ops_available() else False,
)

paddle.set_default_dtype(predictor_args.dtype)
predictor = create_predictor(predictor_args, ModelArgument())

In [4]

input_text = "解释一下温故而知新"

print(predictor.predict(input_text))

RAG向量知识库搭建

文档加载和切片

本代码片段使用 langchain_community 提供的文档加载器和文本分割器，实现了对指定文件的加载和内容分割。

首先根据文件扩展名（.pdf、.txt、.docx）选择合适的加载器加载文档内容，随后使用 RecursiveCharacterTextSplitter 将文档内容分割成固定大小的片段，以便后续处理。

代码成功加载了位于 /home/aistudio/杭创赛收集信息.docx 的文档内容，并将其分割成多个片段，每个片段大小为200字符，重叠部分为50字符。

还打印了加载和分割的结果，包括文档内容、切片数量和第11个切片的内容。

In [9]

!pip install langchain
from langchain.document_loaders import PyPDFLoader,Docx2txtLoader,TextLoader

# 加载文档
file_name='/home/aistudio/ttt.txt'

if file_name.endswith(".pdf"):
    loader = PyPDFLoader(file_name)
elif file_name.endswith(".txt"):
    loader = TextLoader(file_name)
elif file_name.endswith(".docx"):
    loader = Docx2txtLoader(file_name)
else:
    raise BizException("目前只支持pdf文件与txt、docx文件")
data=loader.load()
print("文档加载成功")
print(data)
print("------------------------------------------分割线--------------------------------------------")
from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter=RecursiveCharacterTextSplitter(
    chunk_size=200,
    chunk_overlap=50,
    length_function=len
)


pages=loader.load_and_split(text_splitter=text_splitter)
print("文档切片成功")
print("切片数量：",len(pages))
print(pages[10].page_content)
print("------------------------------------------分割线--------------------------------------------")

向量数据库构建

词嵌入函数定义

词嵌入部分是为了将文本转换为向量形式，向量形式可便于对于文档的检索和搜寻，这里使用最新的ERNIE SDK进行词嵌入函数的定义，可将文本向量化

在自然语言处理（NLP）中，向量知识库可以将文本转换为数值形式的向量，这使得机器可以处理和比较文本数据，嵌入向量能够捕捉文本的语义信息，使得相似的词汇或句子在向量空间中更接近。

本部分用于获取每段文本，将前面切片好后的文档进行整理，全部至于sections列表内，sections是一个二维列表，用于存储数据，便于后续的文本向量化转换

In [10]

import os
from openai import OpenAI
access_token = ""
client_ernie = OpenAI(
     api_key="",  # 含有 AI Studio 访问令牌的环境变量，https://aistudio.baidu.com/account/accessToken,
     base_url="https://aistudio.baidu.com/llm/lmapi/v3",  # aistudio 大模型 api 服务域名
)
def ernie_embedding(text):
    embeddings = client_ernie.embeddings.create(
        model="embedding-v1",
        input=[text])
    all_embeddings = [[float(val) for val in embedding.embedding] for embedding in embeddings.data]
    return all_embeddings
text= pages[10].page_content
re=ernie_embedding(text)
if isinstance(re, list):
    # 假设re是一个嵌套列表，计算其维度
    dimensions = [len(re)] + [len(sublist) for sublist in re]
    print(dimensions)
    print(re)
else:
    # 如果re不是列表，直接打印shape
    print(re.shape)

# 用于存储每个部分的内容
sections = []

# 遍历文档中的段落
for page in pages:
    tt=page.page_content
    text = tt.strip()
    sections.append(text)
print(len(sections))
print(sections[11])

向量数据库构建

主要先进性初始化和配置 Milvus 向量数据库，用于存储和检索向量化的文本数据，首先尝试启动 Milvus 服务器，并连接到本地运行的 Milvus 服务。

接着定义了一个数据集合（Collection）的模式（Schema），包括字段 answer_id（主键）、answer（文本内容）和 answer_vector（向量形式），并创建了一个名为 qadb 的数据集合。

在定义数据集合的模式后，为 answer_vector 字段创建了一个索引，使用了 IVF_FLAT 索引类型和 L2 距离度量，参数 nlist 设置为 1024。

这种索引配置适用于高效的向量检索任务，能够快速找到与查询向量最相似的记录。代码还查询并打印了索引构建的进度信息，确认索引构建状态。

最后加载 qadb 数据集合，确保其处于可查询状态，通过多次调用 collection.load()，确保数据集合被正确加载到内存中，以便后续进行高效的向量相似性搜索操作。

from pymilvus import utility, Collection
from pymilvus import CollectionSchema, FieldSchema, DataType
from milvus import default_server
from pymilvus import connections

try:
    default_server.start()
except:
    default_server.cleanup()
    default_server.start()
    
# 尝试连接到 Milvus 服务器
try:
    connections.connect(host='127.0.0.1', port=default_server.listen_port)
    print(f"成功连接到 Milvus 服务器，端口为：{default_server.listen_port}")
except Exception as e:
    print(f"连接失败：{e}")

# id
answer_id = FieldSchema(
    name="answer_id",
    dtype=DataType.INT64,
    is_primary=True,
    auto_id=True
)
# 答案
answer = FieldSchema(
    name="answer",
    dtype=DataType.VARCHAR,
    max_length=1024,
)
# 答案向量形式
answer_vector = FieldSchema(
    name="answer_vector",
    dtype=DataType.FLOAT_VECTOR,
    dim=384
)
# 定义一个数据集合（Collection）的模式（Schema）-创建一个表
schema = CollectionSchema(
    fields=[answer_id, answer, answer_vector],
    description="vector data"
)
# 表名
collection_name = "qadb"
# 连接数据库
Collection(
    name=collection_name,
    schema=schema,
    using='default',
    shards_num=2
)
collection = Collection("qadb")

index_params = {
    "metric_type":"L2", # COSINE
    "index_type":"IVF_FLAT",
    "params":{"nlist":1024}
}

collection.create_index(
    field_name="answer_vector", 
    index_params=index_params
)

progress_info = utility.index_building_progress("qadb")
# 获取并打印索引构建进度 {'total_rows': 0, 'indexed_rows': 0, 'pending_index_rows': 0}
# total_rows：集合中总行数（或向量数）为0。这表示集合中没有任何数据，或者数据尚未被成功插入到集合中。
# indexed_rows：已成功索引的行数（或向量数）为0。这表示还没有数据被索引。
# pending_index_rows：待索引的行数（或向量数）为0。这表示没有数据正在等待被索引。

print(progress_info)  # 打印进度信息
collection.load() 
collection = Collection("qadb")
collection.load()

数据插入知识库

接着进行信息的插入，这部分通过循环形式，遍历前面处理好的sections数据列表，将每条数据进行向量转换，文本数据和向量数据均放于data列表内

再通过insert形式进行数据的插入，通过result.insert_count属性查看是否插入成功，如果正常插入这里result.insert_count应该是1

In [5]

import time
from pymilvus import Collection

i = 1
# 打开一个文件用于记录错误
with open('error.log', 'w') as error_log:
    for rly in sections[i:]:
        try:
            print(f"这里是第{i}段：")
            rlyEmbedding = ernie_embedding(rly)  # 假设这是一个将文本转换为向量的函数
            print(len(rlyEmbedding))
            data = [
                [rly],  # 文本数据
                rlyEmbedding  # 对应的向量数据
            ]
            print(data)
            result = collection.insert(data)  # 插入数据并获取结果
            print(result)
            # 检查插入结果
            if result.insert_count > 0:
                print(f"成功插入 {result.insert_count} 条数据")
            else:
                print("插入操作失败，没有成功插入数据！！！")
            i += 1
            time.sleep(0.5)
        except Exception as e:
            print(f"插入操作异常：{e}")
            error_log.write(f'Index {i} failed with error: {str(e)}\n')
            continue

检验是否构建成功

主要进行两部分检验，一个是检验数据是否插入，另外一个是知识库查询检验，这里的插入检验貌似会有延迟，前面导入数据后这里如果响应慢不要紧，可以先进行下面的知识库查询校验，如果有数据可查询到代表无问题

这段代码首先使用MilvusClient连接到Milvus服务器，并获取名为qadb的集合的统计信息，然后打印出集合中已插入的总数据量。

接着定义一个函数get_collection_total_entities，该函数通过访问集合的num_entities属性来查询集合中的总实体数，并打印出这个数值。

这里的查询校验使用Milvus数据库进行向量搜索，通过将问题文本转换为向量（使用ernie_embedding函数），然后在集合qadb中查找与该向量最相似的向量（基于欧几里得距离），并输出最匹配的前两个答案的文本内容。

这部分代码重点是两部分，一个是查询参数的定义，另外一个是向量查询，这段代码中的search_params字典定义了在Milvus数据库中进行向量搜索时使用的参数。

In [6]

from pymilvus import MilvusClient

# 设置 Milvus 客户端
client = MilvusClient(uri="http://localhost:19530", token="root:Milvus")

# 获取集合的统计信息
stats = client.get_collection_stats(collection_name="qadb")
print(stats)

# 检查 stats 是否不为空，并且包含键 'row_count'
if stats and 'row_count' in stats:
    total_entities = stats['row_count']
    print(f"集合 'qadb' 中已经插入的总数据量为: {total_entities}")
else:
    print("注意！集合 'qadb' 不存在或为空")

def get_collection_total_entities(collection):
    """
    查询集合中的总实体数（即总数据量）

    :param collection: 向量集合对象
    :return: 总实体数
    """
    # 直接访问集合的 num_entities 属性
    collection.load()
    print(collection.describe())  # 检查集合的元数据
    stats = collection.num_entities
    print(stats)
    return stats

# 调用函数并输出集合中的总数据量
print(f"集合中的总数据量: {get_collection_total_entities(collection)}")

In [7]

question = "比赛参赛要求?"
qEmbedding = ernie_embedding(question)

search_params = {
    "metric_type": "L2",  # 使用余弦相似度  或者 COSINE
    "offset": 0, 
    "ignore_growing": False, 
    "params": {"nprobe": 5}  # 增加 nprobe 值以提高检索范围
}

# 定义要导出的集合名称
collection_name = "qadb"
collection = Collection(collection_name)

results = collection.search(
    data=qEmbedding, 
    anns_field="answer_vector", 
    param=search_params,
    limit=3,  # 查询 5 个结果
    expr=None,
    output_fields=['answer'],
    consistency_level="Strong"
)

# 打印所有返回的结果
for i, result in enumerate(results[0]):
    print(f"Result {i + 1}: {result.entity.get('answer')}")

# 获取第一个结果的 answer 字段
answer = results[0][0].entity.get('answer')
answer

基于DeepSeek-R1构建智能体

本部分代码实现一个基于 PaddleNLP 的 DeepSeek-R1-Distill-Qwen-1.5B 模型的智能问答系统，用于对用户输入的问题进行解答。

代码首先配置了模型参数，并创建了一个预测器实例。系统通过 Milvus 向量数据库检索与用户问题相关的文档内容，并结合这些内容生成简洁的回答。

代码定义了 get_data 函数，用于将用户问题嵌入向量并通过 Milvus 搜索最相关的文档片段。

检索到的内容被整合到提示词中，提示词通过 jinja2 模板生成，结合系统提示词和用户问题，引导模型生成符合要求的回答。系统提示词要求回答简洁（30字以内），并结合检索到的信息。

通过一个循环实现用户交互，用户输入问题后，系统调用 predictor.predict 方法生成回答，并将回答添加到对话历史中。

生成的回答会显示给用户，直到用户输入“结束”为止。整个流程实现了从问题检索到回答生成的完整闭环，为用户提供基于文档内容的智能解答。

In [10]

import paddle
from llm.predict.predictor import PredictorArgument, ModelArgument, create_predictor
from paddlenlp.utils import is_paddlenlp_ops_available
from jinja2 import Template
import os
import json
import asyncio

predictor_args = PredictorArgument(
    model_name_or_path="deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B",
    src_length=1024,
    min_length=32,
    max_length=512,
    top_k=0,
    top_p=0.7,
    temperature=0.95,
    repetition_penalty=1.0,
    dtype="float16",
    inference_model= True if is_paddlenlp_ops_available() else False,
)

paddle.set_default_dtype(predictor_args.dtype)
predictor = create_predictor(predictor_args, ModelArgument())

# 设置系统提示词
systemprompts = """
比赛参赛要求?

"""

# 用于查询相关信息
def get_data(user_input):
    question = user_input
    qEmbedding =  ernie_embedding(question)
    search_params = {
        "metric_type": "L2", 
        "offset": 0, 
        "ignore_growing": False, 
        "params": {"nprobe": 10}  # 提高 nprobe 的值
    }
    collection_name = "qadb"
    collection = Collection(collection_name)
    results = collection.search(
        data=qEmbedding, 
        anns_field="answer_vector", 
        param=search_params,
        limit=2,  # 查询 5 个结果
        expr=None,
        output_fields=['answer'],
        consistency_level="Strong"
    )
    # 打印所有返回的结果
    answer = "原文内容为：\n"
    for i, result in enumerate(results[0]):
        answer = answer + result.entity.get('answer')
    return answer

messages = [
    {
        "role": "system",
        "content":systemprompts,
    }
]


# 用于组合输入内容
def generate_prompt(text,user_input,system_prompt):
    # # 从 messages 中提取历史信息，包括用户和AI的所有对话
    # history_info = str(messages)
    # 设置系统提示词模板
    _DEFAULT_RESULT_ZH = """
    系统提示：{{system_prompt}}
    与用户对话相关的你的宝典信息如下：{{text}}
    本此来访人员的问题是： {{user_input}}
    """ 
    template = Template(_DEFAULT_RESULT_ZH)
    prompt = template.render(system_prompt=system_prompt,text=text, user_input=user_input)
    return prompt

In [11]

user_input = ""
while "结束" not in user_input:
    user_input = input("your：")
    # 将用户问题信息添加到messages列表中
    # 向量查询拿到信息
    text = get_data(user_input)
    # print(text)
    # 提示词组合-信息＋问题
    prompt = generate_prompt(text,user_input,systemprompts)
    # print(prompt)
    # 形成用户提问,并添加到对话中
    messages.append({"role": "user", "content": prompt})
    # 获取大模型回复
    # 调用 predictor.predict 获取结果
    ll = predictor.predict(str(messages))[0]
    
    messages.append({"role": "assistant", "content": ll})
    print(f"小助手：{ll}")
    print("\n")
else:
    print("结束1")

DeepSeek-R1单独调用测试

In [ ]

import re

# 假设 messages 是输入的提示
message = '你是什么，你的基础模型是什么？'

# 调用 predictor.predict 获取结果
ll = predictor.predict(message)[0]

# 确保 ll 是字符串类型
if not isinstance(ll, str):
    raise TypeError(f"Expected a string, but got {type(ll)}")

# 移除 <think> 标签及其内容
cleaned_result = re.sub(r'<think>.*?</think>', '', ll, flags=re.DOTALL).strip()

print(cleaned_result)

参考文献：
DeepSeek实战训练营：从云端模型部署到应用开发 - 飞桨AI Studio星河社区-人工智能学习与实训社区