The Pipeline Pattern: Parallel Processing with Multiple Agents

import Anthropic from '@anthropic-ai/sdk';
import * as dotenv from 'dotenv';
dotenv.config();

const client = new Anthropic();

async function analyzeItem(item, systemPrompt) {
  const response = await client.messages.create({
    model: 'claude-haiku-3-5', // Use Haiku for high-volume parallel tasks
    max_tokens: 512,
    system: systemPrompt,
    messages: [{ role: 'user', content: item }]
  });
  return response.content[0].text;
}

// Process items in parallel with concurrency limit
async function parallelProcess(items, systemPrompt, concurrency = 5) {
  const results = new Array(items.length);
  
  // Process in batches of `concurrency`
  for (let i = 0; i < items.length; i += concurrency) {
    const batch = items.slice(i, i + concurrency);
    const batchPromises = batch.map((item, idx) => 
      analyzeItem(item, systemPrompt).then(result => ({
        index: i + idx,
        result
      }))
    );
    
    const batchResults = await Promise.all(batchPromises);
    batchResults.forEach(({ index, result }) => {
      results[index] = result;
    });
    
    console.log(`Processed ${Math.min(i + concurrency, items.length)}/${items.length}`);
  }
  
  return results;
}

// Example: sentiment analysis of customer reviews
const reviews = [
  "Amazing product, exceeded my expectations!",
  "Terrible quality, broke after one week.",
  "Decent for the price, nothing special.",
  "Best purchase I've made this year.",
  "Would not recommend to anyone.",
  "Solid product, does what it says.",
];

const SENTIMENT_AGENT = 'Classify this text. Respond with JSON only: {"sentiment": "positive|negative|neutral", "confidence": 0.0-1.0, "key_phrase": "..."}'

console.time('parallel');
const results = await parallelProcess(reviews, SENTIMENT_AGENT, 3);
console.timeEnd('parallel');

reviews.forEach((review, i) => {
  const parsed = JSON.parse(results[i]);
  console.log(`"${review.slice(0, 40)}..." → ${parsed.sentiment} (${parsed.confidence})`);
});