qe-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。