🚀 打造高效能智慧推薦系統：Laravel x FastAPI 雙語微服務整合實戰
摘要
想為你的電商網站添加智慧推薦功能，卻苦惱於傳統架構的效能瓶頸和單一技術棧的限制嗎？本文將帶你手把手搭建一個基於 Laravel (PHP) 與 FastAPI (Python) 的跨語言微服務推薦系統。我們將深入探討如何利用 Redis 作為高速快取與服務間的協調中心，並實現 FastAPI 直接以只讀權限連接 Laravel 主資料庫，確保數據一致性。透過 Docker 容器化部署與 GitHub Actions 自動化 CI/CD，讓你掌握構建可擴充、可維護、可容器化的現代化應用程式的關鍵技能！
本專案的所有程式碼已開源並託管於 GitHub：
💡 為什麼選擇 Laravel x FastAPI 雙語微服務？
在現代應用程式開發中，單一技術棧有時難以滿足所有需求。PHP (Laravel) 在 Web 開發領域擁有豐富的生態和快速開發的優勢，而 Python (FastAPI) 則以其高效能、簡潔語法和強大的數據科學庫在 AI/ML 領域獨領風騷。將兩者結合，能達到 1+1 > 2 的效果：
- 專注職責，各司其職：Laravel 負責用戶管理、訂單處理等核心業務邏輯；FastAPI 則專注於執行複雜的推薦演算法，提供高效的預測服務。
- 效能優化：FastAPI 基於 ASGI，天生支援非同步，處理大量推薦請求時效能表現優異。
- 技術棧多元化：利用兩種語言的優勢，為未來的功能擴展和團隊技能組合提供更大的彈性。
- 可擴展性與彈性：微服務架構讓各服務獨立部署、擴縮容，提高系統的整體彈性與可用性。
🏛️ 核心架構概覽
我們的推薦系統將由以下幾個關鍵組件構成：
- Laravel App (PHP)：
- 負責前端展示、用戶介面。
- 處理用戶的產品瀏覽行為（記錄互動數據）。
- 向推薦服務發送請求，獲取推薦結果。
- 將推薦結果渲染到頁面。
- FastAPI Recommender Service (Python)：
- 作為推薦邏輯的核心，執行協同過濾 (Collaborative Filtering) 演算法。
- 直接以只讀權限連接 Laravel 的 MySQL 主資料庫，獲取用戶、產品、訂單、互動等原始數據。
- 將計算出的推薦結果寫入 Redis 快取。
- 提供 API 接口供 Laravel 調用觸發推薦計算或獲取結果。
- MySQL Database：
- Laravel 的主資料庫，儲存所有業務數據（用戶、產品、訂單、互動）。
- FastAPI 以只讀模式訪問這些數據，用於推薦演算法的數據源。
- Redis Cache：
- 高速快取層：儲存 FastAPI 計算出的推薦結果，供 Laravel 快速讀取，降低資料庫負載。
- 跨服務溝通橋樑：Laravel 可以觸發 FastAPI 進行異步推薦計算。
- Docker & Docker Compose：
- 實現各服務的容器化，提供一致的開發和部署環境。
- 一鍵啟動所有服務，簡化環境配置。
- GitHub Actions：
- 實現自動化 CI/CD，確保程式碼品質並自動化測試與部署流程。
以下是我們的架構圖：
程式碼片段
graph TD
A[用戶瀏覽器] -->|HTTP/HTTPS| B(Laravel App);
B --請求推薦<br/>(user_id)--> C(Redis Cache);
C --快取命中?--> B;
C --快取未命中--> D(FastAPI Recommender Service);
D --只讀查詢--> E(MySQL Database<br/>(Laravel 主庫));
E --獲取用戶/商品/互動數據--> D;
D --計算推薦結果--> F(Redis Cache);
F --回寫推薦結果<br/>(TTL)--> C;
C --返回推薦商品ID--> B;
B --從 MySQL 查詢商品詳情<br/>顯示推薦商品--> A;
- 流程說明：
- 用戶瀏覽 Laravel 網站，點擊某商品進入詳情頁。
- Laravel 後端嘗試從 Redis 快取中讀取該用戶的推薦列表。
- 如果 Redis 命中，直接返回推薦商品 ID。
- 如果 Redis 未命中或推薦數據過期：
- Laravel 會向 FastAPI 服務發送請求，觸發推薦邏輯的重新計算（這通常是一個非阻塞的異步觸發）。
- FastAPI 收到請求後，從 MySQL 資料庫（Laravel 的主庫）讀取最新的用戶互動、商品、訂單數據。
- FastAPI 執行協同過濾推薦演算法，計算出推薦商品 ID 列表。
- FastAPI 將計算結果存入 Redis，並設定 TTL (Time To Live)。
- 為了提供即時響應，FastAPI 可能也會立即返回新計算的推薦結果給 Laravel。
- Laravel 收到推薦商品 ID 後，再次從 MySQL 資料庫查詢這些商品的詳細資訊。
- 將商品詳情和推薦列表一起渲染到商品詳情頁面，呈現給用戶。
🛠️ 環境搭建：Docker Compose 一鍵啟動
本專案利用 Docker Compose 管理所有服務，極大簡化了環境搭建。
前置條件
Docker Desktop
1. 複製專案
首先，從 GitHub 克隆本專案到你的本地：
Bash
git clone https://github.com/BpsEason/laravel-fastapi-recommender.git
cd laravel-fastapi-recommender
2. 核心 Docker Compose 配置 docker-compose.yml
這是我們定義所有服務的 Docker Compose 文件，它定義了 laravel-app、recommender-service、mysql 和 redis 服務及其相互依賴關係：
YAML
# docker-compose.yml
version: '3.8'
services:
laravel-app:
build:
context: ./laravel-app
dockerfile: Dockerfile
image: laravel-app:latest
ports:
- "8000:80"
volumes:
- ./laravel-app:/var/www/html
depends_on:
- mysql
- redis
env_file:
- .env.docker
networks:
- app-network
recommender-service:
build:
context: ./recommender-service
dockerfile: Dockerfile
image: recommender-service:latest
ports:
- "8001:8000"
volumes:
- ./recommender-service:/app
depends_on:
- mysql
- redis
env_file:
- .env.docker
command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload # 開發模式下方便熱重載
networks:
- app-network
mysql:
image: mysql:8.0
ports:
- "3306:3306"
environment:
MYSQL_ROOT_PASSWORD: root_password
MYSQL_DATABASE: laravel_db
volumes:
- db_data:/var/lib/mysql # 持久化數據
networks:
- app-network
redis:
image: redis:latest
ports:
- "6379:6379"
volumes:
- ./redis/redis.conf:/usr/local/etc/redis/redis.conf # 掛載 Redis 配置
command: redis-server /usr/local/etc/redis/redis.conf
networks:
- app-network
networks:
app-network:
driver: bridge
volumes:
db_data: # 定義數據卷
3. 環境變數配置 .env.docker
這個文件會被 Docker Compose 自動載入到各個服務中，包含了資料庫和 Redis 的連接資訊，以及服務間通信的 URL：
Ini, TOML
# .env.docker (位於專案根目錄)
# Shared .env for Docker Compose services
# MySQL
MYSQL_ROOT_PASSWORD=root_password
MYSQL_DATABASE=laravel_db
DB_HOST=mysql # 注意這裡是服務名稱，而非 localhost
DB_PORT=3306
DB_USERNAME=root
DB_PASSWORD=root_password
DATABASE_URL=mysql+pymysql://root:root_password@mysql:3306/laravel_db # FastAPI 使用
# Redis
REDIS_HOST=redis # 注意這裡是服務名稱
REDIS_PORT=6379
REDIS_DB=0
REDIS_CLIENT=predis # Laravel 使用 Predis 庫
# Laravel Specific
APP_NAME="Laravel FastAPI Recommender"
APP_ENV=local
APP_KEY= # 稍後會生成
APP_DEBUG=true
APP_URL=http://localhost:8000
LOG_CHANNEL=stack
# FastAPI Specific
FASTAPI_RECOMMENDER_URL=http://recommender-service:8000/api/v1 # Laravel 調用 FastAPI 的內部地址
4. 啟動服務
在專案根目錄執行以下命令，Docker Compose 將會自動構建映像並啟動所有服務：
Bash
docker-compose up --build -d
--build：首次運行會構建 Docker 映像。-d：讓服務在後台運行。
5. Laravel 應用程式設定
在服務啟動後，你需要進入 Laravel 容器內部執行必要的命令，以安裝 Composer 依賴、執行資料庫遷移和填充測試數據，並生成應用程式密鑰：
Bash
docker-compose exec laravel-app composer install
docker-compose exec laravel-app php artisan migrate --force # 執行資料庫遷移
docker-compose exec laravel-app php artisan db:seed # 填充測試數據
docker-compose exec laravel-app php artisan key:generate
6. 存取應用
- Laravel 前台：
http://localhost:8000 - FastAPI 推薦服務文件 (Swagger UI)：
http://localhost:8001/docs
深入程式碼：關鍵文件解析
A. Laravel 端 (laravel-app/)
1. Dockerfile
Laravel 的 Dockerfile 會安裝 PHP 8.2 FPM、Nginx、MySQL 客戶端、Redis 擴展等必要依賴，並配置 Nginx 以服務 Laravel 應用。
2. app/Http/Controllers/ProductController.php
這是處理產品詳情頁的核心控制器。它負責：
- 載入產品資訊。
- 記錄用戶瀏覽互動：這一步非常重要，為推薦系統提供基礎數據。
- 調用
RecommendationService獲取推薦商品。 - 將產品和推薦商品傳遞給視圖渲染。
<!-- end list -->
PHP
<?php
// laravel-app/app/Http/Controllers/ProductController.php
namespace App\Http\Controllers;
use App\Models\Eloquent\Product;
use Illuminate\Http\Request;
use Illuminate\Support\Facades\Auth;
use App\Services\RecommendationService;
use App\Models\Eloquent\UserInteraction;
use Illuminate\Support\Facades\Log;
class ProductController extends Controller
{
protected $recommendationService;
public function __construct(RecommendationService $recommendationService)
{
$this->recommendationService = $recommendationService;
}
public function show(Request $request, $id)
{
$product = Product::with('category')->findOrFail($id);
$recommendedProducts = collect();
if (Auth::check()) { // 檢查用戶是否登入
$user = Auth::user();
try {
// 記錄用戶互動，這裡會自動使用 Laravel 的 created_at/updated_at
UserInteraction::create([
'user_id' => $user->id,
'product_id' => $product->id,
'interaction_type' => 'view', // 假設瀏覽為一種互動類型
]);
Log::info("User interaction logged: user {$user->id} viewed product {$product->id}.");
} catch (\Exception $e) {
Log::error("Failed to log user interaction: {$e->getMessage()}");
}
// 獲取個性化推薦
$recommendedProducts = $this->recommendationService->getRecommendations($user->id);
// 過濾掉當前正在查看的商品，避免重複推薦
$recommendedProducts = $recommendedProducts->filter(function($item) use ($product) {
return $item->id !== $product->id;
});
} else {
Log::info("Guest user viewing product {$product->id}. No personalized recommendations.");
// 訪客或未登入用戶提供熱門商品作為冷啟動/通用推薦
$recommendedProducts = $this->recommendationService->getPopularProducts(5);
$recommendedProducts = $recommendedProducts->filter(function($item) use ($product) {
return $item->id !== $product->id;
});
}
return view('product_detail', compact('product', 'recommendedProducts'));
}
}
3. app/Services/RecommendationService.php
這是 Laravel 端與推薦邏輯互動的服務層，實現了快取優先、異步觸發和容錯機制：
PHP
<?php
// laravel-app/app/Services/RecommendationService.php
namespace App\Services;
use Illuminate\Support\Facades\Redis;
use Illuminate\Support\Facades\Log;
use App\Models\Eloquent\Product;
use App\Services\API\RecommenderClient;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\Cache; // 用於檔案快取回退
class RecommendationService
{
protected $recommenderClient;
protected $redis;
public function __construct(RecommenderClient $recommenderClient)
{
$this->recommenderClient = $recommenderClient;
$this->redis = Redis::connection();
}
public function getRecommendations(int $userId, int $numRecommendations = 5): Collection
{
$cacheKey = "user:{$userId}:recommendations";
$fallbackCacheKey = "user:{$userId}:recommendations_fallback"; // 檔案快取鍵
// 1. 嘗試從 Redis 快取中獲取
try {
$cachedProductIdsJson = $this->redis->get($cacheKey);
if ($cachedProductIdsJson) {
Log::info("RecommendationService: Fetched recommendations for user {$userId} from Redis cache.");
$recommendedProductIds = json_decode($cachedProductIdsJson, true);
return Product::whereIn('id', $recommendedProductIds)->get();
}
} catch (\Exception $e) {
Log::error("RecommendationService: Redis connection error or issue: {$e->getMessage()}. Falling back to file cache/direct API call.");
// Redis 故障時，嘗試從檔案快取讀取
$fallbackProductIds = Cache::get($fallbackCacheKey);
if ($fallbackProductIds) {
Log::info("RecommendationService: Fetched recommendations for user {$userId} from file fallback cache.");
return Product::whereIn('id', $fallbackProductIds)->get();
}
}
// 2. 如果 Redis 沒有快取 (或 Redis 故障且無 fallback 檔案快取)，觸發 FastAPI 重新計算
Log::info("RecommendationService: Cache miss for user {$userId} or Redis issue. Triggering FastAPI recalculation.");
// 觸發 FastAPI 異步重新計算 (recalculateRecommendations 內部會盡量非阻塞發送)
$fastApiCallSuccess = $this->recommenderClient->recalculateRecommendations($userId);
if (!$fastApiCallSuccess) {
Log::warning("RecommendationService: Failed to trigger FastAPI recalculation for user {$userId}.");
// FastAPI 觸發失敗，嘗試返回熱門商品
return $this->getPopularProducts($numRecommendations);
}
// 3. 獲取直接推薦：即使觸發了異步計算，也立即嘗試獲取推薦結果作為快速響應。
// FastAPI 在計算完畢後會將結果回寫到 Redis。這裡的直接獲取可以理解為
// 1) FastAPI 收到請求後立即返回一部分熱門推薦，然後在後台執行複雜計算並更新 Redis。
// 2) 如果是同步調用，則直接等待結果。
// 在這個範例中，`getDirectRecommendations` 其實也是同步等待 FastAPI 的響應。
// 對於真正的異步，這裡應該直接返回熱門商品，等待下次請求時 Redis 快取生效。
$directRecommendations = $this->recommenderClient->getDirectRecommendations($userId, $numRecommendations);
if ($directRecommendations) {
Log::info("RecommendationService: Got direct recommendations for user {$userId} from FastAPI.");
// 收到直接推薦後，也快取一份到檔案，以防 Redis 再次故障
Cache::put($fallbackCacheKey, $directRecommendations, now()->addMinutes(10)); // 檔案快取時間短一點
return Product::whereIn('id', $directRecommendations)->get();
}
Log::warning("RecommendationService: Could not get direct recommendations for user {$userId}. Falling back to popular products.");
return $this->getPopularProducts($numRecommendations);
}
public function getPopularProducts(int $numRecommendations): Collection
{
Log::info("RecommendationService: Falling back to popular products.");
// 從資料庫中獲取熱門商品（例如，根據銷量）
// 這裡僅作示例，實際應有更完善的熱門商品統計邏輯
return Product::orderBy('id', 'asc')->limit($numRecommendations)->get();
}
}
4. app/Services/API/RecommenderClient.php
封裝了 Laravel 調用 FastAPI 服務的 HTTP 請求邏輯。
PHP
<?php
// laravel-app/app/Services/API/RecommenderClient.php
namespace App\Services\API;
use Illuminate\Support\Facades\Http;
use Illuminate\Support\Facades\Log;
class RecommenderClient
{
protected $fastApiUrl;
public function __construct()
{
$this->fastApiUrl = env('FASTAPI_RECOMMENDER_URL', 'http://localhost:8001/api/v1');
}
/**
* 呼叫 FastAPI 服務，觸發推薦邏輯的重新計算。
* 設定較短的超時，使其行為更接近「異步觸發」而不阻塞 Laravel 主請求。
* 實際生產環境應使用 Laravel Queue Job 進行完全異步處理。
*
* @param int $userId
* @return bool
*/
public function recalculateRecommendations(int $userId): bool
{
$endpoint = "{$this->fastApiUrl}/recommendations/recalculate/{$userId}";
try {
// timeout(2) 讓 Guzzle 在 2 秒內沒收到響應就放棄，避免長時間阻塞
$response = Http::timeout(2)->post($endpoint);
if ($response->successful()) {
Log::info("RecommenderClient: Successfully triggered FastAPI recalculation for user {$userId}.");
return true;
} else {
Log::error("RecommenderClient: Failed to trigger FastAPI recalculation for user {$userId}. Status: {$response->status()} Body: {$response->body()}");
return false;
}
} catch (\Illuminate\Http\Client\ConnectionException $e) {
Log::error("RecommenderClient: Connection error to FastAPI for user {$userId} (recalculate): {$e->getMessage()}");
return false;
} catch (\Exception $e) {
Log::error("RecommenderClient: Unexpected error calling FastAPI recalculate API for user {$userId}: {$e->getMessage()}");
return false;
}
}
/**
* 從 FastAPI 服務直接獲取推薦結果。
*
* @param int $userId
* @param int $numRecommendations
* @return array|null
*/
public function getDirectRecommendations(int $userId, int $numRecommendations = 5): ?array
{
$endpoint = "{$this->fastApiUrl}/recommendations/{$userId}?num_recommendations={$numRecommendations}";
try {
$response = Http::timeout(5)->get($endpoint); // 這裡需要等待結果，超時可以稍長
if ($response->successful()) {
return $response->json();
} else {
Log::error("RecommenderClient: Failed to get direct recommendations for user {$userId}. Status: {$response->status()} Body: {$response->body()}");
return null;
}
} catch (\Illuminate\Http\Client\ConnectionException $e) {
Log::error("RecommenderClient: Connection error trying to get direct recommendations for user {$userId}: {$e->getMessage()}");
return null;
} catch (\Exception $e) {
Log::error("RecommenderClient: Unexpected error calling FastAPI direct recommendation API for user {$userId}: {$e->getMessage()}");
return null;
}
}
}
5. Laravel Models (app/Models/Eloquent/)
FastAPI 會讀取這些模型對應的資料表。請確保你的 Laravel Migration 創建了這些資料表，並且它們包含了 id, created_at, updated_at (除了 UserInteraction 中的 timestamp，如果它有單獨的意義)。
User.phpProduct.phpCategory.phpOrder.phpOrderItem.phpUserInteraction.php(用於記錄用戶點擊、收藏等行為，是推薦算法的關鍵輸入)
B. FastAPI 端 (recommender-service/)
1. Dockerfile
FastAPI 的 Dockerfile 將安裝 Python 環境、必要的數據科學庫 (Pandas, NumPy, scikit-learn) 以及 MySQL 和 Redis 客戶端庫。
2. app/main.py
FastAPI 應用的入口點，配置日誌、引入路由、健康檢查等。
Python
# recommender-service/app/main.py
from fastapi import FastAPI, HTTPException, status
import logging
from .api.v1.routes import router as v1_router
from .core.config import settings
from .dependencies import get_db, get_redis_client
# 配置日誌
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
app = FastAPI(
title=settings.PROJECT_NAME,
description=settings.PROJECT_DESCRIPTION,
version=settings.PROJECT_VERSION
)
app.include_router(v1_router, prefix=settings.API_V1_STR)
@app.get("/")
async def root():
logger.info("Root endpoint accessed.")
return {"message": "Welcome to FastAPI Recommender Service!"}
@app.get("/health")
async def health_check():
"""健康檢查端點，檢查資料庫和 Redis 連接狀態。"""
logger.info("Health check requested.")
try:
with get_db() as db:
db.execute("SELECT 1") # 簡單查詢，測試 DB 連接
redis_client_instance = get_redis_client()
if redis_client_instance:
redis_client_instance.ping() # 測試 Redis 連接
else:
raise Exception("Redis client not initialized or connection failed.")
logger.info("Health check successful: Database and Redis connected.")
return {"status": "ok", "database": "connected", "redis": "connected"}
except Exception as e:
logger.error(f"Health check failed: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Service unhealthy: {e}"
)
3. app/core/config.py
統一管理 FastAPI 的配置，如資料庫連接字串、Redis 資訊等。
4. app/models/ (SQLAlchemy ORM Models)
這些是 FastAPI 用於只讀連接 Laravel 主資料庫的 ORM 模型。它們鏡像了 Laravel 資料庫中的表格結構。
db.py：資料庫引擎和會話設置。user.py：對應users表。product.py：對應products和categories表。order.py：對應orders和order_items表。interaction.py：對應user_interactions表。
5. app/data/data_loader.py
負責從 MySQL 資料庫載入用戶互動數據，並轉換為 Pandas DataFrame，以便推薦算法使用。
Python
# recommender-service/app/data/data_loader.py
# ... (省略 import 部分)
class DataLoader:
def __init__(self, db: Session):
self.db = db
def load_interaction_data(self) -> Tuple[pd.DataFrame, dict, dict]:
"""
從資料庫載入用戶互動數據 (訂單和用戶互動表)，並轉換為 Pandas DataFrame。
為每種互動類型賦予分數，並聚合，確保每個用戶-產品對只有一個最高分數。
同時返回 ID 到索引的映射，便於矩陣操作。
"""
try:
# 從 OrderItem 獲取購買數據，賦予高分
order_items_data = self.db.query(OrderItem.user_id, OrderItem.product_id).all()
# 從 UserInteraction 獲取其他互動數據
user_interactions_data = self.db.query(UserInteraction.user_id, UserInteraction.product_id, UserInteraction.interaction_type).all()
except Exception as e:
logger.error(f"Error loading interaction data from DB: {e}")
return pd.DataFrame(columns=['user_id', 'product_id', 'value']), {}, {}
all_interactions = []
for user_id, product_id in order_items_data:
all_interactions.append({'user_id': user_id, 'product_id': product_id, 'value': 5}) # 購買賦予最高分
for user_id, product_id, interaction_type in user_interactions_data:
score = 1 # 預設分數
if interaction_type == 'favorite':
score = 4
elif interaction_type == 'add_to_cart':
score = 3
elif interaction_type == 'click':
score = 2
elif interaction_type == 'view': # 瀏覽行為
score = 1
all_interactions.append({'user_id': user_id, 'product_id': product_id, 'value': score})
if not all_interactions:
logger.info("No interaction data found in database.")
return pd.DataFrame(columns=['user_id', 'product_id', 'value']), {}, {}
df = pd.DataFrame(all_interactions)
# 對相同的 user_id, product_id，取最高的互動分數
df = df.groupby(['user_id', 'product_id'])['value'].max().reset_index()
unique_users = df['user_id'].unique()
unique_products = df['product_id'].unique()
user_to_idx = {user_id: idx for idx, user_id in enumerate(unique_users)}
product_to_idx = {product_id: idx for idx, product_id in enumerate(unique_products)}
logger.info(f"Loaded {len(df)} unique interactions for {len(unique_users)} users and {len(unique_products)} products.")
return df, user_to_idx, product_to_idx
6. app/services/recommender_logic.py
實現核心推薦演算法。這裡使用的是基於用戶的協同過濾。
Python
# recommender-service/app/services/recommender_logic.py
# ... (省略 import 部分)
class Recommender:
def __init__(self, db: Session):
self.db = db
self.data_loader = DataLoader(db)
def get_interaction_matrix_and_mappings(self) -> Tuple[np.ndarray, List[int], List[int]]:
# 載入數據並生成互動矩陣及映射
df, user_to_idx, product_to_idx = self.data_loader.load_interaction_data()
if df.empty:
logger.info("No interaction data loaded. Returning empty matrix and mappings.")
return np.array([]), [], []
num_users = len(user_to_idx)
num_products = len(product_to_idx)
interaction_matrix = np.zeros((num_users, num_products))
for _, row in df.iterrows():
user_idx = user_to_idx[row['user_id']]
product_idx = product_to_idx[row['product_id']]
interaction_matrix[user_idx, product_idx] = row['value'] # 填充互動分數
all_user_ids = [user_id for user_id, _ in sorted(user_to_idx.items(), key=lambda item: item[1])]
all_product_ids = [product_id for product_id, _ in sorted(product_to_idx.items(), key=lambda item: item[1])]
return interaction_matrix, all_user_ids, all_product_ids
def calculate_similarity(self, matrix: np.ndarray) -> np.ndarray:
# 計算用戶之間的餘弦相似度
if matrix.shape[0] < 2 or np.all(matrix == 0):
logger.info("Matrix too small or all zeros for similarity calculation. Returning empty array.")
return np.array([[]])
return cosine_similarity(matrix)
def recommend_for_user(self, target_user_id: int, num_recommendations: int = 5) -> List[int]:
interaction_matrix, all_user_ids, all_product_ids = self.get_interaction_matrix_and_mappings()
# 冷啟動/回退策略：如果用戶沒有互動數據，或數據不足，則推薦熱門商品
if not all_user_ids or not all_product_ids or target_user_id not in all_user_ids:
logger.info(f"User {target_user_id} not in interaction data or no data. Falling back to popular products.")
return self.get_popular_products(num_recommendations)
try:
target_user_idx = all_user_ids.index(target_user_id)
except ValueError:
logger.warning(f"Target user ID {target_user_id} not found in all_user_ids. This should not happen if previous check passed.")
return self.get_popular_products(num_recommendations)
user_similarity = self.calculate_similarity(interaction_matrix)
if user_similarity.size == 0 or target_user_idx >= user_similarity.shape[0]:
logger.info("Similarity matrix is empty or target user index out of bounds. Falling back to popular products.")
return self.get_popular_products(num_recommendations)
# 獲取與目標用戶相似的用戶索引（排除自己）
similar_users_indices = user_similarity[target_user_idx].argsort()[::-1][1:]
recommended_scores: Dict[int, float] = {}
# 獲取目標用戶已經互動過的商品，避免重複推薦
user_interacted_product_indices = np.where(interaction_matrix[target_user_idx] > 0)[0]
user_interacted_product_ids = {all_product_ids[i] for i in user_interacted_product_indices}
for sim_user_idx in similar_users_indices:
similarity_score = user_similarity[target_user_idx, sim_user_idx]
if similarity_score <= 0.0: # 只考慮正相關的相似用戶
continue
sim_user_product_indices = np.where(interaction_matrix[sim_user_idx] > 0)[0]
for product_idx in sim_user_product_indices:
product_id = all_product_ids[product_idx]
# 跳過已經互動過的商品
if product_id in user_interacted_product_ids:
continue
if product_id not in recommended_scores:
recommended_scores[product_id] = 0.0
recommended_scores[product_id] += interaction_matrix[sim_user_idx, product_idx] * similarity_score
sorted_recommendations = sorted(recommended_scores.items(), key=lambda item: item[1], reverse=True)
if not sorted_recommendations:
logger.info(f"No recommendations found via collaborative filtering for user {target_user_id}. Falling back to popular products.")
return self.get_popular_products(num_recommendations)
return [product_id for product_id, score in sorted_recommendations[:num_recommendations]]
def get_popular_products(self, num_recommendations: int = 5) -> List[int]:
# 從 OrderItem 中獲取銷量最高的商品作為熱門推薦
popular_products_by_purchase = self.db.query(
Product.id,
func.sum(OrderItem.quantity).label('total_quantity_sold')
) \
.join(OrderItem, Product.id == OrderItem.product_id) \
.group_by(Product.id) \
.order_by(func.sum(OrderItem.quantity).desc()) \
.limit(num_recommendations) \
.all()
result_ids = [p.id for p in popular_products_by_purchase]
# 如果購買數據不足以提供足夠的熱門商品，則從所有產品中補充
if len(result_ids) < num_recommendations:
remaining_needed = num_recommendations - len(result_ids)
all_product_ids_in_db = [p.id for p in self.db.query(Product.id).all()]
additional_products = []
if all_product_ids_in_db:
available_for_filler = [p_id for p_id in all_product_ids_in_db if p_id not in result_ids]
additional_products = available_for_filler[:remaining_needed] # 簡單取前幾個補充
result_ids.extend(additional_products)
return result_ids[:num_recommendations] # 確保最終返回的數量
7. app/api/v1/routes.py
定義 FastAPI 的 API 端點，包括獲取推薦和觸發重新計算。
Python
# recommender-service/app/api/v1/routes.py
# ... (省略 import 部分)
router = APIRouter()
@router.get("/recommendations/{user_id}", response_model=List[int])
async def get_recommendations_for_user(
user_id: int,
num_recommendations: int = 5,
db: Depends(get_db),
redis_client: Depends(get_redis_client)
):
logger.info(f"Received recommendation request for user_id: {user_id}")
# 檢查用戶是否存在
user_exists = db.query(User).filter(User.id == user_id).first()
if not user_exists:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"User with ID {user_id} not found."
)
redis_key = f"user:{user_id}:recommendations"
if redis_client:
try:
cached_recommendations = redis_client.get(redis_key)
if cached_recommendations:
logger.info(f"Returning cached recommendations for user {user_id}")
return json.loads(cached_recommendations)
except Exception as e:
logger.error(f"Error accessing Redis for user {user_id}: {e}")
# 如果 Redis 讀取失敗，繼續計算推薦
logger.info(f"Cache miss for user {user_id} or Redis error. Calculating recommendations...")
start_time = time.time()
recommender = Recommender(db)
recommended_product_ids = recommender.recommend_for_user(user_id, num_recommendations)
end_time = time.time()
logger.info(f"Recommendation calculation for user {user_id} took {end_time - start_time:.4f} seconds. Result: {recommended_product_ids}")
if redis_client and recommended_product_ids:
try:
redis_client.setex(redis_key, 3600, json.dumps(recommended_product_ids)) # 快取 1 小時
logger.info(f"Recommendations for user {user_id} cached in Redis.")
except Exception as e:
logger.error(f"Error writing recommendations to Redis for user {user_id}: {e}")
return recommended_product_ids
@router.post("/recommendations/recalculate/{user_id}")
async def recalculate_user_recommendations(
user_id: int,
db: Depends(get_db),
redis_client: Depends(get_redis_client)
):
"""
觸發用戶推薦的重新計算，並更新 Redis 快取。
這個接口可以被 Laravel 異步調用。
"""
logger.info(f"Forcing recalculation for user_id: {user_id}")
user_exists = db.query(User).filter(User.id == user_id).first()
if not user_exists:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"User with ID {user_id} not found."
)
recommender = Recommender(db)
recommended_product_ids = recommender.recommend_for_user(user_id, 5) # 重新計算時使用默認數量
if redis_client and recommended_product_ids:
try:
redis_key = f"user:{user_id}:recommendations"
redis_client.setex(redis_key, 3600, json.dumps(recommended_product_ids))
logger.info(f"Recommendations for user {user_id} re-calculated and updated in Redis.")
except Exception as e:
logger.error(f"Error writing re-calculated recommendations to Redis for user {user_id}: {e}")
return {"message": f"Recommendations calculated for user {user_id} but failed to cache.", "status": "error"}
return {"message": f"Recommendations for user {user_id} re-calculated and cached."}
C. Redis 端 (redis/)
redis.conf：Redis 的配置檔，確保可以從任何主機連接。init-redis-data.sh：一個簡單的腳本，用於在 Redis 啟動後填充一些初始測試數據，方便你驗證快取命中邏輯。
D. GitHub Actions (.github/workflows/deploy.yml)
這個 YAML 文件定義了我們的 CI/CD 流程。當程式碼推送到 main 分支時，它會自動觸發以下任務：
- Laravel CI：安裝 Composer 依賴，在 SQLite 資料庫上運行 Migration 和 PHPUnit 測試。
- FastAPI CI：安裝 Python 依賴，運行 Pytest 測試。
這確保了程式碼在合併到主分支前始終是可用的。
🧪 運行測試
為了確保服務的穩定性，我們為 Laravel 和 FastAPI 分別編寫了測試：
Laravel 測試
進入 Laravel 容器執行：
Bash
docker-compose exec laravel-app php artisan test
FastAPI 測試
進入 FastAPI 容器執行：
Bash
docker-compose exec recommender-service pytest
FastAPI 的測試會使用 sqlite:///:memory: 臨時資料庫和 Mock 的 Redis 客戶端，確保測試的隔離性和執行速度。
✨ 專案心得與收穫
完成了這個推薦系統專案，回頭看確實學到了不少東西，也解決了一些實際開發中可能遇到的挑戰。跟大家分享一下這次實作的主要體會：
- 打通跨語言服務的任督二脈：過去可能習慣單一技術棧，但這次 Laravel (PHP) 和 FastAPI (Python) 的整合讓我看到不同語言在各自擅長領域的優勢。透過 Redis 作為中間橋樑，不僅實現了兩邊服務的高效溝通，也讓我對微服務架構有了更深的體會。這就像是讓不同專業的團隊能夠無縫協作，各自發揮所長。
- 數據活用與效率提升：為了讓推薦系統能快速回應，Redis 快取絕對是關鍵。我們不只讓推薦結果能高速存取，FastAPI 直接以只讀權限連接主資料庫，也避免了複雜的數據同步問題，確保推薦邏輯總是基於最新的資料。這對處理大量數據的應用來說，是穩定性的一大保障。
- 面對異常的從容應對：系統不可能永遠完美運行。在設計時，我考量了 Redis 快取失效、FastAPI 服務暫時無法響應，甚至用戶資料不足的冷啟動情境。透過設定回退機制（例如在異常情況下提供熱門商品），可以確保即使主推薦服務暫時有狀況，使用者體驗也不會斷裂。這讓系統在面對真實世界的不確定性時，顯得更為健壯。
- 從手動到自動化：過去可能覺得部署和測試很繁瑣，但這次導入 Docker 進行容器化，讓開發和部署環境保持一致，省去了很多環境配置的麻煩。再搭配 GitHub Actions 自動化 CI/CD，每次程式碼更新都能自動測試和部署，大大提升了開發效率和程式碼品質的信心。這就像為專案建立了一條自動化生產線，讓開發流程更順暢。
- 不只寫程式，更要思考架構：從前端請求、後端處理、微服務溝通、資料庫存取到快取策略，整個專案讓我對「如何將複雜功能拆解成可管理、可擴展的組件」有了更實際的理解。這次的實作，不只是寫程式碼，更多是在思考整個系統的運作邏輯與協調。
總的來說，這個專案是個很棒的學習機會，讓我在多語言微服務、高效能數據處理和自動化部署方面都累積了寶貴的實戰經驗。希望這次的分享，也能對正在學習或考慮類似架構的朋友們有所啟發！
希望這篇文章能幫助你更好地理解和展示這個 Laravel x FastAPI 推薦系統。這個專案非常適合在技術面試中進行闡述，因為它涵蓋了從微服務設計、跨語言通訊、效能優化、快取策略、資料一致性到自動化部署等一系列現代軟體工程的關鍵議題。
沒有留言:
張貼留言