iT邦幫忙

2026 iThome 鐵人賽

DAY 22
0
AI Engineering

從 Prompt 到自主決策:用 Python × Agentic Workflow 實作生活助理系列 第 22 篇

工具註冊表:讓 Agent 的能力可以隨時擴充

  • 分享至 

  • xImage
  •  

昨天我們做出了第一個能用工具的 Agent。但它有個問題——每次要加一個新工具,都得改三個地方:

from tools.new_tool import NewTool          # 1. import

tools = [
    DateTimeTool(),
    WeatherTool(),
    NewTool(),                               # 2. 加進 list
]
tools_map = {t.name: t for t in tools}       # 3. 重建 map

三個地方看起來不多,但只要漏掉一個就會出錯,而且錯誤訊息通常不直觀。

今天要做一個工具註冊表(Tool Registry),讓新增工具變成一件事。


一、設計目標

一個好的工具系統應該做到:

  1. 新增工具只需要寫工具本身,不用改其他檔案
  2. 重複的名字要被抓出來,不要默默覆蓋
  3. 可以依情境挑選工具(有些工具不是每次都要給)
  4. 執行時自動記錄(Day 17 的 trace)
  5. 危險的工具要能標記,之後加確認機制

二、自動註冊

用 Python 的 __init_subclass__,讓子類別一被定義就自動登記:

"""tools/base.py — 工具基底與註冊表。"""

import logging
import time
from abc import ABC, abstractmethod

logger = logging.getLogger(__name__)


class ToolError(Exception):
    """工具執行失敗,訊息會回報給模型。"""


class Tool(ABC):
    """所有工具的基底類別。子類別定義時會自動註冊。"""

    # --- 子類別要覆寫的 ---
    name: str = ""
    description: str = ""
    input_schema: dict = {"type": "object", "properties": {}}

    # --- 行為標記 ---
    dangerous: bool = False      # 是否需要使用者確認
    tags: tuple = ()             # 分類標籤,用來篩選

    # --- 註冊表 ---
    _registry: dict = {}

    def __init_subclass__(cls, **kwargs):
        """每當有人 class X(Tool) 時,這個方法會自動被呼叫。"""
        super().__init_subclass__(**kwargs)

        # 抽象的中介類別不註冊
        if getattr(cls, "abstract", False):
            return

        if not cls.name:
            raise TypeError(f"{cls.__name__} 必須設定 name")
        if not cls.description:
            raise TypeError(f"{cls.__name__} 必須設定 description")
        if cls.name in Tool._registry:
            existing = Tool._registry[cls.name].__name__
            raise TypeError(
                f"工具名稱重複:{cls.name} 已經被 {existing} 使用了"
            )

        Tool._registry[cls.name] = cls
        logger.debug("已註冊工具:%s → %s", cls.name, cls.__name__)

    @abstractmethod
    def run(self, **kwargs) -> str:
        """執行工具,回傳給模型閱讀的文字。"""

    def to_schema(self) -> dict:
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": self.input_schema,
        }

    def safe_run(self, arguments: dict) -> dict:
        """執行工具,保證不拋出例外。"""
        start = time.time()
        try:
            content = self.run(**(arguments or {}))
            ok, content = True, str(content)
        except ToolError as e:
            ok, content = False, f"工具執行失敗:{e}"
            logger.warning("工具 %s 失敗:%s", self.name, e)
        except TypeError as e:
            ok, content = False, f"參數錯誤:{e}"
            logger.warning("工具 %s 參數錯誤:%s", self.name, e)
        except Exception as e:
            ok, content = False, f"未預期的錯誤({type(e).__name__}):{e}"
            logger.exception("工具 %s 發生未預期錯誤", self.name)

        duration = time.time() - start
        logger.info("工具 %s 完成:ok=%s,耗時 %.2fs", self.name, ok, duration)

        return {"ok": ok, "content": content, "duration": duration}

    def __repr__(self):
        return f"<Tool {self.name}>"

__init_subclass__ 是 Python 3.6 之後的功能。它在定義子類別的當下就執行——不是建立物件的時候,是 class WeatherTool(Tool): 這行被讀到的時候。

這代表我們可以在這裡做檢查:名字有沒有填、有沒有重複。問題會在 import 時就爆出來,而不是等到執行才發現。


三、ToolBox:管理一組工具

