feat(asr): 录音文件转写工具 transcribe_audio(讯飞 LFASR,bump 0.53.0)

用户上传录音(会议/访谈/语音备忘)要"转文字/总结"的需求落地:
- core/asr_lfasr.py:讯飞录音文件转写标准版客户端(raasr v2 signa 签名 +
  upload/getResult 异步订单 + lattice 解析 + roleType=1 说话人分离格式化);
  单文件 5h/500MB,服务端解码,本地不需要 ffmpeg
- tools/transcribe_audio.py:host-side tool(look_at_image 同形态,路径复用
  image_ref.resolve_in_root);全文落 <音频名>.transcript.txt 只回预览+路径;
  轮询期响应停止按钮;凭据不齐不挂工具
- 新 env XFYUN_LFASR_SECRET_KEY(APPID 与 IAT 共用,SecretKey 独立)
- 实测:TTS 合成 12.6s 中文语音 13s 出稿逐字全对;originalDuration 是上传参数
  回显,realDuration 才是真实时长
- smoke_transcribe_audio.py(离线 12 项全过)+ diag_lfasr.py 诊断脚本

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

PROGRESS.md

@@ -23,6 +23,7 @@
 
 ### 2026-07
 
+- **07-08 / 0.53.0**:**录音文件转写工具 `transcribe_audio`**(用户上传录音要"转文字/总结"的需求):新 `core/asr_lfasr.py`(讯飞录音文件转写标准版,raasr v2:signa 签名 + upload/getResult 异步订单 + lattice 解析 + 说话人分离格式化)+ `tools/transcribe_audio.py`(look_at_image 同形态 host-side tool,路径复用 image_ref.resolve_in_root)。选 LFASR 弃 IAT 分片方案:单文件 5h/500MB、服务端解码常见格式(本地不需要 ffmpeg)、roleType=1 自带说话人分离(会议场景直接可用);全文落 `<音频名>.transcript.txt` 只回预览+路径,不撑爆 context。凭据 `XFYUN_APPID`(共用)+ 新 env `XFYUN_LFASR_SECRET_KEY`(独立 SecretKey,控制台「录音文件转写标准版」页),没配不挂工具;轮询期响应停止按钮。实测:TTS 合成 12.6s 中文语音 13s 出稿逐字全对;orderInfo.originalDuration 是上传参数回显、realDuration 才是真实时长(踩坑记入代码注释)。smoke `scripts/smoke_transcribe_audio.py`(离线 12 项)+ 诊断 `scripts/diag_lfasr.py`。
 - **07-08 / 0.52.6**:企微语音落地收尾——"文字/语音全无反应"根因定位:**回调注册在企微平台侧失效**(服务器/代码/env 全正常),后台重新保存「接收消息」API 配置即恢复;RUN 故障兜底补该行(判据:`[wecom] inbound` 行没有=平台没投递)。至此两渠道语音全链路用户实测跑通:个人微信(SILK→pilk)+ 企业微信(AMR→ffmpeg)→ 讯飞 IAT → 回显「已识别」→ 进对话。
 - **07-08 / 0.52.5**:企微最终回复推送失败不再静默——用户报企微连文字也无回复:`_bg` 最终 `push_wecom` 返回值一直被忽略,message/send 失败(如企微「企业可信 IP」拦截 60020)完全无痕。改走 `_push_checked` 统一落 `[wecom] push failed: <reason>` 日志(带企微 errcode/errmsg)。
 - **07-08 / 0.52.4**:企微回调加入站留痕日志——用户报企微发语音无反应且 `[wecom-voice]` 日志一行没有(语音分支被调用必打),疑企微侧未投递语音回调;但 text 路径原先零日志无法对照回调通断。解密成功后每条打 `[wecom] inbound msgtype=... from=...`,用"text 行出现、voice 行不出现"即可断定平台侧未投递。个人微信语音已用户实测跑通(0.52.0 链路)。
@@ -219,9 +220,10 @@ core/export_docx.py        383
 core/storage/{__init__,engine,models,usage,utils}.py  ← 4 表(0004-0007 演进);record_chat/image_usage
 core/ark_client.py         105   ← 火山方舟 HTTP 客户端
 core/asr_xfyun.py          170   ← 讯飞语音听写 IAT wss 客户端(整段 PCM→文本;web 语音输入用,diag: scripts/diag_asr.py)
