1.初始化项目 #

uv init

2.安装依赖 #

uv add  llama-index flask  flask-cors

3.运行项目 #

uv run main.py

4.启动服务器 #

1.1. config.py #

config.py

# 定义一个配置类，用于存放API相关的配置信息
class Config:
    # 设置Flask服务监听的主机地址为所有可用IP
    FLASK_HOST = "0.0.0.0"
    # 设置Flask服务监听的端口号为5601
    FLASK_PORT = 5601


# 创建全局配置类的实例，方便在其他模块中引用
config = Config()

1.2. main.py #

main.py

# 从flask库中导入Flask和Response类
+from flask import Flask, Response

# 从config模块导入config配置对象
+from config import config

# 导入json模块，用于数据序列化
+import json

# 创建Flask应用实例
+app = Flask(__name__)


# 定义根路由"/"的处理函数
+@app.route("/")
+def home():
    # 构造返回的数据字典
+   data = {"message": "LlamaIndex智能文档问答系统", "version": "1.0.0"}
    # 将数据序列化为JSON字符串并返回，设置响应类型为application/json
+   return Response(json.dumps(data, ensure_ascii=False), mimetype="application/json")


# 判断是否为主程序入口
if __name__ == "__main__":
    # 打印启动信息，显示访问地址
+   print(f"启动Flask应用: http://{config.FLASK_HOST}:{config.FLASK_PORT}")
    # 启动Flask应用，监听指定主机和端口，开启debug模式
+   app.run(host=config.FLASK_HOST, port=config.FLASK_PORT, debug=True)

5.索引管理器 #

5.1. index_manager.py #

index_manager.py

# 导入OpenAIEmbedding用于生成嵌入向量
from llama_index.embeddings.openai import OpenAIEmbedding

# 导入自定义配置文件
from config import config

# 导入操作系统相关模块
import os

# 从llama_index.core中导入所需的类和方法
from llama_index.core import (
    VectorStoreIndex,  # 向量存储索引类
    StorageContext,  # 存储上下文类
    load_index_from_storage,  # 从存储中加载索引的方法
)


# 定义索引管理器类
class IndexManager:
    """索引管理器"""

    # 构造方法，初始化索引管理器
    def __init__(self):
        # 初始化索引对象为None
        self.index = None
        # 创建OpenAI嵌入模型，模型名称从配置中获取
        self.embed_model = OpenAIEmbedding(model_name=config.EMBEDDING_MODEL)
        # 初始化索引
        self.initialize_index()

    # 初始化索引的方法
    def initialize_index(self):
        # 如果索引目录已存在
        if os.path.exists(config.INDEX_DIR):
            # 从存储中加载已有索引
            self.index = load_index_from_storage(
                StorageContext.from_defaults(persist_dir=config.INDEX_DIR),
                embed_model=self.embed_model,
            )
        else:
            # 如果索引目录不存在，则新建一个空的向量存储索引
            self.index = VectorStoreIndex(nodes=[], embed_model=self.embed_model)
            # 持久化存储索引到指定目录
            self.index.storage_context.persist(persist_dir=config.INDEX_DIR)


# 创建一个全局的索引管理器实例
index_manager = IndexManager()

5.2. config.py #

config.py

# 定义一个配置类，用于存放API相关的配置信息
class Config:
    # 设置Flask服务监听的主机地址为所有可用IP
    FLASK_HOST = "0.0.0.0"
    # 设置Flask服务监听的端口号为5601
    FLASK_PORT = 5601
    # 设置索引目录
+   INDEX_DIR = "./saved_index"
    # 设置嵌入模型
+   EMBEDDING_MODEL = "text-embedding-3-small"


# 创建全局配置类的实例，方便在其他模块中引用
config = Config()

5.3. main.py #

main.py

# 从flask库中导入Flask和Response类
from flask import Flask, Response

# 从config模块导入config配置对象
from config import config

# 导入json模块，用于数据序列化
import json

# 导入索引管理器
+from index_manager import index_manager

# 创建Flask应用实例
app = Flask(__name__)


# 定义根路由"/"的处理函数
@app.route("/")
def home():
    # 构造返回的数据字典
    data = {"message": "LlamaIndex智能文档问答系统", "version": "1.0.0"}
    # 将数据序列化为JSON字符串并返回，设置响应类型为application/json
    return Response(json.dumps(data, ensure_ascii=False), mimetype="application/json")


# 判断是否为主程序入口
if __name__ == "__main__":
    # 打印启动信息，显示访问地址
    print(f"启动Flask应用: http://{config.FLASK_HOST}:{config.FLASK_PORT}")
    # 启动Flask应用，监听指定主机和端口，开启debug模式
    app.run(host=config.FLASK_HOST, port=config.FLASK_PORT, debug=True)

