如何当地安装向量数据库，并搭建rag - qidao123.com技术社区-IT企服评测·应用市场

graph TD
A[用户提出问题] --> B{查询本地向量数据库}
B --> C[根据语义相似度获取本地知识]
C --> D[Prompt 提示词构建]
D --> E{提示词输入大模型}
E --> F[获得最终回答]

复制代码

# 导入所需库
import chromadb # ChromaDB 向量数据库
from openai import OpenAI # OpenAI 客户端
# 初始化 OpenAI 客户端 (替换成自己的 API 信息)
client = OpenAI(
api_key="sk-i3daF6", # 替换为你的 OpenAI API Key , 这里我把自己的 API-KEY 隐藏了
base_url="https://api.xiaoai.plus/v1" # 替换为你的 API 服务端点
)
def get_embeddings(texts, model="text-embedding-ada-002"):
"""将文本转换为向量表示
Args:
texts: 需要编码的文本列表
model: 使用的嵌入模型（默认OpenAI官方推荐模型）
Returns:
包含向量数据的列表，每个元素对应输入文本的768维向量
"""
response = client.embeddings.create(
input=texts,
model=model
)
# 从响应中提取向量数据
return [item.embedding for item in response.data]
# 初始化 ChromaDB 客户端 (持久化到本地目录)
chroma_client = chromadb.PersistentClient(path="chroma_db")
# 创建或获取集合 (相当于数据库表)
collection = chroma_client.get_or_create_collection(
name="news_articles", # 集合名称
metadata={"hnsw:space": "cosine"} # 使用余弦相似度
)
# 原始文本数据集（示例新闻标题）
documents = [
"李彦宏称大模型成本每年降低90%", # 科技类
"乌军大批直升机击落多架俄无人机", # 国际争端
"王力宏回应是否想找新伴侣", # 娱乐新闻
"饺子不知道观众怎么想出的藕饼cp", # 影视相关
"加沙停火协议关键时刻生变" # 国际争端
]
# 批量生成文档向量（OpenAI API调用）
document_embeddings = get_embeddings(documents)
# 生成唯一文档 ID (需要唯一标识符)
document_ids = [str(i) for i in range(len(documents))] # 生成 ["0", "1", ..., "4"]
# 将文档插入数据库
collection.add(
ids=document_ids, # 唯一ID列表
embeddings=document_embeddings, # 文本向量列表
documents=documents # 原始文本列表
)

复制代码

# 初始化向量数据库连接
chroma_client = chromadb.PersistentClient(path="chroma_db")
# 创建或获取集合 (相当于数据库表)
collection = chroma_client.get_or_create_collection(
name="news_articles", # 集合名称
metadata={"hnsw:space": "cosine"} # 使用余弦相似度
)

复制代码

# 1. 向量检索, 生成查询向量, 将文本转为向量
query_embedding = get_embeddings([user_query])[0]

复制代码

# 执行相似性查询
search_results = collection.query(
query_embeddings=[query_embedding], # 查询向量
n_results=2 # 返回前2个最相似结果
)

复制代码

# 提示词模板（控制模型输出格式）
prompt_template = """
你是一名国际军事问题专家。你的任务是根据现有知识库和用户问题给出最佳答案。
已知信息 :
__INFO__
用户提问 :
__QUERY__
"""

复制代码

def build_prompt(**kwargs):
"""动态构建提示词
Args:
prompt_template: 提示词模板字符串
**kwargs: 需要替换的键值对（自动匹配__KEY__占位符）
Returns:
完成替换后的完整提示词
"""
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)
print("构建提示词:" + prompt)
return prompt

复制代码

# 2. 构建增强提示
prompt = build_prompt(
info=search_results['documents'][0], # 取最相关结果
query=user_query
)

复制代码

def get_completion(prompt, model="gpt-3.5-turbo"):
"""封装OpenAI API调用
Args:
prompt: 完整的提示词内容
model: 选择的大模型版本（默认gpt-3.5-turbo）
Returns:
模型生成的文本响应
"""
messages = [{"role": "user", "content": prompt}]
response = client.chat.completions.create(
model=model,
messages=messages,
temperature=0, # 控制输出随机性（0为最确定性输出）
)
return response.choices[0].message.content # 提取响应内容

复制代码

# 3. 调用大模型生成
return self.llm_api(prompt)

复制代码

