LLMアプリを業務に組み込むと、「生成までは自動化できるが、実行まで任せてよいのか」という問題にぶつかります。特にメール送信、チケット更新、データ削除、顧客情報の変更などは、LLMの出力をそのまま実行すると影響範囲が大きくなります。
そこで今回は、LLMにすべてを任せるのではなく、危険度に応じて人間の承認を挟む設計を試しました。結論としては、LLMの処理を「提案」と「実行」に分離し、承認状態を明示的に管理する方法が扱いやすく、監査もしやすくなります。
承認フローを状態機械として考える
LLMの提案から承認、実行までの状態遷移を図にすると次の通りです。
最初に、処理の状態を決めます。単に approved: true/false を持たせるより、状態遷移を定義した方が、二重実行や承認漏れを防ぎやすくなります。
今回の最小構成では、次の状態を使います。
-
pending: LLMが実行案を作成し、人間の確認待ち -
approved: 承認済みで、実行可能 -
rejected: 却下済み -
executed: 実行完了 -
failed: 実行失敗
重要なのは、LLMから返された文章を直接実行しないことです。LLMの出力は、実行対象・引数・理由を持つ構造化された「提案」として保存します。
例えば、次のようなデータです。
{
"action": "send_email",
"args": {
"to": "user@example.com",
"subject": "お問い合わせへの回答",
"body": "ご連絡ありがとうございます。"
},
"reason": "問い合わせ内容に対する返信案を作成した"
}
この時点ではまだメールを送信しません。承認後に、サーバー側で許可された処理だけを実行します。
最小のAPIを実装する
手元ではPythonとFastAPIを使って、承認対象をメモリ上に保存する簡単なAPIを作りました。実際の運用ではPostgreSQLなどの永続DBを使いますが、状態遷移の確認には十分です。
python -m venv .venv
.venv\Scripts\activate
pip install fastapi uvicorn pydantic
main.py は次のようにしました。
from enum import Enum
from typing import Any
from uuid import uuid4
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
approvals: dict[str, dict[str, Any]] = {}
class Status(str, Enum):
PENDING = "pending"
APPROVED = "approved"
REJECTED = "rejected"
EXECUTED = "executed"
FAILED = "failed"
class Proposal(BaseModel):
action: str
args: dict[str, Any]
reason: str
class ApprovalRequest(BaseModel):
approved_by: str
def create_llm_proposal(prompt: str) -> Proposal:
# 実際にはLLM APIを呼び出し、構造化出力を検証する
return Proposal(
action="send_email",
args={
"to": "user@example.com",
"subject": "お問い合わせへの回答",
"body": f"お問い合わせありがとうございます。\n内容: {prompt}",
},
reason="問い合わせへの返信案を作成した",
)
def execute_action(proposal: Proposal) -> None:
if proposal.action != "send_email":
raise ValueError("許可されていないアクションです")
required = {"to", "subject", "body"}
if set(proposal.args) != required:
raise ValueError("メール引数が不正です")
# 本番ではここでメールサービスなどを呼び出す
print("send email:", proposal.args["to"])
@app.post("/proposals")
def create_proposal(prompt: str):
proposal = create_llm_proposal(prompt)
proposal_id = str(uuid4())
approvals[proposal_id] = {
"id": proposal_id,
"status": Status.PENDING,
"proposal": proposal.model_dump(),
"created_by": "llm",
"approved_by": None,
}
return approvals[proposal_id]
@app.get("/proposals/{proposal_id}")
def get_proposal(proposal_id: str):
item = approvals.get(proposal_id)
if not item:
raise HTTPException(status_code=404, detail="not found")
return item
@app.post("/proposals/{proposal_id}/approve")
def approve(proposal_id: str, request: ApprovalRequest):
item = approvals.get(proposal_id)
if not item:
raise HTTPException(status_code=404, detail="not found")
if item["status"] != Status.PENDING:
raise HTTPException(status_code=409, detail="承認待ちではありません")
item["status"] = Status.APPROVED
item["approved_by"] = request.approved_by
return item
@app.post("/proposals/{proposal_id}/execute")
def execute(proposal_id: str):
item = approvals.get(proposal_id)
if not item:
raise HTTPException(status_code=404, detail="not found")
if item["status"] != Status.APPROVED:
raise HTTPException(status_code=409, detail="承認済みではありません")
try:
execute_action(Proposal(**item["proposal"]))
item["status"] = Status.EXECUTED
except Exception as exc:
item["status"] = Status.FAILED
raise HTTPException(status_code=500, detail=str(exc))
return item
起動します。
uvicorn main:app --reload
まず提案を作成します。
curl -X POST "http://127.0.0.1:8000/proposals?prompt=注文内容を確認したい"
返却された id を使って内容を確認し、承認します。
curl -X POST \
"http://127.0.0.1:8000/proposals/PROPOSAL_ID/approve" \
-H "Content-Type: application/json" \
-d "{\"approved_by\":\"reviewer@example.com\"}"
最後に実行します。
curl -X POST \
"http://127.0.0.1:8000/proposals/PROPOSAL_ID/execute"
承認前に実行APIを呼ぶと、409 Conflict になります。この制約をアプリケーション側で必ず強制することがポイントです。フロントエンドのボタンを隠すだけでは、APIを直接呼ばれてしまいます。
すべてを承認待ちにしない
人間の確認を入れると安全になりますが、すべての処理を承認待ちにすると、今度は業務が滞ります。そこで、アクションごとにリスク分類を持たせます。
例えば、次のようなルールです。
RISK_LEVEL = {
"summarize": "low",
"create_draft": "low",
"send_email": "medium",
"update_customer": "high",
"delete_data": "critical",
}
def needs_approval(action: str) -> bool:
return RISK_LEVEL.get(action, "critical") in {"medium", "high", "critical"}
低リスクの要約や下書き作成は自動実行し、外部への送信やデータ変更だけ承認対象にします。
ただし、リスク分類はアクション名だけで決めない方がよい場合もあります。送信先が社外か、金額が一定以上か、個人情報を含むか、といった引数も判定に使います。
def needs_approval_for_email(to: str, body: str) -> bool:
external_domain = not to.endswith("@example.com")
contains_sensitive = any(word in body for word in ["住所", "電話番号", "口座"])
return external_domain or contains_sensitive
LLMに「安全そうだから承認不要」と判断させるのではなく、最終的な判定は通常のプログラムで行います。
実運用で追加したい安全策
最小実装を動かした後、特に重要だと感じた点は次の4つです。
1つ目は、承認対象の内容を固定することです。承認後にLLMへ再問い合わせして内容を変えると、レビューしたものと実行するものが一致しません。提案を保存し、承認時点のハッシュ値も記録すると確認しやすくなります。
2つ目は、承認者の権限確認です。誰でも承認できる状態は避け、対象データや金額に応じて承認者を分けます。承認APIには認証・認可を必ず追加します。
3つ目は、二重実行対策です。実行前に approved から executing へ原子的に遷移させ、同じリクエストが並列で来ても一度しか実行されないようにします。外部APIを呼ぶ場合は、冪等性キーも渡します。
4つ目は、期限切れです。承認待ちのまま長期間残る提案には有効期限を設定します。価格や在庫、顧客情報などは時間が経つと内容が変わるため、期限切れ後は再生成する方が安全です。
監査ログには、少なくとも次の情報を残します。
proposal_id
action
proposal_hash
created_at
approved_by
approved_at
executed_at
result
error
まとめ
LLMアプリに人間の承認を組み込むときは、LLMの出力を直接実行せず、「実行提案」として保存するのが基本です。
状態を pending、approved、executed などに分け、API側で状態遷移を制御すると、承認前の実行や二重実行を防げます。また、リスクの低い処理まで一律に止めるのではなく、アクションと引数に応じて承認要否を決めると、実用性とのバランスを取りやすくなります。
まずはメール送信やチケット更新のような1つの操作から始め、承認内容の固定、権限管理、監査ログ、期限切れを順番に追加するのがおすすめです。人間はLLMのすべてを監視するのではなく、影響の大きい判断に集中できます。