跳到主要內容

【2026 最新】Code-Implemented Tool Calls 是什麼?動手做可暫停、可恢復的 Agent 執行器

最後更新: ·
Code-Implemented Tool Calls 教學:可暫停、可恢復與 deterministic replay 的 Agent 執行器

Code-Implemented Tool Calls 是什麼?傳統 Agent 每拿到一個工具結果,通常要再請模型決定下一步;Code-Implemented Tool Calls 則讓模型先產生一段小程式,交給受限 executor 執行,只有碰到必須由使用者裝置或外部 client 處理的工具時才暫停。client 把結果送回後,executor 恢復流程,最後只把整理過的結果交回模型。

這個名字來自 Mistral AI 在 2026 年 6 月 30 日取得的美國專利 US 12,670,045 B1。8 月 10 日,專利被帶上 Hacker News;2026 年 8 月 12 日 01:39(台北時間)下載的 API 快照是 225 points、188 個 comment-tree descendants。不過我逐則掃過討論後,爭點主要是軟體專利與 prior art,不是 token 效能共識。因此本文不拿專利或留言當性能證明,而是自己做一個可重跑的 Python 實驗。

你會完成一個 119 行、只用標準函式庫的受限 AST evaluator:只允許白名單函式,遇到 pending client call 就 pause;拿到結果後從程式第一個 statement 重播 journal,再 resume 到下一個暫停點或 final result。先說清楚邊界:這是依公開機制獨立重建的教學 toy,沒有使用 Mistral 原始碼,也不代表其內部或產品實作。專利公開不是原始碼、產品認證或開源授權。

先說結論:Code-Implemented Tool Calls 是「程式+broker+ledger+replay」

🧠 記憶把手:可暫停工具程式 = 模型產生的控制流+只開白名單的 broker+工具結果 ledger+從頭 deterministic replay。
程式負責流程,broker 掌握真正權限,ledger 保存已發生事件;replay 只是依同一份紀錄重建結果,不會讓外部世界自動回到過去。

用日常比喻:broker 是有警衛的工具櫃台,ledger 是逐筆蓋章的收據簿,replay 則是從第一頁照收據重走。舊收據吻合就直接讀結果;走到收據簿還沒有的新工具,才停下來向 client 要答案。

若你讀過《AI Agent Harness 是什麼》,可以把它理解成 harness 裡的一個新執行層。模型不再每一步都輸出一個 JSON function call,而是先交出有順序、分支或批次的控制流。這也比《Stateless MCP》多處理一個問題:session 不留在 sandbox 時,怎麼靠外部事件紀錄恢復尚未跑完的程式。

Code-Implemented Tool Calls 與傳統 JSON function-call loop 的模型採樣流程比較
三個彼此相依的步驟,傳統 loop 需要模型逐步決策;程式 loop 把可預測的控制流留在 executor。client pause、工具 RPC 與模型 round trip 必須分開計算。

2026 現況:專利不是唯一實作,各家的 resume 也不相同

「讓模型用程式呼叫工具」並非 2026 年才出現。ICML 2024 的 CodeAct 論文已研究 executable code action。Mistral 專利比較具體的組合,是 pending client call、無狀態 sandbox、從程式開頭 replay,以及把時間、亂數等非決定性結果記錄後重用。這不等於 CodeAct 或其他前例已被法院認定涵蓋所有 claims,也不能據此判定專利有效或無效。

  • Mistral:截至 2026 年 8 月 12 日,其一般 function-calling 指南仍示範 developer 執行工具、把 role: "tool" 結果送回模型,再遞迴下一步。另一方面,Public Preview 的 Mistral Workflows已用 Temporal 提供 event history、crash recovery 與 signal pause/resume;其determinism 文件說明 worker 會從頭重跑 developer 撰寫的 workflow,依 history 跳過已完成 activity。這是高度相鄰的正式產品,但文件描述的不是「LLM 每次先產生 code block+pending client tool」,所以本文不把它當成該專利的 reference implementation。
  • OpenAI:Programmatic Tool Calling讓模型產生 JavaScript,在隔離的 V8 執行;client-owned tool 可多次 pause,app 用原 call_idcaller 回傳結果。文件提到 program fingerprint 用於 resume/replay,但沒有說一定採用 Mistral claims 的 from-start replay 或時間/亂數捕捉。
  • Anthropic:Programmatic Tool Calling在 code execution container 跑 Python;工具呼叫時暫停,client 回傳後延續同一 container,並讓中間資料留在 execution environment。這較像有狀態 continuation,不要和本文的 from-start replay 混稱。
  • Cloudflare/MCP:Cloudflare 標為 experimental 的 Code Mode用 typed tools、sandbox 與程式內的 loop/filter;MCP client best practices則要求 host broker 保管 credentials、逐 call 授權、限制網路與資源。它們說明這已是一個產品家族,不是一份專利的同義詞。

