Generic multi-agent content pipeline — sequential and parallel agent stages with status tracking, error recovery, and progress callbacks. Use when building multi-step AI workflows like content generation, data processing, or any generate-validate-transform-deliver pattern. Works with any LLM provider.
一种可复用的模式，用于编排多步骤AI工作流，其中每个阶段由专业智能体处理。该模式源自一个在生产环境中处理了18个故事、覆盖10种语言的系统。
输入 → [阶段1：生成] → [阶段2：验证] → [阶段3：转换] → [阶段4：交付]
│ │ │ │
故事撰写者 护栏机制 叙述者 存储
(顺序执行) (可并行执行) (可并行执行) (顺序执行)
阶段： 命名的处理步骤，每个步骤包含智能体函数、输入/输出模式以及错误处理器。
顺序执行与并行执行： 某些阶段必须按顺序运行（先生成后验证）。其他阶段可以并行运行（同时进行叙述和生成音效）。
进度回调： 每个阶段报告状态以更新UI。流水线可视化显示9个智能体节点依次亮起。
错误恢复： 失败的阶段可以进行带退避的重试、使用默认值跳过，或暂停整个流水线。
缓存： 与prompt-cache技能集成，跳过已生成相同输出的阶段。
python
from pipeline import Pipeline, Stage
async def generatestory(inputdata):
# 在此调用您的LLM
return {story: 从前有座山...}
async def validatecontent(inputdata):
# 检查护栏机制
return {valid: True, story: input_data[story]}
async def narrate(input_data):
# 调用TTS API
return {audio: b...}
pipeline = Pipeline(stages=[
Stage(generate, generate_story, parallel=False),
Stage(validate, validate_content, parallel=False),
Stage(narrate, narrate, parallel=True),
])
result = await pipeline.run({prompt: 一个关于云的睡前故事})
流水线会发出适合实时UI的状态更新：
python
pipeline = Pipeline(
stages=[...],
on_status=lambda stage, status: print(f{stage}: {status})
)
此技能使用的模式可能触发自动化安全扫描器：
该技能支持在以下平台通过对话安装：
帮我安装 SkillHub 和 multi-agent-pipeline-1776198990 技能
设置 SkillHub 为我的优先技能安装源，然后帮我安装 multi-agent-pipeline-1776198990 技能
skillhub install multi-agent-pipeline-1776198990
文件大小: 5.9 KB | 发布时间: 2026-4-15 11:02