註冊表記錄「有哪些工具類別」,但 Agent 實際要用的是「已經建立好的工具實體」。用一個 ToolBox 管理:

class ToolBox:
    """一組可用的工具。"""

    def __init__(self, tools=None):
        self._tools = {}
        for tool in tools or []:
            self.add(tool)

    def add(self, tool: Tool):
        if tool.name in self._tools:
            raise ValueError(f"工具 {tool.name} 已經在這個 ToolBox 裡了")
        self._tools[tool.name] = tool
        return self

    def get(self, name):
        return self._tools.get(name)

    def schemas(self, exclude_dangerous=False):
        """產生給 API 的工具定義清單。"""
        return [
            t.to_schema()
            for t in self._tools.values()
            if not (exclude_dangerous and t.dangerous)
        ]

    def filter_by_tag(self, *tags):
        """挑出有指定標籤的工具,組成新的 ToolBox。"""
        wanted = set(tags)
        return ToolBox([
            t for t in self._tools.values()
            if wanted & set(t.tags)
        ])

    def execute(self, name, arguments, tracer=None):
        """執行指定的工具,自動記錄 trace。"""
        tool = self.get(name)

        if tool is None:
            available = "、".join(self._tools) or "(沒有可用工具)"
            result = {
                "ok": False,
                "content": f"沒有名為「{name}」的工具。可用的工具有:{available}",
                "duration": 0.0,
            }
        else:
            if tracer:
                tracer.tool_call(name, arguments)
            result = tool.safe_run(arguments)

        if tracer:
            tracer.tool_result(
                name, result["ok"], result["content"], result["duration"]
            )
        return result

    def __len__(self):
        return len(self._tools)

    def __iter__(self):
        return iter(self._tools.values())

    def __repr__(self):
        return f"<ToolBox {len(self._tools)} tools: {', '.join(self._tools)}>"

幾個設計重點:

找不到工具時,告訴模型有哪些可用

f"沒有名為「{name}」的工具。可用的工具有:{available}"

這比單純說「找不到工具」有用太多。模型看到可用清單,就能自己修正。錯誤訊息是給模型的 prompt。

filter_by_tag()

不是每個情境都要給模型全部的工具。工具太多會有兩個問題:

  1. 佔 token(每個工具的 description 都要傳)
  2. 選擇困難(工具越多,選錯的機率越高)

實務上,一次給模型的工具最好控制在 10 個以內。超過的話就要用標籤或情境來篩選。

exclude_dangerous

危險的工具(會刪東西、會發訊息)可以先不給,等使用者確認後再給。


四、改寫工具

有了新基底,工具寫起來更乾淨:

"""tools/weather_tool.py"""

import config
from tools.base import Tool, ToolError
from tools.weather import get_weather, summarize, WeatherError


class WeatherTool(Tool):
    name = "get_weather"
    tags = ("info", "daily")
    description = (
        "查詢台灣某個縣市未來 36 小時的天氣預報,"
        "包含天氣現象、氣溫範圍、降雨機率與舒適度。"
        "當使用者詢問天氣、要不要帶傘、穿什麼衣服,"
        "或在安排戶外行程時,使用這個工具。"
        "只支援台灣的縣市,不支援國外城市或鄉鎮層級。"
    )
    input_schema = {
        "type": "object",
        "properties": {
            "city": {
                "type": "string",
                "description": "台灣的縣市名稱,例如「臺北市」、「高雄市」、「新竹縣」",
            }
        },
        "required": ["city"],
    }

    def __init__(self, api_key=None, default_city=None):
        self.api_key = api_key or config.CWA_API_KEY
        self.default_city = default_city or config.DEFAULT_CITY

    def run(self, city=None):
        try:
            data = get_weather(city or self.default_city, self.api_key)
        except WeatherError as e:
            raise ToolError(str(e)) from e
        return summarize(data)

再加一個危險的工具示範:

"""tools/todo_tool.py(節錄)"""

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):
        todos = load_todos()
        target = next((t for t in todos if t["id"] == todo_id), None)
        if target is None:
            raise ToolError(f"找不到編號 {todo_id} 的待辦事項")
        todos.remove(target)
        save_todos(todos)
        return f"已刪除「{target['title']}」。"

注意 description 裡那句「不要因為事情做完了就刪除(做完請用 complete_todo)」——這是在防止模型誤用。很多 Agent 的問題不是工具寫錯,是模型用錯工具。而解法在 description 裡。


