每年的双十一、618 大促期间，电商平台的客服系统面临着前所未有的挑战。凌晨零点秒杀开启的瞬间，并发咨询量可能在几分钟内从几百飙升至数万，传统的人工客服团队根本无法承接。更糟糕的是，促销规则复杂多变——满减叠加、限时折扣、会员专享价、新用户礼包——客服人员需要实时记忆数十种活动细节，响应速度和准确率都难以保证。
本文将从一个真实的电商场景出发，手把手教你搭建一套完整的 RAG（Retrieval-Augmented Generation）系统，实现基于商品知识库和促销规则的智能问答。整个流程涵盖：文档解析、文本向量化、向量检索、大模型生成四大核心环节，并深度集成 HolySheep AI 的 Embedding 和 Chat API，国内直连延迟低于 50ms，汇率仅 ¥7.3=$1，远低于官方定价。
一、RAG 系统整体架构
在开始编码之前，我们需要先明确整个系统的数据流向：
- 文档解析层：将 PDF、Word、Excel 等格式的促销文档、商品手册解析为结构化文本
- 向量化层：调用 HolySheep Embedding API，将文本切分为语义向量，存入向量数据库
- 检索层：用户提问时，先将问题向量化，从向量库中召回最相关的文档片段
- 生成层：将召回的上下文片段 + 用户问题，组装成 Prompt，调用 HolySheep Chat API 生成最终回答
二、环境准备与依赖安装
pip install openai tiktoken faiss-cpu pypdf python-docx langchain pydantic
我们选择 Faiss 作为向量数据库（支持 CPU 快速检索），LangChain 负责文本分割和 Prompt 组装。核心依赖只需要两个 API 客户端，均通过 HolySheep AI 提供，无需魔法上网。
import os
from openai import OpenAI
HolySheep API 配置
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["HOLYSHEEP_BASE_URL"] = "https://api.holysheep.ai/v1"
初始化客户端
client = OpenAI(
api_key=os.environ["HOLYSHEEP_API_KEY"],
base_url=os.environ["HOLYSHEEP_BASE_URL"]
)
三、文档解析：从多格式到纯文本
电商的促销文档通常散落在各个部门的共享盘里，格式五花八门——运营团队用 Word 撰写活动方案，技术团队用 Excel 维护商品 SKU，财务部门用 PDF 输出对账单。我们先写一个统一的解析器：
import json
from pathlib import Path
from typing import List, Union
from pypdf import PdfReader
from docx import Document
def parse_document(file_path: str) -> List[str]:
"""
统一解析多种文档格式，返回文本段落列表
支持: .pdf, .docx, .txt
"""
path = Path(file_path)
suffix = path.suffix.lower()
if suffix == ".pdf":
return _parse_pdf(file_path)
elif suffix == ".docx":
return _parse_docx(file_path)
elif suffix == ".txt":
return _parse_txt(file_path)
else:
raise ValueError(f"暂不支持的文件格式: {suffix}")
def _parse_pdf(file_path: str) -> List[str]:
"""解析 PDF，提取每页文本"""
reader = PdfReader(file_path)
paragraphs = []
for page in reader.pages:
text = page.extract_text()
if text.strip():
paragraphs.append(text.strip())
return paragraphs
def _parse_docx(file_path: str) -> List[str]:
"""解析 Word 文档，提取段落"""
doc = Document(file_path)
return [para.text.strip() for para in doc.paragraphs if para.text.strip()]
def _parse_txt(file_path: str) -> List[str]:
"""解析纯文本，按行分割"""
with open(file_path, "r", encoding="utf-8") as f:
return [line.strip() for line in f if line.strip()]
使用示例
if __name__ == "__main__":
docs = parse_document("promotion_rules_2024.pdf")
print(f"解析完成，共 {len(docs)} 个段落")
四、文本分块与向量化
文档解析完成后，我们需要将长文本切分为适合检索的小块（Chunk）。分块策略直接影响检索效果：太大则引入噪声，太小则丢失上下文。LangChain 提供了多种文本分割器，这里使用 RecursiveCharacterTextSplitter，按段落、句子递归切分：
from langchain.text_splitter import RecursiveCharacterTextSplitter
def chunk_text(paragraphs: List[str], chunk_size: int = 500, chunk_overlap: int = 50) -> List[str]:
"""
文本分块
Args:
paragraphs: 原始段落列表
chunk_size: 每个块的目标字符数
chunk_overlap: 块之间的重叠字符数（保留上下文连续性）
"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
separators=["\n\n", "\n", "。", "！", "？", " "]
)
full_text = "\n\n".join(paragraphs)
chunks = text_splitter.split_text(full_text)
return chunks
调用 HolySheep Embedding API 获取向量
def get_embeddings(texts: List[str], model: str = "text-embedding-3-small") -> List[List[float]]:
"""
批量获取文本向量（使用 HolySheep API）
价格参考（2026年）：
- text-embedding-3-small: $0.02/1K tokens
- text-embedding-3-large: $0.13/1K tokens
"""
response = client.embeddings.create(
model=model,
input=texts
)
return [item.embedding for item in response.data]
五、构建向量索引与检索
向量数据库的选择上，生产环境推荐 Milvus、Pinecone 等云服务，这里为了演示方便使用 Faiss（Facebook 开源的向量索引库，支持毫秒级召回）：
import faiss
import numpy as np
class VectorStore:
def __init__(self, dimension: int = 1536):
self.dimension = dimension
self.index = faiss.IndexFlatL2(dimension) # L2 距离索引
self.chunks = [] # 存储原始文本块
def add_vectors(self, vectors: List[List[float]], texts: List[str]):
"""批量添加向量和对应文本"""
vectors_array = np.array(vectors).astype('float32')
self.index.add(vectors_array)
self.chunks.extend(texts)
def search(self, query_vector: List[float], top_k: int = 3) -> List[dict]:
"""
向量相似度检索
Returns:
List[dict]: 包含 text 和 score 的结果列表
"""
query_array = np.array([query_vector]).astype('float32')
distances, indices = self.index.search(query_array, top_k)
results = []
for dist, idx in zip(distances[0], indices[0]):
if idx < len(self.chunks):
results.append({
"text": self.chunks[idx],
"score": float(dist)
})
return results
完整示例：构建促销知识库索引
def build_promotion_index(documents: List[str]):
"""构建促销规则知识库"""
# 1. 分块
chunks = chunk_text(documents)
print(f"分块完成: {len(chunks)} 个文本块")
# 2. 向量化
embeddings = get_embeddings(chunks)
print(f"向量化完成: {len(embeddings)} 个向量")
# 3. 构建索引
dimension = len(embeddings[0])
store = VectorStore(dimension=dimension)
store.add_vectors(embeddings, chunks)
print(f"索引构建完成: {store.index.ntotal} 条记录")
return store
六、检索增强生成：组装 Prompt 并调用 LLM
检索环节完成后，我们需要将召回的文档片段作为上下文，组装成结构化的 Prompt。好的 Prompt 设计能显著提升回答质量，尤其是处理促销规则这类需要精确数字的场景：
def retrieve_and_generate(query: str, vector_store: VectorStore, model: str = "gpt-4o-mini"):
"""
RAG 核心流程：检索 + 生成
Args:
query: 用户问题
vector_store: 向量数据库
model: 使用的 LLM 模型
Returns:
str: 生成的答案
"""
# 1. 将用户问题向量化
query_embedding = get_embeddings([query])[0]
# 2. 检索相关文档
relevant_docs = vector_store.search(query_embedding, top_k=3)
# 3. 组装上下文
context = "\n\n".join([doc["text"] for doc in relevant_docs])
# 4. 构建 Prompt（Few-shot 设计，提升格式一致性）
system_prompt = """你是一个专业的电商促销客服。请根据【上下文】中的促销规则回答用户问题。
要求：
1. 只使用【上下文】中提供的信息，不要编造
2. 如果信息不足以回答，明确说明"暂无相关规则"
3. 涉及金额、折扣比例的数字必须精确
4. 回答要简洁有条理，突出关键信息
"""
user_prompt = f"""【上下文】
{context}
【用户问题】
{query}
【回答】"""
# 5. 调用 HolySheep Chat API 生成回答
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.3, # 促销规则回答需要低随机性
max_tokens=500
)
return response.choices[0].message.content
使用示例
if __name__ == "__main__":
# 模拟从 PDF 解析的促销文档
sample_docs = [
"双十一满减规则：全场商品满300减50、满500减100、满1000减250，可与店铺优惠券叠加使用",
"限时折扣专区：11月11日 0:00-2:00，部分商品5折起售，每个用户限购2件",
"会员专享：PLUS会员可额外领取满200减30专属券，与满减活动叠加，上不封顶"
]
# 构建索引
index = build_promotion_index(sample_docs)
# 测试问答
question = "我是 PLUS 会员，双十一当天买了600元的商品，能优惠多少？"
answer = retrieve_and_generate(question, index)
print(f"问题: {question}\n回答: {answer}")
七、性能优化：应对大促并发高峰
双十一零点洪峰对系统的并发处理能力是严峻考验。以下是几项关键优化：
- 连接池复用：避免每次请求都创建新的 HTTP 连接
- 向量缓存：热门商品文档的向量预计算并缓存，减少重复 Embedding 调用
- 异步批处理：批量向量化时使用异步 API，吞吐量提升 5-10 倍
- 限流熔断：接入 HolySheep API 时设置合理的 QPS 限制，防止突发流量打垮下游
from functools import lru_cache
from concurrent.futures import ThreadPoolExecutor
import asyncio
class AsyncRAGPipeline:
"""异步 RAG 流水线，支持高并发"""
def __init__(self, vector_store: VectorStore):
self.vector_store = vector_store
self.embedding_cache = {}
@lru_cache(maxsize=10000)
def cached_embedding(self, text_hash: str, text: str) -> List[float]:
"""向量缓存：基于文本哈希去重"""
return get_embeddings([text])[0]
async def batch_retrieve(self, queries: List[str]) -> List[List[dict]]:
"""批量异步检索"""
loop = asyncio.get_event_loop()
with ThreadPoolExecutor(max_workers=20) as executor:
tasks = [
loop.run_in_executor(executor, self._sync_search, q)
for q in queries
]
results = await asyncio.gather(*tasks)
return results
def _sync_search(self, query: str) -> List[dict]:
embedding = self.cached_embedding(hash(query), query)
return self.vector_store.search(embedding, top_k=3)
async def batch_generate(self, queries: List[str], model: str = "gpt-4o-mini") -> List[str]:
"""批量异步生成"""
contexts = await self.batch_retrieve(queries)
async def generate_one(query: str, context: List[dict]) -> str:
prompt = f"【上下文】\n{' '.join([c['text'] for c in context])}\n\n【问题】{query}"
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
max_tokens=300
)
return response.choices[0].message.content
tasks = [
generate_one(q, c)
for q, c in zip(queries, contexts)
]
return await asyncio.gather(*tasks)
八、生产部署架构建议
单个 Python 进程难以支撑数万 QPS 的并发咨询，建议采用以下微服务架构：
- API 网关层：Nginx / Kong 做流量分发和限流
- Embedding 服务：独立部署，支持水平扩展，HolySheep API 调用频率可设为 500 RPM
- 向量数据库：Milvus 集群 + Redis 缓存热点向量
- LLM 生成服务：异步消息队列（Kafka）削峰，多实例部署
九、常见报错排查
1. API Key 配置错误导致 401 Unauthorized
错误信息：AuthenticationError: Incorrect API key provided
排查步骤：
- 确认环境变量
HOLYSHEEP_API_KEY已正确设置（格式：sk-xxxx） - 检查 base_url 是否为
https://api.holysheep.ai/v1，勿使用其他三方代理地址 - 登录 HolySheep 控制台 确认 Key 是否已激活
2. 向量维度不匹配导致 Faiss 索引报错
错误信息：RuntimeError: Error adding vectors: inconsistent dimension
排查步骤：
- 确认使用的 Embedding 模型维度：
text-embedding-3-small为 1536 维，text-embedding-3-large为 3072 维 - VectorStore 初始化时的 dimension 必须与模型输出一致
- 检查分批添加向量时，模型版本是否被意外切换
3. 模型响应超时或限流 429
错误信息：RateLimitError: Rate limit reached for model
排查步骤：
- 大促期间 API 调用量激增，建议提前联系 HolyShe