AWS部署基于Pickle的Flask搜索应用遇AttributeError求助

问题背景 #

基于Pickle实现搜索功能的Flask应用本地运行正常，部署到AWS服务器后出现AttributeError，提示无法在__main__模块找到Document属性。尝试将Pickle文件加载代码放入if __name__ == '__main__':块后仍未解决，网页显示Internal Server Error。

错误原因 #

1. Pickle反序列化的模块匹配问题：Pickle序列化对象时会记录类所在的模块名。本地运行时脚本以__main__模块执行，序列化的Document对象关联模块为__main__；但AWS上用WSGI服务器（如Gunicorn、Apache）运行时，脚本以实际模块名（如app）加载，反序列化时找不到__main__.Document，触发报错。
2. 变量作用域错误：修改后的代码将docs和myvector的加载放入if __name__ == '__main__':块，但WSGI服务器启动应用时不会执行该代码块，导致search函数中引用myvector和docs时出现未定义错误。
3. 代码冗余：原代码每次搜索都会重新加载JSON、重建FAISS索引，不仅性能低下，还可能引发文件读取异常。

解决方案 #

步骤1：将自定义类拆分到独立模块 #

创建storage.py文件，存放Document和VectorStore类，确保序列化与反序列化时模块名一致：

# storage.py
class Document:
    def __init__(self, title, url, content):
        self.title = title
        self.url = url
        self.content = content

class VectorStore:
    def __init__(self):
        self.documents = []
        self.embeddings = np.empty((0,384))

    def add_to_store(self, document):
        self.documents.append(document)
        # 若需保留该方法，需在当前模块导入generate_embedding或调整逻辑

注：原代码中VectorStore的add_to_store方法存在缩进错误，已修正。

步骤2：修正主应用的变量加载与作用域 #

在主应用中提前加载模型、数据和索引，避免每次搜索重复操作：

# app.py
from flask import Flask, render_template, request, jsonify
import numpy as np
import json
from sentence_transformers import SentenceTransformer
import faiss
import pickle
from storage import Document  # 从独立模块导入Document类

app = Flask(__name__, template_folder='templates')

# 提前加载依赖资源，避免重复开销
model = SentenceTransformer("multi-qa-MiniLM-L6-cos-v1", device="cpu")

# 加载Pickle文档与向量数据
with open('docs.pkl', 'rb') as file:
    docs = pickle.load(file)

with open('output_data.json', 'r') as f:
    myvector = json.load(f)
feature_batch = np.array(myvector)

# 初始化FAISS索引
index = faiss.IndexFlatL2(feature_batch.shape[1])
index.add(feature_batch)
faiss.write_index(index, 'index.faiss')

def generate_embedding(text):
    response = model.encode([text])
    return np.array(response[0])

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/search', methods=['POST'])
def search():
    query = request.form.get('query')
    myresults = []
    similarityThreshold = 1

    query_embedding = generate_embedding(query)
    distances, results = index.search(np.array([query_embedding]), k=3)

    filtered_results = []
    for i, distance in zip(results[0], distances[0]):
        if distance <= similarityThreshold:
            filtered_results.append(i)

    for i in filtered_results:
        myresults.append(docs[i].content)

    return render_template('index.html', myresults=myresults)

if __name__ == '__main__':
    app.run(debug=True)

额外注意事项 #

• 确保AWS服务器安装所有依赖包：flask、numpy、sentence-transformers、faiss-cpu（或对应版本的FAISS）。
• 检查文件路径：docs.pkl、output_data.json、templates文件夹在AWS服务器上的路径正确，且应用有读取权限。
• 生产环境关闭debug模式：将app.run(debug=True)改为app.run(debug=False)，避免安全风险。

内容的提问来源于stack exchange，提问作者gopal