Code-Implemented Tool Calls 如何減少模型往返?先把三種 round trip 分開

假設 A 的結果決定 B 的參數,B 又決定 C。傳統 loop 會是「模型①選 A → A → 模型②選 B → B → 模型③選 C → C → 模型④回答」。若控制流可以預先寫清楚,程式模式可變成「模型①產生程式 → executor 跑 A/B/C → 模型②讀 final result」。它主要減少的是模型重新採樣,不是宣稱所有網路請求都消失。

  • Model round trip:是否要再次呼叫 LLM。可預測的三步串行例,理論上可由 4 次降為 2 次。
  • Host ↔ client pause:client-side tool 仍要把請求與結果送來回;三個相依 call 可能仍是三次。
  • Downstream tool RPC:資料庫、支付、搜尋或內部 API 仍各自有延遲與費用。

若 A、B、C 彼此獨立,傳統 function calling 本來就能在同一回合吐出多個 calls;此時程式模式不一定減少模型回合,價值可能只剩本地篩選、聚合與少傳中間資料。若每一步都需要模型看語意後重新判斷,更不該硬塞進固定程式。判斷 Agent 邊界的完整方法,可搭配《動手做 AI Agent Harness》一起看。

動手做:119 行 allowlisted Python executor

下面版本只用 Python 3.9 以上標準函式庫。它沒有呼叫 evalexec,而是用 ast.parse 後自己走訪少量節點。允許的語法只有單一變數 assignment、JSON-like literals、list/dict、subscript、白名單 local function,以及 assignment 右側的 client_call();最後一行必須指定 result

只要電腦有 Python 3.9 以上、沒寫過 Agent runtime 也能跟做。先確認版本,再建立資料夾並開啟檔案:

python3 --version
mkdir code-tool-lab
cd code-tool-lab
nano executor.py

貼上完整程式:

"""Minimal code-tool loop. AST policy narrows syntax; it is NOT a sandbox.

Independent teaching code, not Mistral code or a reconstruction of its internals.
Each resume starts again at statement zero and substitutes a durable event prefix.
"""
import ast, hashlib, json
from dataclasses import dataclass


def clone(x):
    return json.loads(json.dumps(x, sort_keys=True, separators=(",", ":")))


def digest(x):
    return hashlib.sha256(json.dumps(x, sort_keys=True).encode()).hexdigest()[:20]


@dataclass
class Pending:
    call_id: str
    name: str
    args: list
    journal: list


@dataclass
class Done:
    result: object
    journal: list


