昨天畫好了架構圖，今天開始真的把程式碼接起來。

這一篇會比較多程式碼，但重點不在「抄下來」，而在看這些零件怎麼組合。

一、先把待辦儲存層補完

前面幾天的待辦功能散在各處，先集中成一個模組：

"""storage/todo_store.py — 待辦事項的資料存取。"""

import json
import logging
from datetime import date, datetime
from pathlib import Path

import config

logger = logging.getLogger(__name__)

TODO_FILE = config.DATA_DIR / "todos.json"

VALID_PRIORITIES = ("高", "中", "低")
PRIORITY_ORDER = {"高": 0, "中": 1, "低": 2}


def load_todos():
    """讀取所有待辦事項。"""
    if not TODO_FILE.exists():
        return []
    try:
        return json.loads(TODO_FILE.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        logger.error("待辦檔案格式損壞，備份後重建")
        TODO_FILE.rename(TODO_FILE.with_suffix(".json.bak"))
        return []


def save_todos(todos):
    """寫回待辦事項。"""
    TODO_FILE.parent.mkdir(parents=True, exist_ok=True)
    TODO_FILE.write_text(
        json.dumps(todos, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )


def add(title, priority="中", due=None):
    """新增一筆待辦。"""
    title = (title or "").strip()
    if not title:
        raise ValueError("待辦內容不能是空的")
    if priority not in VALID_PRIORITIES:
        raise ValueError(f"優先度只能是 {'/'.join(VALID_PRIORITIES)}")
    if due:
        try:
            date.fromisoformat(due)
        except ValueError:
            raise ValueError(f"日期格式錯誤（應為 YYYY-MM-DD）：{due}")

    todos = load_todos()
    todo = {
        "id": max((t["id"] for t in todos), default=0) + 1,
        "title": title,
        "priority": priority,
        "due": due,
        "done": False,
        "created_at": datetime.now().isoformat(timespec="seconds"),
        "completed_at": None,
    }
    todos.append(todo)
    save_todos(todos)
    logger.info("新增待辦 #%d：%s", todo["id"], title)
    return todo


def complete(todo_id):
    todos = load_todos()
    for todo in todos:
        if todo["id"] == todo_id:
            if todo["done"]:
                return todo, False
            todo["done"] = True
            todo["completed_at"] = datetime.now().isoformat(timespec="seconds")
            save_todos(todos)
            return todo, True
    raise KeyError(f"找不到編號 {todo_id} 的待辦事項")


def delete(todo_id):
    todos = load_todos()
    target = next((t for t in todos if t["id"] == todo_id), None)
    if target is None:
        raise KeyError(f"找不到編號 {todo_id} 的待辦事項")
    todos.remove(target)
    save_todos(todos)
    logger.info("刪除待辦 #%d：%s", todo_id, target["title"])
    return target


def days_until(due_str, today=None):
    """計算距離期限還有幾天。沒有期限回傳 None。"""
    if not due_str:
        return None
    today = today or date.today()
    try:
        return (date.fromisoformat(due_str) - today).days
    except ValueError:
        return None

幾個實務細節：

檔案壞掉時備份再重建

TODO_FILE.rename(TODO_FILE.with_suffix(".json.bak"))

不要直接當作空的——萬一只是壞了一個字元，備份起來還有機會手動救回。

complete() 回傳 (todo, changed)
區分「本來就完成了」和「剛剛完成」。工具的回覆可以更精準：「這筆本來就已經完成了」vs「已標記完成」。

add() 驗證日期格式
Day 8 的原則：在源頭擋下不合理的資料。模型偶爾會給「2026-13-45」這種東西。

二、把工具補齊

"""tools/todo_tool.py"""

from datetime import date, timedelta

from storage import todo_store
from tools.base import Tool, ToolError


class AddTodoTool(Tool):
    name = "add_todo"
    tags = ("todo", "write")
    description = (
        "新增一筆待辦事項到使用者的清單。"
        "當使用者說「提醒我…」「記一下…」「我要做…」「別忘了…」時使用。"
        "只負責記錄，不會設定鬧鐘或推播通知。"
        "如果使用者提到時間（明天、禮拜五、月底），"
        "請先用 get_current_datetime 確認今天日期，再換算成 YYYY-MM-DD。"
    )
    input_schema = {
        "type": "object",
        "properties": {
            "title": {
                "type": "string",
                "description": "待辦內容，簡短具體，例如「繳電費」",
            },
            "priority": {
                "type": "string",
                "enum": ["高", "中", "低"],
                "description": (
                    "優先度。有明確期限或使用者語氣急迫時為「高」，"
                    "「有空再做」為「低」，其他為「中」。"
                ),
            },
            "due": {
                "type": "string",
                "description": "期限，格式 YYYY-MM-DD。使用者沒提到時間就不要填。",
            },
        },
        "required": ["title"],
    }

    def run(self, title, priority="中", due=None):
        try:
            todo = todo_store.add(title, priority, due)
        except ValueError as e:
            raise ToolError(str(e)) from e

        due_text = f"，期限 {todo['due']}" if todo["due"] else ""
        return f"已新增 #{todo['id']}「{todo['title']}」（{todo['priority']}優先{due_text}）。"


class ListTodosTool(Tool):
    name = "list_todos"
    tags = ("todo",)
    description = (
        "列出使用者的待辦事項，包含編號、內容、優先度與期限。"
        "當使用者問「我有什麼事」「待辦清單」「今天要做什麼」，"
        "或你在規劃行程時需要知道有哪些事項時使用。"
    )
    input_schema = {
        "type": "object",
        "properties": {
            "only_pending": {
                "type": "boolean",
                "description": "只列出未完成的。預設 true。",
            }
        },
    }

    def run(self, only_pending=True):
        todos = todo_store.load_todos()
        if only_pending:
            todos = [t for t in todos if not t["done"]]

        if not todos:
            return "目前沒有待辦事項。"

        # Day 16 的多重排序
        todos.sort(key=lambda t: (
            t["done"],
            todo_store.PRIORITY_ORDER[t["priority"]],
            todo_store.days_until(t["due"]) if t["due"] else 9999,
        ))

        lines = [f"共 {len(todos)} 筆："]
        for t in todos:
            mark = "✓" if t["done"] else "□"
            days = todo_store.days_until(t["due"])
            if days is None:
                due_text = ""
            elif days < 0:
                due_text = f"（已逾期 {-days} 天）"
            elif days == 0:
                due_text = "（今天到期）"
            else:
                due_text = f"（剩 {days} 天）"
            lines.append(f"{mark} #{t['id']} {t['title']}　{t['priority']}優先{due_text}")

        return "\n".join(lines)


class CompleteTodoTool(Tool):
    name = "complete_todo"
    tags = ("todo", "write")
    description = (
        "把一筆待辦事項標記為已完成。"
        "當使用者說「做完了」「已經處理了」「完成第 N 項」時使用。"
        "這不會刪除資料，只是標記狀態。"
        "如果不確定是哪一筆，先用 list_todos 確認編號。"
    )
    input_schema = {
        "type": "object",
        "properties": {
            "todo_id": {"type": "integer", "description": "待辦事項的編號"}
        },
        "required": ["todo_id"],
    }

    def run(self, todo_id):
        try:
            todo, changed = todo_store.complete(todo_id)
        except KeyError as e:
            raise ToolError(str(e)) from e

        if not changed:
            return f"#{todo_id}「{todo['title']}」本來就已經完成了。"
        return f"已完成 #{todo_id}「{todo['title']}」。"


class DeleteTodoTool(Tool):
    name = "delete_todo"
    tags = ("todo", "write")
    dangerous = True
    description = (
        "永久刪除一筆待辦事項。這個操作無法復原。"
        "只在使用者明確要求「刪除」「移除」時使用。"
        "如果事情只是做完了，請用 complete_todo 而不是刪除。"
    )
    input_schema = {
        "type": "object",
        "properties": {
            "todo_id": {"type": "integer", "description": "要刪除的待辦編號"}
        },
        "required": ["todo_id"],
    }

    def run(self, todo_id):
        try:
            todo = todo_store.delete(todo_id)
        except KeyError as e:
            raise ToolError(str(e)) from e
        return f"已刪除 #{todo_id}「{todo['title']}」。"

注意 AddTodoTool 的 description 裡這句：

「如果使用者提到時間（明天、禮拜五、月底），請先用 get_current_datetime 確認今天日期，再換算成 YYYY-MM-DD。」

**這是在 tool description 裡教模型工具之間的先後關係。**沒有這句，模型很容易憑自己的印象猜日期，而它的印象可能是訓練資料的日期，差很遠。

三、今日簡報 Workflow

把 Day 24 的引擎實際用起來：

"""workflows/daily_brief.py"""

import logging
from datetime import datetime

import config
import prompts
from analysis.todo_insight import analyze_todos, make_todo_summary
from analysis.weather_insight import analyze_weather, make_advice
from storage import todo_store
from tools.weather import get_weather
from workflows.engine import ConditionalStep, Step, Workflow

logger = logging.getLogger(__name__)

WEEKDAYS = ["星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日"]


def step_time(state):
    now = datetime.now()
    state.set("now", now)
    state.set("time_text",
              f"{now.strftime('%Y年%m月%d日 %H:%M')}（{WEEKDAYS[now.weekday()]}）")
    return state


def make_step_weather(city):
    def step_weather(state):
        data = get_weather(city, config.CWA_API_KEY)
        insight = analyze_weather(data)
        state.set("weather_insight", insight)
        state.set("weather_text", make_advice(insight))
        return state
    return step_weather


def step_todos(state):
    todos = todo_store.load_todos()
    insight = analyze_todos(todos)
    state.set("todo_insight", insight)
    state.set("todo_text", make_todo_summary(insight))
    return state


def step_urgent_flag(state):
    """有逾期或今天到期的事項時，標記出來。"""
    insight = state.get("todo_insight", {})
    urgent = list(insight.get("overdue", [])) + list(insight.get("due_today", []))
    titles = "、".join(t["title"] for t in urgent)
    state.set("urgent_note", f"⚠️ 今天一定要處理：{titles}")
    return state


def make_step_compose(client, profile_context=""):
    def step_compose(state):
        prompt = prompts.DAILY_BRIEF.substitute(
            current_time=state.get("time_text"),
            weather=state.get("weather_text", "（今天沒有取得天氣資訊）"),
            todos=state.get("todo_text", "（今天沒有取得待辦資訊）"),
        )

        if state.get("urgent_note"):
            prompt += f"\n\n{state.get('urgent_note')}"

        response = client.messages.create(
            model="claude-opus-5",
            max_tokens=1024,
            system=prompts.ASSISTANT_SYSTEM + profile_context,
            messages=[{"role": "user", "content": prompt}],
        )
        state.add_usage(response)

        text = "\n".join(b.text for b in response.content if b.type == "text").strip()

        if state.warnings:
            notes = "\n".join(f"（{w}）" for w in state.warnings)
            text = f"{text}\n\n{notes}"

        state.result = text
        return state
    return step_compose


def build(client, city=None, profile_context=""):
    city = city or config.DEFAULT_CITY
    return Workflow("今日簡報", [
        Step("取得時間", step_time, required=True),
        Step("查詢天氣", make_step_weather(city), required=False, retries=1),
        Step("讀取待辦", step_todos, required=False),
        ConditionalStep(
            "標記緊急事項", step_urgent_flag,
            condition=lambda s: bool(
                s.get("todo_insight", {}).get("overdue")
                or s.get("todo_insight", {}).get("due_today")
            ),
            required=False,
        ),
        Step("合成簡報", make_step_compose(client, profile_context),
             required=True, requires=("time_text",)),
    ])

**整個流程只呼叫一次 AI。**時間、天氣、待辦、緊急判斷全是純程式。

這就是 Day 23 講的「能省的就省」——每天早上跑一次，成本大約是 Agent 版本的五分之一。

四、意圖路由

Day 23 的 Router，這次補完整：

"""workflows/router.py"""

import logging
from typing import Literal, Optional

from pydantic import BaseModel, Field

import config
from analysis.todo_insight import analyze_todos, make_todo_summary
from analysis.weather_insight import analyze_weather, make_advice
from storage import todo_store
from tools.weather import WeatherError, get_weather
from workflows import daily_brief

logger = logging.getLogger(__name__)

CONFIDENCE_THRESHOLD = 0.7


class Intent(BaseModel):
    reasoning: str = Field(description="一句話說明判斷依據")
    action: Literal[
        "daily_brief", "query_weather", "list_todos", "agent"
    ] = Field(description="要走的處理路徑")
    confidence: float = Field(ge=0.0, le=1.0)
    city: Optional[str] = Field(default=None, description="若查天氣，指定的縣市")


INTENT_SYSTEM = """你是一個請求分類器，判斷使用者的請求該走哪條處理路徑。

路徑說明：
- daily_brief：要求今日整體概況，例如「早安」「今天怎麼樣」「幫我看今天」
- query_weather：單純詢問天氣、要不要帶傘、穿什麼
- list_todos：單純想看待辦清單
- agent：其他所有情況，包含新增待辦、修改資料、複雜或多步驟的請求、閒聊

判斷原則：
- 只有在請求「完全符合」前三種單純情境時，才選擇對應路徑
- 只要帶有額外條件、多個動作、或需要判斷，一律選 agent
- 不確定時選 agent，並在 confidence 中反映

舉例：
「明天要不要帶傘」→ query_weather（單純）
「明天要不要帶傘，順便幫我記一下」→ agent（兩個動作）
「早安」→ daily_brief
「我有什麼事」→ list_todos
「幫我把第三件事做完」→ agent（需要修改資料）"""


class Router:
    def __init__(self, client, agent, profile=None):
        self.client = client
        self.agent = agent
        self.profile = profile
        self.stats = {"daily_brief": 0, "query_weather": 0,
                      "list_todos": 0, "agent": 0}

    def classify(self, user_input) -> Intent:
        response = self.client.messages.parse(
            model="claude-haiku-4-5",        # 分類用便宜的模型
            max_tokens=512,
            system=INTENT_SYSTEM,
            messages=[{"role": "user", "content": user_input}],
            output_format=Intent,
        )
        return response.parsed_output

    def handle(self, user_input, history=None, **agent_kwargs):
        try:
            intent = self.classify(user_input)
        except Exception as e:
            logger.warning("意圖分類失敗，改用 Agent：%s", e)
            return self._run_agent(user_input, history, **agent_kwargs)

        logger.info("路由：%s（%.2f）— %s",
                    intent.action, intent.confidence, intent.reasoning)

        if intent.confidence < CONFIDENCE_THRESHOLD:
            logger.info("信心不足，改用 Agent")
            return self._run_agent(user_input, history, **agent_kwargs)

        self.stats[intent.action] += 1

        if intent.action == "daily_brief":
            return self._daily_brief()
        if intent.action == "query_weather":
            return self._weather(intent.city)
        if intent.action == "list_todos":
            return self._todos()

        return self._run_agent(user_input, history, **agent_kwargs)

    # --- 各路徑 ---

    def _daily_brief(self):
        context = self.profile.to_context() if self.profile else ""
        workflow = daily_brief.build(self.client, profile_context=context)
        state = workflow.run("今日簡報")
        logger.info("簡報完成：%s", state.summary())
        return state.result or "抱歉，今天的簡報產生失敗了。"

    def _weather(self, city=None):
        """完全不用 AI。"""
        city = city or (self.profile.data.get("city") if self.profile else None) \
               or config.DEFAULT_CITY
        try:
            data = get_weather(city, config.CWA_API_KEY)
        except WeatherError as e:
            return f"查不到天氣：{e}"
        return make_advice(analyze_weather(data))

    def _todos(self):
        """完全不用 AI。"""
        return make_todo_summary(analyze_todos(todo_store.load_todos()))

    def _run_agent(self, user_input, history, **kwargs):
        self.stats["agent"] += 1
        answer, _ = self.agent.run(user_input, history=history, **kwargs)
        return answer

    def report_stats(self):
        total = sum(self.stats.values())
        if total == 0:
            return "尚未處理任何請求"
        lines = ["請求路由統計："]
        for path, count in sorted(self.stats.items(), key=lambda x: -x[1]):
            lines.append(f"  {path:15s} {count:3d} 次（{count / total:.0%}）")
        return "\n".join(lines)

幾個設計重點：

分類用 Haiku，回答用 Opus
分類是簡單任務，用便宜的模型。這是「多模型分工」的基本應用。

INTENT_SYSTEM 裡給了對照範例
「明天要不要帶傘」vs「明天要不要帶傘，順便幫我記一下」——這種對比範例比抽象規則有效太多（Day 19 的 few-shot）。

分類失敗時 fallback 到 Agent
分類器本身也可能出錯。try/except 包住，出事就走 Agent。降級要平順。

_weather 和 _todos 完全沒有呼叫 AI
資料查詢 + Day 16 的分析函式就夠了。回應是模板化的，但又快又準又免費。

report_stats()
這是為了 Day 29 的驗收——我要知道實際上有多少比例走了 Agent。

五、把工具組裝起來

"""tools/registry.py — 建立完整的 ToolBox。"""

import logging

from tools.base import ToolBox
from tools.calculator_tool import CalculatorTool
from tools.datetime_tool import DateTimeTool
from tools.memory_tool import ForgetTool, RecallTool, RememberTool
from tools.todo_tool import (AddTodoTool, CompleteTodoTool,
                             DeleteTodoTool, ListTodosTool)
from tools.weather_tool import WeatherTool

logger = logging.getLogger(__name__)


def build_toolbox(profile=None, city=None):
    """建立助理用的完整工具箱。"""
    tools = [
        DateTimeTool(),
        WeatherTool(default_city=city),
        AddTodoTool(),
        ListTodosTool(),
        CompleteTodoTool(),
        DeleteTodoTool(),
        CalculatorTool(),
    ]

    if profile is not None:
        tools.extend([
            RememberTool(profile),
            RecallTool(profile),
            ForgetTool(profile),
        ])

    box = ToolBox(tools)
    logger.info("已載入 %d 個工具：%s",
                len(box), "、".join(t.name for t in box))
    return box

10 個工具，剛好在 Day 22 建議的上限。

六、寫幾個測試

Day 29 要驗收，但工具層先自己測一下比較安心：

"""tests/test_tools.py"""

import json
import tempfile
from pathlib import Path

import pytest

from storage import todo_store
from tools.todo_tool import AddTodoTool, CompleteTodoTool, ListTodosTool


@pytest.fixture
def temp_store(monkeypatch):
    """每個測試用獨立的暫存檔，不要動到真實資料。"""
    with tempfile.TemporaryDirectory() as tmpdir:
        path = Path(tmpdir) / "todos.json"
        monkeypatch.setattr(todo_store, "TODO_FILE", path)
        yield path


def test_add_todo(temp_store):
    tool = AddTodoTool()
    result = tool.safe_run({"title": "買牛奶", "priority": "高"})

    assert result["ok"]
    assert "買牛奶" in result["content"]
    assert len(todo_store.load_todos()) == 1


def test_add_empty_title_fails_gracefully(temp_store):
    """空標題應該失敗，但不能拋出例外。"""
    tool = AddTodoTool()
    result = tool.safe_run({"title": "   "})

    assert not result["ok"]
    assert "不能是空的" in result["content"]


def test_add_invalid_date(temp_store):
    tool = AddTodoTool()
    result = tool.safe_run({"title": "測試", "due": "2026-13-45"})

    assert not result["ok"]
    assert "日期格式" in result["content"]


def test_wrong_argument_name(temp_store):
    """模型給錯參數名稱時，要回傳錯誤而不是崩潰。"""
    tool = AddTodoTool()
    result = tool.safe_run({"titel": "打錯字了"})

    assert not result["ok"]
    assert "參數錯誤" in result["content"]


def test_complete_twice(temp_store):
    todo_store.add("測試")
    tool = CompleteTodoTool()

    first = tool.safe_run({"todo_id": 1})
    second = tool.safe_run({"todo_id": 1})

    assert first["ok"] and "已完成" in first["content"]
    assert second["ok"] and "本來就已經完成" in second["content"]


def test_complete_nonexistent(temp_store):
    tool = CompleteTodoTool()
    result = tool.safe_run({"todo_id": 999})

    assert not result["ok"]
    assert "找不到" in result["content"]


def test_list_sorting(temp_store):
    """高優先度應該排在前面。"""
    todo_store.add("低優先", "低")
    todo_store.add("高優先", "高")
    todo_store.add("中優先", "中")

    result = ListTodosTool().safe_run({})
    content = result["content"]

    assert content.index("高優先") < content.index("中優先") < content.index("低優先")

跑起來：

pytest tests/ -v

tests/test_tools.py::test_add_todo PASSED
tests/test_tools.py::test_add_empty_title_fails_gracefully PASSED
tests/test_tools.py::test_add_invalid_date PASSED
tests/test_tools.py::test_wrong_argument_name PASSED
tests/test_tools.py::test_complete_twice PASSED
tests/test_tools.py::test_complete_nonexistent PASSED
tests/test_tools.py::test_list_sorting PASSED

7 passed in 0.14s

**注意這些測試都不需要呼叫 AI。**工具層是純程式，可以完整測試，又快又免費。

這是分層架構的好處——把不確定的部分（AI）隔離在上層，下層全部可測。

最重要的幾個測試是 test_add_empty_title_fails_gracefully 和 test_wrong_argument_name——它們驗證的是 safe_run() 的核心承諾：不管輸入多離譜，都不會拋出例外。

七、今天完成了什麼

✅ storage/todo_store.py     待辦資料存取（含錯誤處理）
✅ tools/todo_tool.py        四個待辦工具
✅ tools/registry.py         工具箱組裝
✅ workflows/daily_brief.py  今日簡報流程
✅ workflows/router.py       意圖路由
✅ tests/test_tools.py       工具層測試

明天要做最後的整合：assistant.py 和 main.py，然後跑完 Day 27 定義的十條驗收標準。

小結

• 資料存取層要處理檔案損壞的情況（備份再重建，不要直接丟掉）
• 在 tool description 裡教模型工具之間的先後關係（「先查日期再換算」）
• complete() 回傳 (物件, 是否改變)，讓工具能給出更精準的回覆
• 今日簡報 Workflow 只呼叫一次 AI，其他步驟全是純程式
• 意圖分類用便宜的模型；分類失敗要能 fallback 到 Agent
• Router 裡的天氣與待辦路徑完全不用 AI，靠 Day 16 的分析函式
• 工具層是純程式，可以完整測試，又快又免費
• 最重要的測試是「離譜的輸入不會讓程式崩潰」

明天：整合、驗收、實際跑一輪。