6.实现文件上传 #

6.1. utils.py #

utils.py

from config import config


def is_supported_file(filename: str) -> bool:
    """检查文件格式是否支持"""
    return any(filename.lower().endswith(ext) for ext in config.SUPPORTED_FORMATS)

6.2. config.py #

config.py

# 定义一个配置类，用于存放API相关的配置信息
class Config:
    # 设置Flask服务监听的主机地址为所有可用IP
    FLASK_HOST = "0.0.0.0"
    # 设置Flask服务监听的端口号为5601
    FLASK_PORT = 5601
    # 设置索引目录
    INDEX_DIR = "./saved_index"
    # 设置嵌入模型
    EMBEDDING_MODEL = "text-embedding-3-small"
    # 支持的文件格式
+   SUPPORTED_FORMATS = [".pdf", ".txt", ".json", ".md"]
    # 设置文档目录
+   DOCUMENTS_DIR = "documents"


# 创建全局配置类的实例，方便在其他模块中引用
config = Config()

6.3. index_manager.py #

index_manager.py

# 导入OpenAIEmbedding类，用于生成文本的嵌入向量
from llama_index.embeddings.openai import OpenAIEmbedding

# 导入自定义的配置对象config
from config import config

# 导入操作系统相关模块，用于文件和目录操作
import os

# 从llama_index.core模块中导入所需的类和方法
from llama_index.core import (
    VectorStoreIndex,  # 向量存储索引类
    StorageContext,  # 存储上下文类
+   load_index_from_storage,  # 从存储目录加载索引的方法
+   SimpleDirectoryReader,  # 简单目录读取器，用于读取文件
)


# 定义索引管理器类
class IndexManager:
    """索引管理器"""

    # 构造方法，初始化索引管理器实例
    def __init__(self):
        # 初始化索引对象为None
        self.index = None
        # 创建OpenAI嵌入模型实例，模型名称从配置中获取
        self.embed_model = OpenAIEmbedding(model_name=config.EMBEDDING_MODEL)
        # 调用初始化索引的方法
        self.initialize_index()

    # 初始化索引的方法
    def initialize_index(self):
        # 判断索引目录是否存在
        if os.path.exists(config.INDEX_DIR):
            # 如果存在，从存储目录加载已有索引
            self.index = load_index_from_storage(
                StorageContext.from_defaults(persist_dir=config.INDEX_DIR),
                embed_model=self.embed_model,
            )
        else:
            # 如果不存在，则新建一个空的向量存储索引
            self.index = VectorStoreIndex(nodes=[], embed_model=self.embed_model)
            # 持久化存储索引到指定目录
            self.index.storage_context.persist(persist_dir=config.INDEX_DIR)

    # 插入文档到索引的方法
+   def insert_document(self, doc_file_path: str):
+       """插入文档到索引"""
        # 使用SimpleDirectoryReader读取指定文件，返回文档对象列表
+       documents = SimpleDirectoryReader(input_files=[doc_file_path]).load_data()

        # 遍历所有文档对象，逐个插入到索引中
+       for document in documents:
+           self.index.insert(document)

        # 持久化存储索引到指定目录
+       self.index.storage_context.persist(persist_dir=config.INDEX_DIR)


# 创建一个全局的索引管理器实例，供其他模块调用
index_manager = IndexManager()

6.4. main.py #

main.py

# 从flask库中导入Flask、Response、request和jsonify
+from flask import Flask, Response, request, jsonify

# 从config模块导入全局配置对象config
from config import config

# 导入json模块，用于数据序列化和反序列化
import json

# 从index_manager模块导入索引管理器实例
from index_manager import index_manager

# 从utils模块导入文件格式校验函数
+from utils import is_supported_file

# 导入os模块，用于文件路径操作
+import os

# 从werkzeug.utils导入secure_filename，用于安全处理文件名
+from werkzeug.utils import secure_filename

# 创建Flask应用实例
app = Flask(__name__)


# 定义根路由"/"的处理函数
@app.route("/")
def home():
    # 构造返回的数据字典
    data = {"message": "LlamaIndex智能文档问答系统", "version": "1.0.0"}
    # 将数据字典序列化为JSON字符串并返回，设置响应类型为application/json
    return Response(json.dumps(data, ensure_ascii=False), mimetype="application/json")


# 定义文件上传接口，路由为/uploadFile，仅支持POST方法
+@app.route("/uploadFile", methods=["POST"])
+def upload_file():
+   """文件上传接口"""
    # 判断请求中是否包含文件
