下面对常见的几种“Query 改写”Prompt 模板进行中英文对照，并在注释中给出中文说明，帮助中国用户快速理解与使用。

---

根据调研，企业级 Query 改写模块需要覆盖多种常见场景，包括拼写纠错、中英混合、省略上下文、多义词扩展、专业术语替换等，从而在 RAG 应用和大模型问答中都能保持高召回和高精度 。为了验证您的改写功能，下面提供 5 个涵盖这些场景的典型测试 Query。

5 个验证用 Query

1. 拼写错误 + 中英混合

   iphon14 pro 价钱

   • 测试拼写纠错（“iphon”→“iphone”）和中英混合分词 。

2. 省略上下文的代词查询

   它昨年营收多少

   • 测试上下文补全，将“它”还原为前文实体（如公司名） 。

3. 中文行业术语替换

   联想笔电多少钱

   • 测试本地词典中“笔电”→“笔记本电脑”的同义词扩展 。

4. 多义词消歧

   apple 更新日志

   • 测试“apple”歧义（品牌 vs. 水果）澄清与实体链接 。

5. 复杂英语专业查询

   best python web framework 性能 比较

   • 测试英文专有名词识别＋中文关键词补全，以及模板改写（补全“有哪些？”） 。

以上 5 个 Query 覆盖了企业级改写中的核心场景，适合作为流水线测试用例。

---

常见提示词工程 

---

1. 简洁零样本重写（Zero-Shot）

// 英文原版
Rewrite the following query to improve its performance and recall:
// 中文翻译
将以下查询重写，以提高其检索性能和召回率：
// 用法说明
//   直接提示模型对用户原始查询进行端到端重写，适合零样本场景。 :contentReference[oaicite:0]{index=0}

---

2. 前缀式调用（Prefix Prompt）

# 英文原版
prompt_calibrated = "query reformulation: " + user_query
# 中文翻译
prompt_calibrated = "query 改写： " + user_query
# 用法说明
#   在 Hugging Face 上的 query-reformulation 模型中常见，直接在查询前加固定前缀。 :contentReference[oaicite:1]{index=1}

import os
import openai
import json# 环境变量中设置 OPENAI_API_KEY
openai.api_key = os.getenv("OPENAI_API_KEY")# ① 待改写的查询列表
queries = ["iphon14 pro 价钱","它昨年营收多少","联想笔电多少钱","apple 更新日志","best python web framework 性能 比较"
]# ② 将列表拼接为多行 Prompt
#    使用 join + 列表推导，每一项前面加 "- "
queries_block = "\n".join(f"- {q}" for q in queries)# ③ 构造完整 Prompt，示例格式引导模型输出 JSON 数组
prompt_str1="重写以下查询，使其更完整、更易检索："
prompt_str="Rewrite the following query to make it more complete and easier to retrieve:"
prompt = f"""
{prompt_str}\n
{queries_block}请**只**返回一个 JSON 数组，格式示例：
["改写1", "改写2", "改写3", "改写4", "改写5"]"""if client is None:ensure_client()
# ④ 调用 OpenAI ChatCompletion 接口
response = client.chat.completions.create(model="gpt-4o",  # 或 "gpt-4-turbo-preview" 支持 response_format=json_objectmessages=[{"role": "system", "content": "你是一个只输出 JSON 数组的助手，不要添加多余文字。"},{"role": "user",   "content": prompt}],temperature=0.0,max_tokens=256
)# ⑤ 解析 JSON 数组并打印
result = json.loads(response.choices[0].message.content)
# print(result)# 输出改写结果
print(f"提示词：{prompt_str}\n")
print("原始查询：")
print(queries)print("\n改写后的查询：")
print(result)

---

3. 提示增强（Hint-Based）

// 英文原版
Here are some hints that you might consider when rewriting the query:
{Hints: a list of rewrite rules, e.g., "Avoid unnecessary JOINs", "Use explicit CTEs"}
Answer:
{Candidate rewrite}
Explanation:
{Which hints were applied}// 中文翻译
下面是改写时可参考的提示：
{提示列表，例如：“避免不必要的 JOIN 操作”、“使用显式的 CTE”}
回答：
{候选重写语句}
解释：
{说明使用了哪些提示}// 用法说明
//   在生成式改写中注入具体规则提示，结合提示工程提高改写质量。 :contentReference[oaicite:3]{index=3}

