iT邦幫忙

2026 iThome 鐵人賽

DAY 8
0
自我挑戰組

AI Agent 從零開始系列 第 8 篇

自己設計第一個 Tool System

  • 分享至 

  • xImage
  •  

設計一個穩定且具擴充性的 Tool System(工具系統),是將 LLM 轉變為 Agent 的關鍵。

一個標準的 Tool System 不僅是「呼叫 Python 函數」,它需要在 API 設計、Schema 轉換、型別檢查、錯誤處理與安全性之間建立完整的架構。

以下為你設計一個採用 Python 裝飾器模式(Decorator Pattern) 打造的現代化 Tool System。


1. Tool System 架構設計

一個好的 Tool System 需要涵蓋四個核心組件:

  1. Tool Definition(聲明):使用 Pydantic 自動將 Python 函數轉換為 LLM 看得懂的 JSON Schema。
  2. Tool Registry(註冊中心):集中管理所有可用的工具,處理名稱對映與調用。
  3. Execution & Error Handling(執行與護欄):攔截參數型別錯誤、執行階段例外,並將 Error 轉化為 LLM 能理解的反饋。
  4. Schema Parser(轉換器):適配不同模型供應商(如 OpenAI、Claude 等)所需的格式。

2. 完整實作程式碼

這個實作使用 pydantic 來做嚴格的型別檢查與 JSON Schema 自動生成(適用於 openai >= 1.0.0)。

import inspect
import json
from typing import Callable, Any, Dict, List
from pydantic import BaseModel, create_model

# ==========================================
# 1. Tool 封裝類別 (Tool Wrapper)
# ==========================================
class Tool:
    def __init__(self, name: str, description: str, func: Callable):
        self.name = name
        self.description = description
        self.func = func
        self.args_schema = self._generate_args_schema(func)

    def _generate_args_schema(self, func: Callable) -> type[BaseModel]:
        """利用 Pydantic 與 inspect 動態提取函數型別與 Docstring 作為 Schema"""
        sig = inspect.signature(func)
        fields = {}
        
        for param_name, param in sig.parameters.items():
            if param.annotation == inspect.Parameter.empty:
                param_type = Any
            else:
                param_type = param.annotation
            
            # 預設值處理
            if param.default == inspect.Parameter.empty:
                fields[param_name] = (param_type, ...)
            else:
                fields[param_name] = (param_type, param.default)

        # 動態建立 Pydantic Model 用於驗證
        return create_model(f"{self.name}Schema", **fields)

    def to_openai_tool(self) -> Dict[str, Any]:
        """轉換為 OpenAI API 相容的 Tool Definition"""
        schema = self.args_schema.model_json_schema()
        
        # 移除 Pydantic 自動生成的 title 屬性以乾淨輸出
        properties = schema.get("properties", {})
        for prop in properties.values():
            prop.pop("title", None)

        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": properties,
                    "required": schema.get("required", [])
                }
            }
        }

    def execute(self, **kwargs) -> str:
        """帶有型別檢查與例外捕獲的安全性執行器"""
        try:
            # 1. 執行參數驗證
            validated_args = self.args_schema(**kwargs)
            # 2. 執行實際函數
            result = self.func(**validated_args.model_dump())
            # 3. 確保輸出永遠為 JSON 字串
            if isinstance(result, str):
                return result
            return json.dumps(result, ensure_ascii=False)
        
        except Exception as e:
            # 將錯誤作為 Observation 回傳給 LLM,讓 LLM 有機會自我修正
            return json.dumps({
                "status": "error",
                "message": f"工具執行失敗 [{type(e).__name__}]: {str(e)}"
            }, ensure_ascii=False)


