7. 基于DeepSeek和智谱清言实现RAG问答

课件链接：https://cloud.189.cn/t/VNvmyimY7Vna（访问码：e4cb）天翼云盘是中国电信推出的云存储服务，为用户提供跨平台的文件存储、备份、同步及分享服务，是国内领先的免费网盘，安全、可靠、稳定、快速。天翼云盘为用户守护数据资产。https://cloud.189.cn/t/VNvmyimY7Vna%EF%BC%88%E8%AE%BF%E9%97%AE%E7%A0%81%EF%BC%9Ae4cb%EF%BC%89

原始数据来源于本课程的课件 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 Settingsload_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 LTTextContainerdef 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.contentdef build_prompt(prompt_template, **kwargs):'''将 Prompt 模板赋值'''prompt = prompt_templatefor k, v in kwargs.items():if isinstance(v, str):val = velif 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 = 0buffer = ''full_text = ''# 提取全部文本for i, page_layout in enumerate(extract_pages(filename)):# 如果指定了页码范围，跳过范围外的页if page_numbers is not None and i not in page_numbers:continuefor 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 += 1if buffer:paragraphs.append(buffer)ids.append(f"id{id_counter}")results = {"documents": paragraphs, "ids": ids}return resultsparagraphs = 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).datareturn [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_fnself.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_dbself.llm_api = llm_apiself.n_results = n_resultsdef chat(self, user_query):# 1. 检索search_results = self.vector_db.search(user_query, self.n_results)# 2. 构建 Promptprompt = build_prompt(prompt_template, info=search_results['documents'][0], query=user_query)# 3. 调用 LLMresponse = 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)

运行效果：