- 本: AIエージェントポケットガイド：LLMを使用した自律システムの構築パターン
- 私の他の著作: Goで考える (2冊シリーズ) — Goプログラミング完全ガイド + Goにおける六角アーキテクチャ
- 私のプロジェクト: Hermes IDE |GitHub — クロードコードや他のAIコーディングツールと一緒にデベロッパーがリリースするためのIDE
- 私： xgabriel.com | GitHub
ドル47Kのループ
A LangChainのユーザーが一週末で約4万7千ドルを燃費悪い理由で使い切った。この話は2023年にTwitterとHNで広まり、失敗の形は変わっていない。エージェントは同じ検索ツールを同じ引数で何度も呼び出し、フレームワークは喜んで各結果を次のプロンプトにフィードバックし、各ラウンドに請求した。
トレースで10秒見ればわかる。47回連続で、同じtool_name、同じargsペイロードだが、タイムスタンプは違う。誰もそんなコードを書かない。どのモデルもそんなコードを書こうとはしない。しかし、少し壊れたツールを使うエージェントに曖昧な質問を突きつけると、同じコールを何度も繰り返し、何かが終了するまで続ける。
殺してやるべきものはPythonの20行です。それはエージェントに存在しません。トレースパイプラインに存在するので、フレームワークの交換、モデルのアップグレード、そして金曜日の午後4時に行うチームの次のリファクタリングを生き延びます.
なぜ max_iterations は間違ったノブです
Googleの最初のページで受け取るアドバイスは「設定してください」というものですmax_iterations=10". これは、住宅街の速度制限が高速道路で間違っているのと同じ理由で間違っている。これは正当な作業を罰している。
同じ製品で実行されている二つのエージェントを考慮してください。
エージェントAは深いリサーチアシスタントです。PDFを取得し、検索を実行し、要約し、3つの引用をフォローし、さらに3つの検索を実行し、検出結果を重複排除し、メモを書きます。80回のツール呼び出し、すべて異なる、すべて有用です。ユーザーはその深さのために支払いました。
エージェントBは不安定なベクトルインデックスを持つ質問応答エージェントです。クエリ#1ではsearch_docs(query="refund policy")を呼び出します。結果が空であるのは、古い埋め込み 때문です。エージェントは「もう一度試すべきだ」と考え、search_docs(query="refund policy")を2回目に呼び出し、その後3回目に呼び出します。ステップ7までに、エージェントは同じツールを同じ引数で7回連続で呼び出しています。
10回の深さ制限でエージェントAが完了する前に切断され、エージェントBが6回の反復を燃焼させる前にトリップする。逆を望む：エージェントAが進歩を続ける限り動作し、エージェントBが反復4で死ぬ。反復がシグナルである、深さではない。
20行の検出器
ここにあります。スライディングウィンドウカウンターが(tool_name, args_hash)にキーを付けます。各ツール呼び出しをプッシュ。キーが何か表示されたらthreshold回数のうちのwindow回の呼び出しで
from collections import deque, Counter
from dataclasses import dataclass, field
import hashlib
import json
class LoopDetected(Exception):
pass
@dataclass
class LoopDetector:
window: int = 10
threshold: int = 4
_calls: deque = field(default_factory=deque)
def observe(self, tool_name: str, args: dict) -> None:
key = (tool_name, _args_hash(args))
self._calls.append(key)
if len(self._calls) > self.window:
self._calls.popleft()
counts = Counter(self._calls)
most_common_key, hits = counts.most_common(1)[0]
if hits >= self.threshold:
raise LoopDetected(
f"{most_common_key[0]} called {hits}x "
f"in last {len(self._calls)} steps"
)
def _args_hash(args: dict) -> str:
canonical = json.dumps(_canonicalize(args), sort_keys=True)
return hashlib.sha256(canonical.encode()).hexdigest()[:16]
_VOLATILE_KEYS = {
"timestamp", "request_id", "trace_id", "span_id",
"nonce", "now", "_ts", "correlation_id",
}
def _canonicalize(value):
# strip keys that change every call but don't change intent
if isinstance(value, dict):
return {
k: _canonicalize(v)
for k, v in value.items()
if k not in _VOLATILE_KEYS
}
if isinstance(value, list):
return [_canonicalize(v) for v in value]
return value
これが全ての検出器です。インポートの数によっては20行程度です。これを投入し、各ツール呼び出しの後observe()を呼び出し、LoopDetectedをキャッチし、何か有用なことをする。
ハッシュは16進数文字16文字に切り詰められます。衝突はここでは問題ではありません。偽陽性（同じハッシュを計算する2つの異なる呼び出し）は何の損害もありませんなぜならループは実在せず次の正当な呼び出しがパターンを破ります。偽陰性（ハッシュが衝突したため実際のループが抜け出す）は統計的に無視できるレベルです16進数文字16文字を10回の呼び出しウィンドウで見ると
どこに置くべきですか
あなたには三つの選択肢があります。最悪から最善までランク付けされています。
エージェントループの内部。 あなたはLoopDetectorをエージェントランナーにインポートし、呼び出します。observe() それぞれのツール呼び出しの後に。簡単だ。しかし脆い。LangChainをLangGraphに置き換える日や、四半期の途中で別のフレームワークに移行する日、検出器は古いコードに付いていく。新しいエージェントをインストゥメント化するのを忘れないようにする必要もある。チームが急いでリリースする3番目のエージェントにはそれがない。
フレームワークコールバック. LangChainにはBaseCallbackHandler、LangGraphにはノードフックがあり、OpenAIのAgents SDKにはライフサイクルイベントがあります。observe()を呼び出すコールバックを一つ書きます。インラインより良いですが、まだフレームワーク固有です。交換するとまだ動かなくなります。
OTel spanエクスポートャー。これが適切な場所です。トレースはすでにエクスポートャーを通じて流れています。追加してください。SpanProcessorはツールコールのスパンを監視し、それらに対して検出器を実行します。フレームワークに依存しません。忘れることはできません。今日出荷されたか、先月の四半期に出荷されたかに関わらず、あなたの艦隊のすべてのエージェントを捉えます。
配置はこのようです：
from opentelemetry.sdk.trace import SpanProcessor
from opentelemetry.sdk.trace.export import (
BatchSpanProcessor,
)
from collections import defaultdict
class LoopDetectingProcessor(SpanProcessor):
def __init__(self, inner: SpanProcessor):
self.inner = inner
# one detector per trace_id
self._detectors = defaultdict(LoopDetector)
def on_start(self, span, parent_context=None):
self.inner.on_start(span, parent_context)
def on_end(self, span) -> None:
# GenAI semconv name for a tool invocation
if span.name == "execute_tool":
attrs = span.attributes or {}
tool_name = attrs.get(
"gen_ai.tool.name", "unknown"
)
# tool args often live under gen_ai.tool.call.arguments
raw_args = attrs.get(
"gen_ai.tool.call.arguments", "{}"
)
try:
args = json.loads(raw_args)
except (TypeError, ValueError):
args = {"_raw": str(raw_args)}
trace_id = format(span.context.trace_id, "032x")
detector = self._detectors[trace_id]
try:
detector.observe(tool_name, args)
except LoopDetected as exc:
span.set_attribute("loop.detected", True)
span.set_attribute("loop.reason", str(exc))
# signal the agent runtime via your own channel:
# Redis pub/sub, a kill flag in DB, etc.
self.inner.on_end(span)
def shutdown(self):
self.inner.shutdown()
def force_flush(self, timeout_millis=30000):
return self.inner.force_flush(timeout_millis)
既存のエクスポートャーをラップして、トレーサー提供者に登録します。これで検出器は、プラットフォームが実行する各エージェントからのすべてのツールスパンを認識します。属性名はOpenTelemetry GenAIのセマンティック規約(gen_ai.tool.name、gen_ai.tool.call.arguments)に従っているため、このコードはそれらのスパンを発行するものと互換性があります.
ウィンドウと閾値の調整
実際に機能するデフォルト値: window=10threshold=4.
その理由。初めての結果が不明瞭だったため、行動を起こすエージェントがツールを再度訪れる場合、2回、場合によっては3回、少し異なる引数でヒットします。10ステップで4回同じ呼び出しがあると、それが探索しているわけではなく、行き詰まっていることを意味します。閾値を3に押し上げると、ループを1ステップ早く捉えることができますが、一部の正当な再試行もフラグが立つ可能性があります。閾値を5に押し上げると、ループごとに1回の無駄な呼び出しが通過するため、GPT-4クラスのトークンレートでは実際のお金になります。
エージェントに指数的バックオフが組み込まれている場合（呼び出し、待機、再度呼び出し、より長く待機）、ウィンドウを15-20に広げ、閾値を4に保ってください。バックオフは繰り返しをより多くのステップに広げるので、より広いウィンドウが正当な再試行を捉えることができ、正当な再試行で過剰にトリガーされないようにします.
ツールカタログが小さい場合（3-5つのツール）で、エージェントが正当に多くのツールを再訪問する場合、例えばread_fileコーディングエージェントであればsearch_web は研究エージェントにおいて、(tool_name, args_hash) に重点を置くだけでなく、tool_name も重要です。args ハッシュが "search_web を8回異なるクエリで呼び出す"（問題ない）と "search_web を8回同じクエリで呼び出す"（問題がある）を区別するものです。
検知された場合にどうするか
三つの選択肢、あなたのエージェントをどれだけ信頼するかの順で並べます。
クイックスイッチ。 デフォルト。例外を発生させ、ループを記録し、呼び出し元に構造化されたエラーを返す。安価で安全。ユーザーが再試行する。
プロンプトでダウングレードする。 システムメッセージを挿入する：「同じ引数でsearch_docsを4回呼び出しました。ツールは同じ結果を返しています。異なるアプローチを試すか、停止して発見したことを報告してください。」 モデルは通常、エラーを発生させます。時にはそうでなく、その後、次の観測時にキルスイッチが作動します。
当番ページ。 ループが実際の停電を意味するエージェント（例えば、ユーザーリトライがない内部自律ツールなど）では、LoopDetected PagerDutyに接続します。まれですが、ループすべきでないエージェントにとって、ページは適切な形です。
キルスイッチから始めてください。回復可能なループに関するデータがある場合にのみ、ダウングレードを促すモードに移行してください.
誤解を招く二つのエッジケース
非決定的な引数. ツールの引数にタイムスタンプ、リクエストID、またはノンスが含まれている場合、ハッシュが各呼び出しで分岐します。上記のコンカナライザーは、一連の変動キー（timestamp、request_id）を削除しますtrace_id、span_id、nonce、now、_ts、correlation_idをハッシュ化する前に追加してください。ご自身のツールスキーマで新しいvolatileフィールドに到達したときにそのセットに追加します。created_at: <now>をそのargsに仕込むエージェントこそ、そうでなければ捕まらないであろうエージェントです。
ストリーミングツールコール。 いくつかのフレームワークは、ツール呼び出しがまだ実行中である間に部分的なスパンを発行します。gen_ai.tool.call.id でフィルタリングし、呼び出しがまだストリーミング中のものは無視します。そうしないと、一つの遅いツール呼び出しを複数の観察としてカウントし、自分自身に偽陽性を報告することになります。
これはあなたのスタックにどのように組み込まれますか
この検出器は、プロダクションエージェントが持つべき3つの実行時ガードの1つです。
トークンバジェット。各エージェント呼び出しごとに累積的な入力＋出力トークンの硬い上限。ループ検知が捉えられない「プロンプトが20万トークンに成長した」という失敗モードを捉える。
ループ検知器。この投稿に記載されているもの。固着した繰り返しを捉える。
目標完了検証器。最後に別途小さなLLM呼び出しを行い、「このエージェントが実際にユーザーの要求通りに行動したか、それとも要点を外れた自信満々な出力を生成したか？」を確認する。最初の二つに見逃される「30ステップ実行したが、ゴミを出力した」という失敗を検出する。
トレースパイプライン内で全ての三つを実行し、エージェント内ではなく。エージェントは信頼性の低い部分である。パイプラインがガードが行われる場所である。
本番環境にデプロイした最もひどいエージェントループは何ですか？トレースをコメントに載せてください。47を超えた人はいるか知りたいです。
これは役に立った면
ランタイムガードの三要素（トークンバジェット、ループ検知器、目標検証器）は、AIエージェントポケットガイド：LLMで自律システムを構築するためのパターンの一つです。。本書は残りの生産チェックリストをカバーしています：ツールカタログの規範、サブエージェントの境界、リプレイとドリフト検出、そしてそれらを観測可能にするトレースレイヤーのインストゥルメンテーション。エージェントをリリースし、一箇所でパターンを整理したいなら、それが本です。
