graphify 视频音频转写流程解析：Whisper 域提示策略与 transcribe 模块源码实现

本文以 graphify 项目 Codex 版技能参考文档 graphify/skills/codex/references/transcribe.md 为主体，完整还原"Step 2.5 视频转写"步骤的触发条件、Whisper 域提示（domain hint）编写策略、环境变量与转写命令，并结合 transcribe.py 源码与 test_transcribe.py 测试用例，讲清转写结果的缓存、容错、URL 下载与安全边界，帮助你在含视频/音频语料上正确跑通 graphify 的"视频 → 文本 → 知识图谱"链路。

1. Step 2.5 在 graphify 构建流水线中的位置

graphify 的技能流水线（以 skill-codex.md 为主流程文档）按步骤推进：Step 1 安装/定位 Python 解释器（路径记录在 graphify-out/.graphify_python），Step 2 运行 detect 探测文件类型并写出 graphify-out/.graphify_detect.json，Step 3 做结构（AST）与语义抽取，Step 4 之后构建图。

参考文档 transcribe.md 定义了其中一条按需加载的支线——Step 2.5（视频/音频转写）：

• 只有当 detect 报告了一个或多个 video 文件时才读取并执行该参考文档；语料中没有任何视频文件时，整个步骤跳过。
• 视频和音频文件无法被直接读取，必须先转写成文本，随后在 Step 3 中把转写稿（transcript）当作普通文档（doc）文件处理。

skill-codex.md 中对应该步骤的入口描述与此一致：

### Step 2.5 - Video and audio (only if video files detected)

Skip this step entirely if `detect` returned zero `video` files. When the corpus has
video or audio, see `references/transcribe.md` to transcribe them to text first,
then treat the transcripts as doc files in Step 3.

文件如何被归类为 video？detect.py 中定义了 VIDEO = "video" 这一文件类别，与 code、document、paper、image 并列；而转写模块 transcribe.py 用扩展名集合识别具体文件：

VIDEO_EXTENSIONS = {'.mp4', '.mov', '.webm', '.mkv', '.avi', '.m4v', '.mp3', '.wav', '.m4a', '.ogg'}

也就是说，detect 输出中 files.video 数组里的每个路径，都会进入本节的转写流程。

2. Whisper 域提示（domain hint）策略：由语言模型自己写，不再发额外 API 请求

参考文档给出的核心策略（Strategy）原文如下：

Read the god nodes from graphify-out/.graphify_detect.json (or the analysis file if it exists from a previous run). You are already a language model — write a one-sentence domain hint yourself from those labels. Then pass it to Whisper as the initial prompt. No separate API call needed.

即：读取 detect 输出（或上次运行留下的分析文件）中的 god node（枢纽节点）标签，由正在执行技能的编码智能体自己（它本身就是语言模型）写出一个单句领域提示，再把它作为 Whisper 的 initial_prompt 传入，无需为生成提示而额外调用任何 API。

文档同时给出了回退规则：如果语料只有视频文件、没有任何其他文档/代码（此时没有 god node 可读），使用通用回退提示：

"Use proper punctuation and paragraph breaks."

2.1 编写提示的两个文档示例

参考文档中给出的标签 → 提示的映射示例（Step 1 - Write the Whisper prompt yourself）：

god node 标签	生成的域提示
transformer, attention, encoder, decoder	"Machine learning research on transformer architectures and attention mechanisms. Use proper punctuation and paragraph breaks."
kubernetes, deployment, pod, helm	"DevOps discussion about Kubernetes deployments and Helm charts. Use proper punctuation and paragraph breaks."

提示必须以 export 的方式导出为 GRAPHIFY_WHISPER_PROMPT——参考文档特别强调这个环境变量名是转写器实际读取的名字，而且必须 export 而不是简单赋值，否则随后的子 Python 进程看不到它。

2.2 源码中的提示优先级

transcribe.py 中的 build_whisper_prompt 实现了与上述策略配套的逻辑：

def build_whisper_prompt(god_nodes: list[dict]) -> str:
    """Build a domain hint for Whisper from god nodes extracted from the corpus. ..."""
    if not god_nodes:
        return _FALLBACK_PROMPT

    override = os.environ.get("GRAPHIFY_WHISPER_PROMPT")
    if override:
        return override

    labels = [n.get("label", "") for n in god_nodes[:10] if n.get("label")]
    if not labels:
        return _FALLBACK_PROMPT

    topics = ", ".join(labels[:5])
    return f"Technical discussion about {topics}. Use proper punctuation and paragraph breaks."

