Tác giả: Lee Hannigan
Ngày phát hành: 13 JAN 2026
Chuyên mục: Advanced (300), Amazon DynamoDB, Artificial Intelligence, Serverless

Tôi đã bị cuốn hút bởi sự phát triển nhanh chóng của các tác nhân AI. Trong năm qua, tôi đã chứng kiến chúng phát triển từ những chatbot đơn giản thành các hệ thống tinh vi có khả năng suy luận qua các vấn đề phức tạp, đưa ra quyết định và duy trì ngữ cảnh trong các cuộc hội thoại dài. Tuy nhiên, một tác nhân chỉ tốt khi bộ nhớ của nó tốt.

Trong bài viết này, chúng tôi sẽ chỉ cho bạn cách xây dựng các tác nhân AI sẵn sàng cho môi trường sản xuất với khả năng quản lý trạng thái bền vững bằng cách sử dụng Amazon DynamoDB và LangGraph với trình kết nối DynamoDBSaver mới, một thư viện checkpoint của LangGraph được AWS duy trì cho Amazon DynamoDB. Nó cung cấp một lớp lưu trữ bền vững sẵn sàng cho môi trường sản xuất, được xây dựng đặc biệt cho DynamoDB và LangGraph, lưu trữ trạng thái tác nhân với khả năng xử lý thông minh các payload dựa trên kích thước của chúng.

Bạn sẽ tìm hiểu cách triển khai này có thể cung cấp cho các tác nhân của bạn khả năng lưu trữ bền vững cần thiết để mở rộng quy mô, phục hồi sau lỗi và duy trì các quy trình làm việc dài hạn.

Tổng quan nhanh về Amazon DynamoDB

Amazon DynamoDB là một cơ sở dữ liệu NoSQL phân tán, không máy chủ, được quản lý hoàn toàn với hiệu suất mili giây một chữ số ở mọi quy mô. Bạn có thể lưu trữ dữ liệu có cấu trúc hoặc bán cấu trúc, truy vấn với độ trễ mili giây nhất quán và tự động mở rộng quy mô mà không cần quản lý máy chủ hoặc cơ sở hạ tầng. Vì DynamoDB được xây dựng cho độ trễ thấp và tính khả dụng cao, nó thường được sử dụng để lưu trữ dữ liệu phiên, hồ sơ người dùng, siêu dữ liệu hoặc trạng thái ứng dụng. Những đặc tính tương tự này làm cho nó trở thành lựa chọn lý tưởng để lưu trữ các checkpoint và siêu dữ liệu luồng cho các tác nhân AI.

Giới thiệu LangGraph

LangGraph là một framework mã nguồn mở từ LangChain được thiết kế để xây dựng các quy trình làm việc AI phức tạp, dựa trên đồ thị. Thay vì xâu chuỗi các lời nhắc và hàm theo một đường thẳng, LangGraph cho phép bạn định nghĩa các node có thể phân nhánh, hợp nhất và lặp lại. Mỗi node thực hiện một tác vụ, và các cạnh kiểm soát luồng giữa chúng.

LangGraph giới thiệu một số khái niệm chính:

• Threads (Luồng): Một luồng là một định danh duy nhất được gán cho mỗi checkpoint chứa trạng thái tích lũy của một chuỗi các lần chạy. Khi một đồ thị thực thi, trạng thái của nó được lưu trữ bền vững vào luồng, điều này yêu cầu chỉ định một thread_id trong cấu hình ({"configurable": {"thread_id": "1"}}). Các luồng phải được tạo trước khi thực thi để lưu trữ trạng thái bền vững.
• Checkpoints (Điểm kiểm tra): Một checkpoint là một ảnh chụp nhanh trạng thái đồ thị được lưu tại mỗi siêu bước, được biểu thị bằng một đối tượng StateSnapshot chứa cấu hình, siêu dữ liệu, giá trị kênh trạng thái, các node tiếp theo để thực thi và thông tin tác vụ (bao gồm lỗi và dữ liệu ngắt). Các checkpoint được lưu trữ bền vững và có thể khôi phục trạng thái luồng sau này. Ví dụ, một đồ thị hai node đơn giản tạo ra bốn checkpoint: một checkpoint trống tại START, một với đầu vào người dùng trước node_a, một với đầu ra của node_a trước node_b, và một checkpoint cuối cùng với đầu ra của node_b tại END.
• Persistence (Tính bền vững): Tính bền vững xác định nơi và cách các checkpoint được lưu trữ (chẳng hạn như trong bộ nhớ, cơ sở dữ liệu hoặc bộ nhớ ngoài) bằng cách sử dụng triển khai checkpointer. Checkpointer lưu trạng thái luồng tại mỗi siêu bước và cho phép truy xuất các trạng thái lịch sử, cho phép các đồ thị tiếp tục từ các checkpoint hoặc khôi phục các trạng thái thực thi trước đó.

