Multi-Agent

import {
  createMultiAgentOrchestrator,
  parallel,
  sequential,
  supervisor,
  selectAgent,
  runAgentRequirement,
  concatResults,
  collectOutputs,
  pickBestResult,
  aggregateTokens,
} from '@directive-run/ai';
import type {
  AgentLike,
  AgentRunner,
  RunResult,
  AgentRegistration,
  MultiAgentOrchestrator,
} from '@directive-run/ai';

// Define specialized agents – each has a distinct role in the pipeline
const researcher: AgentLike = {
  name: 'researcher',
  instructions: 'You are a research assistant. Find relevant information on the given topic.',
  model: 'gpt-4',
};

const writer: AgentLike = {
  name: 'writer',
  instructions: 'You are a technical writer. Write clear, concise content from research notes.',
  model: 'gpt-4',
};

const reviewer: AgentLike = {
  name: 'reviewer',
  instructions: 'You review drafts for accuracy and clarity. Return "approve" or revision notes.',
  model: 'gpt-4',
};

// Wrap your LLM SDK in a standard runner function
const runner: AgentRunner = async (agent, input, options) => {
  const result = await openaiAgentsRun(agent, input, options);

  return result;
};

const orchestrator = createMultiAgentOrchestrator({
  runner,

  // Register each agent with its configuration
  agents: {
    researcher: {
      agent: researcher,
      maxConcurrent: 3,               // Allow 3 parallel research runs
      timeout: 30000,                  // 30s timeout per run
      capabilities: ['search', 'summarize'],
      description: 'Finds and summarizes information on any topic',
    },
    writer: {
      agent: writer,
      maxConcurrent: 1,               // Only one writer at a time
      timeout: 60000,
      guardrails: {                    // Per-agent output guardrails
        output: [createPIIGuardrail({ redact: true })],
      },
    },
    reviewer: {
      agent: reviewer,
      maxConcurrent: 1,
      timeout: 30000,
    },
  },

  // Define reusable execution patterns
  patterns: {
    // Fan out to 3 researchers, merge their outputs with a separator
    research: parallel(
      ['researcher', 'researcher', 'researcher'],
      (results) => concatResults(results, '\n\n---\n\n'),
      { minSuccess: 2 }               // Succeed if at least 2 of 3 complete
    ),

    // Writer drafts, then reviewer checks – output flows from one to the next
    writeAndReview: sequential(
      ['writer', 'reviewer'],
      {
        transform: (output, agentId) =>
          agentId === 'writer'
            ? `Review this draft for accuracy:\n\n${output}`
            : String(output),
      }
    ),
  },

  // Handoff lifecycle hooks
  onHandoff: (request) => {
    console.log(`Handoff: ${request.fromAgent} → ${request.toAgent}`);
  },
  onHandoffComplete: (result) => {
    console.log(`Handoff complete in ${result.completedAt - result.request.requestedAt}ms`);
  },

  // Bounded history – how many completed handoffs to retain (default: 1000)
  maxHandoffHistory: 500,
});

const result = await orchestrator.runAgent<string>('researcher', 'What is WebAssembly?');

console.log(result.output);       // The agent's response
console.log(result.totalTokens);  // Token usage

const controller = new AbortController();
setTimeout(() => controller.abort(), 10000);

const result = await orchestrator.runAgent('researcher', 'Explain WASM', {
  signal: controller.signal,
});

// Execute the named "research" pattern – fans out to 3 researchers
const research = await orchestrator.runPattern<string>(
  'research',
  'Explain the benefits of constraint-driven architecture'
);
// Result is the concatenated outputs from all 3, separated by ---

// Broadcast the same input to multiple agents
const combined = await orchestrator.runParallel(
  ['researcher', 'researcher'],
  'What are WebSockets?',
  (results) => concatResults(results)
);

// Send different inputs to each agent in parallel
const answers = await orchestrator.runParallel(
  ['researcher', 'researcher', 'researcher'],
  ['Explain REST', 'Explain GraphQL', 'Explain gRPC'],
  (results) => collectOutputs(results)  // Returns string[]
);

Not enough successful results: 1/2

const result = await orchestrator.runPattern<string>(
  'writeAndReview',
  'Write a guide to Directive multi-agent orchestration'
);

const results = await orchestrator.runSequential<string>(
  ['researcher', 'writer', 'reviewer'],
  'Create a blog post about AI safety',
  {
    // Shape how each agent's output becomes the next agent's input
    transform: (output, agentId, index) => {
      if (agentId === 'researcher') {
        return `Write a blog post based on this research:\n\n${output}`;
      }
      if (agentId === 'writer') {
        return `Review this blog post draft:\n\n${output}`;
      }

      return String(output);
    },
  }
);

