記憶を運用する GenAIOps 〜 mem0 の観測・評価・是正を Langfuse で回す

こんにちは。ガオ株式会社の黒澤です。この記事では、前回 mem0 を Google Cloud で動かした記事 の続きとして、溜めた記憶を「運用する」段階に踏み込みます。mem0 の内部処理を Langfuse でトレースし、古くなった記憶を評価スコアの低下として見つけ、update で直して回復するまでを実機で確認します。

メモリレイヤーは、入れて終わりの「機能」ではなく、保存された内容を観測し、評価し、修正し続ける「運用対象」です。今回の検証では、古くなった記憶によって既知情報との整合度が平均 5.00 から 2.00 に低下し、update 後に 5.00 へ戻りました。また、add 1回で抽出 LLM への入力が 8,307 トークン、処理全体が約10秒かかることも確認できました。

想定読者

• mem0 などのメモリレイヤーを本番投入しようとしていて、運用面が気になっている方
• エージェントの記憶が「効いているか」を計測したい方
• Langfuse でエージェントの観測・評価をしている方

この記事でわかること

• メモリレイヤーを「運用対象」として見るとは何か
• mem0 が内部で呼ぶ LLM・埋め込みを Langfuse で観測する方法（google-genai の OpenTelemetry 計装）
• 記憶の劣化を、LLM-as-a-judge のスコア低下として Langfuse で検知する方法
• history / update / delete による是正と変更履歴の実挙動

メモリレイヤーは「運用対象」である

#

前回の記事では「観測（オブザーバビリティ）はメモリレイヤーとは別の関心事」と切り分けました。記憶を溜める仕組みそのものの話だったからです。ただし本番運用に入ると、この2つは切り離せなくなります。

理由は、add が LLM で「覚えるべき事実」を自動抽出する仕組みだからです。便利な反面、抽出は必ずしも正しくありません。

• 誤抽出：会話の意図と違う事実を記憶する
• 陳腐化：以前は正しかった事実が古くなる（「React で開発中」がやがて事実でなくなる）
• 矛盾：新旧の記憶が食い違う
• PII 混入：覚えるべきでない個人情報が記憶に入る

これらが起きても、エラーは出ません。問題のある記憶が search で検索されプロンプトに差し込まれると、後続の応答でも古い情報や誤った情報が使われ続けます。

古くなった記憶や誤った記憶は自動では修正されません。そのため、保存内容を観測し、評価結果から問題を見つけ、必要に応じて修正する運用が必要です。本記事ではこの運用を GenAIOps のループとして次のように置きます。