def run(source, local_tools, client_tools, journal=None, returned=None):
    """Run fresh, replay journal, inject returned, then pause or finish."""
    tree = ast.parse(source)
    replaying, old, new = journal is not None, clone(journal or []), clone(returned or {})
    if len(source) > 8000 or sum(1 for _ in ast.walk(tree)) > 200:
        raise ValueError("program too large")

    def check_expr(node):
        if isinstance(node, ast.Constant): return
        if isinstance(node, ast.Name): return
        if isinstance(node, ast.List):
            for item in node.elts: check_expr(item)
            return
        if isinstance(node, ast.Dict):
            for key, value in zip(node.keys, node.values): check_expr(key); check_expr(value)
            return
        if isinstance(node, ast.Subscript): check_expr(node.value); check_expr(node.slice); return
        if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
            if node.func.id not in local_tools or node.keywords: raise ValueError("call denied")
            for arg in node.args: check_expr(arg)
            return
        raise ValueError(f"syntax denied: {type(node).__name__}")

    for stmt in tree.body:
        if not isinstance(stmt, ast.Assign) or len(stmt.targets) != 1 or not isinstance(stmt.targets[0], ast.Name):
            raise ValueError("only x = expression is allowed")
        call = stmt.value
        if isinstance(call, ast.Call) and isinstance(call.func, ast.Name) and call.func.id == "client_call":
            if not call.args or not isinstance(call.args[0], ast.Constant): raise ValueError("literal client name required")
            if call.args[0].value not in client_tools or call.keywords: raise ValueError("client call denied")
            for arg in call.args[1:]: check_expr(arg)
        else: check_expr(stmt.value)
    if not tree.body or tree.body[-1].targets[0].id != "result":
        raise ValueError("last assignment must be result")

    env, out, cursor, crossed, missing = {}, [], 0, False, object()

    def prior_or_none(request):
        nonlocal cursor
        if cursor == len(old): return missing
        event = old[cursor]
        if any(event.get(k) != v for k, v in request.items()): raise ValueError("journal diverged")
        cursor += 1; out.append(clone(event)); return clone(event["result"])

    def value(node, pc):
        if isinstance(node, ast.Constant): return clone(node.value)
        if isinstance(node, ast.Name): return clone(env[node.id])
        if isinstance(node, ast.List): return [value(x, pc) for x in node.elts]
        if isinstance(node, ast.Dict): return {value(k, pc): value(v, pc) for k, v in zip(node.keys, node.values)}
        if isinstance(node, ast.Subscript): return clone(value(node.value, pc)[value(node.slice, pc)])
        name, args = node.func.id, [value(arg, pc) for arg in node.args]
        request = {"kind": "local", "pc": pc, "name": name, "args": args}
        prior = prior_or_none(request)
        if prior is not missing: return prior
        if replaying and not crossed: raise ValueError("journal ended before client boundary")
        result = clone(local_tools[name](*clone(args)))
        out.append({**request, "result": result}); return clone(result)

    for pc, stmt in enumerate(tree.body):
        target, rhs = stmt.targets[0].id, stmt.value
        is_client = isinstance(rhs, ast.Call) and isinstance(rhs.func, ast.Name) and rhs.func.id == "client_call"
        if not is_client: env[target] = value(rhs, pc); continue
        name, args = rhs.args[0].value, [value(arg, pc) for arg in rhs.args[1:]]
        request = {"kind": "client", "pc": pc, "name": name, "args": args}
        call_id = digest({"source": source, **request})
        request["call_id"] = call_id
        prior = prior_or_none(request)
        if prior is not missing: env[target] = prior; continue
        if new:
            if set(new) != {call_id}: raise ValueError("returned call_id does not match boundary")
            result = clone(new.pop(call_id)); out.append({**request, "result": result})
            env[target], crossed = result, True; continue
        if cursor != len(old): raise ValueError("unused journal events")
        return Pending(call_id, name, args, out)
    if cursor != len(old) or new: raise ValueError("unused replay input")
    return Done(clone(env["result"]), out)


