做短视频内容最耗时间的往往不是拍摄本身而是从选题、脚本、分镜到剪辑指令这条链路。每个环节都需要不同角色协作工具之间又是割裂的文案在文档里素材在网盘里剪辑在专业软件里提示词在对话框里。最近在调研和实验 AI 视频创作方案时发现把 MCP 作为工具接入标准再用多智能体系统把策划、脚本、视觉、剪辑、审阅这些角色串成流水线确实能把创作流程压缩到一个可编排、可复现的工程框架里。这篇文章会把整套流程拆开讲清楚先解释 MCP 和多智能体的核心概念再给出视频创作场景下的系统架构设计然后动手实现一个 MCP Server 和多个 Agent最后串联成一条完整流水线并附上常见问题和工程建议。适合有一定 AI 应用开发基础、想在视频内容生产场景里落地 Agent 的开发者也适合对 MCP 感兴趣、想了解它到底怎么用的朋友。1. 背景与核心概念1.1 视频创作流程中的效率瓶颈一条短视频从创意到成片通常要经历这么几个阶段选题策划确定视频主题、目标人群、内容结构。脚本撰写产出口播文案、画面描述、字幕文本。视觉设计规划分镜、镜头语言、画面参考。素材准备调用图片、视频片段、音频、字幕资源。剪辑合成按分镜顺序生成剪辑指令或工程文件。审阅修改检查内容质量、情绪节奏、合规风险。传统做法里这些环节依赖人工在不同软件之间切换信息传递靠文档和口头沟通。即使引入了 AI 工具也只是“单点辅助”用 ChatGPT 写脚本、用 Midjourney 出图、用剪映剪辑各工具之间没有统一的数据交互标准更谈不上多角色协作。如果能有一个系统让不同的 AI 助手分别扮演策划、编剧、视觉设计、剪辑师、审阅官并且它们能调用同一套视频工具集那创作效率的提升会是质的改变。这就引出了两个关键技术点MCP 和多智能体系统。1.2 MCP 是什么模型上下文协议MCP 全称是 Model Context Protocol即“模型上下文协议”。它的作用是解决大语言模型与外部工具、数据源连接时的标准化问题。在没有 MCP 之前每个应用要对接模型和工具都得自己写一套集成代码。比如一个 Agent 要调用数据库、搜索网页、读取本地文件就得分别实现数据库驱动、搜索 API 客户端、文件读写模块。集成方式五花八门换一个模型或换一个工具就要重新适配。MCP 的架构里多了几个标准角色MCP Host运行模型的宿主程序比如 Claude Desktop、Dify、自研的 Agent 应用。MCP ClientHost 内部负责与 MCP Server 通信的组件。MCP Server对外提供工具、资源、提示词的服务工具能力以标准协议暴露。可以把它理解成“AI 世界的 USB 接口”。设备厂商不用关心你的电脑是什么系统只要设备支持 USB 协议就能接入同样MCP Server 只要实现标准协议任何支持 MCP 的宿主都能调用它的工具。MCP 在 2024 年底正式开源后社区生态发展非常快。目前已经有大量现成的 MCP Server覆盖了数据库访问、浏览器自动化如 Playwright、设计稿标注如 Figma、蓝湖、代码托管、搜索引擎等场景。这意味着 Agent 可以主动去连接无限多的外部能力而不局限于模型训练时的静态知识。1.3 多智能体系统的核心思想多智能体系统Multi-Agent System简称 MAS是一套把多个有独立角色和目标的 Agent 组织起来协作完成复杂任务的架构。核心思想是“拆”把一个大任务拆解成多个子任务每个子任务交给最擅长它的 Agent再由调度器或工作流把结果串起来。单 Agent 处理复杂任务时容易遇到几个问题上下文过长。所有历史信息都塞进同一个对话窗口既费 token 又容易超出限制。角色冲突。让同一个模型既当策划又当审阅它很难真正自我批判。工具混乱。一个 Agent 需要关注太多工具调用时容易选错。多智能体系统则把这些问题分而治之每个 Agent 只负责一个相对独立的角色拥有自己的 system prompt、上下文窗口和工具集。Agent 之间传递的是结构化结果而不是漫无边际的对话历史。常见的多智能体协作模式有三种流水线模式一个 Agent 的输出是下一个 Agent 的输入适合流程固定的场景。协作模式多个 Agent 并行工作最后汇总结果。博弈模式一个 Agent 提出方案另一个 Agent 负责挑错再由裁判 Agent 做最终裁决适合需要质量把关的场景。短视频创作流程天然适合流水线加博弈的混合模式这一点后面会详细展开。1.4 MCP 与多智能体的组合优势MCP 解决的是“工具如何被标准化调用”的问题多智能体解决的是“多个角色如何分工协作”的问题两者是互补关系。有了 MCP多智能体系统里的每个 Agent 才能以统一方式调用视频创作所需的外部工具。否则就会出现同时存在 OpenAI 格式、LangChain 格式、自研 API 格式的混乱局面。有了多智能体MCP Server 里的工具不会被一个庞大的、无所不能的 Agent 独占而是按角色分发给不同 Agent让整个系统的可维护性和可扩展性大幅提升。2. 整体架构设计从创意到成片的流水线2.1 视频创作流程拆解把视频创作映射到多智能体系统最关键的一步是明确流程节点和每个节点的输入输出。一条可自动化的视频创作流水线可以拆成阶段输入输出对应 Agent选题策划主题关键词、用户画像内容大纲、切入角度Planner Agent脚本撰写内容大纲口播文案、分镜文本Script Agent视觉设计脚本内容画面描述、分镜提示词Visual Agent剪辑调度分镜与素材信息剪辑指令、素材清单Editor Agent质量审阅成片脚本与指令问题清单、修改建议Reviewer Agent终审裁决修改前后版本是否通过Critic Agent实际项目中还可以增加字幕生成、封面设计、多平台分发等节点但核心骨架就是这六层。2.2 系统总体架构整套系统从上到下可以分成四层用户交互层接收用户输入的主题、风格偏好、时长要求。多智能体编排层由调度器统一管理多个 Agent按流程顺序执行必要时进入博弈循环。MCP 工具层由多个 MCP Server 暴露视频创作相关能力例如素材检索、画面生成、剪辑指令生成、字幕格式化。外部服务层包括大模型 API、视频素材库、图片生成服务、剪辑软件等。这里需要注意多智能体编排层不等于链式调用。好的调度器至少要支持条件分支、循环审校、结果汇总并且能感知每个 Agent 的执行状态。2.3 各 Agent 角色与工具映射并不是每个 Agent 都需要接入全部工具正确思路是按需授权。Planner Agent 一般只需要搜索类工具用于搜索热点、参考竞品。Script Agent 通常只调用文本处理类工具不涉及素材 API。Visual Agent 需要调用图片生成、素材检索类 MCP 工具。Editor Agent 需要调用剪辑指令生成、视频元数据处理类工具。Reviewer Agent 可能需要调用文本质量检查、敏感词检测类工具。这种按角色分配工具的方式能减少 Agent 在调用工具时的选择噪音也能降低误调用风险。3. 环境准备与项目结构3.1 运行环境与版本本文示例以 Python 3.10 以上版本为基准操作系统不限Windows、macOS、Linux 都可以。大模型接口部分为了不让示例绑定某个特定厂商我会用一个自定义的 LLMClient 接口做抽象实际项目里可以替换成任意大模型 API 或本地模型服务。MCP SDK 部分官方提供了 TypeScript SDK 和 Python SDK社区也有 FastMCP 这类更简洁的封装。下面示例中的 MCP Server 代码属于演示骨架实际接入时请以你所选 SDK 的最新文档为准。版本需要根据项目实际情况调整本文重点演示配置思路和工程骨架。3.2 项目结构建议按下面的目录组织项目video-agent-studio/ ├── README.md ├── requirements.txt ├── config/ │ ├── agents.yaml # Agent 角色配置 │ ├── mcp_servers.json # MCP Server 注册信息 │ └── workflow.yaml # 流水线配置 ├── src/ │ ├── main.py # 程序入口 │ ├── llm/ │ │ ├── base.py # LLM 接口抽象 │ │ └── openai_client.py # OpenAI 兼容客户端示例 │ ├── mcp/ │ │ ├── client.py # MCP 客户端封装 │ │ └── video_server.py # 视频创作工具 MCP Server │ ├── agents/ │ │ ├── base.py # Agent 基类 │ │ ├── planner.py # 策划 Agent │ │ ├── script_writer.py # 脚本 Agent │ │ ├── visual_planner.py # 视觉 Agent │ │ ├── editor.py # 剪辑指令 Agent │ │ └── reviewer.py # 审阅 Agent │ └── orchestrator/ │ ├── pipeline.py # 流水线调度器 │ └── review_loop.py # 博弈审阅循环 └── output/ ├── plan.md ├── script.md ├── storyboard.json └── edit_commands.json3.3 基础依赖说明requirements.txt 里的依赖大致是# 根据实际安装环境调整版本号 mcp1.0.0 fastmcp2.0.0 pyyaml6.0 requests2.31.0 pydantic2.5.0MCP 相关库装在 Python 环境后可以通过启动一个独立的 MCP Server 进程来测试。如果你的项目需要和 Dify、Cherry Studio、Trae 这类平台对接MCP Server 的注册方式会是各自平台的配置体系但核心都是“把 server 的启动命令和参数暴露给对方”。4. 实现 MCP Server把视频工具接入标准化协议4.1 为什么需要自定义 MCP Server通用 MCP Server 已经覆盖了浏览器、数据库、设计稿等场景但视频创作领域还很碎片化。一个团队内部的素材库、剪辑模板、字幕规范往往没有现成的 MCP Server 可用。这时候就需要自己写一个 Video MCP Server把团队内部的视频工具能力统一暴露出来。自定义 MCP Server 的核心工作有三部分确定要暴露哪些工具。为每个工具定义输入输出结构。在工具内部对接真实的业务系统。4.2 视频工具的暴露与参数设计视频创作场景里有几个高频工具很适合做成 MCP 工具search_material根据关键词从素材库检索图片、视频片段。generate_visual_prompt根据分镜描述生成画面提示词。build_subtitle根据口播稿生成字幕文件和样式。generate_edit_command把分镜脚本转换成剪辑软件的指令序列。check_content_risk检查文案是否包含敏感词或不合适表达。以 search_material 为例输入参数可以包含 query检索词、material_type素材类型图片或视频、duration目标时长、style风格标签输出则是素材 ID、标题、封面图、时长、资源地址的列表。4.3 MCP Server 核心代码下面是使用 FastMCP 风格封装的一个演示代码。FastMCP 是对官方 MCP Python SDK 的一层高层封装写法更接近普通 Python 函数核心思路是“装饰器注册工具”。# 文件路径src/mcp/video_server.py # 说明这是简化示例实际参数和函数体请根据团队内部系统调整 from fastmcp import FastMCP mcp FastMCP(video-tools) mcp.tool() def search_material(query: str, material_type: str video, style: str ) - list: 根据关键词在素材库中检索可用的图片或视频资源。 Args: query: 检索关键词例如“城市夜景 延时摄影” material_type: 素材类型可选 video / image style: 风格标签例如“科技感”“赛博朋克” Returns: 符合条件的素材列表每个元素包含 id、title、url、duration 等字段 # 真实项目中这里会调用素材库 API 或数据库查询 # 下面是模拟返回仅用于演示工具结构 return [ { id: MTR-001, title: f{query} 素材 1, url: https://example.com/materials/MTR-001.mp4, duration: 8.5, style: style or 通用 }, { id: MTR-002, title: f{query} 素材 2, url: https://example.com/materials/MTR-002.mp4, duration: 12.0, style: style or 通用 } ] mcp.tool() def generate_edit_command(storyboard: list, material_map: dict) - dict: 根据分镜列表和素材映射表生成剪辑指令序列。 Args: storyboard: 分镜列表每个分镜包含序号、画面描述、口播文本 material_map: 分镜序号到素材 ID 的映射表 Returns: 剪辑指令字典包含 scenes 和 transitions scenes [] for item in storyboard: scene_id item.get(id) scenes.append({ scene_id: scene_id, material_id: material_map.get(scene_id, ), text: item.get(text, ), duration: item.get(duration, 5), }) return { scenes: scenes, transitions: [cut, dissolve, cut], audio: {music: background_music_v2.mp3, volume: 0.3} } mcp.tool() def check_content_risk(text: str) - dict: 对文案进行基础内容风险检查。 Args: text: 待检查的文本内容 Returns: 检查结果包含风险等级和问题列表 risk_words [违禁词示例] # 实际项目里用敏感词库代替 hit_words [w for w in risk_words if w in text] return { level: high if hit_words else low, hit_words: hit_words, suggestion: 请删除或替换高亮词汇 if hit_words else 可以发布 } if __name__ __main__: # 以标准输入输出模式启动 MCP Server。 # 这个模式适合本地进程调用便于 Dify、Claude Desktop 等宿主连接。 mcp.run(transportstdio)这段代码有几个设计要点每个工具函数都带有 docstringMCP 协议会把函数签名和描述语法发给模型模型根据这些信息决定何时调用。返回值尽量使用可序列化结构避免返回 Python 对象。工具粒度要适中既不能太粗也不能太细。如果工具太大模型难以理解调用边界如果太细Agent 需要多次调用才能完成一个目标。4.4 MCP 配置与客户端连接MCP Server 编写完成后需要让宿主程序知道如何启动它。标准做法是在配置文件里声明 command 和 args。在 Claude Desktop、Cherry Studio、Dify 这类工具里MCP 配置一般长这样{ mcpServers: { video-tools: { command: python, args: [src/mcp/video_server.py] } } }在 win 系统、macOS 上要注意命令路径和环境变量。推荐用虚拟环境的绝对路径例如{ mcpServers: { video-tools: { command: C:/Users/yourname/.venvs/video-agent/Scripts/python.exe, args: [D:/projects/video-agent-studio/src/mcp/video_server.py] } } }这里有一个隐藏的坑启动宿主程序时尽量让 MCP Server 的命令指向绝对路径否则宿主进程的资源路径不一致会导致 Server 启动失败。在自研 Python 应用里可以用 MCP 客户端 SDK 连接 Server。下面是一个伪代码级别的客户端封装# 文件路径src/mcp/client.py # 说明以 MCP 官方 Python SDK 为示例实际 API 以你安装的版本为准 from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client class McpToolClient: def __init__(self, server_name: str, command: str, args: list): self.server_params StdioServerParameters( commandcommand, argsargs ) async def call_tool(self, tool_name: str, arguments: dict): async with stdio_client(self.server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() result await session.call_tool(tool_name, arguments) return result实际项目中一个 Agent 往往需要连续调用多个工具不可能每次都重新建立连接。正确做法是在应用启动时一次性建立连接把 session 缓存起来复用。后面在编排层会用到这个思路。5. 实现多智能体编排层5.1 Agent 基类与 LLM 适配为了让多个 Agent 的代码风格统一先设计一个轻量级 Agent 基类。基类里维护角色名称、系统提示词、可用工具三个核心字段。# 文件路径src/agents/base.py from typing import Callable, Dict, List class Agent: 所有 Agent 的基类。 Attributes: name: Agent 名称用于日志和编排 system_prompt: 角色系统提示词 tools: 该 Agent 可调用的工具集合键为工具名值为调用函数 llm: 大模型客户端需要实现 complete 方法 def __init__(self, name: str, system_prompt: str, llm, tools: Dict[str, Callable] None): self.name name self.system_prompt system_prompt self.llm llm self.tools tools or {} def run(self, input_data: str) - str: 子类实现具体逻辑默认情况下直接调用模型。 Args: input_data: 输入文本 Returns: Agent 处理后的输出文本 messages [ {role: system, content: self.system_prompt}, {role: user, content: input_data} ] return self.llm.complete(messages) def call_tool(self, tool_name: str, arguments: dict): if tool_name not in self.tools: raise ValueError(fAgent {self.name} 未注册工具 {tool_name}) return self.tools[tool_name](arguments)LLM 客户端抽象如下# 文件路径src/llm/base.py from abc import ABC, abstractmethod class BaseLLMClient(ABC): 大模型客户端抽象类所有模型适配器都要继承它。 abstractmethod def complete(self, messages: list, temperature: float 0.7) - str: 传入消息列表返回模型生成的文本。 Args: messages: OpenAI 风格消息列表 temperature: 采样温度 passOpenAI 兼容客户端的实现可以用 requests也可以用官方 SDK思路是把消息发送到模型接口解析返回内容。# 文件路径src/llm/openai_client.py import requests from .base import BaseLLMClient class OpenAICompatClient(BaseLLMClient): 面向 OpenAI 兼容接口的客户端。 很多模型服务商提供 OpenAI 兼容接口只需要替换 base_url 和 api_key。 def __init__(self, base_url: str, api_key: str, model: str): self.base_url base_url.rstrip(/) /chat/completions self.api_key api_key self.model model def complete(self, messages: list, temperature: float 0.7) - str: payload { model: self.model, messages: messages, temperature: temperature } headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } resp requests.post(self.base_url, jsonpayload, headersheaders, timeout60) resp.raise_for_status() data resp.json() return data[choices][0][message][content]这里刻意不绑定具体厂商因为不同厂商的模型能力和价格差异很大实际项目中要按业务需要选择。比如字幕生成、敏感词检查这类任务可以用快且便宜的模型策划和审阅这类需要综合推理的任务用更强的大模型。5.2 策划 Agent选题与结构策划 Agent 的职责是把一个模糊主题变成内容大纲。它的系统提示词里需要强调目标人群分析、内容切入点、结构节奏、时长控制。# 文件路径src/agents/planner.py from .base import Agent PLANNER_PROMPT 你是一位资深短视频策划。 你的任务是根据用户给定的主题产出一份适合短视频平台的创作大纲。 大纲需要包含 1. 目标人群画像 2. 内容切入点 3. 视频结构开头-中段-结尾 4. 每个环节的预估时长 5. 情绪节奏设计 要求 - 结构清晰每条内容用序号标出 - 控制整条视频在用户期望时长内 - 开头要有钩子结尾要有引导关注或讨论的互动点 class PlannerAgent(Agent): def __init__(self, llm, toolsNone): super().__init__( namePlannerAgent, system_promptPLANNER_PROMPT, llmllm, toolstools ) def run(self, topic: str) - str: user_input f视频主题{topic}\n请输出完整创作大纲。 return super().run(user_input)这个 Agent 输出的是纯文本大纲为了后续 Agent 能稳定解析最好在提示词里要求结构化输出或者在后置处理里把文本转成 JSON。实际项目建议用 JSON 输出并做 schema 校验避免脏数据往下游传播。5.3 脚本 Agent文案与口播稿脚本 Agent 接收策划 Agent 的大纲输出完整口播稿和分镜文本。它还需要把文案按“镜头”拆开方便后续视觉和剪辑 Agent 处理。# 文件路径src/agents/script_writer.py import json from .base import Agent SCRIPT_PROMPT 你是一位短视频编剧。 你会收到一份创作大纲需要把它扩展成完整脚本。 输出格式为 JSON 数组每个元素包含 - id: 编号 - scene_text: 画面描述 - voiceover: 口播台词 - duration: 预估时长秒 要求 - 口播台词要口语化、有节奏感 - 画面描述要具体便于后续生成视觉提示词 - 每条时长之和要接近大纲总时长 class ScriptAgent(Agent): def __init__(self, llm, toolsNone): super().__init__( nameScriptAgent, system_promptSCRIPT_PROMPT, llmllm, toolstools ) def run(self, plan_text: str) - list: raw super().run(plan_text) # 这里做简单解析实际项目需要更健壮的 JSON 提取逻辑 start raw.find([) end raw.rfind(]) 1 json_str raw[start:end] return json.loads(json_str)脚本输出使用 JSON 数组而不是 Markdown是因为多智能体系统里每个 Agent 的输出都要被下一个 Agent 程序化消费。文本读起来方便但程序解析麻烦JSON 读起来稍差但可靠。5.4 视觉 Agent分镜与画面提示词视觉 Agent 的职责是把分镜文本转换成适合图片生成模型使用的提示词同时调用素材检索工具找到可用素材。# 文件路径src/agents/visual_planner.py import json from .base import Agent VISUAL_PROMPT 你是一位视觉导演。 你会收到一个分镜对象需要输出以下内容 1. image_prompt: 适合图片生成模型的英文提示词 2. key_elements: 画面中的关键元素列表 3. material_keyword: 素材库检索用的中文关键词 输出格式为 JSON 对象。 class VisualAgent(Agent): def run(self, scene: dict) - dict: raw super().run(json.dumps(scene, ensure_asciiFalse)) # 解析 JSON并调用素材检索工具 result json.loads(raw) if self.tools and search_material in self.tools: materials self.tools[search_material]( queryresult.get(material_keyword, ), material_typevideo ) result[candidate_materials] materials return result注意这里的 search_material 不是直接调用 MCP Server而是通过工具映射注入进来的。在应用启动时我们会把 MCP 工具包装成普通 Python 函数传给 Agent这样 Agent 层不需要关心工具是本地实现还是远程 MCP 协议通信。5.5 剪辑指令 Agent素材与转场剪辑指令 Agent 是流水线里最接近“最终产物”的节点。它接收分镜列表、画面提示词、素材候选列表输出一段结构化的剪辑指令。# 文件路径src/agents/editor.py import json from .base import Agent EDITOR_PROMPT 你是一位剪辑指导。 你会收到一个分镜列表和素材信息需要输出一条剪辑指令 JSON包含 - scenes: 场景数组每个场景包含镜头号、素材 ID、时长、字幕文本 - transitions: 转场列表 - audio: 音频设置对象 要求 - 根据口播时长确定每个镜头的时长 - 素材 ID 优先从候选素材里选选不到就标记 need_replace - 转场风格要符合内容调性 class EditorAgent(Agent): def run(self, storyboard_with_materials: dict) - dict: raw super().run(json.dumps(storyboard_with_materials, ensure_asciiFalse)) return json.loads(raw)在更复杂的实现里Editor Agent 还可以调用剪辑软件的 MCP Server直接生成可导入剪辑工程的文件。不过这一步依赖具体剪辑软件是否提供接口落地时通常需要定制开发。5.6 审阅 Agent质量与风险检查审阅环节是质量保障的关键也是多智能体系统相比单 Agent 的优势所在。这里我用两个角色实现“正反博弈加裁判”的模式Critic Agent负责挑毛病从逻辑、节奏、文案质量、合规风险等角度提出修改意见。Reviewer Agent负责做最终裁决判断修改后的版本是否达到发布标准。# 文件路径src/agents/reviewer.py import json from .base import Agent CRITIC_PROMPT 你是一位严格的短视频审阅人。 你会收到一份剪辑指令和脚本内容请从以下维度提出具体修改意见 1. 内容逻辑是否连贯 2. 开头前 3 秒是否有吸引力 3. 情绪节奏是否合理 4. 文案是否有语病或敏感表述 5. 镜头时长分配是否合理 输出格式为 JSON 数组每个元素包含 - issue: 问题描述 - severity: 严重程度可选 high / medium / low - suggestion: 修改建议 REVIEWER_PROMPT 你是一位总编审。 你会收到一个视频方案的初版和修改版需要输出最终裁决 - status: passed 或 rejected - final_comment: 最终意见 判断标准 - 如果所有 high 级问题已解决且剩余问题不影响观看可判定 passed - 如果仍有 high 级问题判定 rejected并说明还需修改的原因 class CriticAgent(Agent): def __init__(self, llm, toolsNone): super().__init__( nameCriticAgent, system_promptCRITIC_PROMPT, llmllm, toolstools ) def run(self, edit_commands: dict) - list: raw super().run(json.dumps(edit_commands, ensure_asciiFalse)) start raw.find([) end raw.rfind(]) 1 return json.loads(raw[start:end]) class ReviewerAgent(Agent): def __init__(self, llm, toolsNone): super().__init__( nameReviewerAgent, system_promptREVIEWER_PROMPT, llmllm, toolstools ) def run(self, original: dict, revised: dict, issues: list) - dict: payload { original: original, revised: revised, critic_issues: issues } raw super().run(json.dumps(payload, ensure_asciiFalse)) return json.loads(raw)这个博弈循环的价值在于Critic Agent 不需要“修改”能力Reviewer Agent 不需要“创作”能力每个角色只做自己最擅长的事比让同一个 Agent 既写又改可靠得多。5.7 调度器流水线串联调度器是整个系统的中枢。它负责决定 Agent 的执行顺序、传递中间结果、处理博弈循环以及把最终结果写入输出文件。# 文件路径src/orchestrator/pipeline.py import json from pathlib import Path from typing import Dict from src.agents.editor import EditorAgent from src.agents.planner import PlannerAgent from src.agents.reviewer import CriticAgent, ReviewerAgent from src.agents.script_writer import ScriptAgent from src.agents.visual_planner import VisualAgent class VideoCreationPipeline: 视频创作流水线调度器。 Args: agents: 已经初始化好的 Agent 实例字典 output_dir: 输出目录 max_review_rounds: 审阅最大轮数防止死循环 def __init__(self, agents: Dict[str, object], output_dir: str output, max_review_rounds: int 3): self.agents agents self.output_dir Path(output_dir) self.output_dir.mkdir(parentsTrue, exist_okTrue) self.max_review_rounds max_review_rounds def run(self, topic: str): # 1. 策划阶段 plan self.agents[planner].run(topic) self.output_dir.joinpath(plan.md).write_text(plan, encodingutf-8) print([Pipeline] 策划大纲已生成) # 2. 脚本阶段 script self.agents[script].run(plan) self.output_dir.joinpath(script.json).write_text( json.dumps(script, ensure_asciiFalse, indent2), encodingutf-8 ) print([Pipeline] 脚本已生成) # 3. 视觉阶段 storyboard_with_materials [] for scene in script: visual self.agents[visual].run(scene) storyboard_with_materials.append({ **scene, image_prompt: visual.get(image_prompt, ), candidate_materials: visual.get(candidate_materials, []) }) # 4. 剪辑指令阶段 edit_commands self.agents[editor].run({ storyboard: storyboard_with_materials, total_duration: sum(s.get(duration, 5) for s in script) }) # 5. 审阅博弈阶段 current edit_commands for round_idx in range(self.max_review_rounds): issues self.agents[critic].run(current) if not issues: break # 如果审阅意见为空说明没有高优问题直接通过 high_issues [i for i in issues if i.get(severity) high] if not high_issues: break # 根据意见让 Editor 修改实际项目里需要再注入一个修改 Agent # 这里简化处理本示例直接进入第二轮审阅真实实现请补充分支 break # 角色分配说明真实业务里这里应该有一个 Revise Agent 负责修改 # 由于篇幅限制本示例把修改逻辑放到了 Editor Agent 的二次调用里 final_commands current self.output_dir.joinpath(edit_commands.json).write_text( json.dumps(final_commands, ensure_asciiFalse, indent2), encodingutf-8 ) print([Pipeline] 剪辑指令已输出) # 6. 最终裁决 verdict self.agents[reviewer].run( originaledit_commands, revisedfinal_commands, issuesissues if issues else [] ) self.output_dir.joinpath(review_result.json).write_text( json.dumps(verdict, ensure_asciiFalse, indent2), encodingutf-8 ) return { plan: plan, script: script, edit_commands: final_commands, review: verdict }这里为了演示清晰审阅循环里没有真的把修改后的内容喂回去继续迭代。工程落地时可以增加一个 ReviseAgent专门消费 Critic 的意见并生成修改版然后由 Reviewer 判断是否通过。下面单独给出一个 review_loop.py 模块专门负责这个博弈循环。# 文件路径src/orchestrator/review_loop.py class ReviewLoop: 正反博弈 裁判循环。 执行流程 1. Critic Agent 对当前版本提出问题 2. 如果没有 high 级问题循环结束 3. 如果有问题Revise Agent 根据意见进行修改 4. Reviewer Agent 对修改后的版本做裁决 5. 通过则结束不通过则继续下一轮 def __init__(self, critic_agent, revise_agent, reviewer_agent, max_rounds3): self.critic_agent critic_agent self.revise_agent revise_agent self.reviewer_agent reviewer_agent self.max_rounds max_rounds def run(self, initial_version: dict): current initial_version history [] for round_no in range(1, self.max_rounds 1): issues self.critic_agent.run(current) history.append({ round: round_no, issues: issues, version: current }) high_issues [i for i in issues if i.get(severity) high] if not high_issues: print(f[ReviewLoop] 第 {round_no} 轮无高优问题通过审阅) return current, history print(f[ReviewLoop] 第 {round_no} 轮发现 {len(high_issues)} 个高优问题开始修改) current self.revise_agent.run(current, high_issues) verdict self.reviewer_agent.run({ original: initial_version, revised: current, issues: issues }) if verdict.get(status) passed: print(f[ReviewLoop] 第 {round_no} 轮修改后通过终审) return current, history print(f[ReviewLoop] 达到最大轮数 {self.max_rounds}返回最后版本) return current, history这个 ReviewLoop 模块可以直接复用到不同的内容生产场景。除了视频创作写文章、做海报、生成报告只要有“初稿—批评—修改—裁决”的质量需求都可以套用。6. 运行完整流程与结果验证6.1 启动 MCP Server先在一个终端窗口启动 MCP Servercd video-agent-studio python src/mcp/video_server.py如果配置正确MCP Server 会以 stdio 模式启动不打印日志到标准输出而是等待宿主进程通过标准输入输出发送协议消息。如果你在终端里看到了启动日志通常意味着被误打印到了 stdout需要把调试日志改成输出到 stderr。6.2 编写程序入口并执行流水线在程序入口里做依赖注入把 MCP 工具和 LLM 客户端装配到各 Agent 中然后运行流水线。# 文件路径src/main.py from src.llm.openai_client import OpenAICompatClient from src.mcp.video_server import search_material, check_content_risk from src.agents.planner import PlannerAgent from src.agents.script_writer import ScriptAgent from src.agents.visual_planner import VisualAgent from src.agents.editor import EditorAgent from src.agents.reviewer import CriticAgent, ReviewerAgent from src.orchestrator.pipeline import VideoCreationPipeline # 1. 初始化 LLM 客户端 # 实际使用中把 base_url、api_key、model 替换成自己的配置 llm OpenAICompatClient( base_urlhttps://your-model-endpoint.example.com/v1, api_keyyour-api-key, modelyour-model-name ) # 2. 初始化 MCP 工具映射 # 这里直接把函数引用注入 Agent实际项目可以通过 MCP 客户端动态拉取工具列表 mcp_tools { search_material: lambda args: search_material(**args), check_content_risk: lambda args: check_content_risk(**args), } # 3. 初始化各 Agent planner PlannerAgent(llmllm) script_agent ScriptAgent(llmllm) visual_agent VisualAgent(llmllm) editor EditorAgent( llmllm, tools{ search_material: mcp_tools[search_material], } ) critic CriticAgent(llmllm, tools{ check_content_risk: mcp_tools[check_content_risk], }) reviewer ReviewerAgent(llmllm) agents { planner: planner, script: script_agent, visual: visual_agent, editor: editor, critic: critic, reviewer: reviewer, } # 4. 运行流水线 pipeline VideoCreationPipeline( agentsagents, output_diroutput ) result pipeline.run(如何看待 MCP 协议对 AI 应用开发的影响)运行命令cd video-agent-studio python src/main.py6.3 输出结果说明执行完成后output 目录下会生成以下文件plan.md策划大纲供人工预览。script.json分镜与口播脚本数组。edit_commands.json剪辑指令包含场景、转场、音频设置。review_result.json审阅终裁结果。这套输出的设计理念是“人机协同”机器生成的内容不直接进剪辑软件而是作为半成品让真人快速确认。一方面避免全自动流程出错造成返工另一方面让创作者保留最终控制权。7. 常见问题与排查思路7.1 高频问题清单问题现象常见原因解决思路MCP Server 启动失败Python 路径不对或依赖未安装使用虚拟环境绝对路径确认 mcp / fastmcp 已安装Agent 调用工具时报错“工具不存在”Agent 实例未注入工具映射检查初始化时 tools 参数是否传递脚本 Agent 输出无法解析成 JSON模型返回了 Markdown 格式文本在提示词中强制 JSON 输出增加 JSON 提取与容错逻辑审阅循环死循环没有设置最大轮数设置 max_review_rounds并统计每轮修改是否真正产生 diff上下文过大超出限制多轮 Agent 结果全部塞入同一对话每个 Agent 使用独立上下文工具返回结果只保留摘要字段MCP 工具返回数据太大素材接口返回整个视频 URL 和长描述在 MCP Server 端做字段裁剪和分页某模型对工具描述理解差工具函数名和 docstring 不清晰优化工具命名和描述必要时拆细工具粒度7.2 深挖几个典型问题第一个典型问题是“上下文过大”。在多智能体系统里如果设计不当每一轮的中间结果都会累积到下一轮对话中很快就会触发模型上下文窗口限制报错信息通常类似“上下文大小超出限制”。排查思路是先确认上下文增长来自哪里。如果每个 Agent 都接收前一个 Agent 的完整原始输出那么脚本阶段把策划全文、视觉阶段把脚本全文、剪辑阶段把分镜全文全部堆积到一起token 数量会爆炸。解决方案有三个方向每个 Agent 独立上下文窗口不向前继承。工具返回结果做摘要化处理只保留核心字段。对中间结果做结构化压缩例如把策划大纲沉淀成固定字段的 JSON而不是保留大段自然语言文本。第二个典型问题是“Agent 工具调用不稳定”。同一个模型面对同样的问题有时调用一次工具有时调用三四次甚至选择错误工具。原因是工具描述不够清晰或者工具之间边界模糊。建议为每个工具写完整的 docstring明确参数含义和返回值结构。同时工具数量控制在可理解范围内。如果一个 Agent 挂了 20 个工具模型选择困难会显著增加。第三个典型问题是“多智能体流程调试困难”。单 Agent 还可以看对话日志多 Agent 的失败可能发生在任何一个节点。建议在每个 Agent 的 run 方法入口出口都记录结构化的执行日志内容包括输入摘要、输出摘要、耗时、调用模型、调用的工具列表。用日志系统做全链路追踪才能快速定位是哪个环节出了问题。8. 最佳实践与工程建议8.1 Agent 角色与 prompt 管理Agent 的 system prompt 不要直接写在代码里。虽然示例里为了直观把 prompt 放到了 .py 文件里但工程上更推荐统一管理# 文件路径config/agents.yaml planner: name: PlannerAgent model: strong-model prompt: | 你是一位资深短视频策划... script_writer: name: ScriptAgent model: strong-model prompt: | 你是一位短视频编剧... visual_planner: name: VisualAgent model: fast-model prompt: | 你是一位视觉导演...这样做的价值在于提示词调优是高频操作把 prompt 独立到配置文件里可以避免每次修改都要重新部署代码。8.2 MCP 工具设计原则第一工具命名要“动词加名词”例如 search_material、generate_edit_command避免 ambiguous 的命名。第二工具返回值要精简。很多下游 Agent 只需要素材 ID 和标题不需要完整的资源 URL 和二进制元数据。MCP Server 内部可以只返回必要字段必要时让 Agent 指定是否要详细版本。第三工具要有权限设计意识。视频创作系统里可能涉及删除素材、覆盖工程文件等危险操作这类工具不能暴露给普通 Agent。MCP Server 内部要做身份校验至少在 Host 层控制哪些 Agent 可以调用哪些工具。8.3 上下文与成本控制多智能体系统最大的成本陷阱是 token 浪费。同一份分镜内容在视觉、剪辑、审阅三个 Agent 里重复出现会成倍增加 token 消耗。几个实用的成本控制手段下游 Agent 只接收上游 Agent 的结构化摘要而不是全文。明确每个 Agent 的输入 schema用 Pydantic 做数据校验。对耗时长的流程设置超时和重试上限。不同 Agent 使用不同规格模型。简单抽取任务用快模型复杂推理用强模型。8.4 安全与权限边界这属于所有 AI 应用工程落地都必须重视的部分。MCP Server 暴露的是真实业务能力如果不加控制Agent 可能误调用删除、覆盖、发送等危险操作。至少要做到几点所有非查询类操作都要增加二次确认机制。MCP 工具参数做白名单校验。敏感操作记录操作人、操作 Agent、入参、出参。涉及生产素材库时先在小范围测试验证再放开权限。8.5 日志与可观测性多智能体流程的失败通常不是程序崩溃而是“产出结果不符合预期”。所以日志要能回答几个问题每个 Agent 收到了什么输入它调用哪些工具每个工具返回值是什么它生成了什么输出在哪一轮循环里结果发生了变化推荐使用结构化的 JSON 日志把每个 Agent 的输入输出摘要、token 用量、耗时、工具调用列表全部记录下来。这样即使某个环节出现问题也能根据日志快速回放整个执行过程。模块化的流水线设计可以让不同的 Agent 替换、插拔比如把 ScriptAgent 从“通用模型”切换到“专门微调过文案风格的模型”只需要改配置不改调度逻辑。这才是多智能体系统比单体 Agent 更具工程扩展性的根本原因。如果后续想继续深入建议优先研究两个方向一是怎么让 Agent 的中间产物用更小的 token 传递比如引入分镜元数据 schema 和向量检索让下游 Agent 按需获取详情而不是全量接收。二是尝试把博弈审阅循环用到更长周期的内容生产任务中比如系列视频策划、完整短视频账号的内容矩阵规划这时多智能体的协作优势会比单条视频更明显。动手跑一遍本文的示例代码比看十篇概念解读都有用。跑通之后再试着替换一个真实场景的 MCP 工具整个体系就会慢慢变成你自己的创作基础设施。 MCP+多智能体驱动视频创作流水线的工程实践
做短视频内容最耗时间的往往不是拍摄本身而是从选题、脚本、分镜到剪辑指令这条链路。每个环节都需要不同角色协作工具之间又是割裂的文案在文档里素材在网盘里剪辑在专业软件里提示词在对话框里。最近在调研和实验 AI 视频创作方案时发现把 MCP 作为工具接入标准再用多智能体系统把策划、脚本、视觉、剪辑、审阅这些角色串成流水线确实能把创作流程压缩到一个可编排、可复现的工程框架里。这篇文章会把整套流程拆开讲清楚先解释 MCP 和多智能体的核心概念再给出视频创作场景下的系统架构设计然后动手实现一个 MCP Server 和多个 Agent最后串联成一条完整流水线并附上常见问题和工程建议。适合有一定 AI 应用开发基础、想在视频内容生产场景里落地 Agent 的开发者也适合对 MCP 感兴趣、想了解它到底怎么用的朋友。1. 背景与核心概念1.1 视频创作流程中的效率瓶颈一条短视频从创意到成片通常要经历这么几个阶段选题策划确定视频主题、目标人群、内容结构。脚本撰写产出口播文案、画面描述、字幕文本。视觉设计规划分镜、镜头语言、画面参考。素材准备调用图片、视频片段、音频、字幕资源。剪辑合成按分镜顺序生成剪辑指令或工程文件。审阅修改检查内容质量、情绪节奏、合规风险。传统做法里这些环节依赖人工在不同软件之间切换信息传递靠文档和口头沟通。即使引入了 AI 工具也只是“单点辅助”用 ChatGPT 写脚本、用 Midjourney 出图、用剪映剪辑各工具之间没有统一的数据交互标准更谈不上多角色协作。如果能有一个系统让不同的 AI 助手分别扮演策划、编剧、视觉设计、剪辑师、审阅官并且它们能调用同一套视频工具集那创作效率的提升会是质的改变。这就引出了两个关键技术点MCP 和多智能体系统。1.2 MCP 是什么模型上下文协议MCP 全称是 Model Context Protocol即“模型上下文协议”。它的作用是解决大语言模型与外部工具、数据源连接时的标准化问题。在没有 MCP 之前每个应用要对接模型和工具都得自己写一套集成代码。比如一个 Agent 要调用数据库、搜索网页、读取本地文件就得分别实现数据库驱动、搜索 API 客户端、文件读写模块。集成方式五花八门换一个模型或换一个工具就要重新适配。MCP 的架构里多了几个标准角色MCP Host运行模型的宿主程序比如 Claude Desktop、Dify、自研的 Agent 应用。MCP ClientHost 内部负责与 MCP Server 通信的组件。MCP Server对外提供工具、资源、提示词的服务工具能力以标准协议暴露。可以把它理解成“AI 世界的 USB 接口”。设备厂商不用关心你的电脑是什么系统只要设备支持 USB 协议就能接入同样MCP Server 只要实现标准协议任何支持 MCP 的宿主都能调用它的工具。MCP 在 2024 年底正式开源后社区生态发展非常快。目前已经有大量现成的 MCP Server覆盖了数据库访问、浏览器自动化如 Playwright、设计稿标注如 Figma、蓝湖、代码托管、搜索引擎等场景。这意味着 Agent 可以主动去连接无限多的外部能力而不局限于模型训练时的静态知识。1.3 多智能体系统的核心思想多智能体系统Multi-Agent System简称 MAS是一套把多个有独立角色和目标的 Agent 组织起来协作完成复杂任务的架构。核心思想是“拆”把一个大任务拆解成多个子任务每个子任务交给最擅长它的 Agent再由调度器或工作流把结果串起来。单 Agent 处理复杂任务时容易遇到几个问题上下文过长。所有历史信息都塞进同一个对话窗口既费 token 又容易超出限制。角色冲突。让同一个模型既当策划又当审阅它很难真正自我批判。工具混乱。一个 Agent 需要关注太多工具调用时容易选错。多智能体系统则把这些问题分而治之每个 Agent 只负责一个相对独立的角色拥有自己的 system prompt、上下文窗口和工具集。Agent 之间传递的是结构化结果而不是漫无边际的对话历史。常见的多智能体协作模式有三种流水线模式一个 Agent 的输出是下一个 Agent 的输入适合流程固定的场景。协作模式多个 Agent 并行工作最后汇总结果。博弈模式一个 Agent 提出方案另一个 Agent 负责挑错再由裁判 Agent 做最终裁决适合需要质量把关的场景。短视频创作流程天然适合流水线加博弈的混合模式这一点后面会详细展开。1.4 MCP 与多智能体的组合优势MCP 解决的是“工具如何被标准化调用”的问题多智能体解决的是“多个角色如何分工协作”的问题两者是互补关系。有了 MCP多智能体系统里的每个 Agent 才能以统一方式调用视频创作所需的外部工具。否则就会出现同时存在 OpenAI 格式、LangChain 格式、自研 API 格式的混乱局面。有了多智能体MCP Server 里的工具不会被一个庞大的、无所不能的 Agent 独占而是按角色分发给不同 Agent让整个系统的可维护性和可扩展性大幅提升。2. 整体架构设计从创意到成片的流水线2.1 视频创作流程拆解把视频创作映射到多智能体系统最关键的一步是明确流程节点和每个节点的输入输出。一条可自动化的视频创作流水线可以拆成阶段输入输出对应 Agent选题策划主题关键词、用户画像内容大纲、切入角度Planner Agent脚本撰写内容大纲口播文案、分镜文本Script Agent视觉设计脚本内容画面描述、分镜提示词Visual Agent剪辑调度分镜与素材信息剪辑指令、素材清单Editor Agent质量审阅成片脚本与指令问题清单、修改建议Reviewer Agent终审裁决修改前后版本是否通过Critic Agent实际项目中还可以增加字幕生成、封面设计、多平台分发等节点但核心骨架就是这六层。2.2 系统总体架构整套系统从上到下可以分成四层用户交互层接收用户输入的主题、风格偏好、时长要求。多智能体编排层由调度器统一管理多个 Agent按流程顺序执行必要时进入博弈循环。MCP 工具层由多个 MCP Server 暴露视频创作相关能力例如素材检索、画面生成、剪辑指令生成、字幕格式化。外部服务层包括大模型 API、视频素材库、图片生成服务、剪辑软件等。这里需要注意多智能体编排层不等于链式调用。好的调度器至少要支持条件分支、循环审校、结果汇总并且能感知每个 Agent 的执行状态。2.3 各 Agent 角色与工具映射并不是每个 Agent 都需要接入全部工具正确思路是按需授权。Planner Agent 一般只需要搜索类工具用于搜索热点、参考竞品。Script Agent 通常只调用文本处理类工具不涉及素材 API。Visual Agent 需要调用图片生成、素材检索类 MCP 工具。Editor Agent 需要调用剪辑指令生成、视频元数据处理类工具。Reviewer Agent 可能需要调用文本质量检查、敏感词检测类工具。这种按角色分配工具的方式能减少 Agent 在调用工具时的选择噪音也能降低误调用风险。3. 环境准备与项目结构3.1 运行环境与版本本文示例以 Python 3.10 以上版本为基准操作系统不限Windows、macOS、Linux 都可以。大模型接口部分为了不让示例绑定某个特定厂商我会用一个自定义的 LLMClient 接口做抽象实际项目里可以替换成任意大模型 API 或本地模型服务。MCP SDK 部分官方提供了 TypeScript SDK 和 Python SDK社区也有 FastMCP 这类更简洁的封装。下面示例中的 MCP Server 代码属于演示骨架实际接入时请以你所选 SDK 的最新文档为准。版本需要根据项目实际情况调整本文重点演示配置思路和工程骨架。3.2 项目结构建议按下面的目录组织项目video-agent-studio/ ├── README.md ├── requirements.txt ├── config/ │ ├── agents.yaml # Agent 角色配置 │ ├── mcp_servers.json # MCP Server 注册信息 │ └── workflow.yaml # 流水线配置 ├── src/ │ ├── main.py # 程序入口 │ ├── llm/ │ │ ├── base.py # LLM 接口抽象 │ │ └── openai_client.py # OpenAI 兼容客户端示例 │ ├── mcp/ │ │ ├── client.py # MCP 客户端封装 │ │ └── video_server.py # 视频创作工具 MCP Server │ ├── agents/ │ │ ├── base.py # Agent 基类 │ │ ├── planner.py # 策划 Agent │ │ ├── script_writer.py # 脚本 Agent │ │ ├── visual_planner.py # 视觉 Agent │ │ ├── editor.py # 剪辑指令 Agent │ │ └── reviewer.py # 审阅 Agent │ └── orchestrator/ │ ├── pipeline.py # 流水线调度器 │ └── review_loop.py # 博弈审阅循环 └── output/ ├── plan.md ├── script.md ├── storyboard.json └── edit_commands.json3.3 基础依赖说明requirements.txt 里的依赖大致是# 根据实际安装环境调整版本号 mcp1.0.0 fastmcp2.0.0 pyyaml6.0 requests2.31.0 pydantic2.5.0MCP 相关库装在 Python 环境后可以通过启动一个独立的 MCP Server 进程来测试。如果你的项目需要和 Dify、Cherry Studio、Trae 这类平台对接MCP Server 的注册方式会是各自平台的配置体系但核心都是“把 server 的启动命令和参数暴露给对方”。4. 实现 MCP Server把视频工具接入标准化协议4.1 为什么需要自定义 MCP Server通用 MCP Server 已经覆盖了浏览器、数据库、设计稿等场景但视频创作领域还很碎片化。一个团队内部的素材库、剪辑模板、字幕规范往往没有现成的 MCP Server 可用。这时候就需要自己写一个 Video MCP Server把团队内部的视频工具能力统一暴露出来。自定义 MCP Server 的核心工作有三部分确定要暴露哪些工具。为每个工具定义输入输出结构。在工具内部对接真实的业务系统。4.2 视频工具的暴露与参数设计视频创作场景里有几个高频工具很适合做成 MCP 工具search_material根据关键词从素材库检索图片、视频片段。generate_visual_prompt根据分镜描述生成画面提示词。build_subtitle根据口播稿生成字幕文件和样式。generate_edit_command把分镜脚本转换成剪辑软件的指令序列。check_content_risk检查文案是否包含敏感词或不合适表达。以 search_material 为例输入参数可以包含 query检索词、material_type素材类型图片或视频、duration目标时长、style风格标签输出则是素材 ID、标题、封面图、时长、资源地址的列表。4.3 MCP Server 核心代码下面是使用 FastMCP 风格封装的一个演示代码。FastMCP 是对官方 MCP Python SDK 的一层高层封装写法更接近普通 Python 函数核心思路是“装饰器注册工具”。# 文件路径src/mcp/video_server.py # 说明这是简化示例实际参数和函数体请根据团队内部系统调整 from fastmcp import FastMCP mcp FastMCP(video-tools) mcp.tool() def search_material(query: str, material_type: str video, style: str ) - list: 根据关键词在素材库中检索可用的图片或视频资源。 Args: query: 检索关键词例如“城市夜景 延时摄影” material_type: 素材类型可选 video / image style: 风格标签例如“科技感”“赛博朋克” Returns: 符合条件的素材列表每个元素包含 id、title、url、duration 等字段 # 真实项目中这里会调用素材库 API 或数据库查询 # 下面是模拟返回仅用于演示工具结构 return [ { id: MTR-001, title: f{query} 素材 1, url: https://example.com/materials/MTR-001.mp4, duration: 8.5, style: style or 通用 }, { id: MTR-002, title: f{query} 素材 2, url: https://example.com/materials/MTR-002.mp4, duration: 12.0, style: style or 通用 } ] mcp.tool() def generate_edit_command(storyboard: list, material_map: dict) - dict: 根据分镜列表和素材映射表生成剪辑指令序列。 Args: storyboard: 分镜列表每个分镜包含序号、画面描述、口播文本 material_map: 分镜序号到素材 ID 的映射表 Returns: 剪辑指令字典包含 scenes 和 transitions scenes [] for item in storyboard: scene_id item.get(id) scenes.append({ scene_id: scene_id, material_id: material_map.get(scene_id, ), text: item.get(text, ), duration: item.get(duration, 5), }) return { scenes: scenes, transitions: [cut, dissolve, cut], audio: {music: background_music_v2.mp3, volume: 0.3} } mcp.tool() def check_content_risk(text: str) - dict: 对文案进行基础内容风险检查。 Args: text: 待检查的文本内容 Returns: 检查结果包含风险等级和问题列表 risk_words [违禁词示例] # 实际项目里用敏感词库代替 hit_words [w for w in risk_words if w in text] return { level: high if hit_words else low, hit_words: hit_words, suggestion: 请删除或替换高亮词汇 if hit_words else 可以发布 } if __name__ __main__: # 以标准输入输出模式启动 MCP Server。 # 这个模式适合本地进程调用便于 Dify、Claude Desktop 等宿主连接。 mcp.run(transportstdio)这段代码有几个设计要点每个工具函数都带有 docstringMCP 协议会把函数签名和描述语法发给模型模型根据这些信息决定何时调用。返回值尽量使用可序列化结构避免返回 Python 对象。工具粒度要适中既不能太粗也不能太细。如果工具太大模型难以理解调用边界如果太细Agent 需要多次调用才能完成一个目标。4.4 MCP 配置与客户端连接MCP Server 编写完成后需要让宿主程序知道如何启动它。标准做法是在配置文件里声明 command 和 args。在 Claude Desktop、Cherry Studio、Dify 这类工具里MCP 配置一般长这样{ mcpServers: { video-tools: { command: python, args: [src/mcp/video_server.py] } } }在 win 系统、macOS 上要注意命令路径和环境变量。推荐用虚拟环境的绝对路径例如{ mcpServers: { video-tools: { command: C:/Users/yourname/.venvs/video-agent/Scripts/python.exe, args: [D:/projects/video-agent-studio/src/mcp/video_server.py] } } }这里有一个隐藏的坑启动宿主程序时尽量让 MCP Server 的命令指向绝对路径否则宿主进程的资源路径不一致会导致 Server 启动失败。在自研 Python 应用里可以用 MCP 客户端 SDK 连接 Server。下面是一个伪代码级别的客户端封装# 文件路径src/mcp/client.py # 说明以 MCP 官方 Python SDK 为示例实际 API 以你安装的版本为准 from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client class McpToolClient: def __init__(self, server_name: str, command: str, args: list): self.server_params StdioServerParameters( commandcommand, argsargs ) async def call_tool(self, tool_name: str, arguments: dict): async with stdio_client(self.server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() result await session.call_tool(tool_name, arguments) return result实际项目中一个 Agent 往往需要连续调用多个工具不可能每次都重新建立连接。正确做法是在应用启动时一次性建立连接把 session 缓存起来复用。后面在编排层会用到这个思路。5. 实现多智能体编排层5.1 Agent 基类与 LLM 适配为了让多个 Agent 的代码风格统一先设计一个轻量级 Agent 基类。基类里维护角色名称、系统提示词、可用工具三个核心字段。# 文件路径src/agents/base.py from typing import Callable, Dict, List class Agent: 所有 Agent 的基类。 Attributes: name: Agent 名称用于日志和编排 system_prompt: 角色系统提示词 tools: 该 Agent 可调用的工具集合键为工具名值为调用函数 llm: 大模型客户端需要实现 complete 方法 def __init__(self, name: str, system_prompt: str, llm, tools: Dict[str, Callable] None): self.name name self.system_prompt system_prompt self.llm llm self.tools tools or {} def run(self, input_data: str) - str: 子类实现具体逻辑默认情况下直接调用模型。 Args: input_data: 输入文本 Returns: Agent 处理后的输出文本 messages [ {role: system, content: self.system_prompt}, {role: user, content: input_data} ] return self.llm.complete(messages) def call_tool(self, tool_name: str, arguments: dict): if tool_name not in self.tools: raise ValueError(fAgent {self.name} 未注册工具 {tool_name}) return self.tools[tool_name](arguments)LLM 客户端抽象如下# 文件路径src/llm/base.py from abc import ABC, abstractmethod class BaseLLMClient(ABC): 大模型客户端抽象类所有模型适配器都要继承它。 abstractmethod def complete(self, messages: list, temperature: float 0.7) - str: 传入消息列表返回模型生成的文本。 Args: messages: OpenAI 风格消息列表 temperature: 采样温度 passOpenAI 兼容客户端的实现可以用 requests也可以用官方 SDK思路是把消息发送到模型接口解析返回内容。# 文件路径src/llm/openai_client.py import requests from .base import BaseLLMClient class OpenAICompatClient(BaseLLMClient): 面向 OpenAI 兼容接口的客户端。 很多模型服务商提供 OpenAI 兼容接口只需要替换 base_url 和 api_key。 def __init__(self, base_url: str, api_key: str, model: str): self.base_url base_url.rstrip(/) /chat/completions self.api_key api_key self.model model def complete(self, messages: list, temperature: float 0.7) - str: payload { model: self.model, messages: messages, temperature: temperature } headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } resp requests.post(self.base_url, jsonpayload, headersheaders, timeout60) resp.raise_for_status() data resp.json() return data[choices][0][message][content]这里刻意不绑定具体厂商因为不同厂商的模型能力和价格差异很大实际项目中要按业务需要选择。比如字幕生成、敏感词检查这类任务可以用快且便宜的模型策划和审阅这类需要综合推理的任务用更强的大模型。5.2 策划 Agent选题与结构策划 Agent 的职责是把一个模糊主题变成内容大纲。它的系统提示词里需要强调目标人群分析、内容切入点、结构节奏、时长控制。# 文件路径src/agents/planner.py from .base import Agent PLANNER_PROMPT 你是一位资深短视频策划。 你的任务是根据用户给定的主题产出一份适合短视频平台的创作大纲。 大纲需要包含 1. 目标人群画像 2. 内容切入点 3. 视频结构开头-中段-结尾 4. 每个环节的预估时长 5. 情绪节奏设计 要求 - 结构清晰每条内容用序号标出 - 控制整条视频在用户期望时长内 - 开头要有钩子结尾要有引导关注或讨论的互动点 class PlannerAgent(Agent): def __init__(self, llm, toolsNone): super().__init__( namePlannerAgent, system_promptPLANNER_PROMPT, llmllm, toolstools ) def run(self, topic: str) - str: user_input f视频主题{topic}\n请输出完整创作大纲。 return super().run(user_input)这个 Agent 输出的是纯文本大纲为了后续 Agent 能稳定解析最好在提示词里要求结构化输出或者在后置处理里把文本转成 JSON。实际项目建议用 JSON 输出并做 schema 校验避免脏数据往下游传播。5.3 脚本 Agent文案与口播稿脚本 Agent 接收策划 Agent 的大纲输出完整口播稿和分镜文本。它还需要把文案按“镜头”拆开方便后续视觉和剪辑 Agent 处理。# 文件路径src/agents/script_writer.py import json from .base import Agent SCRIPT_PROMPT 你是一位短视频编剧。 你会收到一份创作大纲需要把它扩展成完整脚本。 输出格式为 JSON 数组每个元素包含 - id: 编号 - scene_text: 画面描述 - voiceover: 口播台词 - duration: 预估时长秒 要求 - 口播台词要口语化、有节奏感 - 画面描述要具体便于后续生成视觉提示词 - 每条时长之和要接近大纲总时长 class ScriptAgent(Agent): def __init__(self, llm, toolsNone): super().__init__( nameScriptAgent, system_promptSCRIPT_PROMPT, llmllm, toolstools ) def run(self, plan_text: str) - list: raw super().run(plan_text) # 这里做简单解析实际项目需要更健壮的 JSON 提取逻辑 start raw.find([) end raw.rfind(]) 1 json_str raw[start:end] return json.loads(json_str)脚本输出使用 JSON 数组而不是 Markdown是因为多智能体系统里每个 Agent 的输出都要被下一个 Agent 程序化消费。文本读起来方便但程序解析麻烦JSON 读起来稍差但可靠。5.4 视觉 Agent分镜与画面提示词视觉 Agent 的职责是把分镜文本转换成适合图片生成模型使用的提示词同时调用素材检索工具找到可用素材。# 文件路径src/agents/visual_planner.py import json from .base import Agent VISUAL_PROMPT 你是一位视觉导演。 你会收到一个分镜对象需要输出以下内容 1. image_prompt: 适合图片生成模型的英文提示词 2. key_elements: 画面中的关键元素列表 3. material_keyword: 素材库检索用的中文关键词 输出格式为 JSON 对象。 class VisualAgent(Agent): def run(self, scene: dict) - dict: raw super().run(json.dumps(scene, ensure_asciiFalse)) # 解析 JSON并调用素材检索工具 result json.loads(raw) if self.tools and search_material in self.tools: materials self.tools[search_material]( queryresult.get(material_keyword, ), material_typevideo ) result[candidate_materials] materials return result注意这里的 search_material 不是直接调用 MCP Server而是通过工具映射注入进来的。在应用启动时我们会把 MCP 工具包装成普通 Python 函数传给 Agent这样 Agent 层不需要关心工具是本地实现还是远程 MCP 协议通信。5.5 剪辑指令 Agent素材与转场剪辑指令 Agent 是流水线里最接近“最终产物”的节点。它接收分镜列表、画面提示词、素材候选列表输出一段结构化的剪辑指令。# 文件路径src/agents/editor.py import json from .base import Agent EDITOR_PROMPT 你是一位剪辑指导。 你会收到一个分镜列表和素材信息需要输出一条剪辑指令 JSON包含 - scenes: 场景数组每个场景包含镜头号、素材 ID、时长、字幕文本 - transitions: 转场列表 - audio: 音频设置对象 要求 - 根据口播时长确定每个镜头的时长 - 素材 ID 优先从候选素材里选选不到就标记 need_replace - 转场风格要符合内容调性 class EditorAgent(Agent): def run(self, storyboard_with_materials: dict) - dict: raw super().run(json.dumps(storyboard_with_materials, ensure_asciiFalse)) return json.loads(raw)在更复杂的实现里Editor Agent 还可以调用剪辑软件的 MCP Server直接生成可导入剪辑工程的文件。不过这一步依赖具体剪辑软件是否提供接口落地时通常需要定制开发。5.6 审阅 Agent质量与风险检查审阅环节是质量保障的关键也是多智能体系统相比单 Agent 的优势所在。这里我用两个角色实现“正反博弈加裁判”的模式Critic Agent负责挑毛病从逻辑、节奏、文案质量、合规风险等角度提出修改意见。Reviewer Agent负责做最终裁决判断修改后的版本是否达到发布标准。# 文件路径src/agents/reviewer.py import json from .base import Agent CRITIC_PROMPT 你是一位严格的短视频审阅人。 你会收到一份剪辑指令和脚本内容请从以下维度提出具体修改意见 1. 内容逻辑是否连贯 2. 开头前 3 秒是否有吸引力 3. 情绪节奏是否合理 4. 文案是否有语病或敏感表述 5. 镜头时长分配是否合理 输出格式为 JSON 数组每个元素包含 - issue: 问题描述 - severity: 严重程度可选 high / medium / low - suggestion: 修改建议 REVIEWER_PROMPT 你是一位总编审。 你会收到一个视频方案的初版和修改版需要输出最终裁决 - status: passed 或 rejected - final_comment: 最终意见 判断标准 - 如果所有 high 级问题已解决且剩余问题不影响观看可判定 passed - 如果仍有 high 级问题判定 rejected并说明还需修改的原因 class CriticAgent(Agent): def __init__(self, llm, toolsNone): super().__init__( nameCriticAgent, system_promptCRITIC_PROMPT, llmllm, toolstools ) def run(self, edit_commands: dict) - list: raw super().run(json.dumps(edit_commands, ensure_asciiFalse)) start raw.find([) end raw.rfind(]) 1 return json.loads(raw[start:end]) class ReviewerAgent(Agent): def __init__(self, llm, toolsNone): super().__init__( nameReviewerAgent, system_promptREVIEWER_PROMPT, llmllm, toolstools ) def run(self, original: dict, revised: dict, issues: list) - dict: payload { original: original, revised: revised, critic_issues: issues } raw super().run(json.dumps(payload, ensure_asciiFalse)) return json.loads(raw)这个博弈循环的价值在于Critic Agent 不需要“修改”能力Reviewer Agent 不需要“创作”能力每个角色只做自己最擅长的事比让同一个 Agent 既写又改可靠得多。5.7 调度器流水线串联调度器是整个系统的中枢。它负责决定 Agent 的执行顺序、传递中间结果、处理博弈循环以及把最终结果写入输出文件。# 文件路径src/orchestrator/pipeline.py import json from pathlib import Path from typing import Dict from src.agents.editor import EditorAgent from src.agents.planner import PlannerAgent from src.agents.reviewer import CriticAgent, ReviewerAgent from src.agents.script_writer import ScriptAgent from src.agents.visual_planner import VisualAgent class VideoCreationPipeline: 视频创作流水线调度器。 Args: agents: 已经初始化好的 Agent 实例字典 output_dir: 输出目录 max_review_rounds: 审阅最大轮数防止死循环 def __init__(self, agents: Dict[str, object], output_dir: str output, max_review_rounds: int 3): self.agents agents self.output_dir Path(output_dir) self.output_dir.mkdir(parentsTrue, exist_okTrue) self.max_review_rounds max_review_rounds def run(self, topic: str): # 1. 策划阶段 plan self.agents[planner].run(topic) self.output_dir.joinpath(plan.md).write_text(plan, encodingutf-8) print([Pipeline] 策划大纲已生成) # 2. 脚本阶段 script self.agents[script].run(plan) self.output_dir.joinpath(script.json).write_text( json.dumps(script, ensure_asciiFalse, indent2), encodingutf-8 ) print([Pipeline] 脚本已生成) # 3. 视觉阶段 storyboard_with_materials [] for scene in script: visual self.agents[visual].run(scene) storyboard_with_materials.append({ **scene, image_prompt: visual.get(image_prompt, ), candidate_materials: visual.get(candidate_materials, []) }) # 4. 剪辑指令阶段 edit_commands self.agents[editor].run({ storyboard: storyboard_with_materials, total_duration: sum(s.get(duration, 5) for s in script) }) # 5. 审阅博弈阶段 current edit_commands for round_idx in range(self.max_review_rounds): issues self.agents[critic].run(current) if not issues: break # 如果审阅意见为空说明没有高优问题直接通过 high_issues [i for i in issues if i.get(severity) high] if not high_issues: break # 根据意见让 Editor 修改实际项目里需要再注入一个修改 Agent # 这里简化处理本示例直接进入第二轮审阅真实实现请补充分支 break # 角色分配说明真实业务里这里应该有一个 Revise Agent 负责修改 # 由于篇幅限制本示例把修改逻辑放到了 Editor Agent 的二次调用里 final_commands current self.output_dir.joinpath(edit_commands.json).write_text( json.dumps(final_commands, ensure_asciiFalse, indent2), encodingutf-8 ) print([Pipeline] 剪辑指令已输出) # 6. 最终裁决 verdict self.agents[reviewer].run( originaledit_commands, revisedfinal_commands, issuesissues if issues else [] ) self.output_dir.joinpath(review_result.json).write_text( json.dumps(verdict, ensure_asciiFalse, indent2), encodingutf-8 ) return { plan: plan, script: script, edit_commands: final_commands, review: verdict }这里为了演示清晰审阅循环里没有真的把修改后的内容喂回去继续迭代。工程落地时可以增加一个 ReviseAgent专门消费 Critic 的意见并生成修改版然后由 Reviewer 判断是否通过。下面单独给出一个 review_loop.py 模块专门负责这个博弈循环。# 文件路径src/orchestrator/review_loop.py class ReviewLoop: 正反博弈 裁判循环。 执行流程 1. Critic Agent 对当前版本提出问题 2. 如果没有 high 级问题循环结束 3. 如果有问题Revise Agent 根据意见进行修改 4. Reviewer Agent 对修改后的版本做裁决 5. 通过则结束不通过则继续下一轮 def __init__(self, critic_agent, revise_agent, reviewer_agent, max_rounds3): self.critic_agent critic_agent self.revise_agent revise_agent self.reviewer_agent reviewer_agent self.max_rounds max_rounds def run(self, initial_version: dict): current initial_version history [] for round_no in range(1, self.max_rounds 1): issues self.critic_agent.run(current) history.append({ round: round_no, issues: issues, version: current }) high_issues [i for i in issues if i.get(severity) high] if not high_issues: print(f[ReviewLoop] 第 {round_no} 轮无高优问题通过审阅) return current, history print(f[ReviewLoop] 第 {round_no} 轮发现 {len(high_issues)} 个高优问题开始修改) current self.revise_agent.run(current, high_issues) verdict self.reviewer_agent.run({ original: initial_version, revised: current, issues: issues }) if verdict.get(status) passed: print(f[ReviewLoop] 第 {round_no} 轮修改后通过终审) return current, history print(f[ReviewLoop] 达到最大轮数 {self.max_rounds}返回最后版本) return current, history这个 ReviewLoop 模块可以直接复用到不同的内容生产场景。除了视频创作写文章、做海报、生成报告只要有“初稿—批评—修改—裁决”的质量需求都可以套用。6. 运行完整流程与结果验证6.1 启动 MCP Server先在一个终端窗口启动 MCP Servercd video-agent-studio python src/mcp/video_server.py如果配置正确MCP Server 会以 stdio 模式启动不打印日志到标准输出而是等待宿主进程通过标准输入输出发送协议消息。如果你在终端里看到了启动日志通常意味着被误打印到了 stdout需要把调试日志改成输出到 stderr。6.2 编写程序入口并执行流水线在程序入口里做依赖注入把 MCP 工具和 LLM 客户端装配到各 Agent 中然后运行流水线。# 文件路径src/main.py from src.llm.openai_client import OpenAICompatClient from src.mcp.video_server import search_material, check_content_risk from src.agents.planner import PlannerAgent from src.agents.script_writer import ScriptAgent from src.agents.visual_planner import VisualAgent from src.agents.editor import EditorAgent from src.agents.reviewer import CriticAgent, ReviewerAgent from src.orchestrator.pipeline import VideoCreationPipeline # 1. 初始化 LLM 客户端 # 实际使用中把 base_url、api_key、model 替换成自己的配置 llm OpenAICompatClient( base_urlhttps://your-model-endpoint.example.com/v1, api_keyyour-api-key, modelyour-model-name ) # 2. 初始化 MCP 工具映射 # 这里直接把函数引用注入 Agent实际项目可以通过 MCP 客户端动态拉取工具列表 mcp_tools { search_material: lambda args: search_material(**args), check_content_risk: lambda args: check_content_risk(**args), } # 3. 初始化各 Agent planner PlannerAgent(llmllm) script_agent ScriptAgent(llmllm) visual_agent VisualAgent(llmllm) editor EditorAgent( llmllm, tools{ search_material: mcp_tools[search_material], } ) critic CriticAgent(llmllm, tools{ check_content_risk: mcp_tools[check_content_risk], }) reviewer ReviewerAgent(llmllm) agents { planner: planner, script: script_agent, visual: visual_agent, editor: editor, critic: critic, reviewer: reviewer, } # 4. 运行流水线 pipeline VideoCreationPipeline( agentsagents, output_diroutput ) result pipeline.run(如何看待 MCP 协议对 AI 应用开发的影响)运行命令cd video-agent-studio python src/main.py6.3 输出结果说明执行完成后output 目录下会生成以下文件plan.md策划大纲供人工预览。script.json分镜与口播脚本数组。edit_commands.json剪辑指令包含场景、转场、音频设置。review_result.json审阅终裁结果。这套输出的设计理念是“人机协同”机器生成的内容不直接进剪辑软件而是作为半成品让真人快速确认。一方面避免全自动流程出错造成返工另一方面让创作者保留最终控制权。7. 常见问题与排查思路7.1 高频问题清单问题现象常见原因解决思路MCP Server 启动失败Python 路径不对或依赖未安装使用虚拟环境绝对路径确认 mcp / fastmcp 已安装Agent 调用工具时报错“工具不存在”Agent 实例未注入工具映射检查初始化时 tools 参数是否传递脚本 Agent 输出无法解析成 JSON模型返回了 Markdown 格式文本在提示词中强制 JSON 输出增加 JSON 提取与容错逻辑审阅循环死循环没有设置最大轮数设置 max_review_rounds并统计每轮修改是否真正产生 diff上下文过大超出限制多轮 Agent 结果全部塞入同一对话每个 Agent 使用独立上下文工具返回结果只保留摘要字段MCP 工具返回数据太大素材接口返回整个视频 URL 和长描述在 MCP Server 端做字段裁剪和分页某模型对工具描述理解差工具函数名和 docstring 不清晰优化工具命名和描述必要时拆细工具粒度7.2 深挖几个典型问题第一个典型问题是“上下文过大”。在多智能体系统里如果设计不当每一轮的中间结果都会累积到下一轮对话中很快就会触发模型上下文窗口限制报错信息通常类似“上下文大小超出限制”。排查思路是先确认上下文增长来自哪里。如果每个 Agent 都接收前一个 Agent 的完整原始输出那么脚本阶段把策划全文、视觉阶段把脚本全文、剪辑阶段把分镜全文全部堆积到一起token 数量会爆炸。解决方案有三个方向每个 Agent 独立上下文窗口不向前继承。工具返回结果做摘要化处理只保留核心字段。对中间结果做结构化压缩例如把策划大纲沉淀成固定字段的 JSON而不是保留大段自然语言文本。第二个典型问题是“Agent 工具调用不稳定”。同一个模型面对同样的问题有时调用一次工具有时调用三四次甚至选择错误工具。原因是工具描述不够清晰或者工具之间边界模糊。建议为每个工具写完整的 docstring明确参数含义和返回值结构。同时工具数量控制在可理解范围内。如果一个 Agent 挂了 20 个工具模型选择困难会显著增加。第三个典型问题是“多智能体流程调试困难”。单 Agent 还可以看对话日志多 Agent 的失败可能发生在任何一个节点。建议在每个 Agent 的 run 方法入口出口都记录结构化的执行日志内容包括输入摘要、输出摘要、耗时、调用模型、调用的工具列表。用日志系统做全链路追踪才能快速定位是哪个环节出了问题。8. 最佳实践与工程建议8.1 Agent 角色与 prompt 管理Agent 的 system prompt 不要直接写在代码里。虽然示例里为了直观把 prompt 放到了 .py 文件里但工程上更推荐统一管理# 文件路径config/agents.yaml planner: name: PlannerAgent model: strong-model prompt: | 你是一位资深短视频策划... script_writer: name: ScriptAgent model: strong-model prompt: | 你是一位短视频编剧... visual_planner: name: VisualAgent model: fast-model prompt: | 你是一位视觉导演...这样做的价值在于提示词调优是高频操作把 prompt 独立到配置文件里可以避免每次修改都要重新部署代码。8.2 MCP 工具设计原则第一工具命名要“动词加名词”例如 search_material、generate_edit_command避免 ambiguous 的命名。第二工具返回值要精简。很多下游 Agent 只需要素材 ID 和标题不需要完整的资源 URL 和二进制元数据。MCP Server 内部可以只返回必要字段必要时让 Agent 指定是否要详细版本。第三工具要有权限设计意识。视频创作系统里可能涉及删除素材、覆盖工程文件等危险操作这类工具不能暴露给普通 Agent。MCP Server 内部要做身份校验至少在 Host 层控制哪些 Agent 可以调用哪些工具。8.3 上下文与成本控制多智能体系统最大的成本陷阱是 token 浪费。同一份分镜内容在视觉、剪辑、审阅三个 Agent 里重复出现会成倍增加 token 消耗。几个实用的成本控制手段下游 Agent 只接收上游 Agent 的结构化摘要而不是全文。明确每个 Agent 的输入 schema用 Pydantic 做数据校验。对耗时长的流程设置超时和重试上限。不同 Agent 使用不同规格模型。简单抽取任务用快模型复杂推理用强模型。8.4 安全与权限边界这属于所有 AI 应用工程落地都必须重视的部分。MCP Server 暴露的是真实业务能力如果不加控制Agent 可能误调用删除、覆盖、发送等危险操作。至少要做到几点所有非查询类操作都要增加二次确认机制。MCP 工具参数做白名单校验。敏感操作记录操作人、操作 Agent、入参、出参。涉及生产素材库时先在小范围测试验证再放开权限。8.5 日志与可观测性多智能体流程的失败通常不是程序崩溃而是“产出结果不符合预期”。所以日志要能回答几个问题每个 Agent 收到了什么输入它调用哪些工具每个工具返回值是什么它生成了什么输出在哪一轮循环里结果发生了变化推荐使用结构化的 JSON 日志把每个 Agent 的输入输出摘要、token 用量、耗时、工具调用列表全部记录下来。这样即使某个环节出现问题也能根据日志快速回放整个执行过程。模块化的流水线设计可以让不同的 Agent 替换、插拔比如把 ScriptAgent 从“通用模型”切换到“专门微调过文案风格的模型”只需要改配置不改调度逻辑。这才是多智能体系统比单体 Agent 更具工程扩展性的根本原因。如果后续想继续深入建议优先研究两个方向一是怎么让 Agent 的中间产物用更小的 token 传递比如引入分镜元数据 schema 和向量检索让下游 Agent 按需获取详情而不是全量接收。二是尝试把博弈审阅循环用到更长周期的内容生产任务中比如系列视频策划、完整短视频账号的内容矩阵规划这时多智能体的协作优势会比单条视频更明显。动手跑一遍本文的示例代码比看十篇概念解读都有用。跑通之后再试着替换一个真实场景的 MCP 工具整个体系就会慢慢变成你自己的创作基础设施。