AIインテリジェント・ボディ・メモリーの深層：コアコンセプトからLangGraphの実践まで

いつも会話の内容を忘れてしまい、毎回最初から話し始めなければならない友人とのコミュニケーションは、間違いなく非効率的で疲れるものだ。しかし、現在のほとんどのAIシステムでは、まさにこれが普通なのだ。それらは強力だが、一般的に重要な要素が欠けている：あんき.

記憶はオプションの付加物ではなく、真に学習し、進化し、協働できるAIインテリジェンス（エージェント）を構築するための中核となる基盤なのだ。

AIメモリーの中核概念

AIが抱える「無国籍」のジレンマ

同等 ChatGPT あるいは、コードアシスタントのようなツールは、その能力にもかかわらず、ユーザーに何度も何度も指示や好みを繰り返させる。この「記憶錯覚」は、大きな言語モデルとプロンプト・エンジニアリングのコンテクスト・ウィンドウによって生み出され、人々はAIがすでに記憶する能力を持っていると考えるようになった。

実際、大多数のAI知能は基本的に**ステートレス**である。そのため、過去の相互作用から学んだり、時間の経過とともに自らを適応させたりすることはできない。ステートレスツールから真に自律的なツールに進化するにはステートフル**知能は、単にコンテクストウィンドウを拡大したり、検索能力を最適化したりするのではなく、真の記憶メカニズムを備えていなければならないだろう。

AI知能における記憶とは一体何なのか？

AIインテリジェンスの文脈では、記憶とは、時間、タスク、セッションを超えて関連情報を保持し、思い出す能力のことである。これにより、インテリジェンスは過去のやりとりを記憶し、この情報を使って将来の行動を導くことができる。

真の記憶とは、チャットを保存する以上のものだ。それは、たとえ2つのやりとりが数週間離れていたとしても、インテリジェンスが下すすべての決断に影響を与える、継続的に進化する内部状態を構築することなのだ。

知的体の記憶は、3つの柱によって定義される：

• 州現在の相互作用に関する瞬時の情報を知覚する。
• 永続性セッションを超えた長期的な知識保持。
• 選択性長く覚えておく価値のある情報は何かを見極める。

これら3つの柱が一体となって、インテリジェンスが**継続性**を達成するための基礎となる。

コンテクスチュアル・ウィンドウ、RAGとメモリの違い

インテリジェンスの主流アーキテクチャにメモリを配置するための典型的なコンポーネントには、大規模言語モデル（LLM）、プランナー（例えば、以下のような）が含まれる。 ReAct フレームワーク）、ツールセット（API）、そしてリトリーバーだ。問題は、これらのコンポーネントのどれもが、昨日何が起こったかを覚えていないことだ。

コンテキストウィンドウ≠メモリー

よくある誤解は、増え続けるコンテクスト・ウィンドウが暗記に完全に取って代わるというものだ。しかし、このアプローチには明らかなボトルネックがある：

• コスト高詳細 トークン は、高い計算コストと待ち時間を意味する。
• 限られた情報セッションをまたいだ永続的な保存は不可能であり、セッションが終了するとすべてのコンテクスト情報は失われる。
• 優先順位の欠如すべての情報は文脈の中で同等の地位を持ち、重要な情報と一過性の情報を区別することはできない。

コンテキスト・ウィンドウは、以下のようなインテリジェンスを保証します。単一セッション内首尾一貫性。クロスセッションインテリジェンス。

RAG≠メモリー

検索強化の生成(RAG)やメモリーシステムもLLMに影響を与えるが、両者が解決する問題はまったく異なる。RAG 推論を行う際、事実情報は外部の知識ベース（文書やデータベースなど）から検索され、答えの精度を高めるために手がかりとなる言葉が注入される。しかし RAG それ自体はステートレスであり、ユーザーのアイデンティティを気にせず、現在のクエリと過去の会話を関連付けることはできない。

一方、記憶は継続性をもたらす。ユーザーの好み、過去の決断、成功や失敗を記録し、その情報を今後のインタラクションに活用する。

簡単に説明する：RAG 知性はより正確に答えることを助け、記憶は知性がより知的に行動することを助ける。 成熟したインテリジェンスには、その両方が必要なのだ：RAG そして記憶が彼らの内的行動を形成する。

この3つをより明確に区別するために、以下の表を参照されたい：

AI メモリーの種類

AI知能における記憶は、短期記憶と長期記憶の2つの基本形態に分けられ、それぞれ異なる認知機能を果たす。

• 短期記憶人間のワーキングメモリーのように、1回のインタラクションでコンテクストを保存する。
• 長期記憶学習と適応のために、セッション、タスク、時間を超えて知識を保持する。

短期記憶