+   if "file" not in request.files:
        # 如果没有文件，返回错误信息
+       return jsonify({"error": "请提供文件"}), 400

    # 获取上传的文件对象
+   uploaded_file = request.files["file"]
    # 判断文件名是否为空
+   if uploaded_file.filename == "":
        # 如果未选择文件，返回错误信息
+       return jsonify({"error": "未选择文件"}), 400

    # 判断文件格式是否受支持
+   if not is_supported_file(uploaded_file.filename):
        # 如果文件格式不支持，返回错误信息
+       return jsonify({"error": "不支持的文件格式"}), 400

+   try:
        # 使用secure_filename处理文件名，防止安全风险
+       filename = secure_filename(uploaded_file.filename)
        # 拼接文件保存路径
+       filepath = os.path.join(config.DOCUMENTS_DIR, filename)
        # 保存文件到指定路径
+       uploaded_file.save(filepath)
        # 将文件插入索引
+       index_manager.insert_document(filepath)
        # 返回上传成功信息
+       return jsonify({"message": "文件上传成功"})

    # 捕获异常并返回错误信息
+   except Exception as e:
+       return jsonify({"error": f"上传失败: {str(e)}"}), 500


# 判断当前模块是否为主程序入口
if __name__ == "__main__":
    # 确保文档目录存在，不存在则自动创建
+   os.makedirs(config.DOCUMENTS_DIR, exist_ok=True)
    # 打印启动信息，显示访问地址
    print(f"启动Flask应用: http://{config.FLASK_HOST}:{config.FLASK_PORT}")
    # 启动Flask应用，监听指定主机和端口，开启debug模式
    app.run(host=config.FLASK_HOST, port=config.FLASK_PORT, debug=True)

7.文档本地存储 #

7.1. config.py #

config.py

# 定义一个配置类，用于存放API相关的配置信息
class Config:
    # 设置Flask服务监听的主机地址为所有可用IP
    FLASK_HOST = "0.0.0.0"
    # 设置Flask服务监听的端口号为5601
    FLASK_PORT = 5601
    # 设置索引目录
    INDEX_DIR = "./saved_index"
    # 设置嵌入模型
    EMBEDDING_MODEL = "text-embedding-3-small"
    # 支持的文件格式
    SUPPORTED_FORMATS = [".pdf", ".txt", ".json", ".md"]
    # 设置文档目录
    DOCUMENTS_DIR = "documents"
    # 设置文档列表文件
+   DOCUMENTS_PKL = "stored_documents.pkl"


# 创建全局配置类的实例，方便在其他模块中引用
config = Config()

7.2. index_manager.py #

index_manager.py

# 导入OpenAIEmbedding类，用于生成文本的嵌入向量
from llama_index.embeddings.openai import OpenAIEmbedding

# 导入自定义的工具函数save_document_info和get_documents_list
+from utils import save_document_info, get_documents_list

# 导入自定义的配置对象config
from config import config

# 导入操作系统相关模块，用于文件和目录操作
import os

# 从llama_index.core模块中导入所需的类和方法
from llama_index.core import (
    VectorStoreIndex,  # 向量存储索引类
    StorageContext,  # 存储上下文类
    load_index_from_storage,  # 从存储目录加载索引的方法
    SimpleDirectoryReader,  # 简单目录读取器，用于读取文件
)


# 定义索引管理器类
class IndexManager:
    """索引管理器"""

    # 构造方法，初始化索引管理器实例
    def __init__(self):
        # 初始化索引对象为None
        self.index = None
        # 创建OpenAI嵌入模型实例，模型名称从配置中获取
        self.embed_model = OpenAIEmbedding(model_name=config.EMBEDDING_MODEL)
        # 调用初始化索引的方法
        self.initialize_index()

    # 初始化索引的方法
    def initialize_index(self):
        # 判断索引目录是否存在
        if os.path.exists(config.INDEX_DIR):
            # 如果索引目录存在，则从存储目录加载已有索引
            self.index = load_index_from_storage(
                StorageContext.from_defaults(persist_dir=config.INDEX_DIR),
                embed_model=self.embed_model,
            )
        else:
            # 如果索引目录不存在，则新建一个空的向量存储索引
            self.index = VectorStoreIndex(nodes=[], embed_model=self.embed_model)
            # 持久化存储索引到指定目录
            self.index.storage_context.persist(persist_dir=config.INDEX_DIR)

    # 插入文档到索引的方法