Tính bền vững là yếu tố cho phép các tính năng nâng cao như xem xét bởi con người (human-in-the-loop review), phát lại, tiếp tục sau lỗi và du hành thời gian giữa các trạng thái.

InMemorySaver là cơ chế checkpointing tích hợp của LangGraph, lưu trữ trạng thái hội thoại và lịch sử thực thi đồ thị trong bộ nhớ, cho phép các tính năng như tính bền vững, gỡ lỗi du hành thời gian và quy trình làm việc có sự tham gia của con người. Bạn có thể sử dụng InMemorySaver để tạo mẫu nhanh chóng, trạng thái chỉ tồn tại trong bộ nhớ và sẽ bị mất khi ứng dụng của bạn khởi động lại.

Hình ảnh sau đây cho thấy kiến trúc checkpointing của LangGraph, nơi một quy trình làm việc cấp cao (super-step) thực thi thông qua các node từ START đến END trong khi một checkpointer liên tục lưu các ảnh chụp nhanh trạng thái vào bộ nhớ (InMemorySaver):

Hình 1: Quy trình trong bộ nhớ

Tại sao tính bền vững lại quan trọng

Theo mặc định, LangGraph lưu trữ các checkpoint trong bộ nhớ bằng cách sử dụng InMemorySaver. Điều này rất tốt cho việc thử nghiệm vì nó không yêu cầu thiết lập và cung cấp quyền truy cập đọc và ghi tức thì.

Tuy nhiên, lưu trữ trong bộ nhớ có hai hạn chế lớn. Nó là tạm thời và cục bộ. Khi quá trình dừng, dữ liệu sẽ bị mất. Nếu bạn chạy nhiều worker, mỗi instance sẽ giữ bộ nhớ riêng của nó. Bạn không thể tiếp tục một phiên đã bắt đầu ở nơi khác và bạn không thể phục hồi nếu một quy trình làm việc bị lỗi giữa chừng.

Đối với môi trường sản xuất, điều này là không thể chấp nhận được. Bạn cần một kho lưu trữ bền vững, chịu lỗi, cho phép các tác nhân tiếp tục từ nơi chúng đã dừng, mở rộng quy mô trên các node và giữ lại lịch sử để phân tích hoặc kiểm toán. Đó là lúc DynamoDBSaver phát huy tác dụng.

Hãy tưởng tượng một kịch bản bạn đang xây dựng một tác nhân hỗ trợ khách hàng xử lý các yêu cầu phức tạp, nhiều bước. Một khách hàng hỏi về đơn hàng của họ, tác nhân truy xuất thông tin, tạo phản hồi và chờ sự chấp thuận của con người trước khi gửi phản hồi.

Nhưng điều gì sẽ xảy ra khi:

• Máy chủ của bạn hết thời gian chờ giữa quy trình làm việc?
• Bạn cần mở rộng quy mô lên nhiều worker?
• Khách hàng quay lại sau nhiều giờ để tiếp tục cuộc trò chuyện?
• Bạn muốn kiểm toán quy trình ra quyết định của tác nhân?

Với lưu trữ trong bộ nhớ, bạn sẽ gặp khó khăn. Khoảnh khắc quá trình của bạn dừng lại, mọi thứ biến mất. Mỗi worker duy trì trạng thái riêng biệt của nó. Không có cách nào để tiếp tục, phát lại hoặc xem xét những gì đã xảy ra.

Giới thiệu DynamoDBSaver