graph LR
  A["観測
何を・どれだけのコストで
記憶したか"] --> B["評価
記憶は効いているか
を測る"]
  B --> C["検知
スコアの低下から
壊れた記憶を見つける"]
  C --> D["是正
update / delete"]
  D --> A

観測と評価には Langfuse を使います。トレース（観測）とスコア（評価）を1つの画面で追えるためです。エージェントの改善ループそのものの位置づけは、LLMOps とは何か の記事もあわせてご覧ください。

この記事で扱うこと・扱わないこと

#

扱うこと

冒頭に挙げた観測・評価・検知・是正を、ローカルの実機検証で確認します。

扱わないこと

• Cloud SQL での本番構築、add の非同期化（Cloud Tasks 等）といったインフラ運用（今後のテーマ）
• 大規模な評価データセットによる厳密なベンチマーク（本記事の評価は挙動確認レベルの小さなサンプル）
• Langfuse 自体の導入手順（セルフホストの記事 等を参照）

検証環境は前回の構成をベースに mem0ai を 2.0.12 へ更新し、観測・評価用のパッケージを足したものです。

観測：mem0 の裏側を Langfuse で見る

#

なぜ計装が必要か

#

Langfuse には mem0 専用のネイティブ連携がありません。Langfuse のメンテナも公式 Discussion でそう明言しています。

we do not have a native integration for this

（出典：mem0 integration（Langfuse GitHub Discussion #9748） ）

一方で、mem0 の add / search は内部で google-genai SDK を呼び、Vertex AI の Gemini に抽出と埋め込みを投げています。なお Vertex AI は現在 Gemini Enterprise Agent Platform に改称されています（公式の製品ページも「Gemini Enterprise Agent Platform (formerly Vertex AI) 」と表記）。ただし SDK や設定名では Vertex AI 表記が残るため、本記事でもそのまま Vertex AI と表記します。そこで、この google-genai の呼び出しを OpenTelemetry で計装し、スパン（呼び出し1回分の入出力・所要時間を記録した単位）を Langfuse に送ります。Langfuse は OpenTelemetry（OTLP）でスパンを受け取れるため、これで mem0 の裏側が Langfuse のトレースに出ます。

graph LR
  A["mem0.add() / search()"] --> B["google-genai SDK
（Gemini 抽出・埋め込み）"]
  B -.計装.-> C["GoogleGenAIInstrumentor
OTel スパン発行"]
  C --> D[("Langfuse")]

計装ライブラリには OpenInference の GoogleGenAIInstrumentor を使います。これは google-genai の呼び出しをラップして OpenTelemetry スパンに変換するもので、mem0 側のコードには手を入れません。前述の Discussion では、この引用に続けてメンテナが OpenLit 経由の計装を代替案として挙げていますが、今回は google-genai に特化した計装が用意されている OpenInference を選びました。OpenTelemetry スパンを Langfuse に送る構図はどちらも同じです。

つなぐ

#

必要なパッケージは Langfuse SDK と計装ライブラリです。前回の構成（mem0ai / google-genai / psycopg）に追加します。

uv add langfuse openinference-instrumentation-google-genai

版は前回同様 uv.lock で固定します（再現性のため）。なお mem0ai 2.0.12 は起動時に「spaCy が無い」という警告を出しますが（任意依存 mem0ai[nlp] 向け）、今回の検証では動作に影響ありませんでした。

環境変数で Langfuse セルフホストと Vertex AI を指定します。Langfuse はプロジェクト固有のキー（pk-lf / sk-lf）、Vertex AI は ADC（サービスアカウント鍵、ローカル開発なら gcloud auth application-default login）で認証し、Google の API キーは使いません。

# Langfuse セルフホスト
export LANGFUSE_BASE_URL=http://localhost:3000
export LANGFUSE_PUBLIC_KEY=pk-lf-...
export LANGFUSE_SECRET_KEY=sk-lf-...
# Vertex AI（前回と同じ）
export GOOGLE_GENAI_USE_VERTEXAI=true
export GOOGLE_CLOUD_PROJECT=<YOUR_PROJECT_ID>
export GOOGLE_CLOUD_LOCATION=global

計装は、アプリのコードより前に一度だけ有効化します。

from langfuse import get_client, observe
from openinference.instrumentation.google_genai import GoogleGenAIInstrumentor

# google-genai の呼び出しを計装（mem0 の内部呼び出しもこれで捕捉される）
GoogleGenAIInstrumentor().instrument()

langfuse = get_client()
assert langfuse.auth_check(), "Langfuse への疎通を確認"

あとは前回どおり Memory.from_config(...) で mem0 を組み、記憶操作を Langfuse の @observe でくくって呼ぶだけです。config は前回記事のものを、LLM だけ gemini-3.6-flash に変えて使います（前回の gemini-3.5-flash のままでは以下の数値になりません）。

from mem0 import Memory
from google import genai
import json, re

m = Memory.from_config(config)
client = genai.Client()

@observe(name="memory-add")
def remember(messages, user_id):
    return m.add(messages, user_id=user_id)

remember(
    [{"role": "user", "content": "私は名古屋在住で、週末はよくロードバイクに乗っています。"},
     {"role": "assistant", "content": "名古屋にお住まいで、ロードバイクが趣味なんですね。覚えておきます。"}],
    user_id="user1",
)
langfuse.flush()

実測：add の中身とコスト

#

実行すると、memory-add トレースの下に mem0 の内部呼び出しがぶら下がりました。1回の add の内訳は次のとおりです。

特に大きいのは、ユーザーの発話が1行なのに、抽出 LLM への入力が 8,000 トークンを超えていることです。これは mem0 の抽出プロンプト（システム指示や例示を含む）が大きいためで、add を呼ぶたびに毎回この入力が積まれます。前回「add は重い」と書いた中身が、トークン数とコスト（抽出 LLM だけで1回あたり約2.7円）として見えるようになりました。Langfuse が算出したコストは、Vertex AI の公開単価による検算とも一致します。8,307 × $1.50 ÷ 100万 + 771 × $7.50 ÷ 100万 = $0.018243 です（Gemini Enterprise Agent Platform 料金 。応答前の内部推論に使われる思考トークンは出力側に計上されます）。同じく前回「チャット応答のなかで同期実行すればユーザーを待たせる」と書いた懸念も、約10秒という実測で裏づけられます。

一点、制約があります。今回の計装では、埋め込み呼び出し（EmbedContent）はスパンとして現れるものの、モデル名とトークン数が Langfuse 側にマッピングされませんでした（モデル名は空欄、トークン数は 0 と表示）。抽出 LLM（GenerateContent）はトークン数まで取れています。埋め込みのコストまで正確に集計したい場合は、この差を踏まえて別途計上する必要があります。本記事に載せた add のコストが抽出 LLM 分のみなのも、この制約によります。

Langfuse のトレースには入力プロンプトと出力も残るため、「どの会話から何が抽出されたか」を後から追えます。抽出元の会話と保存された記憶を突き合わせられるので、誤抽出が起きた箇所を特定できます。

検知対象：壊れた記憶のパターン

#

検知の作りに入る前に、何を見つけたいのかを整理します。運用で実際に出る「壊れた記憶」のパターンです。

• 陳腐化：「React で開発中」のように、記録時点では正しくても後に古くなる事実。放置すると誤った前提で応答が続く。なお今回の 2.0.12 では、抽出された記憶に「as of July 2026」のように記録時点が付与される例が見られ、陳腐化を判断する手がかりになります
• 誤抽出・言語ゆれ：抽出はモデル依存で揺れます。検証では、同じ趣旨の会話が日本語で抽出される回と英語（“User is currently engaged in frontend software development using React …"）で抽出される回がありました。抽出言語を揃えたい場合は custom_instructions で固定します
• 矛盾：新旧の記憶が食い違う。search は両方を引きうる
• PII 混入：覚えるべきでない個人情報。マスキングの考え方は Langfuse の PII マスキング記事 が参考になります

これらは Langfuse のトレース（何が抽出されたか）と、次節の評価（記憶が応答をどう変えたか）の両面から検知します。

評価：記憶は効いているのか測る

#

記憶の良し悪しを毎回目視で判断するのは現実的ではありません。そこで、同じ質問を定期的に実行し、応答と現在のユーザー情報との整合度をスコアとして記録します。この節の目的は「記憶が効く」と示すことではありません（それは前回確認済みです）。記憶の劣化は最終的に応答の品質に現れるため、記憶単体の検査だけでなく、応答側の評価が必要です。

応答がユーザーの既知情報とどれだけ整合しているか（以下、整合度）を、LLM-as-a-judge（別の LLM 呼び出しに応答を採点させる評価手法）で 1〜5 に採点します。そのスコアを Langfuse の該当トレースに紐づける、という方法です。user1 には3つの事実（名古屋在住 / 週末ロードバイク / 人混みが苦手で静かな場所が好き）を add で記憶させておき、評価役の LLM にはユーザーの既知情報を渡して採点させます。評価役は回答と同じ gemini-3.6-flash の別呼び出しで、自己評価の偏りが入りうる簡易構成です。厳密に測る場合は別モデルを評価役に立てます。

評価は、観測とは別のコレクション（collection_name を分けた Memory インスタンス）で行いました。以下の m はその評価用インスタンス、client は genai.Client()、json / re は標準ライブラリです。まず既知情報を登録します。

for msg in [
    "私は名古屋在住です。",
    "週末はよくロードバイクで長距離を走っています。",
    "人混みが苦手で、静かな場所が好きです。",
]:
    m.add([{"role": "user", "content": msg}], user_id="user1")

採点基準は、評価役へのプロンプトにそのまま書きます（1=一般論のみ、5=既知情報を正しく反映、矛盾する内容は減点）。なおこの整合度は、記憶を回答に使えているか（活用度）と、使った記憶が現在の情報と一致しているか（正確性）をまとめて採点する簡易的な複合指標です。記憶なしの応答は、内容として妥当でも一般論であるだけで 1 点になります。両者を個別に測る評価設計ではありません。

@observe(name="answer")
def answer(question, use_memory):
    mems = []
    if use_memory:
        mems = [r["memory"] for r in m.search(question, filters={"user_id": "user1"})
                .get("results", []) if "memory" in r]
    prompt = (("このユーザーについて分かっていること:\n" + "\n".join(f"- {x}" for x in mems)
               + f"\n\n質問: {question}\n日本語で2〜3文で答えてください。")
              if mems else f"質問: {question}\n日本語で2〜3文で答えてください。")
    text = client.models.generate_content(model="gemini-3.6-flash", contents=prompt).text
    # 注入した記憶をトレースの Input に残す（既定では関数の引数がそのまま Input になる）
    langfuse.update_current_span(
        input={"question": question, "use_memory": use_memory, "memories": mems},
        output=text,
    )
    return {"text": text, "trace_id": langfuse.get_current_trace_id()}


KNOWN = "名古屋在住 / 週末ロードバイクで長距離 / 人混みが苦手で静かな場所が好き"

def judge(question, ans):
    prompt = (
        "あなたは、応答がユーザーの既知情報とどれだけ整合しているかを採点する評価者です。\n"
        f"ユーザーの現在の既知情報: {KNOWN}\n"
        f"質問: {question}\n応答: {ans}\n\n"
        "この応答が、ユーザーの現在の既知情報をどれだけ正しく踏まえているかを "
        "1〜5 で採点してください（1=一般論のみ, 5=既知情報を正しく反映, 既知情報と矛盾する内容は減点）。\n"
        'JSON のみで出力: {"score": <int>, "reason": "<20字程度>"}'
    )
    raw = client.models.generate_content(model="gemini-3.6-flash", contents=prompt).text
    mt = re.search(r"\{.*\}", raw, re.S)
    return json.loads(mt.group(0)) if mt else {"score": None, "reason": raw[:40]}
    # => {"score": 5, "reason": "既知情報を完全に反映しているため。"} のような返り値になる

なお簡潔さのため正規表現で JSON を抽出していますが、本番実装では Gemini の Structured Output（response_mime_type="application/json" と response_schema）でスキーマを固定します。

採点結果は、Langfuse のスコアとして各トレースに紐づけます。トレースとスコアが並んで表示されるため、「この応答は記憶を使ったか」「その結果スコアはどうだったか」を1画面で追えます。

question = "週末のおすすめの過ごし方を1つ提案して。"
a = answer(question, use_memory=True)
v = judge(question, a["text"])
langfuse.create_score(
    name="user_context_alignment",
    value=float(v["score"]),
    trace_id=a["trace_id"],
    data_type="NUMERIC",
    comment=v["reason"],
)

動作確認：記憶あり/なしで測る

#

まず、この採点がまともに機能するかを、記憶あり/なしの比較で確かめます。3問での結果です。

記憶なしは一般論（「ウォーキングや軽いランニング」「鎌倉がおすすめ」）に終始し、記憶ありは「木曽川沿いを抜けて岐阜の八百津・恵那方面へ向かう」ロングライドや「早朝の木曽川や庄内川沿いのサイクリングロード」と、既知情報を踏まえた応答になりました。スコアがこの差を正しく捉えています。

劣化を検知する

#

次に、記憶の陳腐化を意図的に起こし、スコアの低下として検知できるかを確認します。

user1 が名古屋から東京へ引っ越したものの、記憶は「名古屋に在住」のまま、という状態を作ります。実世界の情報だけが変わり、保存された記憶が更新されていないケースです。評価役には現在の既知情報（東京在住）を渡し、記憶ありの同じ3問を流します。

平均が 5.00 から 2.00 に落ちました。応答は古い記憶に引きずられて「名古屋から奥三河や岐阜の東濃エリアへ」と提案を続け、評価役のコメントには「東京在住と矛盾する名古屋発を提案したため」と減点理由が残ります。応答の文面だけを見れば自然な提案のため、目視では異変に気づけません。スコアの低下が異変を知らせ、トレースを開けば注入された記憶の中に原因（“User lives in Nagoya”）を特定できます。なお、この回の抽出は英語で保存されており、検知対象の節で述べた言語ゆれがここでも出ています。

今回実装したのはスコアの記録と手動での比較までで、閾値の監視や通知の自動化は含みません。実運用では、評価用のデータセットを用意して継続的に回すことになります。なお、評価役に渡した既知情報は、検証用ユーザーだから書けるものです。実ユーザーには一律の正解リストを用意しにくいため、本記事では正解を固定したテスト用ユーザーを定点観測しています。加えて、確認済みのプロフィール情報がある項目に限定した矛盾検査や、本番トレースのサンプリング評価を組み合わせます。データセットを使った評価や条件比較の進め方は、次の記事が参考になります。

• ragas 最新メトリクス解説と実践的改善ガイド
• Langfuse の Experiments を dataset から独立して使う
• LLM-as-a-judge のカテゴリカル採点

是正：update / delete と変更履歴

#

評価で特定した古い居住地の記憶を、update で現状に合わせて書き換えます。

mems = m.get_all(filters={"user_id": "user1"})["results"]
stale = next(r for r in mems if "名古屋" in r["memory"] or "Nagoya" in r["memory"])
m.update(stale["id"], text="ユーザーは東京に在住している（名古屋から引っ越した）。")
# => {"message": "Memory updated successfully!"}

是正後に同じ3問を再評価すると、平均は 2.00 から 5.00 に戻り、応答も「奥多摩や秩父方面へのロングライド」と新しい居住地を起点にした提案に変わりました。今回の構成で、評価スコアから対象の記憶を特定し、更新後の回復まで確認できました。

検知の節で挙げた「矛盾」も対処は同じです。新旧の記憶が食い違って並んでいる場合は、正しい方に update で寄せるか、誤った方を delete で消します。

なお、内容を渡す text 引数は 2.0.12 で data から改名されたものです（公式ドキュメント ）。data も非推奨エイリアスとして動きますが、実行時に「次のメジャーリリースで削除する」という警告が出ます。前回記事と同じ 2.0.7 で再現する場合は data= を使います。

mem0 に用意されている記憶ごとの操作は次のとおりです。

変更履歴が実際にどう残るかを確認するため、使い捨ての記憶で add → update → delete を一巡させます。

add = m.add([{"role": "user", "content": "私は今、Reactを使ってフロントを書いています。"},
             {"role": "assistant", "content": "React でフロント開発中なんですね。"}],
            user_id="dev1")
mem_id = add["results"][0]["id"]

# 事実が変わったので更新
m.update(mem_id, text="ユーザーは現在 Svelte でフロントエンドを書いている（以前は React）。")
# => {"message": "Memory updated successfully!"}

# 役目を終えたので削除
m.delete(mem_id)
# => {"message": "Memory deleted successfully!"}

各操作は、変更前後の内容つきで履歴に残ります。以下は要点のみの抜粋で、実際の各要素には id / memory_id / created_at / updated_at / role も入り、is_deleted は ADD / UPDATE の行にも false として付きます。

m.history(mem_id)

[
  { "event": "ADD",
    "old_memory": null,
    "new_memory": "User is currently engaged in frontend software development using React as of July 2026.",
    "actor_id": null },
  { "event": "UPDATE",
    "old_memory": "User is currently engaged in frontend software development using React as of July 2026.",
    "new_memory": "ユーザーは現在 Svelte でフロントエンドを書いている（以前は React）。",
    "actor_id": null },
  { "event": "DELETE",
    "old_memory": "ユーザーは現在 Svelte でフロントエンドを書いている（以前は React）。",
    "new_memory": null, "is_deleted": true, "actor_id": null }
]

delete の後も history には削除イベントが残ります（is_deleted: true）。「この記憶がいつ・どう変わり・いつ消えたか」を後から追えるため、監査ログを設計する際の土台になります。

ただし「誰が」は追えません。履歴の actor_id フィールドは、更新や削除を実行したオペレーターではなく、記憶に紐づく発話者・帰属を表すものです（会話メッセージの name から設定され、今回の構成では null。update / delete の履歴にも元の記憶の値が引き継がれるだけです）。誰がいつ操作したか、操作理由や承認情報が必要な場合は、アプリケーション側で別の監査ログを残す必要があります。

もう1つ、上の出力が示すとおり、delete した記憶の内容は old_memory として履歴に残り続けます。delete が消すのは pgvector 側の記憶本体で、DELETE イベントと削除前の内容は別の保存先（標準構成では ~/.mem0/history.db の SQLite）に残るためです。個人情報の完全消去要件に応じる場合は、検索対象の削除だけでは足りず、この履歴側を含めた削除・匿名化の方針を設計し、実挙動を確認する必要があります。なお 2.0.12 の履歴ストアには、ユーザー単位や記憶単位で履歴だけを消す API がありません。用意されているのは履歴テーブルごと破棄する reset() だけなので、ユーザー単位の消去が要件になる場合は、アプリ側で履歴テーブルを直接操作する設計が必要です。

運用観点でのまとめ

#

mem0 を本番で使う場合、記憶を保存する機能だけでなく、保存された内容を観測・評価・修正する運用まで用意する必要があります。今回の検証では、古い記憶によるスコアの低下を検知し、対象の記憶を update で直してスコアが回復するまでを確認できました。各段階でわかったことをまとめます。

今回の検証では、観測・評価・検知は Langfuse 上で行い、記憶の更新や削除は運用スクリプトから実行しました。是正作業が定常化する段階では、アプリ側に管理機能（管理 API や専用画面）を作る開発コストが発生します。マネージド版（mem0 Platform）にはダッシュボードが付属するため、OSS 版の選定にはこの自作コストを織り込んでおく必要があります。

本番化の前に、保持期間、削除範囲、PII 対策、操作監査を決める必要があります。ユーザー単位は delete_all(user_id=...) で削除できますが、完全消去には履歴ストアの扱いも含めて設計します（是正の節を参照）。記憶の共有範囲は前回記事で扱った user_id / agent_id のスコープ設計で決まります。保持期間と PII への備え（マスキングは前述のとおり）も、本記事で実測した範囲の外です。

導入済みの mem0 を観測するだけなら、前回の構成に langfuse と openinference-instrumentation-google-genai を追加するところから始められます。「つなぐ」の節のとおり、環境変数を設定し、GoogleGenAIInstrumentor().instrument() と Langfuse クライアントの初期化を済ませれば、add のトークン数と所要時間を確認できます。そのうえで定期評価と閾値監視を追加すれば、スコアの低下を運用上の確認対象にできます。

今回の評価は、3問の固定質問による挙動確認まででした。次は、評価用のデータセットを用意して継続的に回し、評価軸の分割（記憶の活用度と内容の正確性）や、記憶あり/なしのような条件比較を運用に組み込むところに踏み込みたいと思います。

関連記事（GAO ブログ）