~/himanshu
$whoami
Back to blog

Part 4: Architecting Agent Teams - Hierarchical Workflows and Graph Composition

Building AI systems from hierarchical agent teams and graph composition, modeled on how the brain organizes itself.

March 08, 2024

Part 4: Architecting Agent Teams - Hierarchical Workflows and Graph Composition

Building capable AI systems isn't about crafting one brilliant agent. It's about getting teams of specialized agents to collaborate, delegate, and combine what they find. The human brain is the obvious analogy: a network of specialized regions coordinating toward complex goals. My long-term goal, replicating that through a Mixture of Experts (MoE) architecture, needs a solid approach to agent orchestration.

Why flat chains aren't enough

Simple agent "chains" or linear workflows hit their ceiling fast on complex, real-world problems. Imagine writing a research paper with one person handling everything from literature review to data analysis, coding, writing, and editing. It's inefficient, error-prone, and rarely produces the best result. Humans, by contrast, are good at breaking problems down, handing sub-tasks to specialists, and then integrating what comes back.

That hierarchical model is what advanced agents need too. You want a "manager" agent that can decompose a complex task into sub-tasks, route each one to the right "worker" agent or specialized workflow, orchestrate their execution while managing state and communication, and synthesize the workers' outputs into a coherent result.

Many existing "agent frameworks" try this and fall short. They pile on abstraction that hides the underlying logic, which leads to performance bottlenecks and opaque state. As a competitive programmer I value performance and clarity above all else, and bloated frameworks that dictate too much of how I build (looking at you, LangChain and its ilk) are usually counterproductive. I'd rather have raw API access and direct control, especially around critical state transitions. So we'll design our own composable graph execution system.

This isn't only a design pattern; it edges toward the brain's modularity. Different cortical regions specialize in vision, language, motor control, or abstract reasoning, yet they activate and talk to each other through neural pathways to hit larger goals. Agent teams should work the same way: specialized but integrated.

Architecture: hierarchy with custom graphs

The core of the system is composing and invoking specialized workflow graphs at runtime. Instead of leaning on heavy framework abstractions, here's a direct approach.

The communication bus: WorkflowState

At the heart of any multi-agent system is shared state. WorkflowState is our communication bus, carrying context, inputs, and outputs across every level of the hierarchy. It has to be extensible and precisely typed.

typescript
// src/core/workflowState.ts
 
export type AgentStatus = 'PENDING' | 'RUNNING' | 'COMPLETED' | 'FAILED';
 
export interface WorkflowState {
    // Global context for the entire workflow
    overallTask: string;
    overallResult: string | null;
    overallStatus: AgentStatus;
 
    // Current sub-task being processed by the Manager or a Worker
    currentSubTask: {
        id: string; // Unique ID for the sub-task
        description: string;
        assignedWorkerGraphId: string | null; // Which worker graph is responsible
        input: Record<string, any> | null; // Input payload for the worker
        output: Record<string, any> | null; // Output payload from the worker
        status: AgentStatus;
        errorMessage: string | null;
        startTime: number;
        endTime: number | null;
    } | null;
 
    // History of completed sub-tasks for traceability and error recovery
    subTaskHistory: Array<Omit<NonNullable<WorkflowState['currentSubTask']>, 'input'>>; // Exclude large inputs from history
 
    // Any global metadata or scratchpad space
    metadata: Record<string, any>;
    
    // User-specific information or persistent data
    contextData: Record<string, any>;
}
 
// Initializer for a new workflow state
export function createInitialWorkflowState(overallTask: string, contextData: Record<string, any> = {}): WorkflowState {
    return {
        overallTask,
        overallResult: null,
        overallStatus: 'PENDING',
        currentSubTask: null,
        subTaskHistory: [],
        metadata: {},
        contextData,
    };
}

WorkflowState is the explicit contract between agents, so data passes consistently and predictably. The currentSubTask field matters most: it's a temporary "scratchpad" where the manager prepares inputs for a worker and the worker writes its results.

The GraphEngine: our custom orchestrator

Rather than fight a framework, we define a small GraphEngine that executes a directed acyclic graph (DAG) of nodes. Each node is an atomic operation: an LLM call, a tool execution, a decision, or another graph invocation.

