デモが終了したカスタムAIエージェントの呼び出しループとトークンコスト制御策

フレームワークを導入し、社내データでAIエージェントを作成する作業はプロトタイプ段階までは順調に進みます。本当の問題は、このコードを本番サービスに投入した瞬間に始まります。画面内では正常に動作していたエージェントが、予想外の入力1つに囚われて呼び出しを無限に繰り返し、週末の間に数百万ウォン規模のAPI費用が請求されることもあります。

現場で発生する障害の相当部分は、LLMモデル自体の限界よりも状態管理の失敗と外郭制御の欠如によって発生します。経営陣の指示でカスタムツールを作成したものの、実際には一日中不意に発生するエージェントのハルシネーションをデバッグし、大量のトークン領収書を処理するために本来の業務が進まない状況を終わらせなければなりません。

プロダクション環境で直面する3つの障害物

デモ環境を脱したエージェントは、ブラックボックスオペレーション、非決定論的ループ、データ漏洩リスクという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%のコスト削減を達成しました。

セマンティックキャッシングは以下の手順で構築します。

1. Redis Vector Searchバックエンドの準備: 入力プロンプトをエンベディングベクトルに変換して保存するRedisインスタンスを1つ起動します。
2. 類似度判定およびキャッシュ返却: 新規に入力されたプロンプトベクトルと既存の保存済みベクトルとのコサイン類似度が0.92以上であれば、外部LLMを呼び出さずに保存された応答を即座に返却します。
3. キャッシュミス時の保存: 類似度が0.92に満たない場合のみ外部APIを呼び出し、受信した応答をRedisに新しく更新します。

無限ループを根本的に遮断するには、エージェントの状態を検証するサーキットブレーカーパターンをパイプラインの制御フローに直接組み込む必要があります。フレームワークが提供するデフォルトの再試行オプションに依存すると、例外が発生した際にサービス全体がダウンします。ツール名と引数値を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の実行環境上でこのルーターが条件式を検証するため、ハルシネーションに伴うループ突入を物理的に防御します。

プロンプトバージョンの分離とCI/CD自動化

コード内にプロンプトを放置すると、モデル修正後にそれまで正常に動作していた機能まで不具合を起こす回帰障害（Regression Failure）を防ぐことができません。プロンプトはPythonコードの外へ切り出し、独立したYAMLファイルとして管理します。

`yaml
name: "agent_reasoning"
version: "1.2.0"
model: "gpt-4o"
temperature: 0.1
messages:

• role: "system"
  content: |
  あなたは社内データベースの分析エージェントです。
  指定されたJSONスキーマに従って回答する必要があり、情報が不足している場合は追加の質問を生成してください。
• role: "user"
  content: "{{ user_input }}"

`

このように分離したプロンプトは、オープンソースの評価フレームワークである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

1. 評価基準の定義

correctness_metric = GEval(
name="正確性およびスキーマ準拠",
criteria="LLMの応答が質問に正確に回答し、要求されたJSON形式を完全に遵守しているか？",
evaluation_params=[SingleTurnParams.ACTUAL_OUTPUT, SingleTurnParams.EXPECTED_OUTPUT],
threshold=0.7
)

2. Pytestベースの回帰テスト

@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段階で連携します。

1. 評価データセットの作成: 主要なドメインのクエリと模範解答のセットを作成し、許容基準スコア（Threshold）を設定します。
2. GitHub Actionsワークフローの統合: コード修正後にPRを提出した際、deepeval test runコマンドが自動実行されるようアクションを構築します。
3. シャドウテストの適用: メインブランチへの反映前、本番トラフィックのリクエストを非同期で複製・転送し、前バージョンと応答品質やレイテンシを比較した上で最終デプロイを実施します。

`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

1. Analyzer Engineおよび社員番号ルール登録

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)

2. Anonymizer Engine初期化

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モジュールを代替し、社内システムとして持続的に運用するには定期的な点検ルーティンが必要です。

自社モデルサービング（vLLMベース）と外部APIサブスクリプションの選択に悩んでいる場合は、日々のトラフィックを基準に試算します。

NVIDIA A10G GPUを1基搭載したAWS EC2 g5.2xlargeインスタンスの月額利用料は約$880です。一方、GPT-4o APIの単価は入力100万トークンあたり$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ヶ月で個人情報漏洩を防ぎ、すべてのリクエストがログに残るよう観測ソリューションを構築します。2ヶ月目には反復プロンプトのキャッシングとループ遮断ルーターを接続し、予期せぬ費用発生を削減します。最後の1ヶ月でプロンプトのバージョン管理と自動テスト評価体系を完成させれば、デプロイのたびに発生するエージェントの誤動作の懸念なく、安定した運用が可能になります。