記憶できるエージェントを作ったはずが、なぜか育たない
Claude にツールを渡して「過去のやり取りを覚えてくれるエージェント」を作る。最初の数日はうまくいきます。ユーザーが名前を伝えれば翌日も呼んでくれ、プロジェクトの前提を引き継いでくれる。ここで多くの実装が「できた」と判断して、そのままユーザーに開放してしまいます。
けれど、2週間ほど走らせると景色が変わってきます。なぜか的外れな前提で返答が返ってくる日があり、別のユーザーの発言が混ざっているように見える瞬間が出てくる。ログを見ると、メモリに書き込まれた中身が原型を留めていありません。実運用では、こちらの方が普通です。
私自身も個人開発のエージェントに長期記憶を入れたとき、最初の3週間はひたすら記憶を直していました。直しながら「これは書き込みの設計そのものを見直さないと無理だ」と気づき、そこから組み直して今の設計に落ち着いています。ここではそのときに踏み抜いた落とし穴のうち、本番で特に致命傷になりやすい7つを、対策パターンと併せてお伝えします。
対象とするのは、Claude の Memory 系ツール(Anthropic の Memory Tool、あるいは自作の MCP で同等の読み書き API を提供しているもの)を使って、エージェントに長期記憶を持たせる設計です。RAG とは領域が重なるところもありますが、ここでは「エージェント自身が書き込みを行う記憶」を主題に扱います。
長期記憶設計の全体像 — 読み・書き・忘却の三層で考える
落とし穴を語る前に、全体像を揃えておきます。私は長期記憶を、読み(Retrieve)・書き(Write)・忘却(Forget)の三層で捉えることをおすすめしています。この3つを別物として設計すると、問題が起きたときにどの層の話なのかが切り分けられるようになります。
- 読み: エージェントがリクエストのたびに、どのメモリを取り出してコンテキストに入れるかを決める層。安易に全件読み込むとコンテキストを汚染します。
- 書き: エージェントが「この情報は覚えておくべき」と判断して保存する層。ここが荒れるとすべてが狂います。
- 忘却: 古くなった・誤っている・プライバシー的に保持してはいけないメモリを削除する層。実装されていないケースが多いのが実情です。
実装としての最小構成は、書き込みツール(例: memory_write)・読み出しツール(例: memory_search)・削除ツール(例: memory_forget)の3つを Claude に渡し、背後にフラットファイルか KV ストアを置く形です。Anthropic の Memory Tool を使う場合も、基本構造は同じです。Claude API の Tool Use 完全ガイド の知識があると、この先の話が入りやすいと思います。
# 最小構成の Memory ツール群(Anthropic Python SDK 想定)
from anthropic import Anthropic
import json, time, hashlib, pathlib
MEMORY_DIR = pathlib.Path("./memory_store")
MEMORY_DIR.mkdir(exist_ok=True)
def memory_write(user_id: str, key: str, value: str, source_turn_id: str) -> dict:
"""書き込みは必ず user_id でスコープし、出典(source_turn_id)を必須化する。"""
if len(value) > 2000:
return {"ok": False, "error": "value too long (>2000 chars). split or summarize first."}
path = MEMORY_DIR / user_id
path.mkdir(exist_ok=True)
record = {
"key": key,
"value": value,
"source_turn_id": source_turn_id,
"written_at": int(time.time()),
"checksum": hashlib.sha1(value.encode()).hexdigest(),
}
(path / f"{key}.json").write_text(json.dumps(record, ensure_ascii=False))
return {"ok": True, "key": key}
def memory_search(user_id: str, query: str, limit: int = 5) -> dict:
"""読み出しは必ず user_id スコープ + limit で境界を持たせる。"""
path = MEMORY_DIR / user_id
if not path.exists():
return {"ok": True, "hits": []}
hits = []
for f in sorted(path.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True):
rec = json.loads(f.read_text())
if query.lower() in rec["value"].lower() or query.lower() in rec["key"].lower():
hits.append(rec)
if len(hits) >= limit:
break
return {"ok": True, "hits": hits}
# 期待する出力例:
# memory_write(user_id="u_42", key="preferred_language", value="Japanese", source_turn_id="t_81")
# -> {"ok": True, "key": "preferred_language"}
# memory_search(user_id="u_42", query="language", limit=3)
# -> {"ok": True, "hits": [{"key": "preferred_language", "value": "Japanese", ...}]}ここで重要なのは、どのツールも user_id を第一引数に必須化し、source_turn_id(どの会話ターンに由来するか)を必ず記録している点です。この2つを徹底しておくと、後述する落とし穴の多くを構造的に防げます。
落とし穴1 — 書き込みを「自動化しすぎる」問題
最初の落とし穴は、書き込みをエージェントの裁量に任せ切ってしまう設計です。「気づいたら覚えておいて」というシステムプロンプトを渡すと、Claude は相当の積極性で書き込みに行きます。一見便利ですが、同じ事実が微妙に表現を変えて何十件も保存されたり、会話の流れで仮に出た仮定が「確定事項」として書かれたりします。
対策: 書き込みを2段階にします。第1段階ではエージェントに memory_candidate_propose を提案させるだけで、実際には書き込みません。第2段階で、別の軽量モデル(Haiku など)か、ルールベースの検証器が「本当に覚えるべきか」を判定し、合格したものだけを memory_write に渡します。
# 2段階書き込みの最小実装
def propose_and_validate(proposal: dict) -> dict:
"""重複チェック + 「確定度」の判定。失敗したら書かない。"""
existing = memory_search(proposal["user_id"], proposal["key"], limit=1)
if existing["hits"] and existing["hits"][0]["value"] == proposal["value"]:
return {"ok": False, "reason": "duplicate"}
# 「たぶん」「かもしれない」等の不確実語を含む提案は却下
hedges = ["たぶん", "かもしれ", "perhaps", "maybe", "might"]
if any(h in proposal["value"].lower() for h in hedges):
return {"ok": False, "reason": "uncertain"}
return {"ok": True}
# 期待する挙動:
# propose_and_validate({"user_id": "u_42", "key": "goal", "value": "たぶんダイエットしたい"})
# -> {"ok": False, "reason": "uncertain"}
# propose_and_validate({"user_id": "u_42", "key": "goal", "value": "5kg減量を6月までに達成する"})
# -> {"ok": True}なぜ2段階にするのかというと、書き込みの責任をモデルから「人(あるいは検証器)」へ引き剥がすためです。1段目のモデルは提案するだけなので、積極性の高さが問題になりません。2段目で落ちた提案はログに残し、後から「なぜ覚えようとしたか」を振り返る素材にもなります。このログ自体が、記憶の健全性を監視する最初の指標になります。
落とし穴2 — PII と機密情報の混入が止まらない
長期記憶を運用していてもっとも寒気が走るのは、メモリ内に保存すべきでなかった情報が長期間残っていたと気づく瞬間です。クレジットカード番号、病名、個人住所、企業の内部プロジェクト名——ユーザーは会話の流れで何でも書きます。そしてエージェントは、雑に書き込みを許すと、雑にそれらを保存してしまいます。
対策: 書き込み経路に「3段フィルタ」を入れます。正規表現での明らかな PII ブロック、分類モデルによる機密度スコアリング、そしてユーザー同意の再確認。3段階目は面倒に見えますが、機密度スコアが閾値を超えたときだけ発火する設計にすれば運用負荷は限定的です。
import re
PII_PATTERNS = {
"credit_card": re.compile(r"\b(?:\d[ -]*?){13,19}\b"),
"jp_phone": re.compile(r"0\d{1,4}-\d{1,4}-\d{3,4}"),
"email": re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"),
"my_number": re.compile(r"\b\d{12}\b"), # 日本のマイナンバー想定
}
def pii_redact_or_block(value: str) -> dict:
hits = []
redacted = value
for name, pat in PII_PATTERNS.items():
if pat.search(redacted):
hits.append(name)
redacted = pat.sub(f"[REDACTED:{name}]", redacted)
if "credit_card" in hits or "my_number" in hits:
# 明らかに高リスクは保存拒否
return {"ok": False, "hits": hits}
return {"ok": True, "value": redacted, "hits": hits}
# 期待する挙動:
# pii_redact_or_block("連絡先は 090-1234-5678 です")
# -> {"ok": True, "value": "連絡先は [REDACTED:jp_phone] です", "hits": ["jp_phone"]}
# pii_redact_or_block("カード 4111 1111 1111 1111 を登録")
# -> {"ok": False, "hits": ["credit_card"]}正規表現だけでは取りこぼしが出ますので、私は上記に加えて Haiku に「このテキストに機密情報が含まれる可能性を 0-1 で返してください」とだけ聞く判定器を並列で走らせています。正規表現で取りこぼした自由記述の機密情報(病状の記述、社名入りの未公開案件など)を、このスコアでもう一度救っています。閾値を超えたものは保存せず、ユーザーに対して「この内容は覚えない設計です」と明示します。
落とし穴3 — スコープ分離の甘さが生むクロス汚染
複数ユーザーが同じエージェントに触れる構成では、スコープ分離が甘いだけでクロスユーザー情報漏洩が発生します。「田中さんが昨日話していたプロジェクト X の件」を、別のユーザーの会話に混ぜ込んでしまうパターンです。
対策: 読み書きの全ツールに、user_id あるいは tenant_id を 最初の引数として物理的に強制 します。ツール定義のスキーマで required にするだけでなく、バックエンド側で「ヘッダから取り出した認証済み user_id と、ツール引数の user_id が一致していなければエラー」とする二重チェックを入れます。
# ツール呼び出し前の二重チェック(FastAPI 等のフレームワーク想定)
def guarded_tool_call(authenticated_uid: str, tool_name: str, tool_args: dict) -> dict:
"""認証済み UID とツール引数の UID が食い違ったら即座に失敗させる。"""
requested_uid = tool_args.get("user_id")
if requested_uid is None:
return {"ok": False, "error": "user_id is required"}
if requested_uid != authenticated_uid:
# 監査ログに残す(不審な挙動の早期検知にもなる)
audit_log(event="uid_mismatch", expected=authenticated_uid, got=requested_uid, tool=tool_name)
return {"ok": False, "error": "uid mismatch"}
return dispatch_tool(tool_name, tool_args)さらに、データそのものの配置もユーザーごとのディレクトリや KV プレフィックスで物理的に分離しておくと、仮にコード側でミスがあっても、他ユーザーのデータが読み出されない構造になります。私はフラットファイルで試作する段階から、必ず memory_store/{user_id}/... のディレクトリ分割を入れるようにしています。実装を差し替える頃には、この境界に触らないコードが育っているはずです。
落とし穴4 — データ破損を検知できない運用体制
書き込みの不具合、途中で死んだプロセス、ディスクや KV の障害——どれも現場では起きます。問題は、破損が起きても気づくまでに時間がかかり、気づいたときには「破損メモリを参照して学習を続けたエージェントが出した誤回答」が大量に蓄積していることです。
対策: 書き込み時にチェックサムを必ず記録し、読み出し時に毎回検証します。さらに、定期ジョブで全件の健全性チェックを行い、破損レコードを隔離領域(quarantine)に移動させる仕組みを入れます。隔離はしても即削除はしない、というのがポイントで、復旧の余地を残します。
def memory_read(user_id: str, key: str) -> dict:
path = MEMORY_DIR / user_id / f"{key}.json"
if not path.exists():
return {"ok": False, "error": "not found"}
rec = json.loads(path.read_text())
actual = hashlib.sha1(rec["value"].encode()).hexdigest()
if actual != rec.get("checksum"):
# 破損検知: 隔離領域に移動して、読み出しは失敗させる
quarantine = MEMORY_DIR / "_quarantine" / user_id
quarantine.mkdir(parents=True, exist_ok=True)
path.rename(quarantine / path.name)
audit_log(event="memory_corrupted", user_id=user_id, key=key)
return {"ok": False, "error": "corrupted (quarantined)"}
return {"ok": True, "record": rec}
# 期待する挙動(破損を人工的に作って確認):
# -> 正常時: {"ok": True, "record": {...}}
# -> 破損時: {"ok": False, "error": "corrupted (quarantined)"}運用してみると、破損の多くは「書き込み途中にプロセスが落ちた」か「別プロセスが同じキーを同時に書いた」のどちらかです。ここにファイルロック(あるいはトランザクショナルな KV ストア)を入れると、検知される破損レコードの数がはっきり減ります。ただし、0 にしようとしすぎて書き込みが遅延すると体験が悪化しますので、「検知できる」状態を保ったまま、頻度を下げる方向で詰めていくのが実務的です。
落とし穴5 — コストが肥大化する前に入れるべきガードレール
長期記憶が軌道に乗り始めると、次に襲ってくるのがコストの問題です。メモリ件数が増えると、読み出しでコンテキストに積む情報量が増え、入力トークンが膨らみ、Prompt Caching の恩恵も受けにくくなります。ある月に請求額を見て驚いた、というのは私も一度やりました。
対策: 読み出しに「トークン予算」を設けます。毎リクエストで読み出す記憶の合計トークン数に上限を決め、それを超える場合は重要度順に切り詰めます。加えて、古いメモリは定期的に要約して圧縮する「Consolidation」ジョブを走らせます。Claude API の永続メモリ運用ガイド(pgvector 版) でも触れた考え方ですが、ネイティブの Memory ツール運用でも同じ原則が効きます。
私のおすすめする監視指標は3つです。1つ目は「1 リクエストあたりの memory 由来の入力トークン数」、2つ目は「ユーザーあたりの総メモリ件数」、3つ目は「30日以上参照されていないメモリの割合」。3つ目が4割を超えたら、Consolidation を1段強める合図として扱っています。これをダッシュボードに出し、閾値超えでアラートを飛ばすだけで、突然の高額請求はほぼ防げます。
なお、Extended Thinking を併用する設計の場合は、読み出したメモリが Thinking 入力にも乗るため、一段とコストに跳ねます。Extended Thinking を使うターンでは memory 読み出しを絞る、あるいは要約版を使うといった分岐を入れるのが現実的です。
落とし穴6 — 移行パターン — pgvector / 自作 MCP からの乗り換え
すでに pgvector や自作 MCP でメモリ機構を持っているチームが、Anthropic の Memory Tool 相当の実装(あるいは別のランタイム)へ移行したくなるタイミングは必ず来ます。このとき一気に切り替えると、既存の記憶がすべて「新側からは見えない状態」になり、ユーザー体験が激しく劣化します。
対策: 「シャドー書き込み → 並列読み → 切替 → 旧側削除」の4段で移行します。最初は旧側に書き続けながら、新側にも同じ内容を書き込みます。次に読み出しも新旧両方から行い、差分を計測します。差分が十分に小さくなったら読み出しを新側に一本化し、最後に旧側の削除ジョブを走らせます。
# シャドー書き込みの最小実装
def dual_write(user_id: str, key: str, value: str, source_turn_id: str) -> dict:
old = legacy_memory_write(user_id, key, value, source_turn_id) # 既存実装
try:
new = memory_write(user_id, key, value, source_turn_id) # 新実装
except Exception as e:
# 新側が落ちても旧側書き込みは成立させる
audit_log(event="shadow_write_failed", error=str(e), user_id=user_id, key=key)
new = {"ok": False, "error": str(e)}
return {"old": old, "new": new}
def dual_read(user_id: str, query: str) -> dict:
old_hits = legacy_memory_search(user_id, query)["hits"]
new_hits = memory_search(user_id, query)["hits"]
# 差分をメトリクス化(完全移行の判断材料にする)
metric.increment("memory.migration.old_only", len(set_keys(old_hits) - set_keys(new_hits)))
metric.increment("memory.migration.new_only", len(set_keys(new_hits) - set_keys(old_hits)))
# 移行期間中は旧側を正とする
return {"hits": old_hits}移行期間の目安は、日次アクティブユーザー×最長の会話間隔、あたりを基準に1〜2ヶ月取るのが実務的です。焦って1週間で切ると、長期休眠ユーザーの記憶が突然消えたように見えて、信頼を失います。
本番で一段上の設計にするための次の一歩
長期記憶は、入れたら終わりではなく、ずっと手を入れ続ける領域です。この記事で挙げた7つの落とし穴は、どれも「最初から完璧に設計しておく」ものではなく、「気づいたら直せる観測性を先に入れておく」ものだと捉えるのが実情に合います。
今日ひとつだけ着手するとしたら、私は「書き込みに source_turn_id を必ず付ける」ことをおすすめします。10 分もあれば既存実装に追加でき、以後すべての記憶について「どの会話から生まれたものか」を追えるようになります。破損の追跡、PII の検出、クロス汚染の調査、移行の差分計測——この記事で挙げたあらゆる対策の根になるのが、この一手です。
長期記憶