使用 NVIDIA NeMo Agent Toolkit 和 Amazon S3 Vectors 构建智能体记忆

在我之前的文章《使用 Amazon S3 Vectors 为多智能体 AI 系统构建持久记忆》中，我们探讨了为什么记忆工程是生产级多智能体系统的基础学科。我们展示了 Amazon S3 Vectors（Amazon Simple Storage Service 的一项能力）如何满足智能体记忆的架构要求：语义检索、丰富的元数据、强一致性和弹性扩展。

在本文中，我们从架构转向实现。展示如何在 NVIDIA NeMo Agent Toolkit（NAT）内使用 Amazon S3 Vectors 作为持久记忆层，并部署在 Amazon Elastic Kubernetes Service（Amazon EKS）上以获得完整的运营控制。

读完本文后，你将理解 NAT 的记忆子系统是如何工作的，以及如何将 Amazon S3 Vectors 实现为自定义记忆提供程序。还将学习如何在 Amazon EKS 上部署该技术栈，以多智能体投资研究用例作为贯穿全文的示例。

什么是 NVIDIA NeMo Agent Toolkit？

NVIDIA NeMo Agent Toolkit（NAT）是一个用于构建、性能分析和优化 AI 智能体的开源框架。它与框架无关，可与 Strands Agents、LangChain、LlamaIndex、CrewAI 以及自定义实现配合使用。NAT 为生产级智能体系统提供四项相关能力：

智能体编排 – 将智能体定义为可组合的工作流，配置大型语言模型（LLM）、工具和提示。可以通过 nat run 在本地运行，或通过 nat serve 作为持久服务运行。

性能分析 – 跨智能体和各个工具追踪 token 使用量、延迟、吞吐量和运行时间，以识别多智能体工作流中的瓶颈。

评估 – 内置答案准确性、上下文相关性、响应有据性和智能体轨迹评估器，并支持自定义评估器。

优化 – 自动超参数调优（temperature、top_p、max_tokens），在最大化质量的同时最小化成本和延迟。

NAT 包含一个专门的记忆子系统，旨在跨智能体调用存储和检索对话历史、用户偏好和长期知识。记忆模块可扩展：你通过实现 NAT 的插件接口来创建自定义记忆提供程序（后端）。关键组件包括：

• MemoryEditor – 所有记忆后端必须实现的抽象接口。定义了三个方法：add_items()、search() 和 remove_items()。

• MemoryItem – 表示一条记忆的数据模型，包含对话历史、标签、元数据、user_id 和可选的文本记忆字符串字段。

• MemoryBaseConfig – Pydantic 基类，自定义记忆配置继承自该类。NAT 通过 YAML 配置中的 _type 字段发现提供程序。

• Automatic memory wrapper – auto_memory_agent 工作流类型，自动封装智能体的记忆捕获和检索，无需 LLM 显式调用记忆工具。

NAT 目前内置了以下记忆提供程序：Mem0、MemMachine、Redis 和 Zep。这些覆盖了常见用例。然而，对于需要弹性向量存储、强写入一致性和成本高效扩展到数十亿向量的生产级多智能体系统，基于 Amazon S3 Vectors 的自定义提供程序是最佳选择。

为什么选择 Amazon S3 Vectors 作为持久记忆后端

《使用 Amazon S3 Vectors 为多智能体 AI 系统构建持久记忆》深入介绍了架构原理。下表总结了在 NAT 记忆层中选择 S3 Vectors 的优势：

特性	S3 Vectors	自托管向量数据库
扩展性	原生弹性扩展至数十亿向量	需要手动扩容
一致性	强写入一致性	最终一致性
运维成本	零运维，无需管理基础设施	需要专职 DBA 和运维团队
成本模型	按实际使用付费	固定基础设施成本
可用性	11 个 9 的可用性	依赖部署架构

要跟随本文的实现，你需要：

• 一个 AWS 账户，具有创建 Amazon S3 Vectors 资源和 Amazon EKS 集群的权限。
• 一个现有的 Amazon EKS 集群（参见 Amazon EKS 入门）。
• 已安装 NVIDIA NeMo Agent Toolkit（已使用 1.6 版本测试）。
• Python 3.11 或 3.12（NAT 要求 >=3.11，<3.14）。
• 一个用于生成向量表示的嵌入模型（本文使用 Amazon Titan Text Embeddings V2）。
• 用于容器和集群操作的 kubectl 和 Docker。

