架构：本地轻量预处理 + Dify 纯编排 + 云端 AI API 核心约束：

• ✅ 原视频不上传（10GB → 100MB，缩减 99%）
• ✅ 除 Dify 自身外无任何本地模型部署
• ✅ 本地仅跑传统图像处理算法（无 AI 模型、无 GPU）
• ✅ 每个知识点关联视频时间戳 + PPT 截图 OCR

一、整体架构

1.1 三层分离设计

┌─────────────────────────────────────────────────────────────┐
│ 【Layer 1: 本地预处理器】  你的电脑 / 内网服务器             │
│                                                              │
│  用户双击 run.bat 或命令行:                                  │
│    python preprocess.py "D:\videos\course.mp4"              │
│                                                              │
│  本地 CPU 纯算法处理（无需模型、无需 GPU、无需联网）:         │
│   ① FFmpeg 抽音频 (MP3 64kbps)       ──► 80MB               │
│   ② PySceneDetect 场景检测关键帧     ──► 50~100张 JPG       │
│   ③ ImageHash 感知哈希去重           ──► 传统算法           │
│   ④ OpenCV 规则过滤讲师人像页        ──► 可选                │
│   ⑤ OSS 批量上传（仅音频+截图）      ──► 110MB 总上传       │
│   ⑥ HTTP 触发 Dify Workflow API                             │
│   ⑦ 轮询结果并下载到本地 output/ 目录                        │
└───────────────────────┬─────────────────────────────────────┘
                        │ JSON payload（仅 URL，不含文件）
                        ▼
┌─────────────────────────────────────────────────────────────┐
│ 【Layer 2: Dify Workflow】  Dify 服务器                      │
│                                                              │
│  纯编排 + 纯 JSON 数据变形（不碰文件、不调 FFmpeg）:          │
│   · HTTP 节点：调用各 AI API                                 │
│   · 代码节点：语义分段 / 时间戳对齐 / 去噪 / 聚合 / 格式化    │
│   · 迭代节点（并行模式）：批量调用 Qwen-VL / LLM             │
│   · LLM 节点：知识点提取 / 笔记生成                          │
└───────────────────────┬─────────────────────────────────────┘
                        │ HTTP API
                        ▼
┌─────────────────────────────────────────────────────────────┐
│ 【Layer 3: 云端 AI 服务】  阿里云百炼 / OpenAI 等            │
│                                                              │
│   · Paraformer-v2 ASR (词级时间戳)                          │
│   · Qwen2.5-VL-Max (判图 + OCR + 结构化)                    │
│   · DeepSeek-V3 (噪声分类)                                  │
│   · GPT-4o / Claude 3.5 (笔记生成)                          │
│   · text-embedding-v3 (语义向量)                            │
└─────────────────────────────────────────────────────────────┘

1.2 核心设计原则

二、Layer 1：本地预处理器详细设计

2.1 部署要求（极简）

2.2 依赖安装（一次性，5 分钟）

# 核心依赖
pip install scenedetect[opencv]      # 场景检测（纯算法）
pip install imageio-ffmpeg           # FFmpeg 自动下载
pip install ImageHash Pillow         # 感知哈希去重
pip install opencv-python-headless   # 图像特征提取
pip install oss2                     # 阿里云 OSS 上传
pip install requests tqdm            # 触发 Dify + 进度条

2.3 关键算法说明（无模型版）

① 场景切换检测（PySceneDetect · AdaptiveDetector）

• 不依赖任何模型
• 对 PPT 翻页极其敏感（变化分 0.01 → 0.5 瞬间跳变）
• 对讲师走动、光影变化有免疫力

② 感知哈希去重（ImageHash · pHash）

1. 图片缩小至 32×32 → 灰度化
2. DCT 离散余弦变换
3. 取左上角 8×8 低频分量
4. 大于均值标 1，小于标 0 → 64 位哈希

③ PPT 类型过滤（OpenCV 规则）

