课程讲解：

7. 基于DeepSeek和智谱清言实现RAG问答_哔哩哔哩_bilibilihttps://www.bilibili.com/video/BV1gsNaeyE3b/

 Ads：大模型中转平台，一个key调用国内外四十家公司上千大模型，相应快速，价格优惠，点击链接注册使用

原始数据来源于本课程的课件 RAG.pdf 文件，需要Python解析pdf文件，主要使用智谱清言的 embedding-3 将文本转向量化，存储到 ChromaDB向量数据库中，将向量查询的结果投喂给DeepSeek的deepseek-chat模型，由它根据向量查询结果来回答用户提问。

.env文件

OPENAI_API_KEY=sk-a6******9d
OPENAI_BASE_URL=https://api.deepseek.com/v1
OPENAI_MODEL=deepseek-chat
ZHIPU_BASE_URL=https://open.bigmodel.cn/api/paas/v4/
ZHIPU_API_KEY=4923c4dae*******lTGEx7
EMBEDDING_MODEL=embedding-3

启动向量数据库

chroma run --host 127.0.0.1 --port 5333 --path db_data

初始化

from openai import OpenAI
from dotenv import load_dotenv
import os
import chromadb
from chromadb.config import Settings

load_dotenv()

# 将 api_key 传入到 openAI 当中
client = OpenAI()
ai_model = os.getenv("OPENAI_MODEL")
embed_model = os.getenv("EMBEDDING_MODEL")

from pdfminer.high_level import extract_pages
from pdfminer.layout import LTTextContainer

def get_completion(prompt, model=ai_model):
    """封装 openai 接口"""
    messages = [{"role": "user", "content": prompt}]
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=0,
    )
    return response.choices[0].message.content

def build_prompt(prompt_template, **kwargs):
    '''将 Prompt 模板赋值'''
    prompt = prompt_template
    for k, v in kwargs.items():
        if isinstance(v, str):
            val = v
        elif isinstance(v, list) and all(isinstance(elem, str) for elem in v):
            val = '\n'.join(v)
        else:
            val = str(v)
        prompt = prompt.replace(f"__{k.upper()}__", val)
    return prompt

提示词

prompt_template = """
你是一个问答机器人。
你的任务是根据下述给定的已知信息回答用户问题。
确保你的回复完全依据下述已知信息。不要编造答案。
如果下述已知信息不足以回答用户的问题，请直接回复"我无法回答您的问题"。
已知信息:
__INFO__
用户问：
__QUERY__
请用中文回答用户问题。
"""

解析pdf文件

def extract_text_from_pdf(filename, page_numbers=None, min_line_length=1):
    '''从 PDF 文件中（按指定页码）提取文字'''
    paragraphs = []
    ids = []
    id_counter = 0
    buffer = ''
    full_text = ''
    # 提取全部文本
    for i, page_layout in enumerate(extract_pages(filename)):
        # 如果指定了页码范围，跳过范围外的页
        if page_numbers is not None and i not in page_numbers:
            continue
        for element in page_layout:
            if isinstance(element, LTTextContainer):
                full_text += element.get_text() + '\n'
    # 按空行分隔，将文本重新组织成段落
    lines = full_text.split('\n')
    for text in lines:
        if len(text) >= min_line_length:
            buffer += (' ' + text) if not text.endswith('-') else text.strip('-')
        elif buffer:
            paragraphs.append(buffer)
            ids.append(f"id{id_counter}")
            buffer = ''
            id_counter += 1
    if buffer:
        paragraphs.append(buffer)
        ids.append(f"id{id_counter}")

    results = {"documents": paragraphs, "ids": ids}
    return results


paragraphs = extract_text_from_pdf("RAG.pdf", page_numbers=[4, 5],
                                   min_line_length=10)

文本转向量函数

def get_embeddings(texts, model=embed_model):
    embedClient = OpenAI(
        api_key=os.getenv("ZHIPU_API_KEY"),
        base_url=os.getenv("ZHIPU_BASE_URL")
    )
    data = embedClient.embeddings.create(input=texts, model=model).data
    return [x.embedding for x in data]

向量数据库连接器

class MyVectorDBConnector:
    def __init__(self, collection_name, embedding_fn):
        chroma_client = chromadb.HttpClient(
            host="127.0.0.1",
            port=5333,
        )
        self.embedding_fn = embedding_fn
        self.collection = chroma_client.get_or_create_collection(name=collection_name)
        self.documents = []

    def add_documents(self, contents):
        self.collection.add(
            embeddings=self.embedding_fn(contents.get("documents")),
            documents=contents.get("documents"),
            ids=contents.get("ids"),
        )

    def search(self, query, top_n):
        results = self.collection.query(
            query_embeddings=self.embedding_fn([query]),
            n_results=top_n
        )
        return results

# 创建一个向量数据库对象
vector_db = MyVectorDBConnector("demo", get_embeddings)
# 向向量数据库中添加文档
vector_db.add_documents(paragraphs)

RAG机器人

class RAG_Bot:
    def __init__(self, vector_db, llm_api, n_results=2):
        self.vector_db = vector_db
        self.llm_api = llm_api
        self.n_results = n_results

    def chat(self, user_query):
        # 1. 检索
        search_results = self.vector_db.search(user_query, self.n_results)
        # 2. 构建 Prompt
        prompt = build_prompt(prompt_template, info=search_results['documents'][0], query=user_query)
        # 3. 调用 LLM
        response = self.llm_api(prompt)
        return response


# 创建一个 RAG 机器人
bot = RAG_Bot(
    vector_db,
    llm_api=get_completion
)

最后提问

# 用户查询
user_query = "常用的向量数据库有哪些？"
response = bot.chat(user_query)
print(response)

运行效果：

pdf文档下载

https://cloud.189.cn/t/Yzq2mmJ7J77v（访问码：k9sy）