[React] agent 의 memories 를 sematic search 로 검색하기

---

• 개요
• 메모리 저장소 생성
• 메모리에 저장하기
• NLP 를 이용하여 검색하기
• Agent 에서 사용하기
• create_react_agent 활용하기
• Advanced Usage
  • Multi-vector indexing
• 저장시 Fields 를 Override 하기
• 특정 Memory 에 대한 인덱싱 비활성화
• Reference

개요

LangGraph 에이전트의 메모리 저장소에서 시멘틱 검색 기능을 활성화하는 방법을 설명한다. 이를 통해 단순 키워드 일치가 아닌, 의미 기반으로 메모리를 검색할 수 있다.

메모리 저장소 생성

Semantic 검색 기능이 가능한 메모리 생성

from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore

embedings = init_embeddings('huggingface:BAAI/bge-m3')
store = InMemoryStore(
    index={
        'embed': embedings,
        'dims' : 1024
    }
)

기본적으로 저장소는 시멘틱/벡터 검색이 비활성화된 상태이다. 저장소를 생성할 때 인덱스 구성을 생성자에게 전달하면 인덱싱을 활성화할 수 있다. 만약, 저장소 클래스가 이 인터페이스를 구현하지 않았거나 인덱스 구성을 전달하지 않았다면, 시맨틱 검색은 비활성화되고 put 또는 aput 에 전달된 index 인자는 아무런 효과가 없다.

메모리에 저장하기

• namespace : name space -> Tuple
• key : str : 분류 할 수 있는 key
• value : Dict : index 값들
• index : index 를 지정하거나 , False 을 통해 지정하지 않을 수도 있다.

store.put(('user_123','memories'),'1',{'text': 'I love pizza'})
store.put(('user_123','memories'),'2',{'text': 'I prefer Italian food'})
store.put(('user_123','memories'),'3',{'text': "I don't like spicy food"})
store.put(('user_123','memories'),'3',{'text': 'I am studying econometrics'})
store.put(('user_123','memories'),'3',{'text': 'I am a plumber'})

NLP 를 이용하여 검색하기

memories = store.search(('user_123', 'memories'), query="I'm hungry", limit=3)

for memory in memories:
    print(f"Memory: {memory.value['text']} (similarity: {memory.score})")

Memory: I prefer Italian food (similarity: 0.6429529122454709) 
Memory: I am a plumber (similarity: 0.6412285135022295) 
Memory: I love pizza (similarity: 0.6045187378022228)

• similarity 가 검색되어 결과가 출력된다.

Agent 에서 사용하기

from typing import Optional
from langgraph.store.base import BaseStore
from langgraph.graph import START, MessagesState, StateGraph

llm = model

def chat(state, *, store: BaseStore):
    # Search based on user's last message
    items = store.search(
        ('user_123', 'memories'), query=state['messages'][-1].content, limit=2
    )

    memories = '\n'.join(item.value['text'] for item in items)
    memories = f"## Memories of user\n{memories}" if memories else ""
    response = llm.invoke(
        [
            {'role': 'system' , 'content': f"You are a helpful assistant.\n{memories}"},
            *state['messages'],
        ]
    )
    return {'messages' : [response]}

builder = StateGraph(MessagesState)
builder.add_node(chat)
builder.add_edge(START, 'chat')
graph = builder.compile(store=store)

for message, metadata in graph.stream(
    input = {'messages' : [{'role': 'user' , 'content' : "I'm hungry"}]},
    stream_mode='messages'
):
    print(message.content, end = "")

• items 에서 similarity 가 높은 2개가 선택되어invoke 의 question 에 포함된다.

Since you mentioned that you prefer Italian food, how about trying some pasta or pizza for lunch? If you're looking for something quick and easy, maybe a margherita pizza or spaghetti with marinara sauce would be perfect. Do you have any favorite spots nearby, or should I suggest some places to order from?

create_react_agent 활용하기

import uuid
from typing import Optional

