AIエージェント本番運用で学んだ痛い失敗｜プロンプト忘却とメモリ地獄の脱出記
AIエージェントを本番投入して6ヶ月。コンテキスト爆発、無限ループ、マルチエージェント調停で何度も失敗した実装記録。実際に効いた設計パターンと2026年に向けた課題をまとめました。
本番投入1ヶ月で「あ、これヤバい」と思った瞬間
うちのチームが複数タスクを自律的にこなすAIエージェントを本番に入れたのは昨年秋。最初の1週間は平穏だった。ユーザーからのタスク依頼を自動で分解して、複数のサブエージェントが並列で処理する仕組みだったんだ。
でも2週間目、ちょっとおかしくなり始めた。
エージェントが同じ質問を何度も繰り返すようになったんだ。ユーザーからの質問に対して、A→B→Cの順序で処理するはずが、A→A→A→Bみたいな感じで、コンテキストを忘れたまま無限ループに入る。プロンプトは正しかったのに、実行時に「前のステップで何をしたか」をエージェントが記憶していない。これが、半年かけて解決することになった最初の課題だった。
プロンプト忘却とメモリ管理：見落とされている落とし穴
AIエージェントの本番化で誰もが見落とすのがこれだ。単一のLLMコールと違って、エージェントは複数ステップの推論を重ねる。その過程で、前のステップの「判断理由」をどうやって次のステップに引き継ぐかという問題が発生する。
うちが最初に試した方法は素朴だった。会話履歴をそのままコンテキストに突っ込むやつだ。でもこれ、すぐに爆発する。長いタスクフローでは、コンテキストウィンドウが限界に達する前に、トークン数が月額数百万円に跳ね上がったんだ。
# これは本番で失敗したパターン
class SimpleAgent:
def __init__(self):
self.conversation_history = [] # すべての履歴を保持
async def execute_step(self, task):
# 毎回、全履歴をプロンプトに入れる
full_context = "\n".join(self.conversation_history)
prompt = f"{full_context}\n\n次のステップ: {task}"
response = await llm.call(prompt)
self.conversation_history.append(response)
return responseこのコード、一見合理的に見えるけど、実務的には本当に困る。5ステップ進むと、6ステップ目のプロンプトは前5ステップの全部を含んでる。10ステップ目は10ステップ分の履歴。これ月間1000回実行されると、コスト的に終わる。
僕たちが採用した解決策は「サマリー化」と「スライディングウィンドウ」の組み合わせだ。
class SmartAgent:
def __init__(self):
self.recent_steps = [] # 直近3ステップだけ保持
self.step_summary = "" # 古いステップはサマリー化
async def execute_step(self, task):
# 直近のステップと、古いステップのサマリーだけを使う
context_parts = []
if self.step_summary:
context_parts.append(f"前のステップ概要:\n{self.step_summary}")
if self.recent_steps:
context_parts.append(
f"直近のステップ:\n" +
"\n".join([s["decision"] for s in self.recent_steps[-3:]])
)
context = "\n\n".join(context_parts)
prompt = f"{context}\n\n次のステップ: {task}"
response = await llm.call(prompt)
self.recent_steps.append({"decision": response})
# 5ステップを超えたら古い履歴をサマリーに変える
if len(self.recent_steps) > 5:
summary = await self._summarize_old_steps()
self.step_summary = summary
self.recent_steps = self.recent_steps[-3:] # 最新3つだけ保持
return response
async def _summarize_old_steps(self):
old_decisions = [s["decision"] for s in self.recent_steps[:-3]]
summary_prompt = f"以下の判断過程を1段落で要約してください:\n" + "\n".join(old_decisions)
return await llm.call(summary_prompt)これでトークン数が3割程度削減された。実運用では、月額コストが半分になった。
マルチエージェント：調停地獄から脱出する設計
シングルエージェントで動いてるうちはいい。でも複雑なタスクになると、1つのエージェントには限界がある。データ取得用エージェント、分析用エージェント、レポート生成用エージェント…みたいに分割したくなるんだ。
でも、ここで新しい地獄が待ってた。複数のエージェントが同時に動作するとき、互いに矛盾した判断をしたらどうするのか。
うちの場合、こんなシナリオで爆発した：
- エージェントA：「このデータソースは信頼性が低いから除外すべき」
- エージェントB：「このデータソースを使って分析しました」
- エージェントC：「BのデータがAによって除外されたから、分析は無効」
結果、エージェント同士が何度も言い争う無限ループに入った。
flowchart TD
Task["タスク入力"] --> AgentA["エージェントA<br/>データ検証"]
Task --> AgentB["エージェントB<br/>データ分析"]
AgentA -->|矛盾した判断| Conflict["衝突検出"]
AgentB -->|矛盾した判断| Conflict
Conflict -->|再調整| AgentA
Conflict -->|再調整| AgentB
Conflict -->|ループ| Conflict
AgentA -->|合意| Mediator["調停エージェント"]
AgentB -->|合意| Mediator
Mediator --> Result["最終結果"]これを解決したのが「調停エージェント」パターンだ。複数のエージェントが並列で動作するのではなく、上位の調停エージェントが各エージェントの結果を検証・調整する仕組みにしたんだ。
class MediatorAgent:
def __init__(self, sub_agents: dict):
self.sub_agents = sub_agents # {"validator": AgentA, "analyzer": AgentB, ...}
self.conflict_resolver = LLMConflictResolver()
async def execute(self, task):
# 各エージェントを順序立てて実行
results = {}
for agent_name, agent in self.sub_agents.items():
result = await agent.execute(task, context=results)
results[agent_name] = result
# 結果が矛盾していないか検証
conflicts = await self._check_conflicts(results)
if conflicts:
# 矛盾があれば、調停エージェントが修正を指示
resolved = await self.conflict_resolver.resolve(results, conflicts)
return resolved
return self._merge_results(results)
async def _check_conflicts(self, results):
conflict_prompt = f"""以下の結果に矛盾がないか確認してください:
{json.dumps(results, ensure_ascii=False, indent=2)}
矛盾があれば、具体的に指摘してください。"""
analysis = await llm.call(conflict_prompt)
return self._parse_conflicts(analysis)これで、「一度決まったら覆さない」という意思決定の一貫性が担保された。実務的には、タスク完了に必要な平均ステップ数が35ステップから12ステップに短縮された。
状態管理：エージェント自体が「忘れる」問題
ここからが本当の地獄だった。
エージェントが、実行途中に「自分がどこにいるのか」を失うようになったんだ。同じタスクを2回実行したら、1回目と2回目で全く違う判断をする。ユーザーからは「なぜ同じ入力で違う結果が出るんだ」と文句が来た。
原因は、エージェントのメモリ（LLMの状態）が呼び出しごとにリセットされてたからだ。シングルスレッドならいいけど、複数ユーザーからのリクエストが並行して来ると、各リクエストのコンテキストが混ざる。
# これはダメなパターン
class StatefulAgent:
def __init__(self):
self.state = {} # グローバル状態
async def handle_request(self, request):
# 複数リクエストが同時に来ると、状態が上書きされる
self.state["current_task"] = request["task"]
# ここで別のリクエストが割り込まれると...
result = await self._process()
return resultこれを修正するために、各タスク実行に「実行ID」を割り当てて、その実行ID専用のメモリを持つようにした。
from dataclasses import dataclass
from typing import Dict
import uuid
@dataclass
class ExecutionContext:
execution_id: str
task: str
state: Dict
decision_log: list
created_at: float
class IsolatedAgent:
def __init__(self):
self.executions: Dict[str, ExecutionContext] = {}
self.max_age = 3600 # 1時間で古い実行を削除
async def handle_request(self, request):
execution_id = str(uuid.uuid4())
context = ExecutionContext(
execution_id=execution_id,
task=request["task"],
state={},
decision_log=[]
)
self.executions[execution_id] = context
try:
result = await self._process_with_context(context)
return {"execution_id": execution_id, "result": result}
finally:
# 実行完了後は状態をクリーンアップ
del self.executions[execution_id]
async def _process_with_context(self, context: ExecutionContext):
# コンテキストはこのメソッド内でのみ有効
prompt = f"""タスク: {context.task}
前の判断:
{chr(10).join([str(d) for d in context.decision_log])}
次のステップを決定してください。"""
decision = await llm.call(prompt)
context.decision_log.append(decision)
context.state["last_decision"] = decision
return decision
def _cleanup_old_executions(self):
# 定期的に古い実行を削除してメモリリークを防止
import time
current_time = time.time()
to_delete = []
for exec_id, ctx in self.executions.items():
if current_time - ctx.created_at > self.max_age:
to_delete.append(exec_id)
for exec_id in to_delete:
del self.executions[exec_id]これでメモリリークも防げて、同じ入力には常に同じ結果が返されるようになった。
パフォーマンス比較：本番での改善成果
本番で試した3つのパターンを数値化してみた。
xychart-beta
title AIエージェント実装パターン：コストと完了時間の比較
x-axis [素朴実装, スマート実装, 調停+隔離]
y-axis "月額コスト(万円)" 0 --> 300
line [280, 140, 95]| 指標 | 素朴実装 | スマート実装 | 調停+隔離 | 改善率 |
|---|---|---|---|---|
| 月額LLMコスト | 280万円 | 140万円 | 95万円 | 66%削減 |
| 平均完了時間 | 42秒 | 25秒 | 12秒 | 71%短縮 |
| エラー率 | 3.2% | 1.8% | 0.4% | 87%削減 |
| ユーザー満足度 | 6.2/10 | 7.4/10 | 8.8/10 | +42% |
正直に言うと、最初のコスト計算がめちゃくちゃ甘かった。プロンプトエンジニアリングの知見が不足していて、「LLMを何度も呼ぶから高くつく」というのを軽視していた。
ベクトルDB × RAGとの組み合わせ：現実的な知識管理
エージェントが「知識」を持つには、単にプロンプトに全部ぶち込むだけでは限界がある。大量の文書を参照する必要があるタスクでは、セマンティック検索が必須になるんだ。
うちが本番で採用したのはQdrant。Weaviateも候補だったけど、運用の複雑さとコストを考えると、Qdrantのほうが身軽だった。
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import numpy as np
class KnowledgeAwareAgent:
def __init__(self):
self.qdrant = QdrantClient(":memory:")
self.qdrant.recreate_collection(
collection_name="agent_knowledge",
vectors_config=VectorParams(size=384, distance=Distance.COSINE),
)
self.embedder = SentenceTransformer('intfloat/multilingual-e5-base')
async def retrieve_relevant_knowledge(self, query: str, top_k: int = 5):
# クエリを埋め込みベクトルに変換
query_vector = self.embedder.encode(query).tolist()
# セマンティック類似度でドキュメント検索
search_result = self.qdrant.search(
collection_name="agent_knowledge",
query_vector=query_vector,
limit=top_k,
score_threshold=0.5,
)
# スコアが高い順にドキュメントを取得
documents = [
{
"content": point.payload["text"],
"relevance": point.score,
"source": point.payload.get("source", "unknown")
}
for point in search_result
]
return documents
async def execute_with_knowledge(self, task: str):
# タスク関連の知識を動的に検索
relevant_docs = await self.retrieve_relevant_knowledge(task)
knowledge_context = "\n".join([
f"[{doc['source']}] {doc['content']}\n関連度: {doc['relevance']:.2f}"
for doc in relevant_docs
])
prompt = f"""以下の知識を参考にして、タスクを実行してください:
{knowledge_context}
タスク: {task}"""
result = await llm.call(prompt)
return resultこれで、エージェントは「自分が何を知っているか」を実行時に判断できるようになった。プロンプトの固定化から、動的な知識検索への移行は、エージェントの柔軟性を大幅に向上させたんだ。
本番で起きた「予想外」3つ
1. エージェントが「礼儀正しすぎる」
最初のバージョンでは、LLMの生成テキストをそのままユーザーに返していた。するとエージェントは毎回「申し訳ございませんが…」とか「ご指摘ありがとうございます…」とか、やたらと丁寧な返答をするようになった。
ユーザーは「そんなに丁寧にされたくない」って言う。データドリブンで判断させたいのに、エージェントはポライトネス優先になってた。
プロンプトを「簡潔に、事実のみを返すこと」に修正したら解消された。些細だけど、本番では意外と重要な調整だったんだ。
2. コスト最適化への執着が過ぎた
月額コストを削減することに必死になって、サマリー化を過度にやったら、エージェントが「全体像」を失うようになった。短期的にはコスト削減できたけど、判断の精度が落ちて、結局ユーザー対応コストが増えた。
コスト削減と精度は、トレードオフなんだ。「月額100万円以下」という目標を達成することより、「ユーザーが満足する結果を出す」ことを優先すべきだった。いい反省になった。
3. エージェント間の「相性」が存在する
複数のエージェントを組み合わせるときに気づいたんだけど、エージェントA + エージェントBはうまく行くけど、エージェントA + エージェントCだと相性が悪い、みたいなことが起きる。
原因は、各エージェントのプロンプトテンプレートが微妙に異なる「判断スタイル」を持ってること。一方は「慎重派」で、もう一方は「積極派」だと、調停エージェントが疲れるんだ。プロンプトを標準化することで対応した。
2026年のベストプラクティス：実装チェックリスト
本番運用を通じて、「これは絶対やっておくべき」というのが見えてきた。
実行コンテキストの隔離
- 各タスク実行に一意のIDを割り当て
- その実行専用のメモリを用意
- 実行完了後のクリーンアップを確実に
段階的なトークン削減
- 直近ステップと古いステップのサマリーを分離
- スライディングウィンドウで管理
- 定期的なコスト監視
マルチエージェント設計
- 調停エージェントで矛盾を検出・修正
- エージェント間の相性を事前テスト
- プロンプトテンプレートを標準化
知識管理
- ベクトルDBでセマンティック検索
- スコアしきい値で信頼性を担保
- ドキュメントのメタデータ管理
監視・ロギング
- 各ステップの決定理由をログ
- エージェント間の矛盾を自動検出
- コスト・精度・応答時間を定期的に測定
セキュリティとコンプライアンス：本番で必須な視点
AIエージェントが本番で動く場合、セキュリティとコンプライアンスは切り離せない。うちも本番化するときに、「AIエージェントが自動判断した結果について、誰が責任を持つのか」という問題に直面した。
結果として、重要な判断にはエージェントの推奨に加えて人間による承認ゲートを設置した。特にデータ操作や重要な決定が伴う場合は、エージェント→検証→承認という流れにして、トレーサビリティを確保した。監査ログも自動生成してるから、あとから「なぜこの判断をしたのか」を追跡できるようにしてる。
次のステップ：2026年の課題
ここまで解決してきたけど、新しい課題も見えている。
長期記憶の実装 エージェントが「過去のタスク実行から学習する」ということをまだ本気でやってない。1回のタスク完了時に、「この判断は正しかったのか」をフィードバックして、それを次のタスクに活かすというメカニズムは、実装案の段階なんだ。
分散実行 エージェントが複雑になると、1マシンで動かすのが厳しくなる。複数サーバーで分散実行するときの状態同期をどうするか、これは実装を進めてる最中だ。
リアルタイム最適化 タスク実行中に「このエージェントは今のサブタスクに適していない」と判断して、動的に別のエージェントに切り替えるロジック。これはまだ研究段階だから、今後の大きなテーマだと思ってる。
まとめ
AIエージェント開発は、単なる”LLM API呼び出し”では済まない。本番運用で気づくのは、実は地味だけど極めて重要な実装詳細ばかりなんだ。
メモリ管理が月額コストを決める サマリー化とスライディングウィンドウで66%削減可能。「トークン削減」と「判断精度」のバランスが本当に重要なんだ。
マルチエージェント設計は調停が必須 複数エージェントの矛盾検出と修正フロー。エージェント間の相性を事前テストすることで、本番での無限ループを防げる。
状態隔離なしに本番は走らない 各実行に一意IDを付与。メモリリーク対策としてのクリーンアップが地味に重要。
ベクトルDB + RAGは知識管理の必須要件 セマンティック検索でプロンプトのトークン数削減。スコアしきい値による信頼性担保で、判断の質が大きく変わる。
監視・ログなしに改善できない 各ステップの決定理由を記録。コスト・精度・応答時間の定期測定で、ボトルネックが見える。
うちのチームは今、長期記憶とフィードバックループの実装に取り組んでる。2026年後半には、エージェント自体が学習する仕組みを本番投入したいと思ってる。皆さんのチームはエージェント運用でどんな課題に直面してますか？気になったら、ぜひシェアしてもらいたい。