+core/asr_lfasr.py          250   ← 讯飞录音文件转写 LFASR 客户端(异步订单 + 说话人分离;transcribe_audio 工具底座,diag: scripts/diag_lfasr.py)
 core/agent_builder.py      340   ← 装配 lib(有 ARK_API_KEY 才挂 SeedreamTool);build_skill_registry 装两来源
 core/executor.py / sandbox/{network,pool}.py / executor_docker.py  ← Executor ABC + Docker per-user 容器池
-tools/{base,fs,shell,run_python,skill_tool,skill_authoring,seedream,seedance,web_search,web_fetch,documents,materials_project}.py  ← skill_authoring=save_skill/fork_skill(host-side 写 user .skills)
+tools/{base,fs,shell,run_python,skill_tool,skill_authoring,seedream,seedance,web_search,web_fetch,documents,materials_project,transcribe_audio}.py  ← skill_authoring=save_skill/fork_skill(host-side 写 user .skills)
 main.py                    ~210  ← 入口:web / db / probe / user / sandbox check
 db/migrations/versions/    0001-0008
 web/app.py                ~1360  ← /v1 JSON API + user_id 隔离 + run lock + cancel + files + pptx 预览 + skills(列表/正文/删)

RUN.md

@@ -92,6 +92,12 @@
   # XFYUN_APPID=...
   # XFYUN_API_KEY=...
   # XFYUN_API_SECRET=...
+  # 讯飞录音文件转写(agent 工具 transcribe_audio —— 用户上传的录音/会议音频转文字,
+  # core/asr_lfasr.py):可选,APPID 与上面共用 + 独立 SecretKey(控制台「录音文件转写
+  # 标准版」页,不是听写的 APIKey/APISecret)。未配时 agent 里不出现该工具。
+  # 单文件 ≤5 小时 / ≤500MB,服务端解码 + 说话人分离,不需要本地 ffmpeg。
+  # 验证链路:`.venv/Scripts/python.exe scripts/diag_lfasr.py <音频文件> [cn|en]`。
+  # XFYUN_LFASR_SECRET_KEY=...
   # 企微语音消息解码(core/audio.py)额外需系统装 ffmpeg(Linux `apt install ffmpeg`;
   # 不在 PATH 时用下行指向可执行文件)。web 端麦克风输入不需要 ffmpeg。
   # FFMPEG_PATH=/usr/bin/ffmpeg      # 可选,默认从 PATH 找

core/__init__.py

@@ -1,3 +1,3 @@
 # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。
 # 改版本只动这一行。
-__version__ = "0.52.6"
+__version__ = "0.53.0"

core/agent_builder.py

@@ -60,9 +60,11 @@ from tools.schedule import (
     ScheduleCancelTool, ScheduleCreateTool, ScheduleListTool, ScheduleUpdateTool,
 )
 from tools.send_email import SendEmailTool, smtp_configured
+from tools.transcribe_audio import TranscribeAudioTool
 from tools.wechat_bot import WechatPushTool, wechat_push_available
 
 from core.ark_client import ArkConfig
+from core.asr_lfasr import is_configured as lfasr_configured
 from core.bocha_client import BochaConfig
 
 
@@ -674,6 +676,18 @@ def build_agent(
             )
             tools[look_tool.name] = look_tool
 
+    # 录音文件转写(transcribe_audio / 讯飞 LFASR):仅当 XFYUN_APPID +
+    # XFYUN_LFASR_SECRET_KEY 齐了才挂(沿用"有 key 才注册")。与 IAT 语音听写是两个
+    # 服务、两套 key。cancel_check 同 seedance:轮询期(短则十几秒长则几分钟)响应停止按钮。
+    if lfasr_configured():
+        ta = TranscribeAudioTool(
+            working_dir=working_dir_path,
+            base_dir=tool_base,
+            user_root=ur_path,
+            cancel_check=cancel_check,
+        )
+        tools[ta.name] = ta
+
     # 博查联网搜索:仅当 BOCHA_API_KEY 设了才挂
     bocha_cfg = BochaConfig.load()
     if bocha_cfg is not None:

