释放双内存 Ai 代理的力量：利用长短期记忆构建更智能的客户支持聊天机器人！
大家好，欢迎回到我们关于Agentic AI系统系列的另一篇精彩文章
在本文中，我们将深入探讨长短期记忆的工作，建立在我们之前文章中涵盖的理论理解基础上。现在是时候将这些知识应用于实践，构建一个简单的客户支持代理，利用长短期记忆的能力。
让我们开始吧…
代码示例
class MemoryAgent:
def __init__(self):
self.short_term_memory = {}
self.long_term_memory = {}
def add_short_term_memory(self, key, value):
self.short_term_memory[key] = value
def add_long_term_memory(self, key, value):
self.long_term_memory[key] = value
def retrieve_memory(self, key):
if key in self.short_term_memory:
return self.short_term_memory[key]
elif key in self.long_term_memory:
return self.long_term_memory[key]
else:
return None
结论
在本文中，我们探讨了AI系统中记忆的概念，并提供了一个内存代理的简单实现。通过利用长短期记忆，我们可以增强我们的AI系统在各种应用中的能力，包括客户支持。
聊天机器人代理与长短期记忆
仅回顾一下我们在关于AI代理中的记忆的上一篇文章中讨论的理论内容：
-
对话内记忆: 聊天机器人保持对当前对话上下文的意识，如果有短暂的中断，可以接着之前的内容继续。可以想象成记住你在同一对话中早些时候讨论的内容。为此，我们将使用CheckPointers。这就是我们在上一篇文章中提到的会话范围内的记忆或线程记忆。
-
持久用户记忆: 聊天机器人随着时间的推移建立并维护对每个用户的知识，将这些信息带入所有未来的对话中。这类似于一个人记住与朋友的先前互动，即使这些对话发生在几天或几周之前。为此实现，我们将使用LangGraph Store。
有关LangGraph中记忆管理的更多信息，请阅读这里
导入
我们将通过导入程序中所需的库和包来开始。随着我们的进展，我们会导入更多的内容，但现在让我们引入导入以加载环境变量：
import os
from dotenv import load_dotenv
import os
加载环境变量
要加载我们的环境变量，我们将使用下面的代码。在这种情况下，我们的环境变量将是LLM模型提供者的API密钥和LangSmith凭证——如果您愿意，可以跳过LangSmith凭证。
要加载环境变量，我们使用下面的命令：
os.environ["LANGCHAIN_PROJECT"] = "langgraph_store_customer_support_agent"
os.environ["LANGCHAIN_TRACING_V2"] = "true"
Playground
在我们真正深入探讨AI代理中的记忆管理之前，让我们先快速建立基础知识。我们将简要了解在游乐场场景中如何使用长短期记忆。我的目的是通过一个简单的例子，让您在LangGraph中获得内存存储的实际操作经验。
内存记忆
使用LangGraph中的消息或记忆存储，我们可以将数据存储在更永久的存储位置，例如SQLite数据库或甚至Postgres数据库。现在，我们将保持非常简单，将数据存储在内存中，因此称为“内存记忆”。让我们开始吧…
Please translate the following text to Chinese. Requirements: 1. Keep code blocks (```) unchanged 2. Keep inline code (`) unchanged 3. Keep Markdown formatting 4. Use these technical term translations: Agentic AI systems -> Agentic AI系统 long and short-term memory -> 长短期记忆 customer support agent -> 客户支持代理 CheckPointers -> 检查指针 LangGraph Store -> LangGraph存储 API keys -> API密钥 LLM model provider -> LLM模型提供者 LangSmith credentials -> LangSmith凭证 InMemoryStore -> 内存存储 uuid -> uuid Anthropic -> Anthropic claude-3-opus-20240229 -> claude-3-opus-20240229 ChatAnthropic -> ChatAnthropic MODEL_SYSTEM_MESSAGE -> 模型系统消息 CREATE_HISTORY_INSTRUCTION -> 创建历史指令 MemorySaver -> 内存保存器 StateGraph -> 状态图 MessagesState -> 消息状态 START -> 开始 END -> 结束 BaseStore -> 基础存储 HumanMessage -> 人类消息 SystemMessage -> 系统消息 RunnableConfig -> 可运行配置 stream_mode -> 流模式 values -> 值 updates -> 更新
Text: ## Basic Imports
import uuid
from langgraph.store.memory import InMemoryStore
创建内存存储
in_memory_store = InMemoryStore()
存储记忆
要使用我们刚创建的内存对象存储一些数据或信息，我们将以将客户联系信息（如姓名和电子邮件）存储在长期记忆存储中为例。
customer_id = "1"
namespace_for_memory = (customer_id, "customer_interactions")
key = str(uuid.uuid4())
value = {"name" : "John Doe", "email": "[email protected]"}
in_memory_store.put(namespace_for_memory, key, value)
我们还可以使用与我们希望删除的内存对应的键从内存中删除数据。以下是我们可以做到的方法：
in_memory_store.delete(namespace=namespace_for_memory, key=key)
从上面可以看出，我们只需将 namespace 和 key 传递给删除方法，它将处理内存的删除。
搜索记忆
我们现在可以继续搜索内存存储以检索信息。为此，我们可以使用下面的代码：
memories = in_memory_store.search(namespace_for_memory)
type(memories)
我们可以使用 dict 方法获取记忆数据：
print(type(memories[0]))
print(memories[0].key, memories[0].value)
就像在常规的Python字典中存储键值对一样，我们可以获取每个存储的记忆的键和值。键是我们在存储数据时传入的唯一标识符，而值就是数据本身。
使用键和值获取记忆
通过使用 get 方法按 命名空间 和 键 获取记忆。
您可以看到我们还存储每个记忆对象的创建和更新日期。此信息可用于获取最近的长短期记忆，具体取决于您希望通过长短期记忆存储构建或实现的目标。
创建具有长短期记忆的客户支持代理
现在我们对如何使用记忆有了基本的了解，我们可以继续构建我们的客户支持代理。让我们深入探讨…
创建模型
在这种情况下，我将使用Anthropic的claude-3-opus-20240229模型。确保您拥有Anthropic的API密钥。让我们加载我们的模型：
from langchain_anthropic import ChatAnthropic
model = ChatAnthropic(
temperature=0.0,
model="claude-3-opus-20240229",
api_key=os.environ.get("ANTHROPIC_API_KEY")
)
代理提示
我们需要一个提示，用于利用与客户互动获得的历史数据，为相关客户提供更个性化的支持，我们将把这个提示存储在一个名为 MODEL_SYSTEM_MESSAGE 的变量中。
第二个提示将用于更新上述第一个提示中使用的历史记录。历史记录将不断生成，以收集更多关于客户的数据。我们将收集有关客户的信息，例如：
- 联系历史
- 产品使用/购买
- 以前的问题/解决方案
- 偏好（沟通，产品）
- 特殊情况
让我们继续实现这两个提示。
COMPANY_NAME = "Code With Prince PLC"
MODEL_SYSTEM_MESSAGE = """You are a helpful customer support assistant for {company_name}.
Use the customer's history to provide relevant and personalized support.
Customer profile: {history}"""
CREATE_HISTORY_INSTRUCTION = """Update the customer profile with new support interaction details.
CURRENT PROFILE:
{history}
ANALYZE FOR:
1. Contact history
2. Product usage/purchases
3. Previous issues/resolutions
4. Preferences (communication, products)
5. Special circumstances
Focus on verified support interactions only. Summarize key details clearly.
Update profile based on this conversation:"""
节点
我们的图将有两个主要节点：一个用于调用大型语言模型，另一个用于更新记忆（我们的客户数据）。让我们继续实现这些节点功能，从引入一些导入开始。
from IPython.display import Image, display
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.store.base import BaseStore
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.runnables.config import RunnableConfig
def call_model(state: MessagesState, config: RunnableConfig, store: BaseStore):
"""使用客户上下文和历史生成AI响应。
Args:
state: 当前对话消息
config: 包含customer_id的运行时配置
store: 客户数据的持久存储
Returns:
dict: 生成的响应消息
Flow:
1. 使用ID从存储中获取客户资料
2. 使用客户上下文格式化系统提示
3. 生成个性化响应
"""
customer_id = config["configurable"]["customer_id"]
namespace = ("customer_interactions", customer_id)
key = "customer_data_memory"
memory = store.get(namespace, key)
history = memory.value.get('customer_data_memory') if memory else "未找到现有记忆。"
system_msg = MODEL_SYSTEM_MESSAGE.format(history=history, company_name=COMPANY_NAME)
response = model.invoke([SystemMessage(content=system_msg)] + state["messages"])
return {"messages": response}
现在让我们继续实现我们的最终节点功能。这将用于更新我们的记忆历史——我们在与用户互动时收集的数据。
def write_memory(state: MessagesState, config: RunnableConfig, store: BaseStore):
"""在持久存储中更新客户互动历史。
Args:
state: 当前对话消息
config: 包含customer_id的运行时配置
store: 客户数据的持久存储
Flow:
1. 检索现有客户历史
2. 分析对话以获取新见解
3. 用新数据更新存储的历史
"""
user_id = config["configurable"]["customer_id"]
namespace = ("customer_interactions", user_id)
key = "customer_data_memory"
memory = store.get(namespace=namespace, key=key)
history = memory.value.get(key) if memory else "未找到现有历史。"
system_msg = CREATE_HISTORY_INSTRUCTION.format(history=history)
new_insights = model.invoke([SystemMessage(content=system_msg)] + state['messages'])
store.put(namespace, key, {"customer_interactions": new_insights.content})
构建图
现在我们已经创建了必要的节点，我们可以继续构建实际的代理图，遵循我们在之前文章中使用的相同方法。
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_node("write_memory", write_memory)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", "write_memory")
builder.add_edge("write_memory", END)
across_thread_memory = InMemoryStore()
within_thread_memory = MemorySaver()
graph = builder.compile(
checkpointer=within_thread_memory,
store=across_thread_memory
)
从文章的标题来看，你可能会想知道短期记忆在哪里以及如何使用。如果你仔细阅读代码注释，你会注意到我们将短期记忆存储在MemorySaver对象实例中：
within_thread_memory = MemorySaver()
我们可以在另一篇文章中更深入地探讨短期记忆，在那篇文章中，我们将创建一个类似于我们为长期记忆所做的简单游乐场。
可视化图形
现在我们已经构建了图形，让我们可视化它，以更好地理解流程并查看我们创建了什么。
from IPython.display import display, Image
display(Image(graph.get_graph(xray=1).draw_mermaid_png()))
流式聊天机器人的完整状态
在LangGraph中，像在LangChain中一样，我们有两种主要方式来实现流式传输：
.stream: 用于同步流式传输内容.astream: 用于异步流式传输内容
对于图状态流式传输，LangGraph支持以下方法：
values: 在每个节点被调用后流式传输图的完整状态.updates: 流式传输对图状态所做的更新
我们可以使用stream_mode属性来配置这些。
让我们看看如何实现这一点：
config = {
"configurable": {
"thread_id": "1",
"customer_id": "1"
}
}
input_msg = [HumanMessage(content="Hi, my name is John Doe. I would love to know more about your PCs")]
for chunk in graph.stream(
{"messages": input_msg},
config,
stream_mode="values"
):
chunk["messages"][-1].pretty_print()
让我们添加更多消息以刺激真实世界的对话。
input_messages = [HumanMessage(content="I last purchased the Latest generation Intel processor based PC and it worked really well. What are your prices on this models?")]
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
我们可以使用以下方法访问机器人的长短期记忆：
customer_id = "1"
namespace = ("customer_interactions", customer_id)
existing_memory = across_thread_memory.get(namespace=namespace, key="customer_data_memory")
existing_memory.dict()
我们还可以使用Python中的for循环打印这些消息：
for k, v in existing_memory.dict().get("value").items():
print(k, v)
对于短期记忆，我们可以使用以下方法访问：
短期记忆仅仅是图的当前状态。我们可以使用下面的代码来检查这一点：
thread = {"configurable": {"thread_id": "1"}}
state = graph.get_state(thread).values
for m in state["messages"]:
m.pretty_print()
现在您可以在了解其工作原理后，更多地尝试这个设置。尝试测试机器人回忆长短期信息的能力，并评估其在对话中保持上下文的表现。
结论
恭喜你走到这一步！在本文中，我们迈出了构建一个利用长短期记忆能力的客户支持代理的实际一步。
祝你编码愉快！
