feat: agent_worker DeepSeek fallback & Sprint 003 spec
Deploy t-youtube / Build & Deploy (push) Successful in 1m48s

This commit is contained in:
Gan, Jimmy
2026-07-11 16:09:29 +08:00
parent 40a2666e55
commit 757c4b29c8
4 changed files with 117 additions and 7 deletions
+30 -6
View File
@@ -13,8 +13,12 @@ import db
log = logging.getLogger(__name__)
QWEN_BASE_URL = os.environ.get("QWEN_BASE_URL", "http://localhost:8080")
# LLM config: use local Qwen if QWEN_BASE_URL set, otherwise fall back to DeepSeek
QWEN_BASE_URL = os.environ.get("QWEN_BASE_URL", "")
QWEN_MODEL = os.environ.get("QWEN_MODEL", "current")
DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY", "")
DEEPSEEK_BASE_URL = "https://api.deepseek.com"
DEEPSEEK_MODEL = os.environ.get("DEEPSEEK_MODEL", "deepseek-chat")
POLL_INTERVAL = int(os.environ.get("AGENT_POLL_INTERVAL", "30"))
MAX_CONCURRENT = int(os.environ.get("AGENT_MAX_CONCURRENT", "3"))
SYSTEM_PROMPT = os.environ.get(
@@ -27,8 +31,8 @@ SYSTEM_PROMPT = os.environ.get(
semaphore = asyncio.Semaphore(MAX_CONCURRENT)
async def _call_qwen(prompt: str, video_title: str, video_context: str) -> str:
"""Call the local Qwen27B API with video context + user prompt."""
async def _call_llm(prompt: str, video_title: str, video_context: str) -> str:
"""Call local Qwen or DeepSeek API with video context + user prompt."""
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{
@@ -45,11 +49,27 @@ async def _call_qwen(prompt: str, video_title: str, video_context: str) -> str:
https_proxy = os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy")
proxies = http_proxy or https_proxy
if QWEN_BASE_URL:
# Use local Qwen
base_url = QWEN_BASE_URL
model = QWEN_MODEL
headers = {}
log.debug("Using local Qwen at %s", base_url)
elif DEEPSEEK_API_KEY:
# Fall back to DeepSeek cloud
base_url = DEEPSEEK_BASE_URL
model = DEEPSEEK_MODEL
headers = {"Authorization": f"Bearer {DEEPSEEK_API_KEY}"}
log.debug("Using DeepSeek cloud API")
else:
raise RuntimeError("No LLM configured: set QWEN_BASE_URL or DEEPSEEK_API_KEY")
async with httpx.AsyncClient(proxy=proxies, timeout=180) as client:
resp = await client.post(
f"{QWEN_BASE_URL}/v1/chat/completions",
f"{base_url}/v1/chat/completions",
headers=headers,
json={
"model": QWEN_MODEL,
"model": model,
"messages": messages,
"max_tokens": 2048,
"temperature": 0.3,
@@ -60,6 +80,10 @@ async def _call_qwen(prompt: str, video_title: str, video_context: str) -> str:
return data["choices"][0]["message"]["content"]
# Backward compat alias
_call_qwen = _call_llm
async def _process_item(item: dict) -> None:
"""Process a single agent_queue item."""
video_id = item["video_id"]
@@ -102,7 +126,7 @@ async def _process_item(item: dict) -> None:
video_context = "\n\n".join(context_parts) if context_parts else "No additional context available."
try:
result = await _call_qwen(user_prompt, video["title"], video_context)
result = await _call_llm(user_prompt, video["title"], video_context)
status = "done"
except Exception as e:
log.warning(f"Qwen API call failed for item {item['id']}: {e}")
+4
View File
@@ -31,6 +31,10 @@ async def test_agent_worker_polls_and_processes(test_db, monkeypatch):
item = dict(await cursor.fetchone())
# Mock Qwen API
monkeypatch.setenv("QWEN_BASE_URL", "http://mock-qwen:8080")
import agent_worker
agent_worker.QWEN_BASE_URL = "http://mock-qwen:8080"
mock_response = MagicMock(spec=httpx.Response)
mock_response.status_code = 200
mock_response.json.return_value = {
+1 -1
View File
@@ -1 +1 @@
sprint-002-ux-search-metadata.md
sprint-003-agent-digest.md
+82
View File
@@ -0,0 +1,82 @@
# Sprint 003 — Agent Worker + Daily Telegram Digest
**Depends on:** Sprint 002 (FTS, duration, view count, pagination)
**Duration:** ~8h total
**Goal:** Make the agent queue functional (LLM processing), and deliver a daily Telegram digest of new YouTube summaries.
---
## T3.1 — Implement agent_worker LLM processing
- **Files:** `backend/agent_worker.py`, `backend/main.py`
- **Done means:** `agent_worker.py` processes items from `agent_queue` with status=`pending`:
- Calls local LLM (via `QWEN_BASE_URL` env) or DeepSeek API to process the queued note/prompt.
- Stores result in `agent_queue.result` column.
- Updates status: `pending``processing``done` (or `error`).
- Runs as background loop every 30s.
- **Verify by:**
1. Add item to agent queue via frontend "Save to Agent".
2. Watch status change to `processing` then `done` in queue tab.
3. `result` field populated with LLM response.
- **Risk:** LLM endpoint may be unreachable from NAS. Use `AGENT_BASE_URL` env var with fallback to DeepSeek API.
- [ ] T3.1 — Agent worker LLM processing
## T3.2 — Daily Telegram digest via cron
- **Files:** New cron job in Hermes (not in repo — Hermes-managed)
- **Done means:** Every morning at 8:00 AM CST (00:00 UTC), Hermes:
1. Calls `GET /api/videos?status=pending&limit=20` on t-youtube backend.
2. Groups by channel.
3. Formats and sends a Telegram digest message: channel name, video title, summary snippet.
4. Marks videos as `read` after sending (optional, or just inform).
- **Verify by:**
1. Trigger cron manually → Telegram message received.
2. Message shows channel groupings, titles, and summaries.
- **Risk:** t-youtube API behind Authelia — cron needs API token, not SSO. Use `API_TOKEN` env var.
- [ ] T3.2 — Daily Telegram digest cron
## T3.3 — Duration + view count display in frontend
- **Files:** `frontend/src/routes/YoutubeDashboard.svelte`
- **Done means:** Video list items show duration (formatted as `MM:SS` or `H:MM:SS`) and view count (formatted as `1.2M`, `45K`) in the sidebar list item.
- **Verify by:**
1. Open dashboard → sidebar shows duration + view count for videos where populated.
2. Videos with `duration=null` show nothing (no broken display).
- **Risk:** Most existing videos don't have duration yet (only newly synced ones do). Display only if non-null.
- [ ] T3.3 — Duration + views display in video list
## T3.4 — Search UX polish: debounce + loading state
- **Files:** `frontend/src/routes/YoutubeDashboard.svelte`
- **Done means:**
- Search input debounced 300ms (no API call on every keystroke).
- Loading spinner shows while fetching search results.
- "No results" state when FTS returns empty.
- **Verify by:**
1. Type quickly → only one API call fires (after 300ms pause).
2. Loading state visible while waiting.
3. Type nonsense → "No results" shows cleanly.
- [ ] T3.4 — Search debounce + loading state
## T3.5 — Update active.md symlink to Sprint 003
- **File:** `docs/specs/active.md`
- **Done means:** Symlink points to `sprint-003-agent-digest.md`.
- **Verify by:** `ls -la docs/specs/active.md` → points to sprint-003.
- [ ] T3.5 — Update active.md symlink
---
## Exit criteria
- [ ] Agent queue processes items via LLM — status changes pending → done, result populated
- [ ] Daily Telegram digest cron fires at 8 AM CST with grouped summaries
- [ ] Duration + view count shown in video list (non-null only)
- [ ] Search debounced, has loading state and empty state
- [ ] All 23+ backend tests pass
- [ ] Frontend builds cleanly