LlamaIndex是一个Python库，这意味着将其与全栈Web应用集成会与你习惯的方式略有不同。本指南旨在逐步讲解如何用Python编写一个基础的API服务，并展示其与TypeScript+React前端的交互过程。

此处所有代码示例均可从LlamaIndex入门套件在flask_react文件夹中。

技术栈 #

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

• Python 3.11 - 后端开发语言
• LlamaIndex - 核心AI框架
• Flask - Web框架
• TypeScript - 前端开发语言
• React - 前端框架

Flask 后端 #

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

基础Flask服务器设置 #

使用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文件夹，并将下载的essay文本文件放入其中。

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

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

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)

此函数将初始化我们的索引。如果在启动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由函数装饰器定义的endpoint
• 从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

# 注意：仅用于本地测试，部署时不要硬编码密钥
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上以密码启动服务器。

接着，我们可以按如下方式调整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()函数。

如果允许用户上传自己的文档，我们或许应该从documents文件夹中移除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，或者让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.pyPython文件中，我们有一个Flask API服务器，能够处理将文档插入向量索引的多个请求并响应用户查询！

为了支持前端的一些功能，我对Flask API的部分响应格式进行了调整，同时新增了追踪索引中存储文档的功能（LlamaIndex目前未以用户友好的方式提供此功能，但我们可以自行扩展！）。最后，我不得不使用Flask-corsPython包。

查看完整内容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服务器发起了一个查询（此处假设运行在本地主机上）。请注意，我们需要包含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调用可能要数上传文档了。此处的函数接收一个文件对象并构建POST请求使用FormData。

实际响应文本并未在应用中使用，但可用于在文件上传失败时向用户提供反馈。

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前端代码主要是一些基础组件，以及我为了让界面看起来至少顺眼些所做的努力😊。

我建议阅读其余部分代码库并提交任何改进的PR（Pull Request）！

项目架构总结 #

后端架构 #

1. Flask服务器 (端口5601)

   • 处理HTTP请求
   • 提供RESTful API接口
   • 管理文件上传和查询请求

2. BaseManager服务器 (端口5602)

   • 管理LlamaIndex索引
   • 处理并发访问
   • 确保数据一致性

3. LlamaIndex核心

   • 文档加载和处理
   • 向量索引构建
   • 语义搜索和查询

前端架构 #

1. React + TypeScript

   • 现代化前端框架
   • 类型安全的开发体验
   • 组件化架构

2. API集成

   • 文档获取API
   • 查询API
   • 文件上传API

3. 用户体验

   • 直观的文件上传界面
   • 实时查询响应
   • 文档管理和追踪

部署和扩展 #

生产环境部署 #

1. 环境配置

   • 使用环境变量管理API密钥
   • 配置CORS设置
   • 设置适当的日志级别

2. 容器化部署

   • 使用Docker容器化应用
   • 配置多容器编排
   • 设置健康检查和监控

3. 扩展性考虑

   • 支持多用户和多索引
   • 集成外部向量数据库
   • 添加缓存和负载均衡

功能扩展 #

1. 多用户支持

   • 用户认证和授权
   • 用户隔离的索引
   • 权限管理

2. 高级功能

   • 文档版本控制
   • 查询历史记录
   • 响应质量评估

3. 集成选项

   • S3存储集成
   • Pinecone向量数据库
   • 第三方LLM服务

结论 #

本指南涵盖了海量信息。我们从用Python编写一个基础的"Hello World" Flask服务器开始，逐步构建了一个由LlamaIndex驱动的完整后端系统，并讲解了如何将其连接到前端应用程序。

如您所见，我们可以轻松增强和封装LlamaIndex提供的服务（比如小型外部文档追踪器），从而在前端提供良好的用户体验。

你可以在此基础上添加许多功能（支持多索引/多用户、将对象保存到S3、添加Pinecone向量服务器等）。当你读完本文构建应用后，记得在Discord上分享最终成果！祝你好运！ 💪

下一步建议 #

1. 深入学习

   • 探索LlamaIndex的高级功能
   • 学习更多Flask和React最佳实践
   • 了解向量数据库和RAG系统

2. 项目扩展

   • 添加用户认证系统
   • 集成更多文档格式
   • 实现高级搜索功能

3. 性能优化

   • 优化索引构建速度
   • 改善查询响应时间
   • 实现智能缓存策略

4. 社区贡献

   • 分享你的项目经验
   • 参与开源贡献
   • 帮助改进文档和示例