core/asr_lfasr.py

@@ -0,0 +1,247 @@
+"""讯飞录音文件转写(LFASR,raasr.xfyun.cn v2)客户端 —— 整个录音文件转文字。
+
+接口文档: https://www.xfyun.cn/doc/asr/ifasr_new/API.html
+与 `asr_xfyun`(IAT 语音听写,实时流式,单会话 60s)是**两个独立服务**:LFASR 专为
+已成文件的长录音设计 —— 单文件最长 5 小时 / 500MB,服务端自行解码常见音频格式
+(mp3/m4a/wav/amr/ogg/flac/...,无需本地 ffmpeg),自带说话人分离(roleType=1)。
+转写是异步订单:`upload` 拿 orderId → 轮询 `get_result` 等 status=4。
+
+凭据:env `XFYUN_APPID`(与 IAT 共用)+ `XFYUN_LFASR_SECRET_KEY`(LFASR 服务独立的
+SecretKey,讯飞控制台「语音转写」页查看 —— **不是** IAT 的 APIKey/APISecret)。
+签名:signa = Base64(HmacSHA1(key=secretKey, msg=MD5hex(appid+ts)))。
+
+全 sync(httpx):调用方是 tool.execute,本就跑在 worker 线程里。
+"""
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import os
+import time
+from typing import Callable, Optional
+
+import httpx
+
+_API_BASE = "https://raasr.xfyun.cn/v2/api"
+_POLL_INTERVAL_S = 10.0   # getResult 有频控,10s 一查绰绰有余
+_POLL_ERR_TOLERANCE = 5   # 轮询期连续网络错误容忍次数(订单在讯飞侧,不能因抖动丢单)
+
+# orderInfo.failType → 可读提示(文档错误码页;没列到的走通用兜底)
+_FAIL_HINTS = {
+    1: "音频上传失败",
+    2: "音频转码失败(格式可能不支持,建议转成 mp3/wav 再传)",
+    3: "音频识别失败",
+    4: "音频时长超限(最长 5 小时)",
+    5: "音频文件校验失败(文件损坏或为空)",
+    6: "未检出有效语音(静音 / 纯音乐?)",
+}
+
+
+class LfasrError(RuntimeError):
+    """转写失败(讯飞返回错误 / 连接异常),code 为讯飞错误码字符串(连接类为 None)。"""
+
+    def __init__(self, message: str, code: Optional[str] = None):
+        super().__init__(message)
+        self.code = code
+
+
+class LfasrNotConfigured(LfasrError):
+    """XFYUN_APPID / XFYUN_LFASR_SECRET_KEY 未配置。"""
+
+
+class LfasrCancelled(LfasrError):
+    """用户取消等待(讯飞侧订单不受影响,会继续转完但结果没人取)。"""
+
+
+def is_configured() -> bool:
+    """LFASR 凭据是否齐 —— agent_builder 据此决定挂不挂 transcribe_audio tool。"""
+    return all(
+        (os.getenv(k) or "").strip()
+        for k in ("XFYUN_APPID", "XFYUN_LFASR_SECRET_KEY")
+    )
+
+
+def _load_credentials() -> tuple[str, str]:
+    appid = (os.getenv("XFYUN_APPID") or "").strip()
+    secret = (os.getenv("XFYUN_LFASR_SECRET_KEY") or "").strip()
+    if not (appid and secret):
+        raise LfasrNotConfigured(
+            "录音转写未配置:需在 .env 设 XFYUN_APPID / XFYUN_LFASR_SECRET_KEY"
+            "(后者是讯飞控制台「语音转写」服务的 SecretKey,与语音听写的 key 不同)"
+        )
+    return appid, secret
+
+
+def _auth_params() -> dict:
+    appid, secret = _load_credentials()
+    ts = str(int(time.time()))
+    md5hex = hashlib.md5((appid + ts).encode()).hexdigest()
+    signa = base64.b64encode(
+        hmac.new(secret.encode(), md5hex.encode(), hashlib.sha1).digest()
+    ).decode()
+    return {"appId": appid, "ts": ts, "signa": signa}
+
+
+def _post(path: str, params: dict, content: Optional[bytes] = None,
+          timeout_s: float = 60) -> dict:
+    """POST 一次并校验业务码;code != 000000 抛 LfasrError,返回 content 字段。"""
+    try:
+        resp = httpx.post(
+            f"{_API_BASE}{path}",
+            params={**_auth_params(), **params},
+            content=content,
+            headers={"Content-Type": "application/octet-stream"} if content else None,
+            timeout=timeout_s,
+        )
+        resp.raise_for_status()
+        body = resp.json()
+    except LfasrError:
+        raise
+    except Exception as e:  # 连接 / 超时 / 非 JSON / HTTP 状态码
+        raise LfasrError(f"讯飞录音转写连接失败:{type(e).__name__}: {e}")
+    code = str(body.get("code") or "")
+    if code != "000000":
+        desc = body.get("descInfo") or "未知错误"
+        raise LfasrError(f"讯飞录音转写失败({code}):{desc}", code=code)
+    return body.get("content") or {}
+
+
+def upload(data: bytes, file_name: str, *, language: str = "cn",
+           role_type: int = 1) -> tuple[str, float]:
+    """上传音频建转写订单。返回 (orderId, 预估转写耗时秒)。
+
+    duration 参数按官方 demo 传定值(文档标必填但服务端不校验,只原样回显进
+    orderInfo.originalDuration —— 真实时长看 realDuration);roleType=1 开说话人
+    分离(不额外计费,会议场景直接可用)。
+    """
+    params = {
+        "fileName": file_name,
+        "fileSize": str(len(data)),
+        "duration": "200",
+        "language": language,
+        "roleType": str(role_type),
+    }
+    content = _post("/upload", params, content=data, timeout_s=300)
+    order_id = (content.get("orderId") or "").strip()
+    if not order_id:
+        raise LfasrError("讯飞录音转写:upload 成功但未返回 orderId")
+    estimate_s = float(content.get("taskEstimateTime") or 0) / 1000
+    return order_id, estimate_s
+
+
+def get_result(order_id: str) -> dict:
+    """查一次订单。返回 content(含 orderInfo.status / orderResult)。"""
+    return _post("/getResult", {"orderId": order_id, "resultType": "transfer"},
+                 timeout_s=60)
+
+
+def parse_transcript(order_result: str) -> list[tuple[str, str]]:
+    """orderResult(JSON 串)→ [(role, text)] 按时间顺序。
+
+    lattice 每项的 json_1best 是**再编码一层的 JSON 串**(lattice2 里则已是对象),
+    两种形态都兜。role 是说话人编号字符串("1"/"2"...,未分离时 "0" 或缺失 → "")。
+    """
+    try:
+        parsed = json.loads(order_result or "{}")
+    except json.JSONDecodeError as e:
+        raise LfasrError(f"讯飞录音转写:结果 JSON 解析失败:{e}")
+    lattice = parsed.get("lattice") or parsed.get("lattice2") or []
+    segments: list[tuple[str, str]] = []
+    for item in lattice:
+        best = item.get("json_1best")
+        if isinstance(best, str):
+            try:
+                best = json.loads(best)
+            except json.JSONDecodeError:
+                continue
+        st = (best or {}).get("st") or {}
+        text = "".join(
+            cw.get("w", "")
+            for rt in (st.get("rt") or [])
+            for ws in (rt.get("ws") or [])
+            for cw in (ws.get("cw") or [])
+        )
+        if not text.strip():
+            continue
+        role = str(st.get("rl") or "").strip()
+        segments.append(("" if role == "0" else role, text))
+    return segments
+
+
+def format_transcript(segments: list[tuple[str, str]]) -> str:
+    """[(role, text)] → 成稿文本。
+
+    识别出 ≥2 个说话人时按连续同角色合并成「【说话人N】…」段落(会议/访谈场景);
+    单说话人 / 未分离则纯文本拼接。
+    """
+    roles = {r for r, _ in segments if r}
+    if len(roles) < 2:
+        return "".join(t for _, t in segments).strip()
+    lines: list[str] = []
+    cur_role, buf = None, []
+    for role, text in segments:
+        if role != cur_role:
+            if buf:
+                lines.append(f"【说话人{cur_role or '?'}】{''.join(buf)}")
+            cur_role, buf = role, [text]
+        else:
+            buf.append(text)
+    if buf:
+        lines.append(f"【说话人{cur_role or '?'}】{''.join(buf)}")
+    return "\n\n".join(lines).strip()
+
+
+def transcribe_file(
+    data: bytes,
+    file_name: str,
+    *,
+    language: str = "cn",
+    cancel_check: Optional[Callable[[], bool]] = None,
+) -> dict:
+    """上传 + 轮询到订单完成。返回 {"text", "duration_s", "order_id", "elapsed_s"}。
+
+    等待上限取 max(600s, 3×讯飞预估);轮询期连续网络错误 ≤ _POLL_ERR_TOLERANCE 次
+    只记不抛(订单在讯飞侧,不能因一次抖动丢单)。取消抛 LfasrCancelled。
+    """
+    started = time.monotonic()
+    order_id, estimate_s = upload(data, file_name, language=language)
+    deadline = started + max(600.0, estimate_s * 3)
+    consecutive_errs = 0
+    while True:
+        if cancel_check is not None and cancel_check():
+            raise LfasrCancelled(f"用户取消(讯飞订单 {order_id} 会继续转完,但结果未取)")
+        if time.monotonic() > deadline:
+            raise LfasrError(
+                f"讯飞录音转写超时(订单 {order_id} 等了 "
+                f"{int(time.monotonic() - started)}s 仍未完成)"
+            )
+        time.sleep(_POLL_INTERVAL_S)
+        try:
+            content = get_result(order_id)
+            consecutive_errs = 0
+        except LfasrError as e:
+            if e.code is not None:  # 业务错误码:重试无意义,直接抛
+                raise
+            consecutive_errs += 1
+            if consecutive_errs > _POLL_ERR_TOLERANCE:
+                raise
+            continue
+        info = content.get("orderInfo") or {}
+        status = info.get("status")
+        if status == -1:
+            ft = info.get("failType")
+            hint = _FAIL_HINTS.get(ft, f"转写失败(failType={ft})")
+            raise LfasrError(f"讯飞录音转写失败:{hint}")
+        if status != 4:
+            continue  # 0 已创建 / 3 处理中
+        segments = parse_transcript(content.get("orderResult") or "")
+        # originalDuration 只是 upload 时 duration 参数的回显(我们传的定值);
+        # realDuration 才是讯飞实际解出的音频时长(ms,即计费时长)—— 实测验证。
+        return {
+            "text": format_transcript(segments),
+            "duration_s": float(info.get("realDuration") or 0) / 1000,
+            "order_id": order_id,
+            "elapsed_s": time.monotonic() - started,
+        }

