RAG + Prompt Engineering + 向量数据库:AI 应用开发三件套
引言
随着大语言模型(LLM)在 2023-2026 年的爆发式发展,如何让 AI 真正"落地"到企业生产环境,成为开发者最关注的话题。单纯的 LLM 就像一个博学但容易"胡说八道"的助手——它有知识截止日期、无法访问企业内部数据、还可能产生"幻觉"(Hallucination)。
为了解决这些问题,业界逐渐形成了 AI 应用开发的"三件套":
- Prompt Engineering(提示工程) — 让模型"听话"的艺术
- RAG(Retrieval-Augmented Generation,检索增强生成) — 给模型"外挂知识库"
- 向量数据库(Vector Database) — 让机器"理解语义"的基础设施
本文将系统讲解这三者的技术原理、协同关系,并通过完整的 Python 代码示例,带你从零搭建一个企业级 RAG 应用。
一、Prompt Engineering:与 AI 对话的"正确姿势"
1.1 什么是 Prompt Engineering?
Prompt Engineering 是通过精心设计输入文本(Prompt),引导 LLM 输出更准确、更可控结果的技术。同一个模型,好的 Prompt 和差的 Prompt 效果差距可以达到 10 倍以上。
1.2 Prompt 的核心结构
一个高质量的 Prompt 通常包含以下要素:
Python
# 一个结构化的 Prompt 模板
prompt_template = """
【角色设定】
你是一名资深的{role},拥有10年以上的行业经验。
【任务说明】
{task_description}
【上下文信息】
{context}
【输入数据】
{user_input}
【输出要求】
1. 输出格式:{output_format}
2. 风格要求:{style}
3. 注意事项:{constraints}
"""
1.3 常用 Prompt 技巧
技巧 1:Few-Shot Learning(少样本学习)
Python
from openai import OpenAI
client = OpenAI(api_key="your-api-key")
# Few-shot 示例:通过示例让模型学会模式
few_shot_prompt = """
将以下句子按情感分类为"积极"或"消极":
句子:今天天气真好,心情很棒! -> 积极
句子:这顿饭太难吃了,浪费钱。 -> 消极
句子:这个产品性价比超高,强烈推荐! -> 积极
句子:服务态度极差,再也不来了。 ->
请只输出分类结果。
"""
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": few_shot_prompt}],
temperature=0
)
print(response.choices[0].message.content)
技巧 2:Chain-of-Thought(思维链)
Python
cot_prompt = """
问题:一个商店3天卖了120个苹果。如果每天的销量增加20%,
按照这个增长率,第5天能卖多少个苹果?
让我们一步步思考:
"""
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": cot_prompt}],
temperature=0
)
print(response.choices[0].message.content)
技巧 3:Role Prompting(角色设定)
Python
role_prompt = """
你现在是一位有20年经验的Python后端架构师,专注于高并发系统设计。
请用专业、严谨的语言回答以下问题,必要时给出代码示例。
问题:如何设计一个支持百万QPS的短链生成系统?
"""
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": role_prompt}],
temperature=0.7
)
print(response.choices[0].message.content)
1.4 Prompt Engineering 的最佳实践
- ✅ 明确指令:告诉模型"做什么"和"不做什么"
- ✅ 结构化输入:使用分隔符(如
"""、<>)区分不同部分 - ✅ 指定输出格式:JSON、Markdown、表格等
- ✅ 提供示例:Few-Shot 显著提升准确率
- ❌ 避免歧义:模糊的 Prompt 会得到模糊的结果
- ❌ 避免过长:Prompt 不是越长越好,要精炼
二、向量数据库:让机器"理解"语义的基石
2.1 为什么需要向量数据库?
传统数据库基于精确匹配(关键词、ID),而 AI 应用需要基于语义相似度搜索。
例如:
- 用户搜索:"怎么退款?"
- 知识库中原文:"申请退款的流程如下..."
- 关键词匹配失败,但语义完全一致 → 这就需要向量检索
2.2 Embedding:文本转向量的过程
Python
from openai import OpenAI
import numpy as np
client = OpenAI(api_key="your-api-key")
def get_embedding(text: str, model="text-embedding-3-small"):
"""将文本转换为向量"""
response = client.embeddings.create(
input=text,
model=model
)
return response.data[0].embedding
# 示例:把句子变成向量
text1 = "今天天气真好"
text2 = "今天阳光明媚"
text3 = "Python 是一门编程语言"
vec1 = get_embedding(text1)
vec2 = get_embedding(text2)
vec3 = get_embedding(text3)
print(f"向量维度:{len(vec1)}") # 1536 维
# 计算余弦相似度
def cosine_similarity(a, b):
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
print(f"'{text1}' vs '{text2}': {cosine_similarity(vec1, vec2):.4f}")
# 输出:0.78 左右(语义相近)
print(f"'{text1}' vs '{text3}': {cosine_similarity(vec1, vec3):.4f}")
# 输出:0.30 左右(语义不相关)
2.3 主流向量数据库对比
| 数据库 | 特点 | 适用场景 |
|---|---|---|
| Chroma | 轻量、Python 原生、零配置 | 原型开发、小规模应用 |
| Milvus | 分布式、高性能、功能丰富 | 大规模生产环境 |
| Pinecone | 全托管、SaaS 服务 | 不想运维的公司 |
| Weaviate | 开源、内置向量化和检索 | 中等规模应用 |
| Qdrant | Rust 编写、性能优异 | 实时性要求高的场景 |
| pgvector | PostgreSQL 插件 | 已有 PG 栈、想统一技术栈 |
2.4 使用 Chroma 实战
Python
import chromadb
from chromadb.utils import embedding_functions
# 1. 初始化 Chroma 客户端
client = chromadb.PersistentClient(path="./chroma_db")
# 2. 使用 OpenAI 的 Embedding 模型
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
api_key="your-api-key",
model_name="text-embedding-3-small"
)
# 3. 创建集合(Collection)
collection = client.get_or_create_collection(
name="knowledge_base",
embedding_function=openai_ef,
metadata={"hnsw:space": "cosine"}
)
# 4. 添加文档
documents = [
"公司支持7天无理由退换货,需保持商品完好。",
"退款将在1-3个工作日内原路返回。",
"会员等级分为:青铜、白银、黄金、钻石。",
"钻石会员享受全年免运费特权。",
"客服电话:400-888-8888,工作时间 9:00-21:00。"
]
metadatas = [
{"category": "售后", "source": "退换货政策"},
{"category": "售后", "source": "退款说明"},
{"category": "会员", "source": "会员体系"},
{"category": "会员", "source": "会员权益"},
{"category": "客服", "source": "联系方式"}
]
ids = ["doc1", "doc2", "doc3", "doc4", "doc5"]
collection.add(
documents=documents,
metadatas=metadatas,
ids=ids
)
# 5. 语义检索
results = collection.query(
query_texts=["我想退货怎么办?"],
n_results=3
)
print("查询结果:")
for i, (doc, meta, distance) in enumerate(zip(
results['documents'][0],
results['metadatas'][0],
results['distances'][0]
)):
print(f"\n[{i+1}] 相似度: {1-distance:.4f}")
print(f"分类: {meta['category']} | 来源: {meta['source']}")
print(f"内容: {doc}")
运行结果示例:
Plain Text
查询结果:
[1] 相似度: 0.8523
分类: 售后 | 来源: 退换货政策
内容: 公司支持7天无理由退换货,需保持商品完好。
[2] 相似度: 0.7891
分类: 售后 | 来源: 退款说明
内容: 退款将在1-3个工作日内原路返回。
[3] 相似度: 0.6543
分类: 客服 | 来源: 联系方式
内容: 客服电话:400-888-8888,工作时间 9:00-21:00。
可以看到,即使没有"退货"这个精确关键词,"我想退货怎么办?" 也能精准匹配到相关文档,这就是向量检索的威力。
三、RAG:让 LLM 拥有"专属知识库"
3.1 RAG 的核心思想
RAG(Retrieval-Augmented Generation)的核心流程:
Plain Text
用户提问 → 向量化 → 检索相关文档 → 拼接 Prompt → LLM 生成答案
它解决了 LLM 的三大痛点:
- 🚫 知识过时 → 通过外部知识库实时更新
- 🚫 幻觉问题 → 基于真实文档回答
- 🚫 私有数据 → 让 LLM 访问企业内部知识
3.2 RAG 的完整架构
Plain Text
┌─────────────┐
│ 文档加载层 │ PDF / Word / Markdown / Web
└──────┬──────┘
↓
┌─────────────┐
│ 文本分割层 │ 切片 (Chunking) - 关键步骤
└──────┬──────┘
↓
┌─────────────┐
│ 向量化层 │ Embedding Model
└──────┬──────┘
↓
┌─────────────┐
│ 向量数据库 │ 存储 + 检索
└──────┬──────┘
↓
┌─────────────┐
│ Prompt 拼接 │ Query + Context → LLM
└──────┬──────┘
↓
┌─────────────┐
│ 答案生成 │ LLM 输出
└─────────────┘
3.3 完整 RAG 系统实现
Python
from openai import OpenAI
import chromadb
from chromadb.utils import embedding_functions
from typing import List, Dict
import tiktoken
class RAGSystem:
"""一个生产级的 RAG 系统"""
def __init__(self, api_key: str, collection_name: str = "rag_kb"):
self.client = OpenAI(api_key=api_key)
self.chroma_client = chromadb.PersistentClient(path="./chroma_db")
self.embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
api_key=api_key,
model_name="text-embedding-3-small"
)
self.collection = self.chroma_client.get_or_create_collection(
name=collection_name,
embedding_function=self.embedding_fn
)
self.encoding = tiktoken.encoding_for_model("gpt-4")
def split_text(self, text: str, chunk_size: int = 500, overlap: int = 50) -> List[str]:
"""文本切片:保证语义完整性的滑动窗口"""
tokens = self.encoding.encode(text)
chunks = []
start = 0
while start < len(tokens):
end = min(start + chunk_size, len(tokens))
chunk_tokens = tokens[start:end]
chunk_text = self.encoding.decode(chunk_tokens)
chunks.append(chunk_text)
start += chunk_size - overlap
return chunks
def add_documents(self, documents: List[Dict]):
"""
添加文档到知识库
documents: [{"content": "...", "metadata": {...}}, ...]
"""
all_chunks = []
all_metadatas = []
all_ids = []
for idx, doc in enumerate(documents):
chunks = self.split_text(doc["content"])
for chunk_idx, chunk in enumerate(chunks):
all_chunks.append(chunk)
all_metadatas.append({
**doc.get("metadata", {}),
"doc_id": idx,
"chunk_id": chunk_idx
})
all_ids.append(f"doc_{idx}_chunk_{chunk_idx}")
self.collection.add(
documents=all_chunks,
metadatas=all_metadatas,
ids=all_ids
)
print(f"✅ 成功添加 {len(documents)} 个文档,共 {len(all_chunks)} 个切片")
def retrieve(self, query: str, top_k: int = 3) -> List[Dict]:
"""检索相关文档"""
results = self.collection.query(
query_texts=[query],
n_results=top_k
)
retrieved_docs = []
for i in range(len(results['documents'][0])):
retrieved_docs.append({
"content": results['documents'][0][i],
"metadata": results['metadatas'][0][i],
"distance": results['distances'][0][i]
})
return retrieved_docs
def build_prompt(self, query: str, context_docs: List[Dict]) -> str:
"""构建 RAG Prompt"""
context = "\n\n".join([
f"【参考文档 {i+1}】\n{doc['content']}"
for i, doc in enumerate(context_docs)
])
prompt = f"""你是一名专业的企业知识助手。请根据以下参考文档回答用户问题。
【参考文档】
{context}
【用户问题】
{query}
【回答要求】
1. 仅基于参考文档回答,不要编造信息
2. 如果参考文档不包含答案,请明确告知"知识库中未找到相关信息"
3. 回答要简洁、专业、有条理
4. 必要时引用文档来源
【回答】"""
return prompt
def query(self, question: str, top_k: int = 3, temperature: float = 0) -> str:
"""完整的 RAG 查询流程"""
# 1. 检索
print(f"🔍 正在检索: {question}")
docs = self.retrieve(question, top_k=top_k)
# 2. 构建 Prompt
prompt = self.build_prompt(question, docs)
# 3. 调用 LLM
response = self.client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "你是一名专业的企业知识助手。"},
{"role": "user", "content": prompt}
],
temperature=temperature
)
answer = response.choices[0].message.content
return answer, docs
# ===== 使用示例 =====
if __name__ == "__main__":
# 初始化
rag = RAGSystem(api_key="your-api-key")
# 添加知识库
kb = [
{
"content": "公司提供5种岗位:前端工程师、后端工程师、算法工程师、产品经理、UI设计师。"
"前端工程师要求掌握 HTML、CSS、JavaScript、React 或 Vue。"
"后端工程师要求熟悉 Python/Java、数据库、REST API。",
"metadata": {"category": "招聘", "department": "技术部"}
},
{
"content": "员工每年享有10天带薪年假、5天病假。"
"工作时间为周一至周五 9:00-18:00,午休 12:00-13:30。"
"公司提供五险一金、年度体检、生日礼物。",
"metadata": {"category": "HR", "department": "人事部"}
},
{
"content": "差旅报销标准:一线城市住宿不超过800元/晚,"
"二线城市不超过600元/晚,三线城市不超过400元/晚。"
"机票需提前3天预订,经济舱标准。",
"metadata": {"category": "财务", "department": "财务部"}
}
]
rag.add_documents(kb)
# 提问
questions = [
"前端工程师需要什么技能?",
"年假有几天?",
"去北京出差能住多少钱的酒店?"
]
for q in questions:
print(f"\n{'='*60}")
print(f"❓ 问题: {q}")
answer, refs = rag.query(q, top_k=2)
print(f"\n💡 回答:\n{answer}")
print(f"\n📚 参考文档:")
for i, ref in enumerate(refs):
print(f" [{i+1}] 相似度={1-ref['distance']:.3f} | {ref['content'][:80]}...")
四、三件套的协同关系
Prompt Engineering、RAG、向量数据库并不是孤立的技术,而是一个完整的协作链条:
Plain Text
┌──────────────────────────────────────────────────────┐
│ 用户输入 Query │
└────────────────────┬─────────────────────────────────┘
↓
┌────────────────────────┐
│ 1. 向量化(向量数据库) │ ← Embedding
└────────────┬───────────┘
↓
┌────────────────────────┐
│ 2. 语义检索(向量数据库)│ ← 找最相关的 Top-K
└────────────┬───────────┘
↓
┌────────────────────────┐
│ 3. Prompt 拼接 │ ← Prompt Engineering
│ Query + Context │
└────────────┬───────────┘
↓
┌────────────────────────┐
│ 4. LLM 生成答案 │ ← 大模型推理
└────────────────────────┘
一个都不能少:
- 没有 向量数据库 → RAG 无法做语义检索,退化为关键词匹配
- 没有 RAG → LLM 只能靠"记忆"回答,无法访问私有/最新数据
- 没有 Prompt Engineering → 检索结果给到 LLM,但 LLM 不会用,效果大打折扣
五、生产环境的优化建议
5.1 检索质量优化
Python
# 1. 混合检索:结合关键词 + 向量
def hybrid_search(query, collection, alpha=0.7):
"""alpha 控制向量检索的权重"""
# BM25 关键词检索
keyword_results = bm25_search(query, ...)
# 向量检索
vector_results = collection.query(query_texts=[query], n_results=10)
# 加权融合
final_results = merge_results(keyword_results, vector_results, alpha)
return final_results
# 2. Re-ranking:用更强的模型对初筛结果重排
def rerank(query, candidates, top_k=3):
"""用 Cross-Encoder 精排"""
pairs = [(query, doc) for doc in candidates]
scores = cross_encoder.predict(pairs)
ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
return ranked[:top_k]
5.2 切片策略优化
Python
# 不同的切片策略适用于不同场景
strategies = {
"fixed_size": "固定长度切片 - 简单但可能切断语义",
"semantic": "语义切片 - 按段落/章节切分,更准确",
"sliding_window": "滑动窗口 - 有 overlap,保证上下文连贯",
"structure_aware": "结构感知 - 按标题/代码块/表格分别处理"
}
# 推荐:递归切片 + 元数据
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
separators=["\n\n", "\n", "。", ",", " ", ""]
)
5.3 Prompt 模板优化
Python
# 高级 Prompt 模板:包含 few-shot + 思维链
advanced_prompt = """
你是一名企业知识库助手,擅长从多个文档片段中提取关键信息回答问题。
【参考文档】
{context}
【历史对话】(如有)
{chat_history}
【用户问题】
{question}
【思考步骤】
1. 分析问题需要哪些信息
2. 从参考文档中提取相关信息
3. 如果信息不足,告知用户并建议
4. 组织最终答案
【输出格式】
- 直接答案:<一句话总结>
- 详细说明:<分点说明>
- 参考依据:<引用文档编号>
【回答】
"""
六、进阶方向
掌握三件套后,可以继续深入:
- Advanced RAG:
- HyDE(假设文档嵌入)
- GraphRAG(图结构增强)
- Self-RAG(自反思检索)
- Agentic RAG(智能体驱动的检索)
- 工具与框架:
- LangChain - 事实标准框架
- LlamaIndex - 专为 RAG 设计
- Dify - 国产可视化平台
- Flowise - 拖拽式 AI 工作流
- 评估体系:
- 检索准确率(Recall@K, MRR)
- 答案质量(Faithfulness, Relevance)
- 端到端评估(RAGAS 框架)
结语
RAG + Prompt Engineering + 向量数据库 构成了现代 AI 应用开发的"三件套"。它们三者相辅相成,缺一不可:
- 🎯 Prompt Engineering 决定了 LLM 能否"听懂人话"
- 🗄️ 向量数据库 决定了系统能否"找到正确答案"
- 🔗 RAG 把两者串联起来,让 LLM 拥有"专属知识库"
对于想要进入 AI 应用开发领域的工程师来说,深入理解这三项技术的工作原理和协作机制,是从"会调 API"到"能交付生产级 AI 应用"的关键跨越。
💡 建议学习路径:先掌握 Prompt Engineering → 学习向量数据库基本原理 → 动手实现一个完整 RAG 项目 → 引入 LangChain/LlamaIndex 框架 → 学习 Advanced RAG 优化技巧。
希望本文能帮你建立完整的技术认知。AI 应用的未来已来,而 RAG 正是当下最确定的工程化方向之一。
更多推荐


所有评论(0)