if __name__ == "__main__":
    calls = {"upper": 0, "next_key": 0, "combine": 0}
    def upper(x): calls["upper"] += 1; return x.strip().upper()
    def next_key(x): calls["next_key"] += 1; return x["next"]
    def combine(a, b): calls["combine"] += 1; return [a["key"], b["key"]]
    tools = {"upper": upper, "next_key": next_key, "combine": combine}
    code = 'key = upper(" a ")\na = client_call("lookup", key)\nkey2 = next_key(a)\nb = client_call("lookup", key2)\nresult = combine(a, b)\n'
    one = run(code, tools, {"lookup"}); print("pause 1", one.call_id, calls)
    two = run(code, tools, {"lookup"}, one.journal, {one.call_id: {"key":"A","next":"B"}}); print("pause 2", two.call_id, calls)
    three = run(code, tools, {"lookup"}, two.journal, {two.call_id: {"key":"B","next":"-"}}); print("done", three.result, calls, "events", len(three.journal))

儲存後執行:

python3 -m py_compile executor.py
python3 executor.py

本文實際得到:

pause 1 4127904df82e316bb50f {'upper': 1, 'next_key': 0, 'combine': 0}
pause 2 2bcf8ff4da65a1a7dc41 {'upper': 1, 'next_key': 1, 'combine': 0}
done ['A', 'B'] {'upper': 1, 'next_key': 1, 'combine': 1} events 5

兩次 resume 都重新呼叫 run()、從程式第一個 statement 開始,但 upper/next_key/combine 最後各只真的執行一次:先前事件由 journal 代入,沒有再碰 live tool。第一個 Pending 的 journal 已含 upper 結果;client 回傳 {call_id: result} 後,第二輪先核對並重播它,再走到第二個未知 call。

這個 call_id 只適合 demo 的事件邊界比對:兩次獨立執行相同 source 與參數會得到相同 ID。production identity 還要加入持久化的 tenant/workflow run_id 與授權範圍,不能直接拿 toy ID 當寫入型工具的 idempotency key。

要重跑下方的 batch、5,000 筆 fixture、resource limits 與 12 個測試,可下載完整實驗包(ZIP);SHA-256:f0aa0cb17e4ad74e941b1f6c9551e5886f9bb145f859312371922fa0a2ee3bc8。壓縮檔內含最小版、完整 executor、tests、benchmark、固定輸出、README 與安全清單。

Code-Implemented Tool Calls 從 pending client call 到 deterministic replay 的事件時間線
這個 toy 的 resume 是重新執行,不是把 Python thread 凍結後接著跑;每筆舊事件都要和 PC、工具名、參數與 call ID 對上。

Deterministic replay 做對的關鍵:身份、順序與副作用

專利 claims 4–5 描述從程式開頭 replay evaluation stack,並捕捉、重播非決定性操作的結果;時間與亂數是 detailed description 提供的例子,不是 claims 4–5 的逐字列舉。上面 toy 把每個事件的 kind/pc/name/args 依序附加到 journal,client call 再加 call_id;這些 identity fields 任一不符就中止。教學版的 journal 只是可變的記憶體 list,假設 trusted host 會把它持久化成 append-only 紀錄,而且不驗證 recorded result 的真偽

不過「同一紀錄可得到同一結果」不等於 exactly-once。假設 charge_card() 已扣款,服務卻在 journal commit 前當機,重播仍可能再扣一次。寫入型工具至少要有 durable idempotency key、transaction/outbox 或補償流程;做不到就先只允許 read-only tools。完整測試器中的 unkeyed result hash 只能發現損壞,不是 MAC、簽章或 authenticity 驗證。

實測串行、batch pending 與大量中間資料:不要只看 token 百分比

我把最小版擴成含 client_parallel()、JSON byte limit、step limit、timeout 檢查與 result hash 的完整測試器,固定假資料後跑過 12 個單元測試。這是 protocol accounting experiment,不是模型、網路或供應商 benchmark;測量方法與原始輸入固定,才有資格比較:

  1. 串行/batch pending:同樣三個 quote calls,串行是 3 個 pause/resume cycles、batch widths [1,1,1];batch 版一次交出三個 pending requests,是 1 cycle、batch width [3]。這只量 orchestration;是否真的並行與 latency 改善,由 host scheduler 決定。
  2. 大量中間資料:產生 5,000 筆假訂單,篩到 500 筆再聚合。在預先聲明的 accounting envelope 裡,傳統側計入每個 local tool result 一次,model-visible JSON 是 1,036,451 bytes;程式側只計程式來源加 final result,是 278 bytes。以透明但粗糙的 byte/4 換算為 259,113 vs 70,不是完整 context、tokenizer、帳單或真實 token 用量
  3. Resume ID 驗證:故意送錯 call ID 會 fail closed;保留原 journal、改送正確結果後可完成。最終 5 個 events 的 strict replay 得到相同結果,先前 live tools 重呼叫 0 次;這沒有測 failed-tool retry。
  4. 隱藏成本:為了完整 replay,event log 反而是 2,073,045 bytes。模型 context 變小,不代表 storage、sandbox startup、client RPC 或 tool 費用一起消失。
