适用场景
当你需要产出一段几十秒到几分钟的连续内容（长视频为主，长音频、长序列同理），而单次生成又有明确时长上限时，这套方案把长目标拆成若干段分别生成，再按统一风格、首尾帧衔接和固定参数拼回一条完整成片。适合内容团队批量出片、运营做产品演示片、以及需要把生成成本算清楚的工程团队——它解决的问题不是“能不能生成”，而是“分段之后不跳戏、不重复花钱”。
环境与前置条件
- 操作系统：Linux（推荐 Ubuntu 22.04 及以上）、macOS，或 Windows + WSL2。Windows 原生终端在路径与 ffmpeg 调用上容易踩坑，建议直接用 WSL2。
- 运行时：Python 3.10 及以上（具体最低版本以官方 SDK 文档当前版本为准），pip 可用。
- 内存/磁盘：内存 8 GB 起步，处理 1080p 素材建议 16 GB；磁盘按“成片时长 × 每分钟约 200–600 MB 中间文件”估算，一段 3 分钟的项目留 20 GB 比较稳妥。本项目全部走云端推理，本地不需要 GPU。
- 外部依赖：ffmpeg / ffprobe（拼接、裁剪重叠、抽帧质检）。用包管理器安装即可。
- 账号与凭据：具备调用权限的 API Key，通过环境变量注入，不要写进代码或提交到 Git。
- 网络：需要稳定出网；如果处在受限网络，提前配好代理环境变量。
关于模型 ID、单次生成时长上限、支持的分辨率与帧率、是否支持固定种子和首帧参考，一律以官方文档当前版本为准——这四个参数直接决定分段策略，开始前先查清楚再动手。
分步骤部署
第 1 步：建项目骨架与虚拟环境
这一步做的是把“一次性的脚本”变成可重复运行的项目，后面每段产物、日志、缓存都有固定位置。
```bash
mkdir -p longform && cd longform
python3 -m venv .venv && source .venv/bin/activate
mkdir -p pipeline out cache logs
touch pipeline/__init__.py requirements.txt .env config.yaml
```
成功标志：命令行前缀出现 (.venv)，ls 能看到 pipeline/ out/ cache/ logs/ 四个目录。
第 2 步：安装 SDK 与 ffmpeg
```bash
依赖写在 requirements.txt 里，锁版本，方便回滚
python -m pip install --upgrade pip
python -m pip install pyyaml python-dotenv
模型 SDK 按官方文档给出的包名安装，并在 requirements.txt 中锁定版本
python -m pip install "<官方 SDK 包名>"
ffmpeg
sudo apt-get update && sudo apt-get install -y ffmpeg # Debian/Ubuntu
macOS: brew install ffmpeg
```
成功标志：ffmpeg -version 与 ffprobe -version 都能打印出版本信息；python -c "import yaml, dotenv" 无报错。
第 3 步：写配置文件与密钥文件
把“会变的东西”全部外置。分段参数、成本参数、模型参数分开放，改成本的时候不用动代码。
.env（不要提交到仓库，加进 .gitignore）：
```
API_KEY=你的密钥
MODEL_ID=gemini-omni-1.2-flash # 模型 ID 以官方文档当前版本为准，这里只作占位
```
config.yaml：
```yaml
model:
model_id_env: MODEL_ID
resolution: "1280x720" # 先用低分辨率试跑，定稿再提精度
fps: 24
seed: 20240101 # 模型支持固定种子时才有意义
first_frame_reference: true # 是否支持首帧参考，以官方文档为准
segment:
total_seconds: 60 # 目标成片时长
segment_seconds: 8 # 必须 <= 官方单次时长上限
overlap_seconds: 1 # 相邻段重叠，用于衔接
max_retries: 2
cost:
price_per_generated_second: 0 # 填入你自己的计费口径，或留 0 只看时长
retry_rate_assumed: 0.15
accept_rate_assumed: 0.8
runtime:
concurrency: 2 # 并发越高越容易触发限流
out_dir: out
cache_dir: cache
```
成功标志：python -c "import yaml,os;print(yaml.safe_load(open('config.yaml'))['segment'])" 能打印出分段配置。
第 4 步：实现分段规划器
这是整套方案的核心逻辑：把 60 秒拆成“每段 8 秒、相邻重叠 1 秒”的列表。重叠的作用是给衔接留余量——上一段的最后一帧就是下一段的起点。
pipeline/planner.py：
```python
def plan_segments(total_seconds: float, segment_seconds: float, overlap_seconds: float):
if segment_seconds <= overlap_seconds:
raise ValueError("segment_seconds 必须大于 overlap_seconds")
step = segment_seconds - overlap_seconds
segments, start, idx = [], 0.0, 0
while start < total_seconds - 1e-6:
end = min(start + segment_seconds, total_seconds)
segments.append({
"index": idx,
"start": round(start, 3),
"end": round(end, 3),
"duration": round(end - start, 3),
"trim_tail": round(overlap_seconds, 3) if end < total_seconds else 0.0,
})
if end >= total_seconds:
break
start += step
idx += 1
return segments
```
验证一下逻辑：
```bash
python -c "
from pipeline.planner import plan_segments
s = plan_segments(60, 8, 1)
print(len(s), '段'); print(s[0]); print(s[-1])
print('总生成时长', sum(x['duration'] for x in s))
"
```
预期输出：9 段，首段 start 0.0 / end 8.0 / duration 8.0 / trim_tail 1.0，末段 start 56.0 / end 60.0 / duration 4.0 / trim_tail 0.0，总生成时长 68.0 秒。注意这 8 秒的“额外开销”就是重叠成本，后面算账要用。
第 5 步：写 Provider 适配层，先打通单段
把所有模型调用收敛到一个文件，其他模块只依赖固定签名。这样官方 SDK 参数变化时只改这一处。
pipeline/provider.py：
```python
import os, json, hashlib, pathlib
class Provider:
def __init__(self, model_id: str, api_key: str, cache_dir: str = "cache"):
self.model_id = model_id
self.api_key = api_key
self.cache_dir = pathlib.Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
def _cache_key(self, prompt: str, duration: float, seed, first_frame: str | None) -> str:
raw = json.dumps([self.model_id, prompt, duration, seed, first_frame], ensure_ascii=False)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
def generate(self, prompt: str, duration: float, out_path: str,
seed=None, first_frame: str | None = None):
"""单段生成。缓存命中直接复用，避免调参时重复付费。"""
key = self._cache_key(prompt, duration, seed, first_frame)
cached = self.cache_dir / f"{key}.bin"
if cached.exists():
pathlib.Path(out_path).write_bytes(cached.read_bytes())
return {"cached": True, "path": out_path}
==== 此处按官方 SDK 文档实现一次生成调用 ====
1) 传入 prompt、时长、分辨率/帧率、seed、首帧参考图（若支持）
2) 拿到返回的媒体字节流或下载链接
3) 写入 out_path，并把同一份内容写入 cached 作缓存
raise NotImplementedError("按官方 SDK 在此实现单段生成并落盘")
```
成功标志：python -c "from pipeline.provider import Provider; print('ok')" 无报错，且 cache/ 存在。
第 6 步：设计跨段一致的提示词模板
跨段“跳戏”的根源通常不是模型不行，而是每段提示词各写各的。做法是把提示词拆成“全局锚 + 本段任务 + 衔接约束 + 禁止项”四块，只有“本段任务”逐段变化。
```python
ANCHOR = (
"【全局锚｜每段原样复制】"
"主体：<角色/产品的完整外观描述>；"
"场景：<地点、光源方向、时间>；"
"影像：<画幅、色调、机位高度、镜头焦段、运动方式>；"
)
def build_prompt(i: int, total: int, anchor: str, action: str, prev_end_state: str, next_start_state: str) -> str:
return (
f"{anchor}\n"
f"【本段任务】第 {i+1}/{total} 段。起始状态：{prev_end_state}。"
f"本段需完成：{action}。\n"
f"【衔接约束】首帧延续参考图；结尾停在「{next_start_state}」，为下一段留起手动作。\n"
f"【禁止项】不出现字幕与文字、不切换场景、不改变主体服装与配色、不做快速变焦。"
)
```
分段边界的选取有个实用原则：切在动作停顿处，不要切在动作进行中。比如“人物走到门口 → 停下 → 转身”，边界放在“停下”之后；如果切在“走到一半”，下一段的首帧几乎必然接不上。重叠 1 秒是保守值，动作越快、镜头运动越剧烈，重叠要适当加大。
成功标志：print(build_prompt(0, 9, ANCHOR, "推开木门进入室内", "门外站立", "门开到一半")) 输出四段结构完整的提示词。
第 7 步：编排主流程——生成、抽尾帧、衔接下一段
pipeline/run.py 的关键片段（保留核心逻辑，命令行参数自行补全）：
```python
import os, subprocess, pathlib, yaml
from dotenv import load_dotenv
from pipeline.planner import plan_segments
from pipeline.provider import Provider
load_dotenv()
cfg = yaml.safe_load(open("config.yaml", encoding="utf-8"))
seg_cfg = cfg["segment"]
out = pathlib.Path(cfg["runtime"]["out_dir"]); out.mkdir(exist_ok=True)
plan = plan_segments(seg_cfg["total_seconds"], seg_cfg["segment_seconds"], seg_cfg["overlap_seconds"])
provider = Provider(os.environ[cfg["model"]["model_id_env"]], os.environ["API_KEY"], cfg["runtime"]["cache_dir"])
prev_tail = None
for s in plan:
p = out / f"seg_{s['index']:03d}.mp4"
prompt = build_prompt(s["index"], len(plan), ANCHOR, ACTIONS[s["index"]], STATES[s["index"]], STATES[s["index"] + 1])
provider.generate(prompt, s["duration"], str(p), seed=cfg["model"]["seed"], first_frame=prev_tail)
抽最后一帧，作为下一段的首帧参考
tail = out / f"tail_{s['index']:03d}.jpg"
subprocess.run(["ffmpeg", "-y", "-sseof", "-0.2", "-i", str(p),
"-frames:v", "1", "-q:v", "2", str(tail)], check=True)
prev_tail = str(tail)
按计划裁掉尾部重叠，避免拼接后出现重复画面
if s["trim_tail"] > 0:
trimmed = out / f"trim_{s['index']:03d}.mp4"
subprocess.run(["ffmpeg", "-y", "-i", str(p), "-t", f"{s['duration'] - s['trim_tail']:.3f}",
"-c", "copy", str(trimmed)], check=True)
```
要点：每段生成完立刻抽尾帧，不要等全部生成完再回头抽；重叠在拼接前裁掉，而不是靠播放器“感觉差不多”。
成功标志：out/ 下出现成对的 seg_000.mp4 / tail_000.jpg，段数与上一步规划一致。
第 8 步：拼接与统一编码参数
```bash
生成拼接清单，路径用绝对路径，避免 concat 的相对路径解析问题
ls "$PWD"/out/trim_*.mp4 | sort | sed "s|^|file '|; s|$|'|" > out/concat.txt
ls "$PWD"/out/seg_*.mp4 > /dev/null # 最后一段若未裁剪，记得补进清单末尾
ffmpeg -y -f concat -safe 0 -i out/concat.txt -c copy out/final.mp4
```
如果拼接后出现音画不同步或时长异常，说明各段编码参数不一致，改用重编码方式：
```bash
ffmpeg -y -f concat -safe 0 -i out/concat.txt \
-c:v libx264 -pix_fmt yuv420p -r 24 -c:a aac -ar 48000 out/final.mp4
```
成功标志：ffprobe -v error -show_entries format=duration -of csv=p=0 out/final.mp4 输出的数值与 total_seconds 的偏差在 0.2 秒以内。
验证部署是否成功
按顺序跑这四条检查，全部通过再进入批量生产。
1）时长校验
```bash
ffprobe -v error -show_entries format=duration -of csv=p=0 out/final.mp4
ffprobe -v error -select_streams v:0 -show_entries stream=width,height,r_frame_rate -of csv=p=0 out/final.mp4
```
预期：时长约等于配置的 total_seconds（误差 0.2 秒内），分辨率与帧率与 config.yaml 一致。
2）接缝校验（重点）
在每个拼接点前后各截 3 秒，肉眼过一遍：
```bash
假设第一个接缝在第 7 秒
ffmpeg -y -ss 4 -t 6 -i out/final.mp4 -c copy out/seam_01.mp4
```
预期：画面主体、服装、光线方向、机位高度不发生突变；允许轻微的构图漂移，但不接受“换人”“换场景”“跳色调”。
3）重叠区一致性校验（可选，量化用）
用 SSIM 对比“上一段尾部”和“下一段头部”的重叠画面，分数越高说明衔接越稳：
```bash
ffmpeg -i out/seg_000.mp4 -i out/seg_001.mp4 -filter_complex \
"[0:v]trim=start=7,setpts=PTS-STARTPTS[a];[1:v]trim=end=1,setpts=PTS-STARTPTS[b];[a][b]ssim" -f null -
```
预期：日志里打出 All: 的 SSIM 值，把这个数值记成基线，之后每次改提示词都对比一次，避免“改一处、坏一片”。
4）成本预估表
用规划结果直接算出三段账，先跑 dry-run 不调 API：
```python
from pipeline.planner import plan_segments
plan = plan_segments(60, 8, 1)
generated = sum(s["duration"] for s in plan) # 含重叠的实际生成时长
overlap_ratio = generated / 60 - 1
print(f"段数={len(plan)} 生成时长={generated}s 重叠开销={overlap_ratio:.1%}")
```
预期输出：段数=9 生成时长=68.0s 重叠开销=13.3%。
常见报错与解决
报错：401 Unauthorized / 403 PERMISSION_DENIED，提示 API key 无效或未开通
→ 原因：.env 没有被加载，或密钥复制时带了空格/换行，或该密钥所属项目未开通此模型权限。
→ 解决：
```bash
python -c "import os;from dotenv import load_dotenv;load_dotenv();print(bool(os.getenv('API_KEY')), len(os.getenv('API_KEY','')))"
输出应为 True 加正确长度；为空则检查 .env 是否在运行目录下
export API_KEY="重新粘贴的密钥" # 临时验证是否密钥本身问题
```
报错：429 RESOURCE_EXHAUSTED / rate limit exceeded
→ 原因：并发数超过配额，或短时间内连续重试把配额打满。
→ 解决：把 runtime.concurrency 降到 1，并在重试逻辑里加指数退避，同时给重试次数封顶（配置里的 max_retries: 2），避免失败一段就把预算烧光。
```yaml
runtime:
concurrency: 1
segment:
max_retries: 2
```
```python
import time
def backoff(attempt: int):
time.sleep(min(2 ** attempt, 30))
```
报错：400 INVALID_ARGUMENT，提示时长/分辨率超出支持范围
→ 原因：segment_seconds 或 resolution 超过了官方单次上限，或帧率不被支持。
→ 解决：查官方文档当前版本的参数范围，把 segment_seconds 下调（比如 8 → 6），分辨率降到 720p 先跑通链路，再按需提升。下调后记得重跑第 4 步的规划校验，段数会变多，成本表要重算。
报错：ffmpeg ... unsafe file name 或拼接后只有第一段
→ 原因：concat.txt 里用了相对路径，或 -safe 0 没加。
→ 解决：清单改用绝对路径，命令里带 -safe 0，并检查清单行格式必须是 file '/abs/path/xxx.mp4'。
现象：成片在接缝处画面“倒回去一小段”
→ 原因：重叠帧没有在拼接前裁掉。
→ 解决：确认每个非末段都执行了 -t (duration - overlap) 裁剪，且 concat.txt 里引用的是 trim_*.mp4 而不是 seg_*.mp4。
后续维护
备份：至少备份三样东西——config.yaml、每段的 prompt 文本、以及一个 manifest.json（记录每段的模型 ID、seed、时长、首帧参考文件名、生成时间、是否命中缓存）。成片文件可以重生成，提示词和参数丢了就得从头调。建议按项目日期分目录归档。
升级：SDK 与模型版本都可能变，升级前先锁版本跑一遍小样本（比如 3 段 24 秒），对比接缝 SSIM 与时长偏差，指标没退化再全量切。模型 ID 变更时，在 config.yaml 里改一处即可，不要散落在代码里。
日志与监控：每段至少记录四项指标——生成耗时、重试次数、命中缓存与否、失败原因。这三项指标值得做成看板：单段失败率、接缝返工率、以及下面这个最关键的指标。
单位时长成本：这是长片段项目唯一该盯的成本口径。算法是
```
有效成本/成片秒 = 名义每秒单价 × (1 + 重叠开销率) × (1 + 重试率) ÷ 成片采纳率
```
代入本教程的参数：重叠开销 13.3%、重试率 15%、采纳率 80%，则 1.13 × 1.15 ÷ 0.8 ≈ 1.63——也就是说账面上的每秒单价，实际要乘 1.6 倍左右才是真实成本。想压成本，优先级从高到低是：提高一次通过率（把全局锚和禁止项写扎实）> 降低重试率 > 减少重叠秒数 > 只重生成出问题的那一段而不是整条。分辨率与帧率按最终投放渠道定，预览阶段一律用低档。
清理策略：cache/ 和 out/ 里的中间文件增长很快，设一个保留期（比如 14 天）定时清理，只保留 manifest.json、最终成片和关键尾帧。别让磁盘在批量任务跑到一半时满掉——那是最难排查的一类失败。