一、问题背景：课程制作的"最后一公里"
企业培训平台上最不缺的就是视频。内部讲师录制、外部供应商采购、直播回放沉淀——每个渠道都在持续产出视频内容。但视频内容的"可消费性"很差：员工想快速了解一门课讲了什么，只能拖着进度条听；讲师上传课程后，还要手动撰写课程简介、章节摘要、知识点列表，一门 60 分钟的课程光整理这些就要 2-3 小时。
这个"最后一公里"的问题带来三个直接影响：课程上架慢（讲师拖延整理就不上架）、学习体验差（没有结构化信息，员工找不到重点）、运营成本重（内容团队人力有限，课程量一大就崩）。
我们的目标很明确：给定一段培训视频，自动生成课程章节摘要、核心知识点列表、以及可独立消费的知识卡片，整个过程尽量少人工干预。
二、整体架构：四阶段流水线
整个系统可以拆成四个串联阶段，每个阶段独立可替换：
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ Stage 1 │───▶│ Stage 2 │───▶│ Stage 3 │───▶│ Stage 4 │
│ 音频提取 │ │ 语音识别 │ │ 智能分段 │ │ 知识抽取 │
│ & 预处理 │ │ (ASR) │ │ & 清洗 │ │ & 卡片生成│
└──────────┘ └──────────┘ └──────────┘ └──────────┘
FFmpeg Whisper / 文本预处理 + LLM Prompt
音频标准化 华为云ASR 语义分段 Chain
阿里云ASR
Stage 1 — 音频提取与预处理。 用 FFmpeg 从视频文件中抽取音频轨，统一转为 16kHz 单声道 WAV 格式。这一步同时做降噪处理（低频滤波 + 音量归一化），减少后续 ASR 的识别错误。
Stage 2 — 语音识别。 将音频送入 ASR 引擎，输出带时间戳的逐句转录文本。可选方案包括开源的 Whisper 系列和云厂商的 ASR 服务（华为云语音交互、阿里云智能语音）。
Stage 3 — 智能分段与清洗。 原始 ASR 输出存在标点缺失、口语冗余、段落边界模糊等问题。这一阶段做文本清洗，然后基于语义相似度进行智能分段，还原课程的章节结构。
Stage 4 — 知识抽取与卡片生成。 将分段后的文本送入 LLM，通过精心设计的 Prompt Chain 完成摘要生成、知识点提取、知识卡片结构化输出。
下面逐阶段展开，重点讲工程细节和核心代码。
三、Stage 1：音频提取与预处理
视频文件的格式五花八门（MP4、AVI、MOV、FLV），音频编码也各不相同。我们需要统一抽取为标准格式，同时做基础的音频增强。
import subprocess
import os
def extract_audio(video_path: str, output_dir: str) -> str:
"""从视频文件中提取音频，转为 16kHz 单声道 WAV"""
os.makedirs(output_dir, exist_ok=True)
audio_path = os.path.join(
output_dir,
os.path.splitext(os.path.basename(video_path))[0] + ".wav"
)
cmd = [
"ffmpeg", "-i", video_path,
"-vn", # 不要视频
"-acodec", "pcm_s16le", # 16-bit PCM
"-ar", "16000", # 16kHz 采样率（ASR 标准输入）
"-ac", "1", # 单声道
"-af", "highpass=f=80,lowpass=f=8000,loudnorm=I=-16:TP=-1.5",
# highpass 去低频噪音，lowpass 去高频噪音，loudnorm 做响度归一化
"-y",
audio_path
]
subprocess.run(cmd, check=True, capture_output=True)
return audio_path
几个工程细节值得说明。采样率选 16kHz 而不是 44.1kHz，是因为主流 ASR 模型的训练数据都是 16kHz，更高的采样率不会提升识别率反而增加计算量。响度归一化用 EBU R128 标准（loudnorm 滤镜），把不同录制设备的音量差异压到可接受范围。如果原始视频噪音严重，可以额外接入 noisereduce 库做频谱减法降噪。
对于超过 2 小时的长视频，建议在这里做一刀——按固定时长（比如 30 分钟）切成多段，避免后续 ASR 单次请求超时。
def split_audio(audio_path: str, segment_minutes: int = 30) -> list[str]:
"""将长音频按固定时长切段"""
duration = float(subprocess.run(
["ffprobe", "-v", "error", "-show_entries", "format=duration",
"-of", "default=noprint_wrappers=1:nokey=1", audio_path],
capture_output=True, text=True
).stdout.strip())
segment_seconds = segment_minutes * 60
segments = []
start = 0
idx = 0
while start < duration:
seg_path = audio_path.replace(".wav", f"_seg{idx}.wav")
subprocess.run([
"ffmpeg", "-i", audio_path,
"-ss", str(start), "-t", str(min(segment_seconds, duration - start)),
"-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1",
"-y", seg_path
], check=True, capture_output=True)
segments.append(seg_path)
start += segment_seconds
idx += 1
return segments
四、Stage 2：语音识别（ASR）
ASR 是整个流水线的质量瓶颈——如果转录文本本身错字连篇，后面 LLM 再强也救不回来。选型时我们评估了两条路线。
开源方案：Whisper。 OpenAI 的 Whisper 模型在中文场景表现不错，large-v3 模型在内部测试集上的字错率（CER）约 8-12%。优势是免费、可私有化部署、数据不出域。劣势是推理速度慢——一段 60 分钟的音频在单张 A10 GPU 上需要 8-12 分钟。
云服务方案：华为云 / 阿里云 ASR。 两家都提供带时间戳的长音频转写服务，CER 在 5-8% 左右（针对培训场景的通用口语）。优势是快（60 分钟音频通常 2-3 分钟出结果）、省心。劣势是按量计费，且有数据出域的合规考量。
我们的实际做法是混合策略：对合规要求高的客户用 Whisper 私有化部署，一般场景用云服务。下面给出两种方案的对接代码。
4.1 Whisper 本地推理
import whisper
import torch
def transcribe_with_whisper(audio_path: str) -> list[dict]:
"""
使用 Whisper large-v3 进行转录
返回: [{"text": "...", "start": 0.0, "end": 3.5}, ...]
"""
device = "cuda" if torch.cuda.is_available() else "cpu"
model = whisper.load_model("large-v3", device=device)
result = model.transcribe(
audio_path,
language="zh",
task="transcribe",
verbose=False,
word_timestamps=True # 开启词级时间戳
)
segments = []
for seg in result["segments"]:
segments.append({
"text": seg["text"].strip(),
"start": round(seg["start"], 2),
"end": round(seg["end"], 2),
})
return segments
生产环境建议用 faster-whisper（基于 CTranslate2 推理框架），速度比原版快 4 倍左右，显存占用降低一半：
from faster_whisper import WhisperModel
def transcribe_fast(audio_path: str) -> list[dict]:
model = WhisperModel("large-v3", device="cuda", compute_type="float16")
segments_iter, info = model.transcribe(
audio_path, language="zh",
word_timestamps=True,
vad_filter=True # 启用 VAD，跳过静音段
)
segments = []
for seg in segments_iter:
segments.append({
"text": seg.text.strip(),
"start": round(seg.start, 2),
"end": round(seg.end, 2),
})
return segments
vad_filter=True 值得强调——培训视频中讲师停顿、翻页、切换 PPT 的静音段可能占总时长的 20-30%，开启 VAD 后这些段会被直接跳过，显著减少无效推理。
4.2 华为云 ASR 对接
import requests
import time
class HuaweiCloudASR:
"""华为云语音交互 - 录音文件识别"""
def __init__(self, region: str, project_id: str, ak: str, sk: str):
self.region = region
self.project_id = project_id
self.ak = ak
self.sk = sk
self.base_url = f"https://sis.{region}.myhuaweicloud.com"
def submit_task(self, audio_url: str) -> str:
"""提交录音文件识别任务，返回 task_id"""
url = f"{self.base_url}/v1/{self.project_id}/asr/recit-bgm/tasks"
body = {
"config": {
"audio_format": "wav",
"property": "chinese_8k_common", # 中文通用
"add_punc": True, # 自动加标点
"digit_norm": True, # 数字归一化
},
"data": audio_url # OBS 上的音频 URL
}
resp = requests.post(url, json=body, auth=self._sign())
return resp.json()["task_id"]
def poll_result(self, task_id: str, timeout: int = 600) -> list[dict]:
"""轮询任务结果"""
url = f"{self.base_url}/v1/{self.project_id}/asr/recit-bgm/tasks/{task_id}"
start = time.time()
while time.time() - start < timeout:
resp = requests.get(url, auth=self._sign()).json()
if resp["status"] == "SUCCEEDED":
return self._parse_result(resp["result"])
elif resp["status"] == "FAILED":
raise RuntimeError(f"ASR task failed: {resp}")
time.sleep(5)
raise TimeoutError(f"ASR task {task_id} timed out")
def _parse_result(self, result: dict) -> list[dict]:
"""将华为云返回格式统一为标准 segment 列表"""
segments = []
for item in result.get("segments", []):
segments.append({
"text": item["text"],
"start": item["start_time"] / 1000, # ms -> s
"end": item["end_time"] / 1000,
})
return segments
阿里云 ASR 的对接流程类似，区别主要在认证方式和 API 路径。建议封装一个统一的 ASREngine 接口，让两种后端可无缝切换：
from abc import ABC, abstractmethod
class ASREngine(ABC):
@abstractmethod
def transcribe(self, audio_path_or_url: str) -> list[dict]:
"""返回 [{"text": str, "start": float, "end": float}, ...]"""
...
class WhisperEngine(ASREngine):
def __init__(self, model_size: str = "large-v3"):
self.model_size = model_size
def transcribe(self, audio_path: str) -> list[dict]:
# ... 上面的 Whisper 实现
pass
class HuaweiCloudEngine(ASREngine):
def __init__(self, region, project_id, ak, sk):
self.client = HuaweiCloudASR(region, project_id, ak, sk)
def transcribe(self, audio_url: str) -> list[dict]:
task_id = self.client.submit_task(audio_url)
return self.client.poll_result(task_id)
五、Stage 3：智能分段与文本清洗
ASR 输出的原始文本有几个典型问题需要处理。
第一，标点缺失或错误。Whisper 的中文输出经常缺少逗号句号，或者在不该断句的地方加了标点。第二，口语冗余。"嗯""啊""那个""就是说"这类填充词在培训视频中尤其多。第三，段落边界模糊。ASR 按语音停顿分段，但一个语义完整的段落可能被拆成多个 ASR segment，或者多个话题被合并在一个 segment 里。
5.1 文本清洗
import re
def clean_transcript(segments: list[dict]) -> str:
"""清洗 ASR 原始输出，返回干净的全文"""
texts = [seg["text"] for seg in segments]
full_text = "".join(texts)
# 1. 去除常见口语填充词
fillers = [
r"嗯+", r"啊", r"那个", r"就是说", r"然后呢",
r"对吧", r"是不是", r"怎么说呢", r"额", r"呃"
]
for filler in fillers:
full_text = re.sub(filler, "", full_text)
# 2. 合并连续空格
full_text = re.sub(r"\s+", " ", full_text).strip()
# 3. 修复常见 ASR 错误（可根据业务积累持续补充）
corrections = {
"PPT": "PPT",
"ppt": "PPT",
"API": "API",
"api接口": "API 接口",
}
for wrong, right in corrections.items():
full_text = full_text.replace(wrong, right)
return full_text
5.2 基于语义相似度的智能分段
清洗后的文本是一整块，需要按语义切分成段落。简单的做法是按固定字数切——每 500 字一段，但这样经常在句子中间断开。更好的做法是利用 Embedding 的语义相似度找到话题转换点。
from sentence_transformers import SentenceTransformer
import numpy as np
class SemanticSegmenter:
"""基于语义相似度的文本分段器"""
def __init__(self, model_name: str = "shibing624/text2vec-base-chinese"):
self.model = SentenceTransformer(model_name)
def segment(
self,
text: str,
sentence_splitter: str = None,
min_chars: int = 100,
similarity_threshold: float = 0.65
) -> list[dict]:
"""
将长文本按语义切分为段落。
思路：先按句拆分，计算相邻句的 Embedding 余弦相似度，
相似度低于阈值的位置就是分段点。
"""
# 按标点分句
sentences = re.split(r'([。！？；\n]+)', text)
# 合并标点到前一句
merged = []
for i in range(0, len(sentences) - 1, 2):
merged.append(sentences[i] + (sentences[i+1] if i+1 < len(sentences) else ""))
if len(sentences) % 2 == 1 and sentences[-1].strip():
merged.append(sentences[-1])
# 过滤过短的句子
sentences = [s for s in merged if len(s.strip()) >= 5]
if len(sentences) <= 1:
return [{
"text": text, "sentences": sentences}]
# 计算相邻句的语义相似度
embeddings = self.model.encode(sentences, normalize_embeddings=True)
similarities = []
for i in range(len(embeddings) - 1):
sim = np.dot(embeddings[i], embeddings[i+1])
similarities.append(sim)
# 在相似度低于阈值的位置分段
segments = []
current_segment = [sentences[0]]
for i, sim in enumerate(similarities):
next_sent = sentences[i + 1]
current_text = "".join(current_segment)
if sim < similarity_threshold and len(current_text) >= min_chars:
# 话题转换点，结束当前段落
segments.append({
"text": current_text,
"sentences": list(current_segment)
})
current_segment = [next_sent]
else:
current_segment.append(next_sent)
# 最后一段
if current_segment:
segments.append({
"text": "".join(current_segment),
"sentences": current_segment
})
return segments
similarity_threshold 是关键参数。设太低（比如 0.5）会导致几乎所有内容都在同一段里；设太高（比如 0.85）又会切得太碎。我们在内部课程数据集上测试，0.6-0.7 是比较好的区间。如果你的课程主题跨度大（比如一门课同时讲技术和管理），可以适当调低阈值。
六、Stage 4：知识抽取与卡片生成
这是整个流水线中最依赖 Prompt 工程的阶段。我们需要 LLM 完成三件事：生成章节摘要、提取核心知识点、输出结构化的知识卡片。
直接用一个巨大的 Prompt 做所有事情效果不好——LLM 在多任务混合 Prompt 下容易"偷懒"，漏掉某些知识点或者摘要质量下降。更好的做法是拆成 Prompt Chain，每一步专注一个任务，上一步的输出作为下一步的输入。
6.1 Prompt Chain 设计
from openai import OpenAI
class KnowledgeExtractor:
"""基于 LLM 的知识抽取 Pipeline"""
def __init__(self, model: str = "gpt-4o", base_url: str = None, api_key: str = None):
self.client = OpenAI(base_url=base_url, api_key=api_key)
self.model = model
def run_chain(self, segments: list[dict]) -> dict:
"""
执行完整的知识抽取 Chain：
Step 1: 为每个段落生成结构化摘要
Step 2: 从摘要中提取知识点
Step 3: 生成知识卡片
"""
# Step 1: 章节摘要
chapter_summaries = []
for i, seg in enumerate(segments):
summary = self._summarize_chapter(seg["text"], chapter_index=i+1)
chapter_summaries.append(summary)
# Step 2: 汇总后提取知识点
all_summary_text = "\n\n".join(
f"【第{s['chapter']}章】{s['summary']}"
for s in chapter_summaries
)
knowledge_points = self._extract_knowledge_points(all_summary_text)
# Step 3: 生成知识卡片
cards = self._generate_cards(knowledge_points)
return {
"chapters": chapter_summaries,
"knowledge_points": knowledge_points,
"cards": cards,
}
def _summarize_chapter(self, text: str, chapter_index: int) -> dict:
"""Step 1: 为单个章节段落生成摘要"""
prompt = f"""你是一位企业培训课程的内容编辑。请为以下课程内容段落生成结构化摘要。
要求：
1. 用一句话概括本段核心内容（不超过 30 字）
2. 列出本段涉及的 2-5 个关键概念或术语
3. 用 3-5 个要点总结本段的核心论述（每个要点不超过 25 字）
4. 如果本段包含实操演示或案例，简要说明
课程内容：
{text}
请以 JSON 格式输出：
{
{
"chapter": {chapter_index},
"one_line_summary": "...",
"key_concepts": ["...", "..."],
"key_points": ["...", "...", "..."],
"has_demo": true/false,
"demo_description": "..."
}}"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{
"role": "user", "content": prompt}],
temperature=0.3, # 低温度保证摘要的确定性
response_format={
"type": "json_object"},
)
import json
return json.loads(response.choices[0].message.content)
def _extract_knowledge_points(self, all_summary: str) -> list[dict]:
"""Step 2: 从全局摘要中提取结构化知识点"""
prompt = f"""你是一位企业培训课程的知识管理专家。基于以下课程各章节摘要，提取本课程的核心知识点。
要求：
1. 每个知识点必须是独立、可自解释的知识单元
2. 标注知识点类型：概念（concept）、流程（process）、方法（method）、工具（tool）、案例（case）
3. 标注重要程度：核心（必须掌握）、重要（建议掌握）、了解（拓展认知）
4. 为每个知识点写一句解释（不超过 40 字）
课程摘要：
{all_summary}
请以 JSON 数组格式输出：
[
{
{
"id": "kp_01",
"title": "...",
"type": "concept|process|method|tool|case",
"importance": "核心|重要|了解",
"explanation": "...",
"related_chapters": [1, 2]
}}
]"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{
"role": "user", "content": prompt}],
temperature=0.2,
response_format={
"type": "json_object"},
)
import json
result = json.loads(response.choices[0].message.content)
return result if isinstance(result, list) else result.get("knowledge_points", [])
def _generate_cards(self, knowledge_points: list[dict]) -> list[dict]:
"""Step 3: 将知识点转化为可独立消费的知识卡片"""
import json
kp_text = json.dumps(knowledge_points, ensure_ascii=False, indent=2)
prompt = f"""你是一位企业培训内容设计师。请将以下知识点转化为适合员工在手机上快速浏览的知识卡片。
每张卡片包含：
- card_title：吸引眼球的标题（不超过 15 字）
- core_message：卡片核心内容（2-3 句话，不超过 100 字）
- takeaway：一句话行动建议（学完这个知识点，员工应该做什么）
- scenario：适用场景（什么时候会用到这个知识）
知识点列表：
{kp_text}
请以 JSON 数组格式输出卡片列表。"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{
"role": "user", "content": prompt}],
temperature=0.5, # 稍高温度让卡片文案更有变化
response_format={
"type": "json_object"},
)
result = json.loads(response.choices[0].message.content)
return result if isinstance(result, list) else result.get("cards", [])
6.2 对接国产大模型
如果部署环境不方便调用 OpenAI，可以用华为云 ModelArts 的盘古大模型或阿里云的通义千问替代。以通义千问为例，只需要替换 base_url 和 model：
# 通义千问（DashScope 兼容 OpenAI 接口）
extractor = KnowledgeExtractor(
model="qwen-max",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key="your-dashscope-api-key"
)
# 华为云盘古大模型（如果有 OpenAI 兼容端点）
extractor = KnowledgeExtractor(
model="pangu-large",
base_url="https://pangu-modelarts.{region}.myhuaweicloud.com/v1",
api_key="your-pangu-api-key"
)
Prompt 不需要改动，因为三个模型对中文 JSON 格式 Prompt 的理解能力都在同一水平线上。
七、流水线编排与异步处理
上面的代码是同步的、一步一步跑的。生产环境需要改成异步流水线——用户上传视频后不用等着，后台处理完推送通知。
我们用消息队列做阶段间的解耦，整体编排如下：
import asyncio
from dataclasses import dataclass
from enum import Enum
class TaskStatus(Enum):
PENDING = "pending"
EXTRACTING_AUDIO = "extracting_audio"
TRANSCRIBING = "transcribing"
SEGMENTING = "segmenting"
EXTRACTING_KNOWLEDGE = "extracting_knowledge"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class VideoProcessTask:
task_id: str
video_path: str
status: TaskStatus = TaskStatus.PENDING
error: str = None
class VideoProcessingPipeline:
"""视频处理流水线编排"""
def __init__(self, asr_engine: ASREngine, extractor: KnowledgeExtractor):
self.asr_engine = asr_engine
self.extractor = extractor
async def process(self, task: VideoProcessTask) -> dict:
"""异步执行完整流水线"""
try:
# Stage 1: 音频提取
task.status = TaskStatus.EXTRACTING_AUDIO
audio_path = await asyncio.to_thread(
extract_audio, task.video_path, f"/tmp/{task.task_id}"
)
# Stage 2: 语音识别
task.status = TaskStatus.TRANSCRIBING
segments = await asyncio.to_thread(
self.asr_engine.transcribe, audio_path
)
# Stage 3: 文本清洗 + 智能分段
task.status = TaskStatus.SEGMENTING
full_text = clean_transcript(segments)
segmenter = SemanticSegmenter()
text_segments = await asyncio.to_thread(
segmenter.segment, full_text
)
# Stage 4: 知识抽取
task.status = TaskStatus.EXTRACTING_KNOWLEDGE
result = await asyncio.to_thread(
self.extractor.run_chain, text_segments
)
task.status = TaskStatus.COMPLETED
return result
except Exception as e:
task.status = TaskStatus.FAILED
task.error = str(e)
raise
生产环境中，asyncio.to_thread 应该替换为真正的分布式任务队列（Celery、RQ、华为云 FunctionGraph 等），每个 Stage 独立为一个 Worker，通过消息队列传递中间结果。这样做的好处是：ASR 阶段可以用 GPU 机器，LLM 阶段可以用 CPU 机器，资源按需分配；某个 Stage 失败只需重试该阶段，不用从头跑。
八、质量优化：三个关键经验
流水线跑通不难，难的是把质量做到可交付。以下是我们在实际运营中积累的三条核心优化经验。
经验一：ASR 结果的人工抽检不能省。 完全自动化的流水线意味着没有人在中间检查 ASR 质量。我们采取的策略是：每批处理结果随机抽检 10%，如果某门课的 ASR 字错率超过 15%，自动标记为"需人工校对"，同时触发告警排查是音频质量问题还是 ASR 模型问题。
经验二：Prompt 需要按课程类别调优。 技术类课程（编程、运维、架构）和软技能类课程（沟通、管理、领导力）的知识结构差异很大。技术课的知识点偏向概念和流程，软技能课偏向方法和案例。我们为这两类课程维护了不同的 Prompt 模板，通过课程标签自动路由。
经验三：知识卡片需要"冷却"机制。 LLM 生成的卡片初稿有时会出现重复、过于笼统、或者与原文偏离的情况。我们在 Step 3 之后加了一个自校验步骤——把生成的卡片和原文一起送回 LLM，让它判断每张卡片是否准确反映了原文内容：
def validate_cards(self, cards: list[dict], original_text: str) -> list[dict]:
"""用 LLM 自校验知识卡片的准确性"""
import json
prompt = f"""请校验以下知识卡片是否准确反映了原始课程内容。
对每张卡片判断：
- accurate：卡片内容准确，与原文一致
- vague：卡片过于笼统，缺乏具体信息
- inaccurate：卡片内容与原文不符
- duplicate：与其他卡片内容重复
原始课程内容（摘要）：
{original_text[:3000]}
知识卡片：
{json.dumps(cards, ensure_ascii=False, indent=2)}
请输出修正后的卡片列表，标记每张卡片的校验状态，对 vague 和 inaccurate 的卡片给出修正建议。"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{
"role": "user", "content": prompt}],
temperature=0.1,
response_format={
"type": "json_object"},
)
return json.loads(response.choices[0].message.content)
只保留 accurate 状态的卡片进入最终交付，其余的要么修正后重新校验，要么直接丢弃。这一步大约会过滤掉 10-15% 的低质量卡片，但显著提升了最终交付物的可信度。
九、成本与性能参考
以一门 60 分钟的中文培训视频为例，给出我们实测的成本和耗时数据：
| 阶段 | 耗时 | 成本（云服务方案） | 成本（私有化方案） |
|---|---|---|---|
| 音频提取 | ~10s | 0 | 0 |
| ASR 转录 | ~3min | ¥1.5-3.0 | GPU 算力约 ¥0.8 |
| 文本分段 | ~5s | 0 | 0 |
| LLM 知识抽取 | ~2min | ¥0.5-1.5 | 视模型而定 |
| 合计 | ~6min | ¥2-4.5 | 约 ¥0.8 |
对比人工整理一门课需要 2-3 小时（内容编辑人力成本约 ¥100-200），自动化方案的成本降低了 95% 以上，速度提升了 30 倍。
十、总结与展望
本文介绍了一套 ASR + LLM 的四阶段流水线，实现了企业培训视频的自动摘要和知识点提取。核心思路是：用 ASR 解决"视频转文本"，用语义分段解决"文本结构化"，用 Prompt Chain 解决"知识抽取与卡片生成"。每个阶段独立可替换，可以根据实际场景灵活组合。
后续我们计划探索几个方向：一是引入多模态模型（如 GPT-4o 的视觉能力），直接从视频帧中提取 PPT 内容和板书信息，弥补纯音频路线的信息损失；二是增加课程间的知识图谱关联，让员工学完一门课后能自然跳转到相关课程的知识卡片；三是结合员工的学习行为数据，动态调整知识卡片的展示优先级，实现"千人千面"的课后复习推荐。
企业培训的内容生产不缺量，缺的是让内容变得可消费、可检索、可复用的结构化能力。AI 流水线正好补上这一环。
作者简介：企学宝技术团队，专注于企业培训平台的架构设计与技术实现，关注全球化、AI、多语言等技术领域。