+   def insert_document(self, doc_file_path: str, filename: str):
        """插入文档到索引"""
        # 使用SimpleDirectoryReader读取指定文件，返回文档对象列表
        documents = SimpleDirectoryReader(input_files=[doc_file_path]).load_data()

        # 遍历所有文档对象，逐个插入到索引中
        for document in documents:
            self.index.insert(document)

        # 保存文档信息（文件名和文档前200个字符）
+       save_document_info(filename, documents[0].text[:200])
        # 持久化存储索引到指定目录
        self.index.storage_context.persist(persist_dir=config.INDEX_DIR)

    # 获取文档列表的方法
+   def get_documents_list(self):
+       """获取文档列表"""
        # 调用工具函数获取文档列表
+       return get_documents_list()


# 创建一个全局的索引管理器实例，供其他模块调用
index_manager = IndexManager()

7.3. main.py #

main.py

# 从flask库中导入Flask、Response、request和jsonify
from flask import Flask, Response, request, jsonify

# 从config模块导入全局配置对象config
from config import config

# 导入json模块，用于数据序列化和反序列化
import json

# 从index_manager模块导入索引管理器实例
from index_manager import index_manager

# 从utils模块导入文件格式校验函数
from utils import is_supported_file

# 导入os模块，用于文件路径操作
import os

# 从werkzeug.utils导入secure_filename，用于安全处理文件名
from werkzeug.utils import secure_filename

# 创建Flask应用实例
app = Flask(__name__)


# 定义根路由"/"的处理函数
@app.route("/")
def home():
    # 构造返回的数据字典
    data = {"message": "LlamaIndex智能文档问答系统", "version": "1.0.0"}
    # 将数据字典序列化为JSON字符串并返回，设置响应类型为application/json
    return Response(json.dumps(data, ensure_ascii=False), mimetype="application/json")


# 定义文件上传接口，路由为/uploadFile，仅支持POST方法
@app.route("/uploadFile", methods=["POST"])
def upload_file():
    # 文件上传接口
    """文件上传接口"""
    # 判断请求中是否包含文件
    if "file" not in request.files:
        # 如果没有文件，返回错误信息
        return jsonify({"error": "请提供文件"}), 400

    # 获取上传的文件对象
    uploaded_file = request.files["file"]
    # 判断文件名是否为空
    if uploaded_file.filename == "":
        # 如果未选择文件，返回错误信息
        return jsonify({"error": "未选择文件"}), 400

    # 判断文件格式是否受支持
    if not is_supported_file(uploaded_file.filename):
        # 如果文件格式不支持，返回错误信息
        return jsonify({"error": "不支持的文件格式"}), 400

    try:
        # 使用secure_filename处理文件名，防止安全风险
        filename = secure_filename(uploaded_file.filename)
        # 拼接文件保存路径
        filepath = os.path.join(config.DOCUMENTS_DIR, filename)
        # 保存文件到指定路径
        uploaded_file.save(filepath)
        # 将文件插入索引
+       index_manager.insert_document(filepath, filename)
        # 返回上传成功信息
        return jsonify({"message": "文件上传成功"})

    # 捕获异常并返回错误信息
    except Exception as e:
        return jsonify({"error": f"上传失败: {str(e)}"}), 500


# 定义获取文档列表的接口，路由为/getDocuments，仅支持GET方法
+@app.route("/getDocuments", methods=["GET"])
+def get_documents():
    # 获取文档列表接口
+   """获取文档列表接口"""
+   try:
        # 调用索引管理器获取文档列表
+       documents = index_manager.get_documents_list()
        # 返回文档列表的json数据
+       return jsonify(documents)
+   except Exception as e:
        # 捕获异常并返回错误信息
+       return jsonify({"error": f"获取文档列表失败: {str(e)}"}), 500


# 判断当前模块是否为主程序入口
if __name__ == "__main__":
    # 确保文档目录存在，不存在则自动创建
    os.makedirs(config.DOCUMENTS_DIR, exist_ok=True)
    # 打印启动信息，显示访问地址
    print(f"启动Flask应用: http://{config.FLASK_HOST}:{config.FLASK_PORT}")
    # 启动Flask应用，监听指定主机和端口，开启debug模式
    app.run(host=config.FLASK_HOST, port=config.FLASK_PORT, debug=True)

7.4. utils.py #

utils.py

# 导入类型提示所需的List和Dict
+from typing import List, Dict

# 导入操作系统相关模块
+import os

# 导入全局配置对象
from config import config

# 导入pickle模块用于对象序列化
+import pickle


# 检查文件格式是否受支持
def is_supported_file(filename: str) -> bool:
    # 返回文件名后缀是否在支持的格式列表中
    return any(filename.lower().endswith(ext) for ext in config.SUPPORTED_FORMATS)


