iT邦幫忙

0

# Day 13 整合 SQLite 儲存使用者喜好足跡與回訪推薦

n8n
  • 分享至 

  • xImage
  •  

在昨天的文章中,我們成功串接了 Google Maps Places API (New),讓美食 Agent 具備了搜尋真實店家、獲取即時評分與導航連結的能力。

然而,目前系統仍然屬於「無狀態(Stateless)」的單次對話:每次使用者發問,系統就像第一次認識對方一樣,既不知道使用者上次吃過什麼,也無法記住「不吃辣」、「素食」等長期個人習慣。

今天 Day 13,我們要為微服務植入長期記憶庫:利用 Python 內建的輕量級資料庫 SQLite 搭配 SQLAlchemy ORM,記錄每位 LINE 使用者的點餐足跡與歷史喜好,並在推薦時自動避開剛吃過的店家,打造真正懂你的貼身美食秘書!


一、記憶架構設計:為什麼選擇 SQLite + SQLAlchemy?

  • 零配置、易攜帶:SQLite 無需額外架設資料庫伺服器,直接以單一檔案(.db)存在本地端,極適合中小型 Agent 與本機開發。
  • SQLAlchemy ORM:採用物件關聯對映(ORM),讓我們能用標準的 Python Class 與函式來操作資料表,未來若想無縫遷移至 PostgreSQL 或 MySQL 也無需修改核心業務程式碼。

記憶資料模型規劃:

  1. User 表:儲存使用者的 LINE ID、預設偏好料理、最高預算與飲食忌口。
  2. RecommendationLog 表:記錄每次向該使用者推薦過的餐廳 ID、餐廳名稱、推薦時間,用於避免短時間內重複推薦。

二、環境建置與依賴安裝

請先確認 PowerShell 已啟用虛擬環境((.venv)),安裝 SQLAlchemy:

pip install sqlalchemy

三、資料庫核心與 ORM 模型建立

在 app/ 目錄下建立資料庫與模型的相關模組:

agent-backend/
├── app/
│   ├── database.py       # 資料庫連線引擎與 Session
│   ├── models/           # SQLAlchemy 資料庫模型
│   │   ├── __init__.py
│   │   └── memory.py

Step 1:建立資料庫連線引擎 (app/database.py)

建立 app/database.py,配置本地 SQLite 檔案連線:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DATABASE_URL = "sqlite:///./food_agent_memory.db"