import os
import openai
import json# 环境变量中设置 OPENAI_API_KEY
openai.api_key = os.getenv("OPENAI_API_KEY")# ① 待改写的查询列表
queries = ["iphon14 pro 价钱","它昨年营收多少","联想笔电多少钱","apple 更新日志","best python web framework 性能 比较"
]# ② 将列表拼接为多行 Prompt
#    使用 join + 列表推导，每一项前面加 "- "
queries_block = "\n".join(f"- {q}" for q in queries)# ③ 构造完整 Prompt，示例格式引导模型输出 JSON 数组
prompt_str1="重写以下查询，使其更完整、更易检索："
prompt_str="""Rewrite the following query to make it more complete and easier to retrieve:Here are some hints that you might consider when rewriting the query:
{Hints: a list of rewrite rules, e.g., "Avoid unnecessary JOINs", "Use explicit CTEs"}
Answer:
{Candidate rewrite}
Explanation:
{Which hints were applied}"""
prompt = f"""
{prompt_str}\n
{queries_block}请**只**返回一个 JSON 数组，格式示例：
["改写1", "改写2", "改写3", "改写4", "改写5"]"""if client is None:ensure_client()
# ④ 调用 OpenAI ChatCompletion 接口
response = client.chat.completions.create(model="gpt-4o",  # 或 "gpt-4-turbo-preview" 支持 response_format=json_objectmessages=[{"role": "system", "content": "你是一个只输出 JSON 数组的助手，不要添加多余文字。"},{"role": "user",   "content": prompt}],temperature=0.0,max_tokens=256
)# ⑤ 解析 JSON 数组并打印
result = json.loads(response.choices[0].message.content)
# print(result)# 输出改写结果
print(f"提示词：{prompt_str}\n")
print("原始查询：")
print(queries)print("\n改写后的查询：")
print(result)

---

4. 结合 RAG 的生成式改写

// 英文原版
We use a Generative AI model to rewrite the question… enabling a rewrite-retrieve-read pipeline.
// 中文翻译
我们使用生成式 AI 模型对用户问题进行改写，形成“改写-检索-阅读”流水线。
// 用法说明
//   在 RAG 架构中，先用 LLM 生成改写候选，再并行检索并融合答案，可显著提升覆盖率与一致性。 :contentReference[oaicite:5]{index=5}

 

import os
import json
import numpy as np
from openai import OpenAI, OpenAIError# —— 初始化客户端 ——
# client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))QUERIES = ["iphon14 pro 价钱","它昨年营收多少","联想笔电多少钱","apple 更新日志","best python web framework 性能 比较"
]def generate_candidates(query: str, n: int = 3) -> list[str]:prompt = (f"重写以下查询，使其更完整、更易检索，每条生成{n}个候选：\n"f"1. {query}\n\n只返回 JSON 数组格式：\n""[[\"改写1\",\"改写2\",\"改写3\"]]")resp = client.chat.completions.create(model="gpt-4o",messages=[{"role": "system", "content": "你只输出 JSON 数组，不要多余说明。"},{"role": "user",   "content": prompt}],temperature=0.0,max_tokens=200,)return json.loads(resp.choices[0].message.content)[0]def get_embedding(text: str, model: str = "text-embedding-3-small") -> np.ndarray:resp = client.embeddings.create(input=[text], model=model)return np.array(resp.data[0].embedding)def main():output = {}for q in QUERIES:# 1. 生成候选candidates = generate_candidates(q, n=3)# 2. 获取原查询与候选的嵌入（NumPy 数组）orig_emb = get_embedding(q)                     # shape: (D,)cand_embs = np.stack([get_embedding(c) for c in candidates])  # shape: (3, D)# 3. 向量化计算余弦相似度#    dots: (3,) — 原向量与每个候选的点积dots = cand_embs.dot(orig_emb)#    norms: 标量与向量范数orig_norm = np.linalg.norm(orig_emb)cand_norms = np.linalg.norm(cand_embs, axis=1)scores = dots / (orig_norm * cand_norms + 1e-10)# 4. 排序并保存idxs = np.argsort(scores)[::-1]ranked = [{"rewrite": candidates[i], "score": float(scores[i])} for i in idxs]output[q] = ranked# 输出最终结果print(json.dumps(output, ensure_ascii=False, indent=2))if __name__ == "__main__":try:main()except OpenAIError as e:print("调用 OpenAI API 出错：", e)