# 保存文档信息到pickle文件
+def save_document_info(doc_id: str, doc_text: str) -> None:
    # 初始化存储文档信息的字典
+   stored_docs = {}
    # 如果文档信息文件已存在，则读取已有内容
+   if os.path.exists(config.DOCUMENTS_PKL):
+       with open(config.DOCUMENTS_PKL, "rb") as f:
+           stored_docs = pickle.load(f)
    # 将新的文档信息（只保存前200个字符）添加到字典
+   stored_docs[doc_id] = doc_text[:200]
    # 将更新后的字典写回pickle文件
+   with open(config.DOCUMENTS_PKL, "wb") as f:
+       pickle.dump(stored_docs, f)


# 获取文档列表
+def get_documents_list() -> List[Dict[str, str]]:
    # 初始化存储文档信息的字典
+   stored_docs = {}
    # 如果文档信息文件已存在，则读取已有内容
+   if os.path.exists(config.DOCUMENTS_PKL):
+       with open(config.DOCUMENTS_PKL, "rb") as f:
+           stored_docs = pickle.load(f)
    # 返回文档信息列表，每个元素为包含id和text的字典
+   return [
+       {"id": doc_id, "text": doc_text} for doc_id, doc_text in stored_docs.items()
+   ]

8.查询 #

8.1. config.py #

config.py

# 定义一个配置类，用于存放API相关的配置信息
class Config:
    # 设置Flask服务监听的主机地址为所有可用IP
    FLASK_HOST = "0.0.0.0"
    # 设置Flask服务监听的端口号为5601
    FLASK_PORT = 5601
    # 设置索引目录
    INDEX_DIR = "./saved_index"
    # 设置嵌入模型
    EMBEDDING_MODEL = "text-embedding-3-small"
    # 支持的文件格式
    SUPPORTED_FORMATS = [".pdf", ".txt", ".json", ".md"]
    # 设置文档目录
    DOCUMENTS_DIR = "documents"
    # 设置文档列表文件
    DOCUMENTS_PKL = "stored_documents.pkl"
    # 设置LLM模型
+   LLM_MODEL = "gpt-4o-mini"
    # 设置相似度top_k
+   SIMILARITY_TOP_K = 2


# 创建全局配置类的实例，方便在其他模块中引用
config = Config()

8.2. index_manager.py #

index_manager.py

# 导入OpenAIEmbedding类，用于生成文本的嵌入向量
from llama_index.embeddings.openai import OpenAIEmbedding

# 导入自定义的工具函数save_document_info和get_documents_list
from utils import save_document_info, get_documents_list

# 导入自定义的配置对象config
from config import config

# 导入OpenAI大语言模型
+from llama_index.llms.openai import OpenAI

# 导入操作系统相关模块，用于文件和目录操作
import os

# 从llama_index.core模块中导入所需的类和方法
from llama_index.core import (
    VectorStoreIndex,  # 向量存储索引类
    StorageContext,  # 存储上下文类
    load_index_from_storage,  # 从存储目录加载索引的方法
    SimpleDirectoryReader,  # 简单目录读取器，用于读取文件
)


# 定义索引管理器类
class IndexManager:
    """索引管理器"""

    # 构造方法，初始化索引管理器实例
    def __init__(self):
        # 初始化索引对象为None
        self.index = None
        # 创建OpenAI嵌入模型实例，模型名称从配置中获取
        self.embed_model = OpenAIEmbedding(model_name=config.EMBEDDING_MODEL)
        # 调用初始化索引的方法
        self.initialize_index()

    # 初始化索引的方法
    def initialize_index(self):
        # 判断索引目录是否存在
        if os.path.exists(config.INDEX_DIR):
            # 如果索引目录存在，则从存储目录加载已有索引
            self.index = load_index_from_storage(
                StorageContext.from_defaults(persist_dir=config.INDEX_DIR),
                embed_model=self.embed_model,
            )
        else:
            # 如果索引目录不存在，则新建一个空的向量存储索引
            self.index = VectorStoreIndex(nodes=[], embed_model=self.embed_model)
            # 持久化存储索引到指定目录
            self.index.storage_context.persist(persist_dir=config.INDEX_DIR)

    # 插入文档到索引的方法
    def insert_document(self, doc_file_path: str, filename: str):
        """插入文档到索引"""
        # 使用SimpleDirectoryReader读取指定文件，返回文档对象列表
        documents = SimpleDirectoryReader(input_files=[doc_file_path]).load_data()

        # 遍历所有文档对象，逐个插入到索引中
        for document in documents:
            self.index.insert(document)

        # 保存文档信息（文件名和文档前200个字符）
        save_document_info(filename, documents[0].text[:200])
        # 持久化存储索引到指定目录
        self.index.storage_context.persist(persist_dir=config.INDEX_DIR)

    # 获取文档列表的方法
    def get_documents_list(self):
        """获取文档列表"""
        # 调用工具函数获取文档列表
        return get_documents_list()

    # 定义查询索引的方法，接收查询文本作为参数
