AWS实战：基于Python与Aurora pgvector构建企业级RAG应用

1. 项目概述：在AWS上构建一个基于Python的LLM RAG向量数据库应用

最近在跟几个做AI应用的朋友聊天，大家普遍有个痛点：大语言模型（LLM）本身知识有“保质期”，让它回答最新的、或者你私有的、非公开的资料时，它要么胡说八道，要么直接说“我不知道”。为了解决这个问题，RAG（检索增强生成）架构火了起来。简单说，RAG就是给LLM配一个“外接大脑”——一个向量数据库，里面存着你自己的知识文档。当用户提问时，系统先从这个“外接大脑”里找到最相关的资料，再连同问题和资料一起喂给LLM，让它基于这些“证据”来生成答案，这样既准确又可控。

这个项目，build-on-aws/llm-rag-vectordb-python，就是一个非常典型的实战指南。它手把手教你如何在亚马逊云科技（AWS）上，用Python语言，从零开始搭建一套完整的RAG应用。这不仅仅是跑通一个Demo，而是涉及从文档处理、向量化、存储、检索到最终集成的全链路工程化实践。无论你是想为自己的产品添加智能问答能力，还是想深入理解RAG背后的工程细节，这个项目都能提供一个扎实的起点。接下来，我会结合自己的实操经验，把这个项目拆解透，并补充大量官方文档里不会写的“坑”和技巧。

2. 核心架构与AWS服务选型解析

2.1 为什么选择AWS作为RAG的部署平台？

在云上构建RAG应用，选型第一件事就是云平台。AWS在这个场景下有几个天然优势。首先是生态完整性，从计算、存储、数据库到AI服务，AWS提供了一站式的解决方案，服务之间的集成和权限管理（IAM）非常成熟，能大大降低运维复杂度。其次是Serverless（无服务器）能力，对于RAG这种请求量可能波动很大的应用，采用Serverless服务可以做到按需使用、按量付费，在项目初期或流量不确定时能有效控制成本。最后是AI服务的深度集成，比如Amazon Bedrock，它提供了对多个顶尖基础模型（如Anthropic Claude、Meta Llama 3）的统一、安全的API访问，省去了自己部署和管理模型的麻烦。

2.2 核心服务栈拆解与选型理由

一个生产级的RAG应用通常包含以下几个核心环节，对应到AWS的服务选型，我的方案是这样的：

1. 文档存储与原始数据处理：使用Amazon S3。这是AWS的对象存储服务，几乎是不二之选。它的可靠性极高，成本低廉，非常适合存放原始的PDF、Word、TXT等文档。我们将所有待处理的文档上传到指定的S3存储桶（Bucket）中。

2. 文档处理与向量化（Embedding）：这是RAG的“预处理流水线”。这里我们需要一个计算服务来运行Python代码，进行文本提取、分块（Chunking）和调用Embedding模型生成向量。

   • 计算服务选择：对于灵活、短时间运行的预处理任务，我推荐使用AWS Lambda。它是一个事件驱动的无服务器计算服务。我们可以配置一个Lambda函数，当有新的文档上传到S3时，它自动被触发，执行预处理代码。对于处理时间可能超过15分钟（Lambda单次执行上限）或需要GPU加速的复杂Embedding任务，则可以考虑Amazon SageMaker的Processing Job或托管端点。
   • Embedding模型：可以选择在Lambda或SageMaker中部署开源的Embedding模型（如BAAI/bge-large-en-v1.5），但更省事的方式是直接使用Amazon Bedrock的Titan Embeddings模型。它通过一个API调用即可完成，无需管理模型基础设施，并且其生成的向量与Bedrock的文本生成模型有良好的兼容性。

3. 向量存储与检索：这是RAG的“外接大脑”核心。我们需要一个能高效存储和检索高维向量的数据库。

   • 首选方案：Amazon Aurora PostgreSQL with pgvector。这是一个托管的关系型数据库，通过pgvector扩展支持向量运算。它的优势在于：第一，可以和你现有的业务数据（用户信息、订单记录等）放在同一个数据库里，简化架构；第二，支持完整的ACID事务和SQL查询，能力强大；第三，对于中小规模的向量数据（比如数百万条以内），其性能和成本通常比专用向量数据库更有优势。这是本项目标题中vectordb部分的一个非常主流和务实的选择。
   • 备选方案：专用向量数据库。如果你的数据量极大（数亿以上），对检索延迟要求极苛刻，或者需要复杂的过滤条件，可以考虑像Pinecone（第三方，可通过AWS Marketplace使用）或Amazon OpenSearch Service（内置k-NN功能）这样的专用服务。但在大多数应用场景下，Aurora pgvector已经足够优秀。

