当前位置：首页 > news >正文

基于 Amazon S3 Vectors + OpenClaw 的 RAG 知识库架构与实现

news 2026/5/12 5:28:35

AI 助手有个致命短板：它不认识你的业务。公司文档、部署流程、内部 FAQ——这些东西不在训练数据里，问它只会瞎编。

更烦的是没记忆。聊天记录一清，之前说的全白费。

我花了不少时间研究这个问题。RAG（检索增强生成）是目前公认的解法，但向量数据库的运维成本一直让我犹豫。直到亚马逊云科技推出了 S3 Vectors——在 S3 上直接存向量，不用额外服务。这让整个方案简洁了很多。

本文完整记录我用 S3 Vectors + OpenClaw 搭建企业内部知识库的过程，包括架构设计、代码实现、集成方式和成本分析。

架构概览

整体采用 RAG 架构：

用户提问 → OpenClaw 接收↓文本向量化（Bedrock Embeddings）↓语义搜索（S3 Vectors）↓相关文档片段 + 原始问题 → 大模型生成回答↓返回用户

核心组件：

S3 Vectors：向量存储和检索，亚马逊云科技的 S3 原生能力
Bedrock Embeddings：amazon.titan-embed-text-v2:0，文本转 1024 维向量
OpenClaw：AI 助手框架，通过 Skill 机制集成知识库搜索

S3 Vectors 核心概念

S3 Vectors 的设计很克制，只有三层抽象：

概念	作用	类比
Vector Bucket	顶层容器	S3 Bucket
Vector Index	索引，定义维度和距离度量	数据库的表
Vector	具体数据，含 key、向量、元数据	表里的行

不需要管集群、分片、副本。S3 的可靠性和可用性直接继承。

实现步骤

一、初始化存储

import boto3s3vectors = boto3.client('s3vectors')# 创建向量桶
s3vectors.create_vector_bucket(vectorBucketName='company-kb'
)# 创建向量索引
s3vectors.create_vector_index(vectorBucketName='company-kb',vectorIndexName='docs-index',dimension=1024,distanceMetric='cosine'
)

dimension=1024 对应 Titan Embed Text v2 的输出。cosine 余弦距离，文本语义搜索的标准选择。

二、文档预处理

文档预处理包括两步：切分和向量化。切分策略直接影响搜索质量，这是我踩坑比较多的地方。

import boto3
import jsondef get_embedding(text):"""Bedrock Titan Embed 文本向量化"""bedrock = boto3.client('bedrock-runtime')response = bedrock.invoke_model(modelId='amazon.titan-embed-text-v2:0',body=json.dumps({'inputText': text}))return json.loads(response['body'].read())['embedding']def split_text(text, chunk_size=500, overlap=50):"""固定窗口切分，带重叠"""chunks = []start = 0while start < len(text):end = start + chunk_sizechunk = text[start:end]if chunk.strip():chunks.append(chunk.strip())start = end - overlapreturn chunks

几点经验：

chunk_size=500 对中文技术文档效果不错，太小则上下文不完整，太大则搜索精度下降
overlap=50 防止句子被切断，保持语义连贯
如果文档结构清晰（有标题分节），按节切分效果更好，但实现复杂度也高

三、批量写入

import boto3
import jsondef get_embedding(text):bedrock = boto3.client('bedrock-runtime')response = bedrock.invoke_model(modelId='amazon.titan-embed-text-v2:0',body=json.dumps({'inputText': text}))return json.loads(response['body'].read())['embedding']def split_text(text, chunk_size=500, overlap=50):chunks = []start = 0while start < len(text):end = start + chunk_sizechunk = text[start:end]if chunk.strip():chunks.append(chunk.strip())start = end - overlapreturn chunksdef ingest_documents(documents):"""处理并写入文档集合"""s3vectors = boto3.client('s3vectors')total = 0for doc in documents:chunks = split_text(doc['content'])vectors = []for i, chunk in enumerate(chunks):vectors.append({'key': f"{doc['id']}-{i:03d}",'data': {'float32': get_embedding(chunk)},'metadata': {'title': doc['title'],'source': doc.get('source', 'unknown'),'chunk_index': str(i),'content': chunk}})batch_size = 20for j in range(0, len(vectors), batch_size):batch = vectors[j:j + batch_size]s3vectors.put_vectors(vectorBucketName='company-kb',vectorIndexName='docs-index',vectors=batch)total += len(batch)print(f'完成，共写入 {total} 条向量')# 示例
ingest_documents([{'id': 'deploy-guide','title': '服务部署指南','content': '我们的服务使用 Docker 容器化部署。''拉取镜像后执行 docker-compose up -d。''健康检查端点 /health 返回 200 表示正常。''日志通过 CloudWatch 收集。','source': 'wiki'}
])

元数据设计很重要。content 存原文片段方便回显，source 存来源方便后续过滤和溯源。

四、语义搜索

import boto3
import jsondef get_embedding(text):bedrock = boto3.client('bedrock-runtime')response = bedrock.invoke_model(modelId='amazon.titan-embed-text-v2:0',body=json.dumps({'inputText': text}))return json.loads(response['body'].read())['embedding']def search(query, top_k=5):"""语义搜索知识库"""s3vectors = boto3.client('s3vectors')results = s3vectors.query_vectors(vectorBucketName='company-kb',vectorIndexName='docs-index',queryVector={'float32': get_embedding(query)},topK=top_k)hits = []for vec in results.get('vectors', []):meta = vec.get('metadata', {})hits.append({'key': vec['key'],'score': vec.get('score', 0),'title': meta.get('title', ''),'content': meta.get('content', ''),'source': meta.get('source', '')})return hits# 测试
for h in search('怎么部署到生产环境'):print(f"[{h['score']:.4f}] {h['title']}")print(f"  {h['content'][:100]}")

五、OpenClaw 集成

OpenClaw 通过 Skill 机制扩展能力。创建知识库搜索 Skill：

knowledge-base/SKILL.md：

# Knowledge Base Search搜索企业内部知识库。当用户提问涉及公司内部信息时，
先搜索知识库获取相关文档，再基于文档内容回答。

knowledge-base/scripts/search.py：

#!/usr/bin/env python3
"""知识库搜索脚本"""
import sys
import json
import boto3def get_embedding(text):bedrock = boto3.client('bedrock-runtime')response = bedrock.invoke_model(modelId='amazon.titan-embed-text-v2:0',body=json.dumps({'inputText': text}))return json.loads(response['body'].read())['embedding']def search(query, top_k=5):s3vectors = boto3.client('s3vectors')results = s3vectors.query_vectors(vectorBucketName='company-kb',vectorIndexName='docs-index',queryVector={'float32': get_embedding(query)},topK=top_k)return [{'title': v.get('metadata', {}).get('title', ''),'content': v.get('metadata', {}).get('content', ''),'source': v.get('metadata', {}).get('source', ''),'score': v.get('score', 0)}for v in results.get('vectors', [])]if __name__ == '__main__':q = sys.argv[1] if len(sys.argv) > 1 else ''if not q:print('用法: python search.py "查询内容"')sys.exit(1)print(json.dumps(search(q), ensure_ascii=False, indent=2))

OpenClaw 在识别到知识库相关问题时，自动调用脚本，将返回的文档片段注入到 prompt context 中，模型基于这些"参考资料"生成回答。