from langgraph.prebuilt import InjectedStore
from langgraph.store.base import BaseStore
from typing_extensions import Annotated

from langgraph.prebuilt import create_react_agent

def prepare_messages(state, *, store: BaseStore):
    items = store.search(
        ("user_123", "memories"), query=state['messages'][-1].content, limit=2
    )
    memories = "\n".join(item.value['text'] for item in items)
    memories = f"## Memories of user\n{memories}" if memories else ""

    return [
        {"role": "system", "content" : f"You are a helpful assistant.\n{memories}"}
    ] + state['messages']

def upsert_memory(
        content: str,
        *,
        memory_id: Optional[uuid.UUID] = None,
        store: Annotated[BaseStore, InjectedStore],
):
    """Upsert a memory in the database"""
    # The LLM can use this tool to store a new memory
    mem_id = memory_id or uuid.uuid4()
    store.put(
        ("user_123", "memories"),
        key = str(mem_id),
        value = {'text': content}
    )

    return f"Stroed memory {mem_id}"

agent = create_react_agent(
    model,
    tools=[upsert_memory],
    prompt=prepare_messages,
    store=store
)

for msg, metadata in agent.stream(
    input={"messages" : [{"role": "user", "content" : "I'm hungry"}]},
    stream_mode="messages",
):
    print(msg.content, end ="")

Since you prefer Italian food, how about we find some Italian restaurants nearby? Would you like me to help with that? If so, I'll need your location to proceed. Alternatively, if you're looking for a quick meal, there might be some Italian dishes that are easy to make at home. Let me know what you'd prefer!

• prompt function 은 LLM 이 call 하기 직전에 항상 실행된다.

Advanced Usage

Multi-vector indexing

Memory 의 서로 다른 측면을 개별적으로 저장하고 검색하여 검색을 향상시키거나 특정 필드를 인덱싱 대상에서 제외할 수 있다.

from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore

embeddings = init_embeddings('huggingface:BAAI/bge-m3')

# Configure store to embed both memory content and emotional context
store = InMemoryStore(
    index = {'embed' : embeddings , 'dims': 1024, "fields": ["memory", "emotional_context"]}
)

# Store memories with different content/emotion pairs
store.put(
    ("user_123", "memories"),
    "mem1",
    {
        "memory" : "Had pizza with friends at Mario's",
        "emotional_context" : "felt happy and connected",
        "this_isnt_idexed" : "I perfer ravioli though"
    }
)
  
store.put(
    ("user_123", "memories"),
    "mem2",
    {
        "memory": "Ate alone at home",
        "emotional_context": "felt a bit lonely",
        "this_isnt_indexed": "I like pie",
    },
)

• mem2 와 일치

results = store.search(
    ("user_123","memories"), query="times they felt isolated", limit=1
)
print("Expect mem2")
for r in results:
    print(f"Item: {r.key}; Score ({r.score})")
    print(f"Memory: {r.value['memory']}")
    print(f"Emotion: {r.value['emotional_context']}\n")

Expect mem2 
Item: mem2; Score (0.7748096097731049) 
Memory: Ate alone at home 
Emotion: felt a bit lonely

• mem1 과 일치

print("Expect mem1")
results = store.search(("user_123", "memories"), query="fun pizza", limit=1)
for r in results:
    print(f"Item: {r.key}; Score ({r.score})")
    print(f"Memory: {r.value['memory']}")
    print(f"Emotion: {r.value['emotional_context']}\n")

Expect mem1 
Item: mem1; Score (0.6843336968592125) 
Memory: Had pizza with friends at Mario's 
Emotion: felt happy and connected

• 무작위의 낮은 점수
• ravioli 는 indexing 되지 않았기 때문에 검색이 되지 않는다.

print("Expect random lower score (ravioli not indexed)")
results = store.search(("user_123", "memories"), query="ravioli", limit=1)

