使用 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 设置服务器非常简单。您只需导入包，创建 app 对象，然后创建您的端点。首先，让我们为服务器创建一个基本骨架：

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 中的保罗·格雷厄姆的随笔，创建一个 documents 文件夹，并将随笔文本文件下载并放入其中。

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

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

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

# NOTE: for local testing only, do NOT deploy with your key hardcoded
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=作者从小做了什么（按下回车键后，浏览器会将空格转换为 "%20" 字符）。

一切看起来都很好！我们现在拥有了一个功能齐全的 API。使用您自己的文档，您可以轻松地为任何应用程序提供一个接口，以调用 Flask API 并获取查询答案。

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

一切看起来都相当酷，但我们如何才能更进一步呢？如果我们想允许用户通过上传自己的文档来构建自己的索引怎么办？别担心，Flask 可以处理所有这些问题 :muscle:。

为了让用户上传文档，我们必须采取一些额外的预防措施。索引将不再是只读的，而是变得可变。如果有很多用户向同一个索引添加内容，我们需要考虑如何处理并发性。我们的 Flask 服务器是线程化的，这意味着多个用户可以同时向服务器发送请求，这些请求也将同时处理。

一种选择可能是为每个用户或组创建一个索引，并从 S3 存储和获取内容。但在此示例中，我们将假定存在一个用户正在交互的本地存储索引。

为了处理并发上传并确保按顺序插入索引，我们可以使用 BaseManager Python 包，通过单独的服务器和锁提供对索引的顺序访问。这听起来可能有点吓人，但并没有那么糟！我们只需将所有索引操作（初始化、查询、插入）移动到 BaseManager 的“索引服务器”中，然后从我们的 Flask 服务器调用这些操作。

这是我们将代码移动到 index_server.py 后，它的基本示例：

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

# NOTE: for local testing only, do NOT deploy with your key hardcoded
os.environ["OPENAI_API_KEY"] = "your key here"

index = None
lock = Lock()


def initialize_index():
    global index

    with lock:
        # same as before ...
        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__":
    # init the global index
    print("initializing index...")
    initialize_index()

    # setup server
    # NOTE: you might want to handle the password in a less hardcoded way
    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

# initialize manager connection
# NOTE: you might want to handle the password in a less hardcoded way
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() 函数。

如果我们允许用户上传自己的文档，我们应该先从 documents 文件夹中移除保罗·格雷厄姆的随笔。然后，让我们添加一个用于文件上传的端点！首先，让我们定义我们的 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:
        # cleanup temp file
        if filepath is not None and os.path.exists(filepath):
            os.remove(filepath)
        return "Error: {}".format(str(e)), 500

    # cleanup temp file
    if filepath is not None and os.path.exists(filepath):
        os.remove(filepath)

    return "File inserted!", 200

挺不错的！您会注意到我们将文件写入磁盘。如果只接受像 txt 文件这样的基本文件格式，我们可以跳过这一步，但写入磁盘后，我们可以利用 LlamaIndex 的 SimpleDirectoryReader 来处理许多更复杂的文件格式。另外，我们还可以使用第二个 POST 参数，要么使用文件名作为 doc_id，要么让 LlamaIndex 为我们生成一个。这在我们实现前端后会更容易理解。

对于这些更复杂的请求，我还建议使用 Postman 这样的工具。使用 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 Python 文件，我们就拥有了一个 Flask API 服务器，它可以处理多个请求以将文档插入到向量索引中并响应用户查询！

为了支持前端的一些功能，我调整了 Flask API 的部分响应格式，并添加了一些功能来跟踪索引中存储了哪些文档（LlamaIndex 目前尚未以用户友好的方式支持此功能，但我们可以自行增强它！）。最后，我不得不使用 Flask-cors Python 包为服务器添加 CORS 支持。

请在仓库中查看完整的 flask_demo.py 和 index_server.py 脚本，其中包含最终的微小更改、requirements.txt 文件以及一个用于部署的示例 Dockerfile。

React 前端#

通常，React 和 Typescript 是当今编写 Web 应用程序最流行的库和语言之一。本指南假定您熟悉这些工具的工作原理，否则本指南的篇幅将增加三倍 :smile:。

在仓库中，前端代码组织在 react_frontend 文件夹内。

前端最相关的部分是 src/apis 文件夹。这里是我们调用 Flask 服务器的地方，支持以下查询：

/query -- 对现有索引进行查询
/uploadFile -- 将文件上传到 Flask 服务器，以便插入索引
/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 服务器发起了查询（这里假设它运行在 localhost 上）。请注意，我们需要包含 mode: 'cors' 选项，因为我们正在发出外部请求。

然后，我们检查响应是否正常，如果正常，则获取响应的 JSON 并返回。在这里，响应的 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 前端代码是一些非常基础的 React 组件，以及我尽力使其看起来至少有点漂亮的尝试 :smile:。

我鼓励大家阅读代码库的其余部分，并提交任何改进的 PR！

总结#

本指南涵盖了大量信息。我们从一个用 Python 编写的基本“Hello World” Flask 服务器，到功能齐全的由 LlamaIndex 驱动的后端，以及如何将其连接到前端应用程序。

正如您所见，我们可以轻松地增强和封装 LlamaIndex 提供的服务（例如外部文档跟踪器），以帮助在前端提供良好的用户体验。

您可以在此基础上添加许多功能（多索引/用户支持、将对象保存到 S3、添加 Pinecone 向量服务器等）。当您阅读完本指南后构建应用程序时，请务必在 Discord 中分享最终成果！祝您好运！:muscle