RAG 中的 Prompt 工程

Prompt 的设计直接决定了 RAG 系统回答的质量。本章介绍 RAG 场景下的 Prompt 最佳实践。

RAG Prompt 核心结构

基础 Prompt 模板

"""
RAG Prompt 模板库
"""
# 模板一：基础问答
BASIC_QA_PROMPT = """你是一个专业的知识助手。请根据以下参考资料回答用户的问题。
## 规则
1. 只基于参考资料回答，不要使用你自己的知识
2. 如果参考资料不足以回答问题，请坦诚说明"根据现有资料，我无法回答这个问题"
3. 在回答中使用 [1][2] 等标注引用来源
4. 回答要准确、简洁、有条理
## 参考资料
{context}
## 用户问题
{question}
## 回答"""
# 模板二：带思考过程的问答
CHAIN_OF_THOUGHT_PROMPT = """你是一个专业的知识助手。请根据参考资料回答用户问题。
## 规则
1. 先分析问题需要哪些信息
2. 从参考资料中找到相关内容
3. 组织回答并标注引用来源 [1][2]
4. 如果资料不足，明确说明
## 参考资料
{context}
## 用户问题
{question}
## 思考过程
让我分析一下这个问题：
## 回答"""
# 模板三：多轮对话
CONVERSATIONAL_PROMPT = """你是一个专业的知识助手。请基于对话历史和参考资料回答用户的最新问题。
## 规则
1. 结合对话历史理解用户意图
2. 只基于参考资料回答
3. 标注引用来源 [1][2]
4. 如果是跟进问题，要联系之前的回答
## 对话历史
{chat_history}
## 参考资料
{context}
## 用户最新问题
{question}
## 回答"""

上下文窗口管理

当检索到的文档超过上下文窗口时，需要智能管理：

"""
上下文窗口管理
确保 Prompt 不超过模型的 Token 限制
"""
import tiktoken
class ContextManager:
"""上下文窗口管理器"""
def __init__(
self,
model: str = "gpt-4o-mini",
max_context_tokens: int = 4000,
reserved_for_answer: int = 1000
):
self.encoder = tiktoken.encoding_for_model(model)
self.max_context_tokens = max_context_tokens
self.reserved = reserved_for_answer
def count_tokens(self, text: str) -> int:
"""计算 Token 数"""
return len(self.encoder.encode(text))
def fit_documents(
self,
documents: list[dict],
system_prompt: str,
question: str
) -> list[dict]:
"""
选择能放入上下文窗口的文档
按相关性排序，依次加入直到达到限制
"""
# 计算已用 Token
used_tokens = (
self.count_tokens(system_prompt) +
self.count_tokens(question) +
self.reserved
)
available_tokens = self.max_context_tokens - used_tokens
selected = []
current_tokens = 0
for doc in documents:
doc_tokens = self.count_tokens(doc["content"])
if current_tokens + doc_tokens <= available_tokens:
selected.append(doc)
current_tokens += doc_tokens
else:
# 尝试截断最后一个文档
remaining = available_tokens - current_tokens
if remaining > 100:  # 至少保留100个token
truncated = self._truncate(
doc["content"], remaining
)
doc_copy = dict(doc)
doc_copy["content"] = truncated
doc_copy["truncated"] = True
selected.append(doc_copy)
break
return selected
def _truncate(self, text: str, max_tokens: int) -> str:
"""按 Token 限制截断文本"""
tokens = self.encoder.encode(text)
if len(tokens) <= max_tokens:
return text
truncated_tokens = tokens[:max_tokens]
return self.encoder.decode(truncated_tokens) + "..."
# 使用
manager = ContextManager(max_context_tokens=4000)
documents = [
{"content": "文档1的内容...", "score": 0.95},
{"content": "文档2的内容...", "score": 0.90},
{"content": "文档3的内容...", "score": 0.85},
]
selected = manager.fit_documents(
documents=documents,
system_prompt=BASIC_QA_PROMPT,
question="什么是 RAG？"
)
print(f"选中 {len(selected)}/{len(documents)} 个文档")

引用与溯源

让 AI 的回答可以追溯到原始文档：

