Python AI Agent工具呼叫：2026年ReAct模式讓大模型真正學會使用工具

AI与大数据

大模型會「用工具」和不會「用工具」，差距有多大？

GPT-4o能寫程式碼、做分析、回答問題，但它不會查詢即時資料庫、不會呼叫API、不會操作檔案系統——直到你給它「工具」。但給大模型接上工具，並不意味著它就會用。模型經常選錯工具、傳錯參數、陷入迴圈呼叫，甚至編造不存在的工具。

ReAct（Reasoning + Acting）模式透過「思考-行動-觀察」迴圈，讓大模型真正學會推理和工具使用的協調。2026年，這是構建生產級AI Agent的核心範式。

ReAct模式核心概念

概念 說明 對比
Reasoning 模型先推理當前狀態和下一步行動 對比直接呼叫工具
Acting 執行工具呼叫並獲取結果 對比純文字推理
Observation 觀察工具回傳結果並更新推理 對比忽略結果
Tool Definition 用JSON Schema定義工具介面 對比自然語言描述
Function Calling 模型原生工具呼叫能力 對比正則提取
Tool Orchestration 多工具的編排和依賴管理 對比單工具呼叫

ReAct迴圈流程：

使用者問題 → [Thought] 分析問題 → [Action] 選擇工具 → [Observation] 獲取結果
           → [Thought] 繼續推理 → [Action] 選擇工具 → [Observation] 獲取結果
           → [Thought] 資訊充分 → [Answer] 最終回答

問題深入分析：為什麼工具呼叫這麼難？

問題 原因 ReAct解決方案
選錯工具 模型不理解工具邊界 Thought步驟先推理再選擇
參數錯誤 Schema理解偏差 嚴格Schema驗證+重試
迴圈呼叫 缺乏終止條件 最大步數限制+觀察判斷
幻覺工具 模型編造不存在的工具 工具白名單+強制選擇
結果誤讀 忽略工具回傳的關鍵資訊 Observation步驟顯式處理
上下文遺失 長對話遺忘早期資訊 結構化記憶管理

分步實操：從零構建ReAct Agent

第一步：定義工具Schema

from typing import Optional
from pydantic import BaseModel, Field
import json
import datetime


class ToolParameter(BaseModel):
    name: str
    type: str
    description: str
    required: bool = True
    enum: Optional[list[str]] = None


class ToolDefinition(BaseModel):
    name: str
    description: str
    parameters: list[ToolParameter]

    def to_openai_format(self) -> dict:
        properties = {}
        required = []
        for param in self.parameters:
            prop = {"type": param.type, "description": param.description}
            if param.enum:
                prop["enum"] = param.enum
            properties[param.name] = prop
            if param.required:
                required.append(param.name)
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {"type": "object", "properties": properties, "required": required},
            },
        }


class WeatherInput(BaseModel):
    city: str = Field(description="城市名稱")
    unit: str = Field(default="celsius", description="溫度單位")


class DatabaseQueryInput(BaseModel):
    sql: str = Field(description="SQL查詢語句，僅支援SELECT")
    database: str = Field(default="production", description="資料庫名稱")


class CalculatorInput(BaseModel):
    expression: str = Field(description="數學表達式")
    precision: int = Field(default=2, description="小數精度")

第二步：實作工具執行器

import sqlite3
import re
from typing import Callable


class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, dict] = {}

    def register(self, name: str, description: str, parameters: list[ToolParameter], executor: Callable):
        self._tools[name] = {
            "definition": ToolDefinition(name=name, description=description, parameters=parameters),
            "executor": executor,
        }

    def get_openai_tools(self) -> list[dict]:
        return [tool["definition"].to_openai_format() for tool in self._tools.values()]

    def execute(self, name: str, arguments: dict) -> str:
        if name not in self._tools:
            return json.dumps({"error": f"Tool '{name}' not found", "available": list(self._tools.keys())})
        try:
            result = self._tools[name]["executor"](**arguments)
            return json.dumps(result, ensure_ascii=False, default=str)
        except Exception as e:
            return json.dumps({"error": str(e), "tool": name})


