- Context Compaction System with token counting and summarization - Deterministic State Machine for flow control (no LLM decisions) - Parallel Execution Engine (up to 12 concurrent sessions) - Event-Driven Coordination via Event Bus - Agent Workspace Isolation (tools, memory, identity, files) - YAML Workflow Integration (OpenClaw/Lobster compatible) - Claude Code integration layer - Complete demo UI with real-time visualization - Comprehensive documentation and README Components: - agent-system/: Context management, token counting, subagent spawning - pipeline-system/: State machine, parallel executor, event bus, workflows - skills/: AI capabilities (LLM, ASR, TTS, VLM, image generation, etc.) - src/app/: Next.js demo application Total: ~100KB of production-ready TypeScript code
9.6 KiB
9.6 KiB
Deterministic Multi-Agent Pipeline System
A comprehensive, open-source implementation of Claude Code-level architecture for building deterministic, parallel, event-driven multi-agent pipelines.
🎯 Key Features
| Feature | Description |
|---|---|
| Deterministic Orchestration | State machine controls flow, not LLM decisions |
| Parallel Execution | 4 projects × 3 roles = 12 concurrent agent sessions |
| Event-Driven Coordination | Agents finish work → next step triggers automatically |
| Full Agent Capabilities | Each agent gets tools, memory, identity, workspace |
| YAML Workflow Support | OpenClaw/Lobster-compatible workflow definitions |
📦 Package Contents
pipeline-system/
├── core/
│ └── state-machine.ts # Deterministic state machine engine
├── engine/
│ └── parallel-executor.ts # Parallel execution with worker pools
├── events/
│ └── event-bus.ts # Event-driven coordination system
├── workspace/
│ └── agent-workspace.ts # Isolated agent workspaces
├── workflows/
│ └── yaml-workflow.ts # YAML workflow parser (Lobster-compatible)
├── integrations/
│ └── claude-code.ts # Claude Code integration layer
└── index.ts # Main exports
🚀 Quick Start
Installation
bun add z-ai-web-dev-sdk
Copy pipeline-system/ to your project.
Basic Usage
import { PipelineOrchestrator, runWorkflow } from './pipeline-system';
// Option 1: Create a code pipeline
const orchestrator = new PipelineOrchestrator();
await orchestrator.initialize();
const pipelineId = await orchestrator.createPipeline({
name: 'Code Pipeline',
projects: [
{
id: 'project-1',
name: 'Authentication Module',
tasks: [
{ type: 'implement', description: 'Create auth module', role: 'programmer' },
{ type: 'review', description: 'Review code', role: 'reviewer' },
{ type: 'test', description: 'Test implementation', role: 'tester' }
]
}
],
roles: ['programmer', 'reviewer', 'tester'],
maxConcurrency: 12
});
// Option 2: Run predefined workflow
const workflowId = await runWorkflow('code-pipeline', {
projectId: 'my-project',
requirements: 'Build REST API'
});
// Subscribe to events
orchestrator.onEvent('agent.completed', (event) => {
console.log('Agent completed:', event.payload);
});
📐 Architecture
┌─────────────────────────────────────────────────────────────────┐
│ Pipeline Orchestrator │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ State Machine│ │ Event Bus │ │Parallel Exec │ │
│ │(Deterministic)│ │(Coordination)│ │ (Concurrency)│ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │ │
│ ┌──────┴────────────────┴─────────────────┴──────┐ │
│ │ Agent Workspaces │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │Programmer│ │Reviewer │ │ Tester │ │ │
│ │ │ • Tools │ │ • Tools │ │ • Tools │ │ │
│ │ │ • Memory │ │ • Memory│ │ • Memory│ │ │
│ │ │ • Files │ │ • Files │ │ • Files │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ │ │
│ └────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ LLM Provider (ZAI SDK) │
└─────────────────────────────────────────────────────────────────┘
🔄 State Machine
Define States
const definition: StateMachineDefinition = {
id: 'code-pipeline',
name: 'Code Pipeline',
initial: 'start',
states: {
start: {
id: 'start',
name: 'Start',
type: 'start',
onExit: [{ event: 'start', target: 'code' }]
},
code: {
id: 'code',
name: 'Code',
type: 'action',
agent: 'programmer',
timeout: 300000,
retry: { maxAttempts: 2, backoff: 'exponential' },
onExit: [
{ event: 'completed', target: 'review' },
{ event: 'failed', target: 'failed' }
]
},
review: {
id: 'review',
name: 'Review',
type: 'choice',
onExit: [
{ event: 'approved', target: 'test', condition: { type: 'equals', field: 'approved', value: true } },
{ event: 'rejected', target: 'code' }
]
},
test: {
id: 'test',
name: 'Test',
type: 'action',
agent: 'tester',
onExit: [
{ event: 'passed', target: 'end' },
{ event: 'failed', target: 'failed' }
]
},
end: { id: 'end', name: 'End', type: 'end' },
failed: { id: 'failed', name: 'Failed', type: 'end' }
}
};
⚡ Parallel Execution
const executor = new ParallelExecutionEngine({
maxWorkers: 4,
maxConcurrentPerWorker: 3,
taskTimeout: 300000
});
executor.start();
// Submit parallel tasks
const tasks = executor.submitBatch([
{ projectId: 'p1', role: 'programmer', type: 'implement', description: 'Auth module', priority: 'high' },
{ projectId: 'p2', role: 'programmer', type: 'implement', description: 'Payment module', priority: 'high' },
{ projectId: 'p3', role: 'programmer', type: 'implement', description: 'Dashboard', priority: 'medium' },
{ projectId: 'p4', role: 'programmer', type: 'implement', description: 'API service', priority: 'medium' }
]);
📨 Event Bus
const eventBus = new EventBus();
// Subscribe to events
eventBus.subscribe({
eventType: 'code.written',
handler: async (event) => {
console.log('Code written:', event.payload);
// Trigger review
eventBus.publish({
type: 'review.start',
source: 'coordinator',
payload: { projectId: event.payload.projectId }
});
}
});
// Publish events
eventBus.publish({
type: 'code.written',
source: 'programmer-1',
payload: { projectId: 'p1', files: ['auth.ts', 'auth.test.ts'] }
});
📁 YAML Workflows (Lobster-Compatible)
id: code-pipeline
name: Code Pipeline
initial: start
states:
start:
type: start
on:
start: code
code:
type: action
role: programmer
timeout: 30m
retry:
maxAttempts: 2
backoff: exponential
on:
completed: review
failed: failed
review:
type: choice
on:
approved: test
rejected: code
test:
type: action
role: tester
on:
passed: end
failed: test_failed
test_failed:
type: choice
on:
retry: code
abort: failed
end:
type: end
failed:
type: end
🤝 Integration with Claude Code & OpenClaw
Claude Code Integration
import { PipelineOrchestrator } from './pipeline-system';
const orchestrator = new PipelineOrchestrator();
await orchestrator.initialize();
// Create pipeline for Claude Code project
const pipelineId = await orchestrator.createPipeline({
name: 'Claude Code Pipeline',
projects: [
{
id: 'claude-project',
name: 'Claude Integration',
tasks: [
{ type: 'implement', description: 'Add MCP server', role: 'programmer' },
{ type: 'review', description: 'Review changes', role: 'reviewer' },
{ type: 'test', description: 'Test integration', role: 'tester' }
]
}
]
});
OpenClaw Integration
import { runWorkflow } from './pipeline-system';
// Run Lobster-compatible workflow
const workflowId = await runWorkflow('parallel-projects', {
projects: ['project1', 'project2', 'project3', 'project4'],
roles: ['programmer', 'reviewer', 'tester']
});
📊 Predefined Workflows
| Workflow | Description |
|---|---|
code-pipeline |
Code → Review → Test with max 3 review iterations |
parallel-projects |
Run multiple projects in parallel |
human-approval |
Workflow with human approval gates |
📄 License
MIT License - Free to use, modify, and distribute.
🤝 Contributing
Contributions welcome! This is designed for easy integration with Claude Code and OpenClaw.