デ모が終了したカスタムAIエージェントの呼び出しループとトークンコスト制御策
makedream2026年7月28日
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
フレームワークを導入し、社내データでAIエージェントを作成する作業はプロトタイプ段階までは順調に進みます。本当の問題は、このコードを本番サービスに投入した瞬間に始まります。画面内では正常に動作していたエージェントが、予想外の入力1つに囚われて呼び出しを無限に繰り返し、週末の間に数百万ウォン規模のAPI費用が請求されることもあります。
現場で発生する障害の相当部分は、LLMモデル自体の限界よりも状態管理の失敗と外郭制御の欠如によって発生します。経営陣の指示でカスタムツールを作成したものの、実際には一日中不意に発生するエージェントのハルシネーションをデバッグし、大量のトークン領収書を処理するために本来の業務が進まない状況を終わらせなければなりません。
デモ環境を脱したエージェントは、ブラックボックスオペレーション、非決定論的ループ、データ漏洩リスクという3つの壁にぶつかります。
推論（Chain-of-Thought）プロセス、外部ツール呼び出し（Tool Call）、ベクトルデータベース検索の段階ごとの入出力がログに残らなければ、問題が発生しても原因を特定できません。どこでプロンプトが錯綜したかを追跡するだけで丸一日が潰れてしまいます。
さらに深刻なのは無限ループです。ReAct（Reasoning + Acting）構造を持つエージェントは、ツール実行結果が曖昧だと同じリクエストを繰り返し送信し続けます。
`
┌─────────────────────────────────────────────────────────────────────────┐
│ ReAct Architecture Loop │
│ │
│ ┌────────────┐ User Query ┌────────────┐ Tool Call Request │
│ │ User │ ──────────────> │ Main LLM │ ────────────────────┐ │
│ └────────────┘ └────────────┘ │ │
│ ▲ ▼ │
│ │ Observation ┌──────┐ │
│ │ (Ambiguous/Failed) │ Tool │ │
│ └─────────────────────── │ A │ │
│ └──────┘ │
│ * Problem: When Observation fails, LLM retries Tool A endlessly. │
└─────────────────────────────────────────────────────────────────────────┘
`
終了条件を抜け出せず、同一の引数でAPIを叩き続ける瞬間、わずか数分で1ヶ月分のトークン予算が枯渇します。メインエージェントとサブエージェントが互いを呼び出し合って再試行する連鎖的失敗は、システム全体の障害原因の30%以上を占めます。
対話履歴が蓄積し続けてコンテキストウィンドウが満杯になると、モデルのパフォーマンス自体が低下する問題も発生します。これに加え、Pythonコード内に文字列としてハードコーディングされたプロンプトは、単一の文言を修正する際にもシステム全体の再ビルドおよび再デプロイを余儀なくさせます。
ローカルデバッグソリューションとプロダクション観測ツールを分離し、収集ポイントを特定する必要があります。開発段階ではOpenTelemetryベースのArize PhoenixでRAGエンベディングの品質とツール呼び出しを確認します。本番サービス環境では、ClickHouseをバックエンドとするLangfuseを接続し、トレースログとトークン消費量をリアルタイムでモニタリングします。
`
┌─────────────────────────────────────────────────────────────────────────┐
│ Semantic Caching & Routing Flow │
│ │
│ Client Query │
│ │ │
│ ▼ │
│ ┌───────────┐ Similarity >= 0.92? ┌─────────────────────────────┐ │
│ │ Redis Vector │ ─────────────────────────> │ Return Cached Response │ │
│ │ Cache │ (Cache Hit) │ (Latency -88%, Cost -86%) │ │
│ └───────────┘ └─────────────────────────────┘ │
│ │ │
│ │ (Cache Miss) │
│ ▼ │
│ ┌───────────┐ Execution & Save Cache ┌─────────────────────────────┐ │
│ │ External │ ────────────────────────> │ Store Result as Vector in │ │
│ │ LLM API │ │ Backend Database │ │
│ └───────────┘ └─────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────┘
`
全く同じ問い合わせが繰り返されることで生じるAPI費用は、セマンティックキャッシング（Semantic Caching）レイヤーで防ぎます。
AWSが発表した60,000件のクエリ分析データによると、コサイン類似度基準を適用したセマンティックキャッシュを配置した場合、LLMの推論コストが最大86%削減され、応答レイテンシも88%改善されました。LMSYSのRouteLLMフレームワークも、質問の難易度に応じて高コストモデルと低コストモデルにリクエストを振り分けることで、85%のコスト削減を達成しました。
セマンティックキャッシングは以下の手順で構築します。
無限ループを根本的に遮断するには、エージェントの状態を検証するサーキットブレーカーパターンをパイプラインの制御フローに直接組み込む必要があります。フレームワークが提供するデフォルトの再試行オプションに依存すると、例外が発生した際にサービス全体がダウンします。ツール名と引数値をSHA-256でハッシュ化してリストに格納し、同じハッシュが3回連続で蓄積された場合は強制的によく制御された別ノードへ遷移させるのが安全です。
`python
import hashlib
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
steps: int
tool_hashes: List[str]
def hash_tool_call(tool_name: str, tool_args: str) -> str:
raw_str = f"{tool_name}:{tool_args}"
return hashlib.sha256(raw_str.encode('utf-8')).hexdigest()
def agent_circuit_breaker_router(state: AgentState) -> str:
# 1. 全体の実行回数が5回を超えた場合、あらかじめ決められた退避ノードに遷移
if state["steps"] > 5:
return "fallback_graceful_node"
# 2. 同一のToolとArgumentを3回連続で呼び出した場合は即座に遮断
hashes = state.get("tool_hashes", [])
if len(hashes) >= 3 and hashes[-1] == hashes[-2] == hashes[-3]:
return "fallback_graceful_node"
# 3. 正常終了キーワードの確認
last_message = state["messages"][-1]
if "FINAL_ANSWER" in last_message.content:
return "end"
return "continue_tools"
`
LLMの非決定論的な出力状態に関係なく、Pythonの実行環境上でこのルーターが条件式を検証するため、ハルシネーションに伴うループ突入を物理的に防御します。
コード内にプロンプトを放置すると、モデル修正後にそれまで正常に動作していた機能まで不具合を起こす回帰障害（Regression Failure）を防ぐことができません。プロンプトはPythonコードの外へ切り出し、独立したYAMLファイルとして管理します。
`yaml
name: "agent_reasoning"
version: "1.2.0"
model: "gpt-4o"
temperature: 0.1
messages:
`
このように分離したプロンプトは、オープンソースの評価フレームワークであるDeepEvalとPytestを組み合わせてCI/CDパイプライン上で自動テストします。G-Eval指標を基準点に設定し、デプロイ前にモデルの応答数値を検証します。
`python
import pytest
from deepeval import assert_test
from deepeval.metrics import GEval, TaskCompletionMetric
from deepeval.test_case import LLMTestCase, SingleTurnParams
correctness_metric = GEval(
name="正確性およびスキーマ準拠",
criteria="LLMの応答が質問に正確に回答し、要求されたJSON形式を完全に遵守しているか？",
evaluation_params=[SingleTurnParams.ACTUAL_OUTPUT, SingleTurnParams.EXPECTED_OUTPUT],
threshold=0.7
)
@pytest.mark.parametrize(
"user_input, expected_output",
[
("2024年第1四半期の売上データを要約して。", "第1四半期の総売上は50億ウォンです。"),
("退職金計算規定を教えて。", "退職金は勤続年数1年に対して30日分以上の平均賃金です。")
]
)
def test_agent_regression(user_input, expected_output):
actual_output = run_in_house_agent(user_input)
test_case = LLMTestCase(
input=user_input,
actual_output=actual_output,
expected_output=expected_output
)
# 設定した基準スコアを下回る場合はビルドを中断
assert_test(test_case, [correctness_metric, TaskCompletionMetric(threshold=0.8)])
`
評価体系は次の3段階で連携します。
deepeval test runコマンドが自動実行されるようアクションを構築します。`yaml
name: AI Agent Evaluation Gate
on:
pull_request:
branches: [ main ]
jobs:
eval-gate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install Dependencies
run: |
pip install poetry
poetry install
- name: Run DeepEval Suite
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
poetry run deepeval test run tests/test_evals.py
`
社内データの漏洩を防ぐには、APIリクエストが外部へ送信される前に個人可視情報（PII）を匿名化するガードレールが必要です。ゲートウェイにMicrosoft Presidioエンジンを組み込むことで、住民登録番号、メールアドレス、電話番号、社員番号などを自動でマスキング処理します。
`python
from presidio_analyzer import AnalyzerEngine, PatternRecognizer
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
analyzer = AnalyzerEngine()
employee_id_recognizer = PatternRecognizer(
supported_entity="EMPLOYEE_ID",
regex="EMP-[0-9]{6}",
score=0.95
)
analyzer.registry.add_recognizer(employee_id_recognizer)
anonymizer = AnonymizerEngine()
def sanitize_user_prompt(raw_prompt: str) -> str:
results = analyzer.analyze(
text=raw_prompt,
entities=["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "EMPLOYEE_ID"],
language="en"
)
anonymized_result = anonymizer.anonymize(
text=raw_prompt,
analyzer_results=results,
operators={
"DEFAULT": OperatorConfig("replace", {"new_value": "<REDACTED>"}),
"EMPLOYEE_ID": OperatorConfig("mask", {"chars_to_mask": 6, "masking_char": "*", "from_end": True})
}
)
return anonymized_result.text
`
RAG検索を組み込む際も、ユーザーの権限レベルに応じたドキュメントのみを取得するよう対策を講じる必要があります。QdrantやPineconeなどのベクトルデータベースは、類似度検索の際にメタデータフィルタリング（Metadata Filtering）を用いて権限分離を処理します。
`python
from qdrant_client import QdrantClient
from qdrant_client.http import models
client = QdrantClient(host="localhost", port=6333)
def search_documents_with_rbac(query_vector: list, user_department: str, user_clearance_level: int):
search_result = client.search(
collection_name="enterprise_knowledge_base",
query_vector=query_vector,
query_filter=models.Filter(
must=[
models.FieldCondition(
key="department",
match=models.MatchValue(value=user_department)
),
models.FieldCondition(
key="security_level",
range=models.Range(lte=user_clearance_level)
)
]
),
limit=5
)
return search_result
`
プロンプトインジェクションの防御策としては、システムプロンプトチャネルとユーザー入力チャネルを完全に切り離すことが基本となります。データベースの変更や外部APIの呼び出しなど、危険を伴う作業については人間による直接承認ステップ（Human-in-the-Loop）を挟むか、分離されたサンドボックス内に実行権限を限定することで安全性を担保します。
SaaSモジュールを代替し、社内システムとして持続的に運用するには定期的な点検ルーティンが必要です。
| 周期 | 運用点検項目 | 詳細作業 |
|---|---|---|
| 毎日 | エラー率およびトークン使用量 | LangfuseダッシュボードでHTTP 5xxエラー率と部署別トークン消費量を確認 |
| 毎日 | サーキットブレーカー遮断履歴 | ループ検知により切断されたセッションのツール名と引数パターンを収集して修正 |
| 毎週 | RAG検索失敗クエリの抽出 | 類似度スコア0.6未満のリクエストを抽出してテスト正解セットに追加 |
| 毎週 | 個人情報マスキング誤検知検証 | Presidio処理ログをサンプリング抽出し、マスキング漏れの有無を確認 |
| 毎月 | CI/CD評価基準のアップデート | ビジネスの変更事項を反映して自動テストケースを修正 |
自社モデルサービング（vLLMベース）と外部APIサブスクリプションの選択に悩んでいる場合は、日々のトラフィックを基準に試算します。
NVIDIA A10G GPUを1基搭載したAWS EC2 g5.2xlargeインスタンスの月額利用料は約2.50、出力100万トークンあたり$10.00程度です。日々のトラフィックが5,000万トークン未満であれば、DevOpsエンジニアの管理工数とGPU固定費を考慮した場合、APIサブスクリプション方式が有利です。1日のトラフィックが1億トークンを超えるか、完全な社内網分離が必須である場合に初めて、vLLMベースの自社サービングへの移行を検討するのが適切です。
メインモデルの障害に備えるため、4段階のバックアップルーティング構造を導入します。
`
┌─────────────────────────────────────────────────────────────────────────┐
│ 4-Tier Graceful Degradation Architecture │
│ │
│ [Tier 1] Primary High-Performance Model (e.g., GPT-4o) │
│ │ │
│ ▼ (API Failure / Timeout / Circuit Breaker) │
│ [Tier 2] Lightweight Routing Model (e.g., GPT-4o-mini / On-Prem vLLM) │
│ │ │
│ ▼ (Continuous Outage) │
│ [Tier 3] Deterministic Regex & SQL Rule Engine │
│ │ │
│ ▼ (Unrecoverable Error) │
│ [Tier 4] Static Error Message & Async Admin Ticket Generation │
└─────────────────────────────────────────────────────────────────────────┘
`
カスタムエージェントシステムの最適化作業は、90日単位に区切って進めます。
| 期間 | 適用目標 | 具体的な作業内容 |
|---|---|---|
| 1〜30日目 | 可視性およびデータ保護 | Presidioマスキングエンジンの導入およびLangfuseロギングSDKの移植 |
| 31〜60日目 | ループ遮断およびコスト削減 | Redisセマンティックキャッシングの適用およびPythonベースのサーキットブレーカー接続 |
| 61〜90日目 | 検証自動化および権限制御 | プロンプトのGit管理、DeepEval CI/CD連携、Qdrant RBACメタデータの適用 |
最初の1ヶ月で個人情報漏洩を防ぎ、すべてのリクエストがログに残るよう観測ソリューションを構築します。2ヶ月目には反復プロンプトのキャッシングとループ遮断ルーターを接続し、予期せぬ費用発生を削減します。最後の1ヶ月でプロンプトのバージョン管理と自動テスト評価体系を完成させれば、デプロイのたびに発生するエージェントの誤動作の懸念なく、安定した運用が可能になります。