下面對常見的幾種“Query 改寫”Prompt 模板進行中英文對照,并在注釋中給出中文說明,幫助中國用戶快速理解與使用。
根據調研,企業級 Query 改寫模塊需要覆蓋多種常見場景,包括拼寫糾錯、中英混合、省略上下文、多義詞擴展、專業術語替換等,從而在 RAG 應用和大模型問答中都能保持高召回和高精度 。為了驗證您的改寫功能,下面提供 5 個涵蓋這些場景的典型測試 Query。
5 個驗證用 Query
-
拼寫錯誤 + 中英混合
iphon14 pro 價錢
-
測試拼寫糾錯(“iphon”→“iphone”)和中英混合分詞 。
-
-
省略上下文的代詞查詢
它昨年營收多少
-
測試上下文補全,將“它”還原為前文實體(如公司名) 。
-
-
中文行業術語替換
聯想筆電多少錢
-
測試本地詞典中“筆電”→“筆記本電腦”的同義詞擴展 。
-
-
多義詞消歧
apple 更新日志
-
測試“apple”歧義(品牌 vs. 水果)澄清與實體鏈接 。
-
-
復雜英語專業查詢
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()
?