長期記憶代理

本教學示範如何使用 LangGraph 實作具有長期記憶功能的代理。代理可以儲存、檢索和使用記憶來增強與使用者的互動。

受到 MemGPT 等論文的啟發，並從我們自己關於長期記憶的工作中提煉出來，此圖形從聊天互動中提取記憶並將其持久儲存到資料庫。「記憶」在本教學中將以兩種方式表示

代理產生的文字資訊
代理以 (主詞、謂詞、受詞) 知識三元組的形式提取的關於實體的結構化資訊。

此資訊稍後可以語意方式讀取或查詢，以便在您的機器人回應用戶時提供個人化情境。

關鍵概念是透過儲存記憶，代理會持久儲存關於使用者的資訊，這些資訊在多個對話（執行緒）之間共享，這與 LangGraph 的持久性已啟用的單次對話記憶不同。

您也可以在此儲存庫中查看此代理的完整實作。

安裝相依性

%pip install -U --quiet langgraph langchain-openai langchain-community tiktoken

import getpass
import os


def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("OPENAI_API_KEY")
_set_env("TAVILY_API_KEY")

OPENAI_API_KEY:  ········
TAVILY_API_KEY:  ········

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())

定義工具

接下來，讓我們定義記憶體工具。我們將需要一個工具來儲存記憶體，另一個工具來搜尋它們以找到最相關的記憶體。

import uuid


def get_user_id(config: RunnableConfig) -> str:
    user_id = config["configurable"].get("user_id")
    if user_id is None:
        raise ValueError("User ID needs to be provided to save a memory.")

    return user_id


@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 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]

此外，讓我們讓代理能夠使用 Tavily 搜尋網路。

search = TavilySearchResults(max_results=1)
tools = [save_recall_memory, search_recall_memories, search]

定義狀態、節點和邊緣

我們的圖形狀態將僅包含兩個通道 - messages 用於追蹤聊天記錄，以及 recall_memories - 在調用代理之前提取並傳遞給代理系統提示的情境記憶體。

class State(MessagesState):
    # add memories that will be retrieved based on the conversation context
    recall_memories: List[str]

# Define the prompt template for the agent
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}"),
    ]
)

model = ChatOpenAI(model_name="gpt-4o")
model_with_tools = model.bind_tools(tools)

tokenizer = tiktoken.encoding_for_model("gpt-4o")


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,
    }


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

建立圖形

我們的代理圖形將與簡單的 ReAct 代理非常相似。唯一重要的修改是新增一個節點，以便在第一次調用代理之前載入記憶體。

# 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)

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

執行代理！

讓我們第一次執行代理，並告訴它一些關於使用者的資訊！

def pretty_print_stream_chunk(chunk):
    for node, updates in chunk.items():
        print(f"Update from node: {node}")
        if "messages" in updates:
            updates["messages"][-1].pretty_print()
        else:
            print(updates)

        print("\n")

# NOTE: we're specifying `user_id` to save memories for a given user
config = {"configurable": {"user_id": "1", "thread_id": "1"}}

for chunk in graph.stream({"messages": [("user", "my name is John")]}, config=config):
    pretty_print_stream_chunk(chunk)

Update from node: load_memories
{'recall_memories': []}


