1
import json, math, sys, urllib.request
2

3
URL = 'http://127.0.0.1:11434/v1/embeddings'
4

5
# (anchor, positive=同义改写, hard negative=同话题但语义/极性不同)
6
# 六个类别各若干组，实际用了 60 组，这里节选
7
T = [
8
    # 情感极性反转
9
    ('这家拉面真的绝了', '那面馆味道超好吃', '这家拉面难吃到想吐'),
10
    ('他最近心情不太好', '那家伙这阵子情绪低落', '他最近心情特别好'),
11
    ('这游戏优化稀烂', '这游戏帧数掉得离谱', '这游戏优化好得离谱'),
12
    # 主客体互换
13
    ('我把他删了', '已经把那人拉黑了', '他把我删了'),
14
    ('他欠我钱', '那人还没还我钱', '我欠他钱'),
15
    ('我请他吃饭', '这顿我买单请客', '他请我吃饭'),
16
    # 否定与肯定
17
    ('我不去了', '这次我就不参加了', '我去'),
18
    ('这个方案不可行', '这路子走不通', '这个方案完全可行'),
19
    # 时间/数量
20
    ('这个bug我调了三小时', '排查这问题花了一下午', '这功能我十分钟就写完了'),
21
    ('等了半天才上菜', '上菜慢得离谱', '菜上得特别快'),
22
    # 同话题不同结论
23
    ('我考研上岸了', '研究生考上了', '我考研没考上'),
24
    ('面试通过了', '拿到offer了', '面试挂了'),
25
    # 细微语义
26
    ('这药一天吃三次', '每日三次服用', '这药三天吃一次'),
27
    ('从北京飞上海', '北京出发去上海', '从上海飞北京'),
28
    ('保修期三年', '三年内免费维修', '保修期三个月'),
29
]
30

31
def embed(model, texts, timeout=1200):
32
    req = urllib.request.Request(
33
        URL,
34
        data=json.dumps({'model': model, 'input': texts}).encode(),
35
        headers={'Content-Type': 'application/json'})
36
    return [x['embedding'] for x in json.loads(urllib.request.urlopen(req, timeout=timeout).read())['data']]
37

38
def cos(a, b):
39
    return sum(x * y for x, y in zip(a, b)) / (
40
        math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b)))
41

42
model = sys.argv[1]
43
flat = [t for tri in T for t in tri]
44
vs = []
45
for i in range(0, len(flat), 30):        # 分批，避免单请求过大
46
    vs.extend(embed(model, flat[i:i + 30]))
47

48
hit, margins, fails = 0, [], []
49
for i, tri in enumerate(T):
50
    a, p, n = vs[3 * i], vs[3 * i + 1], vs[3 * i + 2]
51
    sp, sn = cos(a, p), cos(a, n)
52
    if sp > sn:
53
        hit += 1
54
    else:
55
        fails.append(tri[0])
56
    margins.append(sp - sn)
57

58
print(json.dumps({
59
    'model': model, 'dim': len(vs[0]),
60
    'acc': f'{hit}/{len(T)}',
61
    'avg_margin': round(sum(margins) / len(margins), 3),
62
    'fails': fails[:6],
63
}, ensure_ascii=False))

1
HEAD = '今天群里聊起了最近玩的那款开放世界游戏，大家讨论了画面表现和优化情况，' * 50
2
T1 = '后来话题转到了考研，有人说今年报名人数又涨了，竞争压力很大。' * 10
3
T2 = '后来话题转到了房价，有人说楼市终于开始松动，一线城市也在跌。' * 10
4

5
a = HEAD + T1
6
p = HEAD + T1.replace('考研', '研究生考试').replace('报名', '报考')
7
n = HEAD + T2
8

9
v = embed(model, [a, p, n])
10
sp, sn = cos(v[0], v[1]), cos(v[0], v[2])
11
print(f'同尾={sp:.4f} 异尾={sn:.4f} 差={sp-sn:+.4f}',
12
      '看到了尾部' if sp - sn > 0.01 else '★尾部丢失(截断)')

1
import concurrent.futures as cf, time
2

3
def one(model, text):
4
    req = urllib.request.Request(
5
        URL,
6
        data=json.dumps({'model': model, 'input': text}).encode(),
7
        headers={'Content-Type': 'application/json'})
8
    s = time.perf_counter()
9
    urllib.request.urlopen(req, timeout=900).read()
10
    return time.perf_counter() - s
11

12
one(model, '预热')                                        # 先加载模型
13
serial = sorted(one(model, TEXTS[i % 10]) for i in range(10))
14
print(f'单条中位={serial[5]*1000:.0f}ms p90={serial[9]*1000:.0f}ms')
15

16
t0 = time.perf_counter()
17
with cf.ThreadPoolExecutor(max_workers=5) as ex:          # 对应 max_concurrent
18
    list(ex.map(lambda i: one(model, TEXTS[i % 10] + f' 第{i}条'), range(32)))
19
print(f'32条@并发5 = {time.perf_counter()-t0:.2f}s')

1
base = '今天群里聊起了最近玩的那款开放世界游戏，大家讨论了画面表现和优化情况，'
2
for mult in (1, 4, 12, 30, 60):
3
    txt = base * mult
4
    ts = [one(model, txt) for _ in range(3)]
5
    print(f'{len(txt):5d}字 -> 中位 {sorted(ts)[1]*1000:7.0f}ms')

1
URL = 'https://openrouter.ai/api/v1/embeddings'
2
HEADERS = {'Content-Type': 'application/json',
3
           'Authorization': 'Bearer ' + os.environ['OPENROUTER_API_KEY']}   # 别写进源码

1
noise  = cos(embed(m, T), embed(m, T))                       # 同参数两次 = 噪声底
2
effect = cos(embed(m, T), embed(m, T, task_type='...'))      # 带参数
3
# effect 约等于 noise，说明参数被丢了，这个模型的分数不能按"官方最佳用法"解读