使用 LLamaIndex 构建全栈 Web 应用指南#

LlamaIndex 是一个 Python 库，这意味着将其与全栈 Web 应用集成会与您习惯的方式略有不同。

本指南旨在逐步介绍创建一个基础 Python API 服务的步骤，以及如何与 TypeScript+React 前端进行交互。

所有代码示例均可从 llama_index_starter_pack 的 flask_react 文件夹中获取。

本指南使用的主要技术如下：

python3.11
llama_index
flask
typescript
react

Flask 后端#

在本指南中，我们将使用 Flask API 服务器与前端代码通信。如果您愿意，也可以轻松地将其转换为 FastAPI 服务器或任何其他您选择的 Python 服务器库。

使用 Flask 设置服务器非常简单。导入包、创建应用对象，然后创建端点。首先让我们为服务器创建一个基础框架：

from flask import Flask

app = Flask(__name__)


@app.route("/")
def home():
    return "Hello World!"


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5601)

flask_demo.py

如果运行此文件（python flask_demo.py），它将在 5601 端口启动服务器。访问 http://localhost:5601/，您将在浏览器中看到 "Hello World!" 文本。很好！

下一步是决定要在服务器中包含哪些功能，并开始使用 LlamaIndex。

为了保持简单，我们可以提供的最基本操作是查询现有索引。使用 LlamaIndex 中的 paul graham essay，创建一个 documents 文件夹并下载+放置该文章文本文件到其中。

基础 Flask - 处理用户索引查询#

现在，让我们编写一些代码来初始化索引：

import os
from llama_index.core import (
    SimpleDirectoryReader,
    VectorStoreIndex,
    StorageContext,
    load_index_from_storage,
)

# 注意：仅用于本地测试，请勿部署时硬编码密钥
os.environ["OPENAI_API_KEY"] = "your key here"

index = None


def initialize_index():
    global index
    storage_context = StorageContext.from_defaults()
    index_dir = "./.index"
    if os.path.exists(index_dir):
        index = load_index_from_storage(storage_context)
    else:
        documents = SimpleDirectoryReader("./documents").load_data()
        index = VectorStoreIndex.from_documents(
            documents, storage_context=storage_context
        )
        storage_context.persist(index_dir)

此函数将初始化我们的索引。如果在 main 函数中启动 flask 服务器之前调用此函数，我们的索引将准备好处理用户查询！

我们的查询端点将接受带有查询文本作为参数的 GET 请求。以下是完整的端点函数：

from flask import request


@app.route("/query", methods=["GET"])
def query_index():
    global index
    query_text = request.args.get("text", None)
    if query_text is None:
        return (
            "No text found, please include a ?text=blah parameter in the URL",
            400,
        )
    query_engine = index.as_query_engine()
    response = query_engine.query(query_text)
    return str(response), 200

现在，我们为服务器引入了几个新概念：

由函数装饰器定义的新 /query 端点
从 flask 导入的新 request，用于从请求中获取参数
如果缺少 text 参数，则返回错误消息和适当的 HTML 响应代码
否则，我们查询索引，并将响应作为字符串返回

您可以在浏览器中测试的完整查询示例如下：http://localhost:5601/query?text=what did the author do growing up（按下回车后，浏览器会将空格转换为 "%20" 字符）。

看起来相当不错！我们现在有了一个功能性的 API。使用您自己的文档，可以轻松为任何应用程序提供调用 flask API 并获取查询答案的接口。

高级 Flask - 处理用户文档上传#

目前看起来相当不错，但我们如何更进一步呢？如果我们想允许用户通过上传自己的文档来构建索引呢？别担心，Flask 都能搞定 💪。

要让用户上传文档，我们需要采取一些额外的预防措施。与查询现有索引不同，此时索引将变为可变的。如果有多个用户同时向同一索引添加内容，我们需要考虑如何处理并发问题。我们的 Flask 服务器是多线程的，这意味着多个用户可以同时向服务器发送请求。

一个可能的解决方案是为每个用户或群组创建独立的索引，并通过 S3 存储和获取数据。但在本示例中，我们将假设所有用户都在与一个本地存储的索引进行交互。