# connect_args 僅在 SQLite 下使用,避免多執行緒操作衝突
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db():
    """FastAPI 依賴注入取得資料庫 Session"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Step 2:定義資料庫實體模型 (app/models/memory.py)

建立 app/models/memory.py:

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Text
from sqlalchemy.orm import relationship
from datetime import datetime
from app.database import Base

class UserModel(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    line_user_id = Column(String(64), unique=True, index=True, nullable=False)
    favorite_cuisine = Column(String(50), default="全部")
    default_budget = Column(Integer, default=500)
    dietary_notes = Column(Text, default="")
    created_at = Column(DateTime, default=datetime.utcnow)

    # 關聯足跡紀錄
    history = relationship("RecommendationHistoryModel", back_populates="user")

class RecommendationHistoryModel(Base):
    __tablename__ = "recommendation_history"

    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    restaurant_name = Column(String(100), nullable=False)
    cuisine = Column(String(50), nullable=False)
    location = Column(String(100), nullable=False)
    recommended_at = Column(DateTime, default=datetime.utcnow)

    user = relationship("UserModel", back_populates="history")

四、資料庫初始化與服務整合

Step 1:應用程式啟動時自動建立資料表 (app/main.py)

在 app/main.py 加入資料表初始化指令:

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.database import engine, Base
from app.routers import health, recommendation

# 啟動時自動建立尚未存在的資料表
Base.metadata.create_all(bind=engine)

app = FastAPI(
    title="Food Agent Microservice API",
    description="具備長期記憶庫的智慧餐飲推薦微服務",
    version="1.1.0"
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(health.router)
app.include_router(recommendation.router)

@app.get("/")
async def root():
    return {"message": "Food Agent API with Memory is Running!"}

Step 2:在推薦端點注入記憶過濾與儲存邏輯 (app/routers/recommendation.py)

修改 app/routers/recommendation.py,加入依賴注入 db: Session,實現:

1.自動建檔/讀取使用者:若為新用戶則建立記錄。

2.過濾剛吃過的名單:排除近期(如前 3 次)推薦過的重複店家。

3.寫入本次足跡:將本次推薦的首選店家記錄到歷史表中。

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from typing import List
from datetime import datetime
from app.database import get_db
from app.models.memory import UserModel, RecommendationHistoryModel
from app.schemas.food import UserPreference, RecommendationResponse
from app.services.places import search_places

router = APIRouter(
    prefix="/api/v1/recommend",
    tags=["Food Recommendation"]
)

@router.post("/", response_model=RecommendationResponse)
async def recommend_food(preference: UserPreference, db: Session = Depends(get_db)):
    """
    結合記憶機制的即時推薦:排除近期推薦過的重複店家並自動寫入足跡
    """
    # 1. 查詢或建立使用者記憶
    user = db.query(UserModel).filter(UserModel.line_user_id == preference.user_id).first()
    if not user:
        user = UserModel(
            line_user_id=preference.user_id,
            favorite_cuisine=preference.cuisine.value,
            default_budget=preference.budget_max,
            dietary_notes=preference.notes or ""
        )
        db.add(user)
        db.commit()
        db.refresh(user)

    # 2. 獲取近 5 筆推薦過的歷史餐廳名稱
    recent_history = (
        db.query(RecommendationHistoryModel.restaurant_name)
        .filter(RecommendationHistoryModel.user_id == user.id)
        .order_by(RecommendationHistoryModel.recommended_at.desc())
        .limit(5)
        .all()
    )
    visited_names = {h[0] for h in recent_history}

    # 3. 呼叫 Google Places API 搜尋真實店家
    raw_places = await search_places(
        query_location=preference.location,
        query_cuisine=preference.cuisine.value,
        budget_max=preference.budget_max
    )

    # 4. 記憶過濾:排除近期已推薦過的店家
    fresh_places = [p for p in raw_places if p.name not in visited_names]
    final_places = fresh_places if fresh_places else raw_places  # 若全部都去過則保留

    # 5. 將本次首選推薦寫入資料庫
    if final_places:
        top_pick = final_places[0]
        history_entry = RecommendationHistoryModel(
            user_id=user.id,
            restaurant_name=top_pick.name,
            cuisine=top_pick.cuisine.value,
            location=top_pick.location,
            recommended_at=datetime.utcnow()
        )
        db.add(history_entry)
        db.commit()

    return RecommendationResponse(
        status="success" if final_places else "not_found",
        total_found=len(final_places),
        recommendations=final_places
    )

五、實測驗證與記憶檢查

Step 1:啟動伺服器

在終端機啟動後端:

uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

啟動後會發現專案根目錄下自動產生了 food_agent_memory.db 檔案。

Step 2:透過 LINE Bot 連續發問測試

打開手機 LINE 聊天室,用同一個帳號連續發送兩次相同需求的查詢:

1.第一次發問:

「推薦我板橋 300 元左右的拉麵」

➜ 機器人回覆第 1 家拉麵店(例如「双豚 ラーメン」),並在後台寫入歷史足跡。

2.第二次發問:

「我還想看看別家拉麵」

➜ 系統自動檢查歷史庫,排除「双豚 ラーメン」,主動回覆下一間全新的拉麵店(例如「春日豚骨拉麵」)!
今日結語與明日預告
今天我們為原本單純「問答式」的微服務賦予了長期記憶的能力:

透過 SQLite + SQLAlchemy ORM 架構出穩健的本機資料層。

透過 User ID 綁定足跡,實現智能去重與新鮮度推薦。

讓對話 Agent 從單純的搜尋工具,真正進化成具備個人化歷程的智慧助理!

目前為止,所有服務依然跑在我們自己的電腦中,只要關機或關掉 ngrok,LINE 機器人就會停止回應。
明天 Day 14,我們將把整套系統搬上雲端:使用 Dockerfile 打包後端微服務,並免費部署至 Render 雲端平台,實現 24 小時不關機的真正生產級服務! 敬請期待!


*提醒邦友,使用第三方服務/API 時,請務必評估資安風險與隱私保護
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言