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 提供的那篇 paul graham 随笔：建一个 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=what did the author do growing up（回车之后，浏览器会把空格转成 “%20”）。

事情进展得很顺利！我们现在有了一个可用的 API。配上你自己的文档，就能轻松地为任何应用提供一个接口，让它调用 flask API 获得查询答案。

高级 Flask —— 处理用户文档上传

事情进展得挺顺利，但我们还能再进一步吗？如果想让用户通过上传自己的文档来建立各自的索引呢？别担心，这些 Flask 全能应付 💪。

要让用户上传文档，就得采取一些额外措施。索引将不再是只查询既有索引，而是变成可变的。如果有许多用户往同一个索引里添加内容，我们就需要考虑并发处理。我们的 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

# 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 端点里通过 manager 来调用函数。

有一点需要特别注意：BaseManager 服务器返回的对象并不完全如我们所期望。要拿回原始对象，需要调用 _getvalue() 方法。

既然要让用户上传自己的文档，Paul Graham 那篇随笔就该从 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 测试我们各端点的示例放在本项目仓库里。

最后，你会注意到我们给 manager 新增了一个函数。我们在 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 应用最流行的库与语言之一。本指南假定你已经熟悉这些工具的用法，否则这份指南的篇幅要变成三倍 😄。

在仓库里，前端代码组织在 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' 选项，因为我们发的是跨域外部请求。

接着检查响应是否 ok，是的话取出响应 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。之后同样检查响应是否 ok，并按相应的 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 组件，再加上我尽力让界面至少看起来顺眼一点 😄。

鼓励你把其余代码读完，并为改进提交 PR！

结论

本指南讲了很多内容。我们从一个用 Python 写的基础 “Hello World” Flask 服务器，一路做到一个功能完备、由 LlamaIndex 支撑的后端，以及如何把它与前端应用接通。

你可以看到，LlamaIndex 提供的服务（比如那个小巧的外部文档记录器）很容易被我们扩展和包装，从而为前端提供良好的用户体验。

你完全可以在此基础上加很多功能（多索引/多用户支持、把对象存到 S3、接入 Pinecone 向量服务等）。读完这篇之后动手做出应用时，别忘了到 Discord 分享你的成果！祝顺利 💪