Prompt 与 RAG 系统设计要点
High Contrast
Dark Mode
Light Mode
Sepia
Forest
5 min read901 words

Prompt 与 RAG 系统设计要点

Prompt 写得烂,再好的模型也救不了你;RAG 建得乱,知识库就是垃圾场。作为 PM,你不需要亲自调 Prompt,但你必须能判断一个 Prompt 设计是否合理、一套 RAG 架构是否值得投入。


Prompt 结构的 PM 视角

graph TD A[业务需求] --> B{需要外部知识?} B -- 否 --> C[纯 Prompt 方案] B -- 是 --> D{知识量级} D -- 小且稳定 --> E[知识写入 System Prompt] D -- 大或动态 --> F[RAG 系统] C --> G[Prompt 结构设计] G --> G1[角色设定 Role] G --> G2[任务说明 Task] G --> G3[约束条件 Constraint] G --> G4[输出格式 Format] G --> G5[少样本示例 Few-shot] F --> H[知识库建设] H --> H1[文档切分策略] H --> H2[向量化与索引] H --> H3[检索召回设计] H --> H4[上下文注入] G5 --> I[效果评估] H4 --> I I --> J{达标?} J -- 否 --> K[迭代优化] J -- 是 --> L[上线] K --> G

Prompt 设计核心原则

PM 需要知道什么样的 Prompt 是结构化的

要素 作用 PM 判断标准
Role(角色) 锁定模型"身份",防止越界回答 是否明确限定了模型的专业边界
Task(任务) 清晰描述要做什么,一句话一个任务 能否用一句话概括任务目标
Constraint(约束) 禁止做什么、长度限制、语气要求 负向约束是否比正向描述更清晰
Format(格式) JSON / Markdown / 纯文本输出 下游系统能否直接消费输出
Few-shot(示例) 给 2-5 个好的例子比长篇说明有效 示例是否覆盖了典型失败场景
Context(上下文) RAG 检索结果或历史对话注入 上下文窗口是否会被撑爆

何时需要 RAG

不需要 RAG 的场景: - 任务完全基于通用知识(写文案、改语气、分类) - 知识量少于 10 个文档且不会更新 - 对延迟极度敏感(RAG 会增加 200-800ms)

必须上 RAG 的场景: - 知识库超过模型上下文窗口 - 知识需要实时更新(价格、政策、库存) - 需要溯源引用(法律、医疗、金融合规) - 用户问题高度个性化(用户历史、账户信息)


RAG 设计关键决策点

决策 选项 PM 关注的权衡
切分粒度 句子 / 段落 / 页面 太细:语义碎片;太粗:噪声多
检索策略 向量检索 / 关键词 / 混合 专有名词多 → 混合检索更稳
召回数量 Top-3 / Top-5 / Top-10 越多越贵且稀释相关性
重排序 有 / 无 Reranker 精度 +15%,延迟 +200ms
缓存 查询缓存 / 向量缓存 高频相似问题可省 60% 成本

Python 示例:RAG 系统设计评估器