scripts/diag_lfasr.py

@@ -0,0 +1,67 @@
+"""讯飞录音文件转写(LFASR)诊断:验证 XFYUN_APPID/XFYUN_LFASR_SECRET_KEY + 订单链路是否通。
+
+用法(.env 在仓库根):
+  .venv/Scripts/python.exe scripts/diag_lfasr.py <音频文件> [cn|en]
+
+上传 → 轮询到订单完成 → 打印时长/耗时/转写文本。
+ASCII 标签输出(Windows 控制台 GBK,不打 emoji);secret 不打印。
+"""
+import os
+import sys
+import time
+
+_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)
+
+
+def _load_env(path: str) -> None:
+    try:
+        from dotenv import load_dotenv
+        load_dotenv(path)
+        return
+    except Exception:
+        pass
+    try:
+        with open(path, encoding="utf-8") as f:
+            for line in f:
+                line = line.strip()
+                if not line or line.startswith("#") or "=" not in line:
+                    continue
+                k, v = line.split("=", 1)
+                os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))
+    except FileNotFoundError:
+        pass
+
+
+_load_env(os.path.join(_ROOT, ".env"))
+
+from core.asr_lfasr import LfasrError, transcribe_file  # noqa: E402
+
+
+def main() -> int:
+    print(f"[env] XFYUN_APPID={os.getenv('XFYUN_APPID') or '(missing)'} "
+          f"LFASR_SECRET_KEY={'set' if os.getenv('XFYUN_LFASR_SECRET_KEY') else '(missing)'}")
+    if len(sys.argv) < 2:
+        print("[usage] diag_lfasr.py <audio file> [cn|en]")
+        return 2
+    path = sys.argv[1]
+    lang = sys.argv[2] if len(sys.argv) > 2 else "cn"
+    data = open(path, "rb").read()
+    print(f"[file] {path} ({len(data)} bytes) language={lang}")
+
+    t0 = time.monotonic()
+    try:
+        result = transcribe_file(data, os.path.basename(path), language=lang)
+    except LfasrError as e:
+        print(f"[FAIL] {type(e).__name__}: {e}")
+        return 1
+    print(f"[ok] orderId={result['order_id']} duration={result['duration_s']:.1f}s "
+          f"elapsed={result['elapsed_s']:.1f}s (wall {time.monotonic() - t0:.1f}s)")
+    print("[text]")
+    print(result["text"])
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/smoke_transcribe_audio.py