• Canny 边缘密度（PPT 文字产生大量边缘）
• 颜色标准差（PPT 背景色单一）
• Hough 水平线检测（文字基线）

2.4 完整预处理脚本 preprocess.py

# ============================================================
# preprocess.py  ·  本地视频预处理脚本（零模型依赖）
# ============================================================
import os
import sys
import re
import json
import time
import subprocess
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed
import cv2
import numpy as np
import imagehash
import oss2
import requests
from PIL import Image
from tqdm import tqdm
from scenedetect import detect, AdaptiveDetector, open_video
from scenedetect.scene_manager import save_images
import imageio_ffmpeg
========== 配置区（修改为你的实际值）==========
CONFIG = {
"oss": {
"access_key": os.getenv("OSS_AK", "your_ak"),
"secret_key": os.getenv("OSS_SK", "your_sk"),
"endpoint": "https://oss-cn-shanghai.aliyuncs.com",
"bucket": "your-bucket-name",
"cdn_domain": "https://your-cdn.com"  # 可选 CDN
},
"dify": {
"api_url": "https://your-dify.com/v1/workflows/run",
"api_key": "app-xxxxxxxxxx"
},
"scene_detect": {
"adaptive_threshold": 3.0,       # 场景检测灵敏度
"min_scene_len": 30,              # 最小场景长度（帧）
"phash_threshold": 5,             # 去重汉明距离阈值
"filter_non_slide": True          # 是否过滤非PPT画面
},
"audio": {
"bitrate": "64k",                 # MP3 码率
"sample_rate": 16000              # 采样率（ASR 原生）
}
}
FFMPEG = imageio_ffmpeg.get_ffmpeg_exe()  # 自动下载 FFmpeg
========== 模块 1: 音频抽取 ==========
def extract_audio(video_path: str, out_path: str):
"""抽取 MP3 音频，体积比原视频小 100+ 倍"""
print("🎵 抽取音频...")
cmd = [
FFMPEG, '-i', video_path,
'-vn', '-ac', '1',
'-ar', str(CONFIG['audio']['sample_rate']),
'-b:a', CONFIG['audio']['bitrate'],
'-c:a', 'libmp3lame',
'-y', out_path
]
subprocess.run(cmd, check=True, capture_output=True)
size_mb = os.path.getsize(out_path) / 1024 / 1024
print(f"   ✅ 完成: {size_mb:.1f} MB")
return out_path
========== 模块 2: 关键帧检测（零模型）==========
def detect_scenes(video_path: str):
"""用 HSV 直方图差分算法检测场景切换（纯算法）"""
print("🔍 场景切换检测...")
cfg = CONFIG['scene_detect']
scene_list = detect(
video_path,
AdaptiveDetector(
adaptive_threshold=cfg['adaptive_threshold'],
min_scene_len=cfg['min_scene_len']
),
show_progress=True
)
print(f"   ✅ 检测到 {len(scene_list)} 个场景切换点")
return scene_list
def save_keyframes(video_path: str, scene_list, out_dir: str):
"""保存每个场景的首帧"""
os.makedirs(out_dir, exist_ok=True)
video = open_video(video_path)
save_images(
scene_list=scene_list, video=video,
num_images=1, output_dir=out_dir,
image_nametemplate='raw$SCENE_NUMBER',
show_progress=False
)
keyframes = []
for i, (start, end) in enumerate(scene_list):
    raw_path = os.path.join(out_dir, f'raw_{i+1:03d}.jpg')
    if os.path.exists(raw_path):
        keyframes.append({
            "index": i,
            "timestamp_sec": round(start.get_seconds(), 3),
            "timestamp_hms": str(start).split('.')[0],
            "duration_sec": round((end - start).get_seconds(), 3),
            "local_path": raw_path
        })