Code-Implemented Tool Calls toy executor 的串行 batch pending、模型可見 bytes 與 replay 儲存實測
這組數字只回答固定 fixture 的資料搬運問題。真正上線還要同時記錄 correctness、模型採樣次數、client pauses、tool RPC、retries、latency 與 billed tokens。

因此,比起追一個漂亮的「省 99.973%」數字,更重要的是先定義資料邊界。這個 fixture 剛好有大量可在本地丟棄的 rows,所以差距極大;只有一兩個小工具結果時,程式本身與 sandbox 反而可能增加成本。若你真正想省的是 prompt、schema 或輸出 token,可再對照《Claude 怎麼省 Token》,不要把不同成本混成一個百分比。

從 toy 到 production:AST allowlist 絕對不是安全 sandbox

Python 官方甚至提醒,連 ast.literal_eval() 都可能被小型惡意輸入拖垮記憶體、C stack 或 CPU。本文自己解譯 AST,確實縮小了語法與 ambient authority,卻仍在同一個 Python process 呼叫 host function;白名單函式本身若能讀檔、連網或洩漏 secret,攻擊面依然存在。

  • 真正隔離:generated code 放進可 kill 的獨立 process、VM 或 container;rootless、read-only filesystem,預設無網路。搭配 CPU/memory/PID 限制與 wall timeout。seccomp 只能縮小 kernel surface,Linux 文件也明說它本身不是完整 sandbox
  • 語法與 API 雙白名單:拒絕 import、attribute/dunder、reflection、eval/exec/openos/sys/subprocess/socket/ctypes/pickle、thread/process 與任意網路。每個工具再驗參數 schema、使用者身分、資源範圍與 action authorization。
  • Secret isolation:API key、cookie、資料庫密碼只留在 host-side broker。生成程式只看到 typed proxy 與必要參數,不可讀 environment、logs 或 credential store。
  • 有界輸入輸出:限制 source bytes、AST nodes、steps、call count、fan-out、巢狀深度、單筆 result、累積 state/journal 大小;逾時要由外部 supervisor 強制終止,不能只在函式返回後才檢查。
  • 可恢復寫入:pause 邊界用 durable commit 與 compare-and-swap;外部寫入帶 idempotency key。log 可能複製個資與 secrets,要 redaction、加密、TTL、存取稽核與大型 blob reference。

這也是為什麼 production 架構應把 code runtime 和 tool broker 分開:讓 sandbox 不持有 credential、沒有直接網路,只能提出一個結構化 call;broker 再做驗證、授權、執行與輸出過濾。想看更多 harness 的工程分層,可延伸《Hermes Agent 完整解析》與《Context Engineering 是什麼》。

什麼工作流值得用?什麼情況直接 function calling 更好?

適合:控制流可預測、會 fan-out、多個結果需要本地 filter/join/aggregate、中間資料很大,或你需要把 pause/replay 做成可稽核狀態機。例如「查 50 間門市庫存,只回傳低於門檻的 4 間」就很合適。

直接 tool calling 較好:只有一個工具、每步都需要模型讀語意重新判斷、需要模型原生 citations/artifact、或每個動作都要單獨人工批准。程式化不是較高級的預設答案;它是把已知控制流從模型 context 搬到可限制、可測試的執行層。

