python: 3.12.4
langchain: 0.3.1
langchain-openai: 0.2.1
langgraph-checkpoint-postgres: 1.0.9
langgraph: 0.2.39
Checkpointer: グラフの状態のsnapshot, 会話スレッド単位(=thread_id)で管理BaseCheckpointSaverを継承
Store: 会話スレッドをまたいだ記憶保持するKVS
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
user_id = "1"
ns = (user_id, "memories")
memory_id = str(uuid.uuid4())
memory = {"food_preference" : "I like pizza"}
store.put(ns, memory_id, memory)
memories = store.search(ns)
memories[-1].dict()
{
'value': {'food_preference': 'I like pizza'},
'key': '07e0caf4-1631-47b7-b15f-65515d4c1843',
'namespace': ['1', 'memories'],
'created_at': '2024-10-02T17:22:31.590602+00:00',
'updated_at': '2024-10-02T17:22:31.590605+00:00'
}- 埋め込みでも検索出来る
from langchain.embeddings import init_embeddings
store = InMemoryStore(
index={
"embed": init_embeddings("openai:text-embedding-3-small"), # Embedding provider
"dims": 1536, # Embedding dimensions
"fields": ["food_preference", "$"] # Fields to embed
}
)
memories = store.search(
ns,
query="What does the user like to eat?",
limit=3 # Return top 3 matches
)- v0.2系の
langchain.memory.entity.ConversationEntityMemoryはdeprecated-
Please see long-term memory agent tutorial implements an agent that can extract structured information from the conversation history.
- https://python.langchain.com/docs/versions/migrating_memory/long_term_memory_agent/
-
- agentによって情報をknowledge triples
(subject, predicate, object)の形で取り出して永続化する - 記憶は埋め込みで保持される
import json
from typing import List, Literal, Optional
import tiktoken
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_core.messages import get_buffer_string
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import ChatOpenAI
from langchain_openai.embeddings import OpenAIEmbeddings
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode
recall_vector_store = InMemoryVectorStore(OpenAIEmbeddings())
class KnowledgeTriple(TypedDict):
subject: str
predicate: str
object_: str
@tool
def save_recall_memory(memory: str, config: RunnableConfig) -> str:
"""Save memory to vectorstore for later semantic retrieval."""
user_id = get_user_id(config)
document = Document(
page_content=memory, id=str(uuid.uuid4()), metadata={"user_id": user_id}
)
recall_vector_store.add_documents([document])
return memory
@tool
def save_recall_memory(memories: List[KnowledgeTriple], config: RunnableConfig) -> str:
"""Save memory to vectorstore for later semantic retrieval."""
user_id = get_user_id(config)
for memory in memories:
serialized = " ".join(memory.values())
document = Document(
serialized,
id=str(uuid.uuid4()),
metadata={
"user_id": user_id,
**memory,
},
)
recall_vector_store.add_documents([document])
return memories
@tool
def search_recall_memories(query: str, config: RunnableConfig) -> List[str]:
"""Search for relevant memories."""
user_id = get_user_id(config)
def _filter_function(doc: Document) -> bool:
return doc.metadata.get("user_id") == user_id
documents = recall_vector_store.similarity_search(
query, k=3, filter=_filter_function
)
return [document.page_content for document in documents]
search = TavilySearchResults(max_results=1)
tools = [save_recall_memory, search_recall_memories, search]
model = ChatOpenAI(model_name="gpt-4o")
model_with_tools = model.bind_tools(tools)
tokenizer = tiktoken.encoding_for_model("gpt-4o")
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You are a helpful assistant with advanced long-term memory"
" capabilities. Powered by a stateless LLM, you must rely on"
" external memory to store information between conversations."
" Utilize the available memory tools to store and retrieve"
" important details that will help you better attend to the user's"
" needs and understand their context.\n\n"
"Memory Usage Guidelines:\n"
"1. Actively use memory tools (save_core_memory, save_recall_memory)"
" to build a comprehensive understanding of the user.\n"
"2. Make informed suppositions and extrapolations based on stored"
" memories.\n"
"3. Regularly reflect on past interactions to identify patterns and"
" preferences.\n"
"4. Update your mental model of the user with each new piece of"
" information.\n"
"5. Cross-reference new information with existing memories for"
" consistency.\n"
"6. Prioritize storing emotional context and personal values"
" alongside facts.\n"
"7. Use memory to anticipate needs and tailor responses to the"
" user's style.\n"
"8. Recognize and acknowledge changes in the user's situation or"
" perspectives over time.\n"
"9. Leverage memories to provide personalized examples and"
" analogies.\n"
"10. Recall past challenges or successes to inform current"
" problem-solving.\n\n"
"## Recall Memories\n"
"Recall memories are contextually retrieved based on the current"
" conversation:\n{recall_memories}\n\n"
"## Instructions\n"
"Engage with the user naturally, as a trusted colleague or friend."
" There's no need to explicitly mention your memory capabilities."
" Instead, seamlessly incorporate your understanding of the user"
" into your responses. Be attentive to subtle cues and underlying"
" emotions. Adapt your communication style to match the user's"
" preferences and current emotional state. Use tools to persist"
" information you want to retain in the next conversation. If you"
" do call tools, all text preceding the tool call is an internal"
" message. Respond AFTER calling the tool, once you have"
" confirmation that the tool completed successfully.\n\n",
),
("placeholder", "{messages}"),
]
)
def agent(state: State) -> State:
"""Process the current state and generate a response using the LLM.
Args:
state (schemas.State): The current state of the conversation.
Returns:
schemas.State: The updated state with the agent's response.
"""
bound = prompt | model_with_tools
recall_str = (
"<recall_memory>\n" + "\n".join(state["recall_memories"]) + "\n</recall_memory>"
)
prediction = bound.invoke(
{
"messages": state["messages"],
"recall_memories": recall_str,
}
)
return {
"messages": [prediction],
}
def load_memories(state: State, config: RunnableConfig) -> State:
"""Load memories for the current conversation.
Args:
state (schemas.State): The current state of the conversation.
config (RunnableConfig): The runtime configuration for the agent.
Returns:
State: The updated state with loaded memories.
"""
convo_str = get_buffer_string(state["messages"])
convo_str = tokenizer.decode(tokenizer.encode(convo_str)[:2048])
recall_memories = search_recall_memories.invoke(convo_str, config)
return {
"recall_memories": recall_memories,
}
class State(MessagesState):
# add memories that will be retrieved based on the conversation context
recall_memories: List[str]
def route_tools(state: State):
"""Determine whether to use tools or end the conversation based on the last message.
Args:
state (schemas.State): The current state of the conversation.
Returns:
Literal["tools", "__end__"]: The next step in the graph.
"""
msg = state["messages"][-1]
if msg.tool_calls:
return "tools"
return END
# Create the graph and add nodes
builder = StateGraph(State)
builder.add_node(load_memories)
builder.add_node(agent)
builder.add_node("tools", ToolNode(tools))
# Add edges to the graph
builder.add_edge(START, "load_memories")
builder.add_edge("load_memories", "agent")
builder.add_conditional_edges("agent", route_tools, ["tools", END])
builder.add_edge("tools", "agent")
# Compile the graph
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)- agentを実行
config = {"configurable": {"user_id": "1", "thread_id": "1"}}
user : my name is John
---
save_recall_memory: User's name is John.
ai : Nice to meet you, John! How can I assist you today?
---
user : i love pizza
---
load_memories : {'recall_memories': ["User's name is John."]}
save_recall_memory: John loves pizza.
ai : Pizza is amazing! Do you have a favorite type or topping?
---
user : yes -- pepperoni!
---
save_recall_memory: John's favorite pizza topping is pepperoni.
ai : Pepperoni is a classic choice! Do you have a favorite pizza place, or do you enjoy making it at home?
---
user : i also just moved to new york
save_recall_memory: John just moved to New York.
ai: Welcome to New York! That's a fantastic place for a pizza lover. Have you had a chance to explore any of the famous pizzerias there yet?- agentを実行(KnowledgeTriple版)
user : Hi, I'm Alice.
ai : Alice! How can I assist you today?
---
user : My friend John likes Pizza.
save_recall_memory: [
{"subject": "Alice", "predicate": "has a friend", "object_": "John"},
{"subject": "John", "predicate": "likes", "object_": "Pizza"}
]
ai : Got it! If you need any suggestions related to pizza or anything else, feel free to ask. What else is on your mind today?
---
user : What food should I bring to John's party?
---
load_memories : {'recall_memories': ['John likes Pizza', 'Alice has a friend John']}
ai : Since John likes pizza, bringing some delicious pizza would be a great choice for the party.
You might also consider asking if there are any specific toppings he prefers or if there are any dietary restrictions among the guests.
This way, you can ensure everyone enjoys the food!参考文献
- [LangGraph] CheckpointとStoreの使い方
- How to add cross-thread persistence to your graph
- Persistence
- Langchain Entity Memory Explained | Restackio
- langchain v0.2系
ConversationEntityMemory
- langchain v0.2系
- 🦜️🔗 LangChain
- GraphDBのクエリを生成させてQA