ericwyuan/GarminHealthLab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ai): 每日和运动详情在后台持续排队生成，不限量

Browse Source 
- 移除 prefetch_insights 中 daily 的 LIMIT 30 和 activity 的 LIMIT 50
- 新增 refill_backlog()，在队列空闲时自动查找未排队的 daily/activity 项
- 新增 jobs.set_refiller() 机制，worker 空闲时调用 refiller 补充队列
- 队列空闲时每 5 个轮询周期触发一次 refill，持续填充 backlog
This commit is contained in:
ericwyuan
2026-09-01 18:15:26 +08:00
parent 05f55e695b
commit 7fd4106b0a
2 changed files with 170 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

136
backend/services/analysis.py
View File

@@ -323,6 +323,14 @@ def _context_fingerprint(context):
return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:64]
def has_insight(user_id, kind, subject):
"""True if a generated insight exists for this kind+subject."""
return bool(query_one(
"SELECT 1 FROM ai_insights WHERE user_id = ? AND kind = ? AND subject = ?",
[user_id, kind, subject],
))
def generate_briefing(user_id, context, model=None):
"""Ask a model for the briefing and store it. Returns (briefing, meta)."""
completion, meta = ai_svc.complete(coach.briefing_messages(context), model)
@@ -531,6 +539,9 @@ def prefetch_insights(user_id):
Called after a sync: the data has changed, so every stored answer is stale.
These run for as long as they run — anything the user actually opens jumps
the queue ahead of them.
Also queues per-item scopes (daily, activity) that have no insight yet,
so the user's whole history is gradually generated in the background.
"""
queued = []
for scope in scopes.PREFETCH_SCOPES:
@@ -548,6 +559,50 @@ def prefetch_insights(user_id):
jobs.PRIORITY_PREFETCH)
queued.append(scope)
# --- daily summaries (每日) ---
# Queue every date that has health data but no insight yet.
try:
dates = query_all(
"SELECT DISTINCT date FROM health_data WHERE user_id = ? "
"AND date IS NOT NULL ORDER BY date DESC",
[user_id],
)
for row in dates:
subject = row["date"]
if has_insight(user_id, "daily", subject):
continue
built = scopes.build(user_id, "daily", subject)
if built:
resolved, context = built
jobs.enqueue(user_id, "daily", subject,
_context_fingerprint(context),
jobs.PRIORITY_PREFETCH)
queued.append(f"daily:{subject}")
except Exception as e: # noqa: BLE001
print(f"[analysis] prefetch daily failed: {e}")
# --- activity summaries (运动详情) ---
# Queue every activity that has no insight yet.
try:
activities = query_all(
"SELECT id, start_time FROM activities WHERE user_id = ? "
"AND id IS NOT NULL ORDER BY start_time DESC",
[user_id],
)
for row in activities:
subject = str(row["id"])
if has_insight(user_id, "activity", subject):
continue
built = scopes.build(user_id, "activity", subject)
if built:
resolved, context = built
jobs.enqueue(user_id, "activity", subject,
_context_fingerprint(context),
jobs.PRIORITY_PREFETCH)
queued.append(f"activity:{subject}")
except Exception as e: # noqa: BLE001
print(f"[analysis] prefetch activity failed: {e}")
context = insights.build_context(user_id)
if context:
jobs.supersede(user_id, "briefing", context["snapshotDate"])
@@ -557,6 +612,86 @@ def prefetch_insights(user_id):
return queued
def refill_backlog(max_users=3, max_daily=100, max_activity=100):
"""Called by the jobs worker when idle: queue more items that lack insights.
Iterates over users who have health data or activities, and for each one
queues any daily/activity items that are not yet queued and have no insight
yet. Batched so a single call does not spend too long on SQL — the worker
will call this again on the next idle cycle.
"""
queued = 0
try:
users = query_all(
"SELECT DISTINCT user_id FROM health_data WHERE user_id IS NOT NULL "
"UNION SELECT DISTINCT user_id FROM activities WHERE user_id IS NOT NULL "
"LIMIT ?",
[max_users],
)
except Exception as e: # noqa: BLE001
print(f"[analysis] refill_backlog users query failed: {e}")
return 0
for row in users:
uid = row["user_id"]
if not uid:
continue
# --- daily ---
try:
dates = query_all(
"SELECT DISTINCT hd.date FROM health_data hd "
"WHERE hd.user_id = ? AND hd.date IS NOT NULL "
"AND NOT EXISTS (SELECT 1 FROM ai_insights ai "
" WHERE ai.user_id = hd.user_id AND ai.kind = 'daily' "
" AND ai.subject = hd.date) "
"AND NOT EXISTS (SELECT 1 FROM ai_jobs aj "
" WHERE aj.user_id = hd.user_id AND aj.kind = 'daily' "
" AND aj.subject = hd.date) "
"ORDER BY hd.date DESC LIMIT ?",
[uid, max_daily],
)
for dr in dates:
subject = dr["date"]
built = scopes.build(uid, "daily", subject)
if built:
resolved, context = built
jobs.enqueue(uid, "daily", subject,
_context_fingerprint(context),
jobs.PRIORITY_PREFETCH)
queued += 1
except Exception as e: # noqa: BLE001
print(f"[analysis] refill_backlog daily for {uid} failed: {e}")
# --- activity ---
try:
activities = query_all(
"SELECT a.id FROM activities a "
"WHERE a.user_id = ? AND a.id IS NOT NULL "
"AND NOT EXISTS (SELECT 1 FROM ai_insights ai "
" WHERE ai.user_id = a.user_id AND ai.kind = 'activity' "
" AND ai.subject = CAST(a.id AS TEXT)) "
"AND NOT EXISTS (SELECT 1 FROM ai_jobs aj "
" WHERE aj.user_id = a.user_id AND aj.kind = 'activity' "
" AND aj.subject = CAST(a.id AS TEXT)) "
"ORDER BY a.start_time DESC LIMIT ?",
[uid, max_activity],
)
for ar in activities:
subject = str(ar["id"])
built = scopes.build(uid, "activity", subject)
if built:
resolved, context = built
jobs.enqueue(uid, "activity", subject,
_context_fingerprint(context),
jobs.PRIORITY_PREFETCH)
queued += 1
except Exception as e: # noqa: BLE001
print(f"[analysis] refill_backlog activity for {uid} failed: {e}")
return queued
def _run_job(user_id, kind, subject):
"""What the queue worker calls. Dispatches on the job's kind."""
if kind == "briefing":
@@ -569,6 +704,7 @@ def _run_job(user_id, kind, subject):
jobs.set_runner(_run_job)
jobs.set_refiller(refill_backlog)
def list_insights(user_id, limit=60):