短期記憶はワーキングメモリーとも呼ばれ、AIシステムにとって最も基本的な記憶形態であり、以下のような即時的な文脈を保持するために使われる：

• 歴史との対話最近のメッセージとその順序
• 中間状態タスク実行中の一時変数。
• 注目現在の対話の核心。

長期記憶

より複雑なAIアプリケーションでは、セッションをまたいだ知識の保持とパーソナライゼーションのために長期記憶を実装する必要がある。長期記憶には主に3つのタイプがある：

1.手続き記憶

これは知性の "筋肉の記憶 "であり、知性は次のことを知っている。やり方何か。インテリジェンスが実行できるアクションを定義し、ロジックに直接エンコードします。例えば、「技術的な質問に答えるときは、より詳しい説明の仕方をする」とか、「特定の上司からのメールを常に優先する」などです。

2.エピソード記憶

これはインテリジェンスの「インタラクション・アルバム」であり、特定のユーザーに関するインタラクションや特定のイベントの履歴を記録する。これは、パーソナライゼーション、継続性、長期的学習の鍵である。例えば、インテリジェンスは、「前回、顧客が請求の問題について問い合わせた際、対応の遅さが不満につながった」ことを記憶し、将来的に対応戦略を最適化できるようにすることができる。

3.意味記憶

これはインテリジェンスの「事実百科事典」であり、世界とユーザーに関する客観的な事実を保存している。この知識は通常、ベクトル検索や RAG 検索は特定のインタラクションとは無関係に実行される。例えば、"Alice is the technical lead for project A "や "John's work time zone is PST "など。

AIメモリ実装のメカニズム

メモリ書き込みのタイミング

知能はいつ、どのようにして新しい記憶を作り出すのか？主流のアプローチは2つある：

1. リアルタイム同期書き込みユーザーとのインタラクションに応じて、リアルタイムでメモリを作成し、更新する。このアプローチでは、新しいメモリを即座に有効にすることができますが、メインのアプリケーション・ロジックに待ち時間と複雑さを追加する可能性があります。例えばChatGPT 同様のメカニズムが save_memories このツールは、ダイアログで情報を保存するかどうかをリアルタイムで判断する。
2. 非同期バックグラウンド処理メモリーの作成と照合は、別々のバックグラウンド・タスクとして処理する。このアプローチは、メモリ管理をメインのアプリケーション・ロジックから切り離し、待ち時間を短縮し、情報の要約や重複排除などのより複雑なオフライン処理を可能にする。

記憶のマネジメント戦略

ダイアログが長すぎると、短期記憶がLLMのコンテクストウィンドウの限界を超えることがある。一般的な管理方法には次のようなものがある：

• トリミング対話履歴の最初か最後のN個のメッセージを削除する。
• 概要先のダイアログを要約し、原文を要約に置き換える。
• 選択的忘却事前に設定されたルール（重要でないシステムメッセージのフィルタリングなど）やモデルの判断に基づき、メモリの一部を永久的に削除。

AIメモリ実現への挑戦

記憶力は知性にとって極めて重要だが、実際には多くの課題に直面している：

• 記憶の過負荷と忘却メカニズム無限に蓄積される記憶は、システムを肥大化させ非効率にする。陳腐化した情報や無関係な情報を忘れることができるような、効果的な忘却メカニズムを設計することが課題である。
• 情報汚染偽の入力や悪意のある入力は長期記憶に保存され、知能の意思決定システムを「汚染」し、将来の行動に偏りをもたらす可能性がある。
• プライバシーとセキュリティ: プリファレンス、履歴、個人データなど、機密性の高いユーザー情報は長期メモリに大量に保存されます。このデータの暗号化、アクセス制御、コンプライアンスをいかに確保するかが、アプリケーションを軌道に乗せるための鍵となる。
• コストとパフォーマンス大規模なメモリ・ストアを維持し、複雑な検索や書き込み操作を行うには、計算コストやストレージ・コストがかかり、システムの応答性に影響を与える可能性がある。

LangGraphを使って記憶を持つ知的ボディを作る

LangGraph は、ループとステート管理を自然にサポートする、ステートフルでマルチインテリジェンスなアプリケーションを構築するための強力なフレームワークであり、複雑なメモリーシステムの実装に最適である。以下のコード例では LangGraph 短期記憶と長期記憶を知能に加える。

短期記憶の追加

短期記憶（スレッドレベルの永続性）は、インテリジェンスが1つの複数ラウンドの対話を追跡することを可能にする。のための短期記憶を提供することによって可能になる。 LangGraph チェックポインターを設定することで、ステートの自動保存とロードを容易にする。

次の例では、メモリ・チェックポイントを使用しています。 InMemorySaver でセッションの状態を保存する。