4. 大语言模型推理与答案生成：使用Amazon Bedrock。如前所述，它提供了对多个高性能LLM的托管访问。你无需关心服务器、显卡、模型版本升级，只需通过API调用即可。它内置了安全性和负责任的AI功能，对于企业应用很重要。

5. 应用编排与API服务：最终我们需要一个服务来串联整个流程：接收用户问题 -> 检索向量数据库 -> 调用LLM生成答案 -> 返回结果。

   • API后端：可以使用AWS Lambda配合Amazon API Gateway构建一个无服务器的RESTful API。这是非常轻量和成本优化的方案。
   • 复杂工作流：如果业务流程复杂，涉及多个步骤和条件判断，可以使用AWS Step Functions来可视化地编排整个RAG流程，提高可观测性和可维护性。

整个架构的数据流大致是：S3（文档源） -> Lambda（触发处理）-> 文本分块 -> Bedrock（生成向量）-> Aurora pgvector（存储向量）-> API Gateway/Lambda（接收查询）-> pgvector（检索相似向量）-> Bedrock（生成最终答案）-> 返回给用户。

3. 实战搭建：从零到一的详细步骤

3.1 环境准备与基础设施部署

在开始写代码之前，我们需要在AWS上把“舞台”搭好。我强烈建议使用Infrastructure as Code (IaC)工具，比如AWS CloudFormation或者Terraform来部署资源，这能确保环境的一致性，也方便重建。这里我用CloudFormation的核心概念来讲解。

首先，创建一个S3存储桶，名字可以是your-company-rag-source-docs。记住要设置好生命周期策略，比如将30天前的原始文档转移到低频访问层以节省成本。

其次，创建Aurora PostgreSQL数据库集群。关键步骤是启用pgvector扩展。在创建数据库参数组时，需要添加一个参数shared_preload_libraries，其值设置为pgvector。创建完数据库实例后，连接到数据库并执行CREATE EXTENSION vector;来激活扩展。别忘了在安全组（Security Group）中开放5432端口给你的Lambda函数或应用服务器。

然后，创建用于Bedrock访问的IAM角色。Bedrock的模型访问需要显式授权。你需要在IAM控制台创建一个角色，信任实体为你要使用的服务（如Lambda），并附加AWS托管策略AmazonBedrockFullAccess（生产环境建议按需缩小权限）。

最后，准备Lambda函数的执行角色。这个角色需要权限：从上述S3桶读取对象，调用Bedrock的InvokeModelAPI，以及写入Aurora数据库。你需要精心配置其IAM策略。