def execute_weather(city: str, unit: str = "celsius") -> dict:
    mock_data = {
        "台北": {"temp": 30, "condition": "多雲", "humidity": 75},
        "上海": {"temp": 32, "condition": "晴", "humidity": 72},
    }
    data = mock_data.get(city, {"temp": 25, "condition": "未知", "humidity": 50})
    return {"city": city, **data, "unit": unit, "updated_at": datetime.datetime.now().isoformat()}


def execute_database_query(sql: str, database: str = "production") -> dict:
    forbidden = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE"]
    if any(kw in sql.upper() for kw in forbidden):
        return {"error": "僅支援SELECT查詢", "sql": sql}

    conn = sqlite3.connect(":memory:")
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE IF NOT EXISTS orders (id INTEGER, product TEXT, amount REAL, status TEXT)")
    cursor.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", [
        (1, "機械鍵盤", 599.0, "completed"),
        (2, "4K顯示器", 2999.0, "pending"),
        (3, "無線滑鼠", 199.0, "completed"),
    ])
    conn.commit()

    try:
        cursor.execute(sql)
        columns = [desc[0] for desc in cursor.description]
        rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
        return {"results": rows, "row_count": len(rows), "database": database}
    except Exception as e:
        return {"error": str(e), "sql": sql}
    finally:
        conn.close()


def execute_calculator(expression: str, precision: int = 2) -> dict:
    safe_chars = re.sub(r'[^0-9+\-*/().%\s]', '', expression)
    if safe_chars != expression.strip():
        return {"error": "表達式包含不安全字元", "expression": expression}
    try:
        result = eval(safe_chars, {"__builtins__": {}}, {"abs": abs, "round": round, "pow": pow})
        return {"result": round(result, precision), "expression": expression}
    except Exception as e:
        return {"error": str(e), "expression": expression}


registry = ToolRegistry()
registry.register("get_weather", "獲取指定城市的天氣資訊", [
    ToolParameter(name="city", type="string", description="城市名稱"),
    ToolParameter(name="unit", type="string", description="溫度單位", required=False),
], execute_weather)
registry.register("query_database", "執行SQL查詢", [
    ToolParameter(name="sql", type="string", description="SQL查詢語句"),
    ToolParameter(name="database", type="string", description="資料庫名稱", required=False),
], execute_database_query)
registry.register("calculate", "計算數學表達式", [
    ToolParameter(name="expression", type="string", description="數學表達式"),
    ToolParameter(name="precision", type="integer", description="小數精度", required=False),
], execute_calculator)

第三步：實作ReAct Agent核心

from openai import OpenAI
import os


class ReActAgent:
    def __init__(self, tool_registry: ToolRegistry, model: str = "gpt-4o", max_steps: int = 8):
        self.registry = tool_registry
        self.model = model
        self.max_steps = max_steps
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.conversation_history: list[dict] = []

    def _build_system_prompt(self) -> str:
        return """你是一個智慧助手，能夠使用工具來回答使用者問題。

請按照ReAct模式工作：
1. Thought: 分析當前情況，決定下一步行動
2. Action: 選擇合適的工具並呼叫
3. Observation: 觀察工具回傳結果
4. 重複以上步驟直到獲得足夠資訊
5. Answer: 基於收集的資訊給出最終回答

重要規則：
- 每次只呼叫一個工具
- 仔細閱讀工具回傳結果後再決定下一步
- 如果工具回傳錯誤，嘗試修正參數重試
- 不要編造不存在的工具
- 當資訊足夠時，直接給出最終回答"""

    def run(self, user_message: str) -> str:
        self.conversation_history = [
            {"role": "system", "content": self._build_system_prompt()},
            {"role": "user", "content": user_message},
        ]

        tools = self.registry.get_openai_tools()

        for step in range(self.max_steps):
            response = self.client.chat.completions.create(
                model=self.model,
                messages=self.conversation_history,
                tools=tools,
                tool_choice="auto",
                temperature=0.1,
            )

            message = response.choices[0].message
            self.conversation_history.append(message.to_dict())

            if message.content:
                print(f"Thought: {message.content}")

            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)
                print(f"Action: {func_name}({json.dumps(func_args, ensure_ascii=False)})")

                result = self.registry.execute(func_name, func_args)
                print(f"Observation: {result[:200]}...")

                self.conversation_history.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result,
                })

        return "達到最大步數限制，未能完成推理。"


