# Claude Code Workflow多Agent编排全实战
## 概述
本教程深入讲解Claude Code的Workflow引擎——一个支持多Agent并行编排、结构化输出、流水线处理的确定性脚本系统。你将学会使用`pipeline`、`parallel`、`agent`等核心原语构建复杂的多步骤自动化任务，掌握屏障同步、循环直到收敛、对抗验证等高级模式。适合已有Claude Code基础使用经验、希望将AI能力扩展到批量处理和复杂编排场景的开发者。
**学习目标：**
- 理解Workflow引擎的执行模型与脚本结构
- 掌握`pipeline`、`parallel`、`agent`三大编排原语
- 构建完整的代码审查、批量迁移、深度调研等实战Workflow
- 学会结构化输出、对抗验证、预算控制等高级模式
**适用人群：** 已完成Claude Code入门，熟悉基本交互操作，想要自动化复杂开发流程的中高级用户。
**预计教学时长：** 3-4小时（含动手实践）
---
## 前置条件
| 项目 | 要求 |
|------|------|
| Claude Code版本 | 最新版（2025年6月后） |
| Node.js | v18+ |
| 基础知识 | Claude Code基本使用、JavaScript/TypeScript基础 |
| 可选 | 一个测试用的Git仓库 |
| API额度 | 建议至少$50的API额度用于实践 |
---
## 第一章：Workflow引擎架构解析
### 1.1 什么是Workflow引擎
Workflow引擎是Claude Code内置的多Agent编排系统。它允许你用JavaScript脚本定义确定性的执行流程——在多个AI Agent之间分配任务、协调并行执行、收集结果并做出决策。
与普通对话模式的关键区别：
```
普通模式： 用户 → Claude → 单次响应
Workflow： 用户 → 脚本 → [Agent1, Agent2, ...] → 汇总 → 结果
```
Workflow的核心价值在于**确定性控制流**。Agent的调用顺序、并行策略、结果处理逻辑都由你的脚本精确控制，而不是依赖模型的自主决策。
### 1.2 执行模型
```mermaid
graph TB
subgraph "Workflow执行引擎"
S[脚本启动] --> P[解析meta声明]
P --> R[注册phase]
R --> B[执行脚本体]
B --> A1[agent调用1]
B --> A2[agent调用2]
B --> A3[agent调用3]
A1 --> C[结果收集]
A2 --> C
A3 --> C
C --> RET[返回结果]
end
subgraph "Agent池（并发上限 min 16, cpu-2）"
A1
A2
A3
end
RET --> W[写入输出]
W --> LOG[记录到transcript]
```
关键约束：
- **并发上限**：`min(16, cpu_cores - 2)` 个Agent同时运行
- **Agent总数上限**：每个Workflow生命周期最多1000个Agent
- **预算控制**：通过`budget`对象实时监控token消耗
- **隔离模式**：Agent可运行在独立worktree中，避免文件冲突
### 1.3 脚本结构
每个Workflow脚本必须遵循以下结构：
```javascript
// 1. meta声明（必须放在最前面，纯字面量）
export const meta = {
name: 'my-workflow',
description: '一句话描述',
phases: [
{ title: '发现', detail: '扫描代码库' },
{ title: '分析', detail: '逐文件分析' },
{ title: '修复', detail: '应用修改' }
]
}
// 2. 脚本体（异步执行）
phase('发现')
const results = await agent('扫描所有TODO标记', { schema: TODO_SCHEMA })
phase('分析')
const analysis = await pipeline(
results.items,
item => agent(`分析: ${item.file}`, { label: `analyze:${item.file}` }),
detail => agent(`深入分析: ${detail.title}`, { label: `deep:${detail.id}` })
)
phase('修复')
log(`发现 ${analysis.length} 个需要修复的项`)
```
**重要约束：**
- `meta`必须是纯字面量对象，不能包含变量、函数调用、模板字符串
- `phases`数组中的`title`必须与`phase()`调用的字符串完全匹配
- 脚本运行在异步上下文中，直接使用`await`
- `Date.now()`/`Math.random()`/无参`new Date()`不可用（会破坏恢复机制）
---
## 第二章：三大编排原语
### 2.1 agent()—— 基本构建块
`agent()`是最基础的调用单元，启动一个子Agent执行任务并返回结果。
**基础签名：**
```javascript
const result = await agent(prompt, options?)
```
**Options参数详解：**
```javascript
await agent('分析这个文件', {
label: 'analyze:main', // 显示标签，覆盖默认标签
phase: 'Review', // 分配到指定phase组
schema: ZOD_SCHEMA, // 强制结构化输出
model: 'sonnet', // 模型覆盖（一般省略）
isolation: 'worktree', // 在独立worktree中运行
agentType: 'Explore' // 使用特定Agent类型
})
```
**返回值规则：**
- 无`schema`：返回Agent的最终文本（string）
- 有`schema`：返回验证后的JSON对象
- Agent被跳过时返回`null`
- Agent出错时抛出异常（pipeline中会变为`null`）
**schema示例——强制结构化输出：**
```javascript
const REVIEW_SCHEMA = {
type: 'object',
properties: {
bugs: {
type: 'array',
items: {
type: 'object',
properties: {
file: { type: 'string' },
line: { type: 'number' },
severity: { type: 'string', enum: ['high', 'medium', 'low'] },
description: { type: 'string' },
suggestedFix: { type: 'string' }
},
required: ['file', 'line', 'severity', 'description']
}
},
summary: { type: 'string' }
},
required: ['bugs', 'summary']
}
const result = await agent(
'审查 src/auth.ts 中的安全漏洞',
{ schema: REVIEW_SCHEMA, label: 'review:auth' }
)
// result.bugs 是一个结构化的数组
console.log(`发现 ${result.bugs.length} 个问题`)
```
### 2.2 parallel()—— 屏障同步
`parallel()`并行执行多个异步任务，**等待全部完成后返回**。这是一个同步屏障。
```javascript
const results = await parallel([
() => agent('分析前端代码质量', { schema: QUALITY_SCHEMA, phase: 'Review' }),
() => agent('分析后端代码质量', { schema: QUALITY_SCHEMA, phase: 'Review' }),
() => agent('分析测试覆盖率', { schema: COVERAGE_SCHEMA, phase: 'Review' })
])
// results 是一个数组，顺序与输入对应
// 出错的任务返回 null，不会导致整体失败
const validResults = results.filter(Boolean)
```
**何时使用parallel（需要全部结果才能继续）：**
- 多维度审查后需要交叉比对
- 去重操作需要完整结果集
- 需要判断总数后决定是否继续
**何时不使用parallel：**
- 各项之间没有依赖关系 → 用`pipeline`
- 只需要简单并行不需要屏障 → 用`pipeline`
- 中间有纯数据变换 → 在pipeline阶段内处理
### 2.3 pipeline()—— 流水线处理
`pipeline()`是默认推荐的编排模式。每个item独立通过所有阶段，**阶段之间没有屏障**——Item A可以在阶段3时，Item B还在阶段1。
```mermaid
gantt
title Pipeline vs Parallel 执行时序
dateFormat X
axisFormat %s秒
section Pipeline模式
Item A-阶段1 :a1, 0, 3
Item A-阶段2 :a2, 3, 5
Item A-阶段3 :a3, 5, 7
Item B-阶段1 :b1, 1, 4
Item B-阶段2 :b2, 4, 6
Item B-阶段3 :b3, 6, 9
Item C-阶段1 :c1, 2, 5
Item C-阶段2 :c2, 5, 7
Item C-阶段3 :c3, 7, 10
section 如果用Parallel+Sequential
等待所有阶段1完成 :crit, 0, 5
Item A-阶段2 :a2b, 5, 7
Item B-阶段2 :b2b, 5, 7
Item C-阶段2 :c2b, 5, 7
等待所有阶段2完成 :crit2, 7, 7
```
**基础用法：**
```javascript
const FILES = ['src/auth.ts', 'src/api.ts', 'src/db.ts']
const results = await pipeline(
FILES,
// 阶段1：发现
file => agent(`扫描 ${file} 中的问题`, {
label: `scan:${file}`,
phase: 'Scan',
schema: ISSUES_SCHEMA
}),
// 阶段2：分析每个发现
(issues, originalFile, index) =>
issues.issues.length > 0
? agent(`深入分析 ${originalFile} 的问题`, {
label: `analyze:${originalFile}`,
phase: 'Analyze',
schema: ANALYSIS_SCHEMA
})
: null,
// 阶段3：修复
(analysis, originalFile, index) =>
analysis && analysis.needsFix
? agent(`修复 ${originalFile}`, {
label: `fix:${originalFile}`,
phase: 'Fix'
})
: null
)
// 过滤掉 null（被跳过的项）
const fixed = results.filter(Boolean)
log(`成功修复 ${fixed.length} 个文件`)
```
**关键特性：**
- **Wall-clock时间 = 最慢的单条链路**，而非所有阶段的最慢项之和
- 阶段回调签名为`(prevResult, originalItem, index)`
- 任何阶段抛出异常 → 该item变为`null`并跳过后续阶段
- 原始item和index在所有阶段中都可访问
---
## 第三章：高级模式
### 3.1 对抗验证模式
防止AI Agent产生"看起来合理但实际上错误"的发现。为每个发现启动N个独立验证者，要求它们反驳该发现。
```javascript
const ADJUDICATION_SCHEMA = {
type: 'object',
properties: {
refuted: { type: 'boolean' },
reason: { type: 'string' },
confidence: { type: 'number' }
},
required: ['refuted', 'reason', 'confidence']
}
// 对每个发现启动3个独立验证者
const votes = await parallel(
Array.from({ length: 3 }, (_, i) => () =>
agent(
`尝试反驳以下发现。如果不确定，默认refuted=true。发现：${finding.description}`,
{
schema: ADJUDICATION_SCHEMA,
label: `verify:${i}`,
phase: 'Verify'
}
)
)
)
// 多数投票：3个中至少2个认为有效才保留
const validVotes = votes.filter(Boolean).filter(v => !v.refuted).length
const survives = validVotes >= 2
if (survives) {
log(`✅ 发现通过验证：${finding.description}`)
} else {
log(`❌ 发现被反驳：${finding.description}`)
}
```
### 3.2 多视角验证
比简单多数投票更强大的验证方式——每个验证者使用不同的审视维度：
```javascript
const LENSES = [
{ key: 'correctness', prompt: '从代码正确性角度验证' },
{ key: 'security', prompt: '从安全漏洞角度验证' },
{ key: 'performance', prompt: '从性能影响角度验证' },
{ key: 'reproducibility', prompt: '验证问题是否可重现' }
]
const judgements = await parallel(
LENSES.map(lens => () =>
agent(
`通过${lens.prompt}以下发现：${finding.description}。位置：${finding.file}:${finding.line}`,
{ schema: VERDICT_SCHEMA, label: `judge:${lens.key}`, phase: 'Verify' }
)
)
)
const confirmed = judgements
.filter(Boolean)
.filter(j => j.isReal)
.length >= 2 // 至少2个维度确认
```
### 3.3 循环直到收敛（Loop-until-dry）
用于不确定数量的发现任务——持续搜索直到连续K轮没有新发现：
```javascript
const seen = new Set()
const confirmed = []
let dry = 0
const MAX_DRY = 2 // 连续2轮无新发现则停止
while (dry < MAX_DRY) {
const round = await parallel(
FINDERS.map(f => () =>
agent(f.prompt, { phase: 'Find', schema: BUGS_SCHEMA })
)
)
const found = round.filter(Boolean).flatMap(r => r.bugs)
const fresh = found.filter(b => !seen.has(`${b.file}:${b.line}`))
if (fresh.length === 0) {
dry++
log(`第${dry}轮无新发现（共需${MAX_DRY}轮停止）`)
continue
}
dry = 0
fresh.forEach(b => seen.add(`${b.file}:${b.line}`))
// 对新发现进行验证
const verified = await parallel(
fresh.map(b => () =>
agent(`验证此发现：${b.description}`, { schema: VERDICT_SCHEMA, phase: 'Verify' })
)
)
confirmed.push(...verified.filter(Boolean).filter(v => v.isReal))
log(`本轮发现 ${fresh.length} 个新问题，累计确认 ${confirmed.length} 个`)
}
log(`搜索完成，共确认 ${confirmed.length} 个有效问题`)
```
### 3.4 预算控制模式
利用`budget`对象动态控制token消耗：
```javascript
// budget.total: 用户设定的token目标（如 +500k）
// budget.spent(): 已消耗的token
// budget.remaining(): 剩余token
const findings = []
if (!budget.total) {
// 没有预算限制，使用固定数量
const result = await agent('查找代码问题', { schema: BUGS_SCHEMA })
findings.push(...result.bugs)
} else {
// 有预算限制，动态循环
while (budget.remaining() > 50_000) {
const result = await agent('查找更多代码问题', { schema: BUGS_SCHEMA })
const newBugs = result.bugs.filter(b => !findings.some(f => f.id === b.id))
findings.push(...newBugs)
log(`发现 ${findings.length} 个问题，剩余 ${Math.round(budget.remaining() / 1000)}k tokens`)
}
}
log(`预算耗尽前共发现 ${findings.length} 个问题`)
```
### 3.5 Workflow嵌套
在一个Workflow中调用另一个已保存的Workflow：
```javascript
// 调用已保存的workflow
const subResult = await workflow('code-review-workflow', {
files: ['src/auth.ts']
})
// 或调用脚本文件
const deepResult = await workflow({
scriptPath: './deep-analysis.js'
}, { targetModule: 'auth' })
```
**注意：** 嵌套只支持一层——子Workflow中不能再调用`workflow()`。
---
## 第四章：实战案例——全链路代码审查系统
### 4.1 项目结构
我们将构建一个完整的多维度代码审查系统，包含发现、去重、验证、报告四个阶段。
```mermaid
flowchart TB
START[启动审查Workflow] --> DISCOVER[Phase: 发现]
subgraph DISCOVER["Phase: 发现（parallel）"]
D1[安全维度扫描]
D2[性能维度扫描]
D3[可维护性扫描]
end
DISCOVER --> DEDUP[去重合并]
DEDUP --> VERIFY[Phase: 验证]
subgraph VERIFY["Phase: 验证（对抗验证）"]
V1[验证者1: 尝试反驳]
V2[验证者2: 尝试反驳]
V3[验证者3: 尝试反驳]
end
VERIFY --> VOTE{多数投票}
VOTE -->|≥2票确认| KEEP[保留发现]
VOTE -->|<2票确认| DROP[丢弃发现]
KEEP --> REPORT[Phase: 报告]
DROP --> REPORT
REPORT --> END[输出结构化报告]
```
### 4.2 完整脚本
```javascript
export const meta = {
name: 'code-review-system',
description: '多维度代码审查：发现→去重→对抗验证→报告',
phases: [
{ title: '发现', detail: '三维度并行扫描' },
{ title: '验证', detail: '对抗验证每个发现' },
{ title: '报告', detail: '生成结构化报告' }
]
}
// ===== Schema定义 =====
const BUGS_SCHEMA = {
type: 'object',
properties: {
bugs: {
type: 'array',
items: {
type: 'object',
properties: {
id: { type: 'string' },
file: { type: 'string' },
line: { type: 'number' },
severity: { type: 'string', enum: ['critical', 'high', 'medium', 'low'] },
category: { type: 'string' },
title: { type: 'string' },
description: { type: 'string' },
suggestedFix: { type: 'string' }
},
required: ['id', 'file', 'line', 'severity', 'title', 'description']
}
}
},
required: ['bugs']
}
const VERDICT_SCHEMA = {
type: 'object',
properties: {
refuted: { type: 'boolean' },
reason: { type: 'string' },
confidence: { type: 'number' }
},
required: ['refuted', 'reason']
}
// ===== Phase 1: 发现 =====
phase('发现')
const DIMENSIONS = [
{
key: 'security',
prompt: `审查代码库中的安全问题。关注：
- SQL注入、XSS、CSRF
- 认证/授权绕过
- 敏感数据泄露
- 不安全的密码学使用
每个发现需要包含：文件路径、行号、严重程度、描述、建议修复方案`
},
{
key: 'performance',
prompt: `审查代码库中的性能问题。关注：
- N+1查询
- 内存泄漏
- 不必要的同步操作
- 缺少缓存
- 大循环中的低效操作`
},
{
key: 'maintainability',
prompt: `审查代码库中的可维护性问题。关注：
- 代码重复
- 过长的函数/类
- 缺少错误处理
- 不合理的依赖
- 违反SOLID原则`
}
]
const scanResults = await parallel(
DIMENSIONS.map(d => () =>
agent(d.prompt, {
label: `scan:${d.key}`,
phase: '发现',
schema: BUGS_SCHEMA
})
)
)
// ===== 去重 =====
const allBugs = scanResults
.filter(Boolean)
.flatMap(r => r.bugs)
const seen = new Set()
const deduped = allBugs.filter(bug => {
const key = `${bug.file}:${bug.line}:${bug.category || 'general'}`
if (seen.has(key)) return false
seen.add(key)
return true
})
log(`扫描发现 ${allBugs.length} 个问题，去重后 ${deduped.length} 个`)
if (deduped.length === 0) {
log('未发现任何问题，审查完成')
// 无问题，直接返回
} else {
// ===== Phase 2: 验证 =====
phase('验证')
const verified = await pipeline(
deduped,
// 对每个发现进行3人对抗验证
bug => parallel([
() => agent(
`你是一个严格的代码审查验证者。尝试反驳以下发现，如果不确定则默认refuted=true。
发现：${bug.title}
文件：${bug.file}:${bug.line}
描述：${bug.description}`,
{ schema: VERDICT_SCHEMA, label: `verify:1:${bug.id}`, phase: '验证' }
),
() => agent(
`从不同角度尝试反驳此发现。关注是否有误报可能。
发现：${bug.title}
文件：${bug.file}:${bug.line}
描述：${bug.description}
建议修复：${bug.suggestedFix}`,
{ schema: VERDICT_SCHEMA, label: `verify:2:${bug.id}`, phase: '验证' }
),
() => agent(
`验证此发现是否为真正的代码问题，而非设计选择或误报。
发现：${bug.title}
文件：${bug.file}:${bug.line}
严重程度：${bug.severity}`,
{ schema: VERDICT_SCHEMA, label: `verify:3:${bug.id}`, phase: '验证' }
)
]),
// 统计投票结果
(votes, originalBug) => {
const validVotes = votes.filter(Boolean)
const confirmed = validVotes.filter(v => !v.refuted).length
return {
...originalBug,
verified: confirmed >= 2,
confidence: confirmed / validVotes.length,
voteDetails: validVotes
}
}
)
const confirmed = verified.filter(v => v && v.verified)
const rejected = verified.filter(v => v && !v.verified)
log(`验证完成：${confirmed.length} 个确认，${rejected.length} 个被反驳`)
// ===== Phase 3: 报告 =====
phase('报告')
// 按严重程度分组
const grouped = {
critical: confirmed.filter(b => b.severity === 'critical'),
high: confirmed.filter(b => b.severity === 'high'),
medium: confirmed.filter(b => b.severity === 'medium'),
low: confirmed.filter(b => b.severity === 'low')
}
log(`=== 审查报告 ===`)
log(`严重：${grouped.critical.length} 个`)
log(`高：${grouped.high.length} 个`)
log(`中：${grouped.medium.length} 个`)
log(`低：${grouped.low.length} 个`)
// 返回结构化结果
return {
summary: {
totalScanned: allBugs.length,
afterDedup: deduped.length,
confirmed: confirmed.length,
rejected: rejected.length,
bySeverity: Object.fromEntries(
Object.entries(grouped).map(([k, v]) => [k, v.length])
)
},
findings: confirmed,
rejected: rejected.map(r => ({
...r,
rejectionReason: r.voteDetails?.find(v => v.refuted)?.reason
}))
}
}
```
### 4.3 运行Workflow
**方式一：在对话中触发**
在Claude Code对话中，用户可以请求运行Workflow：
```
请运行一个多维度代码审查workflow
```
**方式二：使用Workflow工具调用**
```bash
# Claude Code内部会调用Workflow工具
# 脚本会被持久化到session目录
```
### 4.4 预期输出示例
```
=== Workflow: code-review-system ===
Phase: 发现
├─ scan:security [完成] 发现 5 个问题
├─ scan:performance [完成] 发现 3 个问题
└─ scan:maintainability [完成] 发现 4 个问题
扫描发现 12 个问题，去重后 10 个
Phase: 验证
├─ verify:bug-001 ✅ 3/3 确认
├─ verify:bug-002 ✅ 2/3 确认
├─ verify:bug-003 ❌ 1/3 确认（被反驳）
├─ verify:bug-004 ✅ 3/3 确认
└─ ...
验证完成：7 个确认，3 个被反驳
Phase: 报告
严重：1 个
高：2 个
中：3 个
低：1 个
```
---
## 第五章：实战案例——批量代码迁移系统
### 5.1 场景描述
将一个项目从REST API迁移到GraphQL，涉及数百个文件的修改。Workflow的优势在于：每个文件的发现、修改、验证可以流水线化。
### 5.2 完整脚本
```javascript
export const meta = {
name: 'rest-to-graphql-migration',
description: '将REST API调用批量迁移到GraphQL',
phases: [
{ title: '发现', detail: '扫描所有REST调用点' },
{ title: '迁移', detail: '逐文件迁移到GraphQL' },
{ title: '验证', detail: '验证迁移正确性' }
]
}
const REST_CALL_SCHEMA = {
type: 'object',
properties: {
calls: {
type: 'array',
items: {
type: 'object',
properties: {
file: { type: 'string' },
line: { type: 'number' },
method: { type: 'string', enum: ['GET', 'POST', 'PUT', 'DELETE', 'PATCH'] },
endpoint: { type: 'string' },
surroundingCode: { type: 'string' }
},
required: ['file', 'line', 'method', 'endpoint']
}
}
},
required: ['calls']
}
const MIGRATION_RESULT_SCHEMA = {
type: 'object',
properties: {
success: { type: 'boolean' },
changes: { type: 'array', items: { type: 'string' } },
errors: { type: 'array', items: { type: 'string' } }
},
required: ['success']
}
// ===== Phase 1: 发现 =====
phase('发现')
const discovery = await agent(
`扫描整个代码库，找到所有REST API调用点。
包括：fetch/axios/XMLHttpRequest调用、api client封装、路由定义。
输出每个调用点的文件、行号、HTTP方法、端点路径。`,
{ schema: REST_CALL_SCHEMA, phase: '发现' }
)
// 按文件分组
const filesWithCalls = {}
for (const call of discovery.calls) {
if (!filesWithCalls[call.file]) filesWithCalls[call.file] = []
filesWithCalls[call.file].push(call)
}
const fileEntries = Object.entries(filesWithCalls)
log(`发现 ${discovery.calls.length} 个REST调用点，分布在 ${fileEntries.length} 个文件中`)
// ===== Phase 2: 迁移 =====
phase('迁移')
const migrations = await pipeline(
fileEntries,
// 阶段1：迁移单个文件（在独立worktree中避免冲突）
([file, calls]) => agent(
`将 ${file} 中的以下REST API调用迁移到GraphQL：
${calls.map(c => `- ${c.method} ${c.endpoint} (行${c.line})`).join('\n')}
迁移规则：
1. 将fetch/axios调用替换为GraphQL query/mutation
2. 保持相同的业务逻辑和数据流
3. 更新类型定义
4. 保留错误处理逻辑`,
{
schema: MIGRATION_RESULT_SCHEMA,
label: `migrate:${file}`,
phase: '迁移',
isolation: 'worktree' // 独立worktree避免文件冲突
}
),
// 阶段2：验证迁移结果
(result, [file, calls]) => {
if (!result || !result.success) return null
return agent(
`验证 ${file} 的迁移是否正确：
- 检查GraphQL查询是否等效于原REST调用
- 检查错误处理是否完整
- 检查类型安全性`,
{ schema: MIGRATION_RESULT_SCHEMA, label: `verify:${file}`, phase: '验证' }
)
}
)
const successful = migrations.filter(m => m && m.success)
log(`迁移完成：${successful.length}/${fileEntries.length} 个文件成功`)
```
### 5.3 关键设计决策
| 决策 | 原因 |
|------|------|
| 使用`pipeline`而非`parallel` | 文件迁移相互独立，无需等待全部完成 |
| `isolation: 'worktree'` | 多个Agent同时修改文件需要隔离 |
| 先发现后迁移 | 避免重复扫描，一次性收集所有调用点 |
| 迁移后立即验证 | 验证不需要等所有文件迁移完 |
---
## 第六章：实战案例——深度调研报告生成器
### 6.1 多模态搜索+深度阅读+对抗验证
```javascript
export const meta = {
name: 'deep-research-report',
description: '多源检索+证据汇编+对抗验证的结构化调研报告',
phases: [
{ title: '搜索', detail: '多角度并行搜索' },
{ title: '阅读', detail: '深度阅读关键源' },
{ title: '验证', detail: '对抗验证核心声明' },
{ title: '合成', detail: '生成结构化报告' }
]
}
const CLAIM_SCHEMA = {
type: 'object',
properties: {
claims: {
type: 'array',
items: {
type: 'object',
properties: {
claim: { type: 'string' },
sources: { type: 'array', items: { type: 'string' } },
confidence: { type: 'string', enum: ['high', 'medium', 'low'] }
},
required: ['claim', 'sources', 'confidence']
}
}
},
required: ['claims']
}
const REFUTATION_SCHEMA = {
type: 'object',
properties: {
canRefute: { type: 'boolean' },
evidence: { type: 'string' },
counterSource: { type: 'string' }
},
required: ['canRefute']
}
// args: { topic: "...", depth: 3 }
const topic = args.topic
// ===== Phase 1: 多角度搜索 =====
phase('搜索')
const SEARCH_ANGLES = [
`搜索关于"${topic}"的最新技术发展和论文`,
`搜索"${topic}"的实践案例和行业应用`,
`搜索"${topic}"的批评观点和局限性`
]
const searchResults = await parallel(
SEARCH_ANGLES.map((prompt, i) => () =>
agent(prompt, {
schema: CLAIM_SCHEMA,
label: `search:${i}`,
phase: '搜索'
})
)
)
const allClaims = searchResults
.filter(Boolean)
.flatMap(r => r.claims)
log(`搜索完成，收集 ${allClaims.length} 条声明`)
// ===== Phase 2: 深度阅读 =====
phase('阅读')
const deepReads = await parallel(
allClaims
.filter(c => c.confidence === 'high' || c.confidence === 'medium')
.slice(0, 10) // 限制深度阅读数量
.map(claim => () =>
agent(
`深入分析以下声明，评估其可靠性：${claim.claim}。参考来源：${claim.sources.join(', ')}`,
{ schema: CLAIM_SCHEMA, label: `read:${claim.claim.slice(0, 20)}`, phase: '阅读' }
)
)
)
// ===== Phase 3: 对抗验证 =====
phase('验证')
const verifiedClaims = await pipeline(
allClaims,
claim => parallel([
() => agent(
`尝试找到反驳此声明的证据：${claim.claim}`,
{ schema: REFUTATION_SCHEMA, label: 'refute:1', phase: '验证' }
),
() => agent(
`从不同角度质疑此声明：${claim.claim}`,
{ schema: REFUTATION_SCHEMA, label: 'refute:2', phase: '验证' }
)
]),
(refutations, originalClaim) => ({
...originalClaim,
verified: refutations.filter(Boolean).filter(r => !r.canRefute).length >= 1,
refutationAttempts: refutations.filter(Boolean)
})
)
const confirmedClaims = verifiedClaims.filter(c => c && c.verified)
log(`验证完成：${confirmedClaims.length}/${allClaims.length} 条声明通过验证`)
// ===== Phase 4: 报告合成 =====
phase('合成')
const report = await agent(
`基于以下已验证的声明，生成一份关于"${topic}"的结构化调研报告。
已验证声明：${JSON.stringify(confirmedClaims, null, 2)}
报告要求：
1. 执行摘要
2. 背景与现状
3. 关键发现（每个发现附来源和可信度）
4. 争议与局限性
5. 未来趋势
6. 参考资料`,
{ phase: '合成' }
)
return {
topic,
totalClaims: allClaims.length,
verifiedClaims: confirmedClaims.length,
report
}
```
---
## 第七章：常见问题与最佳实践
### 7.1 常见陷阱
**陷阱1：滥用parallel**
```javascript
// ❌ 错误：中间的map操作不需要屏障
const a = await parallel(items.map(i => () => agent(find(i))))
const b = a.map(transform) // 纯数据操作
const c = await parallel(b.map(i => () => agent(verify(i))))
// ✅ 正确：在pipeline中处理
const result = await pipeline(
items,
i => agent(find(i), { schema: FIND_SCHEMA }),
r => transform(r),
r => agent(verify(r), { schema: VERIFY_SCHEMA })
)
```
**陷阱2：在meta中使用计算值**
```javascript
// ❌ 错误
const phases = [{ title: 'Scan' }]
export const meta = { name: 'test', description: '...', phases } // 引用变量
// ✅ 正确：纯字面量
export const meta = {
name: 'test',
description: '...',
phases: [{ title: 'Scan' }]
}
```
**陷阱3：使用不可用的API**
```javascript
// ❌ 错误：Date.now()不可用
const timestamp = Date.now()
// ✅ 正确：从外部传入
// 在args中传入时间戳，或使用Workflow返回后由调用方添加
```
### 7.2 性能优化
| 策略 | 效果 | 示例 |
|------|------|------|
| 优先使用pipeline | 减少等待时间 | 独立任务不要用parallel |
| 合理设置schema | 减少输出token | 只声明需要的字段 |
| 限制Agent数量 | 控制成本 | `.slice(0, N)`限制批量大小 |
| 使用预算控制 | 动态调整 | `while (budget.remaining() > X)` |
| Worktree隔离 | 仅在需要时使用 | 只在Agent修改文件时才用 |
### 7.3 调试技巧
**1. 查看Workflow进度**
```
# 在另一个终端中
/workflows
```
**2. 使用log()追踪进度**
```javascript
log(`当前进度：${processed}/${total}`)
log(`发现 ${bugs.length} 个问题，剩余 ${Math.round(budget.remaining() / 1000)}k tokens`)
```
**3. 恢复中断的Workflow**
```javascript
// 使用 resumeFromRunId 恢复
// 已完成的Agent调用会返回缓存结果
// 只重新执行修改过或新增的调用
```
### 7.4 错误处理模式
```javascript
// pipeline中的错误处理
const results = await pipeline(
items,
item => {
try {
return agent(process(item), { schema: RESULT_SCHEMA })
} catch (e) {
log(`处理 ${item} 失败：${e.message}`)
return null // 返回null，pipeline会跳过后续阶段
}
},
(result, originalItem) => {
if (!result) return null // 安全检查
return agent(analyze(result), { schema: ANALYSIS_SCHEMA })
}
)
// parallel中的错误处理
const results = await parallel(tasks)
const valid = results.filter(Boolean) // 过滤掉失败的（null）
```
---
## 第八章：Workflow与其他编排方式的对比
```mermaid
graph LR
subgraph "Claude Code编排方式"
A[单次对话] --> B[最简单
单任务]
C["-p 非交互模式"] --> D["自动化单任务
适合管道"]
E[CronCreate] --> F["定时任务
周期执行"]
G[Workflow] --> H["多Agent编排
复杂流程"]
I[Hooks] --> J["事件触发
自动化钩子"]
end
```
| 特性 | 单次对话 | `-p`模式 | CronCreate | Workflow | Hooks |
|------|---------|---------|------------|----------|-------|
| 并行执行 | ❌ | ❌ | ❌ | ✅ | ❌ |
| 确定性控制流 | ❌ | ❌ | ❌ | ✅ | ✅ |
| 结构化输出 | ❌ | ❌ | ❌ | ✅ | ❌ |
| 多步骤编排 | 有限 | 有限 | ❌ | ✅ | 有限 |
| 预算控制 | ❌ | ❌ | ❌ | ✅ | ❌ |
| 持久化 | ❌ | ❌ | ✅ | ✅ | ✅ |
| 交互式 | ✅ | ❌ | ❌ | ❌ | ❌ |
| 适用场景 | 快速问答 | CI/CD | 定时巡检 | 批量处理 | Git hooks |
---
## 第九章：从设计到部署的完整清单
### 启动前检查
```
□ 明确Workflow的输入和输出
□ 定义所有需要的Schema
□ 规划phase结构
□ 评估Agent数量和成本
□ 确定是否需要worktree隔离
□ 准备好测试数据
```
### 脚本编写检查
```
□ meta声明是纯字面量
□ phase()调用的title与meta.phases匹配
□ 没有使用Date.now()/Math.random()
□ 并行策略选择正确（pipeline vs parallel）
□ 错误处理完备（null检查、filter(Boolean)）
□ log()语句充分
□ budget控制到位
```
### 运行后验证
```
□ 检查Agent完成率
□ 验证结果质量
□ 确认token消耗在预期范围
□ 检查是否有被跳过的item
□ 审查transcript日志
```
---
## 训练营适用性评估
| 训练营类型 | 适用性 | 教学建议 |
|-----------|--------|---------|
| AI工程化训练营 | ⭐⭐⭐⭐⭐ | 核心课程，建议6小时深度实践 |
| DevOps自动化训练营 | ⭐⭐⭐⭐ | 重点讲pipeline模式和CI/CD集成 |
| 代码质量训练营 | ⭐⭐⭐⭐ | 重点讲代码审查Workflow |
| 全栈开发训练营 | ⭐⭐⭐ | 作为高级选修，讲解基础编排模式 |
| AI入门训练营 | ⭐⭐ | 难度偏高，建议先掌握基础Claude Code |
**推荐教学路径（3小时精华版）：**
1. **30分钟**：架构讲解 + Mermaid图示
2. **45分钟**：三大原语动手练习（agent/parallel/pipeline）
3. **60分钟**：实战案例——代码审查系统
4. **30分钟**：高级模式（对抗验证、循环直到收敛）
5. **15分钟**：调试技巧 + 最佳实践
---
## 参考来源
1. [Anthropic官方文档 - Claude Code](https://docs.anthropic.com/en/docs/claude-code) — Claude Code核心概念与配置指南
2. [Claude Code GitHub仓库](https://github.com/anthropics/claude-code) — 工具源码与Workflow引擎实现细节
3. [Anthropic Blog - Building with Claude Code](https://www.anthropic.com/blog) — 官方最佳实践与案例分析
4. [Model Context Protocol规范](https://modelcontextprotocol.io/) — MCP协议参考，Agent间通信基础
5. [Claude Code社区示例集](https://github.com/anthropics/claude-code/tree/main/examples) — 官方Workflow脚本示例集合
> [!tip] 51学通信知识星球
> **星主：爱卫生**（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。
> **星球定位**：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。
> **当前规模**：5200 篇主题 / 990 个回答，198 元/年。
> **加入福利**：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。
> **AI 加持**：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。