注意：IAM权限是安全的核心。务必遵循最小权限原则。例如，给文档处理Lambda的S3权限，资源应精确到arn:aws:s3:::your-company-rag-source-docs/*，动作可以是s3:GetObject和s3:ListBucket。不要直接使用*通配符。

3.2 文档处理与向量化流水线实现

这是RAG效果好坏的基础。糟糕的文档分块（Chunking）会直接导致检索质量下降。

步骤一：文本提取当用户上传一个PDF到S3后，会触发S3事件通知，调用我们的处理Lambda。Lambda函数里，我们使用PyPDF2或更强大的pdfplumber库来提取文本。对于Word文档，使用python-docx；对于纯文本和HTML，处理相对简单。这里有个坑：PDF的格式千奇百怪，有些是扫描件（图片）。对于扫描件PDF，你需要先用Amazon Textract（AWS的OCR服务）进行文字识别，这比任何开源库都强大和准确，当然成本也更高。

步骤二：文本分块（Chunking）这是艺术与科学的结合。最简单的是固定大小分块，比如每500个字符一块，用langchain的RecursiveCharacterTextSplitter可以轻松实现。但更好的方法是按语义分块，比如按段落、按标题。我常用的一个策略是“重叠分块”，比如块大小500字符，重叠100字符。这能防止一个完整的语义被硬生生切断，在检索时，上下文信息也能更好地保留。对于长文档（如书籍），可以先按章节分大块，再在每个章节内进行重叠分块。

# 示例：使用LangChain的分块器 from langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=100, length_function=len, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] ) chunks = text_splitter.split_text(extracted_text)

步骤三：生成向量并存储对于每个文本块，我们调用Bedrock的Titan Embeddings模型来生成向量。Bedrock的API返回的是一个浮点数列表，即我们的向量。然后，我们将(文本块, 向量, 元数据)插入到Aurora PostgreSQL中。

import boto3 import json import psycopg2 from psycopg2.extras import execute_values bedrock_runtime = boto3.client('bedrock-runtime', region_name='us-east-1') model_id = 'amazon.titan-embed-text-v1' def get_embedding(text): body = json.dumps({"inputText": text}) response = bedrock_runtime.invoke_model(body=body, modelId=model_id) response_body = json.loads(response.get('body').read()) return response_body['embedding'] # 连接数据库 conn = psycopg2.connect(host=db_host, database=db_name, user=db_user, password=db_password) cur = conn.cursor() # 为每个chunk生成向量并准备数据 data_to_insert = [] for chunk in chunks: embedding = get_embedding(chunk) # 元数据可以包含来源文件、页码、章节等信息 metadata = json.dumps({"source": s3_key, "chunk_index": idx}) data_to_insert.append((chunk, embedding, metadata)) # 使用pgvector扩展插入数据 insert_query = "INSERT INTO document_chunks (text, embedding, metadata) VALUES %s" # 注意：embedding列的类型应为`vector(维度)`，例如`vector(1536)` execute_values(cur, insert_query, data_to_insert) conn.commit()

实操心得：在批量处理大量文档时，直接为每个小块同步调用Bedrock API可能会慢且可能遇到限流。更好的做法是使用异步调用或批量调用（如果服务支持）。此外，务必在代码中加入重试逻辑和指数退避，以处理网络波动或服务临时不可用。

3.3 检索与生成核心逻辑剖析

当用户提出一个问题时，后端API（另一个Lambda函数）需要执行以下步骤：

1. 问题向量化：将用户的问题（Query）用与之前相同的Embedding模型转化为向量。

2. 向量检索：在Aurora pgvector中执行相似度搜索，找出与问题向量最接近的K个文本块。pgvector支持多种距离计算方式，最常用的是余弦相似度（<=>操作符）或L2距离（<->）。

3. 上下文构建：将检索到的Top K个文本块，连同它们的一些元数据（如来源），组合成一个“上下文”字符串。通常会在每个块前加上“出处：”之类的提示。

4. 提示词工程：构建最终发送给LLM（如Claude 3 Haiku）的提示词（Prompt）。这是决定答案质量的关键。一个经典的模板是：

   你是一个专业的助手，请严格根据以下提供的上下文信息来回答问题。如果上下文信息不足以回答问题，请直接说“根据提供的资料，我无法回答这个问题”。不要编造信息。 上下文信息： {context} 问题：{question} 请根据上下文回答：

5. 调用LLM生成答案：将组装好的提示词发送给Bedrock的对话模型，获取生成的答案。

6. 返回结果：将答案、以及可选的引用来源（检索到的文本块及其元数据）一并返回给前端。

-- 在pgvector中检索相似向量的SQL示例 SELECT id, text, metadata, 1 - (embedding <=> %s) as cosine_similarity FROM document_chunks ORDER BY embedding <=> %s LIMIT 5;

注意事项：检索数量K需要权衡。K太小，可能遗漏关键信息；K太大，会引入噪声并增加LLM的上下文长度（Token数），导致成本增加和速度变慢。通常从3-5开始测试。另外，可以考虑在检索时加入元数据过滤，比如“只从某份手册中检索”，这能极大提升精准度。

4. 性能优化与成本控制实战策略

4.1 向量检索性能提升技巧

当你的向量表达到百万级别时，简单的全表扫描排序将变得非常慢。pgvector提供了索引来加速检索。

最常用的索引是IVFFlat索引。它的原理类似于K-Means聚类，先将所有向量分成若干簇（lists），检索时先找到距离目标向量最近的几个簇，然后在这几个簇里进行精确搜索。创建索引的SQL如下：

CREATE INDEX ON document_chunks USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);

这里的lists参数是关键。一个经验法则是：lists = sqrt(行数)。对于100万行数据，lists设为1000是合理的起点。但是，必须在表有足够多的数据（比如每个list至少有1000条记录）后再创建索引，并且索引创建后，如果大量新增数据，需要重建索引（REINDEX）以保持性能。

对于极致性能要求，可以探索HNSW索引，它提供了更优的查询速度和召回率，但创建索引的时间更长，占用的空间也更大。

4.2 多轮对话与历史上下文管理

基础的RAG是“单轮”的。但在聊天场景中，用户的问题往往有上下文关联。例如，用户先问“AWS S3是什么？”，接着问“它怎么收费？”。第二个问题中的“它”指代S3。

实现多轮对话RAG有两种主流思路：

1. 将历史对话也纳入检索：将当前问题与之前的几轮对话（Q&A对）拼接起来，作为一个新的查询语句去检索。这能帮助系统理解指代关系。
2. 让LLM自己管理上下文：使用具有长上下文窗口的模型（如Claude 200K），将检索到的当前上下文和整个对话历史都喂给模型，由模型自行理解。这种方法更简单，但对模型能力要求高，且Token消耗大。

在Lambda无服务器架构下，需要将会话状态（对话历史）存储到外部，如Amazon DynamoDB。每次请求时，先从DynamoDB读取该会话的历史记录，再执行上述流程，最后将本轮新的Q&A对写回DynamoDB。

4.3 成本监控与优化措施

Serverless架构按用量付费，成本透明但也需要关注。

• Bedrock成本：主要来自Embedding调用和文本生成调用。Embedding按Token数计费，文本生成按输入+输出Token总数计费。优化点：优化提示词，减少不必要的指令；控制检索上下文的长度；对于简单问题，可以考虑使用更小、更便宜的模型（如Titan Text Express）。
• Lambda成本：按请求次数和执行时间（GB-秒）计费。优化点：确保函数代码高效（如使用json.loads而非eval）；合理设置内存大小（更大的内存可能带来更快的执行速度，从而降低时间成本）；利用执行环境复用保持数据库连接等。
• Aurora成本：按数据库实例规格、存储和I/O计费。优化点：根据负载选择正确的实例大小；为向量表设置合适的索引，减少全表扫描带来的CPU消耗；定期清理或归档不再需要的旧数据。
• 监控：务必启用AWS Cost Explorer和设置预算告警。在CloudWatch中为Bedrock、Lambda的调用次数和持续时间设置仪表盘，以便直观了解使用模式和发现异常。

5. 常见问题排查与进阶思考

5.1 效果不佳问题诊断清单

如果你的RAG应用回答不准确或胡编乱造，可以按以下清单排查：

5.2 安全性与权限管控

企业级应用必须考虑安全。

• 数据加密：确保S3桶、Aurora数据库都启用了加密（静态加密）。Bedrock的通信默认使用HTTPS（传输中加密）。
• 网络隔离：将Lambda函数、Aurora数据库部署在私有子网（Private Subnet）中。为Lambda配置VPC，通过VPC端点（PrivateLink）访问Bedrock和S3，避免流量经过公网。
• 权限精细化管理：如前所述，为每个Lambda函数创建独立的、权限最小的IAM角色。使用数据库用户和密码（或IAM认证）并限制其只有特定表的读写权限。
• 输入输出审查：对用户输入进行基本的清理和检查，防止提示词注入攻击。对模型输出内容（尤其是面向公众时）可以考虑增加一层安全审查或过滤。

5.3 从项目到产品：可观测性与持续改进

一个实验项目和一个可运营的产品之间，差的是一个完整的可观测性体系。

• 日志记录：在Lambda函数中关键步骤（检索开始、检索结束、调用LLM前、收到回答后）使用print或logging输出结构化日志（JSON格式），并包含请求ID、检索到的文档ID、Token使用量等信息。CloudWatch Logs会自动收集这些日志。
• 指标监控：利用CloudWatch自定义指标，记录每次问答的端到端延迟、检索到的上下文数量、LLM调用Token数、用户反馈（如有）。这能帮助你发现性能退化或成本异常。
• 追踪（Tracing）：使用AWS X-Ray来追踪一次用户请求流经API Gateway、Lambda、Bedrock、Aurora的完整路径，直观定位延迟瓶颈。
• 反馈循环：设计机制收集用户反馈，比如“这个回答是否有用？”的按钮。将用户问题、系统检索到的上下文、生成的答案以及用户反馈一起存储下来。这些数据是优化分块策略、提示词和检索参数的黄金燃料。

搭建这个RAG系统的过程，就像在组装一个精密的仪器。每个环节——文档分块、向量模型、检索策略、提示词——都需要反复调试和校准。没有一劳永逸的“最佳配置”，只有最适合你特定文档和业务场景的配置。我的建议是，先用一个最小可行产品（MVP）跑通全流程，然后用一批典型问题作为测试集，定量地评估回答质量，再针对性地迭代优化其中一个环节，如此循环。在这个过程中，你会对语义搜索和生成式AI有更深刻、更实战的理解。