if __name__ == "__main__":
    agent = ReActAgent(registry)
    answer = agent.run("台北今天天氣怎麼樣？幫我查一下最近完成的訂單總金額。")
    print(f"\n最終回答: {answer}")

完整代碼：帶錯誤恢復的多工具Agent

from openai import OpenAI
from pydantic import BaseModel
from typing import Optional
import json
import os
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class AgentConfig(BaseModel):
    model: str = "gpt-4o"
    max_steps: int = 10
    retry_attempts: int = 2
    retry_delay: float = 1.0
    temperature: float = 0.1


class AgentState(BaseModel):
    step_count: int = 0
    tool_calls_count: int = 0
    errors: list[str] = []
    tool_history: list[dict] = []


class ProductionAgent:
    def __init__(self, tool_registry: ToolRegistry, config: AgentConfig | None = None):
        self.registry = tool_registry
        self.config = config or AgentConfig()
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.state = AgentState()

    def _execute_tool_with_retry(self, name: str, arguments: dict) -> str:
        for attempt in range(self.config.retry_attempts):
            try:
                result = self.registry.execute(name, arguments)
                parsed = json.loads(result)
                if "error" in parsed and attempt < self.config.retry_attempts - 1:
                    logger.warning(f"Tool '{name}' error (attempt {attempt + 1}): {parsed['error']}")
                    time.sleep(self.config.retry_delay)
                    continue
                return result
            except Exception as e:
                if attempt < self.config.retry_attempts - 1:
                    time.sleep(self.config.retry_delay)
                else:
                    return json.dumps({"error": f"Failed after {self.config.retry_attempts} attempts: {e}"})
        return json.dumps({"error": "Unexpected retry loop exit"})

    def _validate_tool_call(self, tool_call) -> Optional[str]:
        if tool_call.function.name not in self.registry._tools:
            return json.dumps({"error": f"工具 '{tool_call.function.name}' 不存在", "available_tools": list(self.registry._tools.keys())})
        try:
            json.loads(tool_call.function.arguments)
        except json.JSONDecodeError as e:
            return json.dumps({"error": f"參數JSON解析失敗: {e}"})
        return None

    def run(self, user_message: str, context: str = "") -> dict:
        start_time = time.time()
        self.state = AgentState()

        messages = [
            {"role": "system", "content": f"你是一個專業AI助手。\n{context}\n按照ReAct模式工作：Thought → Action → Observation → Answer"},
            {"role": "user", "content": user_message},
        ]

        tools = self.registry.get_openai_tools()

        for step in range(self.config.max_steps):
            self.state.step_count = step + 1
            try:
                response = self.client.chat.completions.create(
                    model=self.config.model, messages=messages, tools=tools,
                    tool_choice="auto", temperature=self.config.temperature,
                )
            except Exception as e:
                self.state.errors.append(f"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:
                self.state.tool_calls_count += 1
                validation_error = self._validate_tool_call(tool_call)
                if validation_error:
                    messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": validation_error})
                    continue

                func_args = json.loads(tool_call.function.arguments)
                self.state.tool_history.append({"tool": tool_call.function.name, "args": func_args})
                result = self._execute_tool_with_retry(tool_call.function.name, func_args)
                messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": result})

        return {
            "answer": messages[-1].get("content", ""),
            "steps": self.state.step_count,
            "tool_calls": self.state.tool_calls_count,
            "errors": self.state.errors,
            "tool_history": self.state.tool_history,
            "elapsed_seconds": round(time.time() - start_time, 2),
        }

避坑指南

坑1：模型編造不存在的工具

解決方案：工具白名單驗證，回傳可用工具列表提示模型，在系統提示中明確列出可用工具名稱。

坑2：參數型別不匹配

解決方案：Schema中使用 enum 限制可選值，使用Pydantic模型驗證參數。

坑3：Agent陷入迴圈呼叫

解決方案：設定 max_steps 硬限制（推薦8-10步），在工具歷史中偵測重複呼叫。

坑4：工具回傳結果過長

解決方案：工具執行器中限制回傳條數（如最多20行），對長結果做摘要後再回傳。

坑5：併發工具呼叫順序問題

解決方案：限制每步只呼叫一個工具，在系統提示中明確「逐步執行」。

報錯排查

序號 報錯信息 原因 解決方法
1 Invalid function_call: function name not found 模型呼叫了未定義的工具 添加工具白名單驗證
2 JSON decode error in function arguments 模型生成的參數不是有效JSON 添加JSON解析容錯處理
3 Rate limit exceeded: 429 API呼叫頻率超限 添加指數退避重試
4 Context length exceeded 對話歷史+工具結果超長 截斷歷史或壓縮工具結果
5 Tool execution timeout 工具執行超時 添加超時限制
6 Function argument missing required parameter 缺少必填參數 Schema中設定required並驗證
7 Circular tool call detected 工具迴圈呼叫 添加max_steps和重複偵測
8 Model refused to use tools 模型拒絕使用工具 檢查system prompt和tool_choice
9 Tool returned unexpected format 工具回傳格式異常 統一回傳JSON格式
10 Token usage exceeded budget Token用量超預算 添加token計數和預算控制

進階優化

1. 記憶管理

class ConversationMemory:
    def __init__(self, max_messages: int = 20, summary_threshold: int = 15):
        self.messages: list[dict] = []
        self.max_messages = max_messages
        self.summary_threshold = summary_threshold
        self.summary: str = ""

    def add(self, message: dict):
        self.messages.append(message)
        if len(self.messages) > self.summary_threshold:
            self._compress()

    def _compress(self):
        old = self.messages[:len(self.messages) - 5]
        self.summary += f"\n[歷史摘要] 共{len(old)}條訊息"
        self.messages = self.messages[len(self.messages) - 5:]

    def get_messages(self) -> list[dict]:
        result = []
        if self.summary:
            result.append({"role": "system", "content": f"之前對話的摘要: {self.summary}"})
        result.extend(self.messages)
        return result[-self.max_messages:]

2. 工具編排DSL

class ToolPipeline:
    def __init__(self, steps: list[dict]):
        self.steps = steps

    def execute(self, registry: ToolRegistry, initial_input: dict) -> dict:
        result = initial_input
        for step in self.steps:
            args = {}
            for key, value in step.get("args", {}).items():
                if isinstance(value, str) and value.startswith("$"):
                    args[key] = result.get(value[1:])
                else:
                    args[key] = value
            output = json.loads(registry.execute(step["tool"], args))
            result[step.get("output_as", step["tool"])] = output
        return result

3. Token預算控制

class TokenBudget:
    def __init__(self, max_total: int = 100000, max_per_step: int = 8000):
        self.max_total = max_total
        self.max_per_step = max_per_step
        self.used = 0

    def can_proceed(self, estimated: int = 4000) -> bool:
        return (self.used + estimated) < self.max_total

    def consume(self, tokens: int):
        self.used += tokens

    def remaining(self) -> int:
        return max(0, self.max_total - self.used)

對比分析

維度 ReAct Agent Plan-and-Execute Reflexion AutoGPT
推理方式 逐步推理+行動 先規劃再執行 自我反思修正 自主目標驅動
工具呼叫 每步一個 按計畫批量 帶反思重試 自由呼叫
錯誤恢復 觀察後調整 重新規劃 自我批評修正 試錯
適用場景 多步推理 複雜任務分解 需要高品質輸出 開放式任務
Token消耗 中等 較高 高 很高
可控性 高 中 中 低
實作複雜度 低 中 高 高
穩定性 高 中 中 低

總結：ReAct模式透過「思考-行動-觀察」迴圈，讓大模型從「被動回答」進化為「主動推理+工具使用」。2026年構建生產級AI Agent的關鍵在於：嚴格的工具Schema定義、可靠的重試和錯誤恢復機制、合理的步數和Token預算控制。從單工具Agent開始，逐步引入多工具編排、記憶管理和自我反思，是落地AI Agent的最佳路徑。記住：好的Agent不是讓模型自由發揮，而是透過結構化約束引導模型做出正確決策。

線上工具推薦

本站提供瀏覽器本地工具，免註冊即可試用 →

#Python#AI Agent#Tool Use#Function Calling#ReAct#工具编排#LangChain#智能体