feat: AI Self-Loop 專案骨架 — 爬站健檢版

完整實作建構順序 1-8(rule-based):
- Task Queue(SQLite):tasks/attempts/results + claim 防重複
- LLM 後端:OllamaBackend(num_ctx=32768) + LLMRouter
- Prompt Builder(YAML+Jinja2,重試注入 feedback)
- Context Assembler / Output Judge / Feedback Builder
- Loop Controller + Circuit Breaker
- Driver 層(BaseDriver + ApiClientDriver)
- Agent 層(BaseAgent + SiteHealthAgent:爬站/健檢/issue 生成)
- pipeline.py 入口,驗收:pending→done,output/ 落地

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Yellow
2026-06-14 17:35:54 +08:00
commit 27f98712ee
31 changed files with 1480 additions and 0 deletions

15
.gitignore vendored Normal file
View File

@@ -0,0 +1,15 @@
__pycache__/
*.pyc
.venv/
venv/
# Runtime artifacts
*.db
*.sqlite3
output/*
!output/.gitkeep
review_queue/*
!review_queue/.gitkeep
# Real credentials never get committed; only the .example file does
config/credentials.yaml

66
README.md Normal file
View File

@@ -0,0 +1,66 @@
# AI Self-Loop 骨架
能自我驅動的 AI 處理系統:接收輸入 → 呼叫 LLM 處理 → 評估品質 → 不達標自動修正再跑,
直到達標(`done`)或送人工審核(`review`)。
起步領域:以 `https://www.pluspay.com.tw/` 為起點,自動爬出站內地圖並做逐頁健檢。
## 快速開始
```bash
pip install -r requirements.txt
# 需要本機 Ollama 在 http://localhost:11434且有 environments.yaml 指定的模型
python -m src.pipeline # 預設seed 一筆任務 + 處理(驗收用)
python -m src.pipeline --seed # 只放任務進佇列
python -m src.pipeline --run # 只處理 pending 任務
```
產出:
- `output/sitemap.json` —— 站內頁面地圖
- `output/health_report.json` —— 每頁健檢報告
- `review_queue/task_<id>.json` —— 送審任務(若有)
## 架構:兩個分層
| 層 | 介面 | 職責 | 起步實作 |
|----|------|------|----------|
| **Driver**(怎麼做) | `BaseDriver.connect/execute` | 連線與操作目標,無領域判斷 | `drivers/api_client.py`HTTP |
| **Agent**(做什麼) | `BaseAgent.run` | 領域 AI 判斷,呼叫 Driver + Loop 引擎 | `agents/site_health.py` |
要加 web(Playwright)/mobile driver 時agent 不用動。
## Loop 引擎(五元件,`src/core/`
1. **Context Assembler** — 決定哪些資訊送進 LLM寧可少不要雜
2. **Prompt Builder** — 從 `prompts/*.yaml` 用 Jinja2 渲染;重試時注入上一輪 feedback
3. **Output Judge** — 統一輸出 `passed/score/issues/suggestions`(先 rule-based
4. **Feedback Builder** — 把 Judge 結果轉成下一輪的具體修正指令
5. **Loop Controller** — 重試上限(預設 3+ 每次嘗試完整記錄 + Circuit Breaker
## LLM 後端(`src/llm/`
- `LLMBackend``call(prompt, system)` / `is_available()`
- `OllamaBackend`**每次 call 強制帶 `num_ctx`(預設 32768**,不吃 Ollama 預設的 4K
- `LLMRouter`:多 backend 依序用第一個可用的
## Task Queue`src/db/`SQLite
`tasks`(狀態)/ `attempts`(每次嘗試的完整記錄)/ `results`(定版輸出)。
取任務用原子 compare-and-swap 防重複。
## 設定外部化(`config/`
- `environments.yaml` —— 各環境 URL/端點/Loop 參數(明文,先不加密)
- `credentials.example.yaml` —— 憑證佔位;複製成 `credentials.yaml` 填真值
(已 gitignore。讀取優先序環境變數 > credentials.yaml > 預設。
## 驗收
放一筆任務 → 跑 `pipeline.py` → 任務 `pending → done/review``output/` 出現結果。
已通過3 頁限縮實跑得到 `done`sitemap 與 health_report 落地。
## 尚未做(刻意保留為下一步)
- **LLM-as-Judge**build order #9):目前 Judge 是 rule-based可加一層讓 LLM
評估 issue 的「對人類審核者是否真的有用」。掛點在 `core/judge.py`

View File

@@ -0,0 +1,20 @@
# =============================================================================
# 憑證佔位檔(範本)。
#
# 使用方式:
# 1. 複製成 config/credentials.yaml已被 .gitignore 忽略,不會進版控)
# 2. 填入真實值,或留空改用環境變數
#
# 讀取優先序(見 src/config.py
# 環境變數 > credentials.yaml > 這裡的預設
#
# TODO(security): 日後要加密時,這個檔就是掛載解密層的位置——
# 例如改讀 SOPS / age / Vault介面不變。
# =============================================================================
credentials:
# 範例:未來若目標站需要登入或 API key 才能測
target_basic_auth:
username: "${TARGET_USER:-}"
password: "${TARGET_PASS:-}"
# 範例:若改用雲端 LLM
ollama_api_key: "${OLLAMA_API_KEY:-}"

70
config/environments.yaml Normal file
View File

@@ -0,0 +1,70 @@
# =============================================================================
# 環境設定(明文,先不加密)
# 各環境的 URL / 端點都集中在這裡,程式碼不寫死任何網址。
# 切換環境只要改 active 或設環境變數 APP_ENV。
# =============================================================================
active: ${APP_ENV:-dev} # 預設 dev可被環境變數 APP_ENV 覆蓋
environments:
dev:
# --- 受測目標 ---
target:
start_url: "https://www.pluspay.com.tw/"
# 只爬這個主網域底下的頁面;其餘視為「外部」只檢查可達性
internal_domain: "pluspay.com.tw"
# 記錄但不深入的外部 / 子網域(外連只檢查連得到嗎)
do_not_crawl:
- "104.com.tw"
- "facebook.com"
- "event2023.pluspay.com.tw"
- "member.pluspay.com.tw"
max_pages: 200 # 爬蟲安全上限,避免失控
min_pages_expected: 5 # 驗收sitemap 至少要有這麼多頁,低於視為漏抓
request_timeout_sec: 10
crawl_delay_sec: 0.5 # 禮貌性延遲,別把對方站打掛
# --- LLM 後端(依序嘗試,用第一個可用的)---
llm:
num_ctx: 32768 # 強制覆蓋 Ollama 預設 4K
backends:
- type: ollama
base_url: "http://localhost:11434"
model: "gemma4:e4b"
- type: ollama
base_url: "http://localhost:11434"
model: "minimax-m2.7:cloud"
# --- Loop 引擎參數 ---
loop:
max_attempts: 3
circuit_breaker:
failure_threshold: 5 # 視窗內失敗幾次就跳閘
window_sec: 60 # 觀察視窗
cooldown_sec: 30 # 跳閘後暫停多久
# 之後要加正式環境,照 dev 的結構複製一份即可
prod:
target:
start_url: "https://www.pluspay.com.tw/"
internal_domain: "pluspay.com.tw"
do_not_crawl:
- "104.com.tw"
- "facebook.com"
- "event2023.pluspay.com.tw"
- "member.pluspay.com.tw"
max_pages: 1000
min_pages_expected: 10
request_timeout_sec: 15
crawl_delay_sec: 1.0
llm:
num_ctx: 32768
backends:
- type: ollama
base_url: "http://localhost:11434"
model: "gemma4:e4b"
loop:
max_attempts: 3
circuit_breaker:
failure_threshold: 5
window_sec: 60
cooldown_sec: 30

0
output/.gitkeep Normal file
View File

View File

@@ -0,0 +1,30 @@
# =============================================================================
# 模板:把單一頁面的「原始事實」轉成具體、可逐條檢查的 issue 描述。
# 用 Jinja2 渲染。重試時 feedback 區塊會帶入上一輪失敗原因。
#
# 可用變數:
# url 頁面網址
# facts dictbroken_links / image_problems / redirects / ...
# feedback 上一輪的修正指令(第一次為空)
# =============================================================================
system: |
你是一個網站健檢報告撰寫員。你的任務是把機器掃描出的原始事實,
改寫成「具體到哪個連結、什麼問題」的繁體中文 issue 描述。
規則:
- 每條 issue 必須點名是哪個 URL / 哪張圖 / 哪個表單。
- 嚴禁含糊字眼(例如「有問題」「部分連結異常」)。
- 只描述 facts 裡確實存在的問題,不要臆測。
- 只輸出 JSON格式{"issues": ["...", "..."]}。沒有問題就回 {"issues": []}。
user: |
頁面:{{ url }}
掃描到的原始事實JSON
{{ facts }}
{% if feedback %}
── 上一輪的輸出未達標,請依下列修正後重寫 ──
{{ feedback }}
{% endif %}
請輸出 issues 的 JSON。

6
requirements.txt Normal file
View File

@@ -0,0 +1,6 @@
# --- Loop engine / infra ---
PyYAML>=6.0 # environments.yaml + prompt templates
Jinja2>=3.1 # prompt rendering
requests>=2.31 # api_client driver (HTTP)
beautifulsoup4>=4.12 # HTML parsing for the crawler agent
# stdlib used elsewhere: sqlite3, dataclasses, urllib, json, logging

0
review_queue/.gitkeep Normal file
View File

View File

@@ -0,0 +1,4 @@
{
"_comment": "site_health 任務的輸入範例。實際 start_url 以 config/environments.yaml 為準;這裡只是說明 payload 形狀。",
"start_url": "https://www.pluspay.com.tw/"
}

0
src/__init__.py Normal file
View File

0
src/agents/__init__.py Normal file
View File

19
src/agents/base.py Normal file
View File

@@ -0,0 +1,19 @@
"""Agent 層抽象介面:封裝特定領域的 AI 判斷邏輯(測什麼、怎麼判斷)。
Agent 呼叫 Driver 做操作、呼叫 Loop 引擎處理 AI 判斷。
起步只做一個領域site_health
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any
from ..core.loop_controller import LoopOutcome
class BaseAgent(ABC):
name: str = "base"
@abstractmethod
def run(self, payload: dict, on_attempt: Any = None) -> LoopOutcome:
"""執行一筆任務,回傳 LoopOutcomefinal_state + output"""

249
src/agents/site_health.py Normal file
View File

@@ -0,0 +1,249 @@
"""SiteHealthAgent爬站建地圖 + 逐頁健檢,並用 Loop 引擎把原始事實
轉成具體 issue 描述。
兩階段輸出:
1) sitemap.json —— 站內頁面地圖
2) health_report.json —— 每頁健檢報告list
只爬 internal_domain 主網域;外部 / 子網域只檢查可達性,不深入。
"""
from __future__ import annotations
import json
import re
import time
from collections import deque
from urllib.parse import urldefrag, urljoin, urlparse
from bs4 import BeautifulSoup
from ..core.context_assembler import ContextAssembler
from ..core.judge import OutputJudge, judge_page_status
from ..core.loop_controller import LoopController, LoopOutcome
from ..core.prompt_builder import PromptBuilder
from ..llm.backend import LLMBackend
from ..drivers.base import BaseDriver
from .base import BaseAgent
_SKIP_SCHEMES = ("mailto:", "tel:", "javascript:", "#")
def _parse_issues_json(raw: str) -> list[str]:
"""從 LLM 輸出抽出 {"issues": [...]};容忍 code fence / 前後雜訊。"""
if not raw:
return []
m = re.search(r"\{.*\}", raw, re.DOTALL)
if not m:
return []
try:
data = json.loads(m.group(0))
issues = data.get("issues", [])
return [str(x) for x in issues] if isinstance(issues, list) else []
except json.JSONDecodeError:
return []
class SiteHealthAgent(BaseAgent):
name = "site_health"
def __init__(
self,
driver: BaseDriver,
llm: LLMBackend,
loop: LoopController,
target_cfg: dict,
):
self.driver = driver
self.llm = llm
self.loop = loop
self.cfg = target_cfg
self.assembler = ContextAssembler()
self.prompt_builder = PromptBuilder()
self.judge = OutputJudge()
self._liveness_cache: dict[str, dict] = {} # 連結可達性快取,跨頁共用
# ── URL 工具 ──────────────────────────────────────────────────────────
def _norm(self, url: str) -> str:
return urldefrag(url)[0].rstrip("/") or url
def _host(self, url: str) -> str:
return (urlparse(url).hostname or "").lower()
def _is_crawlable(self, url: str) -> bool:
host = self._host(url)
dom = self.cfg["internal_domain"]
return host in {dom, f"www.{dom}"} and url.startswith("http")
def _is_internal(self, url: str) -> bool:
host = self._host(url)
return host.endswith(self.cfg["internal_domain"])
def _classify(self, url: str, title: str) -> str:
text = (url + " " + title).lower()
rules = [
("使用說明", ["help", "faq", "guide", "說明", "教學", "問答"]),
("公告", ["news", "announce", "notice", "公告", "消息"]),
("活動", ["event", "promo", "campaign", "活動", "優惠"]),
("關於", ["about", "company", "關於", "簡介"]),
("聯絡", ["contact", "聯絡", "客服"]),
]
for label, kws in rules:
if any(k in text for k in kws):
return label
return "其他"
# ── 階段一:爬站建地圖 ─────────────────────────────────────────────────
def crawl(self) -> dict:
start = self._norm(self.cfg["start_url"])
max_pages = int(self.cfg.get("max_pages", 200))
delay = float(self.cfg.get("crawl_delay_sec", 0.5))
seen: set[str] = set()
queue: deque[str] = deque([start])
pages: list[dict] = []
while queue and len(pages) < max_pages:
url = queue.popleft()
if url in seen:
continue
seen.add(url)
res = self.driver.execute("get", url=url)
if not res["ok"] or not res["text"]:
# 連不上的頁面仍記一筆,健檢階段會標 fail
pages.append(self._empty_page(url, res))
continue
soup = BeautifulSoup(res["text"], "html.parser")
title = (soup.title.string or "").strip() if soup.title else ""
internal_links, external_links = [], []
for a in soup.find_all("a", href=True):
href = a["href"].strip()
if not href or href.startswith(_SKIP_SCHEMES):
continue
absolute = self._norm(urljoin(url, href))
if self._is_crawlable(absolute):
internal_links.append(absolute)
if absolute not in seen:
queue.append(absolute)
else:
external_links.append(absolute)
images = [self._norm(urljoin(url, img["src"]))
for img in soup.find_all("img", src=True)]
forms = [self._norm(urljoin(url, f.get("action") or url))
for f in soup.find_all("form")]
pages.append({
"url": url,
"title": title,
"category": self._classify(url, title),
"internal_links": sorted(set(internal_links)),
"external_links": sorted(set(external_links)),
"images": sorted(set(images)),
"forms": sorted(set(forms)),
})
time.sleep(delay)
return {"pages": pages}
@staticmethod
def _empty_page(url: str, res: dict) -> dict:
return {
"url": url, "title": "", "category": "其他",
"internal_links": [], "external_links": [], "images": [],
"forms": [], "_fetch": res,
}
# ── 連結 / 圖片可達性HEAD含快取─────────────────────────────────
def _check_liveness(self, url: str) -> dict:
if url in self._liveness_cache:
return self._liveness_cache[url]
res = self.driver.execute("head", url=url)
# 有些站擋 HEAD退回用 GET 確認
if res.get("status") in (403, 405) or (not res["ok"] and res["status"] is None):
res = self.driver.execute("get", url=url)
self._liveness_cache[url] = res
return res
# ── 階段二:單頁健檢(確定性,不含 AI───────────────────────────────
def _health_check_page(self, page: dict) -> dict:
url = page["url"]
fetch = page.get("_fetch") or self.driver.execute("get", url=url)
page_loads = bool(fetch.get("ok"))
images_ok = [
{"url": img, "ok": self._check_liveness(img)["ok"]}
for img in page.get("images", [])
]
broken_links = []
for link in page.get("internal_links", []) + page.get("external_links", []):
r = self._check_liveness(link)
if not r["ok"]:
reason = (f"HTTP {r['status']}" if r.get("status")
else (r.get("error") or "unreachable"))
broken_links.append({"url": link, "reason": reason})
redirects = []
for link in page.get("internal_links", []):
r = self._liveness_cache.get(link)
if r and r.get("redirected"):
redirects.append({
"url": link, "final_url": r["final_url"],
"ok": r["ok"], "timeout": r.get("timeout", False),
})
forms_reachable = [
{"action": action, "reachable": self._check_liveness(action)["ok"]}
for action in page.get("forms", [])
]
checks = {
"page_loads": page_loads,
"title_exists": bool(page.get("title")),
"images_ok": images_ok,
"broken_links": broken_links,
"redirects": redirects,
"forms_reachable": forms_reachable,
}
return {"url": url, "status": judge_page_status(checks), "checks": checks,
"issues": []}
# ── Loop 串接產出整份報告AI 改寫 issue────────────────────────────
def _produce_report(self, sitemap: dict, base_reports: list[dict],
feedback: str) -> dict:
reports = []
for base in base_reports:
facts = self.assembler.for_health_issues(base)
issues: list[str] = []
if facts: # 只有真的有問題才花 LLM 改寫
prompt = self.prompt_builder.build(
"health_issues", url=base["url"], facts=facts,
feedback=feedback,
)
raw = self.llm.call(prompt.user, prompt.system)
issues = _parse_issues_json(raw)
reports.append({**base, "issues": issues})
return {"sitemap": sitemap, "health_reports": reports}
# ── 對外入口 ──────────────────────────────────────────────────────────
def run(self, payload: dict, on_attempt=None) -> LoopOutcome:
self.driver.connect()
try:
sitemap = self.crawl()
base_reports = [self._health_check_page(p) for p in sitemap["pages"]]
min_pages = int(self.cfg.get("min_pages_expected", 1))
def produce(feedback: str) -> dict:
return self._produce_report(sitemap, base_reports, feedback)
def judge(report: dict):
return self.judge.judge_report(
report["sitemap"], report["health_reports"], min_pages
)
return self.loop.run(produce, judge, on_attempt=on_attempt)
finally:
self.driver.close()

80
src/config.py Normal file
View File

@@ -0,0 +1,80 @@
"""設定載入:把 environments.yaml + credentials 讀進來,並解析 ${VAR:-default} 佔位。
設計重點:
- 程式碼不寫死任何 URL / 端點,一律從 config 取。
- ${VAR} / ${VAR:-default} 佔位讓「憑證外部化」:真實值放環境變數,
檔案裡只留佔位,方便日後換成加密來源而不動程式。
"""
from __future__ import annotations
import os
import re
from pathlib import Path
from typing import Any
import yaml
# 專案根目錄src/ 的上一層)
ROOT = Path(__file__).resolve().parent.parent
CONFIG_DIR = ROOT / "config"
# 比對 ${NAME} 或 ${NAME:-fallback}
_VAR_PATTERN = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)(?::-(.*?))?\}")
def _expand(value: Any) -> Any:
"""遞迴展開字串裡的環境變數佔位。"""
if isinstance(value, str):
def repl(m: re.Match) -> str:
name, default = m.group(1), m.group(2)
return os.environ.get(name, default if default is not None else "")
return _VAR_PATTERN.sub(repl, value)
if isinstance(value, dict):
return {k: _expand(v) for k, v in value.items()}
if isinstance(value, list):
return [_expand(v) for v in value]
return value
def _load_yaml(path: Path) -> dict:
if not path.exists():
return {}
with path.open("r", encoding="utf-8") as f:
return _expand(yaml.safe_load(f) or {})
class Config:
"""單一存取點config.env(...) 取出當前環境的設定區塊。"""
def __init__(self, env_name: str | None = None):
raw = _load_yaml(CONFIG_DIR / "environments.yaml")
self._active = env_name or raw.get("active") or "dev"
self._environments = raw.get("environments", {})
# credentials.yaml 是選用的(被 gitignore不存在就空表
self._credentials = _load_yaml(CONFIG_DIR / "credentials.yaml").get(
"credentials", {}
)
@property
def active(self) -> str:
return self._active
def env(self) -> dict:
if self._active not in self._environments:
raise KeyError(
f"環境 '{self._active}' 不存在於 environments.yaml"
f"可用:{list(self._environments)}"
)
return self._environments[self._active]
def credential(self, *keys: str, default: Any = None) -> Any:
node: Any = self._credentials
for k in keys:
if not isinstance(node, dict) or k not in node:
return default
node = node[k]
return node
def load_config(env_name: str | None = None) -> Config:
return Config(env_name)

0
src/core/__init__.py Normal file
View File

View File

@@ -0,0 +1,57 @@
"""Circuit Breaker短時間內失敗太多次就「跳閘」暫停避免對壞掉的下游持續猛打。
狀態機:
CLOSED正常──視窗內失敗達門檻──▶ OPEN跳閘快速拒絕
OPEN ──冷卻時間到──▶ CLOSED恢復
這是跨任務的全域保護,與「單一任務重試上限」是兩回事。
"""
from __future__ import annotations
import time
from collections import deque
class CircuitBreaker:
def __init__(
self,
failure_threshold: int = 5,
window_sec: float = 60.0,
cooldown_sec: float = 30.0,
):
self.failure_threshold = failure_threshold
self.window_sec = window_sec
self.cooldown_sec = cooldown_sec
self._failures: deque[float] = deque() # 失敗發生的時間戳
self._opened_at: float | None = None
def _now(self) -> float:
return time.monotonic()
def _prune(self, now: float) -> None:
# 丟掉滑出視窗的舊失敗
while self._failures and now - self._failures[0] > self.window_sec:
self._failures.popleft()
def record_failure(self) -> None:
now = self._now()
self._prune(now)
self._failures.append(now)
if len(self._failures) >= self.failure_threshold:
self._opened_at = now
def record_success(self) -> None:
# 成功就清空失敗計數,並關閘
self._failures.clear()
self._opened_at = None
def is_open(self) -> bool:
"""True = 跳閘中,呼叫端應暫停 / 直接送人工審核。"""
if self._opened_at is None:
return False
if self._now() - self._opened_at >= self.cooldown_sec:
# 冷卻結束,半開→直接恢復(簡化版)
self._opened_at = None
self._failures.clear()
return False
return True

View File

@@ -0,0 +1,44 @@
"""Context Assembler決定「哪些資訊」要送進這次 LLM call。
原則:寧可少,不要雜。
對健檢任務而言 —— 只把「有問題的事實」交給 LLM 改寫成 issue
把雜訊(成功載入的圖片、所有站內連結清單)丟掉,省 context 也減少幻覺。
"""
from __future__ import annotations
from typing import Any
class ContextAssembler:
def for_health_issues(self, page_health: dict) -> dict:
"""從單頁的完整健檢結果中,萃取出只跟『問題』有關的事實。"""
checks = page_health.get("checks", {})
facts: dict[str, Any] = {}
broken = checks.get("broken_links", [])
if broken:
facts["broken_links"] = broken
# 只留載不出來的圖OK 的不送
bad_images = [
img for img in checks.get("images_ok", [])
if isinstance(img, dict) and not img.get("ok", True)
]
if bad_images:
facts["image_problems"] = bad_images
# 跳轉只有在「非預期」時才算問題事實
bad_redirects = [
r for r in checks.get("redirects", [])
if isinstance(r, dict) and not r.get("ok", True)
]
if bad_redirects:
facts["redirect_problems"] = bad_redirects
if not checks.get("title_exists", True):
facts["missing_title"] = True
if not checks.get("page_loads", True):
facts["page_unreachable"] = True
return facts

24
src/core/contracts.py Normal file
View File

@@ -0,0 +1,24 @@
"""跨模組共用的資料契約。
JudgeResult 是 Judge 的統一輸出格式spec 要求):
passed / score / issues / suggestions
"""
from __future__ import annotations
from dataclasses import dataclass, field
@dataclass
class JudgeResult:
passed: bool
score: float # 0.0 ~ 1.0
issues: list[str] = field(default_factory=list)
suggestions: list[str] = field(default_factory=list)
def to_dict(self) -> dict:
return {
"passed": self.passed,
"score": self.score,
"issues": self.issues,
"suggestions": self.suggestions,
}

View File

@@ -0,0 +1,24 @@
"""Feedback Builder把 Judge 的輸出轉成下一輪的修正指令。
要具體 —— 直接把不達標的點逐條列出,讓下一輪的 prompt 知道要修什麼。
"""
from __future__ import annotations
from .contracts import JudgeResult
class FeedbackBuilder:
def build(self, judge: JudgeResult) -> str:
if judge.passed:
return ""
lines = ["上一輪輸出未達標,請針對以下問題逐條修正:"]
for i, issue in enumerate(judge.issues, 1):
lines.append(f"{i}. {issue}")
if judge.suggestions:
lines.append("")
lines.append("修正方向:")
for s in judge.suggestions:
lines.append(f"- {s}")
lines.append("")
lines.append("請重新產出,務必讓每條描述都具體到可被人工逐項核對。")
return "\n".join(lines)

118
src/core/judge.py Normal file
View File

@@ -0,0 +1,118 @@
"""Output Judge評估輸出是否達標輸出統一的 JudgeResult。
先做 Rule-based便宜、確定性LLM-as-Judge 是日後的加值層build order #9
兩個層級不可混為一談:
- judge_page_status(): 單頁 pass / warn / failwarn 是有效發現,不是報告失敗)
- judge_report(): 整份報告達不達標覆蓋率、頁數下限、格式、issue 具體度)
"""
from __future__ import annotations
import re
from typing import Any
from .contracts import JudgeResult
# 含糊字眼黑名單issue 出現這些、又沒指名具體 URL視為不夠具體
_VAGUE_PHRASES = ["有問題", "部分", "一些", "可能", "異常", "似乎", "好像"]
_URL_RE = re.compile(r"https?://|/[a-zA-Z0-9_\-./]+")
def judge_page_status(checks: dict) -> str:
"""單頁 pass / warn / fail規則直接對應 spec。"""
if not checks.get("page_loads", False):
return "fail" # 404/500 或完全無法載入
images = checks.get("images_ok", [])
images_all_ok = all(
img.get("ok", True) for img in images if isinstance(img, dict)
)
broken = checks.get("broken_links", [])
has_title = checks.get("title_exists", False)
timeouts = [
r for r in checks.get("redirects", [])
if isinstance(r, dict) and r.get("timeout")
]
if has_title and images_all_ok and not broken:
return "pass" # 200 + 有標題 + 圖全載 + 無死連
if not images_all_ok or timeouts:
return "warn" # 可達,但圖載不出 或 外連逾時
return "warn"
def _is_issue_specific(issue: str) -> bool:
"""一條 issue 夠不夠具體:有指名 URL/路徑,且不只是含糊字眼。"""
has_reference = bool(_URL_RE.search(issue))
only_vague = any(p in issue for p in _VAGUE_PHRASES) and not has_reference
return has_reference and not only_vague
class OutputJudge:
def judge_report(
self,
sitemap: dict,
health_reports: list[dict],
min_pages_expected: int,
) -> JudgeResult:
issues: list[str] = []
suggestions: list[str] = []
pages = sitemap.get("pages", [])
n_pages = len(pages)
# 1) 頁數下限:避免爬蟲漏抓
if n_pages < min_pages_expected:
issues.append(
f"sitemap 只有 {n_pages} 頁,低於預期下限 {min_pages_expected}"
f"疑似漏抓。"
)
suggestions.append("檢查爬蟲的連結抽取與 do_not_crawl 規則是否過嚴。")
# 2) 覆蓋率:每個 page 都要有 health_report
mapped_urls = {p.get("url") for p in pages}
reported_urls = {r.get("url") for r in health_reports}
missing = mapped_urls - reported_urls
if missing:
issues.append(
f"{len(missing)} 個頁面缺健檢報告:"
+ ", ".join(list(missing)[:5]) + ("" if len(missing) > 5 else "")
)
# 3) broken_links 格式正確、可被人工逐條檢查
for r in health_reports:
for bl in r.get("checks", {}).get("broken_links", []):
if not (isinstance(bl, dict) and "url" in bl and "reason" in bl):
issues.append(
f"{r.get('url')} 的 broken_links 格式不符(需含 url/reason"
)
break
# 4) AI issue 描述要具體:不能只寫「有問題」
vague_hits = 0
for r in health_reports:
for issue in r.get("issues", []):
if not _is_issue_specific(issue):
vague_hits += 1
issues.append(
f"{r.get('url')} 的 issue 不夠具體:"
f"{issue}」未指名是哪個連結/圖片/問題。"
)
if vague_hits:
suggestions.append(
"每條 issue 都要點名具體的 URL並說明是什麼問題例如逾時/404"
)
# ── 評分:用「扣分制」聚合,再以門檻決定 passed ──
# 覆蓋率與頁數是硬傷重扣issue 具體度是品質問題(輕扣)。
score = 1.0
if n_pages < min_pages_expected:
score -= 0.4
if missing:
score -= 0.4 * (len(missing) / max(n_pages, 1))
if vague_hits:
score -= min(0.4, 0.05 * vague_hits)
score = max(0.0, round(score, 3))
passed = (not issues) and score >= 0.85
return JudgeResult(
passed=passed, score=score, issues=issues, suggestions=suggestions
)

View File

@@ -0,0 +1,86 @@
"""Loop Controller驅動「產出→評估→修正→再產出」的迴圈。
職責:
- 控制重試上限(預設 3
- 每次嘗試完整記錄(透過 on_attempt 回呼交給 Task Queue
- 超過上限或熔斷時送人工審核review
- 含 Circuit Breaker短時間失敗太多次就暫停。
刻意做成通用agent 提供 produce_fn / judge_fn控制器不關心領域細節。
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Callable, Optional
from .circuit_breaker import CircuitBreaker
from .contracts import JudgeResult
from .feedback_builder import FeedbackBuilder
# produce_fn(feedback) -> output judge_fn(output) -> JudgeResult
ProduceFn = Callable[[str], Any]
JudgeFn = Callable[[Any], JudgeResult]
# on_attempt(attempt_no, output, judge, feedback_used)
OnAttempt = Callable[[int, Any, JudgeResult, str], None]
@dataclass
class LoopOutcome:
final_state: str # done / review
output: Any
last_judge: Optional[JudgeResult]
attempts_made: int
class LoopController:
def __init__(
self,
max_attempts: int = 3,
feedback_builder: FeedbackBuilder | None = None,
breaker: CircuitBreaker | None = None,
):
self.max_attempts = max_attempts
self.feedback_builder = feedback_builder or FeedbackBuilder()
self.breaker = breaker or CircuitBreaker()
def run(
self,
produce_fn: ProduceFn,
judge_fn: JudgeFn,
on_attempt: OnAttempt | None = None,
) -> LoopOutcome:
feedback = ""
last_output: Any = None
last_judge: Optional[JudgeResult] = None
for attempt_no in range(1, self.max_attempts + 1):
# 熔斷中:別再打下游,直接送審
if self.breaker.is_open():
return LoopOutcome("review", last_output, last_judge,
attempt_no - 1)
try:
output = produce_fn(feedback)
judge = judge_fn(output)
except Exception as exc: # 產出/評估炸了也算一次失敗
self.breaker.record_failure()
judge = JudgeResult(
passed=False, score=0.0,
issues=[f"執行例外:{exc}"], suggestions=[],
)
output = None
last_output, last_judge = output, judge
if on_attempt:
on_attempt(attempt_no, output, judge, feedback)
if judge.passed:
self.breaker.record_success()
return LoopOutcome("done", output, judge, attempt_no)
# 沒過:記一次失敗,組下一輪 feedback
self.breaker.record_failure()
feedback = self.feedback_builder.build(judge)
# 用完重試上限仍未過 → 送人工審核
return LoopOutcome("review", last_output, last_judge, self.max_attempts)

View File

@@ -0,0 +1,45 @@
"""Prompt Builder從 prompts/*.yaml 載入模板,用 Jinja2 渲染。
不把任何 prompt 文字寫死在程式裡。重試時把上一輪 feedback 注入模板。
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import yaml
from jinja2 import Environment, StrictUndefined
PROMPTS_DIR = Path(__file__).resolve().parent.parent.parent / "prompts"
@dataclass
class RenderedPrompt:
system: str
user: str
class PromptBuilder:
def __init__(self, prompts_dir: Path = PROMPTS_DIR):
self.prompts_dir = prompts_dir
# StrictUndefined模板用到沒給的變數就報錯早點抓到漏帶
self._jinja = Environment(undefined=StrictUndefined, trim_blocks=True,
lstrip_blocks=True)
def build(self, template_name: str, **context: Any) -> RenderedPrompt:
path = self.prompts_dir / f"{template_name}.yaml"
spec = yaml.safe_load(path.read_text(encoding="utf-8"))
# dict / list 類變數轉成漂亮 JSON 再塞進模板LLM 比較好讀
ctx = {
k: (json.dumps(v, ensure_ascii=False, indent=2)
if isinstance(v, (dict, list)) else v)
for k, v in context.items()
}
ctx.setdefault("feedback", "")
system = self._jinja.from_string(spec.get("system", "")).render(**ctx)
user = self._jinja.from_string(spec.get("user", "")).render(**ctx)
return RenderedPrompt(system=system.strip(), user=user.strip())

0
src/db/__init__.py Normal file
View File

44
src/db/schema.sql Normal file
View File

@@ -0,0 +1,44 @@
-- =============================================================================
-- Task Queue schema (SQLite)
-- 三張表分別對應三種時間尺度:
-- tasks = 當前狀態(可變,一筆任務一列)
-- attempts = 完整歷史(只增不改,一次 loop 迭代一列 → 稽核軌跡)
-- results = 最終輸出(通過或送審後的定版)
-- =============================================================================
CREATE TABLE IF NOT EXISTS tasks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
agent TEXT NOT NULL, -- 哪個 agent 負責domain
payload TEXT NOT NULL, -- 輸入 specJSON 字串)
status TEXT NOT NULL DEFAULT 'pending',
-- pending / running / done / failed / review
attempts_made INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
CHECK (status IN ('pending','running','done','failed','review'))
);
CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status);
CREATE TABLE IF NOT EXISTS attempts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_id INTEGER NOT NULL REFERENCES tasks(id),
attempt_no INTEGER NOT NULL, -- 第幾次嘗試1-based
output TEXT, -- 該次 LLM/agent 產出JSON
passed INTEGER, -- judge 結果 0/1
score REAL, -- judge 分數 0.0~1.0
issues TEXT, -- JSON list[str]
suggestions TEXT, -- JSON list[str]
feedback TEXT, -- 餵給下一輪的修正指令
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_attempts_task ON attempts(task_id);
CREATE TABLE IF NOT EXISTS results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_id INTEGER NOT NULL REFERENCES tasks(id),
output TEXT NOT NULL, -- 定版輸出JSON
final_state TEXT NOT NULL, -- done / review
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);

143
src/db/task_queue.py Normal file
View File

@@ -0,0 +1,143 @@
"""Task QueueSQLite 持久層 + CRUD + 防重複取任務。
對外只暴露幾個動詞enqueue / claim / record_attempt / finish。
其餘模組不直接寫 SQL。
"""
from __future__ import annotations
import json
import sqlite3
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Optional
SCHEMA_PATH = Path(__file__).resolve().parent / "schema.sql"
@dataclass
class Task:
id: int
agent: str
payload: dict
status: str
attempts_made: int
class TaskQueue:
def __init__(self, db_path: str | Path):
self.db_path = str(db_path)
# check_same_thread=False 方便日後多 worker單測夠用
self._conn = sqlite3.connect(self.db_path, check_same_thread=False)
self._conn.row_factory = sqlite3.Row
self._conn.execute("PRAGMA journal_mode=WAL;") # 併發讀寫更穩
self._init_schema()
def _init_schema(self) -> None:
self._conn.executescript(SCHEMA_PATH.read_text(encoding="utf-8"))
self._conn.commit()
# --- 寫入任務 -------------------------------------------------------------
def enqueue(self, agent: str, payload: dict) -> int:
cur = self._conn.execute(
"INSERT INTO tasks(agent, payload) VALUES (?, ?)",
(agent, json.dumps(payload, ensure_ascii=False)),
)
self._conn.commit()
return int(cur.lastrowid)
# --- 取任務(防重複)------------------------------------------------------
def claim(self) -> Optional[Task]:
"""原子地搶下一筆 pending 任務。
靠 compare-and-swapUPDATE ... WHERE id=? AND status='pending'
只有一個 worker 能把該列從 pending 翻走rowcount==1 才算搶到)。
"""
# BEGIN IMMEDIATE 先取得寫鎖,避免兩個 worker 同時讀到同一列
self._conn.execute("BEGIN IMMEDIATE")
try:
row = self._conn.execute(
"SELECT id FROM tasks WHERE status='pending' "
"ORDER BY id LIMIT 1"
).fetchone()
if row is None:
self._conn.execute("ROLLBACK")
return None
task_id = row["id"]
cur = self._conn.execute(
"UPDATE tasks SET status='running', "
"updated_at=datetime('now') "
"WHERE id=? AND status='pending'",
(task_id,),
)
if cur.rowcount != 1:
# 別人先搶走了
self._conn.execute("ROLLBACK")
return None
self._conn.execute("COMMIT")
except Exception:
self._conn.execute("ROLLBACK")
raise
full = self._conn.execute(
"SELECT * FROM tasks WHERE id=?", (task_id,)
).fetchone()
return Task(
id=full["id"],
agent=full["agent"],
payload=json.loads(full["payload"]),
status=full["status"],
attempts_made=full["attempts_made"],
)
# --- 記錄每次嘗試 ---------------------------------------------------------
def record_attempt(
self,
task_id: int,
attempt_no: int,
output: Any,
judge: dict,
feedback: str | None,
) -> None:
self._conn.execute(
"INSERT INTO attempts(task_id, attempt_no, output, passed, "
"score, issues, suggestions, feedback) "
"VALUES (?,?,?,?,?,?,?,?)",
(
task_id,
attempt_no,
json.dumps(output, ensure_ascii=False),
1 if judge.get("passed") else 0,
judge.get("score"),
json.dumps(judge.get("issues", []), ensure_ascii=False),
json.dumps(judge.get("suggestions", []), ensure_ascii=False),
feedback,
),
)
self._conn.execute(
"UPDATE tasks SET attempts_made=?, updated_at=datetime('now') "
"WHERE id=?",
(attempt_no, task_id),
)
self._conn.commit()
# --- 收尾done / review / failed ----------------------------------------
def finish(self, task_id: int, final_state: str, output: Any) -> None:
assert final_state in ("done", "review", "failed")
self._conn.execute(
"INSERT INTO results(task_id, output, final_state) VALUES (?,?,?)",
(task_id, json.dumps(output, ensure_ascii=False), final_state),
)
self._conn.execute(
"UPDATE tasks SET status=?, updated_at=datetime('now') WHERE id=?",
(final_state, task_id),
)
self._conn.commit()
def get_status(self, task_id: int) -> str | None:
row = self._conn.execute(
"SELECT status FROM tasks WHERE id=?", (task_id,)
).fetchone()
return row["status"] if row else None
def close(self) -> None:
self._conn.close()

0
src/drivers/__init__.py Normal file
View File

73
src/drivers/api_client.py Normal file
View File

@@ -0,0 +1,73 @@
"""ApiClientDriver用 HTTP 連線/操作目標站。起步只實作這一個 driver。
execute() 支援兩種 action
- "get": 抓整頁(給爬蟲 / 頁面健檢用)
- "head": 只看連得到嗎(給外部連結可達性檢查用,省頻寬)
回傳統一 dictok / status / final_url / elapsed / text / error。
"""
from __future__ import annotations
import time
from typing import Any
import requests
from .base import BaseDriver
class ApiClientDriver(BaseDriver):
def __init__(self, timeout: int = 10, user_agent: str = "ai-autotest-bot/0.1"):
self.timeout = timeout
self.user_agent = user_agent
self._session: requests.Session | None = None
def connect(self) -> None:
self._session = requests.Session()
self._session.headers.update({"User-Agent": self.user_agent})
def execute(self, action: str, **kwargs: Any) -> dict:
if self._session is None:
raise RuntimeError("driver 尚未 connect()")
url = kwargs["url"]
allow_redirects = kwargs.get("allow_redirects", True)
t0 = time.monotonic()
try:
if action == "get":
resp = self._session.get(
url, timeout=self.timeout, allow_redirects=allow_redirects
)
text = resp.text
elif action == "head":
resp = self._session.head(
url, timeout=self.timeout, allow_redirects=allow_redirects
)
text = ""
else:
raise ValueError(f"未知 action{action}")
elapsed = time.monotonic() - t0
return {
"ok": resp.status_code < 400,
"status": resp.status_code,
"final_url": resp.url,
"redirected": resp.url != url,
"elapsed": round(elapsed, 3),
"text": text,
"error": None,
}
except requests.Timeout:
return self._err(url, "timeout", time.monotonic() - t0, timeout=True)
except requests.RequestException as exc:
return self._err(url, str(exc), time.monotonic() - t0)
@staticmethod
def _err(url: str, msg: str, elapsed: float, timeout: bool = False) -> dict:
return {
"ok": False, "status": None, "final_url": url, "redirected": False,
"elapsed": round(elapsed, 3), "text": "", "error": msg,
"timeout": timeout,
}
def close(self) -> None:
if self._session:
self._session.close()

22
src/drivers/base.py Normal file
View File

@@ -0,0 +1,22 @@
"""Driver 層抽象介面:定義「怎麼連線、怎麼操作」,不含任何領域判斷。
agents 一律透過這個介面操作目標,未來要換成 web(Playwright)/mobile
驅動時agents 不用改。
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any
class BaseDriver(ABC):
@abstractmethod
def connect(self) -> None:
"""建立連線 / 開啟 session。"""
@abstractmethod
def execute(self, action: str, **kwargs: Any) -> Any:
"""執行一個操作(例如 get / head回傳統一格式的結果。"""
def close(self) -> None: # 選用
pass

0
src/llm/__init__.py Normal file
View File

108
src/llm/backend.py Normal file
View File

@@ -0,0 +1,108 @@
"""LLM 後端抽象層 + Ollama 實作 + Router。
抽象介面 LLMBackend
- call(prompt, system) -> str
- is_available() -> bool
OllamaBackend 每次 call 都帶 num_ctx預設 32768不依賴 Ollama 預設的 4K。
LLMRouter 依序嘗試多個 backend用第一個可用的。
"""
from __future__ import annotations
import logging
from abc import ABC, abstractmethod
import requests
log = logging.getLogger(__name__)
DEFAULT_NUM_CTX = 32768
class LLMBackend(ABC):
@abstractmethod
def call(self, prompt: str, system: str = "") -> str:
"""送一次 prompt回傳純文字輸出。"""
@abstractmethod
def is_available(self) -> bool:
"""這個 backend 現在打得通嗎。"""
class OllamaBackend(LLMBackend):
def __init__(
self,
base_url: str,
model: str,
num_ctx: int = DEFAULT_NUM_CTX,
timeout: int = 120,
):
self.base_url = base_url.rstrip("/")
self.model = model
self.num_ctx = num_ctx
self.timeout = timeout
def call(self, prompt: str, system: str = "") -> str:
payload = {
"model": self.model,
"prompt": prompt,
"system": system,
"stream": False,
# ── 關鍵:強制覆蓋 Ollama 預設 4K context ──
"options": {"num_ctx": self.num_ctx},
}
resp = requests.post(
f"{self.base_url}/api/generate", json=payload, timeout=self.timeout
)
resp.raise_for_status()
return resp.json().get("response", "")
def is_available(self) -> bool:
try:
resp = requests.get(f"{self.base_url}/api/tags", timeout=3)
if resp.status_code != 200:
return False
names = {m["name"] for m in resp.json().get("models", [])}
return self.model in names
except requests.RequestException:
return False
class LLMRouter(LLMBackend):
"""依序持有多個 backend呼叫時用第一個 is_available() 的。"""
def __init__(self, backends: list[LLMBackend]):
if not backends:
raise ValueError("LLMRouter 需要至少一個 backend")
self.backends = backends
def _pick(self) -> LLMBackend:
for b in self.backends:
if b.is_available():
return b
log.warning("backend 不可用,跳過:%r", b)
raise RuntimeError("沒有任何可用的 LLM backend")
def call(self, prompt: str, system: str = "") -> str:
return self._pick().call(prompt, system)
def is_available(self) -> bool:
return any(b.is_available() for b in self.backends)
def build_router_from_config(llm_cfg: dict) -> LLMRouter:
"""從 environments.yaml 的 llm 區塊組出 Router。"""
num_ctx = int(llm_cfg.get("num_ctx", DEFAULT_NUM_CTX))
backends: list[LLMBackend] = []
for b in llm_cfg.get("backends", []):
if b.get("type") == "ollama":
backends.append(
OllamaBackend(
base_url=b["base_url"],
model=b["model"],
num_ctx=num_ctx,
)
)
else:
log.warning("未知 backend type%s(略過)", b.get("type"))
return LLMRouter(backends)

133
src/pipeline.py Normal file
View File

@@ -0,0 +1,133 @@
"""pipeline.py —— 系統入口。
用法:
python -m src.pipeline --seed # 放一筆任務進 Task Queue
python -m src.pipeline # 處理所有 pending 任務
python -m src.pipeline --seed --run
流程claim 任務 → 派給對應 agent → 跑 Loop 引擎 → 每次嘗試寫 attempts
→ 收尾 done/review → 輸出落地 output/review 另存 review_queue/)。
"""
from __future__ import annotations
import argparse
import json
import logging
from pathlib import Path
from .config import load_config, ROOT
from .core.circuit_breaker import CircuitBreaker
from .core.feedback_builder import FeedbackBuilder
from .core.loop_controller import LoopController
from .agents.site_health import SiteHealthAgent
from .drivers.api_client import ApiClientDriver
from .llm.backend import build_router_from_config
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
log = logging.getLogger("pipeline")
OUTPUT_DIR = ROOT / "output"
REVIEW_DIR = ROOT / "review_queue"
DB_PATH = ROOT / "tasks.db"
def build_agent(agent_name: str, cfg):
"""Agent 工廠:依任務指定的 agent 名稱組裝(含其 driver 與 loop 引擎)。"""
env = cfg.env()
if agent_name == "site_health":
driver = ApiClientDriver(timeout=env["target"]["request_timeout_sec"])
llm = build_router_from_config(env["llm"])
loop_cfg = env["loop"]
breaker = CircuitBreaker(**loop_cfg["circuit_breaker"])
loop = LoopController(
max_attempts=loop_cfg["max_attempts"],
feedback_builder=FeedbackBuilder(),
breaker=breaker,
)
return SiteHealthAgent(driver, llm, loop, env["target"])
raise ValueError(f"未知 agent{agent_name}")
def seed(queue, cfg) -> int:
"""放一筆 site_health 任務進佇列spec 的驗收起點)。"""
payload = {"start_url": cfg.env()["target"]["start_url"]}
task_id = queue.enqueue("site_health", payload)
log.info("已放入任務 #%sagent=site_health", task_id)
return task_id
def _write_outputs(task_id: int, output: dict, final_state: str) -> None:
OUTPUT_DIR.mkdir(exist_ok=True)
if output:
(OUTPUT_DIR / "sitemap.json").write_text(
json.dumps(output["sitemap"], ensure_ascii=False, indent=2),
encoding="utf-8",
)
(OUTPUT_DIR / "health_report.json").write_text(
json.dumps(output["health_reports"], ensure_ascii=False, indent=2),
encoding="utf-8",
)
if final_state == "review":
REVIEW_DIR.mkdir(exist_ok=True)
(REVIEW_DIR / f"task_{task_id}.json").write_text(
json.dumps(output, ensure_ascii=False, indent=2), encoding="utf-8"
)
def process_all(queue, cfg) -> None:
while True:
task = queue.claim()
if task is None:
log.info("沒有 pending 任務,結束。")
return
log.info("處理任務 #%sagent=%s", task.id, task.agent)
agent = build_agent(task.agent, cfg)
def on_attempt(no, output, judge, feedback):
queue.record_attempt(
task.id, no, output, judge.to_dict(),
feedback=feedback or None,
)
log.info(" attempt #%s passed=%s score=%.2f issues=%d",
no, judge.passed, judge.score, len(judge.issues))
try:
outcome = agent.run(task.payload, on_attempt=on_attempt)
except Exception as exc:
log.exception("任務 #%s 執行失敗", task.id)
queue.finish(task.id, "failed", {"error": str(exc)})
continue
queue.finish(task.id, outcome.final_state, outcome.output)
_write_outputs(task.id, outcome.output, outcome.final_state)
log.info("任務 #%s%s%d 次嘗試)",
task.id, outcome.final_state, outcome.attempts_made)
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--seed", action="store_true", help="放一筆任務進佇列")
parser.add_argument("--run", action="store_true", help="處理 pending 任務")
parser.add_argument("--env", default=None, help="覆蓋環境名dev/prod")
args = parser.parse_args()
cfg = load_config(args.env)
log.info("使用環境:%s", cfg.active)
# 延遲 import避免在沒裝套件時連 --help 都跑不動
from .db.task_queue import TaskQueue
queue = TaskQueue(DB_PATH)
# 沒給任何旗標時預設「seed + run」一次跑完方便驗收
do_seed = args.seed or not (args.seed or args.run)
do_run = args.run or not (args.seed or args.run)
if do_seed:
seed(queue, cfg)
if do_run:
process_all(queue, cfg)
queue.close()
if __name__ == "__main__":
main()