#!/usr/bin/env python3 # -*- coding: utf-8 -*- """本地 qwen3:8b vs DeepSeek：活动记录四项抽取一致性对比 样本：数据/daily_activities_mcp.json（正文 300–1200 字）随机 200 条 输出：销售分析/活动抽取_本地8b_vs_DeepSeek_2026-10-26.xlsx 说明：没有人工标注，因此统计的是「完整率 / 格式合规 / 两模型一致率 / 差异清单」， 不是绝对准确率；差异清单用于人工抽检。 """ import json import os import random import re import time import urllib.request from concurrent.futures import ThreadPoolExecutor import openpyxl import yaml from openpyxl.styles import Alignment, Font, PatternFill from openpyxl.utils import get_column_letter ROOT = '/Users/liuxinyuan/Desktop/Hermes输出-工作类' SRC = os.path.join(ROOT, '数据', 'daily_activities_mcp.json') OUT = os.path.join(ROOT, '销售分析', '活动抽取_本地8b_vs_DeepSeek_2026-10-26.xlsx') N = 200 OLLAMA = 'http://127.0.0.1:11434/api/generate' KEYS = ['客户', '项目或货号', '卡点', '下一步动作'] PROMPT = ('你是销售记录信息抽取器。逐字阅读记录，抽取：客户、项目或货号、卡点、下一步动作。' '必须从原文摘录关键词，不得概括、不得编造；原文确实没有该信息才写“无”。' '只输出 JSON，键名固定：客户/项目或货号/卡点/下一步动作。\n\n记录：\n') def cfg_key(): c = yaml.safe_load(open(os.path.expanduser('~/.hermes/profiles/zone-b/config.yaml'), encoding='utf-8')) return c['model']['api_key'], c['model'].get('base_url', 'https://api.deepseek.com'), c['model'].get('default', 'deepseek-flash') def parse_json(s): s = re.sub(r'^```(json)?|```$', '', s.strip(), flags=re.M).strip() obj = None try: obj = json.loads(s) except Exception: m = re.search(r'\{.*\}|\[.*\]', s, re.S) if m: try: obj = json.loads(m.group(0)) except Exception: obj = None if isinstance(obj, dict): return obj if isinstance(obj, list): # 模型有时输出 [{...}] 或 ["客户...", ...] merged = {} for it in obj: if isinstance(it, dict): merged.update(it) if merged: return merged if all(isinstance(it, str) for it in obj): return dict(zip(KEYS, obj)) return {} def run_local(text): req = {'model': 'qwen3:8b', 'prompt': PROMPT + text, 'stream': False, 'think': False, 'options': {'temperature': 0, 'num_predict': 320}} t = time.time() r = json.load(urllib.request.urlopen(urllib.request.Request( OLLAMA, data=json.dumps(req, ensure_ascii=False).encode(), headers={'Content-Type': 'application/json'}), timeout=300)) return parse_json(r.get('response', '')), time.time() - t def run_deepseek(text, key, base, model): body = json.dumps({'model': model, 'temperature': 0, 'response_format': {'type': 'json_object'}, 'messages': [{'role': 'user', 'content': PROMPT + text}]}).encode() req = urllib.request.Request(base.rstrip('/') + '/chat/completions', data=body, headers={'Content-Type': 'application/json', 'Authorization': 'Bearer ' + key}) t = time.time() r = json.load(urllib.request.urlopen(req, timeout=120)) return parse_json(r['choices'][0]['message']['content']), time.time() - t def sim(a, b): import difflib a, b = str(a or '').strip(), str(b or '').strip() if not a and not b: return 1.0 return difflib.SequenceMatcher(None, a, b).ratio() def main(): items = json.load(open(SRC, encoding='utf-8'))['items'] pool = [x for x in items if 300 <= len(str(x.get('field_b99055') or '')) <= 1200] random.seed(42) sample = random.sample(pool, min(N, len(pool))) print('样本池 %d ｜ 抽取 %d' % (len(pool), len(sample))) key, base, model = cfg_key() cache_path = os.path.join(ROOT, '数据', 'activity_extract_cache.json') cache = json.load(open(cache_path, encoding='utf-8')) if os.path.exists(cache_path) else {} rows = [] for i, x in enumerate(sample, 1): text = str(x.get('field_b99055'))[:800] wid = str(x.get('work_item_id')) if wid in cache and cache[wid].get('local'): local, tl = cache[wid]['local'], cache[wid].get('tl', 0) else: try: local, tl = run_local(text) except Exception as e: local, tl = {'_err': str(e)}, 0 cache[wid] = {'local': local, 'tl': tl} if i % 5 == 0: json.dump(cache, open(cache_path, 'w', encoding='utf-8'), ensure_ascii=False) rows.append({'id': i, 'wid': x.get('work_item_id'), 'name': x.get('name'), 'owner': (x.get('owner') or {}).get('name_cn'), 'date': x.get('start_time'), 'text': text, 'local': local, 'tl': tl}) if i % 20 == 0: print(' 本地 %d/%d' % (i, len(sample)), flush=True) def ds(r): wid = str(r['wid']) if wid in cache and cache[wid].get('ds'): r['ds'], r['td'] = cache[wid]['ds'], cache[wid].get('td', 0) return r try: d, t = run_deepseek(r['text'], key, base, model) r['ds'], r['td'] = d, t except Exception as e: r['ds'], r['td'] = {'_err': str(e)}, 0 cache.setdefault(wid, {}).update({'ds': r['ds'], 'td': r['td']}) return r with ThreadPoolExecutor(max_workers=6) as ex: list(ex.map(ds, rows)) json.dump(cache, open(cache_path, 'w', encoding='utf-8'), ensure_ascii=False) print('DeepSeek 完成') stat = {'local_full': 0, 'ds_full': 0, 'agree': 0, 'local_t': 0.0, 'ds_t': 0.0} for r in rows: if all(str(r['local'].get(k, '')).strip() not in ('', '未提及', '无', 'None') for k in KEYS): stat['local_full'] += 1 if all(str(r['ds'].get(k, '')).strip() not in ('', '未提及', '无', 'None') for k in KEYS): stat['ds_full'] += 1 sc = sum(sim(r['local'].get(k), r['ds'].get(k)) for k in KEYS) / len(KEYS) r['score'] = sc if sc >= 0.75: stat['agree'] += 1 stat['local_t'] += r['tl'] stat['ds_t'] += r['td'] n = len(rows) wb = openpyxl.Workbook() ws = wb.active ws.title = '汇总' ws.append(['指标', '本地 qwen3:8b', 'DeepSeek']) for k, lab in (('full', '四项完整率'), ('t', '平均耗时(秒)')): a = stat['local_full'] / n * 100 if k == 'full' else stat['local_t'] / n b = stat['ds_full'] / n * 100 if k == 'full' else stat['ds_t'] / n ws.append([lab, round(a, 1), round(b, 1)]) ws.append(['两模型字段平均一致率(%)', round(sum(r['score'] for r in rows) / n * 100, 1), '']) ws.append(['一致（≥0.75）条数', stat['agree'], '/%d' % n]) ws.append(['说明', '无人工标注，此处为一致率而非准确率；差异清单供抽检', '']) for c in ws[1]: c.font = Font(color='FFFFFF', bold=True) c.fill = PatternFill('solid', fgColor='1F4E79') for i, w in enumerate([30, 22, 22], 1): ws.column_dimensions[get_column_letter(i)].width = w ws2 = wb.create_sheet('差异清单') ws2.append(['#', '活动名称', '销售', '日期', '字段', '本地8b', 'DeepSeek', '一致度']) for r in sorted(rows, key=lambda z: z['score'])[:40]: for k in KEYS: s = sim(r['local'].get(k), r['ds'].get(k)) if s < 0.75: ws2.append([r['id'], r['name'], r['owner'], r['date'], k, str(r['local'].get(k))[:200], str(r['ds'].get(k))[:200], round(s, 2)]) for c in ws2[1]: c.font = Font(color='FFFFFF', bold=True) c.fill = PatternFill('solid', fgColor='1F4E79') for i, w in enumerate([5, 30, 8, 11, 12, 52, 52, 8], 1): ws2.column_dimensions[get_column_letter(i)].width = w ws3 = wb.create_sheet('明细') ws3.append(['#', '活动名称', '销售', '日期', '原文', '本地8b JSON', 'DeepSeek JSON', '一致度']) for r in rows: ws3.append([r['id'], r['name'], r['owner'], r['date'], r['text'][:500], json.dumps(r['local'], ensure_ascii=False)[:500], json.dumps(r['ds'], ensure_ascii=False)[:500], round(r['score'], 2)]) for c in ws3[1]: c.font = Font(color='FFFFFF', bold=True) c.fill = PatternFill('solid', fgColor='1F4E79') for i, w in enumerate([5, 30, 8, 11, 60, 50, 50, 8], 1): ws3.column_dimensions[get_column_letter(i)].width = w for s in (ws2, ws3): s.freeze_panes = 'A2' for row in s.iter_rows(min_row=2): for c in row: c.alignment = Alignment(vertical='top', wrap_text=True) wb.save(OUT) json.dump(rows, open(os.path.join(ROOT, '数据', 'activity_extract_compare.json'), 'w', encoding='utf-8'), ensure_ascii=False) print('✅', OUT) print('完整率 local %.1f%% / ds %.1f%% ｜ 一致 %.1f%%' % ( stat['local_full'] / n * 100, stat['ds_full'] / n * 100, stat['agree'] / n * 100)) if __name__ == '__main__': main()