feat(voice): 豆包 flash ASR，支持热词与对话上下文

用极速版 HTTP 替代 Whisper；默认热词「小荷」，可选 history/--correct。

AGENTS.md

@@ -77,6 +77,18 @@ mytoolkit voice tts "文本" -v zh_male_wennuanahu_moon_bigtts --format mp3 --sp
 
 语速：`-50` 到 `100`（0 = 正常）
 
+### 语音识别 (STT)
+
+豆包录音文件识别极速版（flash ASR），与 TTS 共用同一组 Volcengine 凭证：
+
+```bash
+mytoolkit voice stt -i clip.webm
+mytoolkit voice stt -i clip.wav -o out.txt
+mytoolkit voice stt -i clip.wav --no-correct   # 跳过大模型纠错
+```
+
+识别请求会带 ASR 原生热词（默认「小荷」）与可选对话上下文（`--history-file`）。`--correct` 可再开文本大模型兜底纠错（默认关闭）。webm / m4a 等会尽量经 `ffmpeg` 转 16 kHz mono wav。
+
 ---
 
 ## 4. LaTeX 脚手架与投稿模板

README.md

@@ -56,7 +56,7 @@ mytoolkit --help
 | `latex` | `compile` / `count` |
 | `bib` | `to-markdown` / `cv-update` |
 | `video` | `avi-to-mp4` |
-| `voice` | Text-to-speech via Volcengine/Doubao (`voice tts`, mp3/wav/pcm) |
+| `voice` | Text-to-speech / speech-to-text via Volcengine/Doubao (`voice tts`, `voice stt`) |
 | `webpage` | Docusaurus build/list/rebuild/push |
 | `mail` | IMAP/SMTP client (read/draft/forward/reply, no auto-send) |
 | `templates` | Inspect template root; external override still possible |

mytoolkit/api/__init__.py

@@ -1,6 +1,13 @@
 """Pure API clients for external services (no CLI deps)."""
 
+from .asr import transcribe_flash
+from .asr_correct import correct_asr_text
 from .image import generate_image
 from .voice import synthesize_tts
 
-__all__ = ["generate_image", "synthesize_tts"]
+__all__ = [
+    "generate_image",
+    "synthesize_tts",
+    "transcribe_flash",
+    "correct_asr_text",
+]

mytoolkit/api/asr.py