将 S3 Vectors 实现为 NAT 记忆提供程序

实现分为三个步骤：

步骤 1：创建 S3 Vectors 基础设施

步骤 2：实现自定义 MemoryEditor 插件

步骤 3：配置智能体工作流

图 1：实现 Amazon S3 Vectors 作为 NAT 记忆提供程序的三个步骤

步骤 1. 创建 Amazon S3 Vectors 基础设施

以下代码创建一个向量存储桶和一个索引，其元数据架构专为智能体记忆设计。索引使用 1024 维以匹配 Amazon Titan Text Embeddings V2 的输出，并将大 content 字段标记为不可过滤元数据：

import boto3

REGION = "us-west-2"
VECTOR_BUCKET = "amzn-s3-demo-research-agent-memory"
INDEX_NAME = "agent-long-term-memory"

s3vectors = boto3.client("s3vectors", region_name=REGION)

# 创建向量存储桶
s3vectors.create_vector_bucket(vectorBucketName=VECTOR_BUCKET)

# 创建索引（1024 维匹配 Amazon Titan Text Embeddings V2）
s3vectors.create_index(
    vectorBucketName=VECTOR_BUCKET,
    indexName=INDEX_NAME,
    dataType="float32",
    dimension=1024,
    distanceMetric="cosine",
    metadataConfiguration={"nonFilterableMetadataKeys": ["content"]},
)

创建存储桶和索引后，你可以实现读写这些资源的记忆提供程序。

步骤 2. 实现自定义 MemoryEditor 插件

以下代码将 MemoryEditor 接口实现为 Amazon S3 Vectors 后端，并注册它以便 NAT 可以发现它：

import boto3
import json
import uuid
from datetime import datetime, timezone
from nat.plugin_api import MemoryBaseConfig, MemoryEditor, MemoryItem, register_memory
from nat.builder.builder import Builder

class S3VectorsMemoryConfig(MemoryBaseConfig, name="s3vectors_memory"):
    """NAT memory provider configuration for Amazon S3 Vectors."""
    vector_bucket: str
    index_name: str
    aws_region: str = "us-west-2"
    default_top_k: int = 5

class S3VectorsMemoryEditor(MemoryEditor):
    """NAT MemoryEditor backed by Amazon S3 Vectors."""

    def __init__(self, config: S3VectorsMemoryConfig):
        self._vector_bucket = config.vector_bucket
        self._index_name = config.index_name
        self._default_top_k = config.default_top_k
        self._s3vectors = boto3.client('s3vectors', region_name=config.aws_region)
        self._bedrock = boto3.client('bedrock-runtime', region_name=config.aws_region)

    def _get_embedding(self, text: str) -> list[float]:
        """Generate embeddings using Amazon Titan Text Embeddings V2."""
        response = self._bedrock.invoke_model(
            modelId='amazon.titan-embed-text-v2:0',
            contentType='application/json',
            accept='application/json',
            body=json.dumps({
                'inputText': text,
                'dimensions': 1024,
                'normalize': True
            })
        )
        return json.loads(response['body'].read())['embedding']

    async def add_items(self, items: list[MemoryItem], **kwargs) -> None:
        """Store memory items as vectors in Amazon S3 Vectors."""
        vectors = []
        for item in items:
            # Build the text to embed from the memory content
            text = item.memory or json.dumps(item.conversation)
            embedding = self._get_embedding(text)
            # Use uuid4 to prevent key collisions within the same second
            key = f"mem_{item.user_id}_{uuid.uuid4().hex[:12]}"
            # Note: S3 Vectors metadata values have size limits.
            # Truncate content to 1024 characters for production use.
            content_for_metadata = text[:1024]
            mem_metadata = {
                'user_id': item.user_id,
                'memory_type': item.metadata.get('memory_type', 'episodic'),
                'agent_id': item.metadata.get('agent_id', ''),
                'team_id': item.metadata.get('team_id', ''),
                'task_id': item.metadata.get('task_id', ''),
                'confidence': item.metadata.get('confidence', 0.8),
                'created_at_epoch': int(datetime.now(timezone.utc).timestamp()),
                'is_shared': item.metadata.get('is_shared', True),
                'source': item.metadata.get('source', 'agent'),
                'content': content_for_metadata,
            }
            # Add domain-specific metadata if present
            if 'ticker' in item.metadata:
                mem_metadata['ticker'] = item.metadata['ticker']
            vectors.append({
                'key': key,
                'data': {'float32': embedding},
                'metadata': mem_metadata
            })
        self._s3vectors.put_vectors(
            vectorBucketName=self._vector_bucket,
            indexName=self._index_name,
            vectors=vectors
        )

