云计算
使用BigQuery作为向量存储构建问答应用
向量存储是检索增强生成（RAG）系统中的关键部分。本质上，“向量存储”只是一个数据库，用于存储向量、文本、元数据、ID以及其他应用特定的数据。通用关系型数据库引擎与向量存储之间的区别在于，后者经过专门设计，以优化高维度数据的检索效率。在大规模应用中，这种差异更加明显；想象一个需要扫描成千上万份文档的应用程序，每份文档包含数千页内容，这些文档存储在多个不同的向量存储中。
存在一些用例不需要最大化效率。有些应用程序仅引用几个私人文档，或者开发人员正在构建一个概念验证（Proof of Concept，简称PoC）来展示功能。对于这些应用程序，很容易陷入无底洞，即评估所有可用工具、比较它们并做出决策（如果你陷入其中，我建议点击这篇文章来了解各种向量数据库的评估标准）。另一种途径是直接使用可用的工具。如果团队有一个 Google Cloud 环境，这些应用程序的开发选择就变得相对简单。我们可以使用 Vertex AI 来构建模型，同时使用 BigQuery 作为向量数据库和检索引擎。幸运的是，LangChain 框架提供了 Python 库，使得开发变得非常简单。
我们将要开发一个简单的Python程序，使用Streamlit前端、FastAPI后端、Langchain和Google Cloud服务来针对2024 NFL规则手册中的问题进行提问。我们将使用Docker容器来运行这个应用。（可在此处下载：2024 NFL规则手册）
此应用程序的完整版本位于GitHub，但在本教程中我们将逐步构建它。需要一个启用了BigQuery和Vertex AI API功能的Google Cloud项目。注意：这将产生数据存储和模型推理相关的费用，建议在教程结束后删除该项目。
以下是期望的完整应用程序结构；但是，本教程主要关注的是支持RAG应用程序的后端代码。
以下为文件和目录结构：
└── bq_vector_store
└── client
└── app.py
└── Dockerfile
└── requirements.txt
└── server
└── app.py
└── Dockerfile
└── requirements.txt
└── .env
└── .env.example
└── .gitignore
└── docker-compose.yaml
└── README.md
首先，我们需要安装所有必要的Python库。这些库可以一个一个地安装，也可以通过一个需求文件进行批量安装。
/server/requirements.txt
fastapi
langchain
langchain-google-vertexai
langchain-google-community[featurestore]
langchain_community
uvicorn
pdfplumber
app.py 文件将是 FastAPI 应用程序。这里发生所有事情。我们将创建一个 BigQuery 向量存储，设置模型接口，接受上传至向量存储的文件，并根据向量存储中的文件提问。运行应用程序只需要这两条路径；/upload_pdf 和 /ask_question。
在我们开始讨论或查看那些API路径之前，我们需要配置我们的模型和BigQuery向量存储。为此，我们可以使用LangChain为托管在Vertex AI上的模型创建客户端对象。请注意，PROJECT_ID将专门针对您的Google Cloud环境。
服务器上的应用程序文件路径 /服务器/app.py （服务器端的应用程序文件） (服务器上的 app.py 文件)（服务器上的应用程序文件）
从langchain_google_vertexai导入VertexAIEmbeddings, VertexAI embedding = VertexAIEmbeddings(model_name='textembedding-gecko@latest', project=PROJECT_ID) llm = VertexAI(model_name='gemini-1.5-flash-002')
接下来，我们可以创建一个利用 BigQuery 的 Vector Store 对象。这些常量将根据应用程序的具体情况而定。创建完 BigQueryVectorStore 对象后，我们可以将其实例化为检索器。这样我们就可以调用检索器来获取基于查询的上下文。search_kwargs 参数定义了从向量存储中返回多少文本片段。
/server/app.py
从 langchain_google_community 导入 BigQueryVectorStore 作为 BigQueryVectorStore
store = BigQueryVectorStore(
project_id=PROJECT_ID, # 项目ID
dataset_name=DATASET, # 数据集名称
table_name=TABLE, # 表名称
location=REGION, # 地域
embedding=embedding, # 嵌入
)
retriever = store.as_retriever(
search_kwargs={ # 搜索参数
"k": 10
}
)
现在基础已经设置好（嵌入模型、大型语言模型和向量存储），我们现在可以开始添加聊天记录和一些标准的模型训练提示。这个“指导”是提示工程中的一个环节，它指示大型语言模型使用我们向量存储中的信息。我们将分步骤来解释每个提示和链，进一步解释每个提示和链。
/server/app.py
from langchain_core.prompts import MessagesPlaceholder
from langchain_core.prompts import ChatPromptTemplate
from langchain.chains import create_history_aware_retriever
system_prompt = (
"你是一个用于问答任务的助手。"
"使用以下检索到的信息来回答问题。"
"如果不知道答案，就说不知道。"
"\n\n"
"{context}"
)
contextualize_q_system_prompt = (
"给定一个聊天历史和最新的用户问题，该问题可能引用了聊天历史中的上下文，"
"制定一个独立且自包含的问题，这个问题可以在没有聊天历史的情况下被理解。"
"不要回答问题，如果需要的话，重新制定问题，否则就按原样返回。"
)
contextualize_q_prompt = ChatPromptTemplate.from_messages(
[
("系统", contextualize_q_system_prompt),
MessagesPlaceholder("chat_history"),
("用户", "{input}"),
]
)
history_aware_retriever = create_history_aware_retriever(
llm, retriever, contextualize_q_prompt
)
上面的 system_prompt 是一个基本提示，告诉模型如何表现。稍后的代码中会插入从向量存储中检索到的 {context}。
contextualize_q_system_prompt将作为我们在聊天历史中的虚拟代理。通过查看提示，我们让模型如果认为我们是在问后续问题的话，重新表述这个问题。
这些提示随后被整合到一个更广泛的提示contextualize_q_prompt中。可以将其视为一系列逐步的指令。首先，我们使用contextualize_q_system_prompt提示模型利用聊天历史，然后我们用MessagesPlaceholder("chat_history")获取之前的聊天消息，最后，("human", "{input}")这一部分将是我们要提问的关于NFL比赛规则的问题的位置。
上述代码块最终以我们的 history_aware_retriever 结束，它负责从向量存储中检索上下文，并用此信息回答 LLM 的问题，使表达更自然流畅。
我们需要采取另外几步，让程序能用我们想要的方式回答问题。
/server/app.py
from langchain.chains 导入 create_retrieval_chain
from langchain.chains.combine_documents 导入 create_stuff_documents_chain
qa_prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt),
MessagesPlaceholder("chat_history"),
("human", "{input}"),
]
)
# 空聊天记录
chat_history = []
# 创建RAG链（相关性感知生成）
question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)
rag_chain = create_retrieval_chain(history_aware_retriever, question_answer_chain)
注意，qa_prompt 与 contextualize_qa_prompt 非常相似，唯一的区别在于 qa_prompt 中我们使用 system_prompt 作为 "system" 的值，而在 contextualize_qa_prompt 中使用的是 contextualize_system_prompt。这是由于我们的应用程序有两个决策点。决策在于询问模型是否需要参考聊天记录来检索上下文，还是只需检索上下文。这些步骤在 question_answer_chain 和 rag_chain 中有详细说明。此外，我们设置了一个空的 chat_history 列表，作为我们的内存中的聊天记录。注：在实际应用中，我们也希望将聊天记录存储在独立的数据存储中（例如可以使用 Redis 或 BigQuery）。
现在我们可以将上面的代码添加到我们的FastAPI应用中。我们可以使用我们的store对象将一个pdf文件添加到向量数据库中，如下：
/server/app.py
从langchain_text_splitters导入RecursiveCharacterTextSplitter
从langchain_community.document_loaders导入PDFPlumberLoader
@app.post("/upload_pdf")
async def upload_pdf(request: Request):
"""
用于将PDF上传到BigQuery向量存储的函数
输入必须是一个url。
参数说明:
request (Request): 客户端发送的POST请求，必须包含带有'pdf_url'键的json数据。
从请求中获取数据如下:
json_data = await request.json()
pdf_url = json_data['pdf_url']
loader = PDFPlumberLoader(pdf_url)
docs = loader.load()
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200)
splits = text_splitter.split_documents(docs)
store.add(splits)
return {"消息": "PDF上传成功"}
这个代码块添加了几个新组件。一个是PDFPlumberLoader类，它非常擅长从PDF文件中提取文本。这个类用来创建docs，从而获取我们想要上传文件的文本版本。然后，RecursiveCharacterTextSplitter接手，将文本分割成每段1000个字符（chunk_size=1000）并且每段之间有200字符的重叠（chunk_overlap=200）的片段。最后，这些片段（splits）被加入到我们的向量存储（store）中。
可以使用下面的 curl 命令访问这个API
curl -X POST \
-H "Content-Type: application/json" \
-d '{"pdf_url": "https://operations.nfl.com/media/24emxacq/2024-nfl-rulebook.pdf"}' \
http://localhost:8000/upload_pdf
此处使用 curl 命令向指定的 URL 发送一个 POST 请求，上传一个包含 PDF 链接的 JSON 数据。
一旦文件上传后，我们希望对这个文件向向量存储库提出一些问题。该端点的定义如下：
/server/app.py
从 langchain_core.messages 导入 AIMessage, HumanMessage
@app.post("/ask_question")
async def ask_question(request: Request):
"""
允许向上面定义的链提出问题。
参数：
request (Request): 客户端的POST请求必须包含一个带有'question'键的JSON数据。
返回：
answer (str): 对用户问题的回答
"""
# 从请求中获取JSON数据
json_data = await request.json()
question = json_data['question']
# 调用RAG链
response = rag_chain.invoke(
{
"input": question,
"chat_history": chat_history
}
)
answer = response['answer']
# 扩展聊天历史记录
chat_history.extend(
[
HumanMessage(content=question),
AIMessage(content=answer),
]
)
return answer
在这里，我们从POST请求中抓取问题，并将其添加到我们的rag_chain中。同时，我们将这个问题加入到我们的chat_history中。rag_chain.invoke()方法将通过模型获取响应。我们只需解析答案（response['answer']），并用用户的提问及回答来扩充我们的chat_history。这个聊天记录会成为后续问题的参考依据。
下面我会包含Streamlit应用的代码，但不会过多关注代码组织和功能。该应用允许输入PDF文件的URL，并包含一个聊天机器人组件。Streamlit比较容易操作，通常的挑战是在熟悉用户界面。客户端的唯一要求是streamlit。
/client/requirements.txt
streamlit
关于代码的几点说明；chat_url 和 backend_url 专用于 Docker 部署，可能需要修改为 localhost。另外，聊天记录存储在客户端 (st.session_state.messages)。我们这么做是为了方便显示消息，同时减少客户端与服务器之间的数据传输。
客户端app.py文件
import streamlit as st
import requests
chat_url = "http://server:8000/ask_question"
backend_url = "http://server:8000/upload_pdf"
# 设置侧边栏输入PDF网址
st.sidebar.title("输入PDF网址")
pdf_url = st.sidebar.text_input("PDF网址:", "")
# 初始化聊天历史
if "messages" not in st.session_state:
st.session_state.messages = []
# 上传PDF到后端按钮
if st.sidebar.button("上传PDF"):
if pdf_url:
# 发送PDF URL到后端API
response = requests.post(backend_url,
json={
"pdf_url": pdf_url
}
)
if response.status_code == 200:
st.sidebar.success("PDF上传成功！")
else:
st.sidebar.error("上传PDF失败，请稍后再试。")
else:
st.sidebar.warning("请输入PDF网址。")
# 设置主页面聊天机器人界面
st.title("关于您的PDF聊天")
st.write("请在侧边栏输入PDF网址以开始聊天。")
# 在重新加载应用程序时显示聊天历史
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
# 对用户输入作出反应
if prompt := st.chat_input("提问"):
# 显示用户消息
st.chat_message("user").markdown(prompt)
# 添加用户消息到聊天历史
st.session_state.messages.append({"role": "user", "content": prompt})
response = requests.post(chat_url, json={"question": prompt})
answer = response.json()
# 显示助手回复
with st.chat_message("assistant"):
st.markdown(answer)
# 添加助手回复到聊天历史
st.session_state.messages.append({"role": "assistant", "content": answer})
我们只需要再找到四个文件就可以运行应用程序了。这些文件的定义及注释如下：
/docker-compose.yaml
服务:
后端:
image: server
build:
context: ./server
ports:
- "8000:8000"
volumes:
- ${GCLOUD_CONFIG}:/root/.config/gcloud/application_default_credentials.json
环境变量:
- GOOGLE_APPLICATION_CREDENTIALS=/root/.config/gcloud/application_default_credentials.json
- PROJECT_ID=${PROJECT_ID}
- REGION=${REGION}
- DATASET=${DATASET}
- TABLE=${TABLE}
前端:
build:
context: ./client
ports:
- "8501:8501"
volumes:
- ${GCLOUD_CONFIG}:/root/.config/gcloud/application_default_credentials.json
环境变量:
- GOOGLE_APPLICATION_CREDENTIALS=/root/.config/gcloud/application_default_credentials.json
卷:
- ${GCLOUD_CONFIG}:/root/.config/gcloud/application_default_credentials.json
这将设置所有必要的环境变量，并让我们能够将 Google Cloud 凭证挂载到容器中。
所有的 ${VAR} 环境变量都是从 .env 文件中拉取的。
/.env
PROJECT_ID = your_project_id
REGION = your_region
DATASET = your_dataset
TABLE = nfl_playbook
# Linux, macOS: `~/.config/gcloud/application_default_credentials.json`
# Windows: `%用户目录%\AppData\gcloud\application_default_credentials.json`
GCLOUD_CONFIG = gcloud应用默认凭证.json的路径
这些变量将包括您的 Google Cloud 账户 ID 和路径。请注意，GCLOUD_CONFIG 变量的注释会因操作系统而有所不同。
我们还需要为每个服务准备Dockerfile。这些文件用来指引容器的创建并启动它们的服务器。
/client/Dockerfile
# 使用官方 Python 运行时镜像
FROM python:3.11-slim
# 设置工作目录为 /app
WORKDIR /app
# 复制 requirements 文件到容器
COPY requirements.txt .
# 安装所需依赖
RUN pip install --no-cache-dir -r requirements.txt
# 复制 Streamlit 应用代码到容器
COPY . .
# 暴露应用端口
EXPOSE 8501
# 启动 Streamlit 应用命令
CMD ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"]
/server/Dockerfile
# 基于官方的 Python 镜像
FROM python:3.11-slim
# 设置工作目录为 /app
WORKDIR /app
# 复制 requirements 文件到容器
COPY requirements.txt .
# 安装依赖项
RUN pip install --no-cache-dir -r requirements.txt
# 复制 FastAPI 应用代码到容器
COPY . .
# 暴露端口 8000
EXPOSE 8000
# 运行 FastAPI 应用使用 uvicorn 的命令
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]
现在所有文件都已经到位，进入 / 目录并执行相关命令。
docker compose up --build
打开浏览器，访问 http://localhost:8051 查看前端页面或访问 http://localhost:8000/docs 查看 API 文档页面。
-
阿里云部署学习：新手入门指南12-18
-
阿里云部署教程：新手入门指南12-18
-
网站项目部署发布的简单教程12-17
-
2024年ETL工程的几大趋势解读12-13
-
别把数据质量搞复杂了12-13
-
构建SaaS或微SaaS的最佳方法，手把手教你！12-13
-
云中的自动驾驶扩展之规模化自主性办公系统BATES12-13
-
现有数据架构的扩展：理想与现实的折中方案12-13
-
阿里云服务器配置类型有哪些？你真的了解吗？12-12
-
Teable 团队 Sealos 最佳实践，创业公司的完美选择12-11
-
分布式流媒体系统教程：入门与实践指南12-11
-
如何搭建高性能流媒体服务器12-11
-
如何搭建你的第一台流媒体服务器12-11
-
分布式流媒体系统项目实战入门教程12-11
-
分布式直播系统项目实战入门教程12-11