为了处理并发上传并确保按顺序插入索引，我们可以使用 Python 的 BaseManager 包，通过独立的服务器和锁机制来提供对索引的顺序访问。听起来有点吓人，但其实并不复杂！我们只需将所有索引操作（初始化、查询、插入）移到 BaseManager 的 "index_server" 中，然后通过 Flask 服务器调用。

以下是迁移代码后 index_server.py 的基本示例：

import os
from multiprocessing import Lock
from multiprocessing.managers import BaseManager
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, Document

# 注意：仅限本地测试，切勿在部署时硬编码密钥
os.environ["OPENAI_API_KEY"] = "your key here"

index = None
lock = Lock()


def initialize_index():
    global index

    with lock:
        # 和之前一样...
        pass


def query_index(query_text):
    global index
    query_engine = index.as_query_engine()
    response = query_engine.query(query_text)
    return str(response)


if __name__ == "__main__":
    # 初始化全局索引
    print("initializing index...")
    initialize_index()

    # 设置服务器
    # 注意：建议采用更安全的方式处理密码
    manager = BaseManager(("", 5602), b"password")
    manager.register("query_index", query_index)
    server = manager.get_server()

    print("starting server...")
    server.serve_forever()

index_server.py

我们迁移了函数，引入了确保顺序访问全局索引的 Lock 对象，在服务器中注册了函数，并在端口 5602 上以密码 password 启动了服务器。

然后，我们可以按如下方式调整 Flask 代码：

from multiprocessing.managers import BaseManager
from flask import Flask, request

# 初始化管理器连接
# 注意：建议采用更安全的方式处理密码
manager = BaseManager(("", 5602), b"password")
manager.register("query_index")
manager.connect()


@app.route("/query", methods=["GET"])
def query_index():
    global index
    query_text = request.args.get("text", None)
    if query_text is None:
        return (
            "No text found, please include a ?text=blah parameter in the URL",
            400,
        )
    response = manager.query_index(query_text)._getvalue()
    return str(response), 200


@app.route("/")
def home():
    return "Hello World!"


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5601)

flask_demo.py

主要变化是连接到现有的 BaseManager 服务器并注册函数，以及在 /query 端点通过管理器调用函数。

需要注意的是，BaseManager 服务器返回的对象与我们预期略有不同。要获取原始对象，我们需要调用 _getvalue() 函数。

如果允许用户上传文档，我们应该先删除文档文件夹中的 Paul Graham 文章。然后添加文件上传端点！首先定义 Flask 端点函数：

...
manager.register("insert_into_index")
...


@app.route("/uploadFile", methods=["POST"])
def upload_file():
    global manager
    if "file" not in request.files:
        return "Please send a POST request with a file", 400

    filepath = None
    try:
        uploaded_file = request.files["file"]
        filename = secure_filename(uploaded_file.filename)
        filepath = os.path.join("documents", os.path.basename(filename))
        uploaded_file.save(filepath)

        if request.form.get("filename_as_doc_id", None) is not None:
            manager.insert_into_index(filepath, doc_id=filename)
        else:
            manager.insert_into_index(filepath)
    except Exception as e:
        # 清理临时文件
        if filepath is not None and os.path.exists(filepath):
            os.remove(filepath)
        return "Error: {}".format(str(e)), 500

    # 清理临时文件
    if filepath is not None and os.path.exists(filepath):
        os.remove(filepath)

    return "File inserted!", 200

还不错！我们先将文件写入磁盘。如果只接受 txt 等简单格式可以跳过这步，但写入磁盘后可以利用 LlamaIndex 的 SimpleDirectoryReader 处理更复杂的文件格式。我们还使用第二个 POST 参数选择是否使用文件名作为 doc_id。

对于复杂请求，建议使用 Postman。测试示例见项目仓库。

最后在 index_server.py 中实现新函数：

def insert_into_index(doc_text, doc_id=None):
    global index
    document = SimpleDirectoryReader(input_files=[doc_text]).load_data()[0]
    if doc_id is not None:
        document.doc_id = doc_id

    with lock:
        index.insert(document)
        index.storage_context.persist()