import chromadbfrom openai import OpenAI# 初始化 OpenAI 客户端 (替换成自己的 API 信息)client = OpenAI( api_key="sk-i3dHqaF6", # 替换为你的 OpenAI API Key , 这里我把自己的 API-KEY 隐蔽了 base_url="https://api.xiaoai.plus/v1" # 替换为你的 API 服务端点)# 提示词模板（控制模型输出格式）
prompt_template = """
你是一名国际军事问题专家。你的任务是根据现有知识库和用户问题给出最佳答案。
已知信息 :
__INFO__
用户提问 :
__QUERY__
"""
def build_prompt(**kwargs):
"""动态构建提示词
Args:
prompt_template: 提示词模板字符串
**kwargs: 需要替换的键值对（自动匹配__KEY__占位符）
Returns:
完成替换后的完整提示词
"""
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)
print("构建提示词:" + prompt)
return prompt
def get_completion(prompt, model="gpt-3.5-turbo"):
"""封装OpenAI API调用
Args:
prompt: 完整的提示词内容
model: 选择的大模型版本（默认gpt-3.5-turbo）
Returns:
模型生成的文本响应
"""
messages = [{"role": "user", "content": prompt}]
response = client.chat.completions.create(
model=model,
messages=messages,
temperature=0, # 控制输出随机性（0为最确定性输出）
)
return response.choices[0].message.content # 提取响应内容
def get_embeddings(texts, model="text-embedding-ada-002"): """将文本转换为向量表现 Args: texts: 需要编码的文本列表 model: 使用的嵌入模子（默认OpenAI官方保举模子） Returns: 包含向量数据的列表，每个元素对应输入文本的768维向量 """ response = client.embeddings.create( input=texts, model=model ) # 从相应中提取向量数据 return [item.embedding for item in response.data]class RAG_Service: def __init__(self, vector_db, llm_api, n_results=2): """初始化RAG机器人 Args: vector_db: 已初始化的向量数据库连接对象 llm_api: 大模子API调用函数 n_results: 默认检索效果数目 """ self.vector_db = vector_db # 向量数据库实例 self.llm_api = llm_api # LLM调用接口 self.n_results = n_results # 检索效果数目 def chat(self, user_query): """处理处罚用户查询的完整流程 Args: user_query: 用户输入的自然语言问题 Returns: 联合知识库天生的答复 """ # 1. 向量检索, 生成查询向量, 将文本转为向量
query_embedding = get_embeddings([user_query])[0]
# 执行相似性查询
search_results = collection.query(
query_embeddings=[query_embedding], # 查询向量
n_results=2 # 返回前2个最相似结果
)
# 2. 构建增强提示
prompt = build_prompt(
info=search_results['documents'][0], # 取最相关结果
query=user_query
)
# 3. 调用大模型生成
return self.llm_api(prompt)
# RAG 使用示例if __name__ == "__main__": # 初始化向量数据库连接
chroma_client = chromadb.PersistentClient(path="chroma_db")
# 创建或获取集合 (相当于数据库表)
collection = chroma_client.get_or_create_collection(
name="news_articles", # 集合名称
metadata={"hnsw:space": "cosine"} # 使用余弦相似度
)
# 创建 RAG 实例 bot = RAG_Service( vector_db=collection, llm_api=get_completion ) # 示例查询 user_query = "你对国际争端新闻的评价" # 天生查询向量 query_embedding = get_embeddings([user_query])[0] # RAG 答复 response = bot.chat(user_query) print("RAG 体系答复：", response)

复制代码

加沙停火协议关键时刻生变
乌军大批直升机击落多架俄无人机

复制代码

你是一名国际军事问题专家。你的使命是根据现有知识库和用户问题给出最佳答案。已知信息 : 加沙停火协议关键时刻生变
乌军大批直升机击落多架俄无人机
用户提问 : 你对国际争端新闻的评价

复制代码

D:\001_Develop\022_Python\Python39\python.exe D:/002_Project/011_Python/OpenAI/rag_service.py构建提示词:你是一名国际军事问题专家。你的使命是根据现有知识库和用户问题给出最佳答案。已知信息 : 加沙停火协议关键时刻生变
乌军大批直升机击落多架俄无人机
用户提问 : 你对国际争端新闻的评价RAG 体系答复：作为国际军事问题专家，我认为国际争端新闻总是令人担忧的。加沙停火协议的变革和乌克兰军队击落俄罗斯无人机的事件都表明国际局势仍然不稳定。这些事件大概会导致更多的紧张局势和辩说，需要各方保持冷静和谨慎，寻求通过对话和外交手段解决分歧。国际争端的解决需要各方共同积极，制止采取激烈的行动，以制止进一步升级局势。Process finished with exit code 0

复制代码