144 lines
6.3 KiB
Python
144 lines
6.3 KiB
Python
"""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}"
|