"""
引用标注系统
让每个回答都有据可查
"""
def format_context_with_citations(documents: list[dict]) -> str:
"""格式化带编号的参考资料"""
formatted = []
for i, doc in enumerate(documents, 1):
title = doc.get("title", f"文档 {i}")
source = doc.get("source", "未知来源")
content = doc["content"]
formatted.append(
f"[{i}] 《{title}》(来源: {source})\n{content}"
)
return "\n\n---\n\n".join(formatted)
def extract_citations(answer: str, documents: list[dict]) -> dict:
"""从回答中提取引用信息"""
import re
# 找到所有引用标记 [1], [2] 等
citation_pattern = re.compile(r"\[(\d+)\]")
cited_indices = set()
for match in citation_pattern.finditer(answer):
idx = int(match.group(1)) - 1
if 0 <= idx < len(documents):
cited_indices.add(idx)
# 构建引用列表
citations = []
for idx in sorted(cited_indices):
doc = documents[idx]
citations.append({
"index": idx + 1,
"title": doc.get("title", f"文档 {idx + 1}"),
"source": doc.get("source", ""),
"url": doc.get("url", "")
})
return {
"answer": answer,
"citations": citations,
"num_citations": len(citations)
}

防幻觉策略

"""
防幻觉 Prompt 策略
减少 LLM 编造信息的风险
"""
# 策略一：严格限制回答范围
STRICT_PROMPT = """你是一个严格基于事实的助手。
## 绝对规则（必须遵守）
- 只使用"参考资料"中的信息回答
- 不要添加任何参考资料中没有的信息
- 不确定的内容用"根据现有资料无法确定"表述
- 每个事实性陈述必须标注来源 [编号]
## 参考资料
{context}
## 问题
{question}"""
# 策略二：分步验证
VERIFY_PROMPT = """请按以下步骤回答问题：
1. 【信息提取】从参考资料中找出与问题相关的所有信息
2. 【相关性判断】判断这些信息是否足以回答问题
3. 【生成回答】基于提取的信息生成回答
4. 【自我检查】检查回答中是否有参考资料之外的信息，如有请删除
## 参考资料
{context}
## 问题
{question}
## 步骤1 - 相关信息："""
def verify_answer_grounding(
answer: str,
documents: list[dict]
) -> dict:
"""
验证回答是否有文档支撑（后处理检查）
使用 LLM 判断回答中的每个声明是否有文档依据
"""
from openai import OpenAI
client = OpenAI()
context = "\n".join([d["content"] for d in documents])
verification_prompt = f"""请检查以下回答中的每个事实性声明是否能在参考资料中找到依据。
## 参考资料
{context}
## 回答
{answer}
## 检查结果
对每个事实性声明，标注：
- SUPPORTED: 有资料支持
- NOT_SUPPORTED: 无资料支持
- PARTIAL: 部分有支持
请列出所有声明及其检查结果："""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": verification_prompt}],
temperature=0
)
return {
"verification": response.choices[0].message.content,
"original_answer": answer
}

多轮对话支持

"""
多轮对话中的 RAG
需要处理代词消解和上下文传递
"""
class ConversationalRAG:
"""支持多轮对话的 RAG"""
def __init__(self, retriever, max_history: int = 5):
self.retriever = retriever
self.max_history = max_history
self.history = []
def chat(self, user_message: str) -> str:
"""处理用户消息"""
# Step 1: 使用历史上下文改写查询
search_query = self._contextualize_query(user_message)
# Step 2: 检索
documents = self.retriever.retrieve(search_query, top_k=3)
# Step 3: 构建 Prompt
chat_history = self._format_history()
context = format_context_with_citations(documents)
prompt = CONVERSATIONAL_PROMPT.format(
chat_history=chat_history,
context=context,
question=user_message
)
# Step 4: 生成回答
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
answer = response.choices[0].message.content
# 保存对话历史
self.history.append({"role": "user", "content": user_message})
self.history.append({"role": "assistant", "content": answer})
# 限制历史长度
if len(self.history) > self.max_history * 2:
self.history = self.history[-self.max_history * 2:]
return answer
def _contextualize_query(self, query: str) -> str:
"""结合历史上下文改写查询"""
if not self.history:
return query
from openai import OpenAI
client = OpenAI()
history_text = self._format_history()
prompt = f"""根据对话历史，将用户的最新问题改写为一个独立的、完整的查询。
如果最新问题已经是独立的，直接返回原问题。
对话历史:
{history_text}
最新问题: {query}
改写后的独立查询:"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
return response.choices[0].message.content.strip()
def _format_history(self) -> str:
"""格式化对话历史"""
lines = []
for msg in self.history[-self.max_history * 2:]:
role = "用户" if msg["role"] == "user" else "助手"
lines.append(f"{role}: {msg['content']}")
return "\n".join(lines)

本章小结

• RAG Prompt 由系统角色、参考资料、用户问题和输出要求四部分组成
• 上下文窗口管理确保 Prompt 不超过 Token 限制
• 引用标注让回答可溯源、可验证
• 防幻觉策略通过严格限制和后处理验证减少编造
• 多轮对话需要查询改写来保持上下文连贯

下一章：我们将学习高级 RAG 技术，包括 Self-RAG 和 GraphRAG。