🎼 Orchestration Example - Multi-Agent Workflow System

@OrchestratorAgent
public class WorkflowOrchestrator extends Agent {
    
    @Inject
    LLMService llmService;
    
    @Inject
    KafkaProducer<String, String> producer;
    
    @Override
    public void initialize(AgentContext context) {
        context.subscribe("orchestration.requests", this::handleRequest);
    }
    
    private void handleRequest(Message message) {
        String request = message.getPayload();
        
        // Use LLM to decompose task
        String decomposition = llmService.chat(
            "Break down this request into subtasks: " + request,
            new ChatConfig()
                .model("gpt-4")
                .temperature(0.3)
        );
        
        // Parse decomposition
        List<Task> tasks = parseDecomposition(decomposition);
        
        // Delegate tasks
        delegateTasks(tasks);
    }
    
    private List<Task> parseDecomposition(String decomposition) {
        List<Task> tasks = new ArrayList<>();
        
        // Example decomposition:
        // 1. Get weather for Paris
        // 2. Get stock prices for AAPL, MSFT
        // 3. Find travel deals to Paris
        // 4. Analyze sentiment of travel reviews
        
        String[] lines = decomposition.split("\n");
        for (String line : lines) {
            if (line.trim().isEmpty()) continue;
            
            Task task = new Task();
            task.setDescription(line);
            
            // Determine task type
            if (line.contains("weather")) {
                task.setType(TaskType.WEATHER);
                task.setAgent("weather-agent");
            } else if (line.contains("stock")) {
                task.setType(TaskType.STOCK);
                task.setAgent("stock-agent");
            } else if (line.contains("travel")) {
                task.setType(TaskType.TRAVEL);
                task.setAgent("travel-agent");
            } else if (line.contains("sentiment")) {
                task.setType(TaskType.SENTIMENT);
                task.setAgent("sentiment-agent");
            }
            
            tasks.add(task);
        }
        
        return tasks;
    }
    
    private void delegateTasks(List<Task> tasks) {
        // Delegate tasks to specialized agents
        for (Task task : tasks) {
            producer.send("orchestration.tasks", serialize(task));
        }
    }
}

@Path("/api/orchestration")
public class OrchestrationResource {
    
    @Inject
    WorkflowOrchestrator orchestrator;
    
    @POST
    @Path("/execute")
    public Response executeWorkflow(WorkflowRequest request) {
        // Decompose workflow
        List<Task> tasks = orchestrator.decomposeTasks(request);
        
        // Execute tasks in parallel
        List<CompletableFuture<TaskResult>> futures = tasks.stream()
            .map(task -> executeTaskAsync(task))
            .collect(toList());
        
        // Wait for all tasks
        List<TaskResult> results = futures.stream()
            .map(CompletableFuture::join)
            .collect(toList());
        
        // Aggregate results
        WorkflowResult aggregated = orchestrator.aggregateResults(results);
        
        return Response.ok()
            .entity(aggregated)
            .build();
    }
    
    private CompletableFuture<TaskResult> executeTaskAsync(Task task) {
        return CompletableFuture.supplyAsync(() -> {
            // Execute task
            return executeTask(task);
        });
    }
    
    private TaskResult executeTask(Task task) {
        // Route to appropriate agent
        switch (task.getType()) {
            case WEATHER:
                return weatherAgent.execute(task);
            case STOCK:
                return stockAgent.execute(task);
            case TRAVEL:
                return travelAgent.execute(task);
            case SENTIMENT:
                return sentimentAgent.execute(task);
            default:
                throw new IllegalArgumentException("Unknown task type");
        }
    }
}

public WorkflowResult aggregateResults(List<TaskResult> results) {
    // Use LLM to synthesize results
    String resultSummary = llmService.chat(
        "Synthesize these results into a coherent response: " +
        results.stream()
            .map(TaskResult::getResult)
            .collect(joining("\n")),
        new ChatConfig()
            .model("gpt-4")
            .temperature(0.7)
    );
    
    WorkflowResult workflow = new WorkflowResult();
    workflow.setSummary(resultSummary);
    workflow.setDetails(results);
    workflow.setStatus("completed");
    workflow.setTimestamp(Instant.now());
    
    return workflow;
}

private TaskResult executeTaskWithRetry(Task task, int maxRetries) {
    for (int attempt = 1; attempt <= maxRetries; attempt++) {
        try {
            return executeTask(task);
        } catch (Exception e) {
            logger.warn("Task execution failed (attempt {}/{}): {}",
                attempt, maxRetries, e.getMessage());
            
            if (attempt == maxRetries) {
                // Final attempt failed
                TaskResult result = new TaskResult();
                result.setStatus("failed");
                result.setError(e.getMessage());
                result.setTask(task);
                
                // Publish error event
                producer.send("orchestration.errors", serialize(result));
                
                return result;
            }
            
            // Wait before retry
            try {
                Thread.sleep(1000 * attempt);  // Exponential backoff
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
    
    return null;
}