設計一個穩定且具擴充性的 Tool System(工具系統),是將 LLM 轉變為 Agent 的關鍵。
一個標準的 Tool System 不僅是「呼叫 Python 函數」,它需要在 API 設計、Schema 轉換、型別檢查、錯誤處理與安全性之間建立完整的架構。
以下為你設計一個採用 Python 裝飾器模式(Decorator Pattern) 打造的現代化 Tool System。
一個好的 Tool System 需要涵蓋四個核心組件:
這個實作使用 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 格式"})
利用 @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]
透過 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"]
}
}
},
...
]
# 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]: 不支援寫入/刪除相關指令!"}
在正式產品中, Tool System 還有三個關鍵考量:
res2 的範例),不要直接中斷程式。將 {"status": "error", "message": "..."} 丟回給 LLM,模型通常具備閱讀錯誤訊息並在下一輪自動修正參數(如改寫 SQL 語法)的能力。send_email、delete_account),在 dispatch 前加入權限攔截,跳出系統提示要求管理員確認。asyncio 支援或 timeout 限制,避免 Agent 被阻塞在單一無回應的外部工具呼叫中。