typescript
// src/core/graphEngine.ts
 
import { WorkflowState } from './workflowState';
 
// A NodeFunction takes the current state and returns a Promise resolving to the updated state.
export type NodeFunction = (state: WorkflowState) => Promise<WorkflowState>;
 
// A WorkflowGraph defines the nodes and their transitions.
export interface WorkflowGraph {
    id: string; // Unique ID for the graph
    nodes: Map<string, NodeFunction>; // Map of nodeName -> function
    edges: Map<string, string | string[]>; // Map of nodeName -> nextNodeName(s) or decision logic
    // Add more complex edge logic if needed (e.g., conditional transitions)
    startNode: string;
}
 
// Global registry for all defined workflow graphs
export const graphRegistry = new Map<string, WorkflowGraph>();
 
// The core engine to execute a workflow graph
export async function runWorkflowGraph(graphId: string, initialState: WorkflowState): Promise<WorkflowState> {
    const graph = graphRegistry.get(graphId);
    if (!graph) {
        throw new Error(`WorkflowGraph with ID '${graphId}' not found.`);
    }
 
    let currentState = { ...initialState }; // Ensure immutability for nodes
    let currentNodeName = graph.startNode;
 
    try {
        while (currentNodeName) {
            const nodeFn = graph.nodes.get(currentNodeName);
            if (!nodeFn) {
                throw new Error(`Node '${currentNodeName}' not found in graph '${graphId}'.`);
            }
            
            // Mark sub-task as running if applicable
            if (currentState.currentSubTask && currentState.currentSubTask.assignedWorkerGraphId === graphId && currentState.currentSubTask.status === 'PENDING') {
                currentState.currentSubTask.status = 'RUNNING';
                currentState.currentSubTask.startTime = Date.now();
            }
 
            console.log(`[${graphId}] Executing node: ${currentNodeName}`);
            currentState = await nodeFn(currentState); // Execute node, update state
 
            const nextEdge = graph.edges.get(currentNodeName);
            if (!nextEdge) {
                // End of graph
                currentNodeName = '';
            } else if (typeof nextEdge === 'string') {
                currentNodeName = nextEdge;
            } else if (Array.isArray(nextEdge)) {
                // Example: simple sequential multiple next nodes, or complex conditional routing
                // For now, let's assume a simple sequential for simplicity or decision-based.
                // A real implementation would have more sophisticated decision nodes.
                throw new Error(`Complex edge logic for node '${currentNodeName}' not yet implemented.`);
            }
            // A more advanced engine would use a router node to pick from `string[]` or based on `currentState`
        }
        
        // Update overall status if this was the top-level graph
        if (graphId === currentState.currentSubTask?.assignedWorkerGraphId) { // Check if this was a root worker
            currentState.currentSubTask.status = 'COMPLETED';
            currentState.currentSubTask.endTime = Date.now();
            currentState.subTaskHistory.push({
                ...currentState.currentSubTask,
                input: undefined // Clear input before pushing to history to save memory
            });
            currentState.currentSubTask = null; // Clear current sub-task after completion
        }
 
        return currentState;
 
    } catch (error: any) {
        console.error(`[${graphId}] Error during graph execution at node ${currentNodeName}:`, error);
        if (currentState.currentSubTask && currentState.currentSubTask.assignedWorkerGraphId === graphId) {
            currentState.currentSubTask.status = 'FAILED';
            currentState.currentSubTask.errorMessage = error.message;
            currentState.currentSubTask.endTime = Date.now();
            currentState.subTaskHistory.push({
                ...currentState.currentSubTask,
                input: undefined
            });
            currentState.currentSubTask = null;
        }
        currentState.overallStatus = 'FAILED';
        currentState.overallResult = `Workflow failed: ${error.message}`;
        return currentState;
    }
}

The GraphEngine is small. It executes nodes and passes WorkflowState around. The key to hierarchy comes next: one node can invoke another WorkflowGraph.

The manager graph (MainOrchestratorGraph)

The manager's job is strategic: break overallTask into smaller independent sub-tasks, pick the right WorkerGraph for each, invoke it through the GraphEngine, and combine the results.

typescript
// src/graphs/mainOrchestratorGraph.ts
 
import { WorkflowState, AgentStatus } from '../core/workflowState';
import { WorkflowGraph, NodeFunction, runWorkflowGraph, graphRegistry } from '../core/graphEngine';
import { generateUniqueId } from '../utils/idGenerator'; // Simple ID generator utility
import { llmCall } from '../utils/llmApi'; // Mock LLM API
 
// --- Manager Nodes ---
 
const decomposeTaskNode: NodeFunction = async (state) => {
    // Use LLM to break down the overall task
    const prompt = `Given the overall task: "${state.overallTask}", identify the essential sub-tasks required to complete it.
    Output a JSON array of objects, where each object has 'id', 'description', and 'requiredWorkerGraph' (e.g., 'CodeGenerator', 'DataAnalyzer', 'ReportWriter').
    Example:
    [
      { "id": "task_1", "description": "Research market trends", "requiredWorkerGraph": "DataAnalyzer" },
      { "id": "task_2", "description": "Generate Python script for analysis", "requiredWorkerGraph": "CodeGenerator" }
    ]`;
 
    const response = await llmCall(prompt, state.contextData);
    const subTasks = JSON.parse(response); // Assume LLM provides valid JSON
 
    // Store decomposed tasks in metadata for sequential processing
    state.metadata.pendingSubTasks = subTasks;
    state.overallStatus = 'RUNNING';
    return state;
};
 
const expertRouterNode: NodeFunction = async (state) => {
    if (!state.metadata.pendingSubTasks || state.metadata.pendingSubTasks.length === 0) {
        // All sub-tasks processed, move to synthesis
        return { ...state, metadata: { ...state.metadata, nextManagerAction: 'synthesize' } };
    }
 
    const nextTask = state.metadata.pendingSubTasks.shift(); // Get next task
    if (!nextTask) throw new Error("No pending sub-tasks, but router node was called.");
 
    // Prepare current sub-task for worker invocation
    state.currentSubTask = {
        id: nextTask.id,
        description: nextTask.description,
        assignedWorkerGraphId: nextTask.requiredWorkerGraph,
        input: { taskDescription: nextTask.description, context: state.contextData }, // Input for worker
        output: null,
        status: 'PENDING',
        errorMessage: null,
        startTime: 0, // Will be set by worker graph
        endTime: null,
    };
 
    return { ...state, metadata: { ...state.metadata, nextManagerAction: 'invokeWorker' } };
};
 
const invokeWorkerGraphNode: NodeFunction = async (state) => {
    if (!state.currentSubTask || !state.currentSubTask.assignedWorkerGraphId) {
        throw new Error("Attempted to invoke worker without a defined currentSubTask or assignedWorkerGraphId.");
    }
 
    const workerGraphId = state.currentSubTask.assignedWorkerGraphId;
    console.log(`Manager invoking worker graph: ${workerGraphId} for sub-task: ${state.currentSubTask.id}`);
 
    // Create a new state instance for the worker to operate on, containing only relevant input
    let workerState: WorkflowState = {
        overallTask: state.currentSubTask.description, // Worker sees its specific task as 'overall'
        overallResult: null,
        overallStatus: 'PENDING',
        currentSubTask: { // This *is* the sub-task for the manager, but the worker treats it as its root
            id: generateUniqueId(), // Worker gets its own root sub-task ID
            description: state.currentSubTask.description,
            assignedWorkerGraphId: workerGraphId,
            input: state.currentSubTask.input,
            output: null,
            status: 'PENDING',
            errorMessage: null,
            startTime: Date.now(),
            endTime: null,
        },
        subTaskHistory: [], // Worker starts with fresh history
        metadata: {},
        contextData: state.contextData, // Pass down relevant context
    };
 
    // --- CRITICAL: Execute the worker graph as a sub-process ---
    workerState = await runWorkflowGraph(workerGraphId, workerState);
 
    // After worker completes, update manager's state
    if (!state.currentSubTask) throw new Error("Current sub-task disappeared after worker invocation.");
    state.currentSubTask.output = workerState.currentSubTask?.output || null;
    state.currentSubTask.status = workerState.currentSubTask?.status || 'FAILED';
    state.currentSubTask.errorMessage = workerState.currentSubTask?.errorMessage || null;
    state.currentSubTask.endTime = workerState.currentSubTask?.endTime || Date.now();
    
    // Add completed sub-task to manager's history
    state.subTaskHistory.push({
        ...state.currentSubTask,
        input: undefined // Clear input before pushing to history
    });
 
    // Clear currentSubTask for the manager to pick the next one
    state.currentSubTask = null;
    return state;
};
 
const synthesizeResultNode: NodeFunction = async (state) => {
    // Collect all outputs from sub-task history
    const workerOutputs = state.subTaskHistory.map(task => ({
        id: task.id,
        description: task.description,
        output: task.output,
        status: task.status
    }));
 
    const prompt = `Synthesize the following sub-task results into a comprehensive final answer for the overall task: "${state.overallTask}"
    Sub-task results: ${JSON.stringify(workerOutputs, null, 2)}
    Provide a concise, professional summary.`;
 
    state.overallResult = await llmCall(prompt, state.contextData);
    state.overallStatus = 'COMPLETED';
    return state;
};
 
// Define the MainOrchestratorGraph
export const MainOrchestratorGraph: WorkflowGraph = {
    id: 'MainOrchestrator',
    startNode: 'decomposeTask',
    nodes: new Map([
        ['decomposeTask', decomposeTaskNode],
        ['expertRouter', expertRouterNode],
        ['invokeWorkerGraph', invokeWorkerGraphNode],
        ['synthesizeResult', synthesizeResultNode],
    ]),
    edges: new Map([
        ['decomposeTask', 'expertRouter'],
        // Dynamic routing based on 'nextManagerAction' in metadata
        ['expertRouter', (state: WorkflowState) => state.metadata.nextManagerAction === 'invokeWorker' ? 'invokeWorkerGraph' : 'synthesizeResult'],
        ['invokeWorkerGraph', 'expertRouter'], // After a worker finishes, go back to router for next task
        ['synthesizeResult', null], // End of graph
    ]) as Map<string, string | ((state: WorkflowState) => string | null)>, // Type assertion for dynamic edges
};
 
graphRegistry.set(MainOrchestratorGraph.id, MainOrchestratorGraph);

A note on dynamic edges: the GraphEngine edges definition (Map<string, string | string[]>) is basic. For the manager I've shown a more advanced edges map with a function that simulates conditional routing. A real GraphEngine would support this directly through a dedicated RouterNode or ConditionalEdge type.

Worker graphs (CodeGenerationGraph, DataAnalysisGraph, etc.)

Worker graphs are focused WorkflowGraph instances. They receive their task and input through WorkflowState.currentSubTask.input when the manager invokes them, and write results back to WorkflowState.currentSubTask.output.

typescript
// src/graphs/codeGenerationGraph.ts
 
import { WorkflowState } from '../core/workflowState';
import { WorkflowGraph, NodeFunction, graphRegistry } from '../core/graphEngine';
import { llmCall } from '../utils/llmApi'; // Mock LLM API
import { runPythonCode } from '../utils/codeExecutor'; // Mock code executor
 
// --- Code Generation Worker Nodes ---
 
const planCodeNode: NodeFunction = async (state) => {
    if (!state.currentSubTask || !state.currentSubTask.input) {
        throw new Error("Code generation worker requires sub-task input.");
    }
    const taskDescription = state.currentSubTask.input.taskDescription;
    const prompt = `Given the task: "${taskDescription}", outline a Python script plan (steps, libraries, expected output).`;
    const plan = await llmCall(prompt, state.contextData);
    state.metadata.codePlan = plan;
    return state;
};
 
const generateCodeNode: NodeFunction = async (state) => {
    if (!state.currentSubTask || !state.metadata.codePlan) {
        throw new Error("Code generation worker needs a plan.");
    }
    const taskDescription = state.currentSubTask.input.taskDescription;
    const plan = state.metadata.codePlan;
    const prompt = `Based on this plan: "${plan}" and the task: "${taskDescription}", generate the full Python code.`;
    const code = await llmCall(prompt, state.contextData);
    state.metadata.generatedCode = code;
    return state;
};
 