...
manager.register("insert_into_index", insert_into_index)
...

很简单！同时运行 index_server.py 和 flask_demo.py，我们就有了一个能处理多请求的 Flask API 服务器！

为了支持前端功能，我调整了部分 API 响应格式，并添加了跟踪索引中文档的功能（LlamaIndex 目前对此支持有限）。最后使用 Flask-cors 包添加了 CORS 支持。

完整代码见仓库，包括 requirements.txt 和示例 Dockerfile。

React 前端#

React 和 TypeScript 是目前最流行的前端开发工具。本指南假设您已熟悉这些工具。

在仓库中，前端代码位于 react_frontend 文件夹。

最核心的部分是 src/apis 文件夹，这里包含对 Flask 服务器的调用：

/query —— 查询现有索引
/uploadFile —— 上传文件到服务器并插入索引
/getDocuments —— 列出当前文档标题和部分内容

通过这三个接口，我们可以构建一个强大的前端，允许用户上传和管理文件、查询索引，并查看响应和来源文本。

fetchDocuments.tsx#

该文件包含获取索引中文档列表的函数：

export type Document = {
  id: string;
  text: string;
};

const fetchDocuments = async (): Promise<Document[]> => {
  const response = await fetch("http://localhost:5601/getDocuments", {
    mode: "cors",
  });

  if (!response.ok) {
    return [];
  }

  const documentList = (await response.json()) as Document[];
  return documentList;
};

我们向 Flask 服务器（假设运行在本地）发送请求。注意需要包含 mode: 'cors' 选项，因为这是跨域请求。检查响应状态后，返回解析后的 JSON 数据（即 Document 对象列表）。

queryIndex.tsx#

该文件将用户查询发送至 Flask 服务器，并获取响应结果以及索引中提供响应的节点详情。

export type ResponseSources = {
  text: string;
  doc_id: string;
  start: number;
  end: number;
  similarity: number;
};

export type QueryResponse = {
  text: string;
  sources: ResponseSources[];
};

const queryIndex = async (query: string): Promise<QueryResponse> => {
  const queryURL = new URL("http://localhost:5601/query?text=1");
  queryURL.searchParams.append("text", query);

  const response = await fetch(queryURL, { mode: "cors" });
  if (!response.ok) {
    return { text: "Error in query", sources: [] };
  }

  const queryResponse = (await response.json()) as QueryResponse;

  return queryResponse;
};

export default queryIndex;

这与 fetchDocuments.tsx 文件类似，主要区别在于我们将查询文本作为 URL 参数传递。随后检查响应状态，并返回带有正确 TypeScript 类型的响应结果。

insertDocument.tsx#

最复杂的 API 调用可能要数文档上传。此函数接收文件对象，并使用 FormData 构建 POST 请求。

虽然实际响应文本未在应用中使用，但可用于向用户反馈文件是否上传成功。

const insertDocument = async (file: File) => {
  const formData = new FormData();
  formData.append("file", file);
  formData.append("filename_as_doc_id", "true");

  const response = await fetch("http://localhost:5601/uploadFile", {
    mode: "cors",
    method: "POST",
    body: formData,
  });

  const responseText = response.text();
  return responseText;
};

export default insertDocument;

其他前端亮点#

至此前端部分已基本完成！剩余的 React 前端代码包含一些基础组件，以及我让界面看起来至少还算顺眼的尝试 :smile:。

建议阅读完整代码库并提交改进的 PR！

结语#

本指南涵盖了海量信息。我们从 Python 编写的基础 Flask 服务器"Hello World"开始，逐步构建了功能完整的 LlamaIndex 驱动后端，并实现其与前端应用的连接。

如你所见，我们可以轻松扩展和封装 LlamaIndex 提供的服务（例如外部文档追踪器），从而为前端提供良好的用户体验。

基于此，你可以添加诸多功能（多索引/用户支持、将对象存储至 S3、接入 Pinecone 向量服务器等）。若你根据本指南构建应用，请务必在 Discord 中分享成果！祝你好运！ :muscle: