引言

随着大语言模型(LLM)在 2023-2026 年的爆发式发展,如何让 AI 真正"落地"到企业生产环境,成为开发者最关注的话题。单纯的 LLM 就像一个博学但容易"胡说八道"的助手——它有知识截止日期、无法访问企业内部数据、还可能产生"幻觉"(Hallucination)。

为了解决这些问题,业界逐渐形成了 AI 应用开发的"三件套"

  1. Prompt Engineering(提示工程) — 让模型"听话"的艺术
  2. RAG(Retrieval-Augmented Generation,检索增强生成) — 给模型"外挂知识库"
  3. 向量数据库(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. 组织最终答案

【输出格式】
- 直接答案:<一句话总结>
- 详细说明:<分点说明>
- 参考依据:<引用文档编号>

【回答】
"""

六、进阶方向

掌握三件套后,可以继续深入:

  1. Advanced RAG
    • HyDE(假设文档嵌入)
    • GraphRAG(图结构增强)
    • Self-RAG(自反思检索)
    • Agentic RAG(智能体驱动的检索)
  1. 工具与框架
    • LangChain - 事实标准框架
    • LlamaIndex - 专为 RAG 设计
    • Dify - 国产可视化平台
    • Flowise - 拖拽式 AI 工作流
  1. 评估体系
    • 检索准确率(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 正是当下最确定的工程化方向之一

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