from langchain_community.chat_models import ChatAnthropic # 假设使用Anthropic模型
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.memory import InMemorySaver
# 为简化，假设模型已初始化
# model = ChatAnthropic(model="claude-3-5-sonnet-20240620")
# 实际使用时请替换为真实模型
class FakeModel:
def invoke(self, messages):
from langchain_core.messages import AIMessage
last_message = messages[-1].content
if "bob" in last_message:
return AIMessage(content="Hi Bob! How can I help you today?")
elif "my name" in last_message:
return AIMessage(content="Your name is Bob.")
else:
return AIMessage(content="Hello there!")
model = FakeModel()
def call_model(state: MessagesState):
response = model.invoke(state["messages"])
return {"messages": [response]}
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
# 使用唯一的 thread_id 来标识一个独立的对话线程
config = {
"configurable": {
"thread_id": "user_1_thread_1"
}
}
# 第一次交互
for chunk in graph.stream(
{"messages": [{"role": "user", "content": "Hi! I'm Bob."}]},
config,
stream_mode="values",
):
chunk["messages"][-1].pretty_print()
# 第二次交互，在同一线程中
for chunk in graph.stream(
{"messages": [{"role": "user", "content": "What's my name?"}]},
config,
stream_mode="values",
):
chunk["messages"][-1].pretty_print()

================================== Ai Message ==================================
Hi Bob! How can I help you today?
================================== Ai Message ==================================
Your name is Bob.

本番環境では InMemorySaver データベースのバックエンドを以下のような永続的なものに置き換える。 PostgresSaver もしかしたら MongoDBSaverサービスを再起動してもダイアログの状態が失われないようにするためである。

経営のチェックポイント

LangGraph チェックポイント（つまりセッションの状態）を管理するためのインターフェイスを提供する。

スレッドの現在の状態を表示します：

# graph.get_state(config)

スレッドのすべての履歴ステータスを表示します：

# list(graph.get_state_history(config))

長期記憶の追加

長期メモリ（クロス・スレッド・パーシステンス）は、ユーザーのプリファレンスなど、長期間保持する必要があるデータを保存するために使用される。LangGraph コンパイルされたグラフに store オブジェクトを使用する。

次の例では InMemoryStore ユーザー情報を保存し、異なる対話スレッド間で共有する。

import uuid
from typing_extensions import TypedDict
from langchain_core.runnables import RunnableConfig
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
# ... (复用之前的模型和 State 定义) ...
def call_model_with_long_term_memory(
state: MessagesState,
config: RunnableConfig,
*,
store: BaseStore,
):
from langchain_core.messages import AIMessage, SystemMessage
user_id = config["configurable"]["user_id"]
namespace = ("memories", user_id)
# 检索长期记忆
memories = store.search(query="", namespace=namespace)
info = "\n".join([d.value["data"] for d in memories])
system_msg_content = f"You are a helpful assistant. User info: {info}"
# 写入长期记忆
last_message = state["messages"][-1]
if "remember" in last_message.content.lower():
memory_to_save = "User's name is Bob"
# 使用唯一ID存储记忆
store.put([(str(uuid.uuid4()), {"data": memory_to_save})], namespace=namespace)
# 调用模型
response = model.invoke([SystemMessage(content=system_msg_content)] + state["messages"])
return {"messages": [response]}
builder_ltm = StateGraph(MessagesState)
builder_ltm.add_node("call_model", call_model_with_long_term_memory)
builder_ltm.add_edge(START, "call_model")
checkpointer_ltm = InMemorySaver()
store = InMemoryStore()
graph_ltm = builder_ltm.compile(
checkpointer=checkpointer_ltm,
store=store,
)
# 在线程1中，用户要求记住名字
config1 = {
"configurable": {
"thread_id": "thread_1",
"user_id": "user_123",
}
}
for chunk in graph_ltm.stream(
{"messages": [{"role": "user", "content": "Hi! Please remember my name is Bob"}]},
config1,
stream_mode="values",
):
chunk["messages"][-1].pretty_print()
# 在新的线程2中，查询名字
config2 = {
"configurable": {
"thread_id": "thread_2",
"user_id": "user_123", # 相同的 user_id
}
}
for chunk in graph_ltm.stream(
{"messages": [{"role": "user", "content": "What is my name?"}]},
config2,
stream_mode="values",
):
chunk["messages"][-1].pretty_print()

================================== Ai Message ==================================
Hi Bob! How can I help you today?
================================== Ai Message ==================================
Your name is Bob.

セマンティック検索を有効にする

を提供する。 store ベクトル埋め込みモデルを設定することで、意味的類似性に基づく記憶検索が可能になり、知能がより賢く関連情報を呼び出せるようになる。