五、自動載入工具模組

最後一塊拼圖:讓 tools/ 資料夾裡的工具自動被載入。

"""tools/__init__.py"""

import importlib
import logging
import pkgutil
from pathlib import Path

from tools.base import Tool, ToolBox, ToolError

logger = logging.getLogger(__name__)


def load_all_tool_modules():
    """匯入 tools/ 底下所有模組,觸發自動註冊。"""
    package_dir = Path(__file__).parent
    for module_info in pkgutil.iter_modules([str(package_dir)]):
        if module_info.name in ("base", "__init__"):
            continue
        importlib.import_module(f"tools.{module_info.name}")


def build_default_toolbox(**init_kwargs):
    """建立包含所有已註冊工具的 ToolBox。"""
    load_all_tool_modules()

    box = ToolBox()
    for name, cls in Tool._registry.items():
        try:
            box.add(cls())
        except TypeError as e:
            logger.warning("工具 %s 無法初始化,已跳過:%s", name, e)
    return box


__all__ = ["Tool", "ToolBox", "ToolError", "build_default_toolbox"]

現在,新增一個工具只要在 tools/ 裡建一個檔案,其他什麼都不用改:

# tools/calculator_tool.py
from tools.base import Tool, ToolError


class CalculatorTool(Tool):
    name = "calculate"
    tags = ("util",)
    description = (
        "計算一個數學算式並回傳結果。"
        "當需要精確的數字運算(加減乘除、百分比、平均)時使用這個工具,"
        "不要自己心算,因為語言模型的算術經常出錯。"
    )
    input_schema = {
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "數學算式,例如「(1200 + 800) * 0.05」",
            }
        },
        "required": ["expression"],
    }

    ALLOWED = set("0123456789+-*/(). ")

    def run(self, expression):
        if not set(expression) <= self.ALLOWED:
            raise ToolError("算式只能包含數字與 + - * / ( ) 符號")
        try:
            result = eval(expression, {"__builtins__": {}}, {})
        except ZeroDivisionError:
            raise ToolError("除數不能是零")
        except Exception as e:
            raise ToolError(f"算式格式錯誤:{e}")
        return f"{expression} = {result}"

⚠️ 這裡用了 eval(),這在一般情況下是危險的。我加了字元白名單和空的 __builtins__ 當防護,但正式環境建議用專門的算式解析函式庫(例如 simpleeval 或 ast.literal_eval 配合自訂求值),不要用 eval。

這點在 Agent 裡尤其重要:工具的輸入來自模型,而模型的輸入可能來自使用者。等於使用者可以間接影響你的程式執行什麼。所有工具都要當成「輸入不可信」來寫。

存檔之後重新執行,它就在了:

box = build_default_toolbox()
print(box)
# <ToolBox 6 tools: get_current_datetime, get_weather, add_todo, list_todos, delete_todo, calculate>

六、重寫 Agent 主迴圈

把所有東西接起來:

"""agent.py — Agent 核心迴圈。"""

import logging

import anthropic

import config
from tools import ToolBox, build_default_toolbox
from tracer import Tracer

logger = logging.getLogger(__name__)

MODEL = "claude-opus-5"
MAX_TURNS = 10


class Agent:
    def __init__(self, system_prompt, toolbox: ToolBox = None, model=MODEL):
        self.client = anthropic.Anthropic(api_key=config.ANTHROPIC_API_KEY)
        self.system = system_prompt
        self.toolbox = toolbox or build_default_toolbox()
        self.model = model

    def run(self, user_input, history=None, max_turns=MAX_TURNS, tracer=None):
        """執行一次完整的任務,回傳最終回答。"""
        tracer = tracer or Tracer()
        tracer.user_input(user_input)

        messages = list(history or [])
        messages.append({"role": "user", "content": user_input})

        schemas = self.toolbox.schemas()

        for turn in range(max_turns):
            tracer.model_call(self.model, len(messages), len(schemas))

            response = self.client.messages.create(
                model=self.model,
                max_tokens=4096,
                system=self.system,
                tools=schemas,
                messages=messages,
            )

            text = self._extract_text(response)
            tracer.model_response(
                response.stop_reason,
                text,
                {
                    "input": response.usage.input_tokens,
                    "output": response.usage.output_tokens,
                },
            )

            # 沒有要呼叫工具 → 這就是最終回答
            if response.stop_reason != "tool_use":
                tracer.final_answer(text)
                messages.append({"role": "assistant", "content": text})
                return text, messages

            # 執行工具
            messages.append({"role": "assistant", "content": response.content})

            results = []
            for block in response.content:
                if block.type != "tool_use":
                    continue
                outcome = self.toolbox.execute(block.name, block.input, tracer)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": outcome["content"],
                    "is_error": not outcome["ok"],
                })

            messages.append({"role": "user", "content": results})

        # 超過上限
        message = f"這個任務我嘗試了 {max_turns} 輪還是沒完成,可能太複雜了。"
        tracer.error("達到最大輪數", {"max_turns": max_turns})
        logger.warning("Agent 達到最大輪數 %d", max_turns)
        return message, messages

    @staticmethod
    def _extract_text(response):
        return "\n".join(
            b.text for b in response.content if b.type == "text"
        ).strip()