从源码结构看，优先级为：GRAPHIFY_WHISPER_PROMPT 环境变量覆盖 > 由 god node 标签拼出的 "Technical discussion about <top5 labels>..." > 空标签/无节点时回退到 _FALLBACK_PROMPT（"Use proper punctuation and paragraph breaks."，见 transcribe.py）。这与参考文档"智能体自己写提示并通过环境变量传入"的设计互为印证：环境变量是显式覆盖通道，标签拼接是无智能体参与时的兜底。test_transcribe.py 中的四组用例分别覆盖了空节点回退、环境变量短路（GRAPHIFY_WHISPER_PROMPT 设置后直接返回自定义提示）、标签主题拼接、以及无 label 键节点的安全跳过。

3. 完整转写命令（Step 2 - Transcribe）与逐行注解

参考文档给出的转写命令是可复制可运行的完整形态，核心是通过 $(cat graphify-out/.graphify_python) 取得正确解释器，再从 detect 结果中提取 files.video 列表并调用 transcribe_all：

export GRAPHIFY_WHISPER_MODEL=base  # or whatever --whisper-model the user passed (must be exported)
export GRAPHIFY_WHISPER_PROMPT="<the one-sentence domain hint you composed in Step 1>"
$(cat graphify-out/.graphify_python) -c "
import json, os, sys
from pathlib import Path
from graphify.transcribe import transcribe_all