@@ -0,0 +1,106 @@
+"""smoke: transcribe_audio / core.asr_lfasr 的纯本地部分(不联讯飞)。
+
+验证:
+  1. asr_lfasr.parse_transcript:json_1best 为串(lattice)与对象(lattice2)两形态、
+     空文本段跳过、rl 角色提取;
+  2. asr_lfasr.format_transcript:多说话人按连续同角色合并「【说话人N】」段落,
+     单说话人纯文本;
+  3. TranscribeAudioTool 入参错误路径(路径不存在 / 扩展名不对 / 视频提示 / 空文件 /
+     越界 / 坏 language),全部在上传前早返,不触网。
+
+跑法: .venv/Scripts/python.exe scripts/smoke_transcribe_audio.py
+"""
+from __future__ import annotations
+
+import json
+import sys
+import tempfile
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from core.asr_lfasr import format_transcript, parse_transcript  # noqa: E402
+from tools.transcribe_audio import TranscribeAudioTool  # noqa: E402
+
+FAIL = 0
+
+
+def check(name: str, ok: bool, detail: str = "") -> None:
+    global FAIL
+    tag = "PASS" if ok else "FAIL"
+    print(f"[{tag}] {name}" + (f" -- {detail}" if detail else ""))
+    if not ok:
+        FAIL = 1
+
+
+def seg(text: str, rl: str = "0") -> dict:
+    """构造一个 lattice 项(json_1best 为再编码 JSON 串,官方返回形态)。"""
+    st = {"rl": rl, "rt": [{"ws": [{"cw": [{"w": ch}]} for ch in text]}]}
+    return {"json_1best": json.dumps({"st": st}, ensure_ascii=False)}
+
+
+def main() -> int:
+    # ---- 1. parse_transcript ----
+    order = json.dumps({"lattice": [seg("今天开会", "1"), seg("好的", "2"), seg("", "1")]},
+                       ensure_ascii=False)
+    segs = parse_transcript(order)
+    check("parse: 2 segments (empty skipped)", len(segs) == 2, str(segs))
+    check("parse: roles extracted", segs == [("1", "今天开会"), ("2", "好的")], str(segs))
+
+    # lattice2 形态:json_1best 已是对象
+    obj_item = json.loads(seg("测试", "0")["json_1best"])
+    segs2 = parse_transcript(json.dumps({"lattice2": [{"json_1best": obj_item}]},
+                                        ensure_ascii=False))
+    check("parse: lattice2 object form", segs2 == [("", "测试")], str(segs2))
+
+    # ---- 2. format_transcript ----
+    multi = [("1", "大家好。"), ("1", "开始吧。"), ("2", "收到。"), ("1", "散会。")]
+    out = format_transcript(multi)
+    check("format: multi-speaker merged",
+          out == "【说话人1】大家好。开始吧。\n\n【说话人2】收到。\n\n【说话人1】散会。",
+          repr(out))
+    single = [("1", "自言"), ("1", "自语")]
+    check("format: single speaker plain", format_transcript(single) == "自言自语")
+    check("format: no-role plain", format_transcript([("", "甲"), ("", "乙")]) == "甲乙")
+
+    # ---- 3. tool 早返错误路径(不触网) ----
+    with tempfile.TemporaryDirectory(prefix="zcbot-smoke-ta-") as td:
+        root = Path(td)
+        wd = root / "taskdir"
+        wd.mkdir()
+        tool = TranscribeAudioTool(working_dir=wd, base_dir=wd, user_root=root)
+
+        r = tool.execute(audio="inbound/nope.m4a")
+        check("missing file -> [Error]", r.startswith("[Error]") and "找不到" in r, r[:60])
+
+        (wd / "notes.docx").write_bytes(b"PK\x03\x04fake")
+        r = tool.execute(audio="notes.docx")
+        check("non-audio ext -> [Error]", r.startswith("[Error]") and "扩展名" in r, r[:60])
+
+        (wd / "clip.mp4").write_bytes(b"\x00\x00\x00\x18ftyp")
+        r = tool.execute(audio="clip.mp4")
+        check("video ext -> hint to extract audio",
+              r.startswith("[Error]") and "音轨" in r, r[:80])
+
+        (wd / "empty.mp3").write_bytes(b"")
+        r = tool.execute(audio="empty.mp3")
+        check("empty file -> [Error]", r.startswith("[Error]") and "空" in r, r[:60])
+
+        outside = root.parent / "zcbot-smoke-outside.mp3"
+        try:
+            outside.write_bytes(b"xx")
+            r = tool.execute(audio=str(outside))
+            check("out-of-root abs path -> [Error]",
+                  r.startswith("[Error]") and "越界" in r, r[:60])
+        finally:
+            outside.unlink(missing_ok=True)
+
+        r = tool.execute(audio="a.mp3", language="fr")
+        check("bad language -> [Error]", r.startswith("[Error]") and "language" in r, r[:60])
+
+    print("RESULT:", "FAIL" if FAIL else "ALL PASS")
+    return FAIL
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/transcribe_audio.py