+   def query_index(self, query_text: str):
        # 方法说明：查询索引
+       """查询索引"""
        # 创建OpenAI大语言模型实例，模型名称从配置中获取
+       llm = OpenAI(model=config.LLM_MODEL)
        # 使用索引的查询引擎进行查询，设置相似度top_k和llm参数
+       response = self.index.as_query_engine(
+           similarity_top_k=config.SIMILARITY_TOP_K,
+           llm=llm,
+       ).query(query_text)
        # 返回查询结果
+       return response


# 创建一个全局的索引管理器实例，供其他模块调用
index_manager = IndexManager()

8.3. main.py #

main.py

# 从flask库中导入Flask、Response、request和jsonify
from flask import Flask, Response, request, jsonify

# 从config模块导入全局配置对象config
from config import config

# 导入json模块，用于数据序列化和反序列化
import json

# 从index_manager模块导入索引管理器实例
from index_manager import index_manager

# 从utils模块导入文件格式校验函数、格式化源节点和格式化响应
+from utils import is_supported_file, format_source_node, format_response

# 导入os模块，用于文件路径操作
import os

# 从werkzeug.utils导入secure_filename，用于安全处理文件名
from werkzeug.utils import secure_filename

# 创建Flask应用实例
app = Flask(__name__)


# 定义根路由"/"的处理函数
@app.route("/")
def home():
    # 构造返回的数据字典
    data = {"message": "LlamaIndex智能文档问答系统", "version": "1.0.0"}
    # 将数据字典序列化为JSON字符串并返回，设置响应类型为application/json
    return Response(json.dumps(data, ensure_ascii=False), mimetype="application/json")


# 定义文件上传接口，路由为/uploadFile，仅支持POST方法
@app.route("/uploadFile", methods=["POST"])
def upload_file():
    # 文件上传接口
    """文件上传接口"""
    # 判断请求中是否包含文件
    if "file" not in request.files:
        # 如果没有文件，返回错误信息
        return jsonify({"error": "请提供文件"}), 400

    # 获取上传的文件对象
    uploaded_file = request.files["file"]
    # 判断文件名是否为空
    if uploaded_file.filename == "":
        # 如果未选择文件，返回错误信息
        return jsonify({"error": "未选择文件"}), 400

    # 判断文件格式是否受支持
    if not is_supported_file(uploaded_file.filename):
        # 如果文件格式不支持，返回错误信息
        return jsonify({"error": "不支持的文件格式"}), 400

    try:
        # 使用secure_filename处理文件名，防止安全风险
        filename = secure_filename(uploaded_file.filename)
        # 拼接文件保存路径
        filepath = os.path.join(config.DOCUMENTS_DIR, filename)
        # 保存文件到指定路径
        uploaded_file.save(filepath)
        # 将文件插入索引
        index_manager.insert_document(filepath, filename)
        # 返回上传成功信息
        return jsonify({"message": "文件上传成功"})

    # 捕获异常并返回错误信息
    except Exception as e:
        return jsonify({"error": f"上传失败: {str(e)}"}), 500


# 定义获取文档列表的接口，路由为/getDocuments，仅支持GET方法
@app.route("/getDocuments", methods=["GET"])
def get_documents():
    # 获取文档列表接口
    """获取文档列表接口"""
    try:
        # 调用索引管理器获取文档列表
        documents = index_manager.get_documents_list()
        # 返回文档列表的json数据
        return jsonify(documents)
    except Exception as e:
        # 捕获异常并返回错误信息
        return jsonify({"error": f"获取文档列表失败: {str(e)}"}), 500


# 定义路由/query，指定仅支持GET方法
+@app.route("/query", methods=["GET"])
# 定义查询接口的处理函数
+def query_index():
    # 查询接口说明
+   """查询接口"""
    # 从请求参数中获取查询文本
+   query_text = request.args.get("text")
    # 如果没有提供查询文本，返回错误信息
+   if not query_text:
+       return jsonify({"error": "请提供查询文本"}), 400

+   try:
        # 调用索引管理器进行查询，获取响应结果
+       response = index_manager.query_index(query_text)

        # 初始化源节点列表
+       sources = []
        # 如果响应对象有source_nodes属性且不为空，则格式化每个源节点
