在昨天的文章中,我們成功串接了 Google Maps Places API (New),讓美食 Agent 具備了搜尋真實店家、獲取即時評分與導航連結的能力。
然而,目前系統仍然屬於「無狀態(Stateless)」的單次對話:每次使用者發問,系統就像第一次認識對方一樣,既不知道使用者上次吃過什麼,也無法記住「不吃辣」、「素食」等長期個人習慣。
今天 Day 13,我們要為微服務植入長期記憶庫:利用 Python 內建的輕量級資料庫 SQLite 搭配 SQLAlchemy ORM,記錄每位 LINE 使用者的點餐足跡與歷史喜好,並在推薦時自動避開剛吃過的店家,打造真正懂你的貼身美食秘書!
.db)存在本地端,極適合中小型 Agent 與本機開發。User 表:儲存使用者的 LINE ID、預設偏好料理、最高預算與飲食忌口。RecommendationLog 表:記錄每次向該使用者推薦過的餐廳 ID、餐廳名稱、推薦時間,用於避免短時間內重複推薦。請先確認 PowerShell 已啟用虛擬環境((.venv)),安裝 SQLAlchemy:
pip install sqlalchemy
在 app/ 目錄下建立資料庫與模型的相關模組:
agent-backend/
├── app/
│ ├── database.py # 資料庫連線引擎與 Session
│ ├── models/ # SQLAlchemy 資料庫模型
│ │ ├── __init__.py
│ │ └── memory.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()
建立 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")
在 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!"}
修改 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
)
在終端機啟動後端:
uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
啟動後會發現專案根目錄下自動產生了 food_agent_memory.db 檔案。
打開手機 LINE 聊天室,用同一個帳號連續發送兩次相同需求的查詢:
1.第一次發問:
「推薦我板橋 300 元左右的拉麵」
➜ 機器人回覆第 1 家拉麵店(例如「双豚 ラーメン」),並在後台寫入歷史足跡。
2.第二次發問:
「我還想看看別家拉麵」
➜ 系統自動檢查歷史庫,排除「双豚 ラーメン」,主動回覆下一間全新的拉麵店(例如「春日豚骨拉麵」)!
今日結語與明日預告
今天我們為原本單純「問答式」的微服務賦予了長期記憶的能力:
透過 SQLite + SQLAlchemy ORM 架構出穩健的本機資料層。
透過 User ID 綁定足跡,實現智能去重與新鮮度推薦。
讓對話 Agent 從單純的搜尋工具,真正進化成具備個人化歷程的智慧助理!
目前為止,所有服務依然跑在我們自己的電腦中,只要關機或關掉 ngrok,LINE 機器人就會停止回應。
明天 Day 14,我們將把整套系統搬上雲端:使用 Dockerfile 打包後端微服務,並免費部署至 Render 雲端平台,實現 24 小時不關機的真正生產級服務! 敬請期待!