![](data:image/svg+xml,<svg xmlns="http://www.w3.org/2000/svg" width="32" height="32"><text y="28" font-size="28">🤑</text></svg>)
知识库查询 —— 商品名确认节点
本文档详细介绍商品名确认节点（item_name_confirm）的设计与实现，该节点是知识库查询流程的入口节点，负责从用户问题中提取并确认商品名称。
学习理念：商品名确认是查询流程的"守门员"——它决定用户到底在问哪个产品。核心是两阶段策略：先用 LLM 从自然语言中提取商品名（语义理解 + 代词消解），再用 Milvus 向量检索对齐到数据库中的精确名称（模糊容错）。确认失败时直接返回提示，避免无效检索。
海外对标：ItemNameConfirmNode 的"LLM 提取 → 向量匹配 → 阈值评分"三阶段模式对标 Google 搜索的 Query Understanding + Entity Linking 流程，以及 Amazon Kendra 的 Query Augmentation + Index Selection 机制。MongoDB 历史对话管理对标 LangChain 的
ConversationBufferMemory。
本节 AI 替代率：~75% | 人工干预率：~25%
| 角色 | 能力范围 |
|---|---|
| 🤖 AI 擅长 | LLM 调用 + JSON 解析、向量检索代码、评分对齐逻辑、MongoDB 读写、测试代码 |
| 👤 人类需理解 | 两阶段策略的设计意图（为什么先用 LLM 再用向量）、多阈值评分对齐（0.63/0.6 阈值的调优）、历史回填的业务含义 |
阅读指引
| 颜色 | 章节 | AI 替代率 | 人工干预 | 说明 |
|---|---|---|---|---|
| 🟡 | §1 任务目标 | ~95% | ~5% | 学习目标明确 |
| 🟢 | §2 核心概念 | ~95% | ~5% | 两阶段策略 / 评分阈值 / 代词消解 |
| 🟡 | §3 整体流程 | ~90% | ~10% | 输入输出数据结构 |
| 🔴 | §4 分步实现 | ~70% | ~30% | 8 步流程中 Step3(LLM提取) + Step4(向量匹配) + Step5(评分对齐) 是核心 |
| 🔴 | §4.4 主代码 | ~65% | ~35% | ItemNameConfirmNode 完整架构，重点是 _extract_item_names + _align_by_score |
| 🟢 | §5 测试运行 | ~95% | ~5% | 看预期输出 + 多轮对话测试 |
| 🟡 | §6 总结 | ~90% | ~10% | 设计要点回顾 |
技术栈健康度标签体系
| 技术 | 健康度 | 建议 |
|---|---|---|
| MongoDB | 🟢 稳定 | 文档型数据库，存储对话历史。get_recent_messages + update_message_item_names 模式固定。 |
| Milvus 混合检索 | 🔥 巅峰 | 稠密+稀疏双向量检索，权重各 0.5。execute_hybrid_search + ranker_weights 是混合检索的标准 API。 |
| JSON Mode | 🔥 巅峰 | LLM 的 json_object 模式确保输出结构化的 JSON。需注意数组需要用对象包装。 |
| LLM 代词消解 | ⏳ 成长期 | LLM 解决代词指代的能力在 2025-2026 年大幅提升，但仍需历史对话作为上下文。 |
体系说明：🟢🟡🟠🔴 标识学习优先级 / AI 替代率；🔥🟢⏳⚠️💀 标识技术栈健康度。
中英文对照表
| English | 中文 | 本质 |
|---|---|---|
| Item Name Confirmation | 商品名确认 | 从用户问题中提取并确认具体产品名称 |
| Anaphora Resolution | 代词指代消解 | 将"这个""它"等代词替换为具体名词 |
| Query Rewriting | 查询改写 | 将代词替换为具体名称，生成独立完整的问题 |
| Vector Match | 向量匹配 | 用向量相似度将用户表述与数据库精确名称对齐 |
| Score Alignment | 评分对齐 | 根据相似度阈值决定确认/候选/忽略 |
| Confidence Threshold | 置信度阈值 | 决定向量匹配结果是否可用的分数门槛 |
| Hybrid Search Weight | 混合搜索权重 | 稠密向量和稀疏向量在最终得分中的占比 |
| History Backfill | 历史回填 | 将新确认的商品名补写到之前缺失的历史记录 |
💡 程序员比喻
- ItemNameConfirmNode 就像 Git 的 pre-receive hook——在代码（查询）进入主流程之前，先检查 commit message（商品名）是否合法。
- LLM 提取 + 向量匹配 就像
grep+fzf——LLM = grep（语义过滤出候选），向量 = fzf（模糊匹配到精确值）。- 多阈值评分 就像 CI 的三个 stage：score > 0.63 = lint 通过（直接合并），0.6~0.63 = 需要人工 review（用户选择），< 0.6 = rejected（忽略）。
- 历史回填 就像
git rebase——把历史 commit 中缺失的信息（item_names）补上，保持完整。
1. 任务目标
1.1 本章目标
通过本章学习，你将掌握：
- 理解商品名确认的业务逻辑：掌握从用户问题到商品确认的完整流程
- 学会 LLM + 向量检索的联合使用：先用 LLM 提取候选名称���再用向量检索对齐数据库
- 理解评分对齐策略：掌握多阈值评分机制的设计思想
- 掌握历史对话管理：学会使用 MongoDB 管理多轮对话上下文
- 实现可测试的节点代码：通过
if __name__ == "__main__"验证节点功能
1.2 涉及文件
knowledge/processor/query_process/
├── nodes/
│ └── item_name_confirm.py # 商品名确认节点（本章重点）
├── prompt.py # 提示词模板
└── ...
knowledge/tools/
├── llm_utils.py # LLM 客户端工具
├── mongo_history_utils.py # MongoDB 历史对话工具
├── embedding_utils.py # 向量嵌入工具
└── milvus_utils.py # Milvus 向量数据库工具1.3 节点在流程中的位置
2. 核心概念扫盲
2.1 为什么需要商品名确认？
在知识库问答场景中，用户的问题往往是模糊的：
| 用户问题 | 问题 | 解决方案 |
|---|---|---|
| "这个怎么用？" | 指代不明 | 需要结合历史对话推断 |
| "万用表怎么测电压" | 商品名不精确 | 需要对齐到具体型号 |
| "苏伯尔RS-12怎么换电池" | 可能有错别字 | 需要向量相似度匹配 |
商品名确认的作用：
- 精准定位知识：将模糊的商品名对齐到数据库中的精确名称
- 支持多轮对话：通过历史记录解决代词指代问题
- 流程分流：确认失败时直接返回提示，避免无效检索
2.2 LLM + 向量检索的两阶段策略
2.3 多阈值评分对齐机制
🟡 【P1 看注释就行】 Step1 获取历史——MongoDB 查询最近 10 条对话记录。异常时返回空列表（降级）。
python
HIGH_CONFIDENCE_THRESHOLD = 0.63 # 高置信阈值
MID_CONFIDENCE_THRESHOLD = 0.6 # 中置信阈值| 分数区间 | 处理策略 | 说明 |
|---|---|---|
| score > 0.63 | 直接确认 | 高置信度，作为最终商品名 |
| 0.6 ≤ score ≤ 0.63 | 作为候选选项 | 中置信度，让用户选择确认 |
| score < 0.6 | 忽略 | 低置信度，匹配不可靠 |
2.4 代词指代消解
LLM 能够根据历史对话解决代词指代问题：
历史对话:
user: 万用表怎么测电压？
assistant: 万用表测电压需要先选择...
当前问题:
user: 那怎么换电池呢？
↓
LLM 提取:
item_names: ["万用表"]
rewritten_query: "万用表怎么换电池？"2.5 JSON Mode
调用 LLM 时使用 json_mode=True，确保输出是严格的 JSON 格式：
🟡 【P1 看注释就行】 Step2 保存用户问题——先 insert 获取
message_id，供后续 update 使用。
python
client = get_llm_client(model, json_mode=True)注意事项：
json_object模式不支持返回 JSON 数组，需要用对象包装- 提示词中需要明确告知期望的 JSON 结构
3. 商品名确认业务处理流程（总）
3.1 整体流程图
3.2 输入输出数据结构
输入状态：
🔥 【P0 必须要学】 Step3 LLM 提取是核心。注意：
json_mode=True确保结构化输出；响应清洗去围栏；降级机制。ITEM_NAME_EXTRACT_TEMPLATE要求同时输出item_names+rewritten_query（代词替换）。
python
{
"session_id": "abc123", # 会话 ID（必需）
"original_query": "万用表怎么用？", # 用户原始问题（必需）
"item_names": [], # 初始为空或已有值
"rewritten_query": "", # 初始为空
"answer": "", # 初始为空
"history": [], # 初始为空，节点内部加载
}输出状态（确认成功）：
🟡 【P1 看注释就行】 Step4 向量匹配——
generate_hybrid_embeddings生成双向量，ranker_weights=(0.5, 0.5)稠密/稀疏各半。
python
{
"session_id": "abc123",
"original_query": "万用表怎么用？",
"item_names": ["万用表RS-12"], # 确认的商品名
"rewritten_query": "万用表RS-12怎么用？", # 改写的问题
"answer": "", # 空，继续检索流程
"history": [...], # 历史对话记录
}输出状态（需要用户选择）：
🔥 【P0 必须要学】 Step5 评分对齐是决策核心。三层：
score>0.63=确认（优先精确匹配）→>=0.6=候选 →<0.6=忽略。两个阈值决定了确认精度。
python
{
"session_id": "abc123",
"original_query": "万用表怎么用？",
"item_names": [], # 未确认
"rewritten_query": "",
"answer": "我不确定您指的是哪款产品。您是在询问以下产品吗：万用表A、万用表B？",
"history": [...],
}4. 商品名确认业务处理流程（分）
4.1 目标
实现一个能够：
- 从用户问题中智能提取商品名称
- 将提取的名称与数据库中的商品精确对齐
- 根据对齐结果决定后续流程走向
- 支持多轮对话的上下文理解
4.2 需求分析
4.2.1 功能需求
- 历史对话获取：从 MongoDB 加载最近的对话记录
- 商品名提取：使用 LLM 从问题中提取候选商品名
- 问题改写：将代词替换为具体名称，形成独立完整的问题
- 向量匹配：使用混合检索在商品名集合中搜索
- 评分对齐：根据相似度分数决定确认、候选或忽略
- 历史回填：将确认的商品名回填到之前的对话记录
4.2.2 技术依赖
- LLM：用于商品名提取和问题改写
- MongoDB：存储和查询历史对话
- Milvus：商品名向量集合，支持混合搜索
- BGE 模型：生成稠密向量和稀疏向量
4.3 实现流程
4.3.1 实现流程图
4.3.2 具体实现步骤
Step1：获取历史对话
目的： 从 MongoDB 加载当前会话的历史对话记录，用于 LLM 理解上下文。
实现逻辑：
- 根据
session_id查询 MongoDB 的chat_message集合 - 按时间戳升序排列，获取最近 10 条记录
- 返回消息列表，每条包含
role、text、item_names等字段
代码片段：
🟡 【P1 看注释就行】 Step6 更新状态——三种结果：confirm→回填+继续 / options→提示选择 / none→无法识别。
python
def _get_history(self, session_id: str, limit: int = 10) -> List[Dict]:
"""获取历史会话记录。"""
from knowledge.tools.mongo_history_utils import get_recent_messages
try:
return get_recent_messages(session_id, limit=limit)
except Exception as e:
self.logger.warning(f"获取历史记录失败: {e}")
return []Step2：保存用户问题
目的： 将当前用户问题先保存到 MongoDB，获取 message_id 供后续更新使用。
实现逻辑：
- 构建消息文档（session_id, role="user", text=query）
- 插入 MongoDB，获取自动生成的
_id - 返回
message_id字符串
代码片段：
🟡 【P1 看注释就行】 Step7 写入历史——更新用户问题（补充 rewritten_query），写入助手回复。
python
message_id = self._save_message(
session_id,
"user",
query,
item_names=state.get("item_names", [])
)Step3：LLM 提取商品名
目的： 使用 LLM 从用户问题和历史对话中提取商品名称，并改写问题。
实现逻辑：
- 获取 LLM 客户端：使用
ITEM_MODEL模型，开启 JSON 模式 - 构建历史文本：将历史对话格式化为 "role: text\n" 形式
- 组装提示词：使用
ITEM_NAME_EXTRACT_TEMPLATE模板 - 调用 LLM：发送 SystemMessage + HumanMessage
- 解析响应：
- 去除首尾空格
- 处理代码块围栏（```json...）
- 解析 JSON
- 返回结果：
{"item_names": [...], "rewritten_query": "..."}
提示词模板：
🟡 【P1 看注释就行】 Step8 历史回填——补写之前
item_names为空的历史记录。
python
ITEM_NAME_EXTRACT_TEMPLATE = """
历史会话：
{history_text}
当前用户问题：{query}
请根据历史会话和当前问题，提取用户正在询问的商品名称（item_names）。
1. 如果用户明确提到了商品名称，请提取出来。可能有一个或多个，但不能重复。
2. 如果用户使用了代词（如"这个"、"它"），请结合历史会话指代消解，确定商品名称。
3. 如果无法确定商品名称，item_names 返回空列表。
4. 请重新改写用户的问题（rewritten_query），使其成为包含商品名称的独立完整问题。
请直接返回JSON格式结果，格式如下：
{{
"item_names": ["商品A", "商品B"],
"rewritten_query": "关于商品A和商品B，..."
}}"""代码片段：
🔥 【P0 必须要学】 ItemNameConfirmNode 完整代码——主流程 5 步：获取历史→LLM提取→向量匹配评分→更新状态→写入历史。重点理解
_update_state的三种分流逻辑和_extract_item_names的 JSON 清洗。
python
def _extract_item_names(self, query: str, history: List[Dict]) -> Dict[str, Any]:
from knowledge.tools.llm_utils import get_llm_client
# 1. 获取LLM客户端
client = get_llm_client(os.getenv("ITEM_MODEL"), json_mode=True)
# 2. 构建历史对话文本
history_text = "".join(
f"{msg.get('role', 'unknown')}: {msg.get('text', '')}\n"
for msg in history
)
# 3. 组装提示词
prompt = ITEM_NAME_EXTRACT_TEMPLATE.format(
history_text=history_text,
query=query
)
try:
# 4. 调用LLM
response = client.invoke([
SystemMessage(content="你是一个专业的客服助手..."),
HumanMessage(content=prompt),
])
# 5. 清洗响应
content = response.content.strip()
if content.startswith("```"):
content = content.split("```")[1].removeprefix("json").strip()
# 6. 解析JSON
result = json.loads(content)
result.setdefault("item_names", [])
result.setdefault("rewritten_query", query)
result["item_names"] = [n.strip() for n in result["item_names"]]
return result
except Exception as e:
self.logger.error(f"LLM 提取商品名称失败: {e}")
return {"item_names": [], "rewritten_query": query}Step4：向量匹配
目的： 将 LLM 提取的候选商品名与 Milvus 中的商品名集合进行向量匹配。
实现逻辑：
- 获取 Milvus 客户端：连接到 Milvus 服务
- 生成混合向量：对每个候选名生成稠密向量和稀疏向量
- 构建检索请求：创建混合检索请求对象
- 执行混合检索：
- 稠密向量权重: 0.5
- 稀疏向量权重: 0.5
- 返回 TopK 结果
- 整理结果：每个候选名对应一组匹配结果（item_name, score）
代码片段：
🟢 【P2 后面可以查】 测试代码——第一轮正常提取 + 第二轮代词指代消解。
python
def _vector_search(self, item_names: List[str]) -> List[Dict[str, Any]]:
from knowledge.tools.embedding_utils import generate_hybrid_embeddings
from knowledge.tools.milvus_utils import (
get_milvus_client,
build_hybrid_search_requests,
execute_hybrid_search,
)
# 1. 获取 Milvus 客户端
client = get_milvus_client()
collection_name = "item_name_collection_test"
# 2. 生成混合向量
embeddings = generate_hybrid_embeddings(item_names)
results = []
# 3. 遍历每个候选商品名
for i, name in enumerate(item_names):
try:
# 3.1 构建混合检索请求
reqs = build_hybrid_search_requests(
dense_vector=embeddings["dense"][i],
sparse_vector=embeddings["sparse"][i],
top_k=self.MAX_OPTIONS,
)
# 3.2 执行混合检索
search_res = execute_hybrid_search(
client=client,
collection_name=collection_name,
search_requests=reqs,
ranker_weights=(0.5, 0.5),
top_k=self.MAX_OPTIONS,
normalize_score=True,
output_fields=["item_name"],
)
# 3.3 整理匹配结果
matches = [
{"item_name": hit["entity"]["item_name"], "score": hit["distance"]}
for hit in (search_res[0] if search_res else [])
]
results.append({"extracted_name": name, "matches": matches})
except Exception as e:
self.logger.error(f"查询商品名称 {name} 失败: {e}")
return resultsStep5：评分对齐
目的： 根据向量匹配的相似度分数，决定商品名的确认状态。
实现逻辑：
- 遍历每个候选的匹配结果
- 按分数降序排序
- 划分分数区间：
- 高置信（score > 0.63）
- 中置信（0.6 ≤ score ≤ 0.63）
- 确认策略：
- 高置信且有精确匹配 → 使用精确匹配
- 高置信无精确匹配 → 使用最高分
- 仅有中置信 → 作为候选选项
- 返回结果：
{"confirmed_item_names": [...], "options": [...]}
代码片段：
python
def _align_by_score(self, query_results: List[Dict]) -> Dict[str, Any]:
confirmed: List[str] = []
options: List[str] = []
for res in query_results:
extracted = (res.get("extracted_name") or "").strip()
# 按分数降序排序
matches = sorted(
res.get("matches") or [],
key=lambda m: m.get("score", 0),
reverse=True,
)
if not matches:
continue
# 划分分数区间
high = [m for m in matches if m["score"] > self.HIGH_CONFIDENCE_THRESHOLD]
mid = [m for m in matches if m["score"] >= self.MID_CONFIDENCE_THRESHOLD]
if high:
# 高置信: 优先精确匹配，否则取最高分
exact = next(
(m for m in high if m["item_name"].strip() == extracted),
None
)
confirmed.append((exact or high[0])["item_name"])
elif mid:
# 中置信: 作为候选选项
options.extend(m["item_name"] for m in mid[:self.MAX_OPTIONS])
return {
"confirmed_item_names": confirmed,
"options": options[:self.MAX_OPTIONS],
}Step6：更新状态
目的： 根据对齐结果更新图状态，决定后续流程走向。
实现逻辑：
有确认的商品名：
- 回填历史记录中缺失的商品名
- 更新
state["item_names"] - 更新
state["rewritten_query"] - 不设置
answer（继续检索流程）
有候选选项：
- 设置
answer为选择提示 - 流程将在此中断，等待用户确认
- 设置
无匹配：
- 设置
answer为无法识别提示 - 流程将在此中断
- 设置
代码片段：
python
def _update_state(
self, state: QueryGraphState, align_result: Dict,
rewritten_query: str, history: List[Dict],
) -> QueryGraphState:
confirmed = align_result.get("confirmed_item_names", [])
options = align_result.get("options", [])
if confirmed:
# 确认成功
self._backfill_history_item_names(history, confirmed)
state["item_names"] = confirmed
state["rewritten_query"] = rewritten_query
elif options:
# 有候选，需要用户选择
state["answer"] = f"我不确定您指的是哪款产品。您是在询问以下产品吗：{'、'.join(options)}"
else:
# 无法识别
state["answer"] = "抱歉，我无法识别您询问的具体产品名称，请提供更准确的产品名称或型号。"
return stateStep7：写入历史
目的： 将本轮对话完整写入历史记录。
实现逻辑：
- 更新用户问题：使用之前获取的
message_id，补充rewritten_query和item_names - 写入助手回复：如果有
answer（确认失败场景），保存助手消息
代码片段：
python
def _write_history(
self, state: QueryGraphState, session_id: str,
rewritten_query: str, message_id: str,
):
query = (state.get("original_query") or "").strip()
answer = (state.get("answer") or "").strip()
item_names = state.get("item_names") or []
# 更新用户问题消息
if query:
self._save_message(
session_id,
"user",
query,
rewritten_query=rewritten_query,
item_names=item_names,
message_id=message_id,
)
# 写入助手回复
if answer:
self._save_message(
session_id,
"assistant",
answer,
item_names=item_names,
)Step8：历史回填
目的： 当商品名首次被确认时，将其回填到之前缺少商品名的历史记录。
实现逻辑：
- 筛选需要更新的记录：找出
item_names为空的历史消息 - 更新内存中的 history：直接修改列表对象
- 批量更新 MongoDB：使用
update_many一次更新所有记录
代码片段：
python
def _backfill_history_item_names(self, history: List[Dict], item_names: List[str]):
from knowledge.tools.mongo_history_utils import update_message_item_names
# 1. 获取要更新的消息 ID
ids_to_update = [
msg["_id"] for msg in history if not msg.get("item_names")
]
if not ids_to_update:
return
# 2. 更新内存中的 history 对象
for msg in history:
if not msg.get("item_names"):
msg["item_names"] = item_names
# 3. 批量更新 MongoDB
try:
update_message_item_names(ids_to_update, item_names)
except Exception as e:
self.logger.warning(f"回填历史商品名称失败: {e}")4.4 代码实现
以下是完整的节点实现代码：
python
# knowledge/processor/query_process/nodes/item_name_confirm.py
"""商品名称确认节点
从用户查询中提取商品名称，通过向量相似度匹配与数据库中已有商品对齐确认。
"""
import os
import json
from typing import List, Dict, Any
from langchain_core.messages import SystemMessage, HumanMessage
from knowledge.processor.query_process.base import BaseNode, setup_logging
from knowledge.processor.query_process.state import QueryGraphState
from knowledge.processor.query_process.prompt import ITEM_NAME_EXTRACT_TEMPLATE
class ItemNameConfirmNode(BaseNode):
"""商品名称确认节点。
流程: 获取历史 → LLM提取商品名 → 向量匹配 → 评分对齐 → 更新状态 → 写入历史
"""
name = "item_name_confirm"
# 对齐阈值
HIGH_CONFIDENCE_THRESHOLD = 0.63 # 高置信阈值
MID_CONFIDENCE_THRESHOLD = 0.6 # 中置信阈值
MAX_OPTIONS = 5 # 最大候选数
# ================================================================== #
# 主流程 #
# ================================================================== #
def process(self, state: QueryGraphState) -> QueryGraphState:
session_id = state["session_id"]
query = state.get("original_query", "")
# 1. 获取历史记录
history = self._get_history(session_id)
# 1.1 保存用户问题（获取 message_id）
message_id = self._save_message(
session_id, "user", query,
item_names=state.get("item_names", [])
)
# 2. LLM 提取商品名称
extract_res = self._extract_item_names(query, history)
item_names = extract_res.get("item_names", [])
rewritten_query = extract_res.get("rewritten_query", query)
# 3. 向量匹配 + 评分对齐
align_result = self._match_and_align(item_names) if item_names else {}
# 4. 更新状态
state = self._update_state(state, align_result, rewritten_query, history)
# 5. 写入历史
self._write_history(state, session_id, rewritten_query, message_id)
state["history"] = history
return state
# ================================================================== #
# 历史记录操作 #
# ================================================================== #
def _get_history(self, session_id: str, limit: int = 10) -> List[Dict]:
"""获取历史会话记录。"""
from knowledge.tools.mongo_history_utils import get_recent_messages
try:
return get_recent_messages(session_id, limit=limit)
except Exception as e:
self.logger.warning(f"获取历史记录失败: {e}")
return []
def _save_message(
self, session_id: str, role: str, text: str,
rewritten_query: str = "", item_names: List[str] = None,
message_id: str = "",
) -> str:
"""保存单条消息到历史记录。"""
from knowledge.tools.mongo_history_utils import save_chat_message
try:
return save_chat_message(
session_id=session_id,
role=role,
text=text,
rewritten_query=rewritten_query,
item_names=item_names or [],
**({"message_id": message_id} if message_id else {}),
)
except Exception as e:
self.logger.warning(f"保存消息失败: {e}")
return ""
# ================================================================== #
# LLM 提取商品名称 #
# ================================================================== #
def _extract_item_names(self, query: str, history: List[Dict]) -> Dict[str, Any]:
"""使用 LLM 从查询和历史中提取商品名称。
Returns:
{"item_names": [...], "rewritten_query": "..."}
"""
from knowledge.tools.llm_utils import get_llm_client
# 1. 获取 LLM 客户端
client = get_llm_client(os.getenv("ITEM_MODEL"), json_mode=True)
# 2. 构建历史对话文本
history_text = "".join(
f"{msg.get('role', 'unknown')}: {msg.get('text', '')}\n"
for msg in history
)
# 3. 组装提示词
prompt = ITEM_NAME_EXTRACT_TEMPLATE.format(
history_text=history_text, query=query
)
try:
# 4. 调用 LLM
response = client.invoke([
SystemMessage(content="你是一个专业的客服助手，擅长理解用户意图和提取关键信息。"),
HumanMessage(content=prompt),
])
# 5. 清洗空格
content = response.content.strip()
# 6. 清洗代码块围栏
if content.startswith("```"):
content = content.split("```")[1].removeprefix("json").strip()
# 7. 解析 JSON
result = json.loads(content)
result.setdefault("item_names", [])
result.setdefault("rewritten_query", query)
result["item_names"] = [n.strip() for n in result["item_names"]]
return result
except Exception as e:
self.logger.error(f"LLM 提取商品名称失败: {e}")
return {"item_names": [], "rewritten_query": query}
# ================================================================== #
# 向量匹配 + 评分对齐 #
# ================================================================== #
def _match_and_align(self, item_names: List[str]) -> Dict[str, Any]:
"""向量检索 + 评分对齐。
Returns:
{"confirmed_item_names": [...], "options": [...]}
"""
# 1. 向量检索
query_results = self._vector_search(item_names)
# 2. 评分对齐
return self._align_by_score(query_results)
def _vector_search(self, item_names: List[str]) -> List[Dict[str, Any]]:
"""批量向量检索商品名称。"""
from knowledge.tools.embedding_utils import generate_hybrid_embeddings
from knowledge.tools.milvus_utils import (
get_milvus_client,
build_hybrid_search_requests,
execute_hybrid_search,
)
# 1. 获取 Milvus 客户端
client = get_milvus_client()
if not client:
self.logger.error("无法连接到 Milvus")
return []
# 2. 集合名称
collection_name = "item_name_collection_test"
# 3. 生成混合向量
embeddings = generate_hybrid_embeddings(item_names)
results = []
# 4. 遍历每个候选商品名
for i, name in enumerate(item_names):
try:
# 4.1 构建混合检索请求
reqs = build_hybrid_search_requests(
dense_vector=embeddings["dense"][i],
sparse_vector=embeddings["sparse"][i],
top_k=self.MAX_OPTIONS,
)
# 4.2 执行混合检索
search_res = execute_hybrid_search(
client=client,
collection_name=collection_name,
search_requests=reqs,
ranker_weights=(0.5, 0.5),
top_k=self.MAX_OPTIONS,
normalize_score=True,
output_fields=["item_name"],
)
# 4.3 整理匹配结果
matches = [
{"item_name": hit["entity"]["item_name"], "score": hit["distance"]}
for hit in (search_res[0] if search_res else [])
]
results.append({"extracted_name": name, "matches": matches})
except Exception as e:
self.logger.error(f"查询商品名称 {name} 失败: {e}")
return results
def _align_by_score(self, query_results: List[Dict]) -> Dict[str, Any]:
"""根据评分对齐商品名称。
规则:
- score > 0.63 且唯一 → 直接确认
- score > 0.63 且多条 → 优先取与提取名完全匹配的，否则取最高分
- 0.6 ≤ score < 0.63 → 作为候选选项
- score < 0.6 → 忽略
"""
confirmed: List[str] = []
options: List[str] = []
for res in query_results:
extracted = (res.get("extracted_name") or "").strip()
# 按分数降序排序
matches = sorted(
res.get("matches") or [],
key=lambda m: m.get("score", 0),
reverse=True,
)
if not matches:
continue
# 划分分数区间
high = [m for m in matches if m["score"] > self.HIGH_CONFIDENCE_THRESHOLD]
mid = [m for m in matches if m["score"] >= self.MID_CONFIDENCE_THRESHOLD]
if high:
# 高置信: 优先精确匹配，否则取最高分
exact = next(
(m for m in high if m["item_name"].strip() == extracted),
None
)
confirmed.append((exact or high[0])["item_name"])
elif mid:
# 中置信: 作为候选选项
options.extend(m["item_name"] for m in mid[:self.MAX_OPTIONS])
return {
"confirmed_item_names": confirmed,
"options": options[:self.MAX_OPTIONS],
}
# ================================================================== #
# 状态更新 & 历史写入 #
# ================================================================== #
def _update_state(
self, state: QueryGraphState, align_result: Dict,
rewritten_query: str, history: List[Dict],
) -> QueryGraphState:
"""根据对齐结果更新 state。"""
confirmed = align_result.get("confirmed_item_names", [])
options = align_result.get("options", [])
if confirmed:
# 确认成功：回填历史，更新状态
self._backfill_history_item_names(history, confirmed)
state["item_names"] = confirmed
state["rewritten_query"] = rewritten_query
elif options:
# 有候选：设置选择提示
state["answer"] = f"我不确定您指的是哪款产品。您是在询问以下产品吗：{'、'.join(options)}？"
else:
# 无匹配：设置无法识别提示
state["answer"] = "抱歉，我无法识别您询问的具体产品名称，请提供更准确的产品名称或型号。"
return state
def _backfill_history_item_names(self, history: List[Dict], item_names: List[str]):
"""将确认的商品名称回填到没有商品名的历史记录。"""
from knowledge.tools.mongo_history_utils import update_message_item_names
# 1. 获取要更新的消息 ID
ids_to_update = [
msg["_id"] for msg in history if not msg.get("item_names")
]
if not ids_to_update:
return
# 2. 更新内存中的 history 对象
for msg in history:
if not msg.get("item_names"):
msg["item_names"] = item_names
# 3. 批量更新 MongoDB
try:
update_message_item_names(ids_to_update, item_names)
except Exception as e:
self.logger.warning(f"回填历史商品名称失败: {e}")
def _write_history(
self, state: QueryGraphState, session_id: str,
rewritten_query: str, message_id: str,
):
"""将本轮对话写入历史（用户问题 + 助手回复）。"""
query = (state.get("original_query") or "").strip()
answer = (state.get("answer") or "").strip()
item_names = state.get("item_names") or []
# 更新用户问题消息
if query:
self._save_message(
session_id,
"user",
query,
rewritten_query=rewritten_query,
item_names=item_names,
message_id=message_id,
)
# 写入助手回复
if answer:
self._save_message(
session_id,
"assistant",
answer,
item_names=item_names,
)
# ================================================================== #
# 兼容 & 测试 #
# ================================================================== #
_node_instance = ItemNameConfirmNode()
def node_item_name_confirm(state: QueryGraphState) -> QueryGraphState:
"""兼容原有调用方式的入口函数。"""
return _node_instance(state)5. 测试入口
5.1 测试代码
在 item_name_confirm.py 文件末尾添加测试代码：
python
if __name__ == "__main__":
import uuid
from dotenv import load_dotenv
load_dotenv()
setup_logging()
print("=" * 60)
print("商品名称确认节点测试")
print("=" * 60)
# 1. 准备测试状态
test_state = {
"session_id": f"test_{uuid.uuid4().hex[:8]}",
"original_query": "你们店里那款苏伯尔RS-12数字万用表怎么测电压？",
"item_names": [],
"rewritten_query": "",
"answer": "",
"history": [],
"is_stream": False,
}
print(f"\n输入状态:")
print(f" session_id: {test_state['session_id']}")
print(f" original_query: {test_state['original_query']}")
print(f" item_names: {test_state['item_names']}")
print("-" * 60)
# 2. 执行节点
try:
result = node_item_name_confirm(test_state)
print("\n输出状态:")
print(f" item_names: {result.get('item_names')}")
print(f" rewritten_query: {result.get('rewritten_query')}")
if result.get("answer"):
print(f"\n拦截回复（流程中断）:")
print(f" {result.get('answer')}")
else:
print(f"\n确认成功，继续检索流程")
print(f"\n历史记录条数: {len(result.get('history', []))}")
except Exception as e:
print(f"\n执行失败: {e}")
import traceback
traceback.print_exc()
# 3. 测试多轮对话场景
print("\n" + "=" * 60)
print("测试多轮对话（代词指代）")
print("=" * 60)
# 假设第一轮已经确认了商品名
test_state_round2 = {
"session_id": test_state["session_id"], # 同一个会话
"original_query": "那它怎么换电池呢？",
"item_names": [],
"rewritten_query": "",
"answer": "",
"history": [],
"is_stream": False,
}
print(f"\n第二轮输入:")
print(f" original_query: {test_state_round2['original_query']}")
print("-" * 60)
try:
result2 = node_item_name_confirm(test_state_round2)
print("\n第二轮输出:")
print(f" item_names: {result2.get('item_names')}")
print(f" rewritten_query: {result2.get('rewritten_query')}")
if result2.get("answer"):
print(f"\n拦截回复: {result2.get('answer')}")
else:
print(f"\n代词已解析，确认成功")
except Exception as e:
print(f"\n执行失败: {e}")5.2 运行测试
bash
# 进入项目目录
cd knowledge
# 激活虚拟环境
source .venv/bin/activate # Linux/Mac
# 或
.venv\Scripts\activate # Windows
# 运行测试
python -m knowledge.processor.query_process.nodes.item_name_confirm5.3 预期输出
============================================================
商品名称确认节点测试
============================================================
输入状态:
session_id: test_a1b2c3d4
original_query: 你们店里那款苏伯尔RS-12数字万用表怎么测电压？
item_names: []
------------------------------------------------------------
2024-01-15 10:30:00 - query.item_name_confirm - INFO - --- item_name_confirm 开始 ---
2024-01-15 10:30:01 - query.item_name_confirm - INFO - --- item_name_confirm 完成 ---
输出状态:
item_names: ['苏伯尔RS-12数字万用表']
rewritten_query: 苏伯尔RS-12数字万用表怎么测电压？
确认成功，继续检索流程
历史记录条数: 1
============================================================
测试多轮对话（代词指代）
============================================================
第二轮输入:
original_query: 那它怎么换电池呢？
------------------------------------------------------------
第二轮输出:
item_names: ['苏伯尔RS-12数字万用表']
rewritten_query: 苏伯尔RS-12数字万用表怎么换电池？
代词已解析，确认成功6. 总结
6.1 节点功能概览
| 功能 | 说明 |
|---|---|
| 商品名提取 | 使用 LLM 从用户问题中提取候选商品名称 |
| 代词指代消解 | 结合历史对话，将"这个"、"它"等代词替换为具体名称 |
| 问题改写 | 生成包含商品名的独立完整问题 |
| 向量匹配 | 使用混合检索（稠密+稀疏）在商品库中搜索 |
| 评分对齐 | 根据相似度分数确定确认、候选或忽略 |
| 历史回填 | 将确认的商品名回填到之前的对话记录 |
| 流程分流 | 确认失败时设置 answer，中断检索流程 |
6.2 节点设计要点
1. 两阶段处理策略
LLM 提取（语义理解）→ 向量匹配（容错对齐）- LLM 擅长理解语义、处理代词
- 向量检索擅长模糊匹配、容忍错别字
2. 多阈值评分机制
高置信 (>0.63) → 直接确认
中置信 (0.6~0.63) → 用户选择
低置信 (<0.6) → 忽略- 避免错误匹配导致的检索偏差
- 保留用户选择的余地
3. 先保存后更新的消息管理
1. 先 insert 获取 message_id
2. 处理完成后 update 补充字段- 确保消息不丢失
- 支持字段渐进式补充
4. 历史回填设计
新确认的商品名 → 回填到之前缺失商品名的消息- 保持历史记录的完整性
- 便于后续会话的上下文理解
企业痛点映射
| 痛点 | 传统方案 | AI Agent 商品名确认方案 | 效率提升 |
|---|---|---|---|
| 用户问题中商品名模糊 | 人工回问"您指的是哪款产品" | LLM 提取 + 向量模糊匹配 | 自动确认率 ~80% |
| 多轮对话代词指代丢失 | 不明白"它"指什么 | LLM 代词消解 + 查询改写 | 指代准确率 ~90% |
| 错别字/不完整匹配失败 | 精确匹配搜不到结果 | 混合检索容错（稠密+稀疏） | 模糊召回率提升 ~60% |
| 低置信匹配返回错误结果 | 误导用户 | 多阈值评分（>0.63确认） | 错误结果降为 ~0% |
Remote & Agent 应用场景价值
Remote 场景价值：依赖 MongoDB / Milvus / LLM API 三个远程服务，通过
.env配置即可运行。历史回填确保分布式环境下多轮对话上下文完整。Agent 落地场景：可封装为"商品名确认 Agent"——三种输出（confirm/options/none）可直接映射到 Agent 的决策路径（继续/询问/拒绝）。
Git Commit 对应
本节商品名确认节点对应的提交记录（参考值，以实际版本为准）：
<待补充 — 建议搜索 "item_name_confirm.py" 相关提交>bash
cd shopkeeper_brain
git log --oneline --all -- knowledge/processor/query_process/nodes/item_name_confirm.py