---

5. 多样化并行改写

// 英文原版
Generate 3 paraphrases of the query, then pick the best by semantic similarity.
// 中文翻译
生成该查询的3条同义改写，然后按语义相似度选取最优。
// 用法说明
//   结合多候选并行检索与重排序，平衡召回和精度。 :contentReference[oaicite:7]{index=7}

 

import os
import json
import math
from typing import List, Dict, Any
from concurrent.futures import ThreadPoolExecutor, as_completed
from openai import OpenAI
from requests.exceptions import RequestException# —— 初始化客户端 ——
# client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))# —— 原始查询列表 ——
QUERIES = ["iphon14 pro 价钱","它昨年营收多少","联想笔电多少钱","apple 更新日志","best python web framework 性能 比较"
]# —— 重试装饰器 ——
def retry_on_exception(max_retries=3):def decorator(func):def wrapper(*args, **kwargs):for attempt in range(1, max_retries + 1):try:return func(*args, **kwargs)except RequestException as e:if attempt == max_retries:raise# should not reach herereturn wrapperreturn decorator# —— 生成改写候选 ——
@retry_on_exception()
def generate_candidates(queries: List[str], n: int = 3) -> List[List[str]]:prompt = "重写以下查询，使其更完整、更易检索，每条生成 {} 个候选：\n".format(n)prompt += "\n".join(f"{i+1}. {q}" for i, q in enumerate(queries))prompt += "\n\n请只返回 JSON 数组，格式示例：\n" \"[[\"改写1a\",\"改写1b\",\"改写1c\"], ...]"resp = client.chat.completions.create(model="gpt-4o",messages=[{"role": "system", "content": "你是只输出 JSON 数组的助手，不要多余说明。"},{"role": "user",   "content": prompt}],temperature=0.0,max_tokens=500,timeout=15.0)return json.loads(resp.choices[0].message.content)# —— 批量获取 Embedding ——
@retry_on_exception()
def batch_embeddings(texts: List[str], model="text-embedding-3-small") -> List[List[float]]:resp = client.embeddings.create(input=texts,model=model,timeout=15.0)return [item.embedding for item in resp.data]# —— 余弦相似度 ——
def cosine_sim(v1: List[float], v2: List[float]) -> float:dot = sum(a*b for a, b in zip(v1, v2))norm1 = math.sqrt(sum(a*a for a in v1))norm2 = math.sqrt(sum(b*b for b in v2))return dot / (norm1 * norm2) if norm1 and norm2 else 0.0# —— 排序任务 ——
def rank_for_query(orig: str, candidates: List[str],orig_emb: List[float], cand_embs: List[List[float]]) -> List[Dict[str, Any]]:scored = [{"rewrite": cand, "score": cosine_sim(orig_emb, emb)}for cand, emb in zip(candidates, cand_embs)]return sorted(scored, key=lambda x: x["score"], reverse=True)def main():# 1. 生成候选candidates = generate_candidates(QUERIES, n=3)# 2. 统一准备所有需要嵌入的文本all_texts = []for q, cands in zip(QUERIES, candidates):all_texts.append(q)all_texts.extend(cands)# 3. 批量请求嵌入embeddings = batch_embeddings(all_texts)# 划分回原始与候选idx = 0orig_embs, cand_embs_list = [], []for cands in candidates:orig_embs.append(embeddings[idx]); idx += 1group = embeddings[idx: idx + len(cands)]; idx += len(cands)cand_embs_list.append(group)# 4. 并行排序output: Dict[str, Any] = {}with ThreadPoolExecutor() as executor:futures = {executor.submit(rank_for_query, q, cands, oe, ce):q for q, cands, oe, ce in zip(QUERIES, candidates, orig_embs, cand_embs_list)}for fut in as_completed(futures):q = futures[fut]output[q] = fut.result()# 5. 输出print(json.dumps(output, ensure_ascii=False, indent=2))if __name__ == "__main__":main()