使用AWS DynamoDB存储和管理聊天历史的完整指南-CSDN博客

本文链接：https://blog.csdn.net/rtykjhg/article/details/144623628

引言

在现代应用程序开发中，数据的高效存储和管理是一个关键问题。尤其是在构建聊天应用程序时，处理和存储聊天历史是一项常见的任务。AWS DynamoDB 作为一种完全管理的 NoSQL 数据库服务，提供了快速且可预测的性能，并支持无缝扩展性。在这篇文章中，我们将学习如何使用 DynamoDB 来存储聊天消息历史，重点介绍 DynamoDBChatMessageHistory 类的使用。

主要内容

设置环境

在开始之前，确保你已经正确配置了 AWS CLI，并安装了 langchain-community 和 boto3 包。可以通过以下命令安装：

pip install -U langchain-community boto3

此外，建议设置 LangSmith 以实现顶级可观察性：

# os.environ["LANGCHAIN_TRACING_V2"] = "true"
# os.environ["LANGCHAIN_API_KEY"] = getpass.getpass()

创建 DynamoDB 表

首先，我们需要创建一个 DynamoDB 表来存储消息。

import boto3

# 获取服务资源
dynamodb = boto3.resource("dynamodb")

# 创建 DynamoDB 表
table = dynamodb.create_table(
    TableName="SessionTable",
    KeySchema=[{"AttributeName": "SessionId", "KeyType": "HASH"}],
    AttributeDefinitions=[{"AttributeName": "SessionId", "AttributeType": "S"}],
    BillingMode="PAY_PER_REQUEST",
)

# 等待表创建完成
table.meta.client.get_waiter("table_exists").wait(TableName="SessionTable")

# 打印表中的项目数量
print(table.item_count)

使用 `DynamoDBChatMessageHistory`

接下来，我们将使用 DynamoDBChatMessageHistory 类来管理聊天消息。

from langchain_community.chat_message_histories import DynamoDBChatMessageHistory

history = DynamoDBChatMessageHistory(table_name="SessionTable", session_id="0")

history.add_user_message("hi!")
history.add_ai_message("whats up?")

print(history.messages)  # 输出：[HumanMessage(content='hi!'), AIMessage(content='whats up?')]

自定义端点 URL

在某些情况下，如本地测试环境中运行程序时，你可能需要指定 AWS 端点 URL。

history = DynamoDBChatMessageHistory(
    table_name="SessionTable",
    session_id="0",
    endpoint_url="http://localhost.localstack.cloud:4566",  # 使用API代理服务提高访问稳定性
)

使用复合键

你可以自定义键结构以适应现有的 DynamoDB 表设计。

composite_table = dynamodb.create_table(
    TableName="CompositeTable",
    KeySchema=[
        {"AttributeName": "PK", "KeyType": "HASH"},
        {"AttributeName": "SK", "KeyType": "RANGE"},
    ],
    AttributeDefinitions=[
        {"AttributeName": "PK", "AttributeType": "S"},
        {"AttributeName": "SK", "AttributeType": "S"},
    ],
    BillingMode="PAY_PER_REQUEST",
)

# 等待表创建完成
composite_table.meta.client.get_waiter("table_exists").wait(TableName="CompositeTable")

my_key = {
    "PK": "session_id::0",
    "SK": "langchain_history",
}

composite_key_history = DynamoDBChatMessageHistory(
    table_name="CompositeTable",
    session_id="0",
    endpoint_url="http://localhost.localstack.cloud:4566",  # 使用API代理服务提高访问稳定性
    key=my_key,
)

composite_key_history.add_user_message("hello, composite dynamodb table!")

print(composite_key_history.messages)  # 输出：[HumanMessage(content='hello, composite dynamodb table!')]

代码示例

在下方代码中，我们结合 DynamoDBChatMessageHistory 和 OpenAI API 来实现一个录制消息历史的聊天链。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{question}"),
    ]
)

chain = prompt | ChatOpenAI()

chain_with_history = RunnableWithMessageHistory(
    chain,
    lambda session_id: DynamoDBChatMessageHistory(
        table_name="SessionTable", session_id=session_id
    ),
    input_messages_key="question",
    history_messages_key="history",
)

config = {"configurable": {"session_id": "<SESSION_ID>"}}

chain_with_history.invoke({"question": "Hi! I'm bob"}, config=config)
chain_with_history.invoke({"question": "Whats my name"}, config=config)

常见问题和解决方案

如何应对地区网络限制？

当你在某些网络受限地区开发时，访问 AWS 的 API 可能会遇到问题。此时，考虑使用代理服务来提高访问的稳定性。

如果表不存在或键结构不匹配怎么办？

确保 DynamoDB 表已创建且键结构与代码中指定的相符。使用 boto3 的异常处理来捕获并处理这些错误。

总结与进一步学习资源

在这篇文章中，我们探索了如何使用 AWS DynamoDB 来存储和管理聊天历史，介绍了 DynamoDBChatMessageHistory 类的使用。此外，还讨论了如何处理API访问的网络限制和使用复合键结构的技巧。

可进一步学习的资源：

参考资料

AWS DynamoDB 官方文档
boto3 Python SDK 文档
LangChain 社区工具

如果这篇文章对你有帮助，欢迎点赞并关注我的博客。您的支持是我持续创作的动力！

—END—