fix(asr): restore computer terminology bias

Rebuild the bundled language model and runtime source filtering around a dedicated computer-term corpus so technical dictation keeps domain coverage.
This commit is contained in:
Rocky
2026-08-15 09:07:15 +08:00
parent 704ac2428c
commit 13214e6601
14 changed files with 10625 additions and 3134 deletions
+130
View File
@@ -0,0 +1,130 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Build OSGKeyboard domain ASR lexicon v1 from the computer-terms scel source.
Sources:
Local 计算机词汇大全【官方推荐】.scel (IT / computer vocabulary only)
Casual network slang and Sogou popular-word dumps are intentionally excluded —
they dilute custom LM phrase biasing without improving domain ASR accuracy.
Output:
OSGKeyboard/Resources/CustomLanguageModel/v1/phrases.tsv
OSGKeyboard/Resources/CustomLanguageModel/v1/manifest.json
The compiled .bin asset is exported separately to
OSGKeyboard/Resources/CustomLanguageModel/v1/ via export_clm.swift.
"""
from __future__ import annotations
import argparse
import json
import sys
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from scel_parser import get_scel_info, parse_scel_file
REPO_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_OUTPUT_DIR = REPO_ROOT / "OSGKeyboard/Resources/CustomLanguageModel/v1"
DEFAULT_COMPUTER_SCEL = Path("/Users/rocky/Downloads/计算机词汇大全【官方推荐】.scel")
SOURCE_KEY = "computer_terms"
SOURCE_LABEL = "计算机词汇大全【官方推荐】"
SOURCE_WEIGHT = 5
@dataclass
class LexiconEntry:
word: str
pinyin: str
source: str
weight: int
def merge_entries(entries: list[tuple[str, str]]) -> list[LexiconEntry]:
merged: dict[str, LexiconEntry] = {}
for word, pinyin in entries:
current = merged.get(word)
candidate = LexiconEntry(
word=word,
pinyin=pinyin,
source=SOURCE_KEY,
weight=SOURCE_WEIGHT,
)
if current is None:
merged[word] = candidate
elif not current.pinyin and pinyin:
merged[word] = candidate
return sorted(merged.values(), key=lambda item: item.word)
def write_outputs(entries: list[LexiconEntry], output_dir: Path, raw_count: int) -> None:
output_dir.mkdir(parents=True, exist_ok=True)
phrases_path = output_dir / "phrases.tsv"
with phrases_path.open("w", encoding="utf-8") as handle:
handle.write("word\tpinyin\tsource\tweight\n")
for entry in entries:
handle.write(f"{entry.word}\t{entry.pinyin}\t{entry.source}\t{entry.weight}\n")
manifest = {
"version": "v1",
"generated_at": datetime.now(timezone.utc).isoformat(),
"locale": "zh-Hans",
"entry_count": len(entries),
"sources": [
{
"key": SOURCE_KEY,
"label": SOURCE_LABEL,
"weight": SOURCE_WEIGHT,
"raw_count": raw_count,
}
],
"notes": [
"Domain-specific computer/IT vocabulary only; casual network slang removed.",
"PhraseCount weights map to SFCustomLanguageModelData relative frequencies.",
"Merged with ai-tech-brands seed at export time for the final .bin asset.",
],
"files": {
"phrases": phrases_path.name,
},
}
manifest_path = output_dir / "manifest.json"
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def build(*, computer_scel: Path, output_dir: Path) -> int:
if not computer_scel.exists():
print(f"Missing computer scel: {computer_scel}", file=sys.stderr)
return 1
computer_info = get_scel_info(computer_scel)
print(f"Computer dict: {computer_info.name} ({computer_info.word_count} header count)")
raw_entries = parse_scel_file(computer_scel)
merged = merge_entries(raw_entries)
write_outputs(merged, output_dir, raw_count=len(raw_entries))
print(f"Raw count: {len(raw_entries)}")
print(f"Merged unique entries: {len(merged)}")
print(f"Wrote {output_dir / 'phrases.tsv'}")
print(f"Wrote {output_dir / 'manifest.json'}")
return 0
def main() -> int:
parser = argparse.ArgumentParser(description="Build OSGKeyboard domain ASR lexicon v1")
parser.add_argument("--computer-scel", type=Path, default=DEFAULT_COMPUTER_SCEL)
parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR)
args = parser.parse_args()
return build(computer_scel=args.computer_scel, output_dir=args.output_dir)
if __name__ == "__main__":
raise SystemExit(main())
+20 -53
View File
@@ -3,9 +3,7 @@
// export_clm.swift
// OSGKeyboard · offline SFCustomLanguageModelData exporter (macOS 14+)
//
// Reads the project-curated AI/tech TSV and writes a .bin training asset via
// Speech framework. The normalized four-column TSV beside the binary also
// powers the Mac runtime bias index.
// Reads merged phrase TSVs and writes a .bin training asset via Speech framework.
// Usage:
// swift Scripts/lexicon/export_clm.swift
// swift Scripts/lexicon/export_clm.swift --max-entries 30000
@@ -17,6 +15,7 @@ import Speech
// MARK: - CLI
struct CLIOptions {
var domainTSV: URL
var aiTechTSV: URL
var outputBin: URL
var localeID: String
@@ -30,6 +29,9 @@ struct CLIOptions {
.deletingLastPathComponent()
.deletingLastPathComponent()
var domain = repoRoot.appendingPathComponent(
"OSGKeyboard/Resources/CustomLanguageModel/v1/phrases.tsv"
)
var aiTech = repoRoot.appendingPathComponent(
"OSGKeyboard/Resources/CustomLanguageModel/ai-tech-brands/v1/phrases.tsv"
)
@@ -38,12 +40,14 @@ struct CLIOptions {
)
var localeID = "zh_CN"
var modelID = "com.osgkeyboard.custom-lm.v1"
var modelVersion = "1.0.1"
var modelVersion = "1.0.0"
var maxEntries: Int?
var iterator = CommandLine.arguments.dropFirst().makeIterator()
while let flag = iterator.next() {
switch flag {
case "--domain-tsv", "--sogou-tsv":
domain = URL(fileURLWithPath: iterator.next() ?? "")
case "--ai-tech-tsv":
aiTech = URL(fileURLWithPath: iterator.next() ?? "")
case "--output":
@@ -67,6 +71,7 @@ struct CLIOptions {
}
return CLIOptions(
domainTSV: domain,
aiTechTSV: aiTech,
outputBin: output,
localeID: localeID,
@@ -81,7 +86,8 @@ struct CLIOptions {
export_clm.swift — build SFCustomLanguageModelData .bin on macOS
Options:
--ai-tech-tsv <path> Project-curated AI/tech phrases TSV
--domain-tsv <path> Domain phrases TSV (computer/IT terms)
--ai-tech-tsv <path> AI/tech seed phrases TSV
--output <path> Output .bin path
--locale <id> Locale identifier (default: zh_CN)
--identifier <id> Custom LM identifier
@@ -96,7 +102,6 @@ struct CLIOptions {
struct PhraseEntry: Hashable {
let phrase: String
let pinyin: String
let weight: Int
let source: String
}
@@ -121,7 +126,7 @@ enum TSVLoader {
}
// Formats:
// normalized: word, pinyin, source, weight
// domain: word, pinyin, source, weight
// ai-tech: word, pinyin, source, category, weight, canonical
let weight: Int
if parts.count >= 6, let parsed = Int(parts[4]) {
@@ -132,16 +137,8 @@ enum TSVLoader {
weight = 1
}
let pinyin = parts.count >= 2 ? parts[1] : ""
let source = parts.count >= 3 ? parts[2] : sourceLabel
entries.append(
PhraseEntry(
phrase: word,
pinyin: pinyin,
weight: max(1, weight),
source: source
)
)
entries.append(PhraseEntry(phrase: word, weight: max(1, weight), source: source))
}
return entries
@@ -174,13 +171,17 @@ enum ExportCLM {
let options = CLIOptions.parse()
let fm = FileManager.default
guard fm.fileExists(atPath: options.domainTSV.path) else {
throw ExportError.missingInput(options.domainTSV.path)
}
guard fm.fileExists(atPath: options.aiTechTSV.path) else {
throw ExportError.missingInput(options.aiTechTSV.path)
}
fputs("Loading phrases…\n", stderr)
let domain = try TSVLoader.load(from: options.domainTSV, sourceLabel: "computer_terms")
let aiTech = try TSVLoader.load(from: options.aiTechTSV, sourceLabel: "ai_tech_seed")
var merged = TSVLoader.merge([aiTech])
var merged = TSVLoader.merge([domain, aiTech])
if let cap = options.maxEntries, merged.count > cap {
merged = Array(merged.prefix(cap))
@@ -188,7 +189,7 @@ enum ExportCLM {
}
fputs(
"Loaded \(merged.count) unique project-curated AI/tech phrases\n",
"Merged \(merged.count) unique phrases (domain=\(domain.count), ai-tech=\(aiTech.count))\n",
stderr
)
fputs("Locale=\(options.localeID) identifier=\(options.modelID) version=\(options.modelVersion)\n", stderr)
@@ -217,15 +218,6 @@ enum ExportCLM {
try fm.removeItem(at: outputURL)
}
let phrasesURL = parent.appendingPathComponent("phrases.tsv")
let normalizedLines = merged.map { entry in
"\(entry.phrase)\t\(entry.pinyin)\t\(entry.source)\t\(entry.weight)"
}
let normalizedTSV = (["word\tpinyin\tsource\tweight"] + normalizedLines)
.joined(separator: "\n") + "\n"
try normalizedTSV.write(to: phrasesURL, atomically: true, encoding: .utf8)
fputs("Wrote \(phrasesURL.path)\n", stderr)
fputs("Exporting to \(outputURL.path)\n", stderr)
try await data.export(to: outputURL)
@@ -244,6 +236,7 @@ enum ExportCLM {
"version": options.modelVersion,
"phrase_count": merged.count,
"sources": [
"computer_terms": domain.count,
"ai_tech_seed": aiTech.count,
],
"bin_file": outputURL.lastPathComponent,
@@ -253,32 +246,6 @@ enum ExportCLM {
let manifestData = try JSONSerialization.data(withJSONObject: manifest, options: [.prettyPrinted, .sortedKeys])
try manifestData.write(to: manifestURL)
fputs("Wrote \(manifestURL.path)\n", stderr)
let sourceManifestURL = parent.appendingPathComponent("manifest.json")
let sourceManifest: [String: Any] = [
"version": "v1",
"generated_at": ISO8601DateFormatter().string(from: Date()),
"locale": "zh-Hans",
"entry_count": merged.count,
"sources": [[
"key": "ai_tech_seed",
"label": "OSGKeyboard curated AI/tech lexicon",
"license": "MIT (curated seed; OSGKeyboard contributors)",
"raw_count": aiTech.count,
]],
"notes": [
"Project-curated bilingual AI brands, technology terms, companies, and names.",
"No third-party cell dictionaries or Sogou-derived data.",
"PhraseCount weights map to SFCustomLanguageModelData relative frequencies.",
],
"files": ["phrases": "phrases.tsv"],
]
let sourceManifestData = try JSONSerialization.data(
withJSONObject: sourceManifest,
options: [.prettyPrinted, .sortedKeys]
)
try sourceManifestData.write(to: sourceManifestURL)
fputs("Wrote \(sourceManifestURL.path)\n", stderr)
}
}
+133
View File
@@ -0,0 +1,133 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Parse Sogou .scel cell dictionaries into (word, pinyin) entries.
Layout follows the classic SCEL format used by imewlconverter / SogouPopularDict.
"""
from __future__ import annotations
import struct
from dataclasses import dataclass
from pathlib import Path
@dataclass(frozen=True)
class ScelInfo:
word_count: int
name: str
type_name: str
description: str
def _read_uint16(handle) -> int:
data = handle.read(2)
if not data or len(data) < 2:
return 0
return struct.unpack("<H", data)[0]
def _read_uint32(handle) -> int:
data = handle.read(4)
if not data or len(data) < 4:
return 0
return struct.unpack("<I", data)[0]
def _read_utf16_str(handle, *, offset: int = -1, length: int = 0) -> str:
if offset >= 0:
handle.seek(offset)
if length > 0:
data = handle.read(length)
end = 0
for index in range(0, len(data), 2):
if index + 1 < len(data) and data[index] == 0 and data[index + 1] == 0:
end = index
break
if end > 0:
data = data[:end]
return data.decode("utf-16le", errors="ignore")
result = bytearray()
while True:
char = handle.read(2)
if not char or len(char) < 2 or (char[0] == 0 and char[1] == 0):
break
result.extend(char)
return result.decode("utf-16le", errors="ignore")
def is_valid_word(word: str) -> bool:
if not word or not (1 <= len(word) <= 10):
return False
allowed_punct = ",。:;?!()【】《》""''"
return all("\u4e00" <= char <= "\u9fff" or char.isdigit() or char in allowed_punct for char in word)
def get_scel_info(scel_path: Path) -> ScelInfo:
with scel_path.open("rb") as handle:
handle.seek(0x124)
word_count = _read_uint32(handle)
handle.seek(0x130)
name = _read_utf16_str(handle, length=64)
handle.seek(0x338)
type_name = _read_utf16_str(handle, length=64)
handle.seek(0x540)
description = _read_utf16_str(handle, length=1024)
return ScelInfo(word_count=word_count, name=name, type_name=type_name, description=description)
def parse_scel_file(scel_path: Path) -> list[tuple[str, str]]:
"""Return ordered (word, pinyin) pairs from a .scel file."""
entries: list[tuple[str, str]] = []
with scel_path.open("rb") as handle:
handle.seek(0x1540)
pinyin_count = _read_uint32(handle)
pinyin_dict: dict[int, str] = {}
for _ in range(pinyin_count):
pinyin_idx = _read_uint16(handle)
pinyin_len = _read_uint16(handle)
pinyin = handle.read(pinyin_len).decode("utf-16le", errors="ignore").strip().lower()
pinyin_dict[pinyin_idx] = pinyin
try:
while True:
same_pinyin_count = _read_uint16(handle)
pinyin_index_len = _read_uint16(handle)
if pinyin_index_len <= 0 or same_pinyin_count <= 0:
break
pinyin_parts: list[str] = []
for _ in range(pinyin_index_len // 2):
idx = _read_uint16(handle)
part = pinyin_dict.get(idx, "")
if part:
pinyin_parts.append(part)
joined_pinyin = " ".join(pinyin_parts).strip()
for _ in range(same_pinyin_count):
word_len = _read_uint16(handle)
word = handle.read(word_len).decode("utf-16le", errors="ignore")
_ = _read_uint16(handle)
_ = _read_uint32(handle)
_ = handle.read(6)
if is_valid_word(word):
entries.append((word, joined_pinyin))
except (struct.error, OSError):
pass
return entries
def load_pinyin_tsv(tsv_path: Path) -> list[tuple[str, str]]:
entries: list[tuple[str, str]] = []
with tsv_path.open(encoding="utf-8") as handle:
for raw_line in handle:
line = raw_line.strip()
if not line:
continue
word, _, pinyin = line.partition("\t")
if word and pinyin:
entries.append((word, pinyin.strip()))
return entries