Python AI Agent Tool Calling實戰：構建可靠函數呼叫系統的5個核心模式
AI与大数据
函數呼叫為什麼總是出問題？
你給AI Agent接了5個工具，結果模型把城市名傳給了數字參數、在3個工具間反覆橫跳、呼叫超時後直接崩潰、回傳的JSON根本解析不了——Tool Calling的痛點遠比想像中多。2026年，OpenAI、Anthropic、Llama三大平台都提供了原生Tool Calling能力，但「能用」和「可靠」之間隔著5個核心模式。
核心概念速查
| 概念 | 說明 | 關鍵點 |
|---|---|---|
| Tool Calling | 模型主動呼叫外部工具的機制 | 區別於純文字生成 |
| Function Calling | OpenAI的函數呼叫協議 | tools參數 + tool_choice |
| JSON Schema參數 | 用Schema定義工具入參結構 | type/required/enum約束 |
| 工具選擇策略 | auto/required/none及指定工具 | 控制模型何時呼叫工具 |
| 並行呼叫 | 模型一次回傳多個tool_call | 需處理並發執行 |
| 呼叫鏈 | 多步驟工具依賴執行 | 前一步輸出作後一步輸入 |
| 冪等性 | 相同參數重複呼叫結果一致 | 重試安全的基礎 |
| 超時重試 | 呼叫失敗後自動重試 | 指數退避 + 最大次數 |
五大挑戰深度分析
| 挑戰 | 典型表現 | 根因 |
|---|---|---|
| 參數生成錯誤 | 字串傳給integer、必填參數缺失、enum值越界 | 模型對Schema理解偏差 |
| 多工具選擇策略 | 模型選錯工具、該用工具時不用、不該用時強用 | 工具描述不夠精確 |
| 呼叫超時與重試 | 網路抖動導致失敗、重試風暴、雪崩效應 | 缺乏退避策略和熔斷 |
| 結果解析與驗證 | 回傳格式異常、欄位缺失、型別不匹配 | 缺少輸出Schema校驗 |
| 工具權限與安全 | 執行危險操作、注入惡意參數、越權存取 | 缺少權限校驗和沙箱 |
分步實作：5個核心模式
模式1：OpenAI Function Calling基礎整合
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Optional
import json
import os
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
tools = [
{
"type": "function",
"function": {
"name": "get_stock_price",
"description": "獲取指定股票的即時價格資訊",
"parameters": {
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "股票代碼，如 AAPL、GOOGL",
"pattern": "^[A-Z]{1,5}$",
},
"exchange": {
"type": "string",
"description": "交易所",
"enum": ["NASDAQ", "NYSE", "SSE", "HKEX"],
},
},
"required": ["symbol"],
},
},
}
]
class StockResult(BaseModel):
symbol: str
price: float
change: float
exchange: str
def execute_get_stock_price(symbol: str, exchange: str = "NASDAQ") -> dict:
mock_prices = {
"AAPL": {"price": 198.5, "change": 2.3},
"GOOGL": {"price": 175.2, "change": -1.1},
"TSLA": {"price": 245.8, "change": 5.7},
}
data = mock_prices.get(symbol, {"price": 0, "change": 0})
return {"symbol": symbol, "price": data["price"], "change": data["change"], "exchange": exchange}
def basic_tool_calling(user_query: str) -> str:
messages = [
{"role": "system", "content": "你是一個股票分析助手，使用工具獲取即時資料。"},
{"role": "user", "content": user_query},
]
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto",
temperature=0,
)
message = response.choices[0].message
if not message.tool_calls:
return message.content or "無法回答"
tool_call = message.tool_calls[0]
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f"呼叫工具: {func_name}({json.dumps(func_args, ensure_ascii=False)})")
if func_name == "get_stock_price":
result = execute_get_stock_price(**func_args)
else:
result = {"error": f"未知工具: {func_name}"}
validated = StockResult(**result) if "error" not in result else result
messages.append(message.to_dict())
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(validated if isinstance(validated, dict) else validated.model_dump(), ensure_ascii=False),
})
final = client.chat.completions.create(
model="gpt-4o",
messages=messages,
temperature=0,
)
return final.choices[0].message.content
if __name__ == "__main__":
answer = basic_tool_calling("蘋果公司股票現在多少錢？")
print(answer)
模式2：多工具註冊與自動選擇
from typing import Callable
import datetime
class ToolRegistry:
def __init__(self):
self._tools: dict[str, dict] = {}
def register(self, name: str, description: str, parameters: dict, executor: Callable):
self._tools[name] = {
"schema": {
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": parameters,
},
},
"executor": executor,
}
def get_tool_schemas(self) -> list[dict]:
return [t["schema"] for t in self._tools.values()]
def execute(self, name: str, arguments: dict) -> str:
if name not in self._tools:
return json.dumps({"error": f"工具'{name}'不存在", "available": list(self._tools.keys())})
try:
result = self._tools[name]["executor"](**arguments)
return json.dumps(result, ensure_ascii=False, default=str)
except TypeError as e:
return json.dumps({"error": f"參數錯誤: {e}", "tool": name})
except Exception as e:
return json.dumps({"error": f"執行失敗: {e}", "tool": name})
registry = ToolRegistry()
registry.register(
name="get_weather",
description="獲取指定城市的天氣資訊，包括溫度、天氣狀況和濕度",
parameters={
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名稱，如台北、上海"},
"unit": {"type": "string", "description": "溫度單位", "enum": ["celsius", "fahrenheit"]},
},
"required": ["city"],
},
executor=lambda city, unit="celsius": {
"city": city, "temp": 28 if city == "台北" else 32,
"condition": "晴" if city == "台北" else "多雲", "unit": unit,
"updated_at": datetime.datetime.now().isoformat(),
},
)
registry.register(
name="search_web",
description="搜尋網際網路獲取最新資訊，適合查詢新聞、文件、技術資料",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜尋關鍵詞"},
"max_results": {"type": "integer", "description": "最大結果數", "minimum": 1, "maximum": 10},
},
"required": ["query"],
},
executor=lambda query, max_results=3: {
"results": [{"title": f"{query} - 相關結果{i}", "url": f"https://example.com/{i}"} for i in range(max_results)],
},
)
registry.register(
name="calculate",
description="計算數學表達式的值，支援四則運算",
parameters={
"type": "object",
"properties": {
"expression": {"type": "string", "description": "數學表達式，如 2+3*4"},
"precision": {"type": "integer", "description": "小數精度", "default": 2},
},
"required": ["expression"],
},
executor=lambda expression, precision=2: {"result": round(eval(expression, {"__builtins__": {}}, {}), precision), "expression": expression},
)
def multi_tool_agent(user_query: str, max_steps: int = 6) -> str:
messages = [
{"role": "system", "content": "你是一個智慧助手，根據使用者問題選擇合適的工具。每次只呼叫一個工具，仔細分析結果後再決定下一步。"},
{"role": "user", "content": user_query},
]
for step in range(max_steps):
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=registry.get_tool_schemas(),
tool_choice="auto",
temperature=0.1,
)
message = response.choices[0].message
messages.append(message.to_dict())
if not message.tool_calls:
return message.content or "無法生成回答"
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
result = registry.execute(func_name, func_args)
messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": result})
return "達到最大步數限制"
if __name__ == "__main__":
answer = multi_tool_agent("台北天氣怎麼樣？順便幫我算一下 (28+32)*1.5")
print(answer)
模式3：呼叫超時與重試機制
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RetryConfig:
max_attempts: int = 3
base_delay: float = 1.0
max_delay: float = 30.0
timeout_seconds: float = 10.0
def exponential_backoff(attempt: int, base_delay: float = 1.0, max_delay: float = 30.0) -> float:
delay = min(base_delay * (2 ** attempt), max_delay)
return delay
def execute_with_retry(
executor: Callable,
arguments: dict,
config: RetryConfig = RetryConfig(),
) -> str:
last_error = None
for attempt in range(config.max_attempts):
try:
result = executor(**arguments)
parsed = json.dumps(result, ensure_ascii=False, default=str)
result_obj = json.loads(parsed)
if "error" in result_obj:
raise ValueError(result_obj["error"])
return parsed
except Exception as e:
last_error = e
delay = exponential_backoff(attempt, config.base_delay, config.max_delay)
logger.warning(f"第{attempt + 1}次呼叫失敗: {e}，{delay:.1f}s後重試")
time.sleep(delay)
return json.dumps({"error": f"重試{config.max_attempts}次後仍失敗: {last_error}"})
def tool_calling_with_retry(user_query: str) -> str:
messages = [
{"role": "system", "content": "你是一個智慧助手，使用工具回答問題。"},
{"role": "user", "content": user_query},
]
for step in range(6):
try:
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=registry.get_tool_schemas(),
tool_choice="auto",
temperature=0.1,
timeout=30.0,
)
except Exception as e:
logger.error(f"API呼叫異常: {e}")
continue
message = response.choices[0].message
messages.append(message.to_dict())
if not message.tool_calls:
return message.content or "無法回答"
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
result = execute_with_retry(
lambda **kwargs: json.loads(registry.execute(func_name, kwargs)),
func_args,
)
messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": result})
return "達到最大步數"
if __name__ == "__main__":
answer = tool_calling_with_retry("查一下上海的天氣")
print(answer)
模式4：工具結果校驗與錯誤恢復
from pydantic import BaseModel, Field, ValidationError
class WeatherOutput(BaseModel):
city: str
temp: float
condition: str
unit: str = "celsius"
class CalculationOutput(BaseModel):
result: float
expression: str
OUTPUT_SCHEMAS: dict[str, type[BaseModel]] = {
"get_weather": WeatherOutput,
"calculate": CalculationOutput,
}
def validate_and_recover(tool_name: str, raw_result: str) -> dict:
try:
parsed = json.loads(raw_result)
except json.JSONDecodeError:
return {"error": "工具回傳非JSON格式", "raw": raw_result[:200]}
if "error" in parsed:
return parsed
schema_class = OUTPUT_SCHEMAS.get(tool_name)
if not schema_class:
return parsed
try:
validated = schema_class(**parsed)
return validated.model_dump()
except ValidationError as e:
recovery = parsed.copy()
for err in e.errors():
field = err["loc"][0] if err["loc"] else None
if field and field not in recovery:
if err["type"] == "missing":
recovery[field] = None
recovery["_recovered"] = True
recovery["_recovery_note"] = f"欄位{field}缺失，已填充預設值"
return recovery
def resilient_tool_agent(user_query: str) -> str:
messages = [
{"role": "system", "content": "你是一個智慧助手。如果工具回傳錯誤，請分析原因並調整參數重試。"},
{"role": "user", "content": user_query},
]
for step in range(8):
response = client.chat.completions.create(
model="gpt-4o", messages=messages,
tools=registry.get_tool_schemas(), tool_choice="auto", temperature=0.1,
)
message = response.choices[0].message
messages.append(message.to_dict())
if not message.tool_calls:
return message.content or "無法回答"
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
raw_result = registry.execute(func_name, func_args)
validated_result = validate_and_recover(func_name, raw_result)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(validated_result, ensure_ascii=False, default=str),
})
return "達到最大步數"
if __name__ == "__main__":
answer = resilient_tool_agent("台北天氣如何？算一下 100/3")
print(answer)
模式5：生產級Tool Calling框架（含監控）
from dataclasses import dataclass, field
from typing import Any
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class ToolCallMetric:
tool_name: str
start_time: float
end_time: float = 0.0
success: bool = False
error: str = ""
retry_count: int = 0
@dataclass
class AgentRunMetric:
total_steps: int = 0
total_tool_calls: int = 0
total_retries: int = 0
total_duration: float = 0.0
tool_metrics: list[ToolCallMetric] = field(default_factory=list)
errors: list[str] = field(default_factory=list)
class ProductionToolCallingFramework:
def __init__(
self,
tool_registry: ToolRegistry,
model: str = "gpt-4o",
max_steps: int = 10,
max_retries: int = 3,
call_timeout: float = 15.0,
enable_validation: bool = True,
):
self.registry = tool_registry
self.model = model
self.max_steps = max_steps
self.max_retries = max_retries
self.call_timeout = call_timeout
self.enable_validation = enable_validation
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def _execute_tool_safely(self, name: str, arguments: dict) -> tuple[str, ToolCallMetric]:
metric = ToolCallMetric(tool_name=name, start_time=time.time())
last_result = ""
for attempt in range(self.max_retries):
try:
raw = self.registry.execute(name, arguments)
parsed = json.loads(raw)
if "error" in parsed:
metric.retry_count = attempt
if attempt < self.max_retries - 1:
time.sleep(min(1.0 * (2 ** attempt), 30.0))
continue
metric.error = parsed["error"]
last_result = raw
break
if self.enable_validation:
validated = validate_and_recover(name, raw)
last_result = json.dumps(validated, ensure_ascii=False, default=str)
else:
last_result = raw
metric.success = True
break
except Exception as e:
metric.retry_count = attempt + 1
metric.error = str(e)
last_result = json.dumps({"error": str(e), "tool": name})
if attempt < self.max_retries - 1:
time.sleep(min(1.0 * (2 ** attempt), 30.0))
metric.end_time = time.time()
return last_result, metric
def run(self, user_query: str, system_prompt: str = "") -> dict:
run_start = time.time()
metrics = AgentRunMetric()
messages = [
{"role": "system", "content": system_prompt or "你是一個專業AI助手，使用工具回答問題。工具出錯時分析原因並調整參數重試。"},
{"role": "user", "content": user_query},
]
tools = self.registry.get_tool_schemas()
for step in range(self.max_steps):
metrics.total_steps = step + 1
try:
response = self.client.chat.completions.create(
model=self.model, messages=messages, tools=tools,
tool_choice="auto", temperature=0.1, timeout=self.call_timeout,
)
except Exception as e:
metrics.errors.append(f"API異常: {e}")
logger.error(f"Step {step + 1} API異常: {e}")
break
message = response.choices[0].message
messages.append(message.to_dict())
if not message.tool_calls:
break
for tool_call in message.tool_calls:
metrics.total_tool_calls += 1
func_name = tool_call.function.name
try:
func_args = json.loads(tool_call.function.arguments)
except json.JSONDecodeError as e:
metrics.errors.append(f"參數解析失敗: {func_name}")
messages.append({"role": "tool", "tool_call_id": tool_call.id,
"content": json.dumps({"error": f"參數JSON解析失敗: {e}"})})
continue
result, tool_metric = self._execute_tool_safely(func_name, func_args)
metrics.tool_metrics.append(tool_metric)
metrics.total_retries += tool_metric.retry_count
if not tool_metric.success:
metrics.errors.append(f"{func_name}: {tool_metric.error}")
messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": result})
metrics.total_duration = round(time.time() - run_start, 3)
final_content = messages[-1].get("content", "") if messages else ""
return {
"answer": final_content,
"metrics": {
"steps": metrics.total_steps,
"tool_calls": metrics.total_tool_calls,
"retries": metrics.total_retries,
"duration_s": metrics.total_duration,
"errors": metrics.errors,
"tool_details": [
{"tool": m.tool_name, "success": m.success, "retries": m.retry_count,
"duration_ms": round((m.end_time - m.start_time) * 1000, 1) if m.end_time else 0}
for m in metrics.tool_metrics
],
},
}
if __name__ == "__main__":
framework = ProductionToolCallingFramework(registry, max_steps=8, max_retries=2)
result = framework.run("台北天氣如何？搜尋Python最新版本，算一下 (28+32)*1.5")
print(f"回答: {result['answer']}")
print(f"指標: {json.dumps(result['metrics'], ensure_ascii=False, indent=2)}")
避坑指南
坑1：Schema描述太模糊
❌ "description": "獲取資料" → 模型無法判斷何時該用
✅ "description": "獲取指定城市的即時天氣，包括溫度和濕度。適合查詢天氣相關問題" → 精確描述觸發場景
坑2：忽略tool_choice的精細控制
❌ 所有場景都用 tool_choice: "auto" → 模型可能不用工具
✅ 明確需要工具時用 tool_choice: {"type": "function", "function": {"name": "xxx"}}，純對話時用 tool_choice: "none"
坑3：工具回傳裸字串
❌ 工具直接回傳 "28度晴" → 模型難以結構化處理
✅ 統一回傳JSON {"temp": 28, "condition": "晴", "unit": "celsius"} → 結構化結果利於後續推理
坑4：重試不做退避
❌ 失敗後立即重試3次 → 加劇服務端壓力，觸發限流 ✅ 指數退避重試（1s → 2s → 4s）+ 最大延遲上限 → 保護下游服務
坑5：缺少呼叫鏈終止條件
❌ Agent無限迴圈呼叫工具 → Token耗盡、成本失控 ✅ 設定max_steps硬限制 + 偵測重複呼叫 + Token預算控制 → 強制終止異常迴圈
報錯排查
| 序號 | 報錯資訊 | 原因 | 解決方法 |
|---|---|---|---|
| 1 | Invalid function name not found |
模型呼叫了未註冊的工具 | 新增工具白名單校驗，回傳可用工具列表 |
| 2 | JSON decode error in function arguments |
模型生成的參數不是合法JSON | 新增JSON解析容錯，捕獲DecodeError |
| 3 | Rate limit exceeded: 429 |
API呼叫頻率超限 | 實作指數退避重試，降低並發 |
| 4 | Context length exceeded |
對話歷史+工具結果超長 | 截斷歷史訊息，壓縮工具回傳 |
| 5 | Tool execution timeout |
工具執行超時 | 設定timeout參數，非同步執行 |
| 6 | Missing required parameter |
必填參數缺失 | Schema中標註required，Pydantic校驗 |
| 7 | Circular tool call detected |
工具迴圈呼叫 | max_steps限制 + 重複呼叫偵測 |
| 8 | Model refused to use tools |
模型拒絕使用工具 | 檢查tool_choice和system prompt |
| 9 | Unexpected tool output format |
工具回傳格式異常 | 統一JSON回傳 + 輸出Schema校驗 |
| 10 | Token budget exceeded |
Token用量超預算 | 新增token計數和預算熔斷 |
進階最佳化
1. 工具結果快取
對冪等工具的相同參數呼叫做快取，避免重複消耗Token和API呼叫：
import hashlib
class ToolResultCache:
def __init__(self, ttl_seconds: int = 300):
self._cache: dict[str, tuple[float, str]] = {}
self._ttl = ttl_seconds
def _cache_key(self, name: str, args: dict) -> str:
raw = json.dumps({"name": name, "args": args}, sort_keys=True)
return hashlib.sha256(raw.encode()).hexdigest()
def get(self, name: str, args: dict) -> str | None:
key = self._cache_key(name, args)
if key in self._cache:
ts, result = self._cache[key]
if time.time() - ts < self._ttl:
return result
del self._cache[key]
return None
def set(self, name: str, args: dict, result: str):
key = self._cache_key(name, args)
self._cache[key] = (time.time(), result)
2. 並行Tool Calling處理
OpenAI支援一次回傳多個tool_call，需並行執行提升效率：
import concurrent.futures
def handle_parallel_tool_calls(tool_calls: list, registry: ToolRegistry) -> list[dict]:
results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor_pool:
futures = {}
for tool_call in tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
future = executor_pool.submit(registry.execute, func_name, func_args)
futures[future] = tool_call.id
for future in concurrent.futures.as_completed(futures):
tool_call_id = futures[future]
try:
result = future.result(timeout=10.0)
except Exception as e:
result = json.dumps({"error": str(e)})
results.append({"role": "tool", "tool_call_id": tool_call_id, "content": result})
return results
3. 工具權限沙箱
對危險工具做權限控制，防止越權操作：
class ToolPermission:
ALLOWED = "allowed"
CONFIRM_REQUIRED = "confirm_required"
DENIED = "denied"
TOOL_PERMISSIONS = {
"get_weather": ToolPermission.ALLOWED,
"search_web": ToolPermission.ALLOWED,
"query_database": ToolPermission.CONFIRM_REQUIRED,
"execute_command": ToolPermission.DENIED,
}
def check_permission(tool_name: str) -> str:
return TOOL_PERMISSIONS.get(tool_name, ToolPermission.DENIED)
4. 呼叫鏈可觀測性
為每次Agent執行生成完整呼叫鏈路追蹤：
class CallTracer:
def __init__(self):
self.trace_id = hashlib.md5(str(time.time()).encode()).hexdigest()[:12]
self.spans: list[dict] = []
def record(self, tool_name: str, args: dict, result: str, duration_ms: float, success: bool):
self.spans.append({
"trace_id": self.trace_id,
"tool": tool_name,
"args_summary": str(args)[:100],
"result_summary": result[:100],
"duration_ms": duration_ms,
"success": success,
"timestamp": time.time(),
})
def export(self) -> dict:
return {"trace_id": self.trace_id, "span_count": len(self.spans), "spans": self.spans}
對比分析
| 維度 | OpenAI Function Calling | Anthropic Tool Use | Llama Tool Calling |
|---|---|---|---|
| 協議格式 | tools陣列 + tool_choice | tools陣列 + tool_choice | 自訂格式，依賴實作 |
| 參數Schema | JSON Schema完整支援 | JSON Schema + input_schema | 基礎JSON Schema |
| 並行呼叫 | 原生支援多tool_call | 支援多tool_use區塊 | 依賴框架實作 |
| 串流輸出 | 支援streaming tool_call | 支援streaming | 部分框架支援 |
| 強制呼叫 | tool_choice: required | tool_choice: any | 需prompt引導 |
| 指定工具 | tool_choice指定function | tool_choice指定tool | 需prompt指定 |
| 錯誤處理 | 需自行實作 | 需自行實作 | 需自行實作 |
| 快取 | Prompt Caching支援 | Prompt Caching支援 | 無原生支援 |
| 成本 | 較高 | 中等 | 低（自部署） |
| 生態成熟度 | 最成熟 | 成熟 | 快速發展中 |
總結展望：2026年構建可靠的Agent Tool Calling系統，核心在於5個模式層層遞進——從基礎整合到多工具選擇，從超時重試到結果校驗，最終落地為帶監控的生產級框架。關鍵原則：Schema描述要精確、重試要有退避、結果要校驗、呼叫要有上限、權限要管控。隨著MCP協議和Agent Protocol的標準化，Tool Calling將從「手寫整合」走向「協議化互操作」，但可靠性工程的核心模式不會過時。
線上工具推薦
- JSON格式化：/zh-TW/json/format — 格式化Tool Calling的Schema和工具回應JSON
- Hash計算：/zh-TW/encode/hash — 計算工具快取Key和呼叫鏈Trace ID
- Curl轉程式碼：/zh-TW/dev/curl-to-code — 將AI API除錯curl命令轉為Python程式碼
本站提供瀏覽器本地工具，免註冊即可試用 →
#Agent Tool Calling#函数调用#OpenAI Function#工具集成#2026#AI与大数据