各位同仁，各位技术爱好者：
大家好！
今天，我们齐聚一堂，共同探讨一个在现代信息检索与问答系统中日益凸显的关键技术——“Query Transformation Circuit”，即“查询转换回路”。在人工智能，特别是大语言模型（LLM）和向量数据库技术飞速发展的当下，我们常常面临一个挑战：用户提出的原始问题，即便表达自然，也可能因为其固有的模糊性、表达习惯的差异，或与后端知识库的“语义鸿沟”，而难以直接从向量数据库中召回最精准、最全面的结果。
传统的向量搜索，尽管在语义匹配方面表现出色，但它本质上是一种“所见即所得”的匹配。如果用户的查询没有很好地捕捉到底层文档的“精髓”，或者使用了不同的词汇、更宽泛的概念，那么即使最先进的向量模型也可能力不从心。试想，用户提问“如何设置用户身份验证？”而我们的文档中更多描述的是“OAuth 2.0 实现”、“JWT 令牌集成”或“单点登录配置”，虽然语义相关，但直接匹配可能效果不佳。
解决这一问题的核心思想，便是今天的主题：利用一个智能的、迭代的“查询转换回路”，不断优化原始问题，直至向量数据库能够召回真正精准且满足用户需求的结果。这就像一位经验丰富的图书馆员，在听到一个模糊的请求后，会通过提问、联想、扩展关键词，甚至尝试不同的分类目录，最终帮助你找到那本你真正需要的书。
1. 原始问题：向量搜索的局限性与挑战
在深入探讨查询转换回路之前，我们有必要先理解为何需要它。向量数据库和嵌入技术无疑是信息检索领域的重大突破，它将文本、图像等非结构化数据转化为高维向量，并通过计算向量间的距离（如余弦相似度）来衡量它们的语义相关性。然而，这种看似完美的匹配机制，在实际应用中依然面临诸多挑战：
- 语义鸿沟与词汇不匹配（Lexical Gap & Semantic Gap）：
- 同义异词：用户可能使用“启动认证”而文档中使用“初始化身份验证”。
- 上下位词：用户查询“水果”，但文档具体描述的是“苹果”或“香蕉”。
- 领域特定术语：用户使用通用语，而文档包含大量专业术语。
- 模糊性：一个词在不同语境下有不同含义。
- 查询的复杂性与多意图：
- 用户查询可能包含多个子意图，例如“如何设置用户登录并保护数据库连接安全？”。
- 长查询或复合查询往往难以一次性映射到单一的、高相关性的向量空间区域。
- 长尾查询与稀疏数据：
- 对于非常具体、罕见的查询（长尾查询），可能缺乏足够多的相似样本进行有效匹配。
- 如果知识库中相关文档内容稀疏，即便有潜在相关性，也难以被召回。
- 用户表达习惯与系统期望不符：
- 用户倾向于使用口语化、不完整的句子。
- 系统（尤其是在没有LLM辅助时）可能期望更结构化、更清晰的表达。
- 缺乏上下文信息：
- 原始查询是孤立的，缺乏用户历史、偏好或当前会话的上下文，导致召回结果不够个性化或相关。
这些局限性使得单纯依赖一次性的向量搜索往往无法满足用户对精准、全面信息的需求。因此，我们需要一个更加智能、动态的机制来“理解”并“优化”用户的查询。
2. 查询转换回路：核心概念与工作原理
“查询转换回路”（Query Transformation Circuit）正是为了应对上述挑战而设计的一种高级检索范式。其核心理念是：将用户的原始查询视为一个初始状态，通过一系列智能的“转换节点”（Transformation Node）对其进行迭代优化、扩展或重构，每次转换后都尝试向向量数据库发起检索，并对结果进行评估，直到找到满足预设条件的精准结果，或者达到最大迭代次数。
这个“回路”不仅仅是简单的查询重写，它是一个包含反馈机制的智能系统，其工作原理可以概括为以下几个关键组件：
- 初始查询（Initial Query）：用户输入的原始问题或需求。
- 转换节点（Transformation Node / 循环节点）：系统的核心智能单元，负责对当前查询（或查询集）进行各种形式的修改，生成一个或多个新的、更优化的查询变体。这通常由大语言模型（LLM）驱动。
- 向量数据库（Vector Database）：存储着大量文档内容的向量嵌入，是查询的目标。
- 检索与聚合（Retrieval & Aggregation）：使用转换后的查询向向量数据库发起检索，并收集所有相关结果。
- 评估机制（Evaluation Mechanism）：对检索到的结果进行量化评估，判断其与用户原始意图的匹配程度、相关性、多样性等。这是回路能否“收敛”的关键。
- 迭代逻辑与终止条件（Iteration Logic & Stopping Criteria）：根据评估结果，决定是继续进行下一轮转换，还是停止回路并返回当前最佳结果。
工作流示意图：
| 阶段 | 描述 | 关键输出 |
|---|---|---|
| 1. 初始输入 | 接收用户原始查询。 | original_query |
| 2. 循环开始 | 设定最大迭代次数，初始化当前查询集、最佳结果集、历史记录。 | current_query_set = {original_query} |
| 3. 查询执行 | 对 current_query_set 中的每个查询变体： |
retrieved_docs_for_variant |
| a. 生成其向量嵌入。 | ||
b. 向向量数据库发起检索，获取 top_k 个文档。 |
||
| 4. 结果评估 | 对所有检索到的文档，结合 original_query 进行评估： |
evaluation_scores (如整体相关性、置信度) |
| a. 判断文档与原始查询的匹配度。 | best_overall_relevance |
|
| b. 识别当前最佳结果集。 | best_results_so_far |
|
| 5. 终止判断 | 检查是否满足终止条件： | continue_iteration (布尔值) |
| a. 达到预设的相关性阈值？ | ||
| b. 达到最大迭代次数？ | ||
| c. 新生成的查询未能显著改善结果？ | ||
| 6. 转换生成 | 如果不满足终止条件： | new_query_variants |
| a. 基于当前最佳查询变体或所有变体，通过“转换节点”生成新的查询变体（扩展、重写、分解等）。 | current_query_set = current_query_set + new_query_variants |
|
b. 将新变体加入 current_query_set。 |
||
| 7. 循环迭代 | 返回第3步，开始下一轮查询执行与评估。 | |
| 8. 结果输出 | 循环终止后，返回最终的最佳结果集和/或历史记录。 | final_results, iteration_history |
3. 组件详解与实现策略
3.1 转换节点（Transformation Node）：回路的智能核心
转换节点是查询转换回路的大脑，它负责根据当前状态（原始查询、已尝试的查询、检索到的结果、会话上下文等）智能地生成新的查询变体。其核心任务是弥合用户意图与向量数据库内容之间的语义或词汇鸿沟。
常见的转换类型：
-
查询扩展（Query Expansion）：
- 目的：增加相关词汇、同义词、上下位词，以扩大召回范围。
- 示例：将“用户认证”扩展为“用户登录”、“身份验证”、“权限管理”、“OAuth”、“JWT”。
- 实现：
- 基于规则/词典：利用同义词词典、领域本体、人工定义的规则。
- 基于统计：从大规模语料中学习词语共现关系。
- 基于LLM：最灵活强大的方式。通过Prompt Engineering，让LLM理解查询意图，并生成多种相关的扩展词汇或短语。
-
查询重写/改写（Query Rewriting/Rephrasing）：
- 目的：在不改变原始意图的前提下，改变查询的表述方式、语法结构或用词，以匹配不同文档的行文风格。
- 示例：将“如何设置用户身份验证？”改写为“用户登录配置步骤”、“身份验证机制实施指南”。
- 实现：
- 基于规则：简单的替换、句式转换。
- 基于LLM：给定原始查询和期望，让LLM生成多个语义等价但表达不同的查询。
-
查询分解（Query Decomposition）：
- 目的：将一个复杂的、包含多意图的查询分解成多个更简单、更聚焦的子查询，分别进行检索，然后聚合结果。
- 示例：将“如何设置用户登录并保护数据库连接安全？”分解为：“如何设置用户登录？”和“如何保护数据库连接安全？”。
- 实现：
- 基于规则/关键词：识别连接词（“和”、“并”），分割查询。
- 基于LLM：让LLM识别并分解查询中的独立意图。
-
查询具体化/澄清（Query Specification/Clarification）：
- 目的：当查询过于模糊时，尝试添加更多细节，使其更具体。如果系统是交互式的，甚至可以生成澄清性问题。
- 示例：对于“如何保护数据？”，LLM可以生成“保护哪种类型的数据？（如个人数据、敏感业务数据）”、“在哪种环境中保护？（如云端、本地）”等。
- 实现：
- 基于知识图谱/本体：识别查询中的实体，并从知识图谱中获取其属性来补充查询。
- 基于LLM：利用LLM的常识和推理能力，推断可能的缺失信息并进行补充。
-
上下文注入（Contextualization）：
- 目的：将用户历史查询、会话状态、用户画像、领域知识等外部上下文信息融入到查询中，提高相关性。
- 示例：用户前一个问题是“OAuth实现”，下一个问题是“如何处理令牌过期？”，系统应将“OAuth”作为上下文注入到第二个查询中。
- 实现：
- 将上下文信息作为LLM Prompt的一部分。
- 在生成查询向量时，结合上下文向量。
LLM-based 转换节点的实现：
LLM在生成查询变体方面展现出无与伦比的灵活性和智能性。关键在于精心设计Prompt。
import json
import random
import numpy as np # 用于模拟嵌入
# --- 模拟 OpenAI Client ---
class MockOpenAIClient:
def chat(self):
return self
def completions(self):
return self
def create(self, model, messages=None, prompt=None, max_tokens=None, response_format=None):
if messages:
input_text = messages[-1]['content']
else:
input_text = prompt
response_content = ""
# 模拟查询扩展
if "generate 3-5 expanded variations" in input_text:
if "user authentication" in input_text.lower():
response_content = json.dumps([
"user login setup",
"identity management configuration",
"implementing authentication systems",
"secure user access setup",
"auth mechanism installation",
"OAuth 2.0 integration",
"JWT token security"
])
elif "user login setup" in input_text.lower():
response_content = json.dumps([
"how to configure login process",
"setting up user accounts",
"user session management",
"login form best practices"
])
elif "database connection security" in input_text.lower():
response_content = json.dumps([
"secure database access",
"database encryption methods",
"protecting data in transit to database",
"SQL injection prevention"
])
else:
base_query = input_text.split("Original Query: "")[1].split(""")[0]
response_content = json.dumps([
f"{base_query} details",
f"guide for {base_query}",
f"advanced {base_query}",
f"troubleshooting {base_query}"
])
# 模拟查询重写
elif "rephrase it in 2-3 different ways" in input_text:
if "user authentication" in input_text.lower():
response_content = json.dumps([
"how do I configure user logins?",
"setting up identity verification for users",
"steps for enabling user security features",
"guide to user access control"
])
elif "user login setup" in input_text.lower():
response_content = json.dumps([
"how to enable user sign-in",
"configuring system access",
"guide to user account setup",
"creating user accounts and login"
])
elif "database connection security" in input_text.lower():
response_content = json.dumps([
"how to secure database links",
"protecting database communication",
"best practices for secure database connectivity"
])
else:
base_query = input_text.split("Original Query: "")[1].split(""")[0]
response_content = json.dumps([
f"alternative phrase for {base_query}",
f"another way to say {base_query}",
f"reword {base_query}"
])
# 模拟查询分解 (简单场景)
elif "decompose the complex query" in input_text:
if "user login and database connection security" in input_text.lower():
response_content = json.dumps([
"how to set up user login",
"how to protect database connections"
])
else:
base_query = input_text.split("Original Query: "")[1].split(""")[0]
response_content = json.dumps([base_query]) # 无法分解时返回原查询
# 模拟结果评估
elif "assess the relevance of the provided search results" in input_text:
overall_relevance_score = random.randint(4, 9)
confidence_score = random.randint(5, 9)
doc_relevance = []
query_keywords = set(input_text.split("Original Query: "")[1].split(""")[0].lower().split())
docs_section = input_text.split("Search Results:")[1]
doc_parts = docs_section.split("n---n")
for part in doc_parts:
if "Document " in part:
doc_id_line = [line for line in part.split('n') if 'ID:' in line][0]
doc_id = doc_id_line.split("ID: ")[1].split(')')[0].strip()
doc_content = part.split(':n', 1)[1].strip()
doc_score = 0
justification = []
content_lower = doc_content.lower()
if "oauth" in content_lower and any(k in content_lower for k in ["auth", "authentication", "login"]):
doc_score = 5
justification.append("Mentions OAuth, highly relevant to authentication.")
elif "jwt" in content_lower and any(k in content_lower for k in ["auth", "authentication", "login"]):
doc_score = 5
justification.append("Mentions JWT, highly relevant to authentication.")
elif "flask-security" in content_lower and any(k in content_lower for k in ["login", "authentication", "user"]):
doc_score = 5
justification.append("Mentions Flask-Security, directly related to user login/auth.")
elif "password" in content_lower and any(k in content_lower for k in ["login", "authentication", "user"]):
doc_score = 4
justification.append("Covers password hashing, which is part of authentication.")
elif "database" in content_lower and "security" in content_lower and any(k in content_lower for k in ["connection", "protect"]):
doc_score = 5
justification.append("Directly addresses database connection security.")
elif "ssl" in content_lower and "database" in content_lower:
doc_score = 4
justification.append("Mentions SSL for database, relevant to security.")
elif any(k in content_lower for k in query_keywords if len(k) > 2): # Avoid very short common words
doc_score = min(4, sum(1 for k in query_keywords if k in content_lower) + 1)
justification.append("Contains several keywords from the query.")
else:
doc_score = random.randint(0, 2)
justification.append("Limited direct relevance to the query.")
doc_relevance.append({"id": doc_id, "score": doc_score, "justification": " ".join(justification)})
if doc_relevance:
avg_doc_score = sum(d['score'] for d in doc_relevance) / len(doc_relevance)
overall_relevance_score = int(min(10, max(0, avg_doc_score * 2)))
# If there's at least one highly relevant doc, bump up overall score
if any(d['score'] >= 4 for d in doc_relevance) and overall_relevance_score < 7:
overall_relevance_score = 7
elif not any(d['score'] >= 4 for d in doc_relevance) and overall_relevance_score > 5:
overall_relevance_score = random.randint(3,5) # Penalize if no good docs
else:
overall_relevance_score = 0
confidence_score = 0
response_content = json.dumps({
"document_relevance": doc_relevance,
"overall_relevance_score": overall_relevance_score,
"confidence_score": confidence_score,
"overall_summary": f"The results contain documents related to the query with an overall relevance score of {overall_relevance_score}."
})
else:
response_content = json.dumps({"error": "Unknown prompt type"})
class MockChoice:
def __init__(self, content):
self.message = type('obj', (object,), {'content': content})()
class MockResponse:
def __init__(self, content):
self.choices = [MockChoice(content)]
return MockResponse(response_content)
class LLMTransformationNode:
def __init__(self, client):
self.client = client
self.model = "gpt-4o" # 实际使用时会调用真实的LLM模型
def _call_llm(self, prompt, response_format="json_object"):
messages = [{"role": "user", "content": prompt}]
try:
if response_format == "json_object":
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
response_format={"type": "json_object"}
)
else:
response = self.client.chat.completions.create(
model=self.model,
messages=messages
)
return json.loads(response.choices[0].message.content)
except Exception as e:
print(f"Error calling LLM for transformation: {e}")
return {}
def expand_query(self, query, context=None):
prompt = f"""Given the following user query and optional context, generate 3-5 expanded variations of the query that cover related concepts, synonyms, and slightly broader or narrower interpretations. Focus on improving recall for a vector database search.
Original Query: "{query}"
{f"Context: {context}" if context else ""}
Output should be a JSON array of strings."""
result = self._call_llm(prompt)
return set(result) if isinstance(result, list) else set()
def rephrase_query(self, query, context=None):
prompt = f"""Given the following user query and optional context, rephrase it in 2-3 different ways that maintain the original intent but use different vocabulary or sentence structures. This is to help a vector database find more relevant documents.
Original Query: "{query}"
{f"Context: {context}" if context else ""}
Output should be a JSON array of strings."""
result = self._call_llm(prompt)
return set(result) if isinstance(result, list) else set()
def decompose_query(self, query, context=None):
prompt = f"""You are an expert at breaking down complex user queries. If the query contains multiple distinct intents, decompose it into 2-3 simpler, atomic sub-queries. If the query is already simple, return it as a single element in the array.
Original Query: "{query}"
{f"Context: {context}" if context else ""}
Output should be a JSON array of strings."""
result = self._call_llm(prompt)
return set(result) if isinstance(result, list) else set()
# --- 模拟嵌入模型 ---
class MockEmbeddingsModel:
def encode(self, text):
# 在实际场景中，这将调用一个真实的嵌入模型，例如SentenceTransformers
# 为了演示，我们生成一个基于文本内容的简单哈希向量
if not text:
return np.zeros(100) # 返回一个零向量或抛出错误
# 简化为每个字符的ASCII值之和作为向量的一部分
hash_val = sum(ord(c) for c in text[:100]) % 1000000 # 避免过大的数值
# 创建一个简单的100维向量，其中大部分为0，少数位置有值
vector = np.zeros(100)
np.random.seed(hash_val % (2**32 - 1)) # 使用哈希值作为随机种子，确保同一文本得到相同向量
indices = np.random.choice(100, size=min(10, 100), replace=False) # 随机选择10个位置
vector[indices] = np.random.rand(min(10, 100)) * 2 - 1 # 填充随机值
return vector
3.2 向量数据库与检索
向量数据库是存储和检索高维向量嵌入的系统。它的效率和准确性直接影响整个回路的性能。
主要功能：
- 存储：将文档内容通过嵌入模型转换为向量，并存储这些向量及其原始文档的元数据。
- 索引：为了高效检索，向量数据库会构建各种索引结构（如HNSW、IVF_FLAT等）。
- 搜索：根据查询向量，快速找到与之最相似的
top_k个文档向量。
检索方法：
- 纯语义搜索：仅基于向量相似度进行最近邻搜索（KNN/ANN）。
- 混合搜索（Hybrid Search）：结合向量搜索（语义）和关键词搜索（词法，如BM25），以兼顾召回率和精确性。这对于处理长尾查询或精确匹配特定关键词的场景非常重要。
# --- 模拟向量数据库 ---
class MockVectorDB:
def __init__(self, embeddings_model):
self.documents = []
self.embeddings = []
self.embeddings_model = embeddings_model
self.doc_id_map = {} # 存储id到文档内容的映射
def add_document(self, doc_id, content):
if doc_id in self.doc_id_map:
print(f"Document ID {doc_id} already exists, skipping.")
return
embedding = self.embeddings_model.encode(content)
self.documents.append({"id": doc_id, "content": content, "embedding": embedding})
self.embeddings.append(embedding)
self.doc_id_map[doc_id] = {"content": content, "embedding": embedding}
def search(self, query_embedding, top_k=5):
if not self.embeddings:
return []
# 简单的余弦相似度计算（为了演示，实际会使用更高效的库或ANN算法）
similarities = []
for i, doc_embedding in enumerate(self.embeddings):
# 避免除以零
norm_query = np.linalg.norm(query_embedding)
norm_doc = np.linalg.norm(doc_embedding)
if norm_query == 0 or norm_doc == 0:
similarity = 0.0
else:
similarity = np.dot(query_embedding, doc_embedding) / (norm_query * norm_doc)
similarities.append((similarity, i))
# 按相似度降序排序
similarities.sort(key=lambda x: x[0], reverse=True)
results = []
for sim, idx in similarities[:top_k]:
doc = self.documents[idx]
results.append({"id": doc["id"], "content": doc["content"], "similarity": sim})
return results
3.3 评估机制：衡量成功的标准
评估机制是查询转换回路的“眼睛”，它告诉我们当前生成的查询和检索到的结果是否足够好，以及是否需要继续迭代。
关键评估指标：
-
相关性分数（Relevance Score）：
- 方法：
- LLM-based评估：让LLM作为一个“裁判”，根据原始查询和检索到的文档内容，打分（例如0-5分）。这是最灵活且能捕捉语义细节的方法。
- 交叉编码器（Cross-encoder）：一个专门训练的模型，输入查询和文档对，输出它们的匹配分数。比LLM更轻量，但需要训练数据。
- 启发式规则：关键词匹配密度、查询词在文档中的位置等（通常用于辅助或作为基线）。
- 聚合：可以对
top_k个文档的相关性分数进行平均、加权求和，或只关注最高分，以得到一个整体的评估。
- 方法：
-
多样性（Diversity）：
- 确保检索结果不仅仅是重复相同内容，而是覆盖了与查询相关的不同方面或视角。
- 方法：计算召回文档向量之间的距离，或者让LLM评估文档内容的差异性。
-
置信度（Confidence Score）：
- 系统对自身检索结果质量的信心。LLM可以根据其评估结果，给出一个置信度分数。
-
下游任务表现：
- 如果查询转换是更大系统（如RAG问答系统）的一部分，那么最终的答案质量（准确性、完整性）是终极指标。
-
用户反馈（User Feedback）：
- 在交互式系统中，用户的点击、点赞/点踩、停留时间等都是重要的隐式或显式反馈信号。
终止条件：
- 达到相关性阈值：评估分数超过预设的满意度阈值。
- 达到最大迭代次数：防止无限循环或资源耗尽。
- 没有显著改善：连续几轮迭代评估分数没有明显提升。
- 查询池耗尽：没有新的、未尝试过的查询变体可以生成。
- 用户满意：在交互式场景中，用户明确表示满意。
class LLMEvaluationNode:
def __init__(self, client):
self.client = client
self.model = "gpt-4o"
def _call_llm(self, prompt, response_format="json_object"):
messages = [{"role": "user", "content": prompt}]
try:
if response_format == "json_object":
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
response_format={"type": "json_object"}
)
else:
response = self.client.chat.completions.create(
model=self.model,
messages=messages
)
return json.loads(response.choices[0].message.content)
except Exception as e:
print(f"Error calling LLM for evaluation: {e}")
return {}
def evaluate_results(self, original_query, search_results):
if not search_results:
return {
"document_relevance": [],
"overall_relevance_score": 0,
"confidence_score": 0,
"overall_summary": "No results found."
}
documents_text = "n---n".join([f"Document {i+1} (ID: {res['id']}):n{res['content']}" for i, res in enumerate(search_results)])
prompt = f"""You are an expert evaluator. Your task is to assess the relevance of the provided search results to the original user query.
Original Query: "{original_query}"
---
Search Results:
{documents_text}
---
For each document, rate its relevance to the original query on a scale of 0-5 (0: completely irrelevant, 5: highly relevant). Also, provide a brief justification.
Finally, provide an overall relevance score for the set of results (0-10, 10 being perfect) and a confidence score (0-10, 10 being very confident that these results address the query).
Output should be a JSON object like this:
{{
"document_relevance": [
{{"id": "doc1", "score": 4, "justification": "..."}},
...
],
"overall_relevance_score": 8,
"confidence_score": 7,
"overall_summary": "..."
}}
"""
evaluation = self._call_llm(prompt)
return evaluation
3.4 迭代逻辑与编排：回路的流程控制
迭代逻辑定义了回路的运行方式，如何从一轮转换到下一轮，以及何时停止。这是一个高级别的控制器，负责协调所有组件。
核心策略：
-
顺序迭代（Sequential Iteration）：
- 每次只选择一个“最佳”查询变体进行下一轮转换。
- 优点：简单，资源消耗少。
- 缺点：可能陷入局部最优，错过其他有潜力的路径。
-
并行/分支迭代（Parallel/Branching Iteration）：
- 在每轮中，生成多个查询变体，并同时对它们进行检索和评估。
- 优点：能更全面地探索查询空间，提高找到最佳结果的可能性。
- 缺点：资源消耗高（需要多次LLM调用和向量数据库查询）。
-
树状搜索（Tree Search）：
- 将查询转换过程视为一个搜索树，每个节点代表一个查询状态，边代表一次转换。
- 可以采用蒙特卡洛树搜索（MCTS）或A*搜索等算法，根据评估分数（奖励）来指导搜索方向。
- 优点：在复杂场景下能进行更智能的探索。
- 缺点：实现复杂，计算成本高。
-
自修正/反射（Self-Correction/Reflection）：
- 让LLM不仅生成查询，还能“反思”前一轮的检索结果，并根据反思结果决定下一步的转换策略。
- 示例：如果发现检索结果过于通用，LLM可以决定进行“查询具体化”；如果发现结果不全面，则进行“查询扩展”。
编排器 (QueryTransformationCircuit) 实现：
编排器将上述所有组件整合到一个统一的流程中。
class QueryTransformationCircuit:
def __init__(self, llm_node, vector_db, eval_node, embeddings_model):
self.llm_node = llm_node
self.vector_db = vector_db
self.eval_node = eval_node
self.embeddings_model = embeddings_model
self.max_iterations = 3 # 最大迭代次数
self.relevance_threshold = 7 # 整体相关性分数阈值 (0-10)
def run_circuit(self, initial_query, context=None):
current_query_set = {initial_query} # 当前待尝试的查询变体集合
all_queries_tried = set() # 记录所有已尝试过的查询，避免重复
best_results_overall = []
best_overall_relevance_score = 0
iteration_history = []
print(f"--- Query Transformation Circuit Started ---")
print(f"Initial Query: '{initial_query}'")
for i in range(self.max_iterations):
print(f"n--- Iteration {i+1}/{self.max_iterations} ---")
queries_for_this_iteration = list(current_query_set - all_queries_tried) # 只处理新的查询
if not queries_for_this_iteration:
print(" No new queries to try in this iteration. Stopping.")
break
all_queries_tried.update(queries_for_this_iteration)
current_iteration_results = []
for query_variant in queries_for_this_iteration:
print(f" Searching with variant: '{query_variant}'")
query_embedding = self.embeddings_model.encode(query_variant)
retrieved_docs = self.vector_db.search(query_embedding, top_k=5)
if not retrieved_docs:
print(f" No documents retrieved for '{query_variant}'.")
continue
eval_data = self.eval_node.evaluate_results(initial_query, retrieved_docs)
overall_relevance = eval_data.get("overall_relevance_score", 0)
print(f" Results for '{query_variant}' (Overall Relevance: {overall_relevance}, Confidence: {eval_data.get('confidence_score',0)}):")
for doc_res in retrieved_docs:
print(f" - Doc ID: {doc_res['id']}, Sim: {doc_res['similarity']:.2f}, Content: '{doc_res['content'][:80]}...'")
current_iteration_results.append({
"query_variant": query_variant,
"retrieved_docs": retrieved_docs,
"evaluation": eval_data,
"overall_relevance": overall_relevance
})
if overall_relevance > best_overall_relevance_score:
best_overall_relevance_score = overall_relevance
best_results_overall = retrieved_docs
if overall_relevance >= self.relevance_threshold:
print(f" Relevance threshold ({self.relevance_threshold}) met with query '{query_variant}'. Stopping circuit.")
iteration_history.append({
"iteration": i + 1,
"queries_processed": queries_for_this_iteration,
"iteration_results": current_iteration_results,
"best_relevance_this_iteration": max([r['overall_relevance'] for r in current_iteration_results]) if current_iteration_results else 0,
"best_overall_relevance": best_overall_relevance_score
})
return best_results_overall, iteration_history
# 记录本轮迭代的历史
iteration_history.append({
"iteration": i + 1,
"queries_processed": queries_for_this_iteration,
"iteration_results": current_iteration_results,
"best_relevance_this_iteration": max([r['overall_relevance'] for r in current_iteration_results]) if current_iteration_results else 0,
"best_overall_relevance": best_overall_relevance_score
})
# 决定下一轮的查询策略
# 如果本轮没有产生任何结果，或者所有结果都很差，则停止
if not current_iteration_results or best_overall_relevance_score < 3: # 设定一个最低的“进步”阈值
print(" No significant improvement or no results in this iteration. Stopping.")
break
# 选出本轮中评估分数最高的查询变体，作为下一轮转换的基准
best_variant_in_current_iter = None
if current_iteration_results:
best_variant_in_current_iter = max(current_iteration_results, key=lambda x: x['overall_relevance'])['query_variant']
print(f" Best performing query in this iteration: '{best_variant_in_current_iter}' (Relevance: {max([r['overall_relevance'] for r in current_iteration_results])})")
else:
# 如果本轮没有结果，尝试对初始查询进行更激进的转换
best_variant_in_current_iter = initial_query
print(f" No results in this iteration, falling back to initial query '{initial_query}' for next transformation.")
# 生成新的查询变体 (扩展、重写、分解)
new_queries_generated = set()
# 优先尝试分解复杂查询
if len(best_variant_in_current_iter.split()) > 5 and any(op in best_variant_in_current_iter.lower() for op in ["and", "or", "but", "with", "how to", "what is"]): # 简单启发式判断是否复杂
decomposed_queries = self.llm_node.decompose_query(best_variant_in_current_iter, context=context)
new_queries_generated.update(decomposed_queries)
print(f" Decomposed '{best_variant_in_current_iter}' into: {list(decomposed_queries)}")
# 始终尝试扩展和重写
expanded_queries = self.llm_node.expand_query(best_variant_in_current_iter, context=context)
rephrased_queries = self.llm_node.rephrase_query(best_variant_in_current_iter, context=context)
new_queries_generated.update(expanded_queries)
new_queries_generated.update(rephrased_queries)
# 将新的且未尝试过的查询加入到下一轮的查询集合中
queries_for_next_iteration = new_queries_generated - all_queries_tried
if not queries_for_next_iteration:
print(" No truly new queries generated for the next iteration. Stopping.")
break
current_query_set = queries_for_next_iteration # 更新为下一轮要处理的新查询
print(f"n--- Query Transformation Circuit Finished ---")
return best_results_overall, iteration_history
4. 高级概念与考量
4.1 上下文管理
仅仅转换当前查询是不够的。一个更智能的回路需要利用丰富的上下文信息：
- 会话历史：用户之前的查询、点击过的文档、交互记录。这有助于理解用户渐进式的意图。
- 用户画像：用户的角色、偏好、专业背景，可以用于个性化查询转换和结果排序。
- 领域知识：将预定义的知识图谱、本体论、业务规则注入LLM的Prompt中，指导其生成更专业的查询。
4.2 混合方法与多模态
- LLM与规则/ML结合：
- LLM生成查询变体，但可以通过规则过滤掉不合法的或低质量的变体。
- 某些简单的查询扩展可以由基于规则的系统快速处理，减少LLM调用成本。
- 多模态检索：如果知识库包含图像、视频等，查询转换回路可以生成多模态的查询（如文本描述+图像特征），以在多模态向量数据库中进行检索。
4.3 多阶段检索（Multi-stage Retrieval）
查询转换回路可以作为多阶段检索流程中的一个环节：
- 初始粗召回：使用原始查询进行快速、宽泛的向量检索。
- 查询转换回路：如果初始召回结果不理想（如相关性低、多样性差），则启动回路进行优化。
- 重排序（Re-ranking）：对回路召回的所有文档，使用更复杂的模型（如交叉编码器）进行精细重排序，提升最终结果的准确性。
- 生成式回答（Generation）：将最优文档提供给LLM，生成最终的用户回答（RAG模式）。
4.4 模糊性处理与意图识别
- 隐式意图推断：LLM可以根据查询的词汇、结构，结合上下文，推断用户更深层次的意图。
- 显式澄清：如果系统是交互式的，转换节点可以生成一系列澄清性问题，由用户选择或回答，从而引导查询走向精确。例如，对于“我需要一个模型”，LLM可以问“你指的是机器学习模型还是3D模型？”。
4.5 性能与可伸缩性
查询转换回路涉及多次LLM调用和向量数据库查询，可能带来显著的延迟和成本：
- LLM优化：
- 使用更小、更快的模型进行某些简单转换。
- 批量处理LLM请求。
- 缓存LLM响应。
- 精简Prompt，减少Token消耗。
- 向量数据库优化：
- 高效的索引结构（HNSW等）。
- 分布式部署。
- 使用
top_k较小的检索，减少数据传输。
- 并行化：尽可能并行执行多个查询变体的检索和评估。
- 智能停止：严格控制迭代次数和评估阈值，避免不必要的计算。
4.6 可解释性与可控性
LLM的“黑箱”特性可能导致转换过程难以理解和控制：
- Prompt设计：在Prompt中明确要求LLM解释其转换逻辑。
- 透明度：向用户展示查询是如何被转换的，以及为什么被转换。
- 人工干预：允许专家配置规则来指导或限制LLM的转换行为。
- 安全性：确保LLM在转换过程中不会引入偏见、不准确信息或安全漏洞。
5. 实践演练：构建文档搜索回路
让我们通过一个具体的场景来整合上述组件：为复杂的软件库构建一个文档搜索系统。
场景描述：
用户查询：“如何设置用户身份验证？”
目标：从大量的开发文档中，召回与用户身份验证相关的最准确、最全面的文档，即使文档使用了“OAuth”、“JWT”、“SSO”等更具体的术语。
我们将使用之前定义的 MockOpenAIClient、MockEmbeddingsModel、MockVectorDB、LLMTransformationNode 和 LLMEvaluationNode 来构建 QueryTransformationCircuit。
# --- 实例化并配置组件 ---
embeddings_model = MockEmbeddingsModel()
mock_openai_client = MockOpenAIClient()
transformation_node = LLMTransformationNode(mock_openai_client)
vector_db = MockVectorDB(embeddings_model)
evaluation_node = LLMEvaluationNode(mock_openai_client)
# --- 添加一些模拟文档 ---
# 文档ID, 内容
docs_to_add = [
("doc1", "Detailed guide on OAuth 2.0 implementation for secure API access in our platform."),
("doc2", "Basic setup for local development environment using Docker and Kubernetes."),
("doc3", "Understanding JWT tokens for stateless authentication in microservices architecture."),
("doc4", "How to configure SAML for enterprise single sign-on (SSO) integration."),
("doc5", "Database migration strategies with Alembic and PostgreSQL."),
("doc6", "Setting up user registration and login forms with Flask-Security library."),
("doc7", "Troubleshooting common network connectivity issues and firewall configurations."),
("doc8", "Best practices for password hashing and storage in web applications."),
("doc9", "Implementing two-factor authentication (2FA) for enhanced security."),
("doc10", "Role-Based Access Control (RBAC) setup and management."),
("doc11", "Secure communication between microservices using mTLS."),
("doc12", "Configuring SSL/TLS for database connections."),
("doc13", "User session management and logout procedures."),
("doc14", "Deep dive into OpenID Connect for identity verification."),
("doc15", "Protecting sensitive data at rest and in transit.")
]
print("--- Populating Mock Vector Database ---")
for doc_id, content in docs_to_add:
vector_db.add_document(doc_id, content)
print(f"Added {len(docs_to_add)} documents.")
# --- 创建查询转换回路实例 ---
circuit = QueryTransformationCircuit(transformation_node, vector_db, evaluation_node, embeddings_model)
# --- 运行回路 ---
initial_query_1 = "how to set up user authentication?"
final_results_1, history_1 = circuit.run_circuit(initial_query_1)
print("nn--- Final Best Results for Query 1 ---")
if final_results_1:
for res in final_results_1:
print(f"ID: {res['id']}, Similarity: {res['similarity']:.2f}, Content: '{res['content'][:120]}...'")
else:
print("No relevant results found after circuit execution.")
print("n--- History of Transformations for Query 1 ---")
for entry in history_1:
print(f"Iteration {entry['iteration']}:")
print(f" Queries Processed: {entry['queries_processed']}")
print(f" Best Relevance This Iteration: {entry['best_relevance_this_iteration']}")
print(f" Best Overall Relevance So Far: {entry['best_overall_relevance']}")
# 可以打印更多 iteration_results 细节
# for res_entry in entry['iteration_results']:
# print(f" - Query: '{res_entry['query_variant']}', Eval: {res_entry['evaluation']['overall_relevance_score']}")
print("n" + "="*80 + "n")
# --- 尝试一个更复杂的查询 ---
initial_query_2 = "how to implement user login and protect database connection security?"
final_results_2, history_2 = circuit.run_circuit(initial_query_2)
print("nn--- Final Best Results for Query 2 ---")
if final_results_2:
for res in final_results_2:
print(f"ID: {res['id']}, Similarity: {res['similarity']:.2f}, Content: '{res['content'][:120]}...'")
else:
print("No relevant results found after circuit execution.")
print("n--- History of Transformations for Query 2 ---")
for entry in history_2:
print(f"Iteration {entry['iteration']}:")
print(f" Queries Processed: {entry['queries_processed']}")
print(f" Best Relevance This Iteration: {entry['best_relevance_this_iteration']}")
print(f" Best Overall Relevance So Far: {entry['best_overall_relevance']}")
代码解释：
- 初始化：我们创建了所有必要的组件实例：LLM转换节点、向量数据库、LLM评估节点和嵌入模型。
- 文档填充：向模拟向量数据库中添加了一系列模拟文档。这些文档故意使用了不同的术语，以模拟真实世界的语义多样性。
- 回路实例：将所有组件传入
QueryTransformationCircuit，构建了整个检索回路。 - 运行回路：
- 首先以一个相对宽泛的查询
"how to set up user authentication?"运行回路。 - 回路会开始迭代。在每次迭代中：
- 它会取出当前
current_query_set中的所有查询变体。 - 对每个变体，生成嵌入，然后查询
MockVectorDB。 LLMEvaluationNode会评估MockVectorDB返回的文档与original_query的相关性。- 如果达到预设的相关性阈值，或者达到最大迭代次数，回路将停止。
- 否则，
LLMTransformationNode会基于本轮表现最好的查询变体，生成新的扩展、重写或分解后的查询。这些新查询将构成下一轮的current_query_set。
- 它会取出当前
- 第二个查询
"how to implement user login and protect database connection security?"是一个复合查询，展示了decompose_query的潜在作用。
- 首先以一个相对宽泛的查询
- 结果输出：打印最终召回的最佳文档以及整个迭代历史，包括每轮使用的查询和评估分数。这有助于我们理解回路是如何逐步逼近最佳结果的。
通过这个演练，我们可以清晰地看到，即使初始查询可能无法直接匹配到包含“OAuth”、“JWT”或“Flask-Security”的文档，查询转换回路也能通过迭代的语义扩展和重写，最终将用户的意图与这些具体的技术文档关联起来，从而召回更精准的结果。
6. 展望未来：智能检索的演进
查询转换回路代表了智能检索系统从被动匹配到主动优化的重要转变。它不再仅仅是等待用户提供完美的查询，而是能够主动地理解、推断和优化用户的意图。
随着大语言模型能力的进一步提升，未来的查询转换回路将更加智能：
- 更深层次的意图理解：LLM将能够更准确地推断用户在模糊查询背后的真实需求，甚至预测其潜在的后续问题。
- 多模态转换：不仅限于文本，未来的查询转换可能包括图像、语音等多种模态的输入和输出。例如，用户上传一张图表，LLM可以将其转换为描述性文本，再进行检索。
- 主动式交互：系统能以更自然的方式与用户进行多轮对话，通过提问、确认，逐步细化查询，实现真正的“人机协作式”检索。
- 自我学习与适应：回路可以通过分析历史查询、用户反馈和检索效果，持续优化其转换策略和评估模型，实现自适应进化。
- 更强的可解释性与控制：虽然LLM是黑箱，但结合可解释AI技术，我们可以更好地理解转换过程，并提供更精细的控制机制。
查询转换回路不仅仅是一个技术方案，它更是一种思维模式，一种将复杂的用户需求与庞大的信息海洋连接起来的智能桥梁。通过持续的迭代优化，我们能够让信息检索系统变得更加聪明、更加人性化。