return keyframes
========== 模块 3: 感知哈希去重 ==========
def dedupe_by_phash(keyframes, threshold=5):
"""感知哈希去重（纯数学，无模型）"""
print("🔁 感知哈希去重...")
deduped, last_hash = [], None
for kf in tqdm(keyframes):
try:
img = Image.open(kf['local_path'])
phash = imagehash.phash(img)
if last_hash is None or (phash - last_hash) > threshold:
kf['phash'] = str(phash)
deduped.append(kf)
last_hash = phash
else:
os.remove(kf['local_path'])
except Exception as e:
print(f"   ⚠️ 跳过 {kf['local_path']}: {e}")
print(f"   ✅ {len(keyframes)} → {len(deduped)} 张")
return deduped
========== 模块 4: PPT 规则过滤（可选）==========
def is_slide_like(image_path: str) -> bool:
"""
规则判断是否像 PPT（三特征融合）

Canny 边缘密度
颜色标准差
水平直线数
"""
img = cv2.imread(image_path)
if img is None:
return False
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)

# 特征1: 边缘密度
edges = cv2.Canny(gray, 100, 200)
edge_ratio = np.sum(edges > 0) / edges.size

# 特征2: 颜色标准差
color_std = np.std(img)

# 特征3: 水平直线数
lines = cv2.HoughLinesP(edges, 1, np.pi/180, 
                         threshold=80, minLineLength=100, maxLineGap=10)
h_lines = 0
if lines is not None:
    h_lines = sum(1 for l in lines if abs(l[0][3] - l[0][1]) < 5)

return edge_ratio > 0.03 and color_std < 75 and h_lines > 3
def filter_non_slide(keyframes):
"""过滤非 PPT 画面（讲师人像等）"""
print("🎯 PPT 画面过滤...")
filtered = []
for kf in tqdm(keyframes):
if is_slide_like(kf['local_path']):
filtered.append(kf)
else:
os.remove(kf['local_path'])
print(f"   ✅ 保留 {len(filtered)} / {len(keyframes)} 张")
return filtered
========== 模块 5: OSS 批量上传 ==========
class OSSUploader:
def init(self):
auth = oss2.Auth(CONFIG['oss']['access_key'], CONFIG['oss']['secret_key'])
self.bucket = oss2.Bucket(auth, CONFIG['oss']['endpoint'], CONFIG['oss']['bucket'])
self.base_url = CONFIG['oss'].get('cdn_domain') or \
f"https://{CONFIG['oss']['bucket']}.{CONFIG['oss']['endpoint'].replace('https://','')}"
def upload(self, local_path: str, oss_key: str) -> str:
    """支持断点续传"""
    oss2.resumable_upload(
        self.bucket, oss_key, local_path,
        multipart_threshold=10*1024*1024,
        part_size=2*1024*1024,
        num_threads=4
    )
    return f"{self.base_url}/{oss_key}"

def upload_batch(self, items, oss_key_fn, max_workers=10):
    """并发批量上传"""
    def _up(item):
        key = oss_key_fn(item)
        url = self.upload(item['local_path'], key)
        item['image_url'] = url
        item.pop('local_path', None)
        return item

    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as ex:
        futures = [ex.submit(_up, item) for item in items]
        for f in tqdm(as_completed(futures), total=len(items)):
            results.append(f.result())
    return sorted(results, key=lambda x: x['index'])
========== 模块 6: 触发 Dify Workflow ==========
def trigger_dify(audio_url: str, manifest_url: str, video_name: str) -> dict:
"""调用 Dify Workflow API"""
print("🚀 触发 Dify Workflow...")
resp = requests.post(
CONFIG['dify']['api_url'],
headers={
'Authorization': f"Bearer {CONFIG['dify']['api_key']}",
'Content-Type': 'application/json'
},
json={
"inputs": {
"audio_url": audio_url,
"manifest_url": manifest_url,
"video_name": video_name
},
"response_mode": "blocking",
"user": "local-preprocess"
},
timeout=3600
)
resp.raise_for_status()
return resp.json()
========== 主流程 ==========
def main(video_path: str):
video_path = os.path.abspath(video_path)
video_name = Path(video_path).stem
workdir = Path(f"./output{video_name}")
work_dir.mkdir(exist_ok=True)
print(f"\n{'='*60}")
print(f"📹 处理视频: {video_path}")
print(f"📁 工作目录: {work_dir}")
print(f"{'='*60}\n")