# ==========================================
# 2. Tool Registry (工具註冊中心)
# ==========================================
class ToolRegistry:
    def __init__(self):
        self._tools: Dict[str, Tool] = {}

    def register(self, description: str, name: str = None):
        """裝飾器:將一般 Python 函數註冊為 Agent 工具"""
        def decorator(func: Callable):
            tool_name = name or func.__name__
            tool = Tool(name=tool_name, description=description, func=func)
            self._tools[tool_name] = tool
            return func
        return decorator

    def get_tool(self, name: str) -> Tool:
        return self._tools.get(name)

    def get_openai_tools(self) -> List[Dict[str, Any]]:
        """匯出所有註冊工具的 OpenAI 格式清單"""
        return [tool.to_openai_tool() for tool in self._tools.values()]

    def dispatch(self, tool_name: str, arguments_json: str) -> str:
        """接收 LLM 傳回的 JSON 參數並派發執行"""
        tool = self.get_tool(tool_name)
        if not tool:
            return json.dumps({"error": f"找不到名為 '{tool_name}' 的工具"})
        
        try:
            args = json.loads(arguments_json) if isinstance(arguments_json, str) else arguments_json
            return tool.execute(**args)
        except json.JSONDecodeError:
            return json.dumps({"error": "LLM 傳入的工具參數非合法 JSON 格式"})


3. 如何使用這個 Tool System

步驟 A:宣告註冊中心並定義工具

利用 @registry.register 裝飾器,只需給予清晰的 description 與 Python 型別標註(Type Hints),系統就會自動推導 Schema:

registry = ToolRegistry()

@registry.register(description="查詢指定地點的當前氣溫與天氣狀態")
def fetch_weather(location: str, unit: str = "celsius") -> dict:
    # 模擬 API 呼叫
    return {"location": location, "temperature": 26, "unit": unit, "condition": "Sunny"}

@registry.register(description="對資料庫進行安全查詢,傳入 SQL 語法與 limit 數量")
def query_database(sql_query: str, limit: int = 10) -> list:
    if "DELETE" in sql_query.upper():
        raise ValueError("不支援寫入/刪除相關指令!")
    return [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}][:limit]

步驟 B:檢視自動生成的 Schema

透過 registry.get_openai_tools() 直接丟給 LLM SDK 使用:

import pprint
pprint.pprint(registry.get_openai_tools())

輸出結果(自動生成規格符合 OpenAI 要求的 JSON):

[
  {
    "type": "function",
    "function": {
      "name": "fetch_weather",
      "description": "查詢指定地點的當前氣溫與天氣狀態",
      "parameters": {
        "type": "object",
        "properties": {
          "location": {"type": "string"},
          "unit": {"type": "string", "default": "celsius"}
        },
        "required": ["location"]
      }
    }
  },
  ...
]

步驟 C:模擬 Agent 分發執行(Dispatch)

# 1. 正常呼叫測試
res1 = registry.dispatch("fetch_weather", '{"location": "Taipei", "unit": "celsius"}')
print("正確執行結果:", res1)
# 輸出: {"location": "Taipei", "temperature": 26, "unit": "celsius", "condition": "Sunny"}

# 2. 測試異常攔截與自我修正回饋 (Self-Correction Capability)
res2 = registry.dispatch("query_database", '{"sql_query": "DELETE FROM users;"}')
print("錯誤攔截結果:", res2)
# 輸出: {"status": "error", "message": "工具執行失敗 [ValueError]: 不支援寫入/刪除相關指令!"}


4. 進階設計考量(生產環境防護網)

在正式產品中, Tool System 還有三個關鍵考量:

  1. 錯誤反饋(Error Feedback Loop):當工具執行失敗時(例如 res2 的範例),不要直接中斷程式。將 {"status": "error", "message": "..."} 丟回給 LLM,模型通常具備閱讀錯誤訊息並在下一輪自動修正參數(如改寫 SQL 語法)的能力。
  2. 人機協同(Human-in-the-Loop, HITL):針對具備破壞性或敏感權限的工具(例如:send_email、delete_account),在 dispatch 前加入權限攔截,跳出系統提示要求管理員確認。
  3. 超時與併發控制(Timeout & Async):耗時的工具 API 需要加上 asyncio 支援或 timeout 限制,避免 Agent 被阻塞在單一無回應的外部工具呼叫中。

上一篇
Tool Calling 到底是什麼?
下一篇
讓 Agent 學會使用多個 Tools
系列文
AI Agent 從零開始 共 25 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言