const executeCodeNode: NodeFunction = async (state) => {
    if (!state.currentSubTask || !state.metadata.generatedCode) {
        throw new Error("Code execution worker needs generated code.");
    }
    const code = state.metadata.generatedCode;
    try {
        const executionResult = await runPythonCode(code, state.currentSubTask.input.context);
        state.currentSubTask.output = {
            code: code,
            result: executionResult,
            success: true,
        };
    } catch (error: any) {
        state.currentSubTask.output = {
            code: code,
            result: null,
            success: false,
            error: error.message,
        };
        throw error; // Propagate error for manager to handle
    }
    return state;
};
 
export const CodeGenerationGraph: WorkflowGraph = {
    id: 'CodeGenerator',
    startNode: 'planCode',
    nodes: new Map([
        ['planCode', planCodeNode],
        ['generateCode', generateCodeNode],
        ['executeCode', executeCodeNode],
    ]),
    edges: new Map([
        ['planCode', 'generateCode'],
        ['generateCode', 'executeCode'],
        ['executeCode', null], // End of graph
    ]),
};
 
graphRegistry.set(CodeGenerationGraph.id, CodeGenerationGraph);
 
// --- Dummy Data Analysis Worker (for illustration) ---
 
const analyzeDataNode: NodeFunction = async (state) => {
    if (!state.currentSubTask || !state.currentSubTask.input) {
        throw new Error("Data analysis worker requires sub-task input.");
    }
    const taskDescription = state.currentSubTask.input.taskDescription;
    console.log(`Simulating data analysis for: ${taskDescription}`);
    // Simulate some work
    await new Promise(resolve => setTimeout(resolve, 1500));
    state.currentSubTask.output = {
        summary: `Analysis complete for "${taskDescription}". Key finding: Data shows a 15% increase in XYZ.`,
        rawOutput: { /* large data payload */ },
    };
    return state;
};
 
export const DataAnalysisGraph: WorkflowGraph = {
    id: 'DataAnalyzer',
    startNode: 'analyzeData',
    nodes: new Map([
        ['analyzeData', analyzeDataNode],
    ]),
    edges: new Map([
        ['analyzeData', null],
    ]),
};
 
graphRegistry.set(DataAnalysisGraph.id, DataAnalysisGraph);

Putting It All Together: The Main Execution Flow

typescript
// src/main.ts
import { createInitialWorkflowState } from './core/workflowState';
import { runWorkflowGraph, graphRegistry } from './core/graphEngine';
import { MainOrchestratorGraph } from './graphs/mainOrchestratorGraph'; // This registers itself
import { CodeGenerationGraph } from './graphs/codeGenerationGraph'; // This registers itself
import { DataAnalysisGraph } from './graphs/dataAnalysisGraph'; // This registers itself
 
async function main() {
    console.log("Starting hierarchical agent workflow...");
 
    const initialTask = "Analyze sales data for Q3 2023, generate a report, and identify top-performing products. Also, write a Python script to automate weekly sales report generation.";
    const userContext = { user: "Alice", preferences: { format: "markdown" } };
 
    let state = createInitialWorkflowState(initialTask, userContext);
 
    try {
        state = await runWorkflowGraph(MainOrchestratorGraph.id, state);
        console.log("\n--- Workflow Completed ---");
        console.log("Overall Status:", state.overallStatus);
        console.log("Overall Result:", state.overallResult);
        console.log("Sub-task History:");
        state.subTaskHistory.forEach(task => {
            console.log(`- [${task.status}] ${task.description} (${task.assignedWorkerGraphId})`);
            console.log(`  Output: ${JSON.stringify(task.output, null, 2).slice(0, 200)}...`); // Truncate for display
        });
    } catch (error) {
        console.error("\n--- Workflow Failed ---");
        console.error("Final State:", state);
        console.error("Error:", error);
    }
}
 
main();
 