t0 = time.time()

# 1. 抽音频
audio_path = work_dir / "audio.mp3"
extract_audio(video_path, str(audio_path))

# 2. 场景检测 + 关键帧保存
slides_dir = work_dir / "slides"
scene_list = detect_scenes(video_path)
keyframes = save_keyframes(video_path, scene_list, str(slides_dir))

# 3. 感知哈希去重
keyframes = dedupe_by_phash(keyframes, CONFIG['scene_detect']['phash_threshold'])

# 4. PPT 过滤（可选）
if CONFIG['scene_detect']['filter_non_slide']:
    keyframes = filter_non_slide(keyframes)

# 5. 上传 OSS
print("\n📤 上传产物到 OSS...")
uploader = OSSUploader()
audio_url = uploader.upload(str(audio_path), f"{video_name}/audio.mp3")
print(f"   ✅ 音频已上传: {audio_url}")

keyframes = uploader.upload_batch(
    keyframes,
    oss_key_fn=lambda k: f"{video_name}/slides/slide_{k['index']:04d}.jpg"
)
print(f"   ✅ {len(keyframes)} 张关键帧已上传")

# 6. 生成 manifest 并上传
manifest_path = work_dir / "keyframes.json"
manifest_path.write_text(
    json.dumps(keyframes, ensure_ascii=False, indent=2),
    encoding='utf-8'
)
manifest_url = uploader.upload(str(manifest_path), f"{video_name}/keyframes.json")
print(f"   ✅ 清单已上传: {manifest_url}")

# 7. 触发 Dify
result = trigger_dify(audio_url, manifest_url, video_name)

# 8. 保存结果
result_path = work_dir / "dify_result.json"
result_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')

# 9. 下载产物（Dify 返回的 MD/JSON/SRT URL）
outputs = result.get('data', {}).get('outputs', {})
for output_name, url in outputs.items():
    if isinstance(url, str) and url.startswith('http'):
        print(f"📥 下载 {output_name}...")
        resp = requests.get(url)
        ext = url.split('.')[-1].split('?')[0]
        (work_dir / f"{output_name}.{ext}").write_bytes(resp.content)

elapsed = time.time() - t0
print(f"\n{'='*60}")
print(f"✨ 全部完成！耗时 {elapsed/60:.1f} 分钟")
print(f"📂 产物目录: {work_dir.absolute()}")
print(f"{'='*60}\n")
if name == "main":
if len(sys.argv) < 2:
print("用法: python preprocess.py <视频路径>")
sys.exit(1)
main(sys.argv[1])

2.5 Windows 一键启动脚本 run.bat

@echo off
chcp 65001 >nul
if "%~1"=="" (
    echo 请将视频文件拖拽到本脚本上
    pause
    exit /b
)
python preprocess.py "%~1"
pause

三、Layer 2：Dify Workflow 设计

3.1 Workflow 输入

inputs:
  audio_url:        # OSS 音频 URL (80MB MP3)
  manifest_url:     # OSS 关键帧清单 JSON URL
  video_name:       # 视频名称

3.2 完整节点结构

