用大模型Agent和FFmpeg复刻Grok剪辑Bot：一句话生成混剪视频

Grok 剪辑 Bot 是一个典型的大模型 Agent 落地项目用户在手机聊天窗口里发一句话后台 Bot 接收消息解析剪辑意图准备素材再由 FFmpeg 完成混剪成片最后把视频原路回传。它把“手机一句话生成混剪成片”这个看似神奇的需求拆成了消息接入、意图解析、素材收集、视频渲染、结果回传这样几个普通工程环节。这篇文章会以复刻和二次开发这个开源项目的角度把完整技术链路、最小实现、部署验证、常见排查和生产级改造写清楚。适合做过基础 Python 服务、想了解大模型如何接入实际业务的开发者也适合负责视频生产工具链的同学拿去设计自动化剪辑服务。为什么要强调“开源”因为这个项目本身的核心逻辑并不复杂复杂的是从手机消息到成片之间的中间链路。直接阅读开源代码时新手容易在三个地方迷失不知道 Bot 框架的回调入口在哪里、不知道大模型返回的 JSON 怎么变成 FFmpeg 参数、不知道服务器上为什么能解析却合成失败。这篇文章不重复粘贴整个仓库源码而是把最小可复刻的骨架拆出来讲。1. 先理解“一句话剪辑”的本质工作链路1.1 核心场景手机端一句话后台完成整条剪辑链路假设用户在手机聊天窗口里发送这样一句话“把素材库里春日出游的片段剪成 15 秒竖屏视频加淡入淡出配字幕。”后端 Bot 收到这句话后需要自动完成理解这句话里的剪辑意图主题是“春日出游”成片时长是 15 秒画面比例是竖屏需要淡入淡出效果需要字幕。从素材库中选出符合条件的素材片段。对素材做切片、缩放、拼接、转场、字幕叠加。输出一个可直接播放的 MP4 文件。把文件通过聊天接口回传给用户。这里没有任何“魔法”。用户的一句话只是入口真正干活的是大模型的意图解析能力加上 FFmpeg 的视频处理能力。这类项目的价值在于它把视频剪辑从“手动打开剪辑软件、拖时间线、调参数”变成“用自然语言描述结果”适合批量生成短视频素材、快速剪辑活动混剪、做内容平台的自动化生产工具。1.2 整体架构与数据流整个系统可以简化成一条单向数据流用户消息 - Bot 接入层 - 语音转文字(可选) - 大模型意图解析 - 结构化 JSON - 素材收集 - FFmpeg 渲染 - 成片回传每一步都会产生中间产物Bot 接入层产生消息事件。语音消息需要先转为文本。大模型意图解析产生一个包含主题、时长、素材来源、效果列表的 JSON。素材收集模块根据 JSON 去本地目录、远程地址或素材索引中找文件。FFmpeg 根据 JSON 生成剪辑命令完成视频合成。最后把成片上传到 IM 平台或对象存储再发送给用户。理解这条链路后排错就会变得有方向。前端发不出消息先查 Bot 接入层解析结果不符合预期先查大模型返回的 JSON合成失败先查 FFmpeg 命令。1.3 为什么选择 Bot 而不是自建 App这个项目选择 Bot 形态是有实际理由的。第一跨平台成本低。用户不需要安装特定 App手机上任意一个能发消息的 IM 客户端都能作为操作入口。第二服务端升级方便。剪辑逻辑全部在服务端更新模型、调整滤镜、增加渲染能力不需要用户更新客户端。第三消息天然适合异步任务。用户发一条指令后后台可以慢慢处理处理完再回传结果。这正好匹配视频渲染这种耗时操作。第四调试门槛低。开发阶段用文本消息就能触发全流程不需要处理移动端 UI、登录、审核等问题。如果换成网页或 App还要额外考虑前端状态管理、视频上传、任务进度轮询等问题。Bot 形态把交互成本降到最低开发者可以把主要精力放在核心的意图解析和视频合成链路上。2. 环境准备与依赖先把本机跑通2.1 软件依赖和版本建议写代码前先确认本机工具链。下面这些依赖是这套项目的基础缺一个都会在运行时报错。依赖版本建议用途Python3.10 及以上运行 Bot 服务pip最新稳定版安装 Python 依赖FFmpeg4.4 及以上视频解码、转码、拼接、滤镜python-telegram-bot20.xTelegram Bot 框架openai最新稳定版调用 OpenAI 兼容的大模型接口python-dotenv最新稳定版读取 .env 配置文件版本确认时以官方最新稳定版为准。不同版本之间如果有接口差异优先看对应版本的官方文档。2.2 获取 Bot Token 和大模型 API Key在 Telegram 中通过 BotFather 创建一个机器人拿到形如123456:ABC...的 Bot Token。这个 Token 是 Bot 进程与 Telegram 服务器通信的凭证必须妥善保管不要提交到公开仓库。大模型侧需要准备一个支持 OpenAI 兼容协议的 API Key 和 base_url。如果项目使用的是 Grok 这类模型需要向服务商确认接口地址、模型名称、上下文长度和计费方式。不同服务商的接口参数不完全一样落地前必须以官方文档为准。2.3 安装依赖并检查 FFmpeg在项目根目录执行python -m venv .venv source .venv/bin/activate pip install -r requirements.txt检查 FFmpeg 是否可用ffmpeg -version如果命令找不到说明 FFmpeg 没有安装。Linux 环境可以使用系统包管理器安装macOS 可以用 HomebrewWindows 需要下载安装包并配置 PATH。安装完成后重新执行ffmpeg -version能看到版本号和编译参数说明视频处理工具链就绪。2.4 初始化配置使用.env文件保存环境变量。参考模板TELEGRAM_BOT_TOKEN123456:ABC... LLM_API_KEYyour_api_key LLM_BASE_URLhttps://api.example.com/v1 LLM_MODELgrok-xxx RAW_MATERIAL_DIRassets/raw OUTPUT_DIRassets/outputs MAX_DURATION60 MAX_FILE_SIZE_MB50各项配置含义配置项含义建议TELEGRAM_BOT_TOKENBot Token从 BotFather 获取LLM_API_KEY大模型接口密钥从服务商获取LLM_BASE_URL大模型接口地址以服务商文档为准LLM_MODEL模型名称使用实际可用模型RAW_MATERIAL_DIR原始素材目录项目内相对路径即可OUTPUT_DIR成片输出目录需要保证可写MAX_DURATION成片最大时长建议 60防止资源耗尽MAX_FILE_SIZE_MB成片最大体积建议 50受 IM 平台限制这些配置并不复杂但有一个常见问题很多人把路径写成相对路径服务启动后工作目录一变素材就找不到了。后面所有涉及路径的代码尽量用项目根目录拼绝对路径。3. 项目结构与最小实现先跑通从消息到响应的闭环3.1 目录设计推荐的最小目录结构grok-clip-bot/ ├── bot/ │ ├── __init__.py │ ├── main.py │ ├── handlers.py │ ├── planner.py │ ├── clip_engine.py │ └── tasks.py ├── assets/ │ ├── raw/ │ └── outputs/ ├── config.py ├── requirements.txt └── .env.example各文件职责main.pyBot 入口负责启动服务、注册处理器。handlers.py消息回调处理接收用户文本和语音。planner.py调用大模型把一句话解析为结构化指令。clip_engine.py封装 FFmpeg 命令完成视频合成。tasks.py任务队列和异步任务管理。config.py读取环境变量统一管理配置。assets/raw原始素材目录。assets/outputs成片输出目录。先不要追求代码量把这个结构跑通再逐步扩展。3.2 最小可运行的 Bot 入口先写一个只负责收发消息的main.py目标是“用户发什么Bot 回什么”。这样能快速验证 Bot 接入层是否正常。import os from dotenv import load_dotenv from telegram import Update from telegram.ext import Application, CommandHandler, MessageHandler, filters load_dotenv() BOT_TOKEN os.getenv(TELEGRAM_BOT_TOKEN) async def start(update: Update, context): await update.message.reply_text(剪辑 Bot 已就绪发送剪辑指令即可。) async def echo(update: Update, context): await update.message.reply_text(f收到{update.message.text}) def main(): app Application.builder().token(BOT_TOKEN).build() app.add_handler(CommandHandler(start, start)) app.add_handler(MessageHandler(filters.TEXT ~filters.COMMAND, echo)) app.run_polling() if __name__ __main__: main()这里的关键点是app.run_polling()。这是本地调试最方便的方式通过长轮询不断向 Telegram 服务器拉取消息。生产环境可以换成 Webhook后面会讲。3.3 先建立“空跑链路”消息进来、日志记录、响应回去很多时候项目失败不是因为大模型不够强而是因为接入层都没有跑通。所以第一步先做空跑链路不接大模型不接 FFmpeg只验证消息能进来、日志能记录、回复能发出去。启动命令python -m bot.main看到控制台输出 Bot 启动日志后在手机或桌面端给 Bot 发/start再发一条普通文本。如果收到回复说明 Bot 接入层正常。这一步做完后再往echo回调里加日志import logging logging.basicConfig( levellogging.INFO, format%(asctime)s %(levelname)s %(message)s ) logger logging.getLogger(__name__) async def echo(update: Update, context): user_text update.message.text chat_id update.effective_chat.id logger.info(message received, chat_id%s, text%s, chat_id, user_text) await update.message.reply_text(f收到{user_text})空跑链路的意义在于缩小排错范围。此时如果消息收发有问题问题一定出在 Bot Token、网络、启动方式或代理配置上与后面的剪辑逻辑无关。4. 核心模块拆解意图解析、素材收集和剪辑合成4.1 意图解析把一句话变成结构化 JSON用户说的话是自然语言FFmpeg 需要的是参数。中间的转换层就是大模型意图解析。核心思路是固定输出 Schema。这样无论用户怎么说下游代码读到的都是同样结构的 JSON。planner.py示例import json import os from openai import OpenAI client OpenAI( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL), ) def parse_edit_plan(user_text: str) - dict: prompt f 你是短视频混剪指令解析器。把用户的需求解析为结构化 JSON只输出 JSON不要解释。 用户指令{user_text} 输出字段 - theme: 主题词 - duration: 期望成片时长整数秒 - source_type: local / url / keyword - source_value: 素材来源目录路径、链接或搜索关键词 - effects: 滤镜或转场效果列表例如 [fade] - with_music: true 或 false - subtitles: true 或 false resp client.chat.completions.create( modelos.getenv(LLM_MODEL), messages[{role: user, content: prompt}], temperature0, ) content resp.choices[0].message.content.strip() # 有些模型会把 JSON 包在代码块里需要清理 if content.startswith(): content content.strip() if content.startswith(json): content content[4:] return json.loads(content)关键点有两个。第一temperature0让模型输出尽量稳定不要每次生成不同的 JSON 结构。第二提示词里明确要求“只输出 JSON不要解释”。即便如此模型仍可能在 JSON 外面加代码块标记所以代码里要做清理。例如用户输入“把春日出游的素材剪成 15 秒竖屏视频加淡入淡出”期望输出{ theme: spring, duration: 15, source_type: local, source_value: spring, effects: [fade], with_music: false, subtitles: false }下游代码只需要读取这个 JSON不需要理解自然语言。4.2 素材收集本地目录、远程链接和版权边界素材来源有三种常见方式。第一种是本地素材库。按主题建立子目录例如assets/raw/spring存放春日出游相关素材。这里需要注意素材文件命名要规范最好用主题加序号例如spring_001.mp4。本地素材扫描函数import glob import os from pathlib import Path def collect_local_materials(source_value: str | None) - list[str]: raw_dir Path(os.getenv(RAW_MATERIAL_DIR)) search_dir raw_dir / source_value if source_value else raw_dir files glob.glob(str(search_dir / *.mp4)) files glob.glob(str(search_dir / *.mov)) return sorted(files)这里有一个坑如果传入的source_value是用户可控的直接拼路径可能造成路径穿越。生产环境必须对source_value做白名单校验或者只允许读取素材库内部目录。第二种是远程链接。用户给出视频 URL服务端先下载到临时目录再进入剪辑流程。下载前必须确认素材来源允许下载和使用不能把版权不明确的素材直接接入生产系统。第三种是关键词搜索。如果接入了免版权素材站或自建素材索引可以用关键词去查询素材。这种方案更复杂但更接近真实的自动化剪辑产品。无论哪种来源下载或扫描完成后都要做一次完整性检查至少确认文件存在、大小不为 0、能被 FFprobe 正常解析。4.3 剪辑引擎用 FFmpeg 完成切片、拼接和滤镜FFmpeg 是这里最核心的依赖。混剪的基础操作包括单素材转竖屏scale加pad。多素材拼接concatdemuxer 或xfade转场。加字幕drawtext或字幕文件。调时长trim和setpts。以两个素材合成为例生成 1080x1920 竖屏视频中间加 1 秒淡入淡出转场ffmpeg -y \ -i assets/raw/spring_001.mp4 \ -i assets/raw/spring_002.mp4 \ -filter_complex \ [0:v]scale1080:1920:force_original_aspect_ratiodecrease,pad1080:1920:(ow-iw)/2:(oh-ih)/2,trimduration6,setptsPTS-STARTPTS[v0];\ [1:v]scale1080:1920:force_original_aspect_ratiodecrease,pad1080:1920:(ow-iw)/2:(oh-ih)/2,trimduration6,setptsPTS-STARTPTS[v1];\ [v0][v1]xfadetransitionfade:duration1:offset5,formatyuv420p[vout] \ -map [vout] -c:v libx264 -preset veryfast -crf 23 \ assets/outputs/demo.mp4这个命令的解释scale1080:1920:force_original_aspect_ratiodecrease保持原比例缩放到能完全放入 1080x1920 画布的最大尺寸。pad1080:1920:(ow-iw)/2:(oh-ih)/2把缩放后的画面居中放到画布中。trimduration6每个片段截取 6 秒。setptsPTS-STARTPTS重置时间戳避免拼接后时间线错乱。xfadetransitionfade:duration1:offset5在第 5 秒开始执行 1 秒转场。Python 侧不建议直接拼 shell 字符串应该使用参数列表方式调用 FFmpeg避免复杂的引号转义和安全问题。示例import asyncio import os import time async def build_clip(material_paths: list[str], output_name: str) - str: output_path os.path.join(os.getenv(OUTPUT_DIR), output_name) cmd [ffmpeg, -y] for path in material_paths: cmd [-i, path] filter_complex build_filter_complex(len(material_paths)) cmd [-filter_complex, filter_complex] cmd [-map, [vout], -c:v, libx264, -preset, veryfast, -crf, 23] cmd [output_path] process await asyncio.create_subprocess_exec( *cmd, stdoutasyncio.subprocess.PIPE, stderrasyncio.subprocess.PIPE ) _, stderr await process.communicate() if process.returncode ! 0: raise RuntimeError(stderr.decode(errorsignore)) return output_pathbuild_filter_complex需要根据素材数量动态生成滤镜字符串这属于容易出错的部分。建议把素材数量限制在 5 个以内先验证 2 个素材的场景再逐步扩展。FFmpeg 常用参数速查表参数含义常用值说明crf画质控制23越大画质越低文件越小preset编码速度veryfast越快 CPU 占用越低体积略大maxrate最大码率4M防止成片体积过大bufsize编码缓冲8M配合 maxrate 使用r帧率30生产环境固定帧率更稳定4.4 结果回传与临时文件清理FFmpeg 产出视频文件后Bot 需要把文件发给用户。python-telegram-bot 的回传方式import io async def send_video(update: Update, video_path: str): with open(video_path, rb) as f: await update.message.reply_video(videoio.BytesIO(f.read()))这个写法适合本地验证和小文件场景。生产环境不建议把整个视频读进内存更好的做法是先将文件上传到对象存储再发送文件 URL 或文件 ID。视频文件会占磁盘空间。每天积累下来assets/outputs会越来越大。建议在任务结束时