// Mock LLM and Code Executor (for demonstration)
// In a real system, these would be proper API calls.
export const llmCall = async (prompt: string, context: Record<string, any>): Promise<string> => {
    console.log(`\n--- LLM Call ---`);
    console.log(`Prompt: ${prompt.slice(0, 300)}...`);
    // Simulate LLM processing time
    await new Promise(resolve => setTimeout(resolve, 500));
    if (prompt.includes("identify the essential sub-tasks")) {
        return JSON.stringify([
            { "id": "sub_1", "description": "Analyze Q3 2023 sales data to identify top products", "requiredWorkerGraph": "DataAnalyzer" },
            { "id": "sub_2", "description": "Generate Python script for weekly sales report automation", "requiredWorkerGraph": "CodeGenerator" },
            { "id": "sub_3", "description": "Write a summary report based on analysis and script", "requiredWorkerGraph": "ReportWriter" } // Assuming another worker
        ]);
    } else if (prompt.includes("outline a Python script plan")) {
        return "Plan: 1. Load sales data. 2. Aggregate by product. 3. Identify top N. 4. Format report. Libraries: pandas, openpyxl.";
    } else if (prompt.includes("generate the full Python code")) {
        return "import pandas as pd\\ndef generate_sales_report(data_path):\\n  df = pd.read_excel(data_path)\\n  # ... (rest of the code)\\n  return 'Report Generated'";
    } else if (prompt.includes("Synthesize the following sub-task results")) {
        return `Comprehensive report for Q3 2023:\n1. Top-performing products identified through data analysis.\n2. Automated script for future reports successfully generated and ready for deployment.`;
    }
    return "LLM generated a generic response.";
};
 
export const runPythonCode = async (code: string, context: Record<string, any>): Promise<string> => {
    console.log(`\n--- Code Execution ---`);
    console.log(`Executing code: ${code.slice(0, 100)}...`);
    await new Promise(resolve => setTimeout(resolve, 1000));
    return "Code executed successfully. Report data generated.";
};
 
export const generateUniqueId = (): string => `id_${Date.now()}_${Math.random().toFixed(5).replace('0.', '')}`;

(Note: The ReportWriter graph is referenced but not fully implemented to keep the example concise. The graphRegistry ensures all workers are known.)

What I learned

Building this reinforced a few principles.

Modularity pays off. Breaking a problem into smaller, specialized sub-graphs makes it easier to maintain, debug, and scale, and each worker graph is a self-contained expert. That mirrors the brain's cortical regions, each handling one kind of information (visual, auditory, motor) while working in concert.

State is the contract. A carefully designed WorkflowState isn't just a data structure, it's the API between every node and every graph. It sets how information flows and catches silent failures, and TypeScript's explicit typing saved me from plenty of runtime errors.

Direct control buys performance. Building my own GraphEngine instead of leaning on a heavy framework meant full control over execution flow and state, which cut overhead and, just as important, let me understand what the system actually does. In research, where you iterate constantly, every millisecond counts.

And the MoE connection gets stronger. This isn't only agent teams, it's a working blueprint for a Mixture of Experts model. The MainOrchestratorGraph is the router, or gating network, deciding which "expert" worker graph fits a given sub-task, and InvokeSubGraphNode is how those experts get activated. That's how I picture brain-like systems: a high-level manager orchestrating a large set of specialized, efficient expert modules.

A few things fought back. Early versions of WorkflowState were too simplistic and leaked data or left it ambiguous; getting to a clean state that held both global context and granular sub-task detail took a couple of passes. The dynamic routing (the conditional logic in expertRouterNode and the edges definition for the MainOrchestratorGraph) needed care to stay flexible without going opaque. And error propagation, catching a failure deep in a sub-graph and surfacing it back up to the manager and into overallStatus, is the part you can't skip.

Plenty left to build. I want to replace the if/else in expertRouterNode with an LLM-driven decision that picks worker graphs based on the sub-task's nuances and past performance, which is basically learning-based routing from MoE. Sub-tasks run sequentially today; running independent sub-graphs in parallel would mirror the brain's parallelism and speed things up. I'd add self-healing nodes that detect worker failures and recover, retrying a sub-task, escalating to a human, or generating an alternative plan. And worker graphs need real tool integration for external APIs, databases, and compute.

This approach, built on explicit state and direct graph orchestration, is a good way to engineer complex agent systems. It's a real step toward building AI that mirrors the modular, efficient architecture of the brain.