+       if hasattr(response, "source_nodes") and response.source_nodes:
+           sources = [format_source_node(node) for node in response.source_nodes]

        # 格式化响应并返回json数据
+       return jsonify(format_response(str(response), sources))

    # 捕获异常并返回错误信息
+   except Exception as e:
+       return jsonify(format_response(f"查询出错: {str(e)}", [])), 500


# 判断当前模块是否为主程序入口
if __name__ == "__main__":
    # 确保文档目录存在，不存在则自动创建
    os.makedirs(config.DOCUMENTS_DIR, exist_ok=True)
    # 打印启动信息，显示访问地址
    print(f"启动Flask应用: http://{config.FLASK_HOST}:{config.FLASK_PORT}")
    # 启动Flask应用，监听指定主机和端口，开启debug模式
    app.run(host=config.FLASK_HOST, port=config.FLASK_PORT, debug=True)

8.4. utils.py #

utils.py

# 导入类型提示所需的List和Dict
+from typing import List, Dict, Any

# 导入操作系统相关模块
import os

# 导入全局配置对象
from config import config

# 导入pickle模块用于对象序列化
import pickle


# 检查文件格式是否受支持
def is_supported_file(filename: str) -> bool:
    # 返回文件名后缀是否在支持的格式列表中
    return any(filename.lower().endswith(ext) for ext in config.SUPPORTED_FORMATS)


# 保存文档信息到pickle文件
def save_document_info(doc_id: str, doc_text: str) -> None:
    # 初始化存储文档信息的字典
    stored_docs = {}
    # 如果文档信息文件已存在，则读取已有内容
    if os.path.exists(config.DOCUMENTS_PKL):
        with open(config.DOCUMENTS_PKL, "rb") as f:
            stored_docs = pickle.load(f)
    # 将新的文档信息（只保存前200个字符）添加到字典
    stored_docs[doc_id] = doc_text[:200]
    # 将更新后的字典写回pickle文件
    with open(config.DOCUMENTS_PKL, "wb") as f:
        pickle.dump(stored_docs, f)


# 获取文档列表
def get_documents_list() -> List[Dict[str, str]]:
    # 初始化存储文档信息的字典
    stored_docs = {}
    # 如果文档信息文件已存在，则读取已有内容
    if os.path.exists(config.DOCUMENTS_PKL):
        with open(config.DOCUMENTS_PKL, "rb") as f:
            stored_docs = pickle.load(f)
    # 返回文档信息列表，每个元素为包含id和text的字典
    return [
        {"id": doc_id, "text": doc_text} for doc_id, doc_text in stored_docs.items()
    ]


# 定义格式化源节点信息的函数，输入为node对象，返回字典
+def format_source_node(node) -> Dict[str, Any]:
    # 返回一个包含节点文本、相似度、文档ID、起始和结束位置的字典
+   return {
        # 节点的文本内容，转为字符串
+       "text": str(node.text),
        # 节点的相似度分数，保留两位小数
+       "similarity": round(node.score, 2),
        # 节点的文档ID，转为字符串
+       "doc_id": str(node.id_),
        # 起始位置，默认为0
+       "start": 0,
        # 结束位置，默认为0
+       "end": 0,
+   }


# 定义格式化API响应的函数，输入为文本和源节点列表，返回字典
+def format_response(text: str, sources: List[Dict[str, Any]]) -> Dict[str, Any]:
    # 返回一个包含响应文本和源节点信息的字典
+   return {"text": text, "sources": sources}

9.支持跨域请求 #

9.1. config.py #

config.py

# 定义一个配置类，用于存放API相关的配置信息
class Config:
    # 设置Flask服务监听的主机地址为所有可用IP
    FLASK_HOST = "0.0.0.0"
    # 设置Flask服务监听的端口号为5601
    FLASK_PORT = 5601
    # 设置索引目录
    INDEX_DIR = "./saved_index"
    # 设置嵌入模型
    EMBEDDING_MODEL = "text-embedding-3-small"
    # 支持的文件格式
    SUPPORTED_FORMATS = [".pdf", ".txt", ".json", ".md"]
    # 设置文档目录
    DOCUMENTS_DIR = "documents"
    # 设置文档列表文件
    DOCUMENTS_PKL = "stored_documents.pkl"
    # 设置LLM模型
    LLM_MODEL = "gpt-4o-mini"
    # 设置相似度top_k
    SIMILARITY_TOP_K = 2
    # 网络配置
+   CORS_ORIGINS = ["*"]


# 创建全局配置类的实例，方便在其他模块中引用
config = Config()

9.2. main.py #

main.py