[开始] audio_url, manifest_url, video_name
    │
    ├─── 【音频通道】────────────────────────────┐
    │                                              │
    │  ① [HTTP 节点] 调用阿里云 Paraformer-v2 ASR  │
    │     提交 → 轮询 → 返回 segments              │
    │                                              │
    │  ② [HTTP 节点] 批量调用 Embedding API         │
    │     (对所有 segment 文本向量化)               │
    │                                              │
    │  ③ [代码节点] 语义分段                       │
    │     (用 embedding 相似度切分话题块)          │
    │                                              │
    ├─── 【视觉通道】────────────────────────────┤
    │                                              │
    │  ④ [HTTP 节点 GET] 读取 manifest.json        │
    │                                              │
    │  ⑤ [迭代节点 · 并行] 对每张关键帧            │
    │     └─ [LLM 视觉节点] Qwen2.5-VL-Max         │
    │        (一次性完成: 判图类型+OCR+结构化)     │
    │                                              │
    │  ⑥ [代码节点] 幻灯片二次去重                 │
    │     (OCR 文本相似度 > 0.85 合并)             │
    │                                              │
    ├─── 【双通道融合】──────────────────────────┤
    │                                              │
    │  ⑦ [代码节点] 时间戳对齐融合                 │
    │     (为每个音频 chunk 匹配覆盖的 PPT)         │
    │                                              │
    │  ⑧ [代码节点] 规则去噪过滤                   │
    │     (正则过滤开场/推广/闲聊)                  │
    │                                              │
    │  ⑨ [迭代节点 · 并行] 对每个 chunk            │
    │     ├─ [LLM 节点] DeepSeek 噪声分类(A/B/C)   │
    │     ├─ [条件分支] category != C              │
    │     └─ [LLM 视觉节点] Qwen-VL-Max            │
    │        多模态融合提取知识点(附PPT图片)       │
    │                                              │
    ├─── 【聚合输出】────────────────────────────┤
    │                                              │
    │  ⑩ [HTTP] Embedding API (知识点向量化)       │
    │                                              │
    │  ⑪ [代码节点] 聚合去重                       │
    │     (相似度 > 0.85 合并，保留最早时间戳)     │
    │                                              │
    │  ⑫ [并行四分支输出]                          │
    │     ├─ [LLM 节点] 生成学习笔记 MD            │
    │     ├─ [代码节点] 生成结构化 JSON            │
    │     ├─ [代码节点] 生成 SRT 字幕              │
    │     └─ [代码节点] 生成可交互 HTML            │
    │                                              │
    │  ⑬ [HTTP 节点] 产物上传 OSS 返回 URL         │
    │                                              │
    └──► [结束] 返回 4 种产物 URL

3.3 核心节点代码（可直接放入 Dify 代码节点）

节点 ③：语义分段

def main(segments: list, embeddings: list) -> dict:
    """在语义断点处切分，保证每个 chunk 是完整话题"""
    import numpy as np
max_duration = 480      # 8分钟上限
min_duration = 120      # 2分钟下限
similarity_threshold = 0.55

chunks, current = [], [segments[0]]
for i in range(1, len(segments)):
    duration = current[-1]['end'] - current[0]['start']
    v1, v2 = np.array(embeddings[i-1]), np.array(embeddings[i])
    sim = float(np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2)))

    if duration >= max_duration or (duration >= min_duration and sim < similarity_threshold):
        chunks.append(current)
        current = [segments[i]]
    else:
        current.append(segments[i])
if current:
    chunks.append(current)

result = []
for idx, c in enumerate(chunks):
    result.append({
        "chunk_id": idx,
        "start": c[0]['start'],
        "end": c[-1]['end'],
        "text": " ".join(s['text'] for s in c),
        "segments": c
    })
return {"chunks": result}

节点 ⑦：时间戳对齐

def main(audio_chunks: list, slides: list) -> dict:
    """为每个音频 chunk 匹配覆盖其时间段的所有 PPT"""
    for chunk in audio_chunks:
        matched = []
        for slide in slides:
            slide_start = slide['timestamp_sec']
            slide_end = slide.get('end_sec', slide['timestamp_sec'] + slide.get('duration_sec', 30))
            overlap = min(chunk['end'], slide_end) - max(chunk['start'], slide_start)
            if overlap > 0:
                matched.append({**slide, "overlap": overlap})
        matched.sort(key=lambda x: x['overlap'], reverse=True)
        chunk['matched_slides'] = matched[:3]  # 最多取前3张
    return {"aligned_chunks": audio_chunks}

节点 ⑧：规则去噪

def main(chunks: list) -> dict:
    """三段式规则过滤"""
    import re
