stream-chain

Stream-Chain 技能

执行复杂的多步骤工作流，让每个 Agent 的输出流入下一个 Agent，从而实现复杂的数据转换和顺序处理管道。

概述

Stream-Chain 提供两种强大的模式，用于编排多 Agent 工作流：

1. 自定义链（run）：完全自主地执行自定义提示词序列
2. 预定义管道（pipeline）：使用经过实践验证的常见任务工作流

链中的每个步骤都会接收上一步的完整输出，通过流式数据流实现复杂的多 Agent 协作。

---

快速开始

运行自定义链

claude-flow stream-chain run \
  "Analyze codebase structure" \
  "Identify improvement areas" \
  "Generate action plan"

执行管道

claude-flow stream-chain pipeline analysis

---

自定义链（run）

使用自定义提示词执行流式链，灵活性最高。

语法

claude-flow stream-chain run <prompt1> <prompt2> [...] [options]

要求：

• 至少需要 2 条提示词
• 每条提示词都会成为链中的一个步骤
• 输出会依次流经所有步骤

选项

选项	说明	默认值
--verbose	显示详细的执行信息	false
--timeout <seconds>	每个步骤的超时时间	30
--debug	启用调试模式并记录完整日志	false

上下文流转方式

每个步骤都会将上一步的输出作为上下文接收：

Step 1: "Write a sorting function"
Output: [function implementation]

Step 2 receives:
  "Previous step output:
  [function implementation]

  Next task: Add comprehensive tests"

Step 3 receives:
  "Previous steps output:
  [function + tests]

  Next task: Optimize performance"

示例

基础开发链

claude-flow stream-chain run \
  "Write a user authentication function" \
  "Add input validation and error handling" \
  "Create unit tests with edge cases"

安全审计工作流

claude-flow stream-chain run \
  "Analyze authentication system for vulnerabilities" \
  "Identify and categorize security issues by severity" \
  "Propose fixes with implementation priority" \
  "Generate security test cases" \
  --timeout 45 \
  --verbose

代码重构链

claude-flow stream-chain run \
  "Identify code smells in src/ directory" \
  "Create refactoring plan with specific changes" \
  "Apply refactoring to top 3 priority items" \
  "Verify refactored code maintains behavior" \
  --debug

数据处理管道

claude-flow stream-chain run \
  "Extract data from API responses" \
  "Transform data into normalized format" \
  "Validate data against schema" \
  "Generate data quality report"

---

预定义管道（pipeline）

执行经过实践验证、针对常见开发任务优化的工作流。

语法

claude-flow stream-chain pipeline <type> [options]

可用管道

1. 分析管道

全面分析代码库并找出改进之处。

claude-flow stream-chain pipeline analysis

工作流步骤：

1. 结构分析：梳理目录结构并识别组件
2. 问题检测：查找潜在的改进点和问题
3. 改进建议：生成可执行的改进报告

适用场景：

• 熟悉新的代码库
• 评估技术债务
• 审查架构
• 审计代码质量

2. 重构管道

系统地重构代码并确定优先级。

claude-flow stream-chain pipeline refactor

工作流步骤：

1. 候选项识别：查找需要重构的代码
2. 优先级排序：制定按优先级排序的重构计划
3. 实施：为优先级最高的项目提供重构后的代码

适用场景：

• 减少技术债务
• 提升代码质量
• 现代化遗留代码
• 实施设计模式

3. 测试管道

全面生成测试并分析覆盖率。

claude-flow stream-chain pipeline test

工作流步骤：

1. 覆盖率分析：识别缺少测试的部分
2. 测试设计：为关键函数创建测试用例
3. 实施：生成包含断言的单元测试

适用场景：

• 提高测试覆盖率
• 支持 TDD 工作流
• 创建回归测试
• 质量保证

4. 优化管道

通过性能分析和实施来优化性能。

claude-flow stream-chain pipeline optimize

工作流步骤：

1. 性能分析：识别性能瓶颈
2. 策略：分析并提出优化方案
3. 实施：提供优化后的代码

适用场景：

• 提升性能
• 优化资源
• 增强可扩展性
• 降低延迟

管道选项

选项	说明	默认值
--verbose	显示详细的执行信息	false
--timeout <seconds>	每个步骤的超时时间	30
--debug	启用调试模式	false

管道示例

快速分析

claude-flow stream-chain pipeline analysis

扩展重构

claude-flow stream-chain pipeline refactor --timeout 60 --verbose

调试测试生成

claude-flow stream-chain pipeline test --debug

全面优化

claude-flow stream-chain pipeline optimize --timeout 90 --verbose

管道输出

每次管道执行都会提供：

• 进度：逐步显示执行状态
• 结果：每个步骤的成功或失败
• 耗时：总执行时间和每个步骤的执行时间
• 摘要：汇总结果和建议

---

自定义流水线定义

在 .claude-flow$config.json 中定义可复用的流水线：

配置格式

{
  "streamChain": {
    "pipelines": {
      "security": {
        "name": "Security Audit Pipeline",
        "description": "Comprehensive security analysis",
        "prompts": [
          "Scan codebase for security vulnerabilities",
          "Categorize issues by severity (critical$high$medium$low)",
          "Generate fixes with priority and implementation steps",
          "Create security test suite"
        ],
        "timeout": 45
      },
      "documentation": {
        "name": "Documentation Generation Pipeline",
        "prompts": [
          "Analyze code structure and identify undocumented areas",
          "Generate API documentation with examples",
          "Create usage guides and tutorials",
          "Build architecture diagrams and flow charts"
        ]
      }
    }
  }
}