"""
RAG 系统设计评估器
PM 视角:评估一套 RAG 方案的合理性与风险点
"""
from dataclasses import dataclass, field
from typing import List, Dict, Tuple
import json
@dataclass
class RAGDesignConfig:
"""RAG 设计配置"""
name: str
# 知识库配置
doc_count: int              # 文档数量
update_frequency: str       # 更新频率: daily/weekly/monthly/static
avg_doc_length: int         # 平均文档字数
# 切分策略
chunk_size: int             # 切分大小(字符数)
chunk_overlap: int          # 重叠字符数
# 检索配置
retrieval_type: str         # vector/keyword/hybrid
top_k: int                  # 召回数量
use_reranker: bool          # 是否使用重排序
# 业务配置
latency_requirement_ms: int # 延迟要求(毫秒)
requires_citation: bool     # 是否需要引用溯源
query_per_day: int          # 日均查询量
@dataclass
class EvaluationResult:
"""评估结果"""
config_name: str
dimension_scores: Dict[str, int] = field(default_factory=dict)
issues: List[str] = field(default_factory=list)
suggestions: List[str] = field(default_factory=list)
overall_score: int = 0
def evaluate_chunk_strategy(config: RAGDesignConfig) -> Tuple[int, List[str], List[str]]:
"""评估切分策略"""
score = 100
issues = []
suggestions = []
# 检查 chunk size 合理性
if config.chunk_size < 100:
score -= 30
issues.append("切分粒度过细(<100字),语义碎片化风险高")
suggestions.append("建议 chunk_size 调整为 300-600 字")
elif config.chunk_size > 2000:
score -= 20
issues.append("切分粒度过粗(>2000字),检索精度下降")
suggestions.append("建议 chunk_size 调整为 500-1000 字")
else:
suggestions.append(f"切分粒度 {config.chunk_size} 字,处于合理区间")
# 检查重叠比例
overlap_ratio = config.chunk_overlap / config.chunk_size
if overlap_ratio < 0.05:
score -= 15
issues.append("重叠率过低,跨块信息可能丢失")
elif overlap_ratio > 0.4:
score -= 10
issues.append("重叠率过高(>40%),存储成本增加且检索噪声多")
else:
suggestions.append(f"重叠率 {overlap_ratio:.0%},配置合理")
return score, issues, suggestions
def evaluate_retrieval_strategy(config: RAGDesignConfig) -> Tuple[int, List[str], List[str]]:
"""评估检索策略"""
score = 100
issues = []
suggestions = []
# 检索类型与场景匹配
if config.retrieval_type == "vector":
if config.doc_count < 100:
score -= 20
issues.append("文档量少(<100篇)时向量检索成本偏高,关键词检索可能更高效")
elif config.retrieval_type == "keyword":
score -= 15
issues.append("纯关键词检索对语义理解弱,推荐升级为混合检索")
suggestions.append("建议使用 hybrid 检索(向量 + BM25)")
elif config.retrieval_type == "hybrid":
suggestions.append("混合检索是当前最佳实践,兼顾语义与精确匹配")
# 召回数量评估
if config.top_k > 10:
score -= 20
issues.append(f"top_k={config.top_k} 过大,上下文注入会稀释相关性且增加 token 成本")
suggestions.append("建议 top_k 控制在 3-7 之间")
elif config.top_k < 3:
score -= 10
issues.append(f"top_k={config.top_k} 过小,召回率不足")
return score, issues, suggestions
def evaluate_performance(config: RAGDesignConfig) -> Tuple[int, List[str], List[str]]:
"""评估性能配置"""
score = 100
issues = []
suggestions = []
# 估算 RAG 延迟
base_latency = 300  # 向量检索基础延迟 ms
if config.use_reranker:
base_latency += 250
suggestions.append("使用了 Reranker,预计额外增加 200-300ms 延迟")
if base_latency > config.latency_requirement_ms * 0.6:
score -= 25
issues.append(
f"RAG 预估延迟 {base_latency}ms 可能超出延迟预算 "
f"(要求 {config.latency_requirement_ms}ms)"
)
suggestions.append("考虑引入查询缓存或降低 top_k")
# 高并发成本
if config.query_per_day > 100000:
suggestions.append(
f"日均 {config.query_per_day:,} 次查询,"
"强烈建议上缓存层,可降低 40-60% 向量检索成本"
)
return score, issues, suggestions
def evaluate_rag_design(config: RAGDesignConfig) -> EvaluationResult:
"""综合评估 RAG 设计方案"""
result = EvaluationResult(config_name=config.name)
# 分维度评估
chunk_score, chunk_issues, chunk_suggestions = evaluate_chunk_strategy(config)
retrieval_score, retrieval_issues, retrieval_suggestions = evaluate_retrieval_strategy(config)
perf_score, perf_issues, perf_suggestions = evaluate_performance(config)
result.dimension_scores = {
"切分策略": chunk_score,
"检索策略": retrieval_score,
"性能配置": perf_score,
}
result.issues = chunk_issues + retrieval_issues + perf_issues
result.suggestions = chunk_suggestions + retrieval_suggestions + perf_suggestions
# 综合评分(加权平均)
result.overall_score = int(
chunk_score * 0.35 + retrieval_score * 0.40 + perf_score * 0.25
)
return result
def print_report(result: EvaluationResult):
"""打印评估报告"""
print(f"\n{'='*55}")
print(f"  RAG 设计评估报告:{result.config_name}")
print(f"{'='*55}")
print(f"综合评分:{result.overall_score}/100")
print("\n各维度得分:")
for dim, score in result.dimension_scores.items():
bar = "█" * (score // 10) + "░" * (10 - score // 10)
print(f"  {dim:<10} {bar} {score}")
if result.issues:
print(f"\n发现 {len(result.issues)} 个问题:")
for i, issue in enumerate(result.issues, 1):
print(f"  [{i}] ⚠ {issue}")
print("\n优化建议:")
for suggestion in result.suggestions:
print(f"  → {suggestion}")
grade = "优秀" if result.overall_score >= 85 else \
"良好" if result.overall_score >= 70 else \
"需优化" if result.overall_score >= 55 else "风险较高"
print(f"\n整体判断:{grade}")
print(f"{'='*55}")
# ── 演示 ──────────────────────────────────────────────
if __name__ == "__main__":
# 方案 A:一个配置较差的设计
config_a = RAGDesignConfig(
name="方案A(初版设计)",
doc_count=5000,
update_frequency="daily",
avg_doc_length=2000,
chunk_size=80,          # 太细
chunk_overlap=5,        # 重叠太少
retrieval_type="keyword", # 纯关键词
top_k=15,               # 太多
use_reranker=False,
latency_requirement_ms=500,
requires_citation=True,
query_per_day=200000,
)
# 方案 B:优化后的设计
config_b = RAGDesignConfig(
name="方案B(优化版)",
doc_count=5000,
update_frequency="daily",
avg_doc_length=2000,
chunk_size=400,         # 合理
chunk_overlap=60,       # ~15% 重叠
retrieval_type="hybrid", # 混合检索
top_k=5,                # 合理
use_reranker=True,
latency_requirement_ms=800,
requires_citation=True,
query_per_day=200000,
)
for cfg in [config_a, config_b]:
result = evaluate_rag_design(cfg)
print_report(result)

本章 checklist

本章小结


下一节03-Workflow与Agent的产品设计 — 从确定性流程到自主 Agent,PM 如何判断边界并设计可降级的产品