Thư viện langgraph-checkpoint-aws cung cấp một lớp lưu trữ bền vững được xây dựng đặc biệt cho AWS. DynamoDBSaver lưu trữ siêu dữ liệu checkpoint nhẹ trong DynamoDB và sử dụng Amazon S3 cho các payload lớn.

Đây là cách nó hoạt động:

1. Checkpoint nhỏ (< 350 KB): Được lưu trữ trực tiếp trong DynamoDB dưới dạng các mục được tuần tự hóa với siêu dữ liệu như thread_id, checkpoint_id, dấu thời gian và trạng thái
2. Checkpoint lớn (≥ 350 KB): Trạng thái được tải lên S3, và DynamoDB lưu trữ một con trỏ tham chiếu đến đối tượng S3
3. Truy xuất: Khi tiếp tục, saver sẽ lấy siêu dữ liệu từ DynamoDB và tải các payload lớn từ S3 một cách minh bạch

Thiết kế này cung cấp tính bền vững, khả năng mở rộng và xử lý hiệu quả cả trạng thái nhỏ và lớn mà không vượt quá giới hạn kích thước mục của DynamoDB.

DynamoDBSaver bao gồm các tính năng tích hợp để giúp bạn quản lý chi phí và vòng đời dữ liệu:

• Time-to-Live (ttl_seconds) cho phép tự động hết hạn các checkpoint theo các khoảng thời gian được chỉ định. Các trạng thái luồng cũ được dọn dẹp mà không cần can thiệp thủ công, lý tưởng cho các quy trình làm việc tạm thời, môi trường thử nghiệm hoặc các ứng dụng mà trạng thái lịch sử vượt quá một độ tuổi nhất định không còn giá trị.
• Nén (enable_checkpoint_compression) giảm kích thước checkpoint trước khi lưu trữ bằng cách tuần tự hóa và nén dữ liệu trạng thái, giúp giảm cả chi phí ghi DynamoDB và chi phí lưu trữ S3 trong khi vẫn duy trì tính toàn vẹn trạng thái đầy đủ khi truy xuất.

Cùng với nhau, các tính năng này giúp cung cấp quyền kiểm soát chi tiết đối với chi phí vận hành và dung lượng lưu trữ của lớp lưu trữ bền vững, cho phép bạn cân bằng các yêu cầu về tính bền vững với các ràng buộc ngân sách khi ứng dụng của bạn mở rộng quy mô.

Bắt đầu

Hãy cùng xây dựng một ví dụ thực tế cho thấy cách lưu trữ trạng thái tác nhân bền vững qua các lần thực thi và truy xuất các checkpoint lịch sử.

Điều kiện tiên quyết

Trước khi bắt đầu, bạn cần thiết lập các tài nguyên AWS cần thiết:

• Bảng DynamoDB: DynamoDBSaver yêu cầu một bảng để lưu trữ siêu dữ liệu checkpoint. Bảng phải có khóa phân vùng tên PK (String) và khóa sắp xếp tên SK (String).
• S3 bucket (tùy chọn): Nếu các checkpoint của bạn có thể vượt quá 350 KB, hãy cung cấp một S3 bucket để lưu trữ payload lớn. Saver sẽ tự động định tuyến các trạng thái quá khổ đến S3 và lưu trữ các tham chiếu trong DynamoDB.

Bạn có thể sử dụng AWS Cloud Development Kit (AWS CDK) để định nghĩa các tài nguyên này:

const table = new dynamodb.Table(this, 'CheckpointTable', {
    tableName: 'my_langgraph_checkpoints_table',
    partitionKey: { name: 'PK', type: dynamodb.AttributeType.STRING },
    sortKey: { name: 'SK', type: dynamodb.AttributeType.STRING },
    timeToLiveAttribute: 'ttl',
    removalPolicy: cdk.RemovalPolicy.DESTROY,
});
const bucket = new s3.Bucket(this, 'CheckpointBucket', {
    bucketName: 'amzn-s3-demo-bucket',
    encryption: s3.BucketEncryption.S3_MANAGED,
    blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
    removalPolicy: cdk.RemovalPolicy.DESTROY
});

Ứng dụng của bạn cần các quyền AWS Identity and Access Management (AWS IAM) sau để sử dụng DynamoDBSaver làm bộ lưu trữ checkpoint của LangGraph:

Truy cập bảng DynamoDB:

