「AIエージェントキャッチアップ #22 - LangMem (LangChain)」を開催しました

from langmem import create_memory_manager
from pydantic import BaseModel

class Triple(BaseModel):
    """Store all new facts, preferences, and relationships as triples."""
    subject: str
    predicate: str
    object: str
    context: str | None = None

manager = create_memory_manager(  
    "anthropic:claude-3-5-sonnet-latest",
    schemas=[Triple], 
    instructions="Extract user preferences and any other useful information",
    enable_inserts=True,
    enable_deletes=True,
)

この例ではTripleというクラスを定義し、「主語」「述語」「目的語」の組み合わせを記憶していくようにしています。

Semantic Memoryのサンプルを動かしてみると、記憶のレコードが追加・更新・削除されていく様子を確認できます。

このcreate_memory_managerでは、enable_inserts=Trueとしている箇所が、追記型で記憶していくという意味のようです。もしも人物のプロフィールなどを更新型で記憶していく場合は、enable_inserts=Falseと設定するようです。

このようにして作成された記憶のレコードを、プロンプトに入れてあげることになります。

def prompt(state):
    """Prepare messages with context from existing memories."""
    memories = store.search(
        ("memories",),
        query=state["messages"][-1].content,
    )
    system_msg = f"""You are a memory manager. Extract and manage all important knowledge, rules, and events using the provided tools.



Existing memories:
<memories>
{memories}
</memories>

Use the manage_memory tool to update and contextualize existing memories, create new ones, or delete old ones that are no longer valid.
You can also expand your search of existing memories to augment using the search tool."""
    return [{"role": "system", "content": system_msg}, *state["messages"]]

Episodic Memory

Episodic Memoryの場合も、おおまかな流れは同じです。

langchain-ai.github.io

大きく異なるのは、プロンプトにFew-shotのような形式で入れるという点のようです。

    # Step 1: Find similar past episodes
    similar = store.search(
        ("memories", "episodes"),
        query=messages[-1]["content"],
        limit=1,
    )

    # Step 2: Build system message with relevant experience
    system_message = "You are a helpful assistant."
    if similar:
        system_message += "\n\n### EPISODIC MEMORY:"
        for i, item in enumerate(similar, start=1):
            episode = item.value["content"]
            system_message += f"""

Episode {i}:
When: {episode['observation']}
Thought: {episode['thoughts']}
Did: {episode['action']}
Result: {episode['result']}
        """

Procedural Memory

Procedural Memoryの場合、システムプロンプトを自動更新するような挙動になります。

そこで、LangMemにはプロンプトの自動最適化の機能が用意されています。

langchain-ai.github.io

create_prompt_optimizer関数でプロンプトのoptimizerを準備し、ユーザーとの過去のやり取り（trajectories）と現在のプロンプトをもとに、プロンプトを自動で更新することになります。

optimizer = create_prompt_optimizer(
    "anthropic:claude-3-5-sonnet-latest",
    kind="metaprompt",
    config={"max_reflection_steps": 1, "min_reflection_steps": 0},
)
updated = optimizer.invoke(
    {"trajectories": trajectories, "prompt": "You are a planetary science expert"}
)