エージェントを1体から3体に増やせば、3倍速くなるはず。ところが実際には、やり直しが増えて1体のときより遅くなる。
精度の問題に見えますが、違います。1体のときに暗黙で効いていたブレーキを、台数を増やすときに外しているだけです。
この記事では、エージェントを増やす前に実装しておく「止める条件」を3つ、Python の最小実装つきで書きます。
起きていることは3つに絞れる
1. やり直しが連鎖する
エージェントAの出力を、エージェントBが入力として受け取ります。Aの出力が7割の出来でも、Bは疑いません。疑う条件を渡していないからです。
7割 × 7割 ≒ 5割。3段重ねると 0.7³ ≒ 0.34 で、3割台まで落ちます。人間が気づくのは最後で、そこから最初に戻ります。
2. 途中の判断を誰も見ていない
1体のときは、出力を人間が毎回見ていました。おかしければ止めていた。
複数にした瞬間、人間が見る場所は「最後」だけになります。エージェント同士は互いの出力を検証しません。検証させる設計をしていないからです。
3. ログはあるのに、原因が特定できない
残しているのが出力だけで、判断の理由が無いからです。出力から読み取れるのは、間違っていたという事実だけです。どこで間違えたかは、出力には残っていません。
なぜ起きるのか
1体で動かしていたとき、ブレーキは人間でした。設計上ブレーキを置いていなかったのではなく、人間が毎回見ていたから、要らなかっただけです。
エージェントを増やすというのは、そのブレーキを外して台数を増やすことです。だから増やすほど壊れます。
先に実装する3つの停止条件
順番が大事です。台数を増やす前に入れます。増やした後に足すと、既存の流れを壊します。
| # | 条件 | 止めたあと |
|---|---|---|
| ① | 確信度の下限を割った | やり直す |
| ② | やり直しが上限（2回）に達した | 人間に返す |
| ③ | 取り返しのつかない操作だった | 最初から人間に返す |
③に入れるのは、間違えたときに取り返しがつかない操作です。
- 外部に何かを送る（メール、投稿、通知）
- 何かを消す
- お金が動く
- 誰かの予定や権限を変える
この4つは、エージェントの精度がどれだけ上がっても人間側に置いたままにします。精度99%は「100回に1回、取り返しのつかないことをする」という意味だからです。
②の上限を2回にしているのは、同じ工程を繰り返させると、エージェントは正しい答えではなく「それらしい答え」に収束していくからです。精度を上げる条件ではなく、気づかないまま回り続けるのを防ぐ条件です。
最小実装
agent(payload) は (出力, 確信度, 判断理由) を返す関数、という前提です。LLM の呼び出し部分は何でも構いません。
import json
from dataclasses import dataclass, asdict
from datetime import datetime
from enum import Enum
class StopReason(str, Enum):
LOW_CONFIDENCE = "low_confidence" # 確信度が下限を割った（やり直しへ）
RETRY_LIMIT = "retry_limit" # やり直し上限に達した（人間へ）
HUMAN_REQUIRED = "human_required" # 人間が決める操作だった（人間へ）
CONFIDENCE_FLOOR = 0.7 # 例。これを割ったら次の工程に渡さない（値は工程ごとに決める）
MAX_ATTEMPTS = 2 # 3回目に入る前に止める
IRREVERSIBLE = {"send", "delete", "payment", "permission_change"}
@dataclass
class StepRecord:
step: str
input: str
output: str
confidence: float
reason: str # 判断の理由（出力とは別に残す）
at: str
stop_reason: str | None = None
def log(rec: StepRecord, path: str = "decisions.jsonl") -> None:
with open(path, "a", encoding="utf-8") as f:
f.write(json.dumps(asdict(rec), ensure_ascii=False) + "\n")
def run_step(step: str, action: str, payload: str, agent) -> StepRecord | None:
"""agent(payload) -> (output, confidence, reason) を返す関数を受け取る"""
# ③ 取り返しのつかない操作は、エージェントに渡す前に止める
if action in IRREVERSIBLE:
log(StepRecord(step, payload, "", 0.0, f"action={action}",
datetime.now().isoformat(), StopReason.HUMAN_REQUIRED.value))
return None
for attempt in range(1, MAX_ATTEMPTS + 1):
output, confidence, reason = agent(payload)
rec = StepRecord(step, payload, output, confidence, reason,
datetime.now().isoformat())
# ① 確信度が下限以上のものだけ、次の工程に渡す
if confidence >= CONFIDENCE_FLOOR:
log(rec)
return rec
# ② 最後の1回で下限を割ったら、やり直さずに人間へ返す
rec.stop_reason = (StopReason.RETRY_LIMIT if attempt == MAX_ATTEMPTS
else StopReason.LOW_CONFIDENCE).value
log(rec)
return None
ポイントは3つです。
-
1工程を1行の JSON で外に出す。
{工程, 入力, 出力, 確信度, 判断理由, 時刻}を、会話ログの中ではなく、あとから機械で読める場所に残す - **止まった理由を別の値にする。**全部まとめて「エラー」にすると、何を直せばいいか分からなくなる
- **判断理由を出力と別の列にする。**出力だけでは、どこで間違えたかが残らない
止まった数を、毎日数える
JSON Lines にしてあれば、理由別の件数は1行で出ます。
grep -o '"stop_reason": "[a-z_]*"' decisions.jsonl | sort | uniq -c
1 "stop_reason": "human_required"
2 "stop_reason": "low_confidence"
1 "stop_reason": "retry_limit"
（上のコードを4工程ぶん動かしたときの出力です）
見方は2つだけです。
- 止まった数がゼロ → 閾値が緩すぎる
- 特定の理由だけ多い → その工程の設計が悪い
この数字は、精度より先に見る指標です。止まらないシステムは、壊れているのに気づけないシステムです。
ブレーキの無い車に、アクセルを3つ付けても速くはなりません。
止まった数を数え始めると、次に困るのは「記録が溜まるだけで、誰も見なくなる」ことでした。見るたびに件数の多さが気になって、ますます開かなくなる。
停止理由ごとに何を1つだけ変えるかを決めて、週15分で改善に回す手順を note に書いています。この記事の続きにあたります。