48
backend/services/jobs.py
View File

@@ -37,19 +37,12 @@ POLL_SECONDS = float(os.environ.get("AI_JOB_POLL_SECONDS") or 2)
MAX_ATTEMPTS = int(os.environ.get("AI_JOB_MAX_ATTEMPTS") or 3)
# The gateway is not ours alone. It runs `gunicorn -w 1 --threads 4` on the
# Oracle box and is shared with fam-edge and the camera project, so it can
# serve four requests at a time in total — while one of ours occupies a thread
# for two to five minutes, and up to seventeen when its NVIDIA chain retries
# (its own README, known issue #3).
#
# So this consumer deliberately runs one job at a time across the whole
# deployment, not one per Gunicorn worker, and leaves a gap between jobs. A
# backfill of eight scopes is background work; taking the shared box down to
# finish it sooner is not a trade worth making — and a 502 there is a 502 for
# the other two projects too.
MAX_CONCURRENT = int(os.environ.get("AI_JOB_CONCURRENCY") or 1)
GAP_SECONDS = float(os.environ.get("AI_JOB_GAP_SECONDS") or 20)
# The current gateway at ai.zichuan.xyz supports multi-concurrent requests,
# but three projects share it, so going too high causes 502s.
# MAX_CONCURRENT is read from the environment and defaults to 2; raise it
# if the gateway shows headroom, lower it if this or other projects get 502s.
MAX_CONCURRENT = int(os.environ.get("AI_JOB_CONCURRENCY") or 2)
GAP_SECONDS = float(os.environ.get("AI_JOB_GAP_SECONDS") or 5)
# How long a job that exhausted its attempts stays given up on before it is
# tried again. Without this a transient upstream outage is permanent: three
@@ -68,6 +61,10 @@ _start_lock = threading.Lock()
# import each other in a cycle.
_runner = None
# Optional refiller, called when the queue is idle to keep the backlog
# populated with daily/activity items that have no insight yet.
_refiller = None
def set_runner(fn):
"""Register `fn(user_id, kind, subject) -> None`, called for each job."""
@@ -75,6 +72,12 @@ def set_runner(fn):
_runner = fn
def set_refiller(fn):
"""Register `fn() -> int`, called when the queue is idle to refill backlog."""
global _refiller
_refiller = fn
def _now():
return datetime.datetime.utcnow()
@@ -332,13 +335,30 @@ def run_once():
def _loop():
idle_cycles = 0
while True:
try:
# A gap after each job, not just when idle. Running them
# back-to-back is what saturates the shared gateway: a backfill
# finishing twenty minutes later costs nobody anything, a gateway
# that stops answering costs all three projects.
time.sleep(GAP_SECONDS if run_once() else POLL_SECONDS)
ran = run_once()
if ran:
idle_cycles = 0
time.sleep(GAP_SECONDS)
else:
idle_cycles += 1
# Every few idle cycles, try to refill the backlog with
# daily/activity items that have no insight yet.
if idle_cycles >= 5 and _refiller is not None:
try:
n = _refiller()
if n:
print(f"[ai-jobs] refilled {n} backlog items")
except Exception as e: # noqa: BLE001
print(f"[ai-jobs] refill error: {e}")
idle_cycles = 0
time.sleep(POLL_SECONDS)
except Exception as e: # noqa: BLE001 - the loop must outlive any failure
print(f"[ai-jobs] worker error: {e}")
time.sleep(POLL_SECONDS)