// Each step's RunResult is available in the array
const finalReview = results[results.length - 1].output;
const totalTokens = aggregateTokens(results);

const pipeline = sequential(
  ['researcher', 'writer', 'reviewer'],
  { continueOnError: true }
);

const manager: AgentLike = {
  name: 'manager',
  instructions: `You are a project manager. Analyze the request and delegate to workers.
    Respond with JSON: { "action": "delegate", "worker": "researcher"|"writer", "workerInput": "..." }
    Or when done: { "action": "complete", "output": "..." }`,
  model: 'gpt-4',
};

const orchestrator = createMultiAgentOrchestrator({
  runner,
  agents: {
    manager: { agent: manager, maxConcurrent: 1 },
    researcher: { agent: researcher, maxConcurrent: 3 },
    writer: { agent: writer, maxConcurrent: 1 },
  },

  patterns: {
    managed: supervisor('manager', ['researcher', 'writer'], {
      maxRounds: 5,    // Safety limit to prevent infinite delegation loops (default: 5)
      extract: (supervisorOutput, workerResults) => ({
        answer: supervisorOutput,
        sources: collectOutputs(workerResults),
        tokens: aggregateTokens(workerResults),
      }),
    }),
  },
});

const result = await orchestrator.runPattern('managed', 'Research and write about WASM');

Invalid worker: unknown-agent

const research = await orchestrator.runAgent('researcher', 'What is Directive?');

// Handoff with optional context metadata
const draft = await orchestrator.handoff(
  'researcher', 'writer',
  `Write an article based on this research:\n\n${research.output}`,
  { sourceTokens: research.totalTokens }  // Attached context
);

const review = await orchestrator.handoff(
  'writer', 'reviewer',
  `Review this article:\n\n${draft.output}`
);

const pending = orchestrator.getPendingHandoffs();
console.log(`${pending.length} handoffs in flight`);

interface HandoffRequest {
  id: string;                          // 'handoff-1', 'handoff-2', ...
  fromAgent: string;
  toAgent: string;
  input: string;
  context?: Record<string, unknown>;   // Optional attached metadata
  requestedAt: number;                 // Date.now() timestamp
}

interface HandoffResult {
  request: HandoffRequest;
  result: RunResult<unknown>;
  completedAt: number;
}

import { selectAgent } from '@directive-run/ai';

// Route complex queries to the expert agent
const routeToExpert = selectAgent(
  (facts) => facts.complexity > 0.8,      // When condition
  'expert',                                // Agent to select (string or function)
  (facts) => String(facts.query),          // Input to send (string or function)
  100                                       // Optional priority
);

// Dynamic agent selection based on facts
const dynamicRoute = selectAgent(
  (facts) => facts.needsProcessing === true,
  (facts) => facts.preferredAgent as string,   // Select agent dynamically
  (facts) => `Process this: ${facts.data}`
);

import { runAgentRequirement } from '@directive-run/ai';

// Use in a Directive constraint definition
const constraints = {
  needsResearch: {
    when: (facts) => facts.hasUnknowns,
    require: runAgentRequirement('researcher', 'Find relevant data', {
      priority: 'high',
    }),
  },
};

interface RunAgentRequirement extends Requirement {
  type: 'RUN_AGENT';
  agent: string;
  input: string;
  context?: Record<string, unknown>;
}

import { createMessageBus } from '@directive-run/ai';
import type { MessageBus, TypedAgentMessage } from '@directive-run/ai';

const bus = createMessageBus({
  maxHistory: 1000,            // Messages to retain in history
  defaultTtlMs: 3600000,      // 1 hour default message TTL
  maxPendingPerAgent: 100,     // Queue cap for offline agents
  onDelivery: (message, recipients) => {
    console.log(`Delivered ${message.type} to ${recipients.join(', ')}`);
  },
  onDeliveryError: (message, error) => {
    console.error(`Failed to deliver ${message.id}:`, error);
  },
});

// Subscribe to messages
const sub = bus.subscribe('writer', (message) => {
  console.log(`Writer received: ${message.type} from ${message.from}`);
}, {
  types: ['DELEGATION', 'REQUEST'],   // Filter by message type
  from: ['researcher'],                // Filter by sender
  priority: ['high', 'urgent'],        // Filter by priority
});