detect = json.loads(Path('graphify-out/.graphify_detect.json').read_text(encoding=\"utf-8\"))
video_files = detect.get('files', {}).get('video', [])
prompt = os.environ.get('GRAPHIFY_WHISPER_PROMPT', 'Use proper punctuation and paragraph breaks.')

transcript_paths = transcribe_all(video_files, initial_prompt=prompt)
# Write the JSON from Python (NOT a shell '>' redirect): transcribe_all/Whisper
# print progress to stdout, which would otherwise corrupt the JSON file (#1392).
Path('graphify-out/.graphify_transcripts.json').write_text(json.dumps(transcript_paths, ensure_ascii=False), encoding=\"utf-8\")
print(f'Transcribed {len(transcript_paths)} file(s)', file=sys.stderr)
"

对关键行的注解：

• export GRAPHIFY_WHISPER_MODEL=base：Whisper 模型默认 base；如果用户在调用 /graphify 时传了 --whisper-model <name>（见 skill-codex.md 的 /graphify <path> --whisper-model medium），则 export GRAPHIFY_WHISPER_MODEL=<name>。必须 export，原因同上——后续命令行里的子 Python 进程只能从环境继承。
• detect.get('files', {}).get('video', [])：转写清单完全来自 Step 2 的 detect 产物，Step 2.5 不重复扫描目录。
• prompt = os.environ.get('GRAPHIFY_WHISPER_PROMPT', 'Use proper punctuation and paragraph breaks.')：与文档的回退规则一致，环境变量缺失时落到通用提示。
• 用 Python 写 JSON，而不是 shell > 重定向：行内注释指出，transcribe_all/Whisper 会向 stdout 打印进度，若用重定向接收会污染 JSON 文件（对应上游 issue #1392）。
• 计数打印写入 stderr（file=sys.stderr），保证 stdout 干净。

4. 转写完成后的收尾动作

参考文档在命令之后规定了四条收尾规则，需要逐条执行：

1. 从 graphify-out/.graphify_transcripts.json 读回转写稿路径列表；
2. 在 Step 3B 分发语义抽取子智能体（dispatching semantic subagents）之前，把这些路径加入 docs 文件列表；
3. 打印创建了多少份转写稿：Transcribed N video file(s) -> treating as docs；
4. 某个文件转写失败时，打印警告并继续处理其余文件——单文件失败不中止整条流水线。

第 4 条在源码中有直接对应：transcribe_all 对每个文件 try/except，失败时打印 warning: could not transcribe {vf}: {exc} 并跳过（见 transcribe.py），最终只返回成功项的路径；全部失败则返回空列表（test_transcribe_all_skips_failed 用 mock 抛错验证了该行为，见 test_transcribe.py）。

5. 源码纵深：transcribe / transcribe_all 的参数、缓存与运行细节

5.1 单文件转写：transcribe()

def transcribe(
    video_path: Path | str,
    output_dir: Path | None = None,
    initial_prompt: str | None = None,
    force: bool = False,
) -> Path:

（transcribe.py）要点：

• 输出目录与命名：默认写入 graphify-out/transcripts/（_TRANSCRIPTS_DIR = str(_out_path("transcripts"))，out_path 定义在 paths.py）。转写稿文件名为 <音频文件主名>.txt（如 lecture.mp4 → lecture.txt）。
• 转写缓存：若目标 .txt 已存在且 force=False，直接返回缓存路径，不加载 Whisper 模型（test_transcribe_uses_cache 验证；test_transcribe_force_reruns 验证 force=True 时强制重跑）。这是增量构建场景下避免重复推理的关键。
• 运行参数：模型通过 _model_name() 读取 GRAPHIFY_WHISPER_MODEL（默认 base）；Whisper 以 device="cpu", compute_type="int8" 加载（纯 CPU、int8 量化），转写使用 beam_size=5 并把域提示作为 initial_prompt。
• 提示参数：initial_prompt 缺省时同样落到 _FALLBACK_PROMPT。
• 产物打印：每完成一个文件打印 transcript saved -> <path> (lang=<检测语言>, N segments)，语言由 Whisper 的 info.language 给出。
• 依赖缺失：未安装 faster-whisper 时抛出带安装提示的 ImportError（pip install 'graphifyy[video]'），test_transcribe_missing_faster_whisper 验证了异常透传。

5.2 批量转写：transcribe_all()

transcribe_all(video_files, output_dir=None, initial_prompt=None)（transcribe.py）的语义：

• 空输入直接返回 []；
• 域提示只构建一次、共享给所有文件（"built once from corpus god nodes"）；
• 已转写文件即时命中缓存；
• 逐文件容错（见第 4 节），返回成功项的转写稿路径列表。

5.3 环境变量的默认值与约束

从源码可确认的取值边界（均以当前仓库 transcribe.py 为准）：

环境变量 / 参数	默认值	读取位置	说明
GRAPHIFY_WHISPER_MODEL	base	_model_name()	需 export，供子进程继承
GRAPHIFY_WHISPER_PROMPT	"Use proper punctuation and paragraph breaks."（_FALLBACK_PROMPT）	build_whisper_prompt() / 命令行内 os.environ.get	域提示覆盖通道
输出目录	graphify-out（可由 GRAPHIFY_OUT 覆盖）	paths.py	转写稿落在 <输出目录>/transcripts/，清单为 <输出目录>/.graphify_transcripts.json
Whisper 运行配置	device="cpu", compute_type="int8", beam_size=5	transcribe()	无 GPU 依赖的本地推理

6. 附加能力：URL 转写与安全校验

参考文档面向本地视频文件；从源码看，transcribe() 还接受 URL：is_url() 判断路径是否以 http://、https://、www. 开头，是则先经 download_audio() 用 yt-dlp 下载纯音频流（格式 bestaudio[ext=m4a]/bestaudio/best，noplaylist: True，不做 ffmpeg 后处理），文件名用 URL 的 SHA-1 前 12 位（yt_<hash>.<ext>）保证稳定且命中已下载缓存。安全上，validate_url（来自 security.py）在 yt-dlp 运行之前拦截私网 IP 与非法 scheme。这与 README 中"Video / Audio … 以及 YouTube / URLs（需 graphifyy[video]）"、/graphify add <youtube-url> 的说明相一致。依赖方面，pyproject.toml 中 video extra 为 faster-whisper（仅 Python ≥ 3.11）+ yt-dlp，即 Python 3.10 及以下无法启用转写能力，安装方式为 uv tool install "graphifyy[video]"。

7. 行为契约的测试佐证

test_transcribe.py 对转写模块形成了完整的行为契约，可作为验证清单：

• test_video_extensions_set：确认 mp4/mp3/wav/mov 等扩展名在集合内、源码扩展名不在；
• test_build_whisper_prompt_*（4 个用例）：空节点回退、GRAPHIFY_WHISPER_PROMPT 环境变量短路、标签主题拼接（含 "punctuation" 指令）、无标签节点安全跳过；
• test_transcribe_uses_cache / test_transcribe_force_reruns：缓存命中不触发 Whisper；force=True 强制重转；
• test_transcribe_missing_faster_whisper：依赖缺失时 ImportError 透传；
• test_transcribe_all_empty / test_transcribe_all_uses_cache / test_transcribe_all_skips_failed：空输入、批量缓存命中、失败跳过并返回 []。

8. 小结：Step 2.5 的执行清单

按参考文档与源码，一次完整的视频转写应满足：

1. detect 的 files.video 非空才执行本步骤，否则跳过；
2. 从 detect/分析结果读取 god node 标签，由智能体自写单句域提示；仅视频语料时用通用回退提示；
3. export GRAPHIFY_WHISPER_PROMPT 与 GRAPHIFY_WHISPER_MODEL（如需非 base 模型）；
4. 运行第 3 节命令，转写稿落入 graphify-out/transcripts/，路径清单写入 graphify-out/.graphify_transcripts.json；
5. 将转写稿并入 Step 3B 的 docs 列表，打印 Transcribed N video file(s) -> treating as docs；
6. 单文件失败仅告警不中断；已有 .txt 转写稿默认走缓存，force=True 才重跑。

整个过程与 graphify 其余能力一样在本地完成（faster-whisper 纯 CPU 推理），视频内容不离开机器，转写稿随后作为普通文档参与语义抽取并进入统一知识图谱。