NOISE_PATTERNS = [
    r'^[嗯啊呃那个这个哦哈]+$',
    r'(大家好|欢迎来到|今天是\d+月|直播间的各位)',
    r'(点个关注|三连|加个微信|扫码|领取资料)',
    r'(喝口水|稍等|不好意思|网络卡|测试一下)'
]

filtered = []
for c in chunks:
    text = c['text'].strip()
    if len(text) < 20:
        continue
    noise_hits = sum(1 for p in NOISE_PATTERNS if re.search(p, text))
    if noise_hits >= 2:
        continue
    filtered.append(c)
return {"filtered_chunks": filtered}

节点 ⑪：知识点聚合去重

def main(kps: list, embeddings: list) -> dict:
    """基于向量相似度的层次聚类合并"""
    import numpy as np
threshold = 0.85
n = len(kps)
assigned = [False] * n
clusters = []

for i in range(n):
    if assigned[i]:
        continue
    cluster = [i]
    assigned[i] = True
    vi = np.array(embeddings[i])
    for j in range(i+1, n):
        if assigned[j]:
            continue
        vj = np.array(embeddings[j])
        sim = float(np.dot(vi, vj) / (np.linalg.norm(vi) * np.linalg.norm(vj)))
        if sim > threshold:
            cluster.append(j)
            assigned[j] = True
    clusters.append(cluster)

merged = []
for cluster in clusters:
    items = [kps[i] for i in cluster]
    primary = max(items, key=lambda x: x.get('importance', 3))
    timestamps = sorted(set((k['start_time_hms'], k['start_time_sec']) for k in items),
                       key=lambda x: x[1])
    slide_urls = list({k['source']['slide_image_url'] for k in items 
                      if k.get('source', {}).get('slide_image_url')})
    merged.append({
        **primary,
        "primary_timestamp_hms": timestamps[0][0],
        "primary_timestamp_sec": timestamps[0][1],
        "all_timestamps": [{"hms": t[0], "sec": t[1]} for t in timestamps],
        "all_slide_images": slide_urls,
        "mention_count": len(items)
    })

merged.sort(key=lambda x: x['primary_timestamp_sec'])
return {"merged_kps": merged}

节点 ⑫-③：SRT 字幕生成