// Publish a message
const messageId = bus.publish({
  type: 'DELEGATION',
  from: 'researcher',
  to: 'writer',
  task: 'Write a summary',
  context: { data: '...' },
  priority: 'high',
});

// Query history
const history = bus.getHistory({ types: ['DELEGATION'] }, 50);
const specific = bus.getMessage(messageId);
const pending = bus.getPending('offline-agent');

// Cleanup
sub.unsubscribe();
bus.dispose();  // Clears all subscriptions, history, and pending queues

const bus = createMessageBus({
  persistence: {
    save: async (message) => { await db.insert('messages', message); },
    load: async (agentId, since) => { return db.query('messages', { to: agentId, after: since }); },
    delete: async (messageId) => { await db.delete('messages', messageId); },
    clear: async (agentId) => { await db.deleteAll('messages', agentId); },
  },
});

import { createAgentNetwork, createMessageBus } from '@directive-run/ai';
import type { AgentNetwork } from '@directive-run/ai';

const network = createAgentNetwork({
  bus: createMessageBus(),
  agents: {
    researcher: { capabilities: ['search', 'summarize'] },
    writer: { capabilities: ['draft', 'edit'] },
    reviewer: { capabilities: ['review', 'approve'] },
  },
  defaultTimeout: 30000,
  onAgentOnline: (agentId) => console.log(`${agentId} connected`),
  onAgentOffline: (agentId) => console.log(`${agentId} disconnected`),
});

// Request-response (waits for correlated RESPONSE)
const answer = await network.request(
  'writer', 'reviewer',
  'check-accuracy',
  { paragraph: 'WebAssembly compiles to...' },
  15000   // Optional per-request timeout
);
console.log(answer.success, answer.result);

// Delegation (waits for correlated DELEGATION_RESULT with metrics)
const result = await network.delegate(
  'researcher', 'writer',
  'Write an article about AI safety',
  { research: findingsData }
);
console.log(result.success, result.metrics?.durationMs);

// Query (request-response shorthand for questions)
const info = await network.query(
  'writer', 'reviewer',
  'Is this paragraph technically accurate?',
  { text: '...' }
);

// Fire-and-forget notification
network.send('researcher', 'writer', {
  type: 'INFORM',
  topic: 'research-complete',
  content: { documentId: 'doc-123' },
});

// Broadcast to all agents
network.broadcast('system', {
  type: 'INFORM',
  topic: 'shutdown',
  content: { reason: 'maintenance' },
});

// Capability-based discovery
const writers = network.findByCapability('draft');
console.log(writers.map((a) => a.id));  // ['writer']

// Dynamic registration
network.register('editor', { capabilities: ['proofread', 'format'] });
network.unregister('editor');

// Cleanup
network.dispose();

import { createResponder } from '@directive-run/ai';

const responder = createResponder(network, 'writer');

responder.onRequest('draft', async (payload) => {
  const draft = await generateDraft(payload.topic as string);

  return { success: true, result: draft };
});

responder.onRequest('edit', async (payload) => {
  const edited = await editDocument(payload.content as string);

  return { success: true, result: edited };
});

// Remove a handler
responder.offRequest('edit');

// Cleanup
responder.dispose();

import { createDelegator } from '@directive-run/ai';

const delegator = createDelegator(network, 'writer');

delegator.onDelegation(async (task, context) => {
  const start = Date.now();
  const result = await executeTask(task, context);

  return {
    success: true,
    result,
    metrics: {
      durationMs: Date.now() - start,
      tokensUsed: 500,
      cost: 0.003,
    },
  };
});

// Remove the handler
delegator.offDelegation();

// Cleanup
delegator.dispose();

import { createPubSub } from '@directive-run/ai';

const pubsub = createPubSub(network, 'analyst');

// Subscribe to topics (returns unsubscribe function)
const unsub = pubsub.subscribe(
  ['market-updates', 'alerts'],
  (topic, content) => {
    console.log(`[${topic}]`, content);
  }
);

// Publish to a topic (broadcasts UPDATE to all agents)
pubsub.publish('market-updates', { price: 100, change: 5 });

// Unsubscribe from specific topics
unsub();

// Cleanup
pubsub.dispose();

import {
  concatResults,
  collectOutputs,
  pickBestResult,
  aggregateTokens,
} from '@directive-run/ai';

// Join all string outputs with a separator (default: '\n\n')
const merged = concatResults(results, '\n\n---\n\n');

// Gather every output into a typed array
const outputs = collectOutputs(results);  // T[]

