Python: 使用 Flask 與 OpenAI API 建立智能問答系統

import os
import json
from flask import Flask, request, jsonify, render_template
from openai import OpenAI
from typing import List, Dict
import sys
import io

# 將默認編碼設置為 UTF-8
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")

# 設置目錄路徑和文件名
dirname = r"D:\user\Python\GPT\json"
basename_api_key = "api_key.json"
basename_database = "database.json"

path_api_key = os.path.join(dirname, basename_api_key)
path_database = os.path.join(dirname, basename_database)

# 獲取 API Key
def get_api_key(path_api_key=path_api_key):
    with open(path_api_key, "r", encoding="UTF-8") as f:
        data = json.load(f)
        api_key = data["api_key"]
        return api_key

api_key = get_api_key(path_api_key=path_api_key)

# 配置 OpenAI 客戶端
client = OpenAI(api_key=api_key)

# 加載 JSON 資料庫
def load_jsonDB(file_path=path_database):
    with open(file_path, "r", encoding="utf-8") as f:
        database = json.load(f)
    return database

# 初始化資料庫
database: List[Dict[str, str]] = load_jsonDB(file_path=path_database)

# 查詢 JSON 資料庫
def query_database(question: str) -> List[Dict[str, str]]:
    # 將問題拆分為關鍵詞，並匹配資料庫內容中是否包含關鍵詞
    results = [dic for dic in database if any(word in dic["content"].lower() for word in question.lower().split())]
    return results

# 初始化 Flask 應用
app = Flask(__name__)

# GPT 問答接口
@app.route("/ask", methods=["POST"])
def ask_gpt():
    try:
        # 獲取用戶問題
        data = request.json  # 從用戶的 POST 請求中提取 JSON 數據
        user_question = data.get("question", "")

        # 查詢 JSON 資料庫
        db_results = query_database(user_question)

        # 構造 GPT 提示
        if db_results:
            # 如果資料庫有匹配結果
            gpt_prompt = f"資料庫查詢結果：{db_results}\n\n用戶問題：{user_question}\n\n請根據資料庫內容回答："
        else:
            # 如果資料庫沒有匹配結果，讓 GPT 自行回答
            gpt_prompt = f"用戶問題：{user_question}\n\n資料庫中沒有相關內容，請根據一般知識回答："

        # 向 GPT 發送請求
        response = client.chat.completions.create(
            model="gpt-4-32k",
            messages=[
                {"role": "system", "content": "你是專業的問題回答助手，請根據用戶的問題給出準確的回答。"},
                {"role": "user", "content": gpt_prompt},
            ],
        )
        print("response:\n",response)
        # 提取 GPT 的回答
        answer = response.choices[0].message.content
        print("GPT 回答：", answer)

        # 返回 JSON 格式的回答
        return jsonify({"answer": answer})

    except Exception as e:
        # 返回錯誤信息
        return jsonify({"error": str(e)}), 500

# 網頁接口
@app.route("/")
def index():
    return render_template("index.html")  # 加載 templates/index.html 文件

# 啟動 Flask 服務器
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=True)

{
    "api_key": "your_openai_api_key_here"
}

GPT/
├── app.py          # 主程式
├── json/           # 資料庫與 API Key 文件存放
│   ├── api_key.json
│   ├── database.json
├── templates/      # 前端 HTML 文件存放
│   └── index.html

# 將默認編碼設置為 UTF-8
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")

dirname = r"D:\user\Python\GPT\json"
basename_api_key = "api_key.json"
basename_database = "database.json"

path_api_key = os.path.join(dirname, basename_api_key)
path_database = os.path.join(dirname, basename_database)

def get_api_key(path_api_key=path_api_key):
    with open(path_api_key, "r", encoding="UTF-8") as f:
        data = json.load(f)
        api_key = data["api_key"]
        return api_key

api_key = get_api_key(path_api_key=path_api_key)

def load_jsonDB(file_path=path_database):
    with open(file_path, "r", encoding="utf-8") as f:
        database = json.load(f)
    return database

database: List[Dict[str, str]] = load_jsonDB(file_path=path_database)

[
{"content": "Flask 是一個 Python 的 Web 框架。"},
{"content": "GPT 是一種基於 Transformer 的大型語言模型。"},
{"content": "JSON Line 是一種每行為一個 JSON 對象的文件格式。"}
]

def query_database(question: str) -> List[Dict[str, str]]:
    results = [dic for dic in database if any(word in dic["content"].lower() for word in question.lower().split())]
    return results

@app.route("/ask", methods=["POST"])
def ask_gpt():
    try:
        data = request.json
        user_question = data.get("question", "")
        db_results = query_database(user_question)

        if db_results:
            gpt_prompt = f"資料庫查詢結果：{db_results}\n\n用戶問題：{user_question}\n\n請根據資料庫內容回答："
        else:
            gpt_prompt = f"用戶問題：{user_question}\n\n資料庫中沒有相關內容，請根據一般知識回答："

        response = client.chat.completions.create(
            model="gpt-4-32k",
            messages=[
                {"role": "system", "content": "你是專業的問題回答助手，請根據用戶的問題給出準確的回答。"},
                {"role": "user", "content": gpt_prompt},
            ],
        )
        answer = response.choices[0].message.content
        return jsonify({"answer": answer})

    except Exception as e:
        return jsonify({"error": str(e)}), 500

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

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=True)