建本地情感關(guān)懷服務(wù)：定時(shí)祝福與情感分析實(shí)踐)
這次我們來看一個(gè)很適合個(gè)人開發(fā)者和家庭場景的服務(wù)型小項(xiàng)目“你的幸福就是我最大的幸?！?。它不是一個(gè)圖像生成工作流也不是語音克隆工具而是一套用 FastAPI 搭起來的本地情感關(guān)懷服務(wù)用來記錄你在意的人的重要日期和情緒狀態(tài)在關(guān)鍵時(shí)刻自動(dòng)生成并推送一條祝福文案。項(xiàng)目的核心理念很樸素把“關(guān)心”變成數(shù)據(jù)、規(guī)則和定時(shí)任務(wù)不讓任何一個(gè)值得記住的日子被漏掉。這個(gè)項(xiàng)目最值得關(guān)注的點(diǎn)有幾個(gè)所有隱私數(shù)據(jù)默認(rèn)存在本地 SQLite不需要把家庭成員記錄傳到外部平臺(tái)提供完整的 REST API可以接自己的任務(wù)流支持批量生成祝福文案和定時(shí)掃描提醒硬件門檻低輕量模式在普通辦公電腦上就能運(yùn)行同時(shí)保留了接入更強(qiáng)情感分析模型的空間。如果后續(xù)要接入本地大模型做語義分析才會(huì)涉及顯卡和顯存評(píng)估基礎(chǔ)版本并不依賴 GPU。本文會(huì)帶讀者從零搭建完整閉環(huán)先初始化項(xiàng)目結(jié)構(gòu)和數(shù)據(jù)庫再啟動(dòng) API 服務(wù)然后錄入用戶和情感記錄跑一輪情感分析與祝福文案生成最后配置定時(shí)任務(wù)和批量任務(wù)隊(duì)列。每一步都給出了可復(fù)制的代碼和使用方法讀者跟著操作就能得到一個(gè)可運(yùn)行的本地服務(wù)。適合的讀者有兩類一類是想要給家人、伴侶、朋友做一套私人提醒系統(tǒng)的個(gè)人開發(fā)者另一類是正在學(xué)習(xí) FastAPI、SQLite、定時(shí)任務(wù)和本地?cái)?shù)據(jù)存儲(chǔ)全棧思路的初學(xué)者。如果只是好奇也可以先看第 5 節(jié)的功能測(cè)試和第 8 節(jié)的資源占用部分能快速判斷這套方案是否符合自己的需求。1. 核心能力速覽先把項(xiàng)目的關(guān)鍵規(guī)格放在前面方便快速判斷這個(gè)方案值不值得花時(shí)間。能力項(xiàng)說明項(xiàng)目類型本地部署的情感關(guān)懷與幸福記錄服務(wù)示例項(xiàng)目主要功能用戶管理、情感記錄、輕量情感分析、祝福文案生成、日期提醒、批量文案生成存儲(chǔ)方式SQLite 本地?cái)?shù)據(jù)庫數(shù)據(jù)默認(rèn)不出本機(jī)推薦運(yùn)行環(huán)境Python 3.9 及以上Windows / Linux / macOS 均可硬件門檻輕量模式普通辦公電腦即可接入本地大模型做語義分析時(shí)才需要評(píng)估顯存啟動(dòng)方式命令行啟動(dòng) FastAPI 服務(wù)接口 API提供 REST API可通過 /docs 查看交互文檔批量任務(wù)支持批量生成祝福文案、定時(shí)掃描提醒日期推送方式默認(rèn)輸出 JSON 文件和日志可自行擴(kuò)展郵件、機(jī)器人等通道適合場景個(gè)人與家庭隱私場景、FastAPI 全棧學(xué)習(xí)、自動(dòng)化關(guān)懷任務(wù)需要強(qiáng)調(diào)一點(diǎn)這個(gè)項(xiàng)目是一個(gè)通用實(shí)現(xiàn)模板不是某個(gè)組織發(fā)布的閉源軟件。讀者拿到的是代碼骨架實(shí)際的情感詞典、文案風(fēng)格、數(shù)據(jù)庫表結(jié)構(gòu)都可以按自己的場景改。表格里的“本地部署”“REST API”等說法都是圍繞這個(gè)模板本身的能力來描述的不綁定任何特定開源模型或第三方平臺(tái)賬號(hào)。2. 適用場景與使用邊界這個(gè)服務(wù)最合適的場景是個(gè)人和家庭內(nèi)部使用。你可以記錄父母的生日、伴侶的結(jié)婚紀(jì)念日、孩子的重要成長節(jié)點(diǎn)然后讓系統(tǒng)到點(diǎn)提醒你該打個(gè)電話該發(fā)條消息該準(zhǔn)備一份小禮物。相比在手機(jī)上單純?cè)O(shè)一個(gè)日歷鬧鐘這套項(xiàng)目的優(yōu)勢(shì)在于它把人物關(guān)系、情感記錄和文案生成串在了一起提醒的不只是一個(gè)“日期”而是一個(gè)“為什么要做這件事”的上下文。它也能用于學(xué)習(xí)完整的后端開發(fā)流程。FastAPI 負(fù)責(zé)接口層SQLite 負(fù)責(zé)持久化APScheduler 負(fù)責(zé)定時(shí)任務(wù)情感分析和模板生成負(fù)責(zé)業(yè)務(wù)邏輯。整個(gè)項(xiàng)目結(jié)構(gòu)足夠小但已經(jīng)覆蓋了一個(gè)服務(wù)端應(yīng)用的主要模塊適合作為練手項(xiàng)目逐步擴(kuò)展。使用邊界要提前說清楚。第一不要在沒有鑒權(quán)的情況下把服務(wù)暴露到公網(wǎng)默認(rèn)監(jiān)聽 127.0.0.1 就是為安全考慮的如果要多設(shè)備訪問應(yīng)該自己加 Token 或身份校驗(yàn)。第二情感分析結(jié)果只做參考它不能替代心理咨詢或醫(yī)療判斷如果發(fā)現(xiàn)記錄對(duì)象長期處于明顯的負(fù)面情緒應(yīng)該用真實(shí)的人際關(guān)懷去介入。第三如果你要把家人的情緒記錄、日期信息錄入系統(tǒng)最好征得對(duì)方同意這些信息屬于個(gè)人敏感數(shù)據(jù)。第四未來如果擴(kuò)展成自動(dòng)發(fā)送短信、微信消息、生成照片或語音祝福必須使用合規(guī)的推送通道并確保肖像權(quán)和聲音權(quán)已經(jīng)獲得授權(quán)。3. 環(huán)境準(zhǔn)備與項(xiàng)目結(jié)構(gòu)先準(zhǔn)備 Python 環(huán)境。推薦使用虛擬環(huán)境隔離依賴避免和系統(tǒng) Python 包沖突。Windows 用戶在命令提示符或 PowerShell 中執(zhí)行Linux / macOS 用戶使用終端。mkdir happiness_service cd happiness_service python -m venv venv激活虛擬環(huán)境# Windows venv\Scripts\activate # Linux / macOS source venv/bin/activate創(chuàng)建依賴文件 requirements.txt寫入以下內(nèi)容fastapi0.110 uvicorn0.29 apscheduler3.10 pydantic2.6然后安裝依賴pip install -r requirements.txt依賴安裝完成后創(chuàng)建項(xiàng)目目錄結(jié)構(gòu)。推薦按下面的方式組織文件后續(xù)擴(kuò)展功能時(shí)不容易亂happiness_service/ ├── app.py ├── config.py ├── database.py ├── analyzer.py ├── template_engine.py ├── scheduler.py ├── requirements.txt ├── data/ └── scripts/ └── send_blessing.py每個(gè)文件的職責(zé)如下config.py 保存路徑和端口配置database.py 負(fù)責(zé) SQLite 初始化和增刪改查analyzer.py 做輕量情感分析template_engine.py 負(fù)責(zé)祝福文案生成scheduler.py 是定時(shí)提醒任務(wù)app.py 是 FastAPI 入口scripts 目錄放獨(dú)立的批量腳本。先寫 config.py把路徑集中管理import os BASE_DIR os.path.dirname(os.path.abspath(__file__)) DATA_DIR os.path.join(BASE_DIR, data) DB_PATH os.path.join(DATA_DIR, happiness.db) HOST os.getenv(HAPPINESS_HOST, 127.0.0.1) PORT int(os.getenv(HAPPINESS_PORT, 8000))然后寫 database.py包含建表和基礎(chǔ)數(shù)據(jù)訪問方法import sqlite3 from pathlib import Path from config import DATA_DIR, DB_PATH def get_connection(): Path(DB_PATH).parent.mkdir(parentsTrue, exist_okTrue) conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row return conn def init_db(): conn get_connection() conn.executescript( CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, relationship TEXT DEFAULT default, birthday TEXT, anniversary_date TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS records ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, content TEXT NOT NULL, mood_type TEXT, mood_score REAL, keywords TEXT, remark TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(id) ); CREATE TABLE IF NOT EXISTS remind_tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, task_type TEXT NOT NULL, task_date TEXT NOT NULL, message TEXT, status TEXT DEFAULT pending, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); ) conn.commit() conn.close()數(shù)據(jù)表設(shè)計(jì)為三張users 保存人物信息records 保存每次情感記錄和分析結(jié)果remind_tasks 保存待觸發(fā)的提醒任務(wù)。birthday 和 anniversary_date 統(tǒng)一使用“YYYY-MM-DD”格式字符串方便定時(shí)任務(wù)按月日匹配。analyzer.py 先實(shí)現(xiàn)一個(gè)離線情感詞典版本不需要聯(lián)網(wǎng)也不需要 GPU。這個(gè)版本用來演示完整流程夠了后續(xù)想提高準(zhǔn)確率可以把 analyze_sentiment 函數(shù)內(nèi)部換成調(diào)用本地部署的情感分類模型。POSITIVE_WORDS {幸福, 開心, 快樂, 溫暖, 感動(dòng), 滿足, 喜歡, 順利, 健康, 值得} NEGATIVE_WORDS {難過, 失落, 疲憊, 焦慮, 生氣, 委屈, 壓力, 糟糕, 失眠, 擔(dān)心} def analyze_sentiment(text: str) - dict: positive_hits [w for w in POSITIVE_WORDS if w in text] negative_hits [w for w in NEGATIVE_WORDS if w in text] if not positive_hits and not negative_hits: return { mood_type: neutral, mood_score: 5.0, keywords: [], } total len(positive_hits) len(negative_hits) ratio len(positive_hits) / total if ratio 0.6: mood_type positive elif ratio 0.4: mood_type negative else: mood_type neutral return { mood_type: mood_type, mood_score: round(ratio * 10, 1), keywords: positive_hits negative_hits, }template_engine.py 使用模板生成祝福文案。這里先做離線模板版本避免把家庭場景的文案請(qǐng)求發(fā)送到外部大模型接口。按場景區(qū)分生日、紀(jì)念日、日常問候后續(xù)可以按人物關(guān)系繼續(xù)擴(kuò)展模板庫。TEMPLATES { birthday: [ {name}生日快樂。愿你想要的都擁有愿你珍惜的人一直陪在身邊。你的幸福就是我最大的幸福。, 今天是你的生日{(diào)name}。不用趕著成為誰今天只做開心的自己。, ], anniversary: [ {name}紀(jì)念日快樂。謝謝你在那些平常日子里給的陪伴愿以后每個(gè)重要日子我們都不缺席。, 屬于我們的紀(jì)念日又到了{(lán)name}。時(shí)間在走想給你的心意不變。, ], daily: [ {name}今天有沒有按時(shí)吃飯稍微停下來歇一會(huì)兒別讓自己太累。, 想跟你說聲{name}今天也辛苦了。你的笑容就是我最想守護(hù)的東西。, ], } RELATIONSHIP_NOTES { partner: 最珍惜的人, parent: 最想孝順的人, child: 最想守護(hù)的人, friend: 重要的朋友, default: 在乎的人, } def build_relationship_prefix(relationship: str) - str: return RELATIONSHIP_NOTES.get(relationship, RELATIONSHIP_NOTES[default]) def generate_blessing(user_name: str, scenario: str daily, relationship: str default) - str: templates TEMPLATES.get(scenario, TEMPLATES[daily]) template templates[0] return template.format(nameuser_name)到這一步項(xiàng)目骨架已經(jīng)完整。沒有 GPU 也能跑因?yàn)槟氵€沒有調(diào)用任何深度學(xué)習(xí)模型。下面進(jìn)入服務(wù)啟動(dòng)和接口驗(yàn)證。4. 啟動(dòng) API 服務(wù)與基礎(chǔ)驗(yàn)證現(xiàn)在寫 FastAPI 入口 app.py。這里直接調(diào)用 init_db 初始化數(shù)據(jù)庫減少啟動(dòng)時(shí)序問題。再定義用戶創(chuàng)建、情感記錄、情感趨勢(shì)和祝福生成等接口。from typing import Optional from fastapi import FastAPI from pydantic import BaseModel from analyzer import analyze_sentiment from config import init_app_config from database import ( add_record, add_user, get_mood_trend, init_db, list_records, list_users, ) from template_engine import generate_blessing app FastAPI(title你的幸福就是我最大的幸福, version0.1.0) init_db() class UserCreate(BaseModel): name: str relationship: str default birthday: Optional[str] None anniversary_date: Optional[str] None class RecordCreate(BaseModel): user_id: int content: str remark: Optional[str] None class BlessingRequest(BaseModel): name: str scenario: str daily relationship: str default app.get(/health) def health_check(): return {status: ok, service: happiness-service} app.post(/users) def create_user(payload: UserCreate): user_id add_user( namepayload.name, relationshippayload.relationship, birthdaypayload.birthday, anniversary_datepayload.anniversary_date, ) return {user_id: user_id, message: 用戶創(chuàng)建成功} app.get(/users) def query_users(): return list_users() app.post(/records) def create_record(payload: RecordCreate): mood_result analyze_sentiment(payload.content) record_id add_record( user_idpayload.user_id, contentpayload.content, mood_resultmood_result, remarkpayload.remark, ) return {record_id: record_id, analysis: mood_result} app.get(/users/{user_id}/records) def query_records(user_id: int): return list_records(user_iduser_id) app.get(/users/{user_id}/mood_trend) def query_mood_trend(user_id: int): return get_mood_trend(user_iduser_id) app.post(/blessing) def create_blessing(payload: BlessingRequest): return { scenario: payload.scenario, message: generate_blessing( user_namepayload.name, scenariopayload.scenario, relationshippayload.relationship, ), }注意這個(gè)版本沒有給接口加鑒權(quán)只適合本機(jī)測(cè)試。啟動(dòng)命令如下uvicorn app:app --host 127.0.0.1 --port 8000啟動(dòng)后終端會(huì)輸出訪問地址。瀏覽器打開 http://127.0.0.1:8000/docs如果能看到 FastAPI 自帶的 Swagger 交互文檔說明服務(wù)已經(jīng)跑起來了。Swagger 頁面會(huì)把所有接口列出并且支持直接在頁面上點(diǎn)擊 Try it out 發(fā)送請(qǐng)求這是最方便的接口驗(yàn)證方式。也可以先驗(yàn)證健康檢查接口curl http://127.0.0.1:8000/health返回結(jié)果{status:ok,service:happiness-service}看到這個(gè)輸出服務(wù)就是可用的。接下來開始錄入真實(shí)數(shù)據(jù)做功能測(cè)試。5. 功能測(cè)試情感記錄與祝福生成先創(chuàng)建一個(gè)用戶。比如要記錄媽媽的信息接口請(qǐng)求如下curl -X POST http://127.0.0.1:8000/users \ -H Content-Type: application/json \ -d {name: 媽媽, relationship: parent, birthday: 1965-08-20}預(yù)期返回{user_id:1,message:用戶創(chuàng)建成功}如果返回其他內(nèi)容先看 Swagger 頁面里的錯(cuò)誤提示通常是 JSON 格式錯(cuò)誤或者日期字段格式不對(duì)。創(chuàng)建用戶后添加一條情感記錄。這里設(shè)計(jì)接口會(huì)先調(diào)用 analyzer.py 的情感分析再把分析結(jié)果和原文一起存入數(shù)據(jù)庫。curl -X POST http://127.0.0.1:8000/records \ -H Content-Type: application/json \ -d {user_id: 1, content: 今天陪媽媽復(fù)查醫(yī)生說恢復(fù)得很好她特別開心。, remark: 復(fù)查日}預(yù)期返回{ record_id: 1, analysis: { mood_type: positive, mood_score: 10.0, keywords: [開心] } }這里 mood_type 有三種可能positive、neutral、negative。mood_score 范圍是 0 到 10分?jǐn)?shù)越高代表文本中包含的積極詞匯占比越高。keywords 是文本中命中的情感詞典詞匯。因?yàn)樵睦镉小伴_心”而沒有負(fù)面詞所以判定為 positive分?jǐn)?shù)是 10。再測(cè)試一條負(fù)面記錄curl -X POST http://127.0.0.1:8000/records \ -H Content-Type: application/json \ -d {user_id: 1, content: 她說最近總是失眠工作壓力很大。, remark: 日常記錄}返回結(jié)果里 mood_type 應(yīng)該是 negativemood_score 偏低keywords 至少包含“失眠”“壓力”。這說明情感分析接口能區(qū)分正向和負(fù)向表達(dá)。情感趨勢(shì)接口能把一段時(shí)間內(nèi)的記錄聚合成折線數(shù)據(jù)。請(qǐng)求curl http://127.0.0.1:8000/users/1/mood_trend返回按日期分組的平均 mood_score。這個(gè)接口適合后續(xù)做一個(gè)小型 Web 看板用來觀察記錄對(duì)象的情緒變化。祝福生成接口測(cè)試curl -X POST http://127.0.0.1:8000/blessing \ -H Content-Type: application/json \ -d {name: 媽媽, scenario: daily, relationship: parent}預(yù)期返回{ scenario: daily, message: 媽媽今天有沒有按時(shí)吃飯稍微停下來歇一會(huì)兒別讓自己太累。 }如果返回的是英文或者亂碼檢查終端編碼。在 Windows 的 PowerShell 里發(fā)送中文 JSON 可能會(huì)出現(xiàn)編碼問題更穩(wěn)妥的辦法是用 Python 腳本調(diào)用接口或者把 JSON 寫入文件再通過 curl 的 --data file 方式發(fā)送。6. 接口 API 調(diào)用示例接口跑通后可以把服務(wù)接到自己的腳本里。下面是 Python 調(diào)用示例適合在本地批量錄入數(shù)據(jù)和生成文案。import requests BASE http://127.0.0.1:8000 # 創(chuàng)建用戶 user requests.post( f{BASE}/users, json{ name: 媽媽, relationship: parent, birthday: 1965-08-20, anniversary_date: None, }, ).json() print(user) # 添加一條情感記錄 record requests.post( f{BASE}/records, json{ user_id: user[user_id], content: 今天陪媽媽復(fù)查醫(yī)生說恢復(fù)得很好她特別開心。, remark: 復(fù)查日, }, ).json() print(record) # 生成祝福文案 blessing requests.post( f{BASE}/blessing, json{ name: 媽媽, scenario: daily, relationship: parent, }, ).json() print(blessing[message])如果只需要生成文案不一定要先創(chuàng)建用戶。祝福接口只依賴 name、scenario、relationship 三個(gè)字段可以直接使用。批量生成祝福文案時(shí)建議把名單放到一個(gè) JSON 文件里例如 people.json[ {name: 爸爸, relationship: parent, scenario: birthday}, {name: 伴侶, relationship: partner, scenario: anniversary}, {name: 朋友, relationship: friend, scenario: daily} ]再寫一個(gè)批量腳本 scripts/send_blessing.pyimport argparse import json import requests BASE http://127.0.0.1:8000 def load_people(path): with open(path, encodingutf-8) as f: return json.load(f) def batch_generate(people): results [] for person in people: response requests.post( f{BASE}/blessing, json{ name: person[name], scenario: person.get(scenario, daily), relationship: person.get(relationship, default), }, timeout10, ) data response.json() results.append( { name: person[name], message: data[message], } ) return results if __name__ __main__: parser argparse.ArgumentParser(description批量生成祝福文案) parser.add_argument(--people, defaultpeople.json, help名單文件) parser.add_argument(--output, defaultblessing_output.json, help輸出文件) args parser.parse_args() people load_people(args.people) results batch_generate(people) with open(args.output, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) print(f已生成 {len(results)} 條文案輸出到 {args.output})運(yùn)行命令python scripts/send_blessing.py --people people.json --output blessing_output.json運(yùn)行后打開 blessing_output.json應(yīng)該能看到三條已經(jīng)填充好人名的祝福文案。這樣即使一次要寫幾十個(gè)人的祝福也只需要維護(hù)一份名單文件剩下的事情交給腳本。批量腳本有一個(gè)明顯的優(yōu)點(diǎn)它只依賴 /blessing 接口改動(dòng)模板時(shí)不需要重新學(xué)腳本。缺點(diǎn)是如果服務(wù)沒有啟動(dòng)所有請(qǐng)求都會(huì)失敗所以腳本里需要增加超時(shí)和異常處理。更完整的版本可以加日志記錄把失敗的用戶單獨(dú)輸出到一個(gè) retry.json 文件方便二次重跑。7. 定時(shí)提醒與批量任務(wù)擴(kuò)展只記錄數(shù)據(jù)和生成文案還不夠真正讓“你的幸福就是我最大的幸?！甭涞氐氖嵌〞r(shí)提醒。APScheduler 可以放在服務(wù)里每隔一段時(shí)間掃描一次數(shù)據(jù)庫把當(dāng)天需要提醒的人物找出來寫入隊(duì)列文件。scheduler.py 實(shí)現(xiàn)定時(shí)掃描邏輯import json import os from datetime import datetime from apscheduler.schedulers.background import BackgroundScheduler from config import DATA_DIR from database import get_connection QUEUE_FILE os.path.join(DATA_DIR, remind_queue.json) def scan_reminders(): today datetime.now().strftime(%m-%d) conn get_connection() rows conn.execute( SELECT u.id, u.name, u.relationship, u.birthday, u.anniversary_date FROM users u WHERE (substr(u.birthday, 6) ?) OR (substr(u.anniversary_date, 6) ?) , (today, today), ).fetchall() pending [] for row in rows: user dict(row) if user[birthday] and user[birthday][5:] today: task_type birthday elif user[anniversary_date] and user[anniversary_date][5:] today: task_type anniversary else: continue pending.append( { user_id: user[id], name: user[name], task_type: task_type, date: datetime.now().strftime(%Y-%m-%d), } ) conn.close() os.makedirs(DATA_DIR, exist_okTrue) with open(QUEUE_FILE, w, encodingutf-8) as f: json.dump(pending, f, ensure_asciiFalse, indent2) if pending: print(f[reminder] 檢測(cè)到 {len(pending)} 條應(yīng)提醒事項(xiàng)) else: print([reminder] 今天沒有匹配的日期) def start_scheduler(): scheduler BackgroundScheduler() scheduler.add_job(scan_reminders, interval, minutes1, idscan_reminders) scheduler.start() print([scheduler] 定時(shí)任務(wù)已啟動(dòng)每 1 分鐘掃描一次) return scheduler這個(gè)邏輯的核心是把日期字符串的“MM-DD”部分和今天的“MM-DD”比較。birthday 如果傳的是 “1965-08-20”substr(birthday, 6) 得到 “08-20”和今天的 “08-20” 匹配就說明今天是這個(gè)人的生日。定時(shí)任務(wù)不建議通過接口啟動(dòng)因?yàn)?uvicorn 開啟 --reload 時(shí)可能重復(fù)啟動(dòng)多個(gè)調(diào)度器。更穩(wěn)妥的做法是把定時(shí)任務(wù)獨(dú)立成一個(gè)進(jìn)程。在 scripts 下新建 run_scheduler.pyimport sys import threading from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from scheduler import start_scheduler if __name__ __main__: scheduler start_scheduler() print(按 CtrlC 退出) try: threading.Event().wait() except KeyboardInterrupt: scheduler.shutdown()運(yùn)行python scripts/run_scheduler.py默認(rèn)情況下掃描結(jié)果寫到 data/remind_queue.json。實(shí)際使用中你可以在 scan_reminders 函數(shù)末尾追加真實(shí)的推送通道比如發(fā)送郵件、調(diào)用企業(yè)微信群機(jī)器人、寫入自己的待辦應(yīng)用。推送通道需要單獨(dú)申請(qǐng)權(quán)限因此這套項(xiàng)目先統(tǒng)一走 JSON 文件方便你自己對(duì)接。批量任務(wù)可以分為兩種。一種是周期性批量生成文案用第 6 節(jié)的 send_blessing.py 就能實(shí)現(xiàn)。另一種是批量發(fā)送提醒需要你把發(fā)送狀態(tài)回寫到數(shù)據(jù)庫的 remind_tasks 表里至少增加一個(gè) processed 字段避免每次掃描都重復(fù)發(fā)送同一條消息。這是從示例走向生產(chǎn)環(huán)境必須補(bǔ)上的細(xì)節(jié)。8. 資源占用與性能觀察這個(gè)項(xiàng)目的基礎(chǔ)版本沒有任何深度學(xué)習(xí)模型資源占用很低。啟動(dòng) uvicorn 和 APScheduler 后主要消耗來自 Python 進(jìn)程本身。普通辦公電腦上內(nèi)存占用通常在幾十 MB 到一兩百 MB 區(qū)間具體數(shù)值會(huì)因 Python 版本和已安裝依賴不同而有差異。CPU 占用在沒有請(qǐng)求時(shí)接近 0每隔一分鐘的定時(shí)掃描任務(wù)也只需要執(zhí)行幾條 SQLite 查詢和少量字符串比較。顯存占用在這個(gè)項(xiàng)目里是可選項(xiàng)。如果你只使用情感詞典和模板生成整個(gè)過程完全不需要 GPU。后續(xù)如果需要更準(zhǔn)確的情感語義分析可以在 analyzer.py 中改為調(diào)用本地部署的開源情感分類模型或本地大模型這時(shí)才需要關(guān)注顯存。顯存占用取決于你接的模型參數(shù)量、輸入文本長度和推理框架沒有統(tǒng)一數(shù)字必須按本機(jī)實(shí)測(cè)為準(zhǔn)。觀察資源占用有三種方式。第一Windows 任務(wù)管理器里看 python 進(jìn)程的 CPU 和內(nèi)存。第二Linux 下使用 top 或 htop 查看。第三如果接入了本地 GPU 模型使用 nvidia-smi 觀察當(dāng)前進(jìn)程的顯存占用。nvidia-smi批量任務(wù)對(duì)性能的影響要分開看。批量生成祝福文案時(shí)瓶頸一般在模板引擎和 JSON 序列化數(shù)千條文案也是秒級(jí)完成。批量寫入情感記錄時(shí)瓶頸可能出現(xiàn)在 SQLite 寫入頻率上建議批量插入時(shí)使用事務(wù)而不是逐條提交。另外定時(shí)掃描任務(wù)如果間隔太短并且掃描邏輯里加了外部推送就要考慮推送通道的頻率限制避免被平臺(tái)當(dāng)成垃圾請(qǐng)求。內(nèi)存泄漏和端口殘留是本地服務(wù)最常見的兩個(gè)問題。開發(fā)調(diào)試時(shí)uvicorn 進(jìn)程沒有正常退出再次啟動(dòng)就會(huì)提示端口被占用。遇到這種情況先找到占用端口的進(jìn)程并終止再重新啟動(dòng)服務(wù)。# Windows netstat -aon | findstr :8000 taskkill /PID 進(jìn)程號(hào) /F # Linux / macOS lsof -i :8000 kill -9 進(jìn)程號(hào)9. 常見問題與排查方法本地部署這個(gè)流程不長但會(huì)遇到一些典型問題。整理成表格方便直接對(duì)照。問題現(xiàn)象可能原因排查方式解決方案啟動(dòng)時(shí)提示端口被占用8000 端口被其他進(jìn)程使用netstat 或 lsof 查看端口占用關(guān)閉占用進(jìn)程或修改 config.py 里的 PORTuvicorn 命令找不到?jīng)]有激活虛擬環(huán)境或依賴未安裝檢查命令行前是否出現(xiàn) venv 前綴激活虛擬環(huán)境后重新安裝 requirements.txt運(yùn)行時(shí)提示 sqlite3.OperationalError: database is locked多個(gè)進(jìn)程同時(shí)寫入數(shù)據(jù)庫檢查是否有多個(gè) API 進(jìn)程或調(diào)度器進(jìn)程關(guān)掉重復(fù)進(jìn)程使用 WAL 模式避免多進(jìn)程寫同一個(gè) SQLite 文件中文返回亂碼Windows 終端編碼不是 UTF-8在 Swagger 頁面看響應(yīng)體是否正常使用 Python 腳本調(diào)用或把 JSON 寫入文件再讀取情感分析結(jié)果全部為 neutral文本中未命中情感詞典詞匯打印 keywords 字段擴(kuò)充 POSITIVE_WORDS 和 NEGATIVE_WORDS或者接入更強(qiáng)模型定時(shí)任務(wù)一直沒有輸出APScheduler 未啟動(dòng)或用戶日期格式不對(duì)確認(rèn) run_scheduler.py 是否在運(yùn)行檢查 users 表中日期格式統(tǒng)一日期為 YYYY-MM-DD確認(rèn)掃描進(jìn)程已啟動(dòng)POST /records 返回 422JSON 字段名稱或類型不匹配查看響應(yīng)中的 detail 錯(cuò)誤信息按 Pydantic 模型傳字段user_id 必須為數(shù)字調(diào)用 /docs 無法打開服務(wù)未啟動(dòng)或?yàn)g覽器代理異常檢查后臺(tái)進(jìn)程和訪問地址確認(rèn) uvicorn 啟動(dòng)成功訪問 127.0.0.1 而不是遠(yuǎn)程地址重啟后數(shù)據(jù)丟失data 目錄未生成或數(shù)據(jù)庫路徑錯(cuò)誤檢查 data/happiness.db 是否存在確認(rèn) config.py 中 DATA_DIR 路徑正確如果接口報(bào) 422通常是請(qǐng)求 JSON 和 Pydantic 模型字段不一致。比如 RecordCreate 要求的是 user_id 和 content你卻傳了 userId 和 text校驗(yàn)直接失敗。先看 Swagger 文檔中的請(qǐng)求示例再對(duì)比自己的請(qǐng)求體。情感分析部分最容易讓使用者誤判。當(dāng)前默認(rèn)的情感詞典只有十幾個(gè)詞不可能覆蓋復(fù)雜的中文表達(dá)。比如“雖然很累但是看到他就踏實(shí)了”這句話可能同時(shí)命中“累”和“踏實(shí)”需要靠情緒分?jǐn)?shù)加權(quán)判斷。更準(zhǔn)確的做法是把 analyzer.py 里的邏輯替換成語義模型而不是繼續(xù)堆關(guān)鍵詞。10. 最佳實(shí)踐、合規(guī)建議與下一步第一次運(yùn)行時(shí)不要急著錄入大量數(shù)據(jù)。先創(chuàng)建 1 到 2 個(gè)測(cè)試用戶分別插入正向和負(fù)向情感記錄確認(rèn)情感分析、祝福生成、定時(shí)掃描三個(gè)主鏈路都正常再開始正式使用。數(shù)據(jù)備份要養(yǎng)成習(xí)慣。SQLite 只有一個(gè)文件直接復(fù)制 data/happiness.db 就能備份。本地服務(wù)沒有自動(dòng)備份機(jī)制建議手動(dòng)備份或者寫一個(gè)簡單的定時(shí)復(fù)制腳本。敏感數(shù)據(jù)不要放到公開倉庫尤其是包含真實(shí)姓名、出生日期、關(guān)系標(biāo)簽的數(shù)據(jù)庫文件。合規(guī)層面控制三點(diǎn)。第一只在獲得當(dāng)事人知情同意后記錄信息不要把工具變成暗中監(jiān)視他人情緒的手段。第二推送消息要控制頻率和措辭避免給對(duì)方造成打擾。如果對(duì)方明確表示不需要應(yīng)該允許關(guān)閉對(duì)該用戶的提醒。第三這個(gè)項(xiàng)目不適合用來做心理狀態(tài)診斷。情感分析只是輔助真實(shí)的關(guān)系維護(hù)仍然需要面對(duì)面溝通。下一步的擴(kuò)展方向比較明確。第一把模板引擎升級(jí)為支持多個(gè)模板隨機(jī)選擇避免每次都發(fā)一模一樣的文案。第二給 analyzer.py 接入本地情感分類模型提高語義判斷能力。第三把數(shù)據(jù)從 SQLite 換成 PostgreSQL增加用戶鑒權(quán)和多端訪問就能從個(gè)人工具升級(jí)成一個(gè)小型家庭服務(wù)。第四增加 Web 看板把情緒趨勢(shì)用圖表展示出來讓記錄變得可視化。第五設(shè)計(jì)發(fā)送確認(rèn)機(jī)制在 send_blessing.py 里加入“已發(fā)送”狀態(tài)防止同一批人重復(fù)收到祝福。如果你也想做一套“幸福守護(hù)”小服務(wù)建議從今天的項(xiàng)目結(jié)構(gòu)開始先跑通 FastAPI 接口再加入定時(shí)任務(wù)最后逐步增強(qiáng)分析和推送能力。代碼不復(fù)雜但至少能把你的關(guān)心從“心里惦記”變成一條準(zhǔn)時(shí)的消息。你的幸福就是我最大的幸福這句話放到代碼里就是一張人物表、一套情感分析規(guī)則、一批定時(shí)任務(wù)以及一條不遲到的祝福。 用FastAPI和SQLite構(gòu)建本地情感關(guān)懷服務(wù)：定時(shí)祝福與情感分析實(shí)踐
這次我們來看一個(gè)很適合個(gè)人開發(fā)者和家庭場景的服務(wù)型小項(xiàng)目“你的幸福就是我最大的幸?！?。它不是一個(gè)圖像生成工作流也不是語音克隆工具而是一套用 FastAPI 搭起來的本地情感關(guān)懷服務(wù)用來記錄你在意的人的重要日期和情緒狀態(tài)在關(guān)鍵時(shí)刻自動(dòng)生成并推送一條祝福文案。項(xiàng)目的核心理念很樸素把“關(guān)心”變成數(shù)據(jù)、規(guī)則和定時(shí)任務(wù)不讓任何一個(gè)值得記住的日子被漏掉。這個(gè)項(xiàng)目最值得關(guān)注的點(diǎn)有幾個(gè)所有隱私數(shù)據(jù)默認(rèn)存在本地 SQLite不需要把家庭成員記錄傳到外部平臺(tái)提供完整的 REST API可以接自己的任務(wù)流支持批量生成祝福文案和定時(shí)掃描提醒硬件門檻低輕量模式在普通辦公電腦上就能運(yùn)行同時(shí)保留了接入更強(qiáng)情感分析模型的空間。如果后續(xù)要接入本地大模型做語義分析才會(huì)涉及顯卡和顯存評(píng)估基礎(chǔ)版本并不依賴 GPU。本文會(huì)帶讀者從零搭建完整閉環(huán)先初始化項(xiàng)目結(jié)構(gòu)和數(shù)據(jù)庫再啟動(dòng) API 服務(wù)然后錄入用戶和情感記錄跑一輪情感分析與祝福文案生成最后配置定時(shí)任務(wù)和批量任務(wù)隊(duì)列。每一步都給出了可復(fù)制的代碼和使用方法讀者跟著操作就能得到一個(gè)可運(yùn)行的本地服務(wù)。適合的讀者有兩類一類是想要給家人、伴侶、朋友做一套私人提醒系統(tǒng)的個(gè)人開發(fā)者另一類是正在學(xué)習(xí) FastAPI、SQLite、定時(shí)任務(wù)和本地?cái)?shù)據(jù)存儲(chǔ)全棧思路的初學(xué)者。如果只是好奇也可以先看第 5 節(jié)的功能測(cè)試和第 8 節(jié)的資源占用部分能快速判斷這套方案是否符合自己的需求。1. 核心能力速覽先把項(xiàng)目的關(guān)鍵規(guī)格放在前面方便快速判斷這個(gè)方案值不值得花時(shí)間。能力項(xiàng)說明項(xiàng)目類型本地部署的情感關(guān)懷與幸福記錄服務(wù)示例項(xiàng)目主要功能用戶管理、情感記錄、輕量情感分析、祝福文案生成、日期提醒、批量文案生成存儲(chǔ)方式SQLite 本地?cái)?shù)據(jù)庫數(shù)據(jù)默認(rèn)不出本機(jī)推薦運(yùn)行環(huán)境Python 3.9 及以上Windows / Linux / macOS 均可硬件門檻輕量模式普通辦公電腦即可接入本地大模型做語義分析時(shí)才需要評(píng)估顯存啟動(dòng)方式命令行啟動(dòng) FastAPI 服務(wù)接口 API提供 REST API可通過 /docs 查看交互文檔批量任務(wù)支持批量生成祝福文案、定時(shí)掃描提醒日期推送方式默認(rèn)輸出 JSON 文件和日志可自行擴(kuò)展郵件、機(jī)器人等通道適合場景個(gè)人與家庭隱私場景、FastAPI 全棧學(xué)習(xí)、自動(dòng)化關(guān)懷任務(wù)需要強(qiáng)調(diào)一點(diǎn)這個(gè)項(xiàng)目是一個(gè)通用實(shí)現(xiàn)模板不是某個(gè)組織發(fā)布的閉源軟件。讀者拿到的是代碼骨架實(shí)際的情感詞典、文案風(fēng)格、數(shù)據(jù)庫表結(jié)構(gòu)都可以按自己的場景改。表格里的“本地部署”“REST API”等說法都是圍繞這個(gè)模板本身的能力來描述的不綁定任何特定開源模型或第三方平臺(tái)賬號(hào)。2. 適用場景與使用邊界這個(gè)服務(wù)最合適的場景是個(gè)人和家庭內(nèi)部使用。你可以記錄父母的生日、伴侶的結(jié)婚紀(jì)念日、孩子的重要成長節(jié)點(diǎn)然后讓系統(tǒng)到點(diǎn)提醒你該打個(gè)電話該發(fā)條消息該準(zhǔn)備一份小禮物。相比在手機(jī)上單純?cè)O(shè)一個(gè)日歷鬧鐘這套項(xiàng)目的優(yōu)勢(shì)在于它把人物關(guān)系、情感記錄和文案生成串在了一起提醒的不只是一個(gè)“日期”而是一個(gè)“為什么要做這件事”的上下文。它也能用于學(xué)習(xí)完整的后端開發(fā)流程。FastAPI 負(fù)責(zé)接口層SQLite 負(fù)責(zé)持久化APScheduler 負(fù)責(zé)定時(shí)任務(wù)情感分析和模板生成負(fù)責(zé)業(yè)務(wù)邏輯。整個(gè)項(xiàng)目結(jié)構(gòu)足夠小但已經(jīng)覆蓋了一個(gè)服務(wù)端應(yīng)用的主要模塊適合作為練手項(xiàng)目逐步擴(kuò)展。使用邊界要提前說清楚。第一不要在沒有鑒權(quán)的情況下把服務(wù)暴露到公網(wǎng)默認(rèn)監(jiān)聽 127.0.0.1 就是為安全考慮的如果要多設(shè)備訪問應(yīng)該自己加 Token 或身份校驗(yàn)。第二情感分析結(jié)果只做參考它不能替代心理咨詢或醫(yī)療判斷如果發(fā)現(xiàn)記錄對(duì)象長期處于明顯的負(fù)面情緒應(yīng)該用真實(shí)的人際關(guān)懷去介入。第三如果你要把家人的情緒記錄、日期信息錄入系統(tǒng)最好征得對(duì)方同意這些信息屬于個(gè)人敏感數(shù)據(jù)。第四未來如果擴(kuò)展成自動(dòng)發(fā)送短信、微信消息、生成照片或語音祝福必須使用合規(guī)的推送通道并確保肖像權(quán)和聲音權(quán)已經(jīng)獲得授權(quán)。3. 環(huán)境準(zhǔn)備與項(xiàng)目結(jié)構(gòu)先準(zhǔn)備 Python 環(huán)境。推薦使用虛擬環(huán)境隔離依賴避免和系統(tǒng) Python 包沖突。Windows 用戶在命令提示符或 PowerShell 中執(zhí)行Linux / macOS 用戶使用終端。mkdir happiness_service cd happiness_service python -m venv venv激活虛擬環(huán)境# Windows venv\Scripts\activate # Linux / macOS source venv/bin/activate創(chuàng)建依賴文件 requirements.txt寫入以下內(nèi)容fastapi0.110 uvicorn0.29 apscheduler3.10 pydantic2.6然后安裝依賴pip install -r requirements.txt依賴安裝完成后創(chuàng)建項(xiàng)目目錄結(jié)構(gòu)。推薦按下面的方式組織文件后續(xù)擴(kuò)展功能時(shí)不容易亂happiness_service/ ├── app.py ├── config.py ├── database.py ├── analyzer.py ├── template_engine.py ├── scheduler.py ├── requirements.txt ├── data/ └── scripts/ └── send_blessing.py每個(gè)文件的職責(zé)如下config.py 保存路徑和端口配置database.py 負(fù)責(zé) SQLite 初始化和增刪改查analyzer.py 做輕量情感分析template_engine.py 負(fù)責(zé)祝福文案生成scheduler.py 是定時(shí)提醒任務(wù)app.py 是 FastAPI 入口scripts 目錄放獨(dú)立的批量腳本。先寫 config.py把路徑集中管理import os BASE_DIR os.path.dirname(os.path.abspath(__file__)) DATA_DIR os.path.join(BASE_DIR, data) DB_PATH os.path.join(DATA_DIR, happiness.db) HOST os.getenv(HAPPINESS_HOST, 127.0.0.1) PORT int(os.getenv(HAPPINESS_PORT, 8000))然后寫 database.py包含建表和基礎(chǔ)數(shù)據(jù)訪問方法import sqlite3 from pathlib import Path from config import DATA_DIR, DB_PATH def get_connection(): Path(DB_PATH).parent.mkdir(parentsTrue, exist_okTrue) conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row return conn def init_db(): conn get_connection() conn.executescript( CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, relationship TEXT DEFAULT default, birthday TEXT, anniversary_date TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS records ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, content TEXT NOT NULL, mood_type TEXT, mood_score REAL, keywords TEXT, remark TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(id) ); CREATE TABLE IF NOT EXISTS remind_tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, task_type TEXT NOT NULL, task_date TEXT NOT NULL, message TEXT, status TEXT DEFAULT pending, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); ) conn.commit() conn.close()數(shù)據(jù)表設(shè)計(jì)為三張users 保存人物信息records 保存每次情感記錄和分析結(jié)果remind_tasks 保存待觸發(fā)的提醒任務(wù)。birthday 和 anniversary_date 統(tǒng)一使用“YYYY-MM-DD”格式字符串方便定時(shí)任務(wù)按月日匹配。analyzer.py 先實(shí)現(xiàn)一個(gè)離線情感詞典版本不需要聯(lián)網(wǎng)也不需要 GPU。這個(gè)版本用來演示完整流程夠了后續(xù)想提高準(zhǔn)確率可以把 analyze_sentiment 函數(shù)內(nèi)部換成調(diào)用本地部署的情感分類模型。POSITIVE_WORDS {幸福, 開心, 快樂, 溫暖, 感動(dòng), 滿足, 喜歡, 順利, 健康, 值得} NEGATIVE_WORDS {難過, 失落, 疲憊, 焦慮, 生氣, 委屈, 壓力, 糟糕, 失眠, 擔(dān)心} def analyze_sentiment(text: str) - dict: positive_hits [w for w in POSITIVE_WORDS if w in text] negative_hits [w for w in NEGATIVE_WORDS if w in text] if not positive_hits and not negative_hits: return { mood_type: neutral, mood_score: 5.0, keywords: [], } total len(positive_hits) len(negative_hits) ratio len(positive_hits) / total if ratio 0.6: mood_type positive elif ratio 0.4: mood_type negative else: mood_type neutral return { mood_type: mood_type, mood_score: round(ratio * 10, 1), keywords: positive_hits negative_hits, }template_engine.py 使用模板生成祝福文案。這里先做離線模板版本避免把家庭場景的文案請(qǐng)求發(fā)送到外部大模型接口。按場景區(qū)分生日、紀(jì)念日、日常問候后續(xù)可以按人物關(guān)系繼續(xù)擴(kuò)展模板庫。TEMPLATES { birthday: [ {name}生日快樂。愿你想要的都擁有愿你珍惜的人一直陪在身邊。你的幸福就是我最大的幸福。, 今天是你的生日{(diào)name}。不用趕著成為誰今天只做開心的自己。, ], anniversary: [ {name}紀(jì)念日快樂。謝謝你在那些平常日子里給的陪伴愿以后每個(gè)重要日子我們都不缺席。, 屬于我們的紀(jì)念日又到了{(lán)name}。時(shí)間在走想給你的心意不變。, ], daily: [ {name}今天有沒有按時(shí)吃飯稍微停下來歇一會(huì)兒別讓自己太累。, 想跟你說聲{name}今天也辛苦了。你的笑容就是我最想守護(hù)的東西。, ], } RELATIONSHIP_NOTES { partner: 最珍惜的人, parent: 最想孝順的人, child: 最想守護(hù)的人, friend: 重要的朋友, default: 在乎的人, } def build_relationship_prefix(relationship: str) - str: return RELATIONSHIP_NOTES.get(relationship, RELATIONSHIP_NOTES[default]) def generate_blessing(user_name: str, scenario: str daily, relationship: str default) - str: templates TEMPLATES.get(scenario, TEMPLATES[daily]) template templates[0] return template.format(nameuser_name)到這一步項(xiàng)目骨架已經(jīng)完整。沒有 GPU 也能跑因?yàn)槟氵€沒有調(diào)用任何深度學(xué)習(xí)模型。下面進(jìn)入服務(wù)啟動(dòng)和接口驗(yàn)證。4. 啟動(dòng) API 服務(wù)與基礎(chǔ)驗(yàn)證現(xiàn)在寫 FastAPI 入口 app.py。這里直接調(diào)用 init_db 初始化數(shù)據(jù)庫減少啟動(dòng)時(shí)序問題。再定義用戶創(chuàng)建、情感記錄、情感趨勢(shì)和祝福生成等接口。from typing import Optional from fastapi import FastAPI from pydantic import BaseModel from analyzer import analyze_sentiment from config import init_app_config from database import ( add_record, add_user, get_mood_trend, init_db, list_records, list_users, ) from template_engine import generate_blessing app FastAPI(title你的幸福就是我最大的幸福, version0.1.0) init_db() class UserCreate(BaseModel): name: str relationship: str default birthday: Optional[str] None anniversary_date: Optional[str] None class RecordCreate(BaseModel): user_id: int content: str remark: Optional[str] None class BlessingRequest(BaseModel): name: str scenario: str daily relationship: str default app.get(/health) def health_check(): return {status: ok, service: happiness-service} app.post(/users) def create_user(payload: UserCreate): user_id add_user( namepayload.name, relationshippayload.relationship, birthdaypayload.birthday, anniversary_datepayload.anniversary_date, ) return {user_id: user_id, message: 用戶創(chuàng)建成功} app.get(/users) def query_users(): return list_users() app.post(/records) def create_record(payload: RecordCreate): mood_result analyze_sentiment(payload.content) record_id add_record( user_idpayload.user_id, contentpayload.content, mood_resultmood_result, remarkpayload.remark, ) return {record_id: record_id, analysis: mood_result} app.get(/users/{user_id}/records) def query_records(user_id: int): return list_records(user_iduser_id) app.get(/users/{user_id}/mood_trend) def query_mood_trend(user_id: int): return get_mood_trend(user_iduser_id) app.post(/blessing) def create_blessing(payload: BlessingRequest): return { scenario: payload.scenario, message: generate_blessing( user_namepayload.name, scenariopayload.scenario, relationshippayload.relationship, ), }注意這個(gè)版本沒有給接口加鑒權(quán)只適合本機(jī)測(cè)試。啟動(dòng)命令如下uvicorn app:app --host 127.0.0.1 --port 8000啟動(dòng)后終端會(huì)輸出訪問地址。瀏覽器打開 http://127.0.0.1:8000/docs如果能看到 FastAPI 自帶的 Swagger 交互文檔說明服務(wù)已經(jīng)跑起來了。Swagger 頁面會(huì)把所有接口列出并且支持直接在頁面上點(diǎn)擊 Try it out 發(fā)送請(qǐng)求這是最方便的接口驗(yàn)證方式。也可以先驗(yàn)證健康檢查接口curl http://127.0.0.1:8000/health返回結(jié)果{status:ok,service:happiness-service}看到這個(gè)輸出服務(wù)就是可用的。接下來開始錄入真實(shí)數(shù)據(jù)做功能測(cè)試。5. 功能測(cè)試情感記錄與祝福生成先創(chuàng)建一個(gè)用戶。比如要記錄媽媽的信息接口請(qǐng)求如下curl -X POST http://127.0.0.1:8000/users \ -H Content-Type: application/json \ -d {name: 媽媽, relationship: parent, birthday: 1965-08-20}預(yù)期返回{user_id:1,message:用戶創(chuàng)建成功}如果返回其他內(nèi)容先看 Swagger 頁面里的錯(cuò)誤提示通常是 JSON 格式錯(cuò)誤或者日期字段格式不對(duì)。創(chuàng)建用戶后添加一條情感記錄。這里設(shè)計(jì)接口會(huì)先調(diào)用 analyzer.py 的情感分析再把分析結(jié)果和原文一起存入數(shù)據(jù)庫。curl -X POST http://127.0.0.1:8000/records \ -H Content-Type: application/json \ -d {user_id: 1, content: 今天陪媽媽復(fù)查醫(yī)生說恢復(fù)得很好她特別開心。, remark: 復(fù)查日}預(yù)期返回{ record_id: 1, analysis: { mood_type: positive, mood_score: 10.0, keywords: [開心] } }這里 mood_type 有三種可能positive、neutral、negative。mood_score 范圍是 0 到 10分?jǐn)?shù)越高代表文本中包含的積極詞匯占比越高。keywords 是文本中命中的情感詞典詞匯。因?yàn)樵睦镉小伴_心”而沒有負(fù)面詞所以判定為 positive分?jǐn)?shù)是 10。再測(cè)試一條負(fù)面記錄curl -X POST http://127.0.0.1:8000/records \ -H Content-Type: application/json \ -d {user_id: 1, content: 她說最近總是失眠工作壓力很大。, remark: 日常記錄}返回結(jié)果里 mood_type 應(yīng)該是 negativemood_score 偏低keywords 至少包含“失眠”“壓力”。這說明情感分析接口能區(qū)分正向和負(fù)向表達(dá)。情感趨勢(shì)接口能把一段時(shí)間內(nèi)的記錄聚合成折線數(shù)據(jù)。請(qǐng)求curl http://127.0.0.1:8000/users/1/mood_trend返回按日期分組的平均 mood_score。這個(gè)接口適合后續(xù)做一個(gè)小型 Web 看板用來觀察記錄對(duì)象的情緒變化。祝福生成接口測(cè)試curl -X POST http://127.0.0.1:8000/blessing \ -H Content-Type: application/json \ -d {name: 媽媽, scenario: daily, relationship: parent}預(yù)期返回{ scenario: daily, message: 媽媽今天有沒有按時(shí)吃飯稍微停下來歇一會(huì)兒別讓自己太累。 }如果返回的是英文或者亂碼檢查終端編碼。在 Windows 的 PowerShell 里發(fā)送中文 JSON 可能會(huì)出現(xiàn)編碼問題更穩(wěn)妥的辦法是用 Python 腳本調(diào)用接口或者把 JSON 寫入文件再通過 curl 的 --data file 方式發(fā)送。6. 接口 API 調(diào)用示例接口跑通后可以把服務(wù)接到自己的腳本里。下面是 Python 調(diào)用示例適合在本地批量錄入數(shù)據(jù)和生成文案。import requests BASE http://127.0.0.1:8000 # 創(chuàng)建用戶 user requests.post( f{BASE}/users, json{ name: 媽媽, relationship: parent, birthday: 1965-08-20, anniversary_date: None, }, ).json() print(user) # 添加一條情感記錄 record requests.post( f{BASE}/records, json{ user_id: user[user_id], content: 今天陪媽媽復(fù)查醫(yī)生說恢復(fù)得很好她特別開心。, remark: 復(fù)查日, }, ).json() print(record) # 生成祝福文案 blessing requests.post( f{BASE}/blessing, json{ name: 媽媽, scenario: daily, relationship: parent, }, ).json() print(blessing[message])如果只需要生成文案不一定要先創(chuàng)建用戶。祝福接口只依賴 name、scenario、relationship 三個(gè)字段可以直接使用。批量生成祝福文案時(shí)建議把名單放到一個(gè) JSON 文件里例如 people.json[ {name: 爸爸, relationship: parent, scenario: birthday}, {name: 伴侶, relationship: partner, scenario: anniversary}, {name: 朋友, relationship: friend, scenario: daily} ]再寫一個(gè)批量腳本 scripts/send_blessing.pyimport argparse import json import requests BASE http://127.0.0.1:8000 def load_people(path): with open(path, encodingutf-8) as f: return json.load(f) def batch_generate(people): results [] for person in people: response requests.post( f{BASE}/blessing, json{ name: person[name], scenario: person.get(scenario, daily), relationship: person.get(relationship, default), }, timeout10, ) data response.json() results.append( { name: person[name], message: data[message], } ) return results if __name__ __main__: parser argparse.ArgumentParser(description批量生成祝福文案) parser.add_argument(--people, defaultpeople.json, help名單文件) parser.add_argument(--output, defaultblessing_output.json, help輸出文件) args parser.parse_args() people load_people(args.people) results batch_generate(people) with open(args.output, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) print(f已生成 {len(results)} 條文案輸出到 {args.output})運(yùn)行命令python scripts/send_blessing.py --people people.json --output blessing_output.json運(yùn)行后打開 blessing_output.json應(yīng)該能看到三條已經(jīng)填充好人名的祝福文案。這樣即使一次要寫幾十個(gè)人的祝福也只需要維護(hù)一份名單文件剩下的事情交給腳本。批量腳本有一個(gè)明顯的優(yōu)點(diǎn)它只依賴 /blessing 接口改動(dòng)模板時(shí)不需要重新學(xué)腳本。缺點(diǎn)是如果服務(wù)沒有啟動(dòng)所有請(qǐng)求都會(huì)失敗所以腳本里需要增加超時(shí)和異常處理。更完整的版本可以加日志記錄把失敗的用戶單獨(dú)輸出到一個(gè) retry.json 文件方便二次重跑。7. 定時(shí)提醒與批量任務(wù)擴(kuò)展只記錄數(shù)據(jù)和生成文案還不夠真正讓“你的幸福就是我最大的幸?！甭涞氐氖嵌〞r(shí)提醒。APScheduler 可以放在服務(wù)里每隔一段時(shí)間掃描一次數(shù)據(jù)庫把當(dāng)天需要提醒的人物找出來寫入隊(duì)列文件。scheduler.py 實(shí)現(xiàn)定時(shí)掃描邏輯import json import os from datetime import datetime from apscheduler.schedulers.background import BackgroundScheduler from config import DATA_DIR from database import get_connection QUEUE_FILE os.path.join(DATA_DIR, remind_queue.json) def scan_reminders(): today datetime.now().strftime(%m-%d) conn get_connection() rows conn.execute( SELECT u.id, u.name, u.relationship, u.birthday, u.anniversary_date FROM users u WHERE (substr(u.birthday, 6) ?) OR (substr(u.anniversary_date, 6) ?) , (today, today), ).fetchall() pending [] for row in rows: user dict(row) if user[birthday] and user[birthday][5:] today: task_type birthday elif user[anniversary_date] and user[anniversary_date][5:] today: task_type anniversary else: continue pending.append( { user_id: user[id], name: user[name], task_type: task_type, date: datetime.now().strftime(%Y-%m-%d), } ) conn.close() os.makedirs(DATA_DIR, exist_okTrue) with open(QUEUE_FILE, w, encodingutf-8) as f: json.dump(pending, f, ensure_asciiFalse, indent2) if pending: print(f[reminder] 檢測(cè)到 {len(pending)} 條應(yīng)提醒事項(xiàng)) else: print([reminder] 今天沒有匹配的日期) def start_scheduler(): scheduler BackgroundScheduler() scheduler.add_job(scan_reminders, interval, minutes1, idscan_reminders) scheduler.start() print([scheduler] 定時(shí)任務(wù)已啟動(dòng)每 1 分鐘掃描一次) return scheduler這個(gè)邏輯的核心是把日期字符串的“MM-DD”部分和今天的“MM-DD”比較。birthday 如果傳的是 “1965-08-20”substr(birthday, 6) 得到 “08-20”和今天的 “08-20” 匹配就說明今天是這個(gè)人的生日。定時(shí)任務(wù)不建議通過接口啟動(dòng)因?yàn)?uvicorn 開啟 --reload 時(shí)可能重復(fù)啟動(dòng)多個(gè)調(diào)度器。更穩(wěn)妥的做法是把定時(shí)任務(wù)獨(dú)立成一個(gè)進(jìn)程。在 scripts 下新建 run_scheduler.pyimport sys import threading from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from scheduler import start_scheduler if __name__ __main__: scheduler start_scheduler() print(按 CtrlC 退出) try: threading.Event().wait() except KeyboardInterrupt: scheduler.shutdown()運(yùn)行python scripts/run_scheduler.py默認(rèn)情況下掃描結(jié)果寫到 data/remind_queue.json。實(shí)際使用中你可以在 scan_reminders 函數(shù)末尾追加真實(shí)的推送通道比如發(fā)送郵件、調(diào)用企業(yè)微信群機(jī)器人、寫入自己的待辦應(yīng)用。推送通道需要單獨(dú)申請(qǐng)權(quán)限因此這套項(xiàng)目先統(tǒng)一走 JSON 文件方便你自己對(duì)接。批量任務(wù)可以分為兩種。一種是周期性批量生成文案用第 6 節(jié)的 send_blessing.py 就能實(shí)現(xiàn)。另一種是批量發(fā)送提醒需要你把發(fā)送狀態(tài)回寫到數(shù)據(jù)庫的 remind_tasks 表里至少增加一個(gè) processed 字段避免每次掃描都重復(fù)發(fā)送同一條消息。這是從示例走向生產(chǎn)環(huán)境必須補(bǔ)上的細(xì)節(jié)。8. 資源占用與性能觀察這個(gè)項(xiàng)目的基礎(chǔ)版本沒有任何深度學(xué)習(xí)模型資源占用很低。啟動(dòng) uvicorn 和 APScheduler 后主要消耗來自 Python 進(jìn)程本身。普通辦公電腦上內(nèi)存占用通常在幾十 MB 到一兩百 MB 區(qū)間具體數(shù)值會(huì)因 Python 版本和已安裝依賴不同而有差異。CPU 占用在沒有請(qǐng)求時(shí)接近 0每隔一分鐘的定時(shí)掃描任務(wù)也只需要執(zhí)行幾條 SQLite 查詢和少量字符串比較。顯存占用在這個(gè)項(xiàng)目里是可選項(xiàng)。如果你只使用情感詞典和模板生成整個(gè)過程完全不需要 GPU。后續(xù)如果需要更準(zhǔn)確的情感語義分析可以在 analyzer.py 中改為調(diào)用本地部署的開源情感分類模型或本地大模型這時(shí)才需要關(guān)注顯存。顯存占用取決于你接的模型參數(shù)量、輸入文本長度和推理框架沒有統(tǒng)一數(shù)字必須按本機(jī)實(shí)測(cè)為準(zhǔn)。觀察資源占用有三種方式。第一Windows 任務(wù)管理器里看 python 進(jìn)程的 CPU 和內(nèi)存。第二Linux 下使用 top 或 htop 查看。第三如果接入了本地 GPU 模型使用 nvidia-smi 觀察當(dāng)前進(jìn)程的顯存占用。nvidia-smi批量任務(wù)對(duì)性能的影響要分開看。批量生成祝福文案時(shí)瓶頸一般在模板引擎和 JSON 序列化數(shù)千條文案也是秒級(jí)完成。批量寫入情感記錄時(shí)瓶頸可能出現(xiàn)在 SQLite 寫入頻率上建議批量插入時(shí)使用事務(wù)而不是逐條提交。另外定時(shí)掃描任務(wù)如果間隔太短并且掃描邏輯里加了外部推送就要考慮推送通道的頻率限制避免被平臺(tái)當(dāng)成垃圾請(qǐng)求。內(nèi)存泄漏和端口殘留是本地服務(wù)最常見的兩個(gè)問題。開發(fā)調(diào)試時(shí)uvicorn 進(jìn)程沒有正常退出再次啟動(dòng)就會(huì)提示端口被占用。遇到這種情況先找到占用端口的進(jìn)程并終止再重新啟動(dòng)服務(wù)。# Windows netstat -aon | findstr :8000 taskkill /PID 進(jìn)程號(hào) /F # Linux / macOS lsof -i :8000 kill -9 進(jìn)程號(hào)9. 常見問題與排查方法本地部署這個(gè)流程不長但會(huì)遇到一些典型問題。整理成表格方便直接對(duì)照。問題現(xiàn)象可能原因排查方式解決方案啟動(dòng)時(shí)提示端口被占用8000 端口被其他進(jìn)程使用netstat 或 lsof 查看端口占用關(guān)閉占用進(jìn)程或修改 config.py 里的 PORTuvicorn 命令找不到?jīng)]有激活虛擬環(huán)境或依賴未安裝檢查命令行前是否出現(xiàn) venv 前綴激活虛擬環(huán)境后重新安裝 requirements.txt運(yùn)行時(shí)提示 sqlite3.OperationalError: database is locked多個(gè)進(jìn)程同時(shí)寫入數(shù)據(jù)庫檢查是否有多個(gè) API 進(jìn)程或調(diào)度器進(jìn)程關(guān)掉重復(fù)進(jìn)程使用 WAL 模式避免多進(jìn)程寫同一個(gè) SQLite 文件中文返回亂碼Windows 終端編碼不是 UTF-8在 Swagger 頁面看響應(yīng)體是否正常使用 Python 腳本調(diào)用或把 JSON 寫入文件再讀取情感分析結(jié)果全部為 neutral文本中未命中情感詞典詞匯打印 keywords 字段擴(kuò)充 POSITIVE_WORDS 和 NEGATIVE_WORDS或者接入更強(qiáng)模型定時(shí)任務(wù)一直沒有輸出APScheduler 未啟動(dòng)或用戶日期格式不對(duì)確認(rèn) run_scheduler.py 是否在運(yùn)行檢查 users 表中日期格式統(tǒng)一日期為 YYYY-MM-DD確認(rèn)掃描進(jìn)程已啟動(dòng)POST /records 返回 422JSON 字段名稱或類型不匹配查看響應(yīng)中的 detail 錯(cuò)誤信息按 Pydantic 模型傳字段user_id 必須為數(shù)字調(diào)用 /docs 無法打開服務(wù)未啟動(dòng)或?yàn)g覽器代理異常檢查后臺(tái)進(jìn)程和訪問地址確認(rèn) uvicorn 啟動(dòng)成功訪問 127.0.0.1 而不是遠(yuǎn)程地址重啟后數(shù)據(jù)丟失data 目錄未生成或數(shù)據(jù)庫路徑錯(cuò)誤檢查 data/happiness.db 是否存在確認(rèn) config.py 中 DATA_DIR 路徑正確如果接口報(bào) 422通常是請(qǐng)求 JSON 和 Pydantic 模型字段不一致。比如 RecordCreate 要求的是 user_id 和 content你卻傳了 userId 和 text校驗(yàn)直接失敗。先看 Swagger 文檔中的請(qǐng)求示例再對(duì)比自己的請(qǐng)求體。情感分析部分最容易讓使用者誤判。當(dāng)前默認(rèn)的情感詞典只有十幾個(gè)詞不可能覆蓋復(fù)雜的中文表達(dá)。比如“雖然很累但是看到他就踏實(shí)了”這句話可能同時(shí)命中“累”和“踏實(shí)”需要靠情緒分?jǐn)?shù)加權(quán)判斷。更準(zhǔn)確的做法是把 analyzer.py 里的邏輯替換成語義模型而不是繼續(xù)堆關(guān)鍵詞。10. 最佳實(shí)踐、合規(guī)建議與下一步第一次運(yùn)行時(shí)不要急著錄入大量數(shù)據(jù)。先創(chuàng)建 1 到 2 個(gè)測(cè)試用戶分別插入正向和負(fù)向情感記錄確認(rèn)情感分析、祝福生成、定時(shí)掃描三個(gè)主鏈路都正常再開始正式使用。數(shù)據(jù)備份要養(yǎng)成習(xí)慣。SQLite 只有一個(gè)文件直接復(fù)制 data/happiness.db 就能備份。本地服務(wù)沒有自動(dòng)備份機(jī)制建議手動(dòng)備份或者寫一個(gè)簡單的定時(shí)復(fù)制腳本。敏感數(shù)據(jù)不要放到公開倉庫尤其是包含真實(shí)姓名、出生日期、關(guān)系標(biāo)簽的數(shù)據(jù)庫文件。合規(guī)層面控制三點(diǎn)。第一只在獲得當(dāng)事人知情同意后記錄信息不要把工具變成暗中監(jiān)視他人情緒的手段。第二推送消息要控制頻率和措辭避免給對(duì)方造成打擾。如果對(duì)方明確表示不需要應(yīng)該允許關(guān)閉對(duì)該用戶的提醒。第三這個(gè)項(xiàng)目不適合用來做心理狀態(tài)診斷。情感分析只是輔助真實(shí)的關(guān)系維護(hù)仍然需要面對(duì)面溝通。下一步的擴(kuò)展方向比較明確。第一把模板引擎升級(jí)為支持多個(gè)模板隨機(jī)選擇避免每次都發(fā)一模一樣的文案。第二給 analyzer.py 接入本地情感分類模型提高語義判斷能力。第三把數(shù)據(jù)從 SQLite 換成 PostgreSQL增加用戶鑒權(quán)和多端訪問就能從個(gè)人工具升級(jí)成一個(gè)小型家庭服務(wù)。第四增加 Web 看板把情緒趨勢(shì)用圖表展示出來讓記錄變得可視化。第五設(shè)計(jì)發(fā)送確認(rèn)機(jī)制在 send_blessing.py 里加入“已發(fā)送”狀態(tài)防止同一批人重復(fù)收到祝福。如果你也想做一套“幸福守護(hù)”小服務(wù)建議從今天的項(xiàng)目結(jié)構(gòu)開始先跑通 FastAPI 接口再加入定時(shí)任務(wù)最后逐步增強(qiáng)分析和推送能力。代碼不復(fù)雜但至少能把你的關(guān)心從“心里惦記”變成一條準(zhǔn)時(shí)的消息。你的幸福就是我最大的幸福這句話放到代碼里就是一張人物表、一套情感分析規(guī)則、一批定時(shí)任務(wù)以及一條不遲到的祝福。