用起來:

from logger_setup import setup_logging
from agent import Agent
from prompts import ASSISTANT_SYSTEM

setup_logging()
agent = Agent(ASSISTANT_SYSTEM)

print(f"載入了 {len(agent.toolbox)} 個工具")

answer, history = agent.run("我明天在台北要不要帶傘?")
print(answer)

# 接著問,帶上歷史
answer, history = agent.run("那幫我記一下", history=history)
print(answer)

七、偵測重複呼叫

一個常見的問題:模型陷入「呼叫工具 → 失敗 → 再呼叫同一個工具」的迴圈。

加一個簡單的偵測:

import json
from collections import Counter

# 在 run() 裡面
call_signatures = Counter()

# 執行工具時
signature = f"{block.name}:{json.dumps(block.input, sort_keys=True)}"
call_signatures[signature] += 1

if call_signatures[signature] > 2:
    outcome = {
        "ok": False,
        "content": (
            f"你已經用完全相同的參數呼叫過 {block.name} 兩次了,"
            f"結果不會改變。請改用其他方式,"
            f"或直接告訴使用者目前遇到的困難。"
        ),
        "duration": 0.0,
    }
else:
    outcome = self.toolbox.execute(block.name, block.input, tracer)

注意這個錯誤訊息——它不只說「不要再試了」,還給了下一步的建議(改用其他方式,或告訴使用者)。這樣模型才知道該怎麼繼續。

json.dumps(..., sort_keys=True) 是為了讓 {"a":1,"b":2} 和 {"b":2,"a":1} 產生相同的簽章。


八、我們現在有什麼

life-assistant/
├── config.py            設定管理           (Day 12)
├── logger_setup.py      日誌               (Day 17)
├── tracer.py            決策追蹤           (Day 17)
├── prompts.py           Prompt 模板        (Day 19)
├── agent.py             Agent 核心迴圈     (Day 22) ⭐
└── tools/
    ├── __init__.py      自動載入           (Day 22)
    ├── base.py          Tool / ToolBox     (Day 22) ⭐
    ├── weather.py       氣象署 API         (Day 14)
    ├── weather_tool.py
    ├── datetime_tool.py
    ├── todo_tool.py
    └── calculator_tool.py

一個可以自主使用工具完成任務的程式。距離 Day 1 設定的目標,已經走了大半。

但它還是「被動的」——我們問一句,它答一句。真正的「自主決策」還差了什麼?

接下來四天(Day 23–26)要討論這個問題:從 Workflow 到 Agent 的光譜、控制流、自主性的分界、以及記憶。


小結

  • __init_subclass__ 讓工具在定義時自動註冊,並在 import 階段就檢查錯誤
  • ToolBox 管理一組工具,支援依標籤篩選、排除危險工具
  • 一次給模型的工具建議控制在 10 個以內,太多會佔 token 也容易選錯
  • 找不到工具時,把可用清單告訴模型——錯誤訊息就是 prompt
  • 用 pkgutil + importlib 自動載入模組,新增工具只要建一個檔案
  • 工具的輸入來自模型,要當成不可信的輸入處理
  • 偵測重複呼叫,並在錯誤訊息裡給出下一步建議

明天開始換個層次,討論什麼是 Workflow,以及它跟 Agent 的差別。


上一篇
讓 AI 動手:Tool Use 的原理與第一次實作
下一篇
從單次呼叫到流程:什麼是 Workflow
系列文
從 Prompt 到自主決策:用 Python × Agentic Workflow 實作生活助理 共 24 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言