From 5d1da0672ed42dca5f063487a882583c2d6fca5c Mon Sep 17 00:00:00 2001 From: caoqianming Date: Wed, 8 Jul 2026 15:33:14 +0800 Subject: [PATCH] =?UTF-8?q?feat(asr):=20=E5=BD=95=E9=9F=B3=E6=96=87?= =?UTF-8?q?=E4=BB=B6=E8=BD=AC=E5=86=99=E5=B7=A5=E5=85=B7=20transcribe=5Fau?= =?UTF-8?q?dio(=E8=AE=AF=E9=A3=9E=20LFASR,bump=200.53.0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用户上传录音(会议/访谈/语音备忘)要"转文字/总结"的需求落地: - 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 --- PROGRESS.md | 4 +- RUN.md | 6 + core/__init__.py | 2 +- core/agent_builder.py | 14 ++ core/asr_lfasr.py | 247 ++++++++++++++++++++++++++++++ scripts/diag_lfasr.py | 67 ++++++++ scripts/smoke_transcribe_audio.py | 106 +++++++++++++ tools/transcribe_audio.py | 143 +++++++++++++++++ 8 files changed, 587 insertions(+), 2 deletions(-) create mode 100644 core/asr_lfasr.py create mode 100644 scripts/diag_lfasr.py create mode 100644 scripts/smoke_transcribe_audio.py create mode 100644 tools/transcribe_audio.py diff --git a/PROGRESS.md b/PROGRESS.md index 218cf2b..fb46732 100644 --- a/PROGRESS.md +++ b/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: ` 日志(带企微 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(列表/正文/删) diff --git a/RUN.md b/RUN.md index 37bfa30..bf0908b 100644 --- a/RUN.md +++ b/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 找 diff --git a/core/__init__.py b/core/__init__.py index 9f29f42..55568cb 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,3 +1,3 @@ # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。 # 改版本只动这一行。 -__version__ = "0.52.6" +__version__ = "0.53.0" diff --git a/core/agent_builder.py b/core/agent_builder.py index 799fd30..065f352 100644 --- a/core/agent_builder.py +++ b/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: diff --git a/core/asr_lfasr.py b/core/asr_lfasr.py new file mode 100644 index 0000000..33bac28 --- /dev/null +++ b/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, + } diff --git a/scripts/diag_lfasr.py b/scripts/diag_lfasr.py new file mode 100644 index 0000000..e9a079b --- /dev/null +++ b/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