From 27f98712ee915753232cc13986ec44ccb9671065 Mon Sep 17 00:00:00 2001 From: Yellow Date: Sun, 14 Jun 2026 17:35:54 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20AI=20Self-Loop=20=E5=B0=88=E6=A1=88?= =?UTF-8?q?=E9=AA=A8=E6=9E=B6=20=E2=80=94=20=E7=88=AC=E7=AB=99=E5=81=A5?= =?UTF-8?q?=E6=AA=A2=E7=89=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 完整實作建構順序 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 --- .gitignore | 15 ++ README.md | 66 +++++++++ config/credentials.example.yaml | 20 +++ config/environments.yaml | 70 +++++++++ output/.gitkeep | 0 prompts/health_issues.yaml | 30 ++++ requirements.txt | 6 + review_queue/.gitkeep | 0 specs/site_health.example.json | 4 + src/__init__.py | 0 src/agents/__init__.py | 0 src/agents/base.py | 19 +++ src/agents/site_health.py | 249 ++++++++++++++++++++++++++++++++ src/config.py | 80 ++++++++++ src/core/__init__.py | 0 src/core/circuit_breaker.py | 57 ++++++++ src/core/context_assembler.py | 44 ++++++ src/core/contracts.py | 24 +++ src/core/feedback_builder.py | 24 +++ src/core/judge.py | 118 +++++++++++++++ src/core/loop_controller.py | 86 +++++++++++ src/core/prompt_builder.py | 45 ++++++ src/db/__init__.py | 0 src/db/schema.sql | 44 ++++++ src/db/task_queue.py | 143 ++++++++++++++++++ src/drivers/__init__.py | 0 src/drivers/api_client.py | 73 ++++++++++ src/drivers/base.py | 22 +++ src/llm/__init__.py | 0 src/llm/backend.py | 108 ++++++++++++++ src/pipeline.py | 133 +++++++++++++++++ 31 files changed, 1480 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 config/credentials.example.yaml create mode 100644 config/environments.yaml create mode 100644 output/.gitkeep create mode 100644 prompts/health_issues.yaml create mode 100644 requirements.txt create mode 100644 review_queue/.gitkeep create mode 100644 specs/site_health.example.json create mode 100644 src/__init__.py create mode 100644 src/agents/__init__.py create mode 100644 src/agents/base.py create mode 100644 src/agents/site_health.py create mode 100644 src/config.py create mode 100644 src/core/__init__.py create mode 100644 src/core/circuit_breaker.py create mode 100644 src/core/context_assembler.py create mode 100644 src/core/contracts.py create mode 100644 src/core/feedback_builder.py create mode 100644 src/core/judge.py create mode 100644 src/core/loop_controller.py create mode 100644 src/core/prompt_builder.py create mode 100644 src/db/__init__.py create mode 100644 src/db/schema.sql create mode 100644 src/db/task_queue.py create mode 100644 src/drivers/__init__.py create mode 100644 src/drivers/api_client.py create mode 100644 src/drivers/base.py create mode 100644 src/llm/__init__.py create mode 100644 src/llm/backend.py create mode 100644 src/pipeline.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..cba7eb2 --- /dev/null +++ b/.gitignore @@ -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 diff --git a/README.md b/README.md new file mode 100644 index 0000000..713f1db --- /dev/null +++ b/README.md @@ -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_.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`。 diff --git a/config/credentials.example.yaml b/config/credentials.example.yaml new file mode 100644 index 0000000..e4586b6 --- /dev/null +++ b/config/credentials.example.yaml @@ -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:-}" diff --git a/config/environments.yaml b/config/environments.yaml new file mode 100644 index 0000000..aa4db13 --- /dev/null +++ b/config/environments.yaml @@ -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 diff --git a/output/.gitkeep b/output/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/prompts/health_issues.yaml b/prompts/health_issues.yaml new file mode 100644 index 0000000..8738686 --- /dev/null +++ b/prompts/health_issues.yaml @@ -0,0 +1,30 @@ +# ============================================================================= +# 模板：把單一頁面的「原始事實」轉成具體、可逐條檢查的 issue 描述。 +# 用 Jinja2 渲染。重試時 feedback 區塊會帶入上一輪失敗原因。 +# +# 可用變數： +# url 頁面網址 +# facts dict（broken_links / image_problems / redirects / ...） +# feedback 上一輪的修正指令（第一次為空） +# ============================================================================= +system: | + 你是一個網站健檢報告撰寫員。你的任務是把機器掃描出的原始事實， + 改寫成「具體到哪個連結、什麼問題」的繁體中文 issue 描述。 + 規則： + - 每條 issue 必須點名是哪個 URL / 哪張圖 / 哪個表單。 + - 嚴禁含糊字眼（例如「有問題」「部分連結異常」）。 + - 只描述 facts 裡確實存在的問題，不要臆測。 + - 只輸出 JSON，格式：{"issues": ["...", "..."]}。沒有問題就回 {"issues": []}。 + +user: | + 頁面：{{ url }} + + 掃描到的原始事實（JSON）： + {{ facts }} + + {% if feedback %} + ── 上一輪的輸出未達標，請依下列修正後重寫 ── + {{ feedback }} + {% endif %} + + 請輸出 issues 的 JSON。 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..693eb97 --- /dev/null +++ b/requirements.txt @@ -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 diff --git a/review_queue/.gitkeep b/review_queue/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/specs/site_health.example.json b/specs/site_health.example.json new file mode 100644 index 0000000..eb00c35 --- /dev/null +++ b/specs/site_health.example.json @@ -0,0 +1,4 @@ +{ + "_comment": "site_health 任務的輸入範例。實際 start_url 以 config/environments.yaml 為準；這裡只是說明 payload 形狀。", + "start_url": "https://www.pluspay.com.tw/" +} diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/agents/__init__.py b/src/agents/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/agents/base.py b/src/agents/base.py new file mode 100644 index 0000000..e296ba0 --- /dev/null +++ b/src/agents/base.py @@ -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: + """執行一筆任務，回傳 LoopOutcome（final_state + output）。""" diff --git a/src/agents/site_health.py b/src/agents/site_health.py new file mode 100644 index 0000000..9f1363f --- /dev/null +++ b/src/agents/site_health.py @@ -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() diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..fdf8913 --- /dev/null +++ b/src/config.py @@ -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) diff --git a/src/core/__init__.py b/src/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/core/circuit_breaker.py b/src/core/circuit_breaker.py new file mode 100644 index 0000000..edb6f83 --- /dev/null +++ b/src/core/circuit_breaker.py @@ -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 diff --git a/src/core/context_assembler.py b/src/core/context_assembler.py new file mode 100644 index 0000000..44f0ab5 --- /dev/null +++ b/src/core/context_assembler.py @@ -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 diff --git a/src/core/contracts.py b/src/core/contracts.py new file mode 100644 index 0000000..df917cd --- /dev/null +++ b/src/core/contracts.py @@ -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, + } diff --git a/src/core/feedback_builder.py b/src/core/feedback_builder.py new file mode 100644 index 0000000..e18d44a --- /dev/null +++ b/src/core/feedback_builder.py @@ -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) diff --git a/src/core/judge.py b/src/core/judge.py new file mode 100644 index 0000000..10b34d0 --- /dev/null +++ b/src/core/judge.py @@ -0,0 +1,118 @@ +"""Output Judge：評估輸出是否達標，輸出統一的 JudgeResult。 + +先做 Rule-based（便宜、確定性）；LLM-as-Judge 是日後的加值層（build order #9）。 + +兩個層級不可混為一談： + - judge_page_status(): 單頁 pass / warn / fail（warn 是有效發現，不是報告失敗） + - 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 + ) diff --git a/src/core/loop_controller.py b/src/core/loop_controller.py new file mode 100644 index 0000000..cb76647 --- /dev/null +++ b/src/core/loop_controller.py @@ -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) diff --git a/src/core/prompt_builder.py b/src/core/prompt_builder.py new file mode 100644 index 0000000..edf2818 --- /dev/null +++ b/src/core/prompt_builder.py @@ -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()) diff --git a/src/db/__init__.py b/src/db/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/db/schema.sql b/src/db/schema.sql new file mode 100644 index 0000000..3d2ddcf --- /dev/null +++ b/src/db/schema.sql @@ -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, -- 輸入 spec（JSON 字串） + 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')) +); diff --git a/src/db/task_queue.py b/src/db/task_queue.py new file mode 100644 index 0000000..0ab10f4 --- /dev/null +++ b/src/db/task_queue.py @@ -0,0 +1,143 @@ +"""Task Queue：SQLite 持久層 + 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-swap：UPDATE ... 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() diff --git a/src/drivers/__init__.py b/src/drivers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/drivers/api_client.py b/src/drivers/api_client.py new file mode 100644 index 0000000..82725ba --- /dev/null +++ b/src/drivers/api_client.py @@ -0,0 +1,73 @@ +"""ApiClientDriver：用 HTTP 連線/操作目標站。起步只實作這一個 driver。 + +execute() 支援兩種 action： + - "get": 抓整頁（給爬蟲 / 頁面健檢用） + - "head": 只看連得到嗎（給外部連結可達性檢查用，省頻寬） + +回傳統一 dict：ok / 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() diff --git a/src/drivers/base.py b/src/drivers/base.py new file mode 100644 index 0000000..14d6329 --- /dev/null +++ b/src/drivers/base.py @@ -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 diff --git a/src/llm/__init__.py b/src/llm/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/llm/backend.py b/src/llm/backend.py new file mode 100644 index 0000000..53c2422 --- /dev/null +++ b/src/llm/backend.py @@ -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) diff --git a/src/pipeline.py b/src/pipeline.py new file mode 100644 index 0000000..02a7ae7 --- /dev/null +++ b/src/pipeline.py @@ -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("已放入任務 #%s（agent=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("處理任務 #%s（agent=%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()