用 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,
)

# 注意：仅供本地测试使用，不要在部署时硬编码您的密钥
os.environ["OPENAI_API_KEY"] = "your key here"

index = None


def initialize_index():
    global index
    storage_context = StorageContext.from_defaults()
    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 函数开始之前调用它，那么我们的索引将准备好接受用户查询！

我们的查询端点将接受带有查询文本作为参数的 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 可以处理一切 :muscle:。

为了让用户上传文档，我们必须采取一些额外的预防措施。与查询现有索引不同，索引将变得可变。如果您有许多用户添加到同一个索引，我们需要考虑如何处理并发。我们的 Flask 服务器是多线程的，这意味着多个用户可以同时向服务器发送请求，这些请求将同时处理。一种选择是为每个用户或组创建一个索引，并从 S3 存储和获取内容。但在这个例子中，我们假设有一个本地存储的索引，用户正在与之交互。

为了处理并发上传并确保将内容按顺序插入索引，我们可以使用 BaseManager Python 包，通过一个单独的服务器和锁提供对索引的顺序访问。听起来有点吓人，但其实并不那么复杂！我们只需将所有的索引操作（初始化、查询、插入）移到 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("初始化索引...")
    initialize_index()

    # 设置服务器
    # 注意：可能需要以不那么硬编码的方式处理密码
    manager = BaseManager(("", 5602), b"password")
    manager.register("query_index", query_index)
    server = manager.get_server()

    print("启动服务器...")
    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 (
            "未找到文本，请在 URL 中包含 ?text=blah 参数",
            400,
        )
    response = manager.query_index(query_text)._getvalue()
    return str(response), 200


@app.route("/")
def home():
    return "你好，世界！"


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 "请使用带有文件的 POST 请求", 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 "错误：{}".format(str(e)), 500

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

    return "文件已插入！", 200

还好！你会注意到我们将文件写入磁盘。如果我们只接受基本文件格式如 txt 文件，我们可以跳过这一步，但是将文件写入磁盘后，我们可以利用 LlamaIndex 的 SimpleDirectoryReader 来处理更复杂的文件格式。另外，我们还使用第二个 POST 参数来决定是否使用文件名作为文档 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: ```