for r in results:
    print(f"Item: {r.key}; Score ({r.score})")
    print(f"Memory: {r.value['memory']}")
    print(f"Emotion: {r.value['emotional_context']}\n") ## fields 에 포함시켜준 애만 Indexing이 된다.

Expect random lower score (ravioli not indexed) 
Item: mem1; Score (0.536263103505899) 
Memory: Had pizza with friends at Mario's 
Emotion: felt happy and connected

저장시 Fields 를 Override 하기

특정 기억을 저장할 때, store의 기본 설정과 관계없이 put(..., index=[...fields]) 를 사용하여 어떤 필드를 임베딩할지 재정의 할수 있다.

store = InMemoryStore(
    index=(
        {
            'embed': embeddings,
            'dims' : 1024,
            'fileds' : ['memory'],
        }
    )
)

# Store one memory with default indexing
store.put(
    ("user_123", "memories"),
    "mem1",
    {"memory": "I love spicy food", "context": "At a Thai restaurant"},
)

# Store another overriding which fields to embed
store.put(
    ("user_123", "memories"),
    "mem2",
    {"memory": "The restaurant was too loud", "context": "Dinner at an Italian place"},
    index=["context"],  # Override: only embed the context
)

# Search about food - matches mem1 (using default field)
print("Expect mem1")
results = store.search(
    ("user_123", "memories"), query="what food do they like", limit=1
)
for r in results:
    print(f"Item: {r.key}; Score ({r.score})")
    print(f"Memory: {r.value['memory']}")
    print(f"Context: {r.value['context']}\n")

Expect mem1 Item: mem2; Score (0.6082269820388908) 
Memory: The restaurant was too loud 
Context: Dinner at an Italian place

• 기본 필드 내에 음식에 관한 이야기가 있음

# Search about restaurant atmosphere - matches mem2 (using overridden field)
print("Expect mem2")
results = store.search(
    ("user_123", "memories"), query="restaurant environment", limit=1
)
for r in results:
    print(f"Item: {r.key}; Score ({r.score})")
    print(f"Memory: {r.value['memory']}")
    print(f"Context: {r.value['context']}\n")

Expect mem2 Item: mem2; Score (0.6742498352381826) 
Memory: The restaurant was too loud 
Context: Dinner at an Italian place

• 레스토랑 분위기에 대한 검색으로 재정의된 필드를 사용하였다.

특정 Memory 에 대한 인덱싱 비활성화

일부 mEMORY 는 내용 기반으로 검색되지 않아야 한다. 이러한 경우 put(..., index=False) 를 사용해 인덱싱은 비활성화하고 저장만 할 수 있다.

store = InMemoryStore(index={"embed": embeddings, "dims": 1536, "fields": ["memory"]})

# Store a normal indexed memory
store.put(
    ("user_123", "memories"),
    "mem1",
    {"memory": "I love chocolate ice cream", "type": "preference"},
)
  
# Store a system memory without indexing
store.put(
    ("user_123", "memories"),
    "mem2",
    {"memory": "User completed onboarding", "type": "system"},
    index=False,  # Disable indexing entirely
)

# Search about food preferences - finds mem1
print("Expect mem1")
results = store.search(("user_123", "memories"), query="what food preferences", limit=1)
for r in results:
    print(f"Item: {r.key}; Score ({r.score})")
    print(f"Memory: {r.value['memory']}")
    print(f"Type: {r.value['type']}\n")

• 음식 선호도에 관한 기억을 찾음

# Search about onboarding - won't find mem2 (not indexed)
print("Expect low score (mem2 not indexed)")
results = store.search(("user_123", "memories"), query="onboarding status", limit=1)
for r in results:
    print(f"Item: {r.key}; Score ({r.score})")
    print(f"Memory: {r.value['memory']}")
    print(f"Type: {r.value['type']}\n")

• 온보딩에 관한 검색은 인덱싱이 되지 않아서 발견하지 못하였다.

• 코드 주소

---

Reference

• How to add semantic search to your agent’s memory