# 将代理式人工智能与 DynamoDB 结合使用 Amazon DynamoDB 是一种无服务器、完全托管的分布式 NoSQL 数据库，在任何规模下都能提供个位数毫秒级性能。DynamoDB 已针对高吞吐量工作负载进行了优化，您可以通过与生成式人工智能模型集成来扩展其功能。利用生成式人工智能模型，您可以实时处理存储在 DynamoDB 表中的数据，并构建具有上下文感知功能且高度个性化的应用程序。您还可以充分利用业务、用户和应用程序数据来定制生成式人工智能解决方案，从而增强终端用户体验。 有关生成式人工智能以及 AWS 提供的生成式人工智能应用程序构建解决方案的更多信息，请参阅[使用生成式人工智能实现业务转型](https://aws.amazon.com/ai/generative-ai/)。 **Topics** + [DynamoDB 的生成式人工智能应用场景](#gen-ai-use-case-ddb) + [DynamoDB 的生成式人工智能博客](#gen-ai-blogs) + [将 DynamoDB 零 ETL 集成与 OpenSearch Service 结合使用](ddb-and-amazon-bedrock.md) + [将 DynamoDB 用作 LangGraph 代理的检查点存储](ddb-langgraph-checkpoint.md) ## DynamoDB 的生成式人工智能应用场景 DynamoDB 广泛用于人工智能驱动的对话应用程序中，例如聊天机器人和使用[基础模型（FM）](https://aws.amazon.com/what-is/foundation-models/)构建的呼叫中心。可以通过 Amazon Bedrock、Amazon SageMaker AI 或其它模型提供商访问 FM。此类应用程序通常使用 DynamoDB 来改进个性化并增强三种数据模式的用户体验：应用程序数据、业务数据和用户数据。以下是这些数据模式的一些示例： + 通过与 [LangChain](https://js.langchain.com/v0.1/docs/integrations/chat_memory/dynamodb/)、[LlamaIndex](https://docs.llamaindex.ai/en/stable/examples/docstore/DynamoDBDocstoreDemo/) 或自定义代码集成来存储应用程序数据，例如聊天消息历史记录。此上下文可让模型与用户来回*对话*，从而增强用户体验。 + 利用库存、定价和文档等业务数据创建定制用户体验。 + 使用用户数据（例如，Web 历史记录、过去的订单和用户偏好）来提供个性化的答案。 例如，保险公司可以使用 DynamoDB 构建聊天机器人，让基于[检索增强生成（RAG）](https://docs.aws.amazon.com/sagemaker/latest/dg/jumpstart-foundation-models-customize-rag.html)的生成式人工智能模型能够访问近实时的数据。此类数据的示例包括实时抵押贷款利率、产品定价、合规/标准合同副本、用户 Web 历史记录和用户偏好。通过将 DynamoDB 与 RAG 结合使用，可以获得有关保险产品和用户数据的深入的更新信息。这将丰富提示和答案，为终端用户提供准确、个性化且近实时的体验。 同样，金融服务行业的客户可使用 DynamoDB、[Amazon Bedrock 知识库](https://docs.aws.amazon.com/bedrock/latest/userguide/knowledge-base.html)和 [Amazon Bedrock 代理](https://aws.amazon.com/bedrock/agents/)来构建基于 RAG 的生成式人工智能应用程序。这些应用程序可以使用开源财报和通话记录。他们还可以使用用户特定的投资组合和交易历史记录来生成投资组合的按需摘要，包括对未来的展望。 ## DynamoDB 的生成式人工智能博客 以下文章提供了详细的用例、最佳实践和分步指南，有助于您利用 DynamoDB 的功能来构建人工智能驱动的高级应用程序。 + [Amazon DynamoDB data models for generative AI chatbots ](https://aws.amazon.com/blogs/database/amazon-dynamodb-data-models-for-generative-ai-chatbots/) + [Build a scalable, context-aware chatbot with Amazon DynamoDB, Amazon Bedrock, and LangChain](https://aws.amazon.com/blogs/database/build-a-scalable-context-aware-chatbot-with-amazon-dynamodb-amazon-bedrock-and-langchain/) + [Build durable AI agents with LangGraph and Amazon DynamoDB](https://aws.amazon.com/blogs/database/build-durable-ai-agents-with-langgraph-and-amazon-dynamodb/) # 将 DynamoDB 零 ETL 集成与 OpenSearch Service 结合使用 您可以将 Amazon Bedrock 与 DynamoDB 结合使用，以提供对[基础模型（FM）](https://aws.amazon.com/what-is/foundation-models/)（例如 Amazon Titan 和其他第三方模型）的无服务器访问。在构建生成式人工智能应用程序时，您可以利用与 Amazon OpenSearch Service 的零 ETL 集成来启用向量搜索功能。[采用 DynamoDB 与 OpenSearch 的零 ETL 集成和 Amazon Bedrock 的生成式人工智能](https://catalog.workshops.aws/dynamodb-labs/en-US/dynamodb-opensearch-zetl)讲习会为您提供设置 DynamoDB 与 OpenSearch 的零 ETL 集成的实践经验。此讲习会将执行以下任务： + 创建从 DynamoDB 表到 OpenSearch 的管道。 + 在 OpenSearch 中创建 Amazon Bedrock Connector。 + 使用 OpenSearch 作为向量存储来查询 Amazon Bedrock。 + 使用 Amazon Bedrock 中的 Claude FM 用简单英文创建书面回复，从而解释 OpenSearch 返回的搜索结果。 此讲习会使您能够将 DynamoDB 与 OpenSearch 集成，以构建生成式人工智能应用程序。它还演示了跨数据库引擎的灵活查询功能，协助您针对传统应用场景集成 DynamoDB 和 OpenSearch。此讲习会是 [Amazon DynamoDB Immersion Day](https://catalog.workshops.aws/dynamodb-labs/en-US) 的七个模块之一。您可以在任何 AWS 账户中运行此讲习会。 您也可以参考以下博客文章，了解如何在 DynamoDB 和 OpenSearch Service 之间设置零 ETL 集成。这篇博客文章还介绍了如何在 OpenSearch Service 中设置模型连接器，以使用 Amazon Bedrock 自动为传入数据生成嵌入。[Amazon DynamoDB 与 Amazon OpenSearch Service 的零 ETL 集成的向量搜索](https://aws.amazon.com/blogs/database/vector-search-for-amazon-dynamodb-with-zero-etl-for-amazon-opensearch-service/) # 将 DynamoDB 用作 LangGraph 代理的检查点存储 [LangGraph](https://langchain-ai.github.io/langgraph/) 是一个使用大语言模型（LLM）构建有状态、多参与者人工智能应用程序的框架。LangGraph 代理需要永久存储来维护对话状态、实现人机协同工作流程、支持容错能力以及提供时空旅行调试功能。DynamoDB 的无服务器架构、个位数毫秒延迟和自动扩展等特性使其成为在 AWS 上实现生产 LangGraph 部署的理想检查点存储。 `langgraph-checkpoint-aws` 程序包提供了一个用于实现 LangGraph 检查点接口的 `DynamoDBSaver` 类，使您能够对大型检查点使用可选的 Amazon Simple Storage Service 卸载来在 DynamoDB 中保持代理状态。 ## 主要 功能 状态持久性 在每个步骤之后自动保存代理状态，使代理能够从中断中恢复并从故障中恢复。 基于生存时间的清理 使用 DynamoDB 生存时间来管理存储成本，自动使旧检查点过期。 压缩 （可选）使用 gzip 压缩检查点数据，以降低存储成本并提高吞吐量。 Amazon S3 卸载 自动将大型检查点（大于 350 KB）卸载到 Amazon Simple Storage Service，以便在 DynamoDB 项目大小限制内工作。 同步和异步支持 同步和异步 API，用于在不同的应用程序架构中实现灵活性。 ## 先决条件 + Python 3.10 或更高版本 + 一个 AWS 账户，具有创建 DynamoDB 表（以及可选的 Amazon S3 存储桶）的权限 + 配置了 AWS 凭证（有关凭证设置选项，请参阅 AWS 文档） **重要** 本指南创建的 AWS 资源可能会产生费用。默认情况下，DynamoDB 使用按请求支付计费，如果您启用大型检查点卸载功能，则会收取 Amazon S3 费用。完成后，按照[清理](#langgraph-cleanup)部分以删除资源。 ## 安装 从 PyPI 安装检查点程序包： ``` pip install langgraph-checkpoint-aws ``` ## 基本用法 以下示例演示如何将 DynamoDB 配置为 LangGraph 代理的检查点存储： ``` from langgraph.graph import StateGraph from langgraph_checkpoint_aws import DynamoDBSaver from typing import TypedDict # Define your state schema class State(TypedDict): input: str result: str # Initialize the DynamoDB checkpoint saver checkpointer = DynamoDBSaver( table_name="langgraph-checkpoints", region_name="us-east-1" ) # Build your LangGraph workflow builder = StateGraph(State) builder.add_node("process", lambda state: {"result": "processed"}) builder.set_entry_point("process") builder.set_finish_point("process") # Compile the graph with the DynamoDB checkpointer graph = builder.compile(checkpointer=checkpointer) # Invoke the graph with a thread ID to enable state persistence config = {"configurable": {"thread_id": "session-123"}} result = graph.invoke({"input": "data"}, config) ``` 配置中的 `thread_id` 充当 DynamoDB 中的分区键，支持您维护单独的对话线程并检索任何线程的历史状态。 ## 生产配置 对于生产部署，您可以启用生存时间、压缩和 Amazon S3 卸载。也可以使用 `endpoint_url` 参数来指向本地 DynamoDB 实例以进行测试： ``` import boto3 from botocore.config import Config from langgraph_checkpoint_aws import DynamoDBSaver # Production configuration session = boto3.Session( profile_name="production", region_name="us-east-1" ) checkpointer = DynamoDBSaver( table_name="langgraph-checkpoints", session=session, ttl_seconds=86400 * 7, # Expire checkpoints after 7 days enable_checkpoint_compression=True, # Enable gzip compression boto_config=Config( retries={"mode": "adaptive", "max_attempts": 6}, max_pool_connections=50 ), s3_offload_config={ "bucket_name": "my-checkpoint-bucket" } ) # Local testing with DynamoDB Local local_checkpointer = DynamoDBSaver( table_name="langgraph-checkpoints", region_name="us-east-1", endpoint_url="http://localhost:8000" ) ``` ## DynamoDB 表配置 检查点保护程序需要一个具有复合主键的 DynamoDB 表。使用以下 AWS CloudFormation 模板创建表： ``` AWSTemplateFormatVersion: '2010-09-09' Description: 'DynamoDB table for LangGraph checkpoint storage' Parameters: TableName: Type: String Default: langgraph-checkpoints Resources: CheckpointTable: Type: AWS::DynamoDB::Table DeletionPolicy: Retain UpdateReplacePolicy: Retain Properties: TableName: !Ref TableName BillingMode: PAY_PER_REQUEST AttributeDefinitions: - AttributeName: PK AttributeType: S - AttributeName: SK AttributeType: S KeySchema: - AttributeName: PK KeyType: HASH - AttributeName: SK KeyType: RANGE TimeToLiveSpecification: AttributeName: ttl Enabled: true PointInTimeRecoverySpecification: PointInTimeRecoveryEnabled: true SSESpecification: SSEEnabled: true ``` 使用 AWS CLI 部署模板： ``` aws cloudformation deploy \ --template-file template.yaml \ --stack-name langgraph-checkpoint \ --parameter-overrides TableName=langgraph-checkpoints ``` ## 所需的 IAM 权限 以下 IAM 策略提供了 DynamoDB 检查点保护程序所需的最小权限。将 *111122223333* 替换为您的 AWS 账户 ID，并更新区域以匹配您的环境。 ``` { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:Query", "dynamodb:BatchGetItem", "dynamodb:BatchWriteItem" ], "Resource": "arn:aws:dynamodb:us-east-1:111122223333:table/langgraph-checkpoints" } ] } ``` 如果启用 Amazon S3 卸载，请在策略中添加以下语句： ``` { "Effect": "Allow", "Action": [ "s3:PutObject", "s3:GetObject", "s3:DeleteObject", "s3:PutObjectTagging" ], "Resource": "arn:aws:s3:::my-checkpoint-bucket/*" }, { "Effect": "Allow", "Action": [ "s3:GetBucketLifecycleConfiguration", "s3:PutBucketLifecycleConfiguration" ], "Resource": "arn:aws:s3:::my-checkpoint-bucket" } ``` ## 异步使用 对于异步应用程序，请使用检查点保护程序提供的异步方法： ``` import asyncio from langgraph.graph import StateGraph from langgraph_checkpoint_aws import DynamoDBSaver from typing import TypedDict class State(TypedDict): input: str result: str async def main(): checkpointer = DynamoDBSaver( table_name="langgraph-checkpoints", region_name="us-east-1" ) builder = StateGraph(State) builder.add_node("process", lambda state: {"result": "processed"}) builder.set_entry_point("process") builder.set_finish_point("process") graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "async-session-123"}} result = await graph.ainvoke({"input": "data"}, config) return result asyncio.run(main()) ``` ## 清理 为避免持续产生费用，请删除您创建的资源： ``` # Delete the DynamoDB table aws dynamodb delete-table --table-name langgraph-checkpoints # Delete the CloudFormation stack (if you used the template above) aws cloudformation delete-stack --stack-name langgraph-checkpoint # If you created an S3 bucket for large checkpoint offloading, empty and delete it aws s3 rm s3://my-checkpoint-bucket --recursive aws s3 rb s3://my-checkpoint-bucket ``` ## 错误处理 常见错误情景： + **找不到表**：验证 `table_name` 和 `region_name` 与您的 DynamoDB 表匹配。 + **节流**：如果您看到 `ProvisionedThroughputExceededException`，请考虑切换到按需计费模式或增加预置容量。 + **超出项目大小**：如果检查点超过 350 KB，请启用 Amazon S3 卸载（请参阅[生产配置](#langgraph-production-config)）。 + **凭证错误**：验证您的 AWS 凭证有效且具有[所需的权限](#langgraph-iam-permissions)。 ## 其他资源 + [langgraph-checkpoint-aws on PyPI](https://pypi.org/project/langgraph-checkpoint-aws/) + [langgraph-checkpoint-aws on GitHub](https://github.com/langchain-ai/langchain-aws/blob/main/libs/langgraph-checkpoint-aws/docs/dynamodb/DynamoDBSaver.md) + [LangGraph documentation](https://langchain-ai.github.io/langgraph/) + [DynamoDB 最佳实践](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/best-practices.html) + [Build durable AI agents with LangGraph and Amazon DynamoDB](https://aws.amazon.com/blogs/database/build-durable-ai-agents-with-langgraph-and-amazon-dynamodb/)