第5章 学生Notebook：美食语义匹配器¶
同样的字不等于同样的意思
本Notebook完成以下任务：
- 用共同字基线匹配菜品
- 用嵌入向量 + 余弦相似度匹配菜品
- 对比两种方法，分析各自的失败情况
- 修改一个变量（阈值或数据集），观察变化
准备：确认 data/candidates.json 存在。
1. 加载数据¶
读取菜品和查询数据。
import json
import numpy as np
from pathlib import Path
# 加载数据
data_path = Path('../data/candidates.json')
with open(data_path, 'r', encoding='utf-8') as f:
data = json.load(f)
dishes = data['dishes']
queries = data['queries']
print(f'菜品数量: {len(dishes)}')
print(f'查询数量: {len(queries)}')
print()
print('前3道菜品:')
for d in dishes[:3]:
print(f' {d["id"]} {d["name"]}: {d["description"]}')
print()
print('第1条查询:')
q = queries[0]
print(f' {q["id"]}: {q["text"]}')
print(f' 相关: {q["relevant"]}')
print(f' 不相关: {q["irrelevant"]}')
def char_overlap_score(query, text):
"""计算两段文字的共同字数量"""
query_chars = set(query)
text_chars = set(text)
overlap = query_chars & text_chars
# 去掉标点空格
overlap = {c for c in overlap if c.strip()}
return len(overlap)
def char_overlap_ranking(query_text, dishes):
"""对菜品按共同字数量排序"""
scores = []
for d in dishes:
combined = d['name'] + d['description']
score = char_overlap_score(query_text, combined)
scores.append((d['id'], d['name'], score))
scores.sort(key=lambda x: x[2], reverse=True)
return scores
# 对第1条查询运行
q = queries[0]
print(f'查询: {q["text"]}')
print(f'预期相关: {q["relevant"]}')
print()
print('共同字排序 Top-5:')
ranking = char_overlap_ranking(q['text'], dishes)
for i, (did, name, score) in enumerate(ranking[:5]):
marker = '✓' if did in q['relevant'] else ' '
print(f' {i+1}. [{marker}] {did} {name} (共同字: {score})')
思考：对Q02"想吃点酸的开胃菜"运行共同字方法，观察结果。
在下方写下你的预测：共同字方法会排前3的是什么？
# 运行 Q02
q2 = queries[1]
print(f'查询: {q2["text"]}')
print(f'预期相关: {q2["relevant"]}')
print()
ranking2 = char_overlap_ranking(q2['text'], dishes)
print('共同字排序 Top-5:')
for i, (did, name, score) in enumerate(ranking2[:5]):
marker = '✓' if did in q2['relevant'] else ' '
print(f' {i+1}. [{marker}] {did} {name} (共同字: {score})')
# 在下方写下你的观察
# 我的观察：
def cosine_similarity(a, b):
"""计算两个向量的余弦相似度"""
a = np.array(a, dtype=float)
b = np.array(b, dtype=float)
dot_product = np.dot(a, b)
norm_a = np.linalg.norm(a)
norm_b = np.linalg.norm(b)
if norm_a == 0 or norm_b == 0:
return 0.0
return dot_product / (norm_a * norm_b)
# 演示：两个已知向量的余弦相似度
v1 = [1, 0, 0] # 指向x轴
v2 = [1, 0, 0] # 相同方向
v3 = [0, 1, 0] # 垂直方向
print(f'相同方向: cos = {cosine_similarity(v1, v2):.4f}') # 应接近 1.0
print(f'垂直方向: cos = {cosine_similarity(v1, v3):.4f}') # 应为 0.0
import os
import hashlib
EMBED_DIM = 64 # 预计算向量的维度（简化演示用）
def get_embedding_local_model(text):
"""尝试通过本地 Ollama 获取嵌入向量"""
try:
import urllib.request
url = 'http://localhost:11434/api/embed'
payload = json.dumps({'model': 'nomic-embed-text', 'input': text}).encode()
req = urllib.request.Request(url, data=payload, headers={'Content-Type': 'application/json'})
with urllib.request.urlopen(req, timeout=10) as resp:
result = json.loads(resp.read())
return result['embeddings'][0]
except Exception:
return None
def get_embedding_precomputed(text):
"""离线兜底：用文本哈希生成确定性伪向量"""
h = hashlib.sha256(text.encode('utf-8')).digest()
rng = np.random.RandomState(int.from_bytes(h[:4], 'big'))
vec = rng.randn(EMBED_DIM).astype(float)
vec = vec / np.linalg.norm(vec) # 归一化
return vec.tolist()
def get_embedding(text):
"""自动选择嵌入路径"""
vec = get_embedding_local_model(text)
if vec is not None:
return vec, 'local'
return get_embedding_precomputed(text), 'precomputed'
# 测试嵌入路径
test_vec, source = get_embedding('测试文本')
print(f'嵌入路径: {source}')
print(f'向量维度: {len(test_vec)}')
print(f'前5个值: {test_vec[:5]}')
计算所有菜品的嵌入并排序¶
运行：对每条查询，用余弦相似度排序菜品。
def embedding_ranking(query_text, dishes):
"""对菜品按嵌入余弦相似度排序"""
q_vec, source = get_embedding(query_text)
scores = []
for d in dishes:
combined = d['name'] + ' ' + d['description']
d_vec, _ = get_embedding(combined)
sim = cosine_similarity(q_vec, d_vec)
scores.append((d['id'], d['name'], sim))
scores.sort(key=lambda x: x[2], reverse=True)
return scores, source
# 对 Q01 运行
q = queries[0]
ranking_emb, source = embedding_ranking(q['text'], dishes)
print(f'查询: {q["text"]}')
print(f'嵌入路径: {source}')
print(f'预期相关: {q["relevant"]}')
print()
print('嵌入排序 Top-5:')
for i, (did, name, sim) in enumerate(ranking_emb[:5]):
marker = '✓' if did in q['relevant'] else ' '
print(f' {i+1}. [{marker}] {did} {name} (相似度: {sim:.4f})')
4. 两种方法全面对照¶
运行：对全部6条查询，对比两种方法。
def evaluate_ranking(ranking, relevant_ids, top_k=5):
"""计算Top-K中命中相关菜品的数量"""
top_ids = [r[0] for r in ranking[:top_k]]
hits = sum(1 for did in top_ids if did in relevant_ids)
return hits
print(f'{"查询":<20} {"共同字命中":<10} {"嵌入命中":<10} {"差异"}')
print('-' * 60)
for q in queries:
char_rank = char_overlap_ranking(q['text'], dishes)
emb_rank, _ = embedding_ranking(q['text'], dishes)
char_hits = evaluate_ranking(char_rank, q['relevant'])
emb_hits = evaluate_ranking(emb_rank, q['relevant'])
diff = emb_hits - char_hits
diff_str = f'+{diff}' if diff > 0 else str(diff)
print(f'{q["text"]:<20} {char_hits:<10} {emb_hits:<10} {diff_str}')
思考：
- 哪些查询嵌入方法明显优于共同字？为什么？
- 有没有共同字方法更好或持平的查询？为什么？
- 有没有两种方法都失败的查询？为什么？
在下方写下你的分析。
# 我的分析：
#
# 嵌入明显优于共同字的查询：
#
# 两种方法都失败的查询：
#
5. 可视化：排序对照¶
运行：生成两种方法的排序对照图。
try:
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
HAS_MPL = True
except ImportError:
HAS_MPL = False
print('Matplotlib 未安装，跳过可视化。核心计算不受影响。')
if HAS_MPL:
fig, axes = plt.subplots(2, 3, figsize=(15, 10))
fig.suptitle('共同字 vs 嵌入向量：菜品排序对照', fontsize=14)
for idx, q in enumerate(queries):
ax = axes[idx // 3][idx % 3]
char_rank = char_overlap_ranking(q['text'], dishes)
emb_rank, _ = embedding_ranking(q['text'], dishes)
top5_char = char_rank[:5]
top5_emb = emb_rank[:5]
names = [d['name'] for d in dishes]
char_scores = [0] * len(dishes)
emb_scores = [0] * len(dishes)
for rank, (did, _, score) in enumerate(char_rank):
i = int(did[1:]) - 1
char_scores[i] = len(dishes) - rank
for rank, (did, _, score) in enumerate(emb_rank):
i = int(did[1:]) - 1
emb_scores[i] = len(dishes) - rank
colors = ['green' if did in q['relevant'] else 'gray' for did, _, _ in char_rank]
ax.barh(range(min(8, len(dishes))),
[char_scores[i] for i in range(min(8, len(dishes)))],
alpha=0.5, label='共同字', color='steelblue')
ax.set_title(f'{q["id"]}: {q["text"][:12]}...', fontsize=10)
ax.set_yticks(range(min(8, len(dishes))))
ax.set_yticklabels([dishes[i]['name'] for i in range(min(8, len(dishes)))], fontsize=8)
ax.invert_yaxis()
plt.tight_layout()
plt.savefig('../outputs/ch05_comparison.png', dpi=100, bbox_inches='tight')
print('对照图已保存到 outputs/ch05_comparison.png')
plt.show()
6. B档修改：修改一个变量¶
从以下两项中选择一项修改，修改前先写预测。
选项A：新增3道菜品到数据集
- 至少一道能被"想吃点酸的"匹配到
- 至少一道是"不要辣的"应该排除的
选项B：添加阈值判断
- 当最高相似度低于阈值时，输出"请人工确认"
- 测试不同阈值（0.3, 0.5, 0.7）的影响
# 我选择：选项___
#
# 修改前预测：
#
# === 选项A：新增菜品 ===
# 取消下面的注释并修改
# new_dishes = [
# {"id": "D21", "name": "___", "description": "___", "category": "___"},
# {"id": "D22", "name": "___", "description": "___", "category": "___"},
# {"id": "D23", "name": "___", "description": "___", "category": "___"},
# ]
# dishes_extended = dishes + new_dishes
# === 选项B：阈值判断 ===
# 取消下面的注释并修改
# THRESHOLD = 0.5 # 修改这个值
# for q in queries:
# emb_rank, _ = embedding_ranking(q['text'], dishes)
# top_sim = emb_rank[0][2]
# status = '推荐' if top_sim >= THRESHOLD else '请人工确认'
# print(f'{q["text"]}: 最高相似度={top_sim:.4f} → {status}')
7. 验收¶
运行下面的检查，确认你完成了所有必做项。
checks = {
'运行了共同字基线': True, # 你已经运行了第2节
'运行了嵌入排序': True, # 你已经运行了第3-4节
'写了对比分析': False, # 检查第4节的分析单元格是否有内容
'完成了B档修改': False, # 检查第6节是否有修改
}
print('验收清单:')
for check, status in checks.items():
print(f' [{"✓" if status else " "}] {check}')
print()
print('请手动把 False 改为 True，确认你完成了对应项目。')
print()
print('关键概念检查:')
print(' 1. 余弦相似度的值域是 ___ 到 ___')
print(' 2. 嵌入向量的作用是 ___')
print(' 3. 共同字方法的局限是 ___')
print(' 4. 嵌入方法也可能失败的情况是 ___')