Conductor

name: company-intel
description: Fetch and analyze company data

flow:
  # Fetch company data
  - name: fetch
    operation: http
    config:
      url: https://api.company-data.com/lookup?domain=${input.domain}
      cache_ttl: 3600  # Cache for 1 hour

  # Analyze with AI
  - name: analyze
    operation: think
    component: [email protected]  # Versioned prompt via Edgit
    config:
      model: claude-3-5-sonnet-20241022
    input:
      company_data: ${fetch.output}

  # Validate quality
  - name: score
    agent: validator  # Pre-built agent
    input:
      content: ${analyze.output}

output:
  domain: ${input.domain}
  analysis: ${analyze.output}
  quality_score: ${score.output.score}
  cached: ${fetch.cached}

ensemble conductor deploy

# Execute via API (requires authentication)
curl https://your-worker.workers.dev/api/v1/execute/ensemble/company-intel \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"input": {"domain": "stripe.com"}}'

# Or define an HTTP trigger for public access
# See triggers documentation for auth options

flow:
  - name: analyzer
    operation: think
    component: [email protected]
    config:
      model: claude-3-5-sonnet-20241022

flow:
  - name: scraper
    agent: scraper  # Use pre-built
    config:
      url: https://example.com

name: my-workflow

flow:
  # Sequential by default
  - name: fetch
    operation: http

  - name: analyze
    operation: think

  # Parallel execution
  - parallel:
      - name: financial-analysis
        operation: think
      - name: legal-analysis
        operation: think

  # Conditional execution
  - name: manual-review
    agent: hitl
    condition: ${analyze.score < 0.8}

output:
  result: ${analyze.output}
  required_review: ${manual-review.completed}

state:
  schema:
    company_data: object
    analyses: array

flow:
  - name: fetch
    operation: http
    state:
      set: [company_data]  # Write to state

  - name: analyze
    operation: think
    state:
      use: [company_data]  # Read from state
      set: [analyses]       # Append to array

flow:
  - name: fetch
    operation: http
    config:
      cache_ttl: 3600  # KV cache for 1 hour

  - name: analyze
    operation: think
    config:
      model: claude-3-5-sonnet-20241022
      # AI Gateway caches automatically

flow:
  # Sequential
  - name: step1
    operation: http

  - name: step2
    operation: think

  # Parallel
  - parallel:
      - name: task-a
        operation: think
      - name: task-b
        operation: think

  # Conditional
  - name: fallback
    operation: think
    condition: ${step2.failed}

  # Loop
  - name: process-batch
    operation: code
    loop:
      items: ${input.items}

  # Retry
  - name: api-call
    operation: http
    retry:
      max_attempts: 3
      backoff: exponential

name: ab-test

flow:
  # Variant A: GPT-4 with conservative prompt
  - name: variant-a
    operation: think
    component: [email protected]
    config:
      model: gpt-4

  # Variant B: Claude with aggressive prompt
  - name: variant-b
    operation: think
    component: [email protected]
    config:
      model: claude-3-5-sonnet-20241022

output:
  variant_a_result: ${variant-a.output}
  variant_b_result: ${variant-b.output}
  # Compare and pick winner

# ensembles/blog-post.yaml
name: blog-post
description: Render a blog post page

trigger:
  - type: http
    path: /blog/:slug
    methods: [GET]
    public: true
    rateLimit:
      requests: 100
      window: 60

flow:
  # Fetch post data
  - name: fetch-post
    operation: http
    config:
      url: https://api.cms.com/posts/${params.slug}

  # Render the page
  - name: render
    operation: html
    config:
      templateEngine: handlebars
      template: |
        <h1>{{fetch-post.output.title}}</h1>
        <div class="content">
          {{fetch-post.output.content}}
        </div>

output:
  _raw: ${render.output}

name: versioned-workflow

flow:
  - name: scraper
    agent: [email protected]          # Versioned agent via Edgit

  - name: analyzer
    operation: think
    component: [email protected]  # Versioned component via Edgit
    config:
      model: claude-3-5-sonnet-20241022

name = "my-ai-workflow"
main = "src/index.ts"
compatibility_date = "2024-11-01"

# KV for caching
[[kv_namespaces]]
binding = "CACHE"
id = "your_kv_namespace_id"

# D1 for structured data
[[d1_databases]]
binding = "DB"
database_name = "analytics"
database_id = "your_d1_database_id"

# AI Gateway
[ai]
binding = "AI_GATEWAY"
gateway_id = "your_gateway_id"

import { describe, it, expect } from 'vitest';
import { Executor, MemberLoader } from '@ensemble-edge/conductor';
import { stringify } from 'yaml';

describe('My Ensemble', () => {
  it('should execute successfully', async () => {
    // Setup with proper ExecutionContext mock
    const env = {} as Env;
    const ctx = {
      waitUntil: (promise: Promise<any>) => promise,
      passThroughOnException: () => {}
    } as ExecutionContext;

    const executor = new Executor({ env, ctx });
    const loader = new MemberLoader({ env, ctx });

    // Register your agents
    const agent = loader.registerAgent(agentConfig, agentFunction);
    executor.registerAgent(agent);

    // Execute ensemble
    const result = await executor.executeFromYAML(
      stringify(myEnsemble),
      { input: 'test data' }
    );

    // Assert
    expect(result.success).toBe(true);
    expect(result.value.output).toBeDefined();
  });
});