执行自定义流水线

claude-flow stream-chain pipeline security
claude-flow stream-chain pipeline documentation

---

高级用例

多 Agent 协调

针对复杂工作流串联不同类型的 Agent：

claude-flow stream-chain run \
  "Research best practices for API design" \
  "Design REST API with discovered patterns" \
  "Implement API endpoints with validation" \
  "Generate OpenAPI specification" \
  "Create integration tests" \
  "Write deployment documentation"

数据转换流水线

通过多个阶段处理和转换数据：

claude-flow stream-chain run \
  "Extract user data from CSV files" \
  "Normalize and validate data format" \
  "Enrich data with external API calls" \
  "Generate analytics report" \
  "Create visualization code"

代码迁移工作流

通过系统化流程迁移代码并进行验证：

claude-flow stream-chain run \
  "Analyze legacy codebase dependencies" \
  "Create migration plan with risk assessment" \
  "Generate modernized code for high-priority modules" \
  "Create migration tests" \
  "Document migration steps and rollback procedures"

质量保证链

全面的代码质量工作流：

claude-flow stream-chain pipeline analysis
claude-flow stream-chain pipeline refactor
claude-flow stream-chain pipeline test
claude-flow stream-chain pipeline optimize

---

最佳实践

1. 使用清晰具体的提示词

推荐：

"Analyze authentication.js for SQL injection vulnerabilities"

避免：

"Check security"

2. 合理安排流程

按顺序排列提示词，让后续步骤基于前面的输出：

1. "Identify the problem"
2. "Analyze root causes"
3. "Design solution"
4. "Implement solution"
5. "Verify implementation"

3. 设置合适的超时时间

• 简单任务：30 秒（默认值）
• 分析任务：45-60 秒
• 实现任务：60-90 秒
• 复杂工作流：90-120 秒

4. 添加验证步骤

在链中加入验证：

claude-flow stream-chain run \
  "Implement feature X" \
  "Write tests for feature X" \
  "Verify tests pass and cover edge cases"

5. 迭代优化

使用链进行迭代改进：

claude-flow stream-chain run \
  "Generate initial implementation" \
  "Review and identify issues" \
  "Refine based on issues found" \
  "Final quality check"

---

与 Claude Flow 集成

与 Swarm 协调结合

# Initialize swarm for coordination
claude-flow swarm init --topology mesh

# Execute stream chain with swarm agents
claude-flow stream-chain run \
  "Agent 1: Research task" \
  "Agent 2: Implement solution" \
  "Agent 3: Test implementation" \
  "Agent 4: Review and refine"

内存集成

Stream 链会自动将上下文存储在内存中，以便跨会话持久保存：

# Execute chain with memory
claude-flow stream-chain run \
  "Analyze requirements" \
  "Design architecture" \
  --verbose

# Results stored in .claude-flow$memory$stream-chain/

神经模式训练

成功的链会训练神经模式，从而提升性能：

# Enable neural training
claude-flow stream-chain pipeline optimize --debug

# Patterns learned and stored for future optimizations

---

故障排除

链超时

如果步骤超时，请增大超时时间：

claude-flow stream-chain run "complex task" --timeout 120

上下文丢失

如果上下文传递不正常，请使用 --debug：

claude-flow stream-chain run "step 1" "step 2" --debug

未找到流水线

请确认流水线名称和自定义定义：

# Check available pipelines
cat .claude-flow$config.json | grep -A 10 "streamChain"

---

性能特征

• 吞吐量：每分钟 2-5 个步骤（视复杂度而异）
• 上下文大小：每个步骤最多 100K 个 token
• 内存使用量：每条活动链约 50MB
• 并发：支持并行执行链

---

相关技能

• SPARC 方法论：系统化开发工作流
• Swarm 协调：多 Agent 编排
• 内存管理：持久化上下文存储
• 神经模式：自适应学习

---

示例代码库

完整开发工作流

# Full feature development chain
claude-flow stream-chain run \
  "Analyze requirements for user profile feature" \
  "Design database schema and API endpoints" \
  "Implement backend with validation" \
  "Create frontend components" \
  "Write comprehensive tests" \
  "Generate API documentation" \
  --timeout 60 \
  --verbose

代码审查流水线

# Automated code review workflow
claude-flow stream-chain run \
  "Analyze recent git changes" \
  "Identify code quality issues" \
  "Check for security vulnerabilities" \
  "Verify test coverage" \
  "Generate code review report with recommendations"

迁移助手

# Framework migration helper
claude-flow stream-chain run \
  "Analyze current Vue 2 codebase" \
  "Identify Vue 3 breaking changes" \
  "Create migration checklist" \
  "Generate migration scripts" \
  "Provide updated code examples"

---

总结

Stream-Chain 可通过以下方式实现复杂的多步骤工作流：

• 顺序处理：每个步骤都基于前面的结果
• 保留上下文：完整的输出历史会沿链传递
• 灵活编排：使用自定义链或预定义流水线
• Agent 协调：自然的多 Agent 协作模式
• 数据转换：通过简单步骤完成复杂处理

自定义工作流使用 run，经过实战验证的解决方案使用 pipeline。