LLMが正しいツールを選び、引数も正しく生成できても、create_order() を安全に実行できるとは限りません。結果を返す直前に通信が切れ、同じ操作が再送されれば、注文だけが2件作られる可能性があります。
副作用のあるツールでは、モデルの精度より外側にもう一つ境界が必要です。同じ業務操作が複数回届くことを前提に、実行層を冪等にします。
以下は Python 3.12 以上と標準ライブラリの sqlite3 だけで再現できます。追加パッケージはありません。
正しいfunction callでも二重登録できる
まずLLMを外して、問題だけを小さくします。ツール呼び出しを固定JSONとして扱えば、事故の原因がモデルなのか実行層なのかを分離できます。
import json
import sqlite3
con = sqlite3.connect(":memory:")
con.execute("""
CREATE TABLE orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sku TEXT NOT NULL,
quantity INTEGER NOT NULL
)
""")
def create_order(arguments_json: str) -> int:
args = json.loads(arguments_json)
cur = con.execute(
"INSERT INTO orders (sku, quantity) VALUES (?, ?)",
(args["sku"], args["quantity"]),
)
con.commit()
return cur.lastrowid
tool_call = '{"sku":"A-001","quantity":2}'
print(create_order(tool_call))
print(create_order(tool_call))
print(con.execute("SELECT * FROM orders").fetchall())
出力はこうなります。
1
2
[(1, 'A-001', 2), (2, 'A-001', 2)]
同じ引数を2回処理したので、当然2件できます。
ここで「同じ引数なら2回目を無視すればよい」と考えるのも危険です。同じ商品を同じ数量で、本当に2回注文したいケースと区別できません。
必要なのは引数の同一性ではなく、「これは同じ業務操作の再送である」と判断できる識別子です。
HTTPやLLMの呼び出し境界をまたぐ処理では、こちらがレスポンスを受け取れなかったことと、相手側で処理されなかったことは同じではありません。
たとえば注文INSERTが成功した直後、結果をLLM側へ返す前に接続が失われたとします。
tool call
|
v
INSERT order 成功
|
v
通信失敗
|
v
再実行
|
v
INSERT order もう一度成功
呼び出し側から見ると最初の処理結果が分からないため、retry自体は自然です。問題はretryではなく、受け手が「初回」と「再送」を区別できないことです。
call_idと業務上の操作IDは役割が違う
OpenAIのFunction Callingでは、function_call に call_id が含まれ、その結果を function_call_output として返す際にも同じ call_id を指定します。
Programmatic Tool Callingでも、プログラム自身のcall_idと、そこから発生するfunction_callのcall_idは別の項目として扱われます。
ここでAPI上のcall_idを、そのまま注文の冪等性キーにしたくなります。しかし私は、APIプロトコル上の識別子と業務操作の識別子を分けて考えるほうが安全だと考えています。
call_idは「このfunction callの結果をどこへ返すか」を対応付ける識別子です。一方、こちらが欲しいoperation_idは「この注文作成は以前受け付けた注文作成と同一か」を判断するための識別子です。
境界を分けると、構造はこうなります。
LLM / API
|
| call_id
v
tool adapter
|
| operation_id
v
execution record
|
v
business operation
会話を作り直した、ジョブを再投入した、あるいは別の実行経路から同じ業務操作を再開した、といった場合まで同一操作として扱いたいなら、その同一性はアプリケーション側で定義しなければなりません。
引数のハッシュも保存しておくと、さらに一つ事故を防げます。同じoperation_idなのに引数だけ変わった場合、それを再送として成功扱いするのではなく、衝突として検出できます。
executionテーブルを副作用の手前に置く
最小構成として、実行記録を次の状態で持たせます。
pending -> completed
|
+----> failed
今回はSQLiteで、operation_idにUNIQUE制約を持たせます。ポイントは注文テーブルではなく、その手前のexecutionテーブルで操作を識別することです。
import hashlib
import json
import sqlite3
con = sqlite3.connect(":memory:")
con.executescript("""
CREATE TABLE orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sku TEXT NOT NULL,
quantity INTEGER NOT NULL
);
CREATE TABLE executions (
operation_id TEXT PRIMARY KEY,
args_hash TEXT NOT NULL,
status TEXT NOT NULL,
result TEXT
);
""")
def hash_args(args: dict) -> str:
canonical = json.dumps(
args, sort_keys=True, separators=(",", ":")
)
return hashlib.sha256(canonical.encode()).hexdigest()
def create_order_once(operation_id: str, args: dict) -> dict:
digest = hash_args(args)
with con:
row = con.execute(
"""
SELECT args_hash, status, result
FROM executions
WHERE operation_id = ?
""",
(operation_id,),
).fetchone()
if row:
old_hash, status, result = row
if old_hash != digest:
raise ValueError("operation_id reused with different arguments")
if status == "completed":
return json.loads(result)
raise RuntimeError(f"operation already {status}")
con.execute(
"""
INSERT INTO executions
(operation_id, args_hash, status, result)
VALUES (?, ?, 'pending', NULL)
""",
(operation_id, digest),
)
cur = con.execute(
"INSERT INTO orders (sku, quantity) VALUES (?, ?)",
(args["sku"], args["quantity"]),
)
result = {"order_id": cur.lastrowid}
con.execute(
"""
UPDATE executions
SET status = 'completed', result = ?
WHERE operation_id = ?
""",
(json.dumps(result), operation_id),
)
return result
同じ操作を2回渡してみます。
args = {"sku": "A-001", "quantity": 2}
print(create_order_once("op-20261001-001", args))
print(create_order_once("op-20261001-001", args))
print(con.execute("SELECT * FROM orders").fetchall())
print(con.execute(
"SELECT operation_id, status FROM executions"
).fetchall())
結果は次の形になります。
{'order_id': 1}
{'order_id': 1}
[(1, 'A-001', 2)]
[('op-20261001-001', 'completed')]
2回目は注文処理を繰り返さず、保存済みの結果を返します。
さらに、同じ操作IDで内容だけ変更してみます。
create_order_once(
"op-20261001-001",
{"sku": "B-999", "quantity": 5},
)
これはValueErrorになります。単にoperation_idの存在だけを見る実装だと、間違った引数を「処理済み」と誤認するため、args_hashにも意味があります。
ここではexecutionsへのINSERT、注文作成、completedへの更新が一つのSQLiteトランザクションに入っています。途中で例外が発生すればまとめてロールバックされるため、この小さな例ではpendingが残りません。
では、なぜpendingを設計に入れたのか。
実システムでは副作用が同じDB内で完結するとは限らないからです。メール送信、決済API、外部SaaSへの登録などをSQLiteのトランザクションに参加させることはできません。
DBトランザクションだけでは外部副作用を守れない
たとえば次の順序には、避けられない隙間があります。
pendingを保存
|
外部APIで処理成功
|
| ここでプロセス停止
v
completedを保存
再起動後にDBを見るとpendingですが、外部サービスでは処理済みかもしれません。この状態で無条件に再実行すると二重処理になります。
つまりexecutionテーブルを置けば何でもexactly-onceになるわけではありません。
外部サービス自身がidempotency keyを受け付けるなら、同じoperation_idをその境界まで伝播させる設計が扱いやすくなります。受け付けない場合は、処理結果を問い合わせられるか、pendingを人または別のreconciliation処理で確認するかなど、対象サービスに合わせた回復方法が必要です。
failedも同様です。「失敗したから安全に再実行できる」とは限りません。外部への要求が到達する前に失敗したのか、相手では成功した後にタイムアウトしたのかで意味が変わります。
そのため、実運用では単純なfailedだけでなく、エラー種別や外部参照ID、試行回数、更新時刻などを保存したくなるはずです。ただし、それらは対象の副作用が決まってから足せばよく、最初に守る境界は変わりません。
LLMに副作用のあるツールを渡すとき、私はモデル側に「二度呼ばないで」と期待するより、ツール実行層を通常の分散システムと同じように扱うほうがよいと考えています。モデルが正しいJSONを一度生成することと、その業務操作が一度だけ成立することは別問題だからです。
今回のSQLite実装は、同一DBトランザクション内で完結する副作用を理解するための最小例です。複数ワーカーが同時実行する構成では、SELECTしてからINSERTする競合も考える必要があるため、UNIQUE制約による競合を実際に発生させ、DBごとのトランザクション分離レベルと合わせて確認するのが次の境界になります。