Code-Implemented Tool Calls FAQ

Mistral 取得專利,代表功能已正式上線嗎?

不代表。專利說明一組 claims 與可能實施方式,不是 release note、reference implementation 或效能驗證。本文只能確認公開專利與現行文件各自寫了什麼。

這和一般 Code Interpreter 有什麼不同?

重點是外部工具邊界。一般 Code Interpreter 多半在受控環境內算資料;這個模式還讓程式透過 broker 呼叫 client-owned tools,遇到外部權限或資料才 pause,再把結果代回控制流。

一定要用 Python 嗎?

不用。OpenAI 文件示範隔離 V8 的 JavaScript,Cloudflare 走 TypeScript/JavaScript;語言不是核心,權限 broker、可驗證事件與 resume semantics 才是。

每次 client pause 都會再呼叫模型嗎?

不一定,而且本文 toy 不會。client 結果直接交回 executor,只有程式完成或需要模型修錯時才回模型。實際 API 要依供應商協定計算,不能把 pause 次數直接當模型 round trips。

Stateless replay 是不是完全不保存狀態?

不是。至少要由某一方保存可恢復資料。本文 toy 由 sandbox 外的 trusted host 保存 source、journal 與工具結果;供應商也可能把 opaque replay state 留在服務端,而且不是每種實作都會捕捉非決定性值。

Deterministic replay 能保證工具只執行一次嗎?

不能。它能避免已記錄結果被再次呼叫;若外部副作用完成、紀錄卻未 commit,仍有重複風險。寫入工具需要 idempotency key 或交易式流程。

AST allowlist 能直接拿去執行不可信程式嗎?

不能。它只是教學用語法 policy。production 還要獨立 OS boundary、無網路預設、資源上限、可強制 kill 的 timeout,以及 broker 端逐 call 授權。

Code-Implemented Tool Calls 一定比較省 token 嗎?

不一定。中間資料越大、可本地壓縮的步驟越多,越可能有利;少量小結果或每步需要模型判斷時,程式與 runtime overhead 可能抵銷收益。要用同一任務實測 billed tokens、correctness、latency 與 retries。

給新手的七個重點

  1. Code-Implemented Tool Calls 把可預測控制流從模型回合搬進 executor。
  2. 模型 round trip、client pause 與 tool RPC 要分開量。
  3. pending call 用穩定 ID;resume 必須精確對應同一事件邊界。
  4. from-start replay 靠 journal 代入舊結果,不是凍結 thread。
  5. 減少 model-visible context 不代表 storage 與網路成本消失。
  6. replay 不提供 exactly-once;寫入型工具一定要做冪等。
  7. AST allowlist 是 policy,不是 production sandbox。

接著閱讀

左右滑動查看更多推薦

結語:先把一條串行流程做成可重播的帳本

今天先不要接支付、寄信或資料刪除。複製 119 行 toy,把 lookup 換成一個無副作用的假工具,確認兩次 pause 都能用同一份 journal 從頭恢復;接著故意改動一個參數或 call ID,確定系統會拒絕 divergent replay,而不是默默算出另一個答案。

通過後,再加入三個彼此獨立的 calls,量 serial 與 batch pending 的 pause cycles;最後放入一批大資料,只把聚合結果送回模型。若結果真的更正確、更省 context,而且 resume 與 failed-tool retry 都已各自驗證,才值得升級成獨立 worker 與 host broker。回到開頭的記憶把手:控制流、broker、ledger、replay 四件事同時成立,才是一個可暫停、可恢復的工具程式。想把它接成完整專案,可到 AlphaLab《線上課程》與《AI 專區》繼續實作。

ALPHALAB 社群

有問題?來 Telegram 聊

和 Terry、編輯、其他網友一起討論這篇文章。提問、分享觀點,回覆更即時。

加入 Telegram 討論

📩 訂閱 AlphaLab 電子報

每週一封,第一時間收到新文章與投資觀察。

我們不會 spam,隨時可退訂。