Update from node: agent
==================================[1m Ai Message [0m==================================
Tool Calls:
  save_recall_memory (call_OqfbWodmrywjMnB1v3p19QLt)
 Call ID: call_OqfbWodmrywjMnB1v3p19QLt
  Args:
    memory: User's name is John.


Update from node: tools
=================================[1m Tool Message [0m=================================
Name: save_recall_memory

User's name is John.


Update from node: agent
==================================[1m Ai Message [0m==================================

Nice to meet you, John! How can I assist you today?

您可以看到代理已儲存關於使用者名稱的記憶體。讓我們新增更多關於使用者的資訊！

for chunk in graph.stream({"messages": [("user", "i love pizza")]}, config=config):
    pretty_print_stream_chunk(chunk)

Update from node: load_memories
{'recall_memories': ["User's name is John."]}


Update from node: agent
==================================[1m Ai Message [0m==================================
Tool Calls:
  save_recall_memory (call_xxEivMuWCURJrGxMZb02Eh31)
 Call ID: call_xxEivMuWCURJrGxMZb02Eh31
  Args:
    memory: John loves pizza.


Update from node: tools
=================================[1m Tool Message [0m=================================
Name: save_recall_memory

John loves pizza.


Update from node: agent
==================================[1m Ai Message [0m==================================

Pizza is amazing! Do you have a favorite type or topping?

for chunk in graph.stream(
    {"messages": [("user", "yes -- pepperoni!")]},
    config={"configurable": {"user_id": "1", "thread_id": "1"}},
):
    pretty_print_stream_chunk(chunk)

Update from node: load_memories
{'recall_memories': ["User's name is John.", 'John loves pizza.']}


Update from node: agent
==================================[1m Ai Message [0m==================================
Tool Calls:
  save_recall_memory (call_AFrtCVwIEr48Fim80zlhe6xg)
 Call ID: call_AFrtCVwIEr48Fim80zlhe6xg
  Args:
    memory: John's favorite pizza topping is pepperoni.


Update from node: tools
=================================[1m Tool Message [0m=================================
Name: save_recall_memory

John's favorite pizza topping is pepperoni.


Update from node: agent
==================================[1m Ai Message [0m==================================

Pepperoni is a classic choice! Do you have a favorite pizza place, or do you enjoy making it at home?

for chunk in graph.stream(
    {"messages": [("user", "i also just moved to new york")]},
    config={"configurable": {"user_id": "1", "thread_id": "1"}},
):
    pretty_print_stream_chunk(chunk)

Update from node: load_memories
{'recall_memories': ["User's name is John.", 'John loves pizza.', "John's favorite pizza topping is pepperoni."]}


Update from node: agent
==================================[1m Ai Message [0m==================================
Tool Calls:
  save_recall_memory (call_Na86uY9eBzaJ0sS0GM4Z9tSf)
 Call ID: call_Na86uY9eBzaJ0sS0GM4Z9tSf
  Args:
    memory: John just moved to New York.


Update from node: tools
=================================[1m Tool Message [0m=================================
Name: save_recall_memory

John just moved to New York.


Update from node: agent
==================================[1m Ai Message [0m==================================

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?

現在我們可以在不同的執行緒上使用關於使用者的已儲存資訊。讓我們試試看

config = {"configurable": {"user_id": "1", "thread_id": "2"}}

for chunk in graph.stream(
    {"messages": [("user", "where should i go for dinner?")]}, config=config
):
    pretty_print_stream_chunk(chunk)

Update from node: load_memories
{'recall_memories': ['John loves pizza.', "User's name is John.", 'John just moved to New York.']}


Update from node: agent
==================================[1m Ai Message [0m==================================

Considering you just moved to New York and love pizza, I'd recommend checking out some of the iconic pizza places in the city. Some popular spots include:

1. **Di Fara Pizza** in Brooklyn – Known for its classic New York-style pizza.
2. **Joe's Pizza** in Greenwich Village – A historic pizzeria with a great reputation.
3. **Lucali** in Carroll Gardens, Brooklyn – Often ranked among the best for its delicious thin-crust pies.

Would you like more recommendations or information about any of these places?

請注意，代理如何在回答之前載入最相關的記憶體，並且在我們的案例中，根據食物偏好和位置建議晚餐推薦。

最後，讓我們將搜尋工具與對話情境和記憶體的其餘部分一起使用，以尋找披薩店的位置

for chunk in graph.stream(
    {"messages": [("user", "what's the address for joe's in greenwich village?")]},
    config=config,
):
    pretty_print_stream_chunk(chunk)

Update from node: load_memories
{'recall_memories': ['John loves pizza.', 'John just moved to New York.', "John's favorite pizza topping is pepperoni."]}


Update from node: agent
==================================[1m Ai Message [0m==================================
Tool Calls:
  tavily_search_results_json (call_aespiB28jpTFvaC4d0qpfY6t)
 Call ID: call_aespiB28jpTFvaC4d0qpfY6t
  Args:
    query: Joe's Pizza Greenwich Village NYC address


Update from node: tools
=================================[1m Tool Message [0m=================================
Name: tavily_search_results_json

[{"url": "https://www.joespizzanyc.com/locations-1-1", "content": "Joe's Pizza Greenwich Village (Original Location) 7 Carmine Street New York, NY 10014 (212) 366-1182﻿ Joe's Pizza Times Square 1435 Broadway New York, NY 10018 (646) 559-4878. TIMES SQUARE MENU. ORDER JOE'S TIMES SQUARE Joe's Pizza Williamsburg 216 Bedford Avenue Brooklyn, NY 11249"}]


Update from node: agent
==================================[1m Ai Message [0m==================================

The address for Joe's Pizza in Greenwich Village is:

**7 Carmine Street, New York, NY 10014**

Enjoy your pizza!

如果您傳遞不同的使用者 ID，則代理的回應將不會個人化，因為我們尚未儲存任何關於其他使用者的資訊

新增結構化記憶體

到目前為止，我們將記憶體表示為字串，例如 "John loves pizza"。這是將記憶體持久儲存到向量儲存區時的自然表示。如果您的用例可以從其他持久性後端（例如圖形資料庫）中受益，我們可以更新我們的應用程式以產生具有額外結構的記憶體。

在下面，我們更新 save_recall_memory 工具以接受「知識三元組」列表，或包含 主詞、謂詞 和 受詞 的 3 元組，適用於儲存在知識圖譜中。然後，我們的模型將產生這些表示作為其工具調用的一部分。

為了簡單起見，我們使用與之前相同的向量資料庫，但可以進一步更新 save_recall_memory 和 search_recall_memories 工具以與圖形資料庫互動。目前，我們只需要更新 save_recall_memory 工具

recall_vector_store = InMemoryVectorStore(OpenAIEmbeddings())

from typing_extensions import TypedDict


class KnowledgeTriple(TypedDict):
    subject: str
    predicate: str
    object_: str


@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

然後我們可以像以前一樣編譯圖形

tools = [save_recall_memory, search_recall_memories, search]
model_with_tools = model.bind_tools(tools)


# 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)

config = {"configurable": {"user_id": "3", "thread_id": "1"}}

for chunk in graph.stream({"messages": [("user", "Hi, I'm Alice.")]}, config=config):
    pretty_print_stream_chunk(chunk)

Update from node: load_memories
{'recall_memories': []}


Update from node: agent
==================================[1m Ai Message [0m==================================

Hello, Alice! How can I assist you today?

請注意，應用程式選擇從使用者的陳述中提取知識三元組

for chunk in graph.stream(
    {"messages": [("user", "My friend John likes Pizza.")]}, config=config
):
    pretty_print_stream_chunk(chunk)

Update from node: load_memories
{'recall_memories': []}


Update from node: agent
==================================[1m Ai Message [0m==================================
Tool Calls:
  save_recall_memory (call_EQSZlvZLZpPa0OGS5Kyzy2Yz)
 Call ID: call_EQSZlvZLZpPa0OGS5Kyzy2Yz
  Args:
    memories: [{'subject': 'Alice', 'predicate': 'has a friend', 'object_': 'John'}, {'subject': 'John', 'predicate': 'likes', 'object_': 'Pizza'}]


Update from node: tools
=================================[1m Tool Message [0m=================================
Name: save_recall_memory

[{"subject": "Alice", "predicate": "has a friend", "object_": "John"}, {"subject": "John", "predicate": "likes", "object_": "Pizza"}]


Update from node: agent
==================================[1m Ai Message [0m==================================

Got it! If you need any suggestions related to pizza or anything else, feel free to ask. What else is on your mind today?

與之前一樣，從一個執行緒產生的記憶體可以從同一使用者的另一個執行緒中存取

config = {"configurable": {"user_id": "3", "thread_id": "2"}}

for chunk in graph.stream(
    {"messages": [("user", "What food should I bring to John's party?")]}, config=config
):
    pretty_print_stream_chunk(chunk)

Update from node: load_memories
{'recall_memories': ['John likes Pizza', 'Alice has a friend John']}


Update from node: agent
==================================[1m Ai Message [0m==================================

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!

或者，為了說明目的，我們可以視覺化模型提取的知識圖譜

%pip install -U --quiet matplotlib networkx

import matplotlib.pyplot as plt
import networkx as nx

# Fetch records
records = recall_vector_store.similarity_search(
    "Alice", k=2, filter=lambda doc: doc.metadata["user_id"] == "3"
)


# Plot graph
plt.figure(figsize=(6, 4), dpi=80)
G = nx.DiGraph()

for record in records:
    G.add_edge(
        record.metadata["subject"],
        record.metadata["object_"],
        label=record.metadata["predicate"],
    )

pos = nx.spring_layout(G)
nx.draw(
    G,
    pos,
    with_labels=True,
    node_size=3000,
    node_color="lightblue",
    font_size=10,
    font_weight="bold",
    arrows=True,
)
edge_labels = nx.get_edge_attributes(G, "label")
nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels, font_color="red")
plt.show()

安裝相依性​

定義記憶體的向量儲存區​

定義工具​

定義狀態、節點和邊緣​

建立圖形​

執行代理！​

新增結構化記憶體​

此頁面是否有幫助？