@@ -0,0 +1,206 @@
+"""Doubao / Volcengine flash ASR (录音文件识别极速版)."""
+
+from __future__ import annotations
+
+import base64
+import json
+import shutil
+import subprocess
+import tempfile
+import uuid
+from pathlib import Path
+from typing import Any, Sequence
+
+import urllib.error
+import urllib.request
+
+_FLASH_URL = "https://openspeech.bytedance.com/api/v3/auc/bigmodel/recognize/flash"
+_RESOURCE_ID = "volc.bigasr.auc_turbo"
+_OK_CODE = "20000000"
+
+_FORMAT_BY_SUFFIX = {
+    ".wav": "wav",
+    ".mp3": "mp3",
+    ".ogg": "ogg",
+    ".webm": "webm",
+    ".m4a": "mp4",
+    ".mp4": "mp4",
+    ".aac": "aac",
+    ".flac": "flac",
+}
+
+# Product defaults — ASR-side boosting, not post-hoc string replace.
+DEFAULT_HOTWORDS: tuple[str, ...] = ("小荷", "Xiaohe")
+DEFAULT_SCENE_HINTS: tuple[str, ...] = (
+    "我在使用个人助理小荷（Xiaohe）",
+)
+
+
+def _to_wav_if_needed(path: Path) -> tuple[bytes, str]:
+    """Return (audio_bytes, format). Convert unsupported containers via ffmpeg."""
+    suf = path.suffix.lower()
+    fmt = _FORMAT_BY_SUFFIX.get(suf)
+    data = path.read_bytes()
+    if fmt in {"wav", "mp3", "ogg"}:
+        return data, fmt
+    ffmpeg = shutil.which("ffmpeg")
+    if not ffmpeg:
+        if fmt:
+            return data, fmt
+        raise RuntimeError(
+            f"unsupported audio format {suf or '(unknown)'}; install ffmpeg "
+            "or upload wav/mp3/ogg"
+        )
+    with tempfile.TemporaryDirectory(prefix="mytoolkit-asr-") as tmp:
+        out = Path(tmp) / "out.wav"
+        proc = subprocess.run(
+            [
+                ffmpeg,
+                "-y",
+                "-i",
+                str(path),
+                "-ac",
+                "1",
+                "-ar",
+                "16000",
+                str(out),
+            ],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        if proc.returncode != 0 or not out.is_file():
+            detail = (proc.stderr or proc.stdout or "").strip()[:300]
+            raise RuntimeError(f"ffmpeg convert failed: {detail or 'unknown'}")
+        return out.read_bytes(), "wav"
+
+
+def build_asr_corpus(
+    *,
+    hotwords: Sequence[str] | None = None,
+    history: Sequence[str] | None = None,
+    scene_hints: Sequence[str] | None = None,
+    max_history: int = 8,
+) -> dict[str, Any] | None:
+    """Build ``request.corpus`` for Doubao ASR (hotwords + dialog context)."""
+    if hotwords is None:
+        hotwords = DEFAULT_HOTWORDS
+    if scene_hints is None:
+        scene_hints = DEFAULT_SCENE_HINTS
+
+    corpus: dict[str, Any] = {}
+
+    words = [w.strip() for w in hotwords if w and w.strip()]
+    if words:
+        corpus["context"] = json.dumps(
+            {
+                "hotwords": [
+                    {"word": w, "scale": 2.0} for w in words[:200]
+                ]
+            },
+            ensure_ascii=False,
+        )
+
+    # Newest-first context snippets (ASR truncates old ones when over limit).
+    snippets: list[str] = []
+    for h in scene_hints:
+        t = (h or "").strip()
+        if t:
+            snippets.append(t)
+    hist = [((h or "").strip()) for h in (history or ())]
+    hist = [h for h in hist if h]
+    # Caller may pass oldest→newest; reverse to newest-first for the API.
+    for h in reversed(hist[-max_history:]):
+        snippets.append(h[:200])
+    if snippets:
+        corpus["context_data"] = json.dumps(
+            [{"text": s} for s in snippets[:20]],
+            ensure_ascii=False,
+        )
+
+    return corpus or None
+
+
+def transcribe_flash(
+    appid: str,
+    token: str,
+    audio_path: str | Path,
+    *,
+    uid: str | None = None,
+    timeout_sec: float = 60.0,
+    hotwords: Sequence[str] | None = None,
+    history: Sequence[str] | None = None,
+    scene_hints: Sequence[str] | None = None,
+) -> str:
+    """Transcribe a local audio file via Doubao flash ASR. Returns text."""
+    path = Path(audio_path)
+    if not path.is_file():
+        raise FileNotFoundError(str(path))
+
+    audio_bytes, fmt = _to_wav_if_needed(path)
+    if not audio_bytes:
+        raise RuntimeError("empty audio")
+
+    headers = {
+        "Content-Type": "application/json",
+        "X-Api-App-Key": appid,
+        "X-Api-Access-Key": token,
+        "X-Api-Resource-Id": _RESOURCE_ID,
+        "X-Api-Request-Id": str(uuid.uuid4()),
+        "X-Api-Sequence": "-1",
+    }
+    request_body: dict[str, Any] = {
+        "model_name": "bigmodel",
+        "enable_itn": True,
+        "enable_punc": True,
+        "enable_ddc": True,
+    }
+    corpus = build_asr_corpus(
+        hotwords=hotwords, history=history, scene_hints=scene_hints
+    )
+    if corpus:
+        request_body["corpus"] = corpus
+
+    body = {
+        "user": {"uid": uid or appid},
+        "audio": {
+            "data": base64.b64encode(audio_bytes).decode("ascii"),
+            "format": fmt,
+        },
+        "request": request_body,
+    }
+    req = urllib.request.Request(
+        _FLASH_URL,
+        data=json.dumps(body).encode("utf-8"),
+        headers=headers,
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=timeout_sec) as resp:
+            status = resp.headers.get("X-Api-Status-Code", "")
+            message = resp.headers.get("X-Api-Message", "")
+            raw = resp.read().decode("utf-8", errors="replace")
+    except urllib.error.HTTPError as exc:
+        detail = exc.read().decode("utf-8", errors="replace")[:400]
+        raise RuntimeError(
+            f"ASR HTTP {exc.code}: {detail or exc.reason}"
+        ) from exc
+    except urllib.error.URLError as exc:
+        raise RuntimeError(f"ASR network error: {exc.reason}") from exc
+
+    if status and status != _OK_CODE:
+        raise RuntimeError(
+            f"ASR failed: status={status} message={message or 'unknown'}"
+        )
+
+    try:
+        payload = json.loads(raw) if raw else {}
+    except json.JSONDecodeError as exc:
+        raise RuntimeError("ASR response is not JSON") from exc
+
+    result = payload.get("result") if isinstance(payload, dict) else None
+    if isinstance(result, dict):
+        text = (result.get("text") or "").strip()
+        if text:
+            return text
+    raise RuntimeError("ASR returned empty text")

mytoolkit/api/asr_correct.py

@@ -0,0 +1,128 @@
+"""LLM post-correction for ASR transcripts (Ark / DeepSeek)."""
+
+from __future__ import annotations
+
+import os
+from typing import Callable
+
+# Soft domain hints for the model — not string-replace rules.
+DEFAULT_CONTEXT = (
+    "这是个人助理「小荷」（Xiaohe）客户端的语音输入。"
+    "识别结果可能把专有名词听错成同音字（例如产品名）。"
+)
+
+DEFAULT_ARK_MODEL = os.environ.get(
+    "MYTOOLKIT_ASR_CORRECT_MODEL", "doubao-seed-1-6-flash-250828"
+)
+DEFAULT_DEEPSEEK_MODEL = os.environ.get(
+    "MYTOOLKIT_ASR_CORRECT_DEEPSEEK_MODEL", "deepseek-chat"
+)
+
+
+def _system_prompt(context: str) -> str:
+    system = (
+        "你是语音识别结果校对助手。根据上下文，轻量修正明显的识别错误"
+        "（同音字、专有名词、标点），保留原意与口语风格。"
+        "不要新增原句没有的信息，不要翻译，不要解释。"
+        "只输出修正后的完整句子。"
+    )
+    if context.strip():
+        system += f"\n背景：{context.strip()}"
+    return system
+
+
+def _unwrap(text: str) -> str:
+    out = (text or "").strip()
+    if len(out) >= 2 and out[0] == out[-1] and out[0] in "\"'“”":
+        out = out[1:-1].strip()
+    return out
+
+
+def _chat_correct(
+    *,
+    base_url: str,
+    api_key: str,
+    model: str,
+    text: str,
+    context: str,
+    timeout_sec: float,
+) -> str:
+    from openai import OpenAI
+
+    client = OpenAI(base_url=base_url, api_key=api_key, timeout=timeout_sec)
+    resp = client.chat.completions.create(
+        model=model,
+        temperature=0.1,
+        max_tokens=min(800, max(64, len(text) * 3)),
+        messages=[
+            {"role": "system", "content": _system_prompt(context)},
+            {"role": "user", "content": text},
+        ],
+    )
+    return _unwrap(resp.choices[0].message.content or "") or text
+
+
+def correct_asr_text(
+    text: str,
+    *,
+    context: str = DEFAULT_CONTEXT,
+    ark_api_key: str | None = None,
+    deepseek_api_key: str | None = None,
+    ark_model: str | None = None,
+    deepseek_model: str | None = None,
+    timeout_sec: float = 20.0,
+    on_skip: Callable[[str], None] | None = None,
+) -> str:
+    """Correct an ASR transcript with a chat model. Returns corrected text.
+
+    Tries Ark first (if key present), then DeepSeek. On failure, returns the
+    original text. Domain hints are prompt context only — no hardcoded replaces.
+    """
+    cleaned = (text or "").strip()
+    if not cleaned:
+        return ""
+
+    attempts: list[tuple[str, str, str, str]] = []
+    if ark_api_key:
+        attempts.append(
+            (
+                "ark",
+                "https://ark.cn-beijing.volces.com/api/v3",
+                ark_api_key,
+                (ark_model or DEFAULT_ARK_MODEL).strip() or DEFAULT_ARK_MODEL,
+            )
+        )
+    if deepseek_api_key:
+        attempts.append(
+            (
+                "deepseek",
+                "https://api.deepseek.com",
+                deepseek_api_key,
+                (deepseek_model or DEFAULT_DEEPSEEK_MODEL).strip()
+                or DEFAULT_DEEPSEEK_MODEL,
+            )
+        )
+
+    if not attempts:
+        if on_skip:
+            on_skip("no LLM API key for ASR correction")
+        return cleaned
+
+    last_err = ""
+    for name, base_url, key, model in attempts:
+        try:
+            return _chat_correct(
+                base_url=base_url,
+                api_key=key,
+                model=model,
+                text=cleaned,
+                context=context,
+                timeout_sec=timeout_sec,
+            )
+        except Exception as exc:  # noqa: BLE001
+            last_err = f"{name}: {exc}"
+            continue
+
+    if on_skip:
+        on_skip(last_err or "ASR correction failed")
+    return cleaned

mytoolkit/commands/voice.py

@@ -1,4 +1,4 @@
-"""Voice / TTS utilities via Volcengine (ByteDance) WebSocket API."""
+"""Voice / TTS / STT utilities via Volcengine (ByteDance) APIs."""
 
 import os
 import re
@@ -6,6 +6,8 @@ from pathlib import Path
 
 import click
 
+from mytoolkit.api.asr import transcribe_flash
+from mytoolkit.api.asr_correct import correct_asr_text
 from mytoolkit.api.voice import synthesize_tts
 from mytoolkit.config import config
 from mytoolkit.utils import handle_errors
@@ -59,7 +61,7 @@ def _synthesize_sync(
 
 @click.group()
 def voice():
-    """Voice synthesis (TTS) commands."""
+    """Voice synthesis (TTS) and recognition (STT) commands."""
     pass
 
 
@@ -97,6 +99,84 @@ def tts_cmd(text, file, speaker, fmt, speed, output):
     click.echo(os.path.abspath(result))
 
 
+@voice.command("stt")
+@click.option(
+    "-i",
+    "--input",
+    "input_path",
+    type=click.Path(exists=True, dir_okay=False),
+    required=True,
+    help="输入音频文件",
+)
+@click.option(
+    "-o",
+    "--output",
+    type=click.Path(dir_okay=False),
+    help="写出识别文本（默认打印到 stdout）",
+)
+@click.option(
+    "--history-file",
+    type=click.Path(exists=True, dir_okay=False),
+    help="近期对话文本（每行一条，旧→新），传给 ASR corpus 上下文",
+)
+@click.option(
+    "--hotword",
+    multiple=True,
+    help="ASR 热词（可重复）；默认含「小荷」",
+)
+@click.option(
+    "--correct/--no-correct",
+    default=False,
+    show_default=True,
+    help="识别后再用文本大模型纠错（可选兜底；需 ark/deepseek key）",
+)
+@handle_errors
+def stt_cmd(input_path, output, history_file, hotword, correct):
+    """使用豆包/火山引擎极速 ASR 将语音转为文本。"""
+    appid, access_token = _get_credentials()
+    if not appid or not access_token:
+        raise click.UsageError(
+            "Volcengine credentials not set. Run:\n"
+            "  mytoolkit env set secrets.volcengine.app_id <appid>\n"
+            "  mytoolkit env set secrets.volcengine.access_token <token>"
+        )
+    history = None
+    if history_file:
+        history = [
+            ln.strip()
+            for ln in Path(history_file).read_text(encoding="utf-8").splitlines()
+            if ln.strip()
+        ]
+    hotwords = list(hotword) if hotword else None
+    text = transcribe_flash(
+        appid,
+        access_token,
+        input_path,
+        history=history,
+        hotwords=hotwords,
+    )
+    if correct:
+        ark_key = config.resolve_key("secrets.api_keys.ark", "ARK_API_KEY")
+        deepseek_key = config.resolve_key(
+            "secrets.api_keys.deepseek", "DEEPSEEK_API_KEY"
+        )
+        ark_model = config.resolve_key(
+            "settings.asr_correct_ark_model", "MYTOOLKIT_ASR_CORRECT_MODEL"
+        )
+        text = correct_asr_text(
+            text,
+            ark_api_key=ark_key,
+            deepseek_api_key=deepseek_key,
+            ark_model=ark_model,
+            on_skip=lambda msg: click.echo(f"ASR correct skipped: {msg}", err=True),
+        )
+    if output:
+        Path(output).write_text(text + "\n", encoding="utf-8")
+        click.echo(os.path.abspath(output))
+    else:
+        click.echo(text)
+
+
 def _natural_sort_key(path: Path):
     name = path.stem
     return [int(text) if text.isdigit() else text.lower()