人工智慧技術的快速發展為網頁應用程式開發帶來了革命性的變化。OpenAI 推出的 GPT 系列模型和 DALL-E 圖片生成模型，讓開發者能夠輕鬆地為應用程式添加自然語言處理和圖像生成能力。Flask 作為 Python 生態系中最受歡迎的微框架之一，以其輕量和靈活的特性，成為整合這些 AI 能力的理想選擇。
本文將深入探討如何使用 Flask 框架整合 OpenAI 的各項服務，建立功能完善的互動式網頁應用程式。我們將從基礎的聊天機器人開始，逐步擴展到圖片生成功能，最後整合 Elasticsearch 實現全文檢索。過程中，我們會特別關注 API 安全性、錯誤處理、效能最佳化等企業級開發實踐，確保所建立的應用程式能夠在生產環境中穩定運作。
專案架構與環境設定
在開始開發之前，良好的專案架構設計是成功的關鍵。一個設計良好的專案架構不僅讓程式碼更容易維護，也讓團隊協作更加順暢。對於整合多種外部服務的應用程式來說，適當的模組劃分尤為重要。
@startuml
!define DISABLE_LINK
!define PLANTUML_FORMAT svg
!theme _none_
skinparam dpi auto
skinparam shadowing false
skinparam linetype ortho
skinparam roundcorner 5
skinparam defaultFontName "Microsoft JhengHei UI"
skinparam defaultFontSize 16
skinparam minClassWidth 100
package "Flask GPT 應用架構" {
folder "my_app" {
file "__init__.py" as init
file "config.py" as config
folder "catalog" {
file "models.py" as models
file "views.py" as views
file "forms.py" as forms
}
folder "services" {
file "openai_service.py" as openai
file "elasticsearch_service.py" as es
}
folder "templates" {
file "base.html" as base
file "chat.html" as chat
file "product.html" as product
}
folder "static" {
file "js/chat.js" as js
file "css/style.css" as css
}
}
}
init -down-> config : 載入設定
views -right-> openai : 呼叫 AI 服務
views -right-> es : 呼叫搜尋服務
views -down-> models : 資料操作
@enduml專案架構採用了服務層（Service Layer）設計模式，將與外部 API 的互動封裝在獨立的服務模組中。這種設計有幾個重要的優點。首先，它實現了關注點分離，讓視圖層專注於處理 HTTP 請求和回應，而服務層負責業務邏輯和外部整合。其次，它提高了程式碼的可測試性，因為服務層可以獨立進行單元測試。最後，當需要替換外部服務或添加快取層時，只需要修改服務模組，不影響其他部分。
# config.py
# 這個模組包含了應用程式的所有配置設定
# 使用環境變數來管理敏感資訊是安全最佳實踐
import os
from datetime import timedelta
class Config:
"""
應用程式基礎配置類別
所有的敏感資訊都從環境變數讀取，
這樣可以避免將密鑰硬編碼在程式碼中。
"""
# Flask 基本設定
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key-change-in-production'
# 資料庫設定
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///app.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False
# OpenAI API 設定
# 這是與 OpenAI 服務通訊的核心配置
OPENAI_API_KEY = os.environ.get('OPENAI_API_KEY')
OPENAI_MODEL = os.environ.get('OPENAI_MODEL', 'gpt-4')
OPENAI_MAX_TOKENS = int(os.environ.get('OPENAI_MAX_TOKENS', 2000))
OPENAI_TEMPERATURE = float(os.environ.get('OPENAI_TEMPERATURE', 0.7))
# 圖片生成設定
DALL_E_MODEL = os.environ.get('DALL_E_MODEL', 'dall-e-3')
DALL_E_SIZE = os.environ.get('DALL_E_SIZE', '1024x1024')
DALL_E_QUALITY = os.environ.get('DALL_E_QUALITY', 'standard')
# 檔案上傳設定
UPLOAD_FOLDER = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'static', 'uploads')
MAX_CONTENT_LENGTH = 16 * 1024 * 1024 # 限制上傳檔案大小為 16MB
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif'}
# Elasticsearch 設定
ELASTICSEARCH_URL = os.environ.get('ELASTICSEARCH_URL', 'https://localhost:9200')
ELASTICSEARCH_USER = os.environ.get('ELASTICSEARCH_USER', 'elastic')
ELASTICSEARCH_PASSWORD = os.environ.get('ELASTICSEARCH_PASSWORD')
ELASTICSEARCH_CA_CERTS = os.environ.get('ELASTICSEARCH_CA_CERTS')
ELASTICSEARCH_INDEX = os.environ.get('ELASTICSEARCH_INDEX', 'catalog')
# 快取設定
CACHE_TYPE = 'RedisCache'
CACHE_REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
CACHE_DEFAULT_TIMEOUT = 300
# 速率限制設定
# 這些設定用於防止 API 被濫用
RATELIMIT_DEFAULT = "100 per day;30 per hour"
RATELIMIT_STORAGE_URL = os.environ.get('REDIS_URL', 'memory://')
class DevelopmentConfig(Config):
"""開發環境配置"""
DEBUG = True
OPENAI_MODEL = 'gpt-3.5-turbo' # 開發時使用較便宜的模型
class ProductionConfig(Config):
"""生產環境配置"""
DEBUG = False
# 在生產環境中強制要求 HTTPS
SESSION_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SAMESITE = 'Lax'
class TestingConfig(Config):
"""測試環境配置"""
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
WTF_CSRF_ENABLED = False
# 配置對映
config = {
'development': DevelopmentConfig,
'production': ProductionConfig,
'testing': TestingConfig,
'default': DevelopmentConfig
}
配置管理是任何應用程式的基礎。透過將配置分離為不同的類別，我們可以輕鬆地在不同環境之間切換，同時確保敏感資訊不會洩漏到版本控制系統中。使用環境變數來管理 API 金鑰和密碼是業界標準做法，這樣可以在不修改程式碼的情況下部署到不同的環境。
建立 OpenAI 服務層
服務層的設計是整個應用程式的核心。一個設計良好的服務層應該封裝所有與 OpenAI API 的互動細節，包括錯誤處理、重試邏輯、日誌記錄等。這樣，視圖層就可以專注於處理使用者請求，而不需要關心底層 API 的細節。
# my_app/services/openai_service.py
# 這個模組封裝了所有與 OpenAI API 的互動
import openai
import logging
import time
from functools import wraps
from flask import current_app
from typing import Generator, Dict, List, Optional
import httpx
# 設定日誌記錄器
logger = logging.getLogger(__name__)
class OpenAIServiceError(Exception):
"""OpenAI 服務相關錯誤的基礎類別"""
pass
class RateLimitError(OpenAIServiceError):
"""API 速率限制錯誤"""
pass
class APIConnectionError(OpenAIServiceError):
"""API 連線錯誤"""
pass
def retry_on_error(max_retries=3, delay=1, backoff=2):
"""
重試裝飾器
當 API 呼叫失敗時自動重試，使用指數退避策略。
這對於處理暫時性的網路問題或 API 速率限制特別有用。
參數:
max_retries: 最大重試次數
delay: 初始延遲時間（秒）
backoff: 退避倍數
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
retries = 0
current_delay = delay
while retries < max_retries:
try:
return func(*args, **kwargs)
except openai.RateLimitError as e:
retries += 1
if retries >= max_retries:
logger.error(f'API 速率限制錯誤，已達最大重試次數: {e}')
raise RateLimitError(f'API 速率限制: {str(e)}')
logger.warning(f'API 速率限制，{current_delay} 秒後重試...')
time.sleep(current_delay)
current_delay *= backoff
except openai.APIConnectionError as e:
retries += 1
if retries >= max_retries:
logger.error(f'API 連線錯誤，已達最大重試次數: {e}')
raise APIConnectionError(f'API 連線失敗: {str(e)}')
logger.warning(f'API 連線錯誤，{current_delay} 秒後重試...')
time.sleep(current_delay)
current_delay *= backoff
return func(*args, **kwargs)
return wrapper
return decorator
class OpenAIService:
"""
OpenAI 服務類別
封裝所有與 OpenAI API 的互動，包括聊天完成和圖片生成。
提供錯誤處理、重試邏輯和日誌記錄功能。
"""
def __init__(self, app=None):
"""
初始化 OpenAI 服務
參數:
app: Flask 應用程式實例
"""
self.client = None
if app is not None:
self.init_app(app)
def init_app(self, app):
"""
使用 Flask 應用程式配置初始化服務
這個方法會讀取應用程式配置中的 OpenAI 相關設定，
並建立 OpenAI 客戶端實例。
"""
api_key = app.config.get('OPENAI_API_KEY')
if not api_key:
logger.warning('OPENAI_API_KEY 未設定，OpenAI 功能將無法使用')
return
# 建立 OpenAI 客戶端
# 使用 httpx 客戶端可以自定義超時和重試設定
self.client = openai.OpenAI(
api_key=api_key,
timeout=httpx.Timeout(60.0, connect=5.0),
max_retries=3
)
# 儲存配置
self._model = app.config.get('OPENAI_MODEL', 'gpt-4')
self._max_tokens = app.config.get('OPENAI_MAX_TOKENS', 2000)
self._temperature = app.config.get('OPENAI_TEMPERATURE', 0.7)
self._dalle_model = app.config.get('DALL_E_MODEL', 'dall-e-3')
self._dalle_size = app.config.get('DALL_E_SIZE', '1024x1024')
self._dalle_quality = app.config.get('DALL_E_QUALITY', 'standard')
logger.info(f'OpenAI 服務已初始化，使用模型: {self._model}')
@retry_on_error(max_retries=3, delay=1, backoff=2)
def chat_completion(
self,
messages: List[Dict[str, str]],
model: Optional[str] = None,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None,
stream: bool = False
):
"""
呼叫 ChatCompletion API
這是與 GPT 模型互動的主要方法。支援一般模式和串流模式。
參數:
messages: 對話訊息列表，每個訊息包含 role 和 content
model: 使用的模型，預設使用配置中的設定
temperature: 生成的隨機性，0 最確定，1 最隨機
max_tokens: 回應的最大 token 數
stream: 是否使用串流模式
回傳:
如果 stream=False，回傳完整的回應文字
如果 stream=True，回傳一個生成器
"""
if not self.client:
raise OpenAIServiceError('OpenAI 客戶端未初始化')
# 使用預設值或傳入的參數
model = model or self._model
temperature = temperature if temperature is not None else self._temperature
max_tokens = max_tokens or self._max_tokens
logger.info(f'呼叫 ChatCompletion API，模型: {model}, 訊息數: {len(messages)}')
response = self.client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
stream=stream
)
if stream:
# 串流模式回傳生成器
return self._stream_response(response)
else:
# 一般模式回傳完整文字
content = response.choices[0].message.content
logger.info(f'ChatCompletion 完成，回應長度: {len(content)} 字元')
return content
def _stream_response(self, response) -> Generator[str, None, None]:
"""
處理串流回應
將 OpenAI 的串流回應轉換為字串生成器，
方便視圖層進行 Server-Sent Events (SSE) 處理。
參數:
response: OpenAI 的串流回應物件
產出:
每次產出一個文字片段
"""
try:
for chunk in response:
if chunk.choices[0].delta.content is not None:
yield chunk.choices[0].delta.content
except Exception as e:
logger.error(f'串流回應處理錯誤: {e}')
raise OpenAIServiceError(f'串流處理失敗: {str(e)}')
@retry_on_error(max_retries=2, delay=2, backoff=2)
def generate_image(
self,
prompt: str,
model: Optional[str] = None,
size: Optional[str] = None,
quality: Optional[str] = None,
n: int = 1
) -> List[str]:
"""
使用 DALL-E 生成圖片
根據文字提示生成圖片，回傳圖片的 URL 列表。
參數:
prompt: 圖片描述文字
model: 使用的模型（dall-e-2 或 dall-e-3）
size: 圖片尺寸
quality: 圖片品質（standard 或 hd）
n: 生成的圖片數量
回傳:
圖片 URL 列表
"""
if not self.client:
raise OpenAIServiceError('OpenAI 客戶端未初始化')
model = model or self._dalle_model
size = size or self._dalle_size
quality = quality or self._dalle_quality
logger.info(f'呼叫 Image API，提示: {prompt[:50]}...')
response = self.client.images.generate(
model=model,
prompt=prompt,
size=size,
quality=quality,
n=n
)
urls = [image.url for image in response.data]
logger.info(f'圖片生成完成，數量: {len(urls)}')
return urls
def create_embeddings(self, texts: List[str], model: str = 'text-embedding-3-small') -> List[List[float]]:
"""
建立文字嵌入向量
將文字轉換為數值向量，可用於語義搜尋和相似度比對。
參數:
texts: 要轉換的文字列表
model: 嵌入模型
回傳:
嵌入向量列表
"""
if not self.client:
raise OpenAIServiceError('OpenAI 客戶端未初始化')
response = self.client.embeddings.create(
model=model,
input=texts
)
return [item.embedding for item in response.data]
# 建立全域服務實例
openai_service = OpenAIService()
這個服務層的設計包含了幾個重要的特性。首先，重試機制使用指數退避策略，這是處理 API 速率限制和暫時性錯誤的最佳實踐。其次，完善的日誌記錄幫助追蹤 API 呼叫和診斷問題。第三，串流支援允許即時顯示 AI 生成的回應，提升使用者體驗。最後，靈活的配置允許在執行時覆蓋預設設定。
實作聊天機器人功能
聊天機器人是 GPT 整合最常見的應用場景之一。一個好的聊天機器人不僅要能夠理解使用者的問題並給出有用的回答，還需要維護對話上下文，提供流暢的對話體驗。
# my_app/catalog/views.py
# 這個模組包含了所有的視圖函式
from flask import (
Blueprint, request, jsonify, render_template,
current_app, Response, stream_with_context, session
)
from flask_login import login_required, current_user
from my_app.services.openai_service import openai_service, OpenAIServiceError
from my_app import db, cache
from my_app.catalog.models import ChatHistory
from datetime import datetime
import json
catalog = Blueprint('catalog', __name__)
@catalog.route('/chat', methods=['GET'])
@login_required
def chat_page():
"""
顯示聊天頁面
載入使用者的歷史對話記錄，提供持續的對話體驗。
"""
# 從資料庫載入最近的對話記錄
history = ChatHistory.query.filter_by(
user_id=current_user.id
).order_by(
ChatHistory.created_at.desc()
).limit(50).all()
# 反轉順序以便按時間順序顯示
history = list(reversed(history))
return render_template('chat.html', history=history)
@catalog.route('/chat/message', methods=['POST'])
@login_required
def chat_message():
"""
處理聊天訊息
接收使用者訊息，呼叫 GPT API 生成回應，
並儲存對話記錄到資料庫。
請求主體:
message: 使用者的訊息
conversation_id: 對話 ID（可選）
回傳:
AI 生成的回應
"""
data = request.get_json()
if not data or 'message' not in data:
return jsonify({'error': '請提供訊息內容'}), 400
user_message = data['message'].strip()
if not user_message:
return jsonify({'error': '訊息不能為空'}), 400
if len(user_message) > 4000:
return jsonify({'error': '訊息長度超過限制'}), 400
# 取得或建立對話 ID
conversation_id = data.get('conversation_id') or session.get('conversation_id')
if not conversation_id:
conversation_id = f"{current_user.id}_{datetime.utcnow().timestamp()}"
session['conversation_id'] = conversation_id
try:
# 建立對話訊息列表
# 系統訊息定義了 AI 助手的行為和限制
messages = [
{
"role": "system",
"content": """你是一個專業的電子商務網站客服助手。你的職責是：
1. 回答關於產品的問題
2. 協助處理訂單查詢
3. 提供購物建議
4. 解答常見問題
請保持專業、友善的態度。如果遇到無法回答的問題，
請建議使用者聯繫人工客服。回答請使用繁體中文。"""
}
]
# 載入最近的對話歷史以維護上下文
recent_history = ChatHistory.query.filter_by(
user_id=current_user.id,
conversation_id=conversation_id
).order_by(
ChatHistory.created_at.desc()
).limit(10).all()
# 將歷史記錄添加到訊息列表
for record in reversed(recent_history):
messages.append({"role": "user", "content": record.user_message})
messages.append({"role": "assistant", "content": record.assistant_message})
# 添加當前使用者訊息
messages.append({"role": "user", "content": user_message})
# 呼叫 OpenAI API
assistant_message = openai_service.chat_completion(
messages=messages,
temperature=0.7,
max_tokens=1000
)
# 儲存對話記錄
chat_record = ChatHistory(
user_id=current_user.id,
conversation_id=conversation_id,
user_message=user_message,
assistant_message=assistant_message
)
db.session.add(chat_record)
db.session.commit()
current_app.logger.info(f'使用者 {current_user.id} 聊天完成')
return jsonify({
'message': assistant_message,
'conversation_id': conversation_id
})
except OpenAIServiceError as e:
current_app.logger.error(f'OpenAI 服務錯誤: {e}')
return jsonify({'error': '抱歉，AI 服務暫時無法使用，請稍後再試'}), 503
except Exception as e:
current_app.logger.error(f'聊天處理錯誤: {e}')
db.session.rollback()
return jsonify({'error': '處理訊息時發生錯誤'}), 500
@catalog.route('/chat/stream', methods=['POST'])
@login_required
def chat_stream():
"""
串流模式聊天
使用 Server-Sent Events (SSE) 即時傳送 AI 生成的回應。
這提供了更好的使用者體驗，讓使用者可以看到 AI 逐字生成回應。
回傳:
SSE 串流回應
"""
data = request.get_json()
if not data or 'message' not in data:
return jsonify({'error': '請提供訊息內容'}), 400
user_message = data['message'].strip()
def generate():
"""
生成 SSE 事件
這個生成器函式會逐個產出 AI 的回應片段，
讓前端可以即時顯示生成的內容。
"""
try:
messages = [
{
"role": "system",
"content": "你是一個專業的電子商務網站客服助手，使用繁體中文回答。"
},
{"role": "user", "content": user_message}
]
# 使用串流模式呼叫 API
full_response = []
for chunk in openai_service.chat_completion(
messages=messages,
stream=True
):
full_response.append(chunk)
# 使用 SSE 格式傳送資料
yield f"data: {json.dumps({'content': chunk})}\n\n"
# 傳送結束信號
yield f"data: {json.dumps({'done': True})}\n\n"
# 儲存完整的回應到資料庫
complete_message = ''.join(full_response)
chat_record = ChatHistory(
user_id=current_user.id,
user_message=user_message,
assistant_message=complete_message
)
db.session.add(chat_record)
db.session.commit()
except OpenAIServiceError as e:
yield f"data: {json.dumps({'error': str(e)})}\n\n"
except Exception as e:
current_app.logger.error(f'串流處理錯誤: {e}')
yield f"data: {json.dumps({'error': '處理訊息時發生錯誤'})}\n\n"
return Response(
stream_with_context(generate()),
mimetype='text/event-stream',
headers={
'Cache-Control': 'no-cache',
'X-Accel-Buffering': 'no' # 禁用 Nginx 緩衝
}
)
@catalog.route('/chat/clear', methods=['POST'])
@login_required
def clear_chat_history():
"""
清除對話歷史
刪除使用者的所有對話記錄，開始新的對話。
"""
try:
ChatHistory.query.filter_by(user_id=current_user.id).delete()
db.session.commit()
# 清除 session 中的對話 ID
session.pop('conversation_id', None)
return jsonify({'message': '對話記錄已清除'})
except Exception as e:
current_app.logger.error(f'清除對話歷史錯誤: {e}')
db.session.rollback()
return jsonify({'error': '清除對話記錄時發生錯誤'}), 500
聊天功能的實作包含了幾個重要的特性。對話上下文維護是透過載入最近的對話記錄來實現的，這讓 AI 能夠理解之前的對話內容，提供更連貫的回答。串流模式使用 Server-Sent Events 技術，讓使用者可以看到 AI 逐字生成回應，這大大改善了等待體驗。錯誤處理確保即使 API 呼叫失敗，應用程式也能優雅地處理並向使用者提供有用的錯誤訊息。
前端聊天介面實作
一個好的使用者介面對於聊天應用程式至關重要。前端需要處理使用者輸入、顯示對話歷史、處理串流回應，並提供良好的視覺回饋。
<!-- my_app/templates/chat.html -->
<!-- 聊天介面範本 -->
{% extends 'base.html' %}
{% block content %}
<div class="chat-container">
<div class="chat-header">
<h2>AI 客服助手</h2>
<button id="clear-chat" class="btn btn-outline-secondary btn-sm">
清除對話
</button>
</div>
<div class="chat-messages" id="chat-messages">
<!-- 顯示歷史訊息 -->
{% for record in history %}
<div class="message user-message">
<div class="message-content">{{ record.user_message }}</div>
<div class="message-time">{{ record.created_at.strftime('%H:%M') }}</div>
</div>
<div class="message assistant-message">
<div class="message-content">{{ record.assistant_message }}</div>
<div class="message-time">{{ record.created_at.strftime('%H:%M') }}</div>
</div>
{% endfor %}
</div>
<div class="chat-input-container">
<div class="typing-indicator" id="typing-indicator" style="display: none;">
<span></span>
<span></span>
<span></span>
</div>
<form id="chat-form" class="chat-input-form">
<textarea
id="message-input"
class="chat-input"
placeholder="輸入您的問題..."
rows="1"
maxlength="4000"
></textarea>
<button type="submit" id="send-button" class="send-button">
<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="currentColor">
<path d="M2.01 21L23 12 2.01 3 2 10l15 2-15 2z"/>
</svg>
</button>
</form>
</div>
</div>
{% endblock %}
{% block scripts %}
<script>
// 聊天功能的 JavaScript 實作
// 這段程式碼處理使用者輸入、API 呼叫和介面更新
(function() {
'use strict';
// DOM 元素參照
const chatMessages = document.getElementById('chat-messages');
const chatForm = document.getElementById('chat-form');
const messageInput = document.getElementById('message-input');
const sendButton = document.getElementById('send-button');
const typingIndicator = document.getElementById('typing-indicator');
const clearChatButton = document.getElementById('clear-chat');
// 對話 ID
let conversationId = null;
// 捲動到最新訊息
function scrollToBottom() {
chatMessages.scrollTop = chatMessages.scrollHeight;
}
// 將訊息添加到聊天視窗
function appendMessage(content, isUser, isStreaming = false) {
const messageDiv = document.createElement('div');
messageDiv.className = `message ${isUser ? 'user-message' : 'assistant-message'}`;
const contentDiv = document.createElement('div');
contentDiv.className = 'message-content';
if (isStreaming) {
// 串流模式下先建立空的內容容器
contentDiv.id = 'streaming-message';
} else {
// 處理 Markdown 格式（簡單版本）
contentDiv.innerHTML = formatMessage(content);
}
const timeDiv = document.createElement('div');
timeDiv.className = 'message-time';
timeDiv.textContent = new Date().toLocaleTimeString('zh-TW', {
hour: '2-digit',
minute: '2-digit'
});
messageDiv.appendChild(contentDiv);
messageDiv.appendChild(timeDiv);
chatMessages.appendChild(messageDiv);
scrollToBottom();
return contentDiv;
}
// 簡單的訊息格式化（處理換行和程式碼區塊）
function formatMessage(content) {
// 轉義 HTML
content = content
.replace(/&/g, '&')
.replace(/</g, '<')
.replace(/>/g, '>');
// 處理程式碼區塊
content = content.replace(/```(\w*)\n([\s\S]*?)```/g,
'<pre><code class="language-$1">$2</code></pre>');
// 處理行內程式碼
content = content.replace(/`([^`]+)`/g, '<code>$1</code>');
// 處理換行
content = content.replace(/\n/g, '<br>');
return content;
}
// 使用串流模式傳送訊息
async function sendMessageStream(message) {
// 顯示使用者訊息
appendMessage(message, true);
// 顯示正在輸入指示器
typingIndicator.style.display = 'flex';
// 建立 AI 回應的容器
const responseContainer = appendMessage('', false, true);
try {
// 使用 fetch API 發送請求
const response = await fetch('{{ url_for("catalog.chat_stream") }}', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
message: message,
conversation_id: conversationId
})
});
if (!response.ok) {
throw new Error('API 請求失敗');
}
// 處理 SSE 串流
const reader = response.body.getReader();
const decoder = new TextDecoder();
let fullContent = '';
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value);
const lines = chunk.split('\n');
for (const line of lines) {
if (line.startsWith('data: ')) {
try {
const data = JSON.parse(line.slice(6));
if (data.error) {
responseContainer.innerHTML = formatMessage(
`錯誤: ${data.error}`
);
} else if (data.content) {
fullContent += data.content;
responseContainer.innerHTML = formatMessage(fullContent);
scrollToBottom();
} else if (data.done) {
// 串流完成
responseContainer.removeAttribute('id');
}
} catch (e) {
// 忽略解析錯誤
}
}
}
}
} catch (error) {
console.error('傳送訊息錯誤:', error);
responseContainer.innerHTML = formatMessage(
'抱歉，傳送訊息時發生錯誤，請稍後再試。'
);
} finally {
typingIndicator.style.display = 'none';
}
}
// 表單提交處理
chatForm.addEventListener('submit', async function(e) {
e.preventDefault();
const message = messageInput.value.trim();
if (!message) return;
// 清空輸入框
messageInput.value = '';
messageInput.style.height = 'auto';
// 禁用傳送按鈕
sendButton.disabled = true;
try {
await sendMessageStream(message);
} finally {
sendButton.disabled = false;
messageInput.focus();
}
});
// 清除對話歷史
clearChatButton.addEventListener('click', async function() {
if (!confirm('確定要清除所有對話記錄嗎？')) return;
try {
const response = await fetch('{{ url_for("catalog.clear_chat_history") }}', {
method: 'POST'
});
if (response.ok) {
// 清空聊天視窗
chatMessages.innerHTML = '';
conversationId = null;
} else {
alert('清除對話記錄失敗，請稍後再試。');
}
} catch (error) {
console.error('清除對話錯誤:', error);
alert('清除對話記錄時發生錯誤。');
}
});
// 自動調整輸入框高度
messageInput.addEventListener('input', function() {
this.style.height = 'auto';
this.style.height = Math.min(this.scrollHeight, 150) + 'px';
});
// 支援 Enter 鍵傳送（Shift+Enter 換行）
messageInput.addEventListener('keydown', function(e) {
if (e.key === 'Enter' && !e.shiftKey) {
e.preventDefault();
chatForm.dispatchEvent(new Event('submit'));
}
});
// 頁面載入完成後捲動到底部
scrollToBottom();
})();
</script>
{% endblock %}
前端實作包含了幾個重要的使用者體驗最佳化。串流回應處理讓使用者可以即時看到 AI 生成的內容，不需要等待整個回應完成。自動調整輸入框高度讓使用者可以方便地輸入多行訊息。鍵盤快捷鍵支援（Enter 傳送，Shift+Enter 換行）符合大多數聊天應用程式的習慣。
實作 DALL-E 圖片生成功能
DALL-E 是 OpenAI 推出的圖片生成模型，能夠根據文字描述生成高品質的圖片。在電子商務應用程式中，這個功能可以用於自動生成產品圖片，大大簡化產品上架流程。
# my_app/catalog/views.py（續）
# 圖片生成相關視圖
import os
import requests
import uuid
from werkzeug.utils import secure_filename
from my_app.catalog.models import Product, Category, ProductForm
@catalog.route('/product/create-with-ai', methods=['GET', 'POST'])
@login_required
def create_product_with_ai():
"""
使用 AI 生成圖片建立產品
這個功能讓使用者只需要輸入產品名稱和基本資訊，
系統會自動使用 DALL-E 生成產品圖片。
"""
form = ProductForm()
# 載入類別選項
form.category.choices = [
(c.id, c.name) for c in Category.query.order_by(Category.name).all()
]
if form.validate_on_submit():
try:
name = form.name.data
price = form.price.data
description = form.description.data
category_id = form.category.data
# 取得類別資訊用於生成更好的提示
category = Category.query.get_or_404(category_id)
# 建立圖片生成提示
# 提示的品質直接影響生成圖片的品質
prompt = f"""
為電子商務網站生成一張專業的產品圖片：
產品名稱：{name}
產品類別：{category.name}
要求：
- 白色或淺色背景
- 專業的產品攝影風格
- 高品質、清晰的細節
- 適合用於線上商店展示
- 正面視角，良好的光線
"""
# 呼叫 DALL-E API 生成圖片
current_app.logger.info(f'開始為產品 "{name}" 生成圖片')
image_urls = openai_service.generate_image(
prompt=prompt,
size='1024x1024',
quality='standard',
n=1
)
if not image_urls:
raise OpenAIServiceError('圖片生成失敗，未回傳任何圖片')
# 下載並儲存圖片
image_url = image_urls[0]
filename = save_generated_image(image_url, name)
# 建立產品記錄
product = Product(
name=name,
price=price,
description=description,
category_id=category_id,
image_filename=filename,
created_by=current_user.id
)
db.session.add(product)
db.session.commit()
# 將產品索引到 Elasticsearch
product.add_to_search_index()
current_app.logger.info(f'產品 "{name}" 建立成功，ID: {product.id}')
flash(f'產品 "{name}" 已成功建立', 'success')
return redirect(url_for('catalog.product_detail', id=product.id))
except OpenAIServiceError as e:
current_app.logger.error(f'圖片生成錯誤: {e}')
flash('圖片生成失敗，請稍後再試', 'error')
except Exception as e:
current_app.logger.error(f'產品建立錯誤: {e}')
db.session.rollback()
flash('建立產品時發生錯誤', 'error')
return render_template('product_create_ai.html', form=form)
def save_generated_image(image_url: str, product_name: str) -> str:
"""
下載並儲存 AI 生成的圖片
從 DALL-E 回傳的 URL 下載圖片，
並儲存到應用程式的上傳目錄中。
參數:
image_url: 圖片的 URL
product_name: 產品名稱，用於生成檔案名稱
回傳:
儲存的檔案名稱
"""
# 生成唯一的檔案名稱
# 使用 UUID 確保檔案名稱不會重複
safe_name = secure_filename(product_name)
unique_id = str(uuid.uuid4())[:8]
filename = f"{safe_name}_{unique_id}.png"
# 確保上傳目錄存在
upload_folder = current_app.config['UPLOAD_FOLDER']
if not os.path.exists(upload_folder):
os.makedirs(upload_folder)
# 下載圖片
response = requests.get(image_url, timeout=30)
response.raise_for_status()
# 儲存圖片
filepath = os.path.join(upload_folder, filename)
with open(filepath, 'wb') as f:
f.write(response.content)
current_app.logger.info(f'圖片已儲存: {filename}')
return filename
@catalog.route('/product/<int:id>/regenerate-image', methods=['POST'])
@login_required
def regenerate_product_image(id):
"""
重新生成產品圖片
如果使用者對自動生成的圖片不滿意，
可以重新生成新的圖片。
"""
product = Product.query.get_or_404(id)
# 檢查權限
if product.created_by != current_user.id and not current_user.is_admin:
return jsonify({'error': '您沒有權限執行此操作'}), 403
try:
# 取得自訂提示（如果有的話）
custom_prompt = request.json.get('prompt') if request.is_json else None
# 建立提示
if custom_prompt:
prompt = custom_prompt
else:
prompt = f"""
為電子商務網站生成一張專業的產品圖片：
產品名稱：{product.name}
產品類別：{product.category.name}
要求：白色背景、專業攝影風格、高品質
"""
# 生成新圖片
image_urls = openai_service.generate_image(
prompt=prompt,
size='1024x1024',
quality='standard',
n=1
)
if not image_urls:
return jsonify({'error': '圖片生成失敗'}), 500
# 刪除舊圖片
if product.image_filename:
old_filepath = os.path.join(
current_app.config['UPLOAD_FOLDER'],
product.image_filename
)
if os.path.exists(old_filepath):
os.remove(old_filepath)
# 儲存新圖片
filename = save_generated_image(image_urls[0], product.name)
product.image_filename = filename
db.session.commit()
return jsonify({
'message': '圖片已重新生成',
'image_url': url_for('static', filename=f'uploads/{filename}')
})
except OpenAIServiceError as e:
current_app.logger.error(f'重新生成圖片錯誤: {e}')
return jsonify({'error': '圖片生成服務暫時無法使用'}), 503
except Exception as e:
current_app.logger.error(f'重新生成圖片錯誤: {e}')
return jsonify({'error': '重新生成圖片時發生錯誤'}), 500
圖片生成功能的實作重點在於建立高品質的提示（Prompt）。提示的品質直接影響生成圖片的品質，因此需要仔細設計提示的內容，包括產品名稱、類別、風格要求等。此外，還需要處理圖片下載和儲存，確保檔案名稱的唯一性，以及在需要時能夠重新生成圖片。
整合 Elasticsearch 全文檢索
Elasticsearch 是一個強大的搜尋引擎，能夠提供快速、準確的全文檢索功能。對於電子商務應用程式來說，良好的搜尋功能是提升使用者體驗的關鍵。
# my_app/services/elasticsearch_service.py
# Elasticsearch 服務層
from elasticsearch import Elasticsearch, exceptions as es_exceptions
from flask import current_app
import logging
logger = logging.getLogger(__name__)
class ElasticsearchService:
"""
Elasticsearch 服務類別
封裝所有與 Elasticsearch 的互動，
包括索引管理、文件操作和搜尋功能。
"""
def __init__(self, app=None):
self.client = None
self.index_name = None
if app is not None:
self.init_app(app)
def init_app(self, app):
"""
使用 Flask 應用程式配置初始化服務
"""
es_url = app.config.get('ELASTICSEARCH_URL')
if not es_url:
logger.warning('ELASTICSEARCH_URL 未設定，搜尋功能將無法使用')
return
try:
# 建立 Elasticsearch 客戶端
self.client = Elasticsearch(
es_url,
ca_certs=app.config.get('ELASTICSEARCH_CA_CERTS'),
verify_certs=app.config.get('ELASTICSEARCH_VERIFY_CERTS', False),
basic_auth=(
app.config.get('ELASTICSEARCH_USER', 'elastic'),
app.config.get('ELASTICSEARCH_PASSWORD', '')
),
request_timeout=30
)
self.index_name = app.config.get('ELASTICSEARCH_INDEX', 'catalog')
# 確保索引存在
self._ensure_index_exists()
logger.info(f'Elasticsearch 服務已初始化，索引: {self.index_name}')
except Exception as e:
logger.error(f'Elasticsearch 初始化失敗: {e}')
self.client = None
def _ensure_index_exists(self):
"""
確保索引存在，如果不存在則建立
同時設定索引的映射（Mapping），
定義欄位類型和分析器。
"""
if not self.client:
return
try:
if not self.client.indices.exists(index=self.index_name):
# 定義索引設定和映射
settings = {
"settings": {
"number_of_shards": 1,
"number_of_replicas": 0,
"analysis": {
"analyzer": {
"chinese_analyzer": {
"type": "custom",
"tokenizer": "standard",
"filter": ["lowercase", "asciifolding"]
}
}
}
},
"mappings": {
"properties": {
"name": {
"type": "text",
"analyzer": "chinese_analyzer",
"fields": {
"keyword": {"type": "keyword"}
}
},
"description": {
"type": "text",
"analyzer": "chinese_analyzer"
},
"category": {
"type": "keyword"
},
"price": {
"type": "float"
},
"created_at": {
"type": "date"
}
}
}
}
self.client.indices.create(index=self.index_name, body=settings)
logger.info(f'索引 {self.index_name} 已建立')
except es_exceptions.RequestError as e:
if 'resource_already_exists_exception' not in str(e):
raise
def index_document(self, doc_id: str, document: dict):
"""
索引文件
將文件添加到 Elasticsearch 索引中。
如果文件已存在，則更新。
參數:
doc_id: 文件的唯一識別碼
document: 要索引的文件內容
"""
if not self.client:
logger.warning('Elasticsearch 未初始化，跳過索引操作')
return
try:
self.client.index(
index=self.index_name,
id=doc_id,
document=document
)
# 重新整理索引以便立即可搜尋
self.client.indices.refresh(index=self.index_name)
logger.debug(f'文件 {doc_id} 已索引')
except Exception as e:
logger.error(f'索引文件失敗: {e}')
def delete_document(self, doc_id: str):
"""
刪除文件
從索引中移除指定的文件。
參數:
doc_id: 要刪除的文件 ID
"""
if not self.client:
return
try:
self.client.delete(index=self.index_name, id=doc_id)
logger.debug(f'文件 {doc_id} 已刪除')
except es_exceptions.NotFoundError:
logger.warning(f'文件 {doc_id} 不存在')
except Exception as e:
logger.error(f'刪除文件失敗: {e}')
def search(
self,
query: str,
page: int = 1,
per_page: int = 20,
category: str = None,
min_price: float = None,
max_price: float = None
) -> dict:
"""
搜尋產品
執行全文搜尋，支援分頁和過濾。
參數:
query: 搜尋關鍵字
page: 頁碼
per_page: 每頁數量
category: 類別過濾
min_price: 最低價格
max_price: 最高價格
回傳:
搜尋結果，包含命中的文件和總數
"""
if not self.client:
return {'hits': [], 'total': 0}
# 計算分頁偏移量
from_offset = (page - 1) * per_page
# 建立搜尋查詢
must_clauses = []
filter_clauses = []
# 全文搜尋
if query:
must_clauses.append({
"multi_match": {
"query": query,
"fields": ["name^3", "description", "category"],
"type": "best_fields",
"fuzziness": "AUTO"
}
})
else:
must_clauses.append({"match_all": {}})
# 類別過濾
if category:
filter_clauses.append({
"term": {"category": category}
})
# 價格範圍過濾
if min_price is not None or max_price is not None:
price_range = {}
if min_price is not None:
price_range["gte"] = min_price
if max_price is not None:
price_range["lte"] = max_price
filter_clauses.append({
"range": {"price": price_range}
})
# 組合查詢
search_body = {
"query": {
"bool": {
"must": must_clauses,
"filter": filter_clauses
}
},
"from": from_offset,
"size": per_page,
"sort": [
{"_score": "desc"},
{"created_at": "desc"}
],
"highlight": {
"fields": {
"name": {},
"description": {"fragment_size": 150}
}
}
}
try:
response = self.client.search(
index=self.index_name,
body=search_body
)
# 處理搜尋結果
hits = []
for hit in response['hits']['hits']:
item = {
'id': hit['_id'],
'score': hit['_score'],
**hit['_source']
}
# 添加高亮結果
if 'highlight' in hit:
item['highlight'] = hit['highlight']
hits.append(item)
return {
'hits': hits,
'total': response['hits']['total']['value']
}
except Exception as e:
logger.error(f'搜尋失敗: {e}')
return {'hits': [], 'total': 0}
def get_suggestions(self, query: str, size: int = 5) -> list:
"""
取得搜尋建議
根據部分輸入提供自動完成建議。
參數:
query: 部分搜尋詞
size: 建議數量
回傳:
建議列表
"""
if not self.client or not query:
return []
try:
response = self.client.search(
index=self.index_name,
body={
"query": {
"match_phrase_prefix": {
"name": {
"query": query,
"max_expansions": 10
}
}
},
"size": size,
"_source": ["name"]
}
)
return [hit['_source']['name'] for hit in response['hits']['hits']]
except Exception as e:
logger.error(f'取得建議失敗: {e}')
return []
# 建立全域服務實例
es_service = ElasticsearchService()
Elasticsearch 服務層的設計考慮了多種搜尋場景。全文搜尋使用 multi_match 查詢，可以同時搜尋多個欄位，並使用權重來提高名稱欄位的重要性。模糊搜尋（Fuzziness）允許搜尋時有少量拼寫錯誤。高亮功能可以標示搜尋結果中的匹配部分，提升使用者體驗。
# my_app/catalog/views.py（續）
# 搜尋相關視圖
from my_app.services.elasticsearch_service import es_service
@catalog.route('/search')
def search():
"""
產品搜尋頁面
支援全文搜尋、類別過濾和價格範圍過濾。
"""
# 取得搜尋參數
query = request.args.get('q', '')
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 20, type=int)
category = request.args.get('category', '')
min_price = request.args.get('min_price', type=float)
max_price = request.args.get('max_price', type=float)
# 執行搜尋
results = es_service.search(
query=query,
page=page,
per_page=per_page,
category=category if category else None,
min_price=min_price,
max_price=max_price
)
# 取得所有類別供過濾使用
categories = Category.query.order_by(Category.name).all()
return render_template(
'search.html',
query=query,
results=results['hits'],
total=results['total'],
page=page,
per_page=per_page,
categories=categories,
selected_category=category,
min_price=min_price,
max_price=max_price
)
@catalog.route('/search/suggestions')
def search_suggestions():
"""
搜尋自動完成建議
根據使用者輸入提供即時的搜尋建議。
"""
query = request.args.get('q', '')
suggestions = es_service.get_suggestions(query, size=5)
return jsonify({'suggestions': suggestions})
搜尋功能的實作包含了完整的過濾支援，讓使用者可以按類別和價格範圍縮小搜尋結果。自動完成建議功能提供即時的搜尋輔助，幫助使用者更快找到想要的產品。
資料模型設計
良好的資料模型設計是應用程式的基礎。資料模型不僅要正確地表示業務實體，還要與外部服務（如 Elasticsearch）緊密整合。
# my_app/catalog/models.py
# 資料模型定義
from datetime import datetime
from my_app import db
from my_app.services.elasticsearch_service import es_service
class Category(db.Model):
"""
產品類別模型
"""
__tablename__ = 'categories'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100), nullable=False, unique=True)
description = db.Column(db.Text)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
# 關聯
products = db.relationship('Product', backref='category', lazy='dynamic')
def __repr__(self):
return f'<Category {self.name}>'
class Product(db.Model):
"""
產品模型
包含與 Elasticsearch 整合的方法。
"""
__tablename__ = 'products'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(200), nullable=False, index=True)
description = db.Column(db.Text)
price = db.Column(db.Numeric(10, 2), nullable=False)
image_filename = db.Column(db.String(255))
category_id = db.Column(db.Integer, db.ForeignKey('categories.id'), nullable=False)
created_by = db.Column(db.Integer, db.ForeignKey('users.id'))
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
def __repr__(self):
return f'<Product {self.name}>'
def add_to_search_index(self):
"""
將產品添加到 Elasticsearch 索引
在建立或更新產品後呼叫此方法，
確保搜尋索引與資料庫同步。
"""
document = {
'name': self.name,
'description': self.description or '',
'price': float(self.price),
'category': self.category.name,
'created_at': self.created_at.isoformat()
}
es_service.index_document(str(self.id), document)
def remove_from_search_index(self):
"""
從 Elasticsearch 索引中移除產品
在刪除產品時呼叫此方法。
"""
es_service.delete_document(str(self.id))
@property
def image_url(self):
"""
取得產品圖片的 URL
"""
if self.image_filename:
from flask import url_for
return url_for('static', filename=f'uploads/{self.image_filename}')
return None
class ChatHistory(db.Model):
"""
聊天歷史記錄模型
儲存使用者與 AI 助手的對話記錄。
"""
__tablename__ = 'chat_history'
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
conversation_id = db.Column(db.String(100), index=True)
user_message = db.Column(db.Text, nullable=False)
assistant_message = db.Column(db.Text, nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True)
# 關聯
user = db.relationship('User', backref='chat_history')
def __repr__(self):
return f'<ChatHistory {self.id}>'
# 事件監聽器
@db.event.listens_for(Product, 'after_insert')
def product_after_insert(mapper, connection, target):
"""
產品插入後自動索引到 Elasticsearch
"""
# 注意：這裡不能直接呼叫 add_to_search_index
# 因為事務還沒提交，需要使用 after_commit 事件
pass
@db.event.listens_for(Product, 'after_delete')
def product_after_delete(mapper, connection, target):
"""
產品刪除後從 Elasticsearch 移除
"""
es_service.delete_document(str(target.id))
資料模型的設計將 Elasticsearch 整合封裝在模型方法中，讓其他部分的程式碼可以方便地維護搜尋索引的同步。這種設計遵循了關注點分離的原則，讓資料存取邏輯集中在模型層。
安全性考量與最佳實踐
在開發 AI 驅動的應用程式時，安全性是一個特別重要的考量。API 金鑰管理、輸入驗證、速率限制等都需要妥善處理。
# my_app/__init__.py
# 應用程式工廠和安全設定
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from flask_wtf.csrf import CSRFProtect
db = SQLAlchemy()
login_manager = LoginManager()
csrf = CSRFProtect()
limiter = Limiter(key_func=get_remote_address)
def create_app(config_name='default'):
"""
應用程式工廠函式
"""
app = Flask(__name__)
# 載入配置
from config import config
app.config.from_object(config[config_name])
# 初始化擴充套件
db.init_app(app)
login_manager.init_app(app)
csrf.init_app(app)
limiter.init_app(app)
# 初始化服務
from my_app.services.openai_service import openai_service
from my_app.services.elasticsearch_service import es_service
openai_service.init_app(app)
es_service.init_app(app)
# 註冊藍圖
from my_app.catalog.views import catalog
app.register_blueprint(catalog)
# 設定全域速率限制
# 這對於防止 API 濫用特別重要
@app.errorhandler(429)
def ratelimit_handler(e):
return jsonify({'error': '請求過於頻繁，請稍後再試'}), 429
return app
# 在視圖中應用速率限制
@catalog.route('/chat/message', methods=['POST'])
@login_required
@limiter.limit("30 per minute") # 每分鐘最多 30 次聊天請求
def chat_message():
# ... 聊天邏輯 ...
pass
@catalog.route('/product/create-with-ai', methods=['GET', 'POST'])
@login_required
@limiter.limit("10 per hour") # 每小時最多生成 10 張圖片
def create_product_with_ai():
# ... 產品建立邏輯 ...
pass
安全性措施包括多個層面。API 金鑰透過環境變數管理，避免在程式碼中硬編碼。速率限制防止 API 被濫用，這對於控制成本和防止攻擊都很重要。CSRF 保護防止跨站請求偽造攻擊。輸入驗證確保所有使用者輸入都是安全的。
總結
本文深入探討了如何使用 Flask 框架整合 OpenAI 的各項服務，建立功能完善的互動式網頁應用程式。我們從專案架構設計開始，逐步實作了聊天機器人、圖片生成和全文檢索等核心功能。
在聊天機器人的實作中，我們不僅處理了基本的對話功能，還加入了對話上下文維護和串流回應支援，提供了流暢的使用者體驗。服務層的設計封裝了所有與 OpenAI API 的互動，包括錯誤處理和重試邏輯，確保應用程式的穩定性。
DALL-E 圖片生成功能展示了如何根據產品資訊自動生成高品質的產品圖片，大幅簡化了電子商務網站的產品上架流程。透過精心設計的提示，我們可以生成符合專業標準的產品圖片。
Elasticsearch 整合提供了強大的全文檢索能力，包括模糊搜尋、過濾和自動完成建議等功能。搜尋服務層的設計讓搜尋邏輯可以方便地複用和測試。
在整個開發過程中，我們特別關注了安全性、效能和可維護性。API 金鑰管理、速率限制、輸入驗證等措施確保了應用程式的安全。服務層設計和完善的錯誤處理確保了程式碼的可維護性。透過這些最佳實踐，開發者可以建立出企業級的 AI 驅動網頁應用程式。