• dynamodb:GetItem – Truy xuất các checkpoint riêng lẻ
• dynamodb:PutItem – Lưu trữ các checkpoint mới
• dynamodb:Query – Tìm kiếm các checkpoint theo ID luồng
• dynamodb:BatchGetItem – Truy xuất nhiều checkpoint một cách hiệu quả
• dynamodb:BatchWriteItem – Lưu trữ nhiều checkpoint trong một thao tác duy nhất

Thao tác đối tượng S3 (đối với các checkpoint lớn hơn 350KB):

• s3:PutObject – Tải lên dữ liệu checkpoint
• s3:GetObject – Truy xuất dữ liệu checkpoint
• s3:DeleteObject – Xóa các checkpoint đã hết hạn
• s3:PutObjectTagging – Gắn thẻ đối tượng để quản lý vòng đời

Cấu hình S3 Bucket:

• s3:GetBucketLifecycleConfiguration – Đọc các quy tắc vòng đời
• s3:PutBucketLifecycleConfiguration – Cấu hình tự động hết hạn dữ liệu

Cài đặt

Cài đặt LangGraph và thư viện lưu trữ checkpoint của AWS bằng pip:

pip install langgraph langgraph-checkpoint-aws

Thiết lập cơ bản

Cấu hình DynamoDB checkpoint saver với bảng của bạn và S3 bucket tùy chọn cho các checkpoint lớn:

from langgraph.graph import StateGraph, END
from langgraph_checkpoint_aws import DynamoDBSaver
from typing import TypedDict, Annotatedimport operator
# Define your state
class State(TypedDict):
    foo: str
    bar: Annotated[list[str], add]
# Configure DynamoDB persistence
checkpointer = DynamoDBSaver(
    table_name="my_langgraph_checkpoints_table",
    region_name="us-east-1",
    ttl_seconds=86400 * 30,  # 30 days
    enable_checkpoint_compression=True,
    s3_offload_config={
        "bucket_name": "amzn-s3-demo-bucket",
    }
)

Xây dựng quy trình làm việc

Tạo đồ thị của bạn và biên dịch nó với checkpointer để kích hoạt trạng thái bền vững qua các lần gọi:

# thread_id for session
THREAD_ID = "99"
workflow = StateGraph(State)
workflow.add_node(node_a)
workflow.add_node(node_b)
workflow.add_edge(START, "node_a")
workflow.add_edge("node_a", "node_b")
workflow.add_edge("node_b", END)
graph = workflow.compile(checkpointer=checkpointer)
config: RunnableConfig = {"configurable": {"thread_id": THREAD_ID}}
graph.invoke({"foo": "", "bar": []}, config)

Lấy trạng thái

Truy xuất trạng thái hiện tại hoặc truy cập các checkpoint trước đó để gỡ lỗi du hành thời gian:

# get the latest state snapshot
config = {"configurable": {"thread_id": THREAD_ID}}
latest_checkpoint = graph.get_state(config)
print(latest_checkpoint)
# get a state snapshot for a specific checkpoint_id
checkpoint_id = latest_checkpoint.config.get("configurable", {}).get("checkpoint_id")
config = {"configurable": {"thread_id": THREAD_ID, "checkpoint_id": checkpoint_id}}
specific_checkpoint = graph.get_state(config)
print(specific_checkpoint)

Các trường hợp sử dụng thực tế

1. Xem xét bởi con người (Human-in-the-loop review)

Đối với các hoạt động nhạy cảm (giao dịch tài chính, tài liệu pháp lý, tư vấn y tế), bạn có thể tạm dừng quy trình làm việc để con người giám sát:

# Agent generates a response
workflow.invoke({"query": "Approve my loan"}, config)
# Human reviews in a separate process/UI
# Checkpoint is safely stored in DynamoDB
# After approval, resume
workflow.invoke({"approved": True}, config)

2. Phục hồi sau lỗi

Trong các hệ thống sản xuất, lỗi là điều không thể tránh khỏi. Gián đoạn mạng, hết thời gian chờ API hoặc lỗi tạm thời có thể làm dừng quá trình thực thi giữa chừng.

