基于某中心S3向量与DeepSeek R1构建企业级RAG应用

企业级RAG应用的技术挑战

组织正在采用DeepSeek R1等大语言模型(LLM)来改造业务流程，但独立LLM存在幻觉、知识过时等局限。检索增强生成(RAG)通过将语义搜索与生成式AI结合，使模型能够从企业知识库中检索相关信息后再生成响应。然而随着RAG解决方案的普及，生产环境中的扩展性挑战也随之增加，包括向量数据库带来的成本与基础设施复杂性。

传统向量数据库存在四大痛点：

不可预测的成本 - 传统方案常需过度配置
运维复杂性 - 需投入工程资源管理专用集群
扩展限制 - 向量集合增长时容量规划困难
集成开销 - 与现有数据管道对接增加摩擦

解决方案架构

核心组件

文档摄取：使用LangChain文档加载器处理PDF
文本分块：实验不同分块策略(固定大小/递归)
向量嵌入：通过SageMaker部署的嵌入模型生成
向量存储：使用某中心S3向量服务存储向量及元数据
检索逻辑：基于向量相似度实现语义搜索
响应生成：结合检索上下文生成最终响应
评估体系：使用SageMaker托管MLflow进行性能评估

技术实现

1. 模型部署

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


# 部署嵌入模型
EMBEDDING_ENDPOINT_NAME = deploy_jumpstart_model(
    model_id="huggingface-textembedding-gte-qwen2-7b-instruct",
    instance_type="ml.g5.xlarge"
)

# 部署文本生成模型
TEXT_GENERATION_ENDPOINT_NAME = deploy_jumpstart_model(
    model_id="deepseek-llm-r1-distill-qwen-7b",
    instance_type="ml.g6.12xlarge"
)

2. S3向量存储配置

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


# 创建向量桶
s3vectors_client.create_vector_bucket(vectorBucketName=vectors_bucket_name)

# 创建向量索引(支持5000万向量/索引)
s3vectors_client.create_index(
    vectorBucketName=vectors_bucket_name,
    indexName="financial_docs",
    dimension=1024,
    distanceMetric="cosine"
)

3. 文档处理与向量化

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


for chunk in chunks:
    embedding = embeddings_model_endpoint.predict({'inputs': chunk["text"]})
    s3vectors_client.put_vectors(
        vectorBucketName=bucket_name,
        indexName=index_name,
        vectors=[{
            "key": str(uuid.uuid4()),
            "vector": embedding[0],
            "metadata": {"domain": "年报", "year": 2023}
        }]
    )

4. LangGraph实现RAG工作流

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18


# 定义检索函数
def retrieve(state: State):
    query_vector = embedding_model.predict(state["question"])
    results = s3vectors_client.query_vectors(
        vectorBucketName=bucket_name,
        indexName=index_name,
        vector=query_vector,
        topK=5
    )
    return {"context": results}

# 定义生成函数
def generate(state: State):
    response = text_generation_model.predict({
        "messages": format_prompt(state["context"]),
        "temperature": 0.2
    })
    return {"answer": response["choices"][0]["message"]["content"]}

性能评估与优化

使用MLflow跟踪关键指标：

回答正确性(通过Claude 3评估)
延迟性能
分块策略参数
嵌入模型版本

1
2
3
4
5
6
7
8


with mlflow.start_run():
    mlflow.evaluate(
        model=rag_pipeline,
        metrics=[answer_correctness(), latency()],
        evaluator_config={"col_mapping": {"inputs": "question"}}
    )
    mlflow.log_param("chunk_size", 512)
    mlflow.log_metric("correctness", 0.87)

S3向量核心优势

成本效益：比传统方案节省高达90%成本
弹性扩展：从GB到PB级无缝扩展
统一存储：向量与元数据共同存储
灵活查询：支持元数据过滤(MongoDB语法)

适用场景

医疗：可搜索的医学研究数据库
金融：历史交易数据分析
零售：商品目录增强
制造：技术文档管理
法律：合同文档检索

清理资源

完成实验后应及时删除：

SageMaker终端节点
S3向量桶和索引
MLflow跟踪服务器

该方案展示了如何利用某中心S3向量服务与SageMaker AI构建可扩展、经济高效的企业级RAG应用，同时通过MLflow实现全流程的实验跟踪和性能优化。