# 从flask库中导入Flask、Response、request和jsonify
from flask import Flask, Response, request, jsonify

# 从config模块导入全局配置对象config
from config import config

# 导入json模块，用于数据序列化和反序列化
import json

# 从index_manager模块导入索引管理器实例
from index_manager import index_manager

# 从utils模块导入文件格式校验函数、格式化源节点和格式化响应
from utils import is_supported_file, format_source_node, format_response

# 导入os模块，用于文件路径操作
import os

# 从werkzeug.utils导入secure_filename，用于安全处理文件名
from werkzeug.utils import secure_filename

# 导入CORS扩展，用于支持跨域请求
+from flask_cors import CORS

# 创建Flask应用实例
app = Flask(__name__)
# 配置CORS，允许所有来源的请求
+CORS(app, origins=config.CORS_ORIGINS)


# 定义根路由"/"的处理函数
@app.route("/")
def home():
    # 构造返回的数据字典
    data = {"message": "LlamaIndex智能文档问答系统", "version": "1.0.0"}
    # 将数据字典序列化为JSON字符串并返回，设置响应类型为application/json
    return Response(json.dumps(data, ensure_ascii=False), mimetype="application/json")


# 定义文件上传接口，路由为/uploadFile，仅支持POST方法
@app.route("/uploadFile", methods=["POST"])
def upload_file():
    # 文件上传接口
    """文件上传接口"""
    # 判断请求中是否包含文件
    if "file" not in request.files:
        # 如果没有文件，返回错误信息
        return jsonify({"error": "请提供文件"}), 400

    # 获取上传的文件对象
    uploaded_file = request.files["file"]
    # 判断文件名是否为空
    if uploaded_file.filename == "":
        # 如果未选择文件，返回错误信息
        return jsonify({"error": "未选择文件"}), 400

    # 判断文件格式是否受支持
    if not is_supported_file(uploaded_file.filename):
        # 如果文件格式不支持，返回错误信息
        return jsonify({"error": "不支持的文件格式"}), 400

    try:
        # 使用secure_filename处理文件名，防止安全风险
        filename = secure_filename(uploaded_file.filename)
        # 拼接文件保存路径
        filepath = os.path.join(config.DOCUMENTS_DIR, filename)
        # 保存文件到指定路径
        uploaded_file.save(filepath)
        # 将文件插入索引
        index_manager.insert_document(filepath, filename)
        # 返回上传成功信息
        return jsonify({"message": "文件上传成功"})

    # 捕获异常并返回错误信息
    except Exception as e:
        return jsonify({"error": f"上传失败: {str(e)}"}), 500


# 定义获取文档列表的接口，路由为/getDocuments，仅支持GET方法
@app.route("/getDocuments", methods=["GET"])
def get_documents():
    # 获取文档列表接口
    """获取文档列表接口"""
    try:
        # 调用索引管理器获取文档列表
        documents = index_manager.get_documents_list()
        # 返回文档列表的json数据
        return jsonify(documents)
    except Exception as e:
        # 捕获异常并返回错误信息
        return jsonify({"error": f"获取文档列表失败: {str(e)}"}), 500


# 定义路由/query，指定仅支持GET方法
@app.route("/query", methods=["GET"])
# 定义查询接口的处理函数
def query_index():
    # 查询接口说明
    """查询接口"""
    # 从请求参数中获取查询文本
    query_text = request.args.get("text")
    # 如果没有提供查询文本，返回错误信息
    if not query_text:
        return jsonify({"error": "请提供查询文本"}), 400

    try:
        # 调用索引管理器进行查询，获取响应结果
        response = index_manager.query_index(query_text)

        # 初始化源节点列表
        sources = []
        # 如果响应对象有source_nodes属性且不为空，则格式化每个源节点
        if hasattr(response, "source_nodes") and response.source_nodes:
            sources = [format_source_node(node) for node in response.source_nodes]

        # 格式化响应并返回json数据
        return jsonify(format_response(str(response), sources))

    # 捕获异常并返回错误信息
    except Exception as e:
        return jsonify(format_response(f"查询出错: {str(e)}", [])), 500


# 判断当前模块是否为主程序入口
if __name__ == "__main__":
    # 确保文档目录存在，不存在则自动创建
    os.makedirs(config.DOCUMENTS_DIR, exist_ok=True)
    # 打印启动信息，显示访问地址
    print(f"启动Flask应用: http://{config.FLASK_HOST}:{config.FLASK_PORT}")
    # 启动Flask应用，监听指定主机和端口，开启debug模式
    app.run(host=config.FLASK_HOST, port=config.FLASK_PORT, debug=True)