// Select the single best result using a custom scoring function
const best = pickBestResult(results, (r) => {
  return typeof r.output === 'string' ? r.output.length : 0;
});

// Sum token usage across every result
const totalTokens = aggregateTokens(results);

import { Semaphore } from '@directive-run/ai';

const sem = new Semaphore(3);  // 3 concurrent permits

const release = await sem.acquire();  // Waits if all 3 permits are taken
try {
  await doWork();
} finally {
  release();
}

// Inspect state
console.log(sem.available);  // Permits currently free
console.log(sem.waiting);    // Callers queued up
console.log(sem.max);        // Total permits (3)

// Reject all pending waiters and reset permits
sem.drain();

const state = orchestrator.getAgentState('researcher');
console.log(state.status);      // 'idle' | 'running' | 'completed' | 'error'
console.log(state.runCount);    // How many times this agent has run
console.log(state.totalTokens); // Cumulative token usage
console.log(state.lastInput);   // Last input string
console.log(state.lastOutput);  // Last output value
console.log(state.lastError);   // Last error message (if status is 'error')

// Iterate over all registered agents
const allStates = orchestrator.getAllAgentStates();
for (const [id, s] of Object.entries(allStates)) {
  console.log(`${id}: ${s.status} (${s.runCount} runs, ${s.totalTokens} tokens)`);
}

interface MultiAgentState {
  __agents: Record<string, {
    status: 'idle' | 'running' | 'completed' | 'error';
    lastInput?: string;
    lastOutput?: unknown;
    lastError?: string;
    runCount: number;
    totalTokens: number;
  }>;
  __handoffs: HandoffRequest[];
  __handoffResults: HandoffResult[];
}

// Reset all agent states, drain semaphores, clear handoff history
orchestrator.reset();

// Alias for reset – use when you're done with the orchestrator
orchestrator.dispose();

// Tolerates 1 failure out of 3
const research = parallel(
  ['researcher', 'researcher', 'researcher'],
  (results) => concatResults(results),
  { minSuccess: 2 }
);

No successful results in sequential pattern

Invalid worker: unknown-agent

// Throws: 'Unknown agent: nonexistent'
await orchestrator.runAgent('nonexistent', 'hello');

// Throws: 'Unknown pattern: nonexistent'
await orchestrator.runPattern('nonexistent', 'hello');

import {
  createMockAgentRunner,
  createMultiAgentOrchestrator,
  parallel,
  sequential,
  concatResults,
} from '@directive-run/ai';

const runner = createMockAgentRunner({
  responses: {
    researcher: 'Research findings about WebAssembly...',
    writer: 'WebAssembly (WASM) is a binary instruction format...',
    reviewer: 'APPROVED. The article is technically accurate.',
  },
  tokenCount: 150,
  latencyMs: 50,
});

const orchestrator = createMultiAgentOrchestrator({
  runner,
  agents: {
    researcher: { agent: { name: 'researcher' }, maxConcurrent: 3 },
    writer: { agent: { name: 'writer' }, maxConcurrent: 1 },
    reviewer: { agent: { name: 'reviewer' }, maxConcurrent: 1 },
  },
  patterns: {
    research: parallel(
      ['researcher', 'researcher'],
      (results) => concatResults(results)
    ),
    pipeline: sequential(['researcher', 'writer', 'reviewer']),
  },
});

// Test a parallel pattern
const research = await orchestrator.runPattern('research', 'Explain WASM');
expect(research).toContain('Research findings');

// Test a sequential pipeline
const article = await orchestrator.runPattern('pipeline', 'Write about WASM');

// Inspect runner call history
expect(runner.getCallCount('researcher')).toBe(3);  // 2 parallel + 1 sequential
expect(runner.getCallCount('writer')).toBe(1);
expect(runner.calls.length).toBe(5);

// Reset between tests
runner.reset();
orchestrator.reset();

const flakyRunner = createMockAgentRunner({
  defaultResponse: 'OK',
  shouldFail: (agent) => agent.name === 'researcher',
  failureError: new Error('Rate limited'),
});

const orchestrator = createMultiAgentOrchestrator({
  runner: flakyRunner,
  agents: {
    researcher: { agent: { name: 'researcher' }, maxConcurrent: 1 },
    writer: { agent: { name: 'writer' }, maxConcurrent: 1 },
  },
  patterns: {
    tolerant: parallel(
      ['researcher', 'writer'],
      (results) => concatResults(results),
      { minSuccess: 1 }
    ),
  },
});