@@ -0,0 +1,143 @@
+"""transcribe_audio: 用户上传的录音文件 → 文字稿(讯飞 LFASR 录音文件转写)。
+
+主模型听不见音频 —— 用户上传录音(会议 / 访谈 / 语音备忘)要"转成文字 / 总结"时调这个。
+host-side tool:XFYUN_* 凭据在宿主(沙箱内没有),与 look_at_image 同形态:workspace
+文件路径进、文本出。路径解析复用 tools/image_ref.resolve_in_root(同一套三形态解析
++ user_root 越界校验)。
+
+底座是 `core.asr_lfasr`(讯飞录音文件转写,单文件 5 小时 / 500MB,服务端解码 + 说话人
+分离),原始文件字节直传,本地不需要 ffmpeg。转写全文落 `<音频名>.transcript.txt`
+(与音频同目录),tool result 只带预览 + 路径 —— 长文进文件不进 context,后续总结 /
+检索由主模型 read 文件自取。
+"""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Callable, Optional
+
+from core import asr_lfasr
+
+from .base import Tool
+from .image_ref import resolve_in_root
+
+# LFASR 服务端支持的音频格式(文档列的常见项)。视频容器(mp4 等)不支持 ——
+# 拒绝并提示先抽音轨,而不是静默失败在讯飞侧。
+AUDIO_EXTS = {
+    ".mp3", ".wav", ".pcm", ".aac", ".opus", ".flac", ".ogg", ".m4a", ".amr",
+}
+_MAX_RAW_MB = 500  # LFASR 单文件上限
+_PREVIEW_CHARS = 500
+
+
+def _fmt_dur(seconds: float) -> str:
+    m, s = divmod(int(seconds + 0.5), 60)
+    return f"{m}m{s:02d}s" if m else f"{s}s"
+
+
+class TranscribeAudioTool(Tool):
+    name = "transcribe_audio"
+    description = (
+        "Transcribe a speech recording file (meeting / interview / voice memo) in the workspace "
+        "to text via Xfyun long-form ASR, with automatic speaker separation. You (the main model) "
+        "cannot hear audio — call this when the user uploads an audio file (look for a "
+        "`[用户上传的文件] <path>` line with an audio extension) and asks to 转文字/转写/整理/总结"
+        "录音. Supports mp3/wav/m4a/aac/opus/flac/ogg/amr/pcm, up to 5 hours / 500MB per file "
+        "(video files are rejected — extract the audio track first). Transcription is asynchronous "
+        "on Xfyun's side: expect roughly 10-30s for short clips and up to several minutes for "
+        "hour-long recordings. The FULL transcript is saved to `<audio>.transcript.txt` next to "
+        "the file; the tool result only carries a preview + that path — read the saved file before "
+        "summarizing. Mandarin (with mixed English) by default; pass language=en for pure English."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "audio": {
+                "type": "string",
+                "description": (
+                    "要转写的录音文件相对路径(task_dir 内,如 'inbound/会议录音.m4a',"
+                    "或用户消息里 `[用户上传的文件]` 行给的路径)。"
+                ),
+            },
+            "language": {
+                "type": "string",
+                "enum": ["cn", "en"],
+                "description": "语种(可选):cn 中文普通话含中英混合(默认)/ en 纯英文。",
+            },
+        },
+        "required": ["audio"],
+    }
+
+    def __init__(
+        self,
+        *,
+        working_dir: Path,
+        base_dir: Optional[Path] = None,
+        user_root: Optional[Path] = None,
+        cancel_check: Optional[Callable[[], bool]] = None,
+    ) -> None:
+        super().__init__(base_dir, user_root=user_root)
+        self.working_dir = Path(working_dir)
+        self.cancel_check = cancel_check  # 轮询期间响应用户停止按钮(长录音要等几分钟)
+
+    def execute(self, audio: str, language: Optional[str] = None) -> str:
+        if not (audio or "").strip():
+            return "[Error] audio(录音文件路径)不能为空"
+        lang = (language or "cn").strip() or "cn"
+        if lang not in ("cn", "en"):
+            return f"[Error] language 不支持: {lang!r}(仅 cn / en)"
+
+        resolved = resolve_in_root(audio.strip(), self.working_dir, self.user_root)
+        if resolved is None:
+            return (
+                f"[Error] 录音文件找不到或越界: {audio!r}。请传 task_dir 内已存在文件的"
+                f"相对路径(如 'inbound/xxx.m4a',或用户消息里 `[用户上传的文件]` 行给的路径)。"
+            )
+        ext = resolved.suffix.lower()
+        if ext not in AUDIO_EXTS:
+            hint = "视频文件请先抽出音轨(如 ffmpeg -i in.mp4 -vn out.m4a)再传。" \
+                if ext in (".mp4", ".mov", ".avi", ".mkv", ".webm") else ""
+            return (
+                f"[Error] 扩展名不支持: {ext or '(无)'}。"
+                f"仅支持 {'/'.join(sorted(e.lstrip('.') for e in AUDIO_EXTS))}。{hint}"
+            )
+        try:
+            raw = resolved.read_bytes()
+        except OSError as e:
+            return f"[Error] 读取文件失败: {type(e).__name__}: {e}"
+        if not raw:
+            return "[Error] 文件是空的(0 字节)"
+        if len(raw) > _MAX_RAW_MB * 1024 * 1024:
+            return (
+                f"[Error] 文件 {len(raw) / 1024 / 1024:.0f}MB 超过讯飞 {_MAX_RAW_MB}MB 上限。"
+                f"先压缩(如转 mp3)或切段再传。"
+            )
+
+        try:
+            result = asr_lfasr.transcribe_file(
+                raw, resolved.name, language=lang, cancel_check=self.cancel_check,
+            )
+        except asr_lfasr.LfasrCancelled as e:
+            return f"[Cancelled] {e}"
+        except asr_lfasr.LfasrError as e:
+            return f"[Error] {e}"
+
+        text = result["text"]
+        if not text:
+            return (
+                "[transcribe_audio] 转写完成但没有文字 —— 录音可能是纯音乐 / 静音 / "
+                f"语种不符(当前 language={lang})。"
+            )
+
+        out_path = resolved.with_suffix(".transcript.txt")
+        out_path.write_text(text + "\n", encoding="utf-8")
+
+        banner = (
+            f"[transcribe_audio] audio={self._display(resolved)}"
+            f" · duration={_fmt_dur(result['duration_s'])}"
+            f" · elapsed={_fmt_dur(result['elapsed_s'])}"
+            f" · saved={self._display(out_path)}"
+        )
+        preview = text[:_PREVIEW_CHARS]
+        if len(text) > _PREVIEW_CHARS:
+            preview += f"\n……(预览截断,全文 {len(text)} 字见 saved 文件,总结前先 read 它)"
+        return f"{banner}\n\n{preview}"