def main(kps: list) -> dict:
    """生成带知识点标注的 SRT 字幕轨"""
    def sec_to_srt_time(sec):
        h = int(sec // 3600)
        m = int((sec % 3600) // 60)
        s = int(sec % 60)
        ms = int((sec - int(sec)) * 1000)
        return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
srt_lines = []
for i, k in enumerate(kps, 1):
    start = sec_to_srt_time(k['primary_timestamp_sec'])
    end = sec_to_srt_time(k['primary_timestamp_sec'] + 15)
    srt_lines.append(f"{i}\n{start} --> {end}\n【💡 知识点 {i}】{k['knowledge_point']}\n")
return {"srt_content": "\n".join(srt_lines)}

3.4 关键 Prompt 模板

Prompt A：Qwen-VL 判图+OCR+结构化

你是PPT内容识别专家。请分析这张直播截图。
第一步 判断画面类型:
A=幻灯片/PPT  B=代码演示  C=图表架构图  D=讲师人像  E=过渡/黑屏
第二步 若为A/B/C,提取结构化内容。
严格输出 JSON:
{
"page_type": "A|B|C|D|E",
"is_valuable": true|false,
"title": "主标题",
"bullets": ["要点1", "要点2"],
"formulas": ["LaTeX公式"],
"code_blocks": ["代码原文(保留缩进)"],
"tables": ["Markdown表格"],
"figure_description": "图形含义一句话",
"raw_text": "完整文字",
"confidence": 0.0-1.0
}
D/E类直接返回 {"page_type":"D","is_valuable":false}

Prompt B：多模态知识点融合提取（核心）

你是知识笔记整理专家。融合【讲师语音】+【PPT结构化OCR】+【PPT原图】提取结构化知识点。
【时间范围】{{chunk.start_hms}} ~ {{chunk.end_hms}}
【讲师语音】
{{chunk.text}}
【词级时间戳】
{{chunk.words_json}}
【同时段PPT】
{% for slide in chunk.matched_slides %}
PPT #{{loop.index}} @{{slide.timestamp_hms}}
标题: {{slide.title}}
要点: {{slide.bullets}}
公式: {{slide.formulas}}
代码: {{slide.code_blocks}}
原图: {{slide.image_url}}
{% endfor %}
【融合规则】

PPT标题作为知识点主题依据
语音补充讲师口头强调(常是考点)
公式/代码优先用PPT原文(OCR比ASR准)
语音与PPT矛盾以PPT为准,note字段标注
start_time 精确到知识点实际开始的那句话
无有效知识点返回 []

严格输出 JSON 数组:
[{
"knowledge_point": "标题(≤20字)",
"category": "概念|方法|案例|结论|公式|代码|工具",
"detail": "200-400字融合描述",
"formula_or_code": "PPT原文",
"source": {
"audio_evidence": "50字内语音引用",
"slide_evidence": "50字内PPT引用",
"slide_image_url": "对应截图URL"
},
"start_time_hms": "HH:MM:SS",
"start_time_sec": 数字,
"end_time_hms": "HH:MM:SS",
"end_time_sec": 数字,
"keywords": [],
"importance": 1-5
}]

四、Layer 3：云端 AI API 选型

4.1 最终选型表

五、工具选型与算法原理映射表

六、成本与性能评估

6.1 单视频端到端耗时（3 小时视频）

6.2 单视频成本

6.3 质量指标（目标）

• 专业术语准确率：98%+（ASR + PPT OCR 双通道互补）
• 公式/代码保真度：95%+（Qwen-VL 多模态识别）
• 知识点时间戳误差：< 3 秒（词级 ASR 对齐）
• 知识点可回看率：100%（全部带 PPT 截图 + 视频跳转链接）

七、部署架构（极简）

┌─────────────────────────────────────────┐
│ 本地电脑 (Windows/Mac/Linux)             │
│  - Python 3.8+                          │
│  - preprocess.py + 依赖 (200MB)         │
│  - 无 GPU 无模型                         │
└──────────────┬──────────────────────────┘
               │ 仅上传 110MB 音频+截图
               ▼
      ┌────────────────┐
      │ 阿里云 OSS      │
      └────────┬───────┘
               │
               ▼
┌─────────────────────────────────────────┐
│ Dify 服务器 (2核4G 起步)                 │
│  - Dify 本体 (Docker Compose)           │
│  - 无任何本地 AI 模型                     │
└──────────────┬──────────────────────────┘
               │ 全部 HTTP API 调用
   ┌───────────┴───────────────┐
   ▼                           ▼
┌──────────┐            ┌────────────┐
│阿里云百炼 │            │ OpenAI 等  │
│  Paraformer           │  GPT-4o    │
│  Qwen-VL-Max          │  Claude    │
│  Embedding            │            │
└──────────┘            └────────────┘

1. ✅ 本地机器安装 Python 3.8+ 和依赖（pip install 一次搞定）
2. ✅ 一台 Dify 服务器（Docker Compose 一键起）
3. ✅ 阿里云账号开通百炼、OSS
4. ✅ DeepSeek / OpenAI API Key
5. ❌ 无需 GPU 服务器
6. ❌ 无需自建 AI 服务

八、输出产物示例

8.1 学习笔记 Markdown

# 《Transformer原理精讲》学习笔记
视频时长: 03:25:40

知识点数: 42

生成时间: 2025-01-15
📑 索引



#
知识点
时间
重要度




1
Self-Attention计算公式
[00:15:32]
⭐⭐⭐⭐⭐


2
多头注意力机制
[00:22:18]
⭐⭐⭐⭐




1. Self-Attention 计算公式 ⭐⭐⭐⭐⭐
🕐 时间点: 00:15:32

🔁 重复讲解: [01:23:45] · [02:05:10]

📸 PPT 截图


🧮 公式 (来自PPT)

**📝 详细说明**  
Self-Attention 的核心是通过 Query、Key、Value 三个矩阵计算注意力权重...
💬 讲师原话

> 除以根号 d_k 是为了防止 softmax 后的梯度消失,这是面试常考点
🏷 关键词: 注意力机制 · Transformer · 面试考点

8.2 结构化 JSON

{
  "video_meta": {
    "video_name": "Transformer原理精讲",
    "total_duration_sec": 12340,
    "duration_hms": "03:25:40",
    "knowledge_points_count": 42
  },
  "knowledge_points": [{
    "id": 1,
    "title": "Self-Attention 计算公式",
    "category": "公式",
    "detail": "...",
    "formula_or_code": "Attention(Q,K,V) = softmax(QK^T/√d_k)V",
    "primary_timestamp": {"hms": "00:15:32", "sec": 932},
    "all_timestamps": [
      {"hms": "00:15:32", "sec": 932},
      {"hms": "01:23:45", "sec": 5025}
    ],
    "slide_images": ["https://oss.../slide_0012.jpg"],
    "slide_ocr_text": "Self-Attention 机制...",
    "audio_quote": "除以根号 d_k 是为了防止梯度消失",
    "keywords": ["注意力", "Transformer"],
    "importance": 5,
    "mention_count": 3
  }]
}

8.3 SRT 字幕轨（可挂载到播放器）

1
00:15:32,000 --> 00:15:47,000
【💡 知识点 1】Self-Attention 计算公式
2
00:22:18,000 --> 00:22:33,000
【💡 知识点 2】多头注意力机制

8.4 可交互 HTML 回看页面

┌─────────────────────────────┬─────────────────┐
│                             │  📚 知识点列表   │
│   视频播放器 (Plyr.js)       │  1. Self-Atten  │
│   点击时间跳转              │     📸[缩略图]  │
│                             │     🕐 15:32    │
│                             │  2. 多头注意力  │
│                             │     📸[缩略图]  │
│                             │     🕐 22:18    │
└─────────────────────────────┴─────────────────┘

九、避坑清单

十、实施路线图（2 周极简版）

Week 1：基础设施 + 本地工具

• Day 1：部署 Dify（Docker Compose），开通阿里云百炼/OSS
• Day 2-3：编写并调试 preprocess.py，5 分钟短视频跑通
• Day 4-5：调参（场景阈值 3.0、pHash 阈值 5、PPT 过滤规则）
• Day 6-7：批量测试 30 分钟视频，优化上传稳定性

Week 2：Dify 工作流 + 联调

• Day 8-9：搭建 Dify Workflow（音频通道 + 视觉通道）
• Day 10：时间戳对齐、去噪、聚合代码节点
• Day 11-12：调优多模态融合 Prompt（3 轮迭代）
• Day 13：4 种输出格式 + HTML 模板
• Day 14：3 小时真实视频全流程验证 + 成本评估

十一、交付清单

十二、核心优势总结

十三、核心设计哲学

“碰到文件、进程、硬件资源” → 本地跑 “纯数据变形、调 API” → Dify 跑 “AI 推理” → 云端 API 跑

• 本地：有文件系统权限，有 FFmpeg，没有模型依赖 —— 适合做预处理
• Dify：有编排能力，有代码沙箱，有 HTTP 网关 —— 适合做流程控制
• 云端：有 GPU 算力，有 SOTA 模型，有弹性扩容 —— 适合做 AI 推理

方案结束。

如需下一步产出，可任选：

1. 📦 preprocess.py 完整可执行版（含 GUI 进度条、断点续传、异常恢复、日志系统）
2. 📋 Dify Workflow DSL YAML 文件（可直接导入 Dify）
3. 🎨 HTML 回看页面模板（Plyr.js + Vue3，带 PPT 缩略图和点击跳转）
4. 🔧 PyInstaller 打包脚本（生成 Windows 单文件 exe，非技术同事可用）
5. 📊 不同场景的调参对比表（CS 课/数学课/商科课 的阈值推荐）