from langchain_openai import OpenAIEmbeddings
# 假设已初始化 OpenAIEmbeddings
# embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# store = InMemoryStore(
#     index={
#         "embed": embeddings,
#         "dims": 1536, # 嵌入维度
#     }
# )
# 
# store.put([
#     ("1", {"text": "I love pizza"}),
#     ("2", {"text": "I am a plumber"})
# ], namespace=("user_123", "memories"))
# 在调用时，通过 query 参数进行语义搜索
# items = store.search(
#     query="I'm hungry",
#     namespace=("user_123", "memories"),
#     limit=1
# )
# 搜索结果将是与 "I'm hungry" 最相关的 "I love pizza"。

この機能により、インテリジェンスは単にキーワードのマッチングを行うのではなく、対話の文脈に基づいてメモリーバンクから最も関連性の高い情報を動的に取り出すことができる。

ケーススタディ：メモリーを強化したメール・インテリジェンス・ボディの構築

以下では、複数のタイプのメモリを持つメール処理インテリジェンスを構築することによって、この理論を実践する。

7.1 インテリジェントボディ状態の定義

まず、短期記憶と長期記憶の両方を含む、知能が把握する必要のある状態を定義する。

from typing import TypedDict, Dict, Any
from langgraph.graph import MessagesState
class EmailAgentState(TypedDict, total=False):
"""邮件智能体的状态定义"""
messages: MessagesState  # 短期记忆：对话历史
user_preferences: Dict[str, Any]  # 长期语义记忆：用户偏好
email_threads: Dict[str, Any]  # 长期情景记忆：历史邮件线程
action: str # 决策动作
reply: str # 生成的回复

7.2 決定分類器の実装（状況記憶の使用）

メールの内容と過去のやり取り（状況記憶）に基づいて次のアクションを決定する分類器ノードを作成する。

def classify_email(state: EmailAgentState):
# 此处应调用 LLM 进行分类
# 为简化，直接返回一个硬编码的决策
email_content = state["messages"][-1].content
if "urgent" in email_content.lower():
return {"action": "escalate"}
return {"action": "reply"}

7.3 道具の定義（意味記憶を使う）

例えば、意味記憶からの連絡先情報など。

from langgraph.prebuilt import ToolNode
def get_contacts_from_memory(user_id: str):
# 模拟从用户数据库或语义记忆中获取联系人
contacts_db = {
"user_123": {
"developers": ["alice@company.com"],
"managers": ["charlie@company.com"]
}
}
return contacts_db.get(user_id, {})
def send_email(to: str, subject: str, body: str):
# 模拟发送邮件API
print(f"Email sent to {to} with subject '{subject}'")
return "Email sent successfully."
tools = [get_contacts_from_memory, send_email]
tool_node = ToolNode(tools)

7.4 グラフの構築とノードの接続

利用する LangGraph すべてのコンポーネントを実行可能なワークフローに接続します。

from langgraph.graph import StateGraph, START, END
builder = StateGraph(EmailAgentState)
# 1. 定义节点
builder.add_node("classify", classify_email)
builder.add_node("generate_reply", lambda state: {"reply": "Generated reply based on user tone."}) # 简化版
builder.add_node("escalate_task", lambda state: print("Task Escalated!"))
# 2. 定义边
builder.add_edge(START, "classify")
builder.add_conditional_edges(
"classify",
lambda state: state["action"],
{
"reply": "generate_reply",
"escalate": "escalate_task",
}
)
builder.add_edge("generate_reply", END)
builder.add_edge("escalate_task", END)
# 3. 编译图
email_graph = builder.compile()

この単純化された例は、（状態やツールを介して）異なるタイプの記憶を組み合わせることで、インテリジェンスのワークフローをどのように推進できるかを示している。このフレームワークを拡張することで、複雑なタスクを処理でき、高度にパーソナライズされ、継続的な学習が可能な強力なAIアシスタントを構築することができる。

以下の質問は、自分の記憶力を高めるインテリジェンスを設計するときに考える効果的なガイドとなる：

1. 学習内容知能はどのような情報を学ぶべきか？それは客観的事実なのか、歴史的出来事の要約なのか、それとも行動の規則なのか？
2. 思い出の時間メモリー形成はいつ、誰がトリガーすべきなのか？リアルタイムのインタラクションなのか、バックグラウンドのバッチ処理なのか？
3. ストレージ・ソリューション思い出をどこに保存すべきか？ローカルのデータベースか、ベクターストレージか、それともクラウドサービスか？
4. セキュリティとプライバシーメモリーのデータを漏洩や悪用から守るには？
5. アップデートとメンテナンス時代遅れの記憶や誤った記憶が、インテリジェンスの意思決定を汚染するのを防ぐにはどうすればいいのでしょうか？