// Researcher fails but writer succeeds – pattern completes
const result = await orchestrator.runPattern('tolerant', 'test');
expect(result).toBe('OK');

import { createAgentStack, parallel, sequential } from '@directive-run/ai';

const stack = createAgentStack({
  runner,
  agents: {
    researcher: { agent: researcher, maxConcurrent: 3 },
    writer: { agent: writer, maxConcurrent: 1 },
    reviewer: { agent: reviewer, maxConcurrent: 1 },
  },
  patterns: {
    research: parallel(
      ['researcher', 'researcher'],
      (results) => concatResults(results)
    ),
    pipeline: sequential(['researcher', 'writer', 'reviewer']),
  },
  // Stack-level features apply to all agent runs
  memory: { strategy: 'sliding', maxMessages: 50 },
  circuitBreaker: { maxFailures: 5, resetMs: 60000 },
  guardrails: {
    input: [createPIIGuardrail({ redact: true })],
    output: [createModerationGuardrail({ checkFn: moderate })],
  },
  messageBus: { maxHistory: 500 },
});

// Run patterns through the stack
const research = await stack.runPattern('research', 'Explain WASM');

// Access the underlying multi-agent orchestrator
const coordinator = stack.coordinator;
const state = coordinator.getAgentState('researcher');

// Access the message bus
const bus = stack.messageBus;

import { useAgentOrchestrator, useFact, useSelector } from '@directive-run/react';

function MultiAgentPanel() {
  const orchestrator = useAgentOrchestrator({ runner, autoApproveToolCalls: true });
  const { system } = orchestrator;

  const agent = useFact(system, '__agent');
  const summary = useSelector(system, (state) => ({
    status: state.__agent?.status,
    tokens: state.__agent?.totalTokens,
  }));

  return (
    <div>
      <p>Status: {agent?.status}</p>
      <p>Tokens: {summary.tokens}</p>
    </div>
  );
}

<script setup>
import { createMultiAgentOrchestrator } from '@directive-run/ai';
import { useFact, useInspect } from '@directive-run/vue';
import { onUnmounted } from 'vue';

const orchestrator = createMultiAgentOrchestrator({ runner, agents: { /* ... */ } });
onUnmounted(() => orchestrator.dispose());

const agent = useFact(orchestrator.system, '__agent');
const { isSettled } = useInspect(orchestrator.system);
</script>

<template>
  <p>Status: {{ agent?.status }}</p>
  <p>{{ isSettled ? 'Idle' : 'Working...' }}</p>
</template>

<script>
import { createMultiAgentOrchestrator } from '@directive-run/ai';
import { useFact, useInspect } from '@directive-run/svelte';
import { onDestroy } from 'svelte';

const orchestrator = createMultiAgentOrchestrator({ runner, agents: { /* ... */ } });
onDestroy(() => orchestrator.dispose());

const agent = useFact(orchestrator.system, '__agent');
const inspect = useInspect(orchestrator.system);
</script>

<p>Status: {$agent?.status}</p>
<p>{$inspect.isSettled ? 'Idle' : 'Working...'}</p>

import { createMultiAgentOrchestrator } from '@directive-run/ai';
import { useFact, useInspect } from '@directive-run/solid';
import { onCleanup } from 'solid-js';

function MultiAgentPanel() {
  const orchestrator = createMultiAgentOrchestrator({ runner, agents: { /* ... */ } });
  onCleanup(() => orchestrator.dispose());

  const agent = useFact(orchestrator.system, '__agent');
  const inspect = useInspect(orchestrator.system);

  return (
    <div>
      <p>Status: {agent()?.status}</p>
      <p>{inspect().isSettled ? 'Idle' : 'Working...'}</p>
    </div>
  );
}

import { LitElement, html } from 'lit';
import { createMultiAgentOrchestrator } from '@directive-run/ai';
import { FactController, InspectController } from '@directive-run/lit';

class MultiAgentPanel extends LitElement {
  private orchestrator = createMultiAgentOrchestrator({ runner, agents: { /* ... */ } });
  private agent = new FactController(this, this.orchestrator.system, '__agent');
  private inspect = new InspectController(this, this.orchestrator.system);

  disconnectedCallback() {
    super.disconnectedCallback();
    this.orchestrator.dispose();
  }

  render() {
    return html`
      <p>Status: ${this.agent.value?.status}</p>
      <p>${this.inspect.value?.isSettled ? 'Idle' : 'Working...'}</p>
    `;
  }
}