From 757c4b29c8ea3138a0bc42a90fd3f141d96f9559 Mon Sep 17 00:00:00 2001 From: "Gan, Jimmy" Date: Sat, 11 Jul 2026 16:09:29 +0800 Subject: [PATCH] feat: agent_worker DeepSeek fallback & Sprint 003 spec --- backend/agent_worker.py | 36 ++++++++++-- backend/tests/test_agent_worker.py | 4 ++ docs/specs/active.md | 2 +- docs/specs/sprint-003-agent-digest.md | 82 +++++++++++++++++++++++++++ 4 files changed, 117 insertions(+), 7 deletions(-) create mode 100644 docs/specs/sprint-003-agent-digest.md diff --git a/backend/agent_worker.py b/backend/agent_worker.py index 97782d8..dd10071 100644 --- a/backend/agent_worker.py +++ b/backend/agent_worker.py @@ -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}") diff --git a/backend/tests/test_agent_worker.py b/backend/tests/test_agent_worker.py index 3de3186..fb28e3b 100644 --- a/backend/tests/test_agent_worker.py +++ b/backend/tests/test_agent_worker.py @@ -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 = { diff --git a/docs/specs/active.md b/docs/specs/active.md index 936cc56..c02bf07 120000 --- a/docs/specs/active.md +++ b/docs/specs/active.md @@ -1 +1 @@ -sprint-002-ux-search-metadata.md \ No newline at end of file +sprint-003-agent-digest.md \ No newline at end of file diff --git a/docs/specs/sprint-003-agent-digest.md b/docs/specs/sprint-003-agent-digest.md new file mode 100644 index 0000000..d4d7307 --- /dev/null +++ b/docs/specs/sprint-003-agent-digest.md @@ -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