Với các checkpoint trong bộ nhớ, bạn sẽ mất tiến độ. Với DynamoDBSaver, quy trình làm việc có thể truy vấn checkpoint thành công cuối cùng và tiếp tục từ đó. Điều này giúp giảm tính toán lại, tăng tốc độ phục hồi và cải thiện độ tin cậy.

try:
    workflow.invoke({"input": "complex query"}, config)
except Exception as e:
    # Log error, alert ops team
    pass
# Later, retry from the last successful checkpoint
# No need to re-execute completed steps
workflow.invoke({}, config)

3. Các cuộc hội thoại dài hạn

Một số quy trình làm việc kéo dài hàng giờ hoặc hàng ngày. Tính bền vững của DynamoDB đảm bảo các cuộc hội thoại được lưu trữ bền vững:

# Day 1: Customer starts inquiry
workflow.invoke({"messages": ["I need help"]}, config)
# Day 2: Customer provides more info
workflow.invoke({"messages": ["Here's my account number"]}, config)
# Day 3: Agent completes the task
workflow.invoke({"action": "resolve"}, config)

Chuyển từ nguyên mẫu sang sản xuất đơn giản như việc thay đổi checkpointer của bạn. Thay thế MemorySaver bằng DynamoDBSaver để có được khả năng quản lý trạng thái bền vững, có thể mở rộng:

Hình 2: Quy trình DynamoDB

Dọn dẹp

Để tránh phát sinh chi phí liên tục, hãy xóa các tài nguyên bạn đã tạo:

Nếu bạn đã sử dụng AWS CDK để triển khai, hãy chạy lệnh sau:

cdk destroy

Nếu bạn đã sử dụng CLI, hãy chạy các lệnh sau:

• Xóa bảng DynamoDB:
  bash aws dynamodb delete-table --table-name my_langgraph_checkpoints_table
• Làm trống và xóa S3 bucket:
  bash aws s3 rm s3://amzn-s3-demo-bucket --recursive aws s3 rb s3://amzn-s3-demo-bucket

Kết luận

LangGraph giúp việc xây dựng các tác nhân thông minh, có trạng thái trở nên đơn giản. DynamoDBSaver giúp việc chạy chúng trong môi trường sản xuất trở nên an toàn.

Bằng cách tích hợp DynamoDBSaver vào các ứng dụng LangGraph của bạn, bạn có thể đạt được tính bền vững, khả năng mở rộng và khả năng tiếp tục các quy trình làm việc phức tạp từ một thời điểm cụ thể. Bạn có thể xây dựng các hệ thống có sự giám sát của con người, duy trì các phiên dài hạn và phục hồi một cách linh hoạt sau các gián đoạn.

Bắt đầu ngay hôm nay

Bắt đầu với các checkpoint trong bộ nhớ khi tạo mẫu. Khi bạn sẵn sàng triển khai, hãy chuyển sang DynamoDBSaver và để các tác nhân của bạn ghi nhớ, phục hồi và mở rộng quy mô một cách tự tin. Cài đặt thư viện bằng pip install langgraph-checkpoint-aws.

Tìm hiểu thêm về DynamoDBSaver trên tài liệu langgraph-checkpoint-aws để xem các tùy chọn cấu hình có sẵn.

Đối với các workload sản xuất, hãy cân nhắc việc lưu trữ các tác nhân LangGraph của bạn bằng cách sử dụng Amazon Bedrock AgentCore Runtime. AgentCore cung cấp một môi trường runtime được quản lý hoàn toàn, xử lý việc mở rộng quy mô, giám sát và quản lý cơ sở hạ tầng, cho phép bạn tập trung vào việc xây dựng logic tác nhân trong khi AWS quản lý sự phức tạp trong vận hành.

Về tác giả

Lee Hannigan

Lee là Kỹ sư Cơ sở dữ liệu DynamoDB cấp cao có trụ sở tại Donegal, Ireland. Anh ấy mang đến nhiều kinh nghiệm chuyên môn về các hệ thống phân tán, với nền tảng vững chắc về công nghệ dữ liệu lớn và phân tích. Trong vai trò của mình, Lee tập trung vào việc nâng cao hiệu suất, khả năng mở rộng và độ tin cậy của DynamoDB, đồng thời giúp khách hàng và các nhóm nội bộ tận dụng tối đa các khả năng của nó.