```python
async def search(self, query: str, top_k: int = None, **kwargs) -> list[MemoryItem]:
        """从 Amazon S3 Vectors 中检索语义相关的记忆。"""
        query_embedding = self._get_embedding(query)
        effective_top_k = top_k or self._default_top_k
        # 从 kwargs 构建元数据过滤器
        filter_expr = {}
        for field in ('agent_id', 'memory_type', 'ticker', 'team_id', 'user_id'):
            if field in kwargs:
                filter_expr[field] = {'$eq': kwargs[field]}
        # 支持 is_shared 的布尔过滤器
        if 'is_shared' in kwargs:
            filter_expr['is_shared'] = {'$eq': kwargs['is_shared']}
        response = self._s3ve

该插件使用 Amazon Titan Text Embeddings V2 生成嵌入向量，将每条记忆存储为带作用域元数据的向量，并将搜索过滤器转换为 Amazon S3 Vectors 元数据查询。

Step 3. 配置 NAT 智能体工作流

定义了插件后，在 NAT 的 YAML 配置中进行配置，并将其接入智能体工作流：

# config.yml - NAT 智能体配置，包含 Amazon S3 Vectors 记忆层
memory:
  agent_memory:
    _type: s3vectors_memory
    vector_bucket: "amzn-s3-demo-research-agent-memory"
    index_name: "agent-long-term-memory"
    aws_region: "us-west-2"

functions:
  add_memory:
    _type: add_memory
    memory: agent_memory
    description: |
      将重要发现、模式或事实存储到长期记忆。
      在研究过程中发现新信息后使用。
  get_memory:
    _type: get_memory
    memory: agent_memory
    description: |
      开始研究前检索相关先验知识。
      使用当前研究主题进行查询，以召回相关发现。
  web_search:
    _type: web_search
  financial_data:
    _type: financial_data_api

workflow:
  _type: auto_memory_agent
  inner_agent_name: research_agent
  memory_name: agent_memory
  llm_name: bedrock_llm
  save_user_messages_to_memory: true
  retrieve_memory_for_every_response: true
  save_ai_messages_to_memory: true

llm:
  bedrock_llm:
    _type: bedrock
    model_id: "anthropic.claude-sonnet-4-20250514"
    temperature: 0.3

通过 auto_memory_agent 包装器，可以自动捕获和检索记忆。用户消息和智能体响应都会被存储，相关上下文在每次智能体调用前被注入。这种设计无需 LLM 显式调用记忆工具。

负责任的 AI 与数据处理注意事项

由于此解决方案会持久化对话历史和用户记忆，需要规划这些数据的保留和访问方式。为存储的记忆设置保留策略，并使用本文中的记忆整合与删除路径来清理不再需要的数据。避免在记忆元数据中存储个人身份信息（PII），在嵌入敏感字段前先进行脱敏或令牌化处理。使用每租户索引和最小权限 IAM 策略来限制访问范围，使每个智能体只能读写其拥有的记忆。

将 Amazon S3 Vectors 投入实践：多智能体投资研究

以下用例演示了这种模式在实际中的应用。假设三个专业智能体协作完成一个投资研究工具：

• 研究智能体（Research Agent） — 收集市场数据、财报和新闻。
• 分析智能体（Analysis Agent） — 执行定量分析并识别模式。
• 综合智能体（Synthesis Agent） — 整合两个智能体的发现生成报告。

借助持久化记忆，每个智能体都能在前人工作的基础上继续推进。研究智能体可以召回之前收集的数据，避免重复的 API 调用。分析智能体建立在先前会话中识别的模式之上。综合智能体访问累积的发现，以生成内容日渐丰富的报告。

多智能体记忆协调

每个智能体使用同一个 S3 Vectors 索引，但在写入时在元数据中带上自己的 agent_id。智能体使用元数据过滤器检索共享知识：

# 分析智能体检索研究智能体共享的发现
research_findings = await memory_client.search(
    query=f"关于 {ticker} 的最新研究发现",
    top_k=10,
    team_id='investment-research',
    is_shared=True
)

# 综合智能体查询团队全部知识
all_team_knowledge = await memory_client.search(
    query=f"{ticker} 2026 年 Q2 的完整分析研究",
    top_k=20,
    team_id='investment-research'
)

NAT 内置的多租户记忆隔离使用 user_id 按用户划分记忆范围。对于多智能体团队协作，team_id 元数据字段在同一 S3 Vectors 索引内提供了额外的分组维度。

随着时间推移，情景记忆会不断积累。定期将其整合为语义记忆（泛化知识）可以保持检索的准确性和效率。你可以通过定时 cron 任务、向量数量阈值或智能体在可配置的若干研究周期后主动发起来触发整合：

async def consolidate_memories(ticker: str, memory_client, workflow):
    """将情景记忆提炼为持久的语义知识。"""
    episodes = await memory_client.search(
        query=f"关于 {ticker} 的所有发现",
        top_k=50,
        memory_type='episodic',
        ticker=ticker
    )
    episode_texts = [e.memory for e in episodes if e.memory]
    consolidation_prompt = f"""给定关于 {ticker} 的这 {len(episode_texts)} 条观察，
请识别出持久性模式和泛化知识。

{chr(10).join(episode_texts)}

返回一个洞察字符串的 JSON 数组。请勿包含具体日期或一次性事件。"""
    response = await workflow.run(consolidation_prompt)
    # 将响应解析为洞察列表
    insights = json.loads(response)
    # 将每个整合后的洞察存储为语义记忆
    await memory_client.add_items([
        MemoryItem(
            conversation=[],
            tags=['semantic'],
            metadata={
                'memory_type': 'semantic',
                'confidence': 0.9,
                'ticker': ticker,
                'source': 'consolidation',
                'is_shared': True,
                'team_id': 'investment-research'
            },
            user_id='system',
            memory=insight
        )
        for insight in insights
    ])

在 Amazon EKS 上部署智能体工作负载

对于需要完全掌控智能体工作负载运营控制的团队来说，Amazon EKS 是一个深思熟虑的选择。通过 Amazon EKS，你可以控制扩展、网络和生命周期管理，并与 AWS Identity and Access Management（IAM）原生集成以访问 Amazon S3 Vectors。如果你偏好全托管运行时，Amazon Bedrock AgentCore 提供了一种移除运营开销的托管替代方案。你使用 nat serve 将 NAT 智能体作为容器化服务运行。每种智能体类型都是一个 Kubernetes Deployment，通过 IAM Roles for Service Accounts（IRSA）授予 IAM 访问权限。

Step 1: 构建智能体容器

以下 Dockerfile 将智能体及其配置和记忆插件打包：

FROM python:3.12-slim
RUN pip install nvidia-nat[langchain] boto3
COPY config.yml /app/config.yml
COPY s3vectors_memory.py /app/s3vectors_memory.py
WORKDIR /app
CMD ["nat", "serve", "--config_file", "config.yml"]

Step 2: 使用 Kubernetes manifest 部署

以下 manifest 部署了带自动扩展的研究智能体：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: research-agent
  namespace: agent-team
spec:
  replicas: 2
  selector:
    matchLabels:
      app: research-agent
  template:
    metadata:
      labels:
        app: research-agent
        team: investment-research
    spec:
      serviceAccountName: agent-sa
      containers:
      - name: agent
        image: ${ECR_REGISTRY}/research-agent:latest # Amazon Elastic Container Registry (Amazon ECR)
        ports:
        - containerPort: 8000
        env:
        - name: VECTOR_BUCKET
          value: "amzn-s3-demo-research-agent-memory"
        - name: INDEX_NAME
          value: "agent-long-term-memory"
        - name: AWS_REGION
          value: "us-west-2"
        resources:
          requests:
            cpu: "500m"
            memory: "1Gi"
          limits:
            cpu: "2000m"
            memory: "4Gi"
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: research-agent-hpa
  namespace: agent-team
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: research-agent
  minReplicas: 1
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70

用于 S3 Vectors 访问的 IAM 策略

以下 IAM 策略授予记忆层的限定访问权限。通过 IRSA 将其附加到与 agent-sa 服务账号关联的 IAM 角色：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3vectors:PutVectors",
                "s3vectors:QueryVectors",
                "s3vectors:GetVectors",
                "s3vectors:DeleteVectors"
            ],
            "Resource": "arn:aws:s3vectors:us-west-2:${ACCOUNT_ID}:vector-bucket/amzn-s3-demo-research-agent-memory/*"
        }
    ]
}

构建并推送容器镜像后，部署智能体：

# 创建命名空间
kubectl create namespace agent-team

# 应用清单文件
kubectl apply -f kubernetes/research-agent-deployment.yaml

# 验证 pods 运行状态
kubectl get pods -n agent-team

# 测试智能体端点
kubectl port-forward -n agent-team svc/research-agent 8000:8000
curl -X POST http://localhost:8000/generate \
  -H "Content-Type: application/json" \
  -d '{"inputs": "What are the latest AAPL earnings trends?"}'

S3 Vectors 提供强写入一致性。Pods 和智能体类型可以立即看到智能体 pods 存储的记忆，无需缓存失效。HorizontalPodAutoscaler 基于 CPU 利用率扩展智能体副本。每个副本通过 IRSA 连接到同一个 S3 Vectors 索引，支持无论哪个 pod 处理请求都能实现一致的内存访问。

评估记忆的影响

通过 NAT 的评估工具（nat eval），你可以量化记忆对智能体性能的影响。配置两次评估运行：一次启用记忆，一次禁用，使用相同的数据集：

# 启用记忆评估
nat eval --config_file config_with_memory.yml \
  --dataset eval_dataset.jsonl \
  --metrics accuracy,groundedness,token_usage,latency

# 禁用记忆评估（基准）
nat eval --config_file config_no_memory.yml \
  --dataset eval_dataset.jsonl \
  --metrics accuracy,groundedness,token_usage,latency

基于该设计的架构属性，在比较这些运行时，你可以预期以下定性结果。这些是方向性预期，而非基准测量：

• 有据性提升 —— 召回的记忆提供可验证的来源上下文，智能体可以引用它，而不仅依赖参数化知识。
• Token 用量减少 —— 智能体跳过重新推导先前会话中已建立的事实。减少幅度与工作流中重复上下文的多少成正比。
• 延迟适度增加 —— 每次记忆召回都会增加一次 S3 Vectors 查询（在标准访问模式下为亚秒级），通常是整体 LLM 推理时间的一小部分。
• 重复工作减少 —— 有了共享记忆，智能体避免重复团队成员已完成的研究或分析。

这些效果的幅度取决于你的工作负载：跨会话重用多少上下文、多少个智能体协调，以及你的检索预算（top-k）有多大。NAT 的超参数优化器可以系统地调优 top_k 和相似度阈值，为你的用例找到最佳成本-质量权衡。

为避免本演练创建的资源产生持续费用，请删除以下内容：

删除 Amazon EKS 部署及相关资源：kubectl delete namespace agent-team

kubectl delete namespace agent-team

删除 S3 Vectors 索引和向量桶：s3vectors.delete_index(vectorBucketName='amzn-s3-demo-research-agent-memory', indexName='agent-long-term-memory') s3vectors.delete_vector_bucket(vectorBucketName='amzn-s3-demo-research-agent-memory')

s3vectors.delete_index(vectorBucketName='amzn-s3-demo-research-agent-memory', indexName='agent-long-term-memory')
s3vectors.delete_vector_bucket(vectorBucketName='amzn-s3-demo-research-agent-memory')

如果不再需要，删除为 IRSA 创建的 IAM 角色和策略。

这篇文章展示了如何实现由 Amazon S3 Vectors 支持的自定义 NAT 记忆提供程序，并在 Amazon EKS 上部署它。关键构建块包括：

• NAT 的 MemoryEditor 接口为自定义记忆后端提供插件契约。
• auto_memory_agent 包装器自动处理捕获和检索。
• Amazon S3 Vectors 提供持久化、可语义查询的后端，具有强一致性，支持多智能体协调。
• Amazon EKS 让你对智能体生命周期、扩展和网络隔离拥有完整的运营控制。

这些模式适用于智能体从积累知识中受益的领域。包括将记忆分类为情景记忆、语义记忆或程序记忆，通过元数据过滤在智能体间共享记忆，以及用 LLM 整合记忆。示例包括客户支持、DevOps 自动化、法律研究和科学发现。

有关本文引用的记忆架构基础，请参阅 Building persistent memory for multi-agent AI systems with Amazon S3 Vectors。要开始使用 NAT，请参阅 NVIDIA NeMo Agent Toolkit 文档和 Adding a Memory Provider 指南。