Back to blogs

Part 5: Production-Ready Agents - Implementing Human-in-the-Loop Supervision

April 26, 2024

Part 5: Production-Ready Agents - Implementing Human-in-the-Loop Supervision

In our journey to build sophisticated AI agents—particularly with a modular, MoE-inspired architecture—we often chase autonomy. We envision systems that think, reason, and act independently. Yet, the brutal truth of production systems, especially in critical domains, is that full autonomy is a liability. It’s a risk that no responsible engineer should tolerate without robust safeguards.

My passion for AI research, particularly in distributed intelligence akin to the human brain’s multi-expert system, drives me towards ever more capable agents. But even the human brain has inherent safety mechanisms, error correction, and, critically, conscious oversight. As AI researchers, we can learn from this. We need to bridge the gap between impressive academic prototypes and bulletproof, real-world deployments.

This post isn't about the next groundbreaking LLM fine-tune or a new agentic loop. It's about engineering reality. It's about designing systems that don't fail catastrophically when they encounter the unexpected, the ambiguous, or the downright dangerous. We're talking about Human-in-the-Loop (HITL) supervision, architected for performance and maintainability.

Why Human-in-the-Loop? Because Trust isn't Built on Blind Faith.

Imagine an agent tasked with financial transactions, medical diagnoses, or infrastructure management. A small error, an hallucination, or a misinterpretation can have catastrophic consequences. While MoE architectures offer resilience and specialized expertise, they aren't infallible. They operate within a model of the world, and that model can be incomplete or incorrect.

Full autonomy in these scenarios is not just risky; it's irresponsible. HITL isn't a sign of weakness in our AI; it's a testament to robust engineering. It acknowledges the current limitations of AI and proactively integrates human intelligence for:

  1. Validation: Confirming critical decisions before execution.
  2. Correction: Modifying agent outputs when they stray.
  3. Learning: Providing explicit feedback for future model improvements.
  4. Compliance: Meeting regulatory requirements that mandate human oversight.

Our goal: build an agent workflow that can be paused at designated critical junctures, hand off context to a human for review/modification, and then seamlessly resume execution, potentially incorporating that human feedback.

Architecting Robust Interruption: Beyond Bloated Frameworks

Many frameworks attempt to abstract away state management and workflow orchestration. While they offer quick starts, they often introduce significant overhead, opinionated structures, and limit the granular control vital for high-performance, production-ready systems. We value raw APIs, direct control, and zero-overhead abstractions.

For this kind of stateful, interruptible workflow, we're not reaching for an existing "agent framework." We're building a lightweight, TypeScript-first orchestrator. Why? Because we need:

  • Explicit State Management: No hidden global variables or magically passed contexts.
  • Performance: Minimize serialization/deserialization, function call overhead.
  • Flexibility: Define interruption points with precision, not just at predefined "tool calls."
  • Auditability: Every state transition and decision point is logged and traceable.

Our core idea revolves around a persistent WorkflowState and an AgentOrchestrator that manages transitions between AgentNode functions. Interruptions are simply a special state that requires an external trigger to resolve.

Core Components:

  1. WorkflowState: A structured object persisted to a durable store (e.g., PostgreSQL JSONB, Redis, DynamoDB). It contains:

    • workflowId: string
    • currentState: AgentState (e.g., INITIALIZING, PROCESSING_STEP_1, WAITING_FOR_HUMAN, COMPLETED, FAILED)
    • data: Record<string, any> (the mutable context passed between nodes)
    • lastExecutedNode: string | null
    • interruptionDetails?: { nodeId: string; reason: string; humanInputRequired: boolean }

  2. AgentNode: A simple asynchronous function that takes the current WorkflowState and returns an updated WorkflowState. It encapsulates a discrete piece of agent logic.

    type AgentNodeFn = (state: WorkflowState) => Promise<WorkflowState>;

  3. AgentOrchestrator: The central engine. It retrieves the current WorkflowState, determines the next node to execute based on a predefined graph/sequence, executes it, and persists the updated state. Its key methods will be execute and resume.

The Workflow Definition

Instead of a complex graph library, we'll define our workflow as a sequence of nodes, with explicit conditional logic for transitions and interruption points. This is effectively a state machine defined in code.

// types.ts
interface WorkflowState {
    workflowId: string;
    currentState: string; // e.g., 'start', 'analyze_request', 'propose_action', 'await_human_review', 'execute_action', 'complete'
    data: Record<string, any>; // Dynamic data for the workflow
    lastExecutedNode: string | null;
    interruptionDetails?: {
        nodeId: string;
        reason: string;
        humanInputRequired: boolean;
    };
}
 
// nodes.ts
const initialRequestNode: AgentNodeFn = async (state) => {
    console.log(`[${state.workflowId}] Initializing request...`);
    // Example: parse initial input, set up context
    state.data.initialQuery = state.data.inputQuery; // Assume inputQuery comes from initial trigger
    state.currentState = 'analyze_request';
    state.lastExecutedNode = 'initialRequestNode';
    return state;
};
 
const analyzeRequestNode: AgentNodeFn = async (state) => {
    console.log(`[${state.workflowId}] Analyzing request: ${state.data.initialQuery}`);
    // Use an LLM or specific expert to analyze
    // const analysisResult = await callLLMExpert(state.data.initialQuery);
    state.data.analysis = "LLM analysis results here..."; // Placeholder
    
    // Simple conditional for interruption AFTER this node
    if (state.data.analysis.includes("critical financial operation")) {
        console.log(`[${state.workflowId}] Critical operation detected. Requiring human review.`);
        state.currentState = 'await_human_review';
        state.interruptionDetails = {
            nodeId: 'analyzeRequestNode',
            reason: 'Critical operation identified, requires human approval',
            humanInputRequired: true
        };
    } else {
        state.currentState = 'propose_action';
    }
    state.lastExecutedNode = 'analyzeRequestNode';
    return state;
};
 
const proposeActionNode: AgentNodeFn = async (state) => {
    // This node could be interrupted BEFORE execution if human already intervened
    // (e.g., human explicitly marked 'no_action_needed' in previous step)
    if (state.interruptionDetails && state.interruptionDetails.humanInputRequired && !state.data.humanApproved) {
        // This is a "guard" for interrupt_before functionality
        // The orchestrator would typically handle this, but for clarity, showing it here.
        throw new Error("Workflow is paused, human input pending or disapproved.");
    }
    
    console.log(`[${state.workflowId}] Proposing action based on analysis: ${state.data.analysis}`);
    // const actionPlan = await generateActionPlan(state.data.analysis);
    state.data.actionPlan = "Generated action plan here..."; // Placeholder
    state.currentState = 'execute_action';
    state.lastExecutedNode = 'proposeActionNode';
    return state;
};
 
const executeActionNode: AgentNodeFn = async (state) => {
    if (!state.data.humanApproved) {
        // This could be another 'interrupt_before' scenario if the human explicitly rejected
        // and the orchestrator didn't update the state correctly.
        throw new Error("Action not approved by human. Cannot execute.");
    }
    
    console.log(`[${state.workflowId}] Executing action: ${state.data.actionPlan}`);
    // await runAction(state.data.actionPlan); // External API call
    state.data.executionResult = "Action completed successfully."; // Placeholder
    state.currentState = 'complete';
    state.lastExecutedNode = 'executeActionNode';
    return state;
};
 
const completionNode: AgentNodeFn = async (state) => {
    console.log(`[${state.workflowId}] Workflow completed.`);
    state.currentState = 'complete';
    state.lastExecutedNode = 'completionNode';
    return state;
};
 
// Map node names to functions
const nodes: Record<string, AgentNodeFn> = {
    initialRequestNode,
    analyzeRequestNode,
    proposeActionNode,
    executeActionNode,
    completionNode
};

The AgentOrchestrator Implementation

This is where the magic happens. The orchestrator needs to:

  1. Load the current state.
  2. Determine the next node based on currentState.
  3. Execute the node.
  4. Persist the new state.
  5. Handle interruptions by not moving to the next node if currentState signals a pause.
// orchestrator.ts
import { WorkflowState, AgentNodeFn } from './types'; // Assuming types defined above
 
interface WorkflowDefinition {
    [stateName: string]: {
        node: string; // Name of the node function
        nextState: string | ((state: WorkflowState) => string); // Next state or a function to determine it
    };
}
 
// A simple in-memory store for demonstration. In production, use Redis/DB.
const workflowStore = new Map<string, WorkflowState>();
 
class AgentOrchestrator {
    private workflowDefinition: WorkflowDefinition;
    private nodeFunctions: Record<string, AgentNodeFn>;
 
    constructor(workflowDefinition: WorkflowDefinition, nodeFunctions: Record<string, AgentNodeFn>) {
        this.workflowDefinition = workflowDefinition;
        this.nodeFunctions = nodeFunctions;
    }
 
    private async saveState(state: WorkflowState): Promise<void> {
        workflowStore.set(state.workflowId, { ...state }); // Deep copy to prevent mutation issues
        // console.log(`[${state.workflowId}] State saved: ${state.currentState}`);
    }
 
    private async loadState(workflowId: string): Promise<WorkflowState | undefined> {
        return workflowStore.get(workflowId);
    }
 
    public async initializeWorkflow(workflowId: string, initialData: Record<string, any>): Promise<WorkflowState> {
        const initialState: WorkflowState = {
            workflowId,
            currentState: 'start', // Entry point
            data: initialData,
            lastExecutedNode: null,
        };
        await this.saveState(initialState);
        return initialState;
    }
 
    public async execute(workflowId: string): Promise<WorkflowState> {
        let state = await this.loadState(workflowId);
        if (!state) {
            throw new Error(`Workflow ${workflowId} not found.`);
        }
 
        while (state.currentState !== 'complete' && state.currentState !== 'failed' && !state.interruptionDetails) {
            const currentStep = this.workflowDefinition[state.currentState];
            if (!currentStep) {
                throw new Error(`Invalid state in workflow definition: ${state.currentState}`);
            }
 
            const nodeFn = this.nodeFunctions[currentStep.node];
            if (!nodeFn) {
                throw new Error(`Node function '${currentStep.node}' not found.`);
            }
 
            try {
                // Execute the node
                state = await nodeFn(state);
                
                // If the node itself set an interruption, break the loop
                if (state.interruptionDetails) {
                    console.log(`[${state.workflowId}] Workflow paused for human review at node: ${state.lastExecutedNode}`);
                    await this.saveState(state);
                    return state;
                }
 
                // Determine next state
                if (typeof currentStep.nextState === 'function') {
                    state.currentState = currentStep.nextState(state);
                } else if (currentStep.nextState) {
                    state.currentState = currentStep.nextState;
                } else {
                    // Implicitly move to next state if not defined, or assume completion
                    // For a robust system, this should be explicitly handled in definition.
                    state.currentState = 'complete';
                }
 
                await this.saveState(state);
 
            } catch (error: any) {
                console.error(`[${workflowId}] Error executing node ${currentStep.node}:`, error.message);
                state.currentState = 'failed';
                state.data.error = error.message;
                await this.saveState(state);
                return state;
            }
        }
        console.log(`[${workflowId}] Workflow execution finished. Current state: ${state.currentState}`);
        return state;
    }
    
    // This is the external API endpoint handler for resuming a workflow
    public async resumeWithHumanInput(workflowId: string, humanInput: Record<string, any>): Promise<WorkflowState> {
        let state = await this.loadState(workflowId);
        if (!state) {
            throw new Error(`Workflow ${workflowId} not found.`);
        }
 
        if (!state.interruptionDetails) {
            throw new Error(`Workflow ${workflowId} is not currently interrupted.`);
        }
 
        // Apply human input to the workflow data
        state.data = { ...state.data, ...humanInput };
        
        // Clear interruption status
        state.interruptionDetails = undefined;
        
        // Decide the next step based on human input or resume from where it left off
        // For simplicity, we'll assume human input dictates approval and we resume from the next logical step
        // In a real system, humanInput might include a 'decision' field like 'approved'/'rejected'
        if (humanInput.decision === 'approved') {
            state.data.humanApproved = true; // Flag for subsequent nodes
            // Resume from the node that would have been next if no interruption
            const interruptedNodeConfig = this.workflowDefinition[state.currentState];
            if (typeof interruptedNodeConfig.nextState === 'function') {
                 state.currentState = interruptedNodeConfig.nextState(state);
            } else {
                 state.currentState = interruptedNodeConfig.nextState || 'complete'; // Or whatever makes sense
            }
            console.log(`[${workflowId}] Human approved. Resuming from state: ${state.currentState}`);
        } else {
            // Human rejected, potentially move to a 'rejected' or 'failed' state
            state.data.humanApproved = false;
            state.currentState = 'failed';
            state.data.error = 'Human rejected the proposed action.';
            console.log(`[${workflowId}] Human rejected. Workflow moved to state: ${state.currentState}`);
        }
 
        await this.saveState(state);
        return this.execute(workflowId); // Attempt to continue execution
    }
}
 
// Define the workflow's state transitions
const agentWorkflowDefinition: WorkflowDefinition = {
    'start': { node: 'initialRequestNode', nextState: 'analyze_request' },
    'analyze_request': {
        node: 'analyzeRequestNode',
        nextState: (state) => state.interruptionDetails ? 'await_human_review' : 'propose_action'
    },
    'await_human_review': { // This is a "waiting" state, orchestrator won't execute further
        node: 'analyzeRequestNode', // This node already ran, this just signifies we're paused AFTER it
        nextState: (state) => state.data.humanApproved ? 'propose_action' : 'failed' // After human input
    },
    'propose_action': { node: 'proposeActionNode', nextState: 'execute_action' },
    'execute_action': { node: 'executeActionNode', nextState: 'complete' },
    'complete': { node: 'completionNode', nextState: 'complete' } // Terminal state
};
 
// --- Example Usage ---
async function runExample() {
    const orchestrator = new AgentOrchestrator(agentWorkflowDefinition, nodes);
    const workflowId = `wf-${Date.now()}`;
    
    console.log("--- Starting Workflow 1 (requires human approval) ---");
    let state1 = await orchestrator.initializeWorkflow(workflowId, { inputQuery: "Perform critical financial operation: Transfer $100,000 to account X" });
    state1 = await orchestrator.execute(workflowId);
 
    console.log("\nCurrent state after initial run:", state1.currentState, state1.interruptionDetails);
 
    if (state1.currentState === 'await_human_review' && state1.interruptionDetails) {
        console.log("\n--- Simulating human review and approval ---");
        // An external API call would hit `resumeWithHumanInput`
        // For example: POST /api/workflow/{workflowId}/resume { decision: 'approved', comment: 'Looks good' }
        state1 = await orchestrator.resumeWithHumanInput(workflowId, { decision: 'approved', comment: 'Review passed.' });
        console.log("State after human approval and resumption attempt:", state1.currentState);
    }
    
    console.log("\n--- Starting Workflow 2 (no human approval needed) ---");
    const workflowId2 = `wf-${Date.now()}-no-interrupt`;
    let state2 = await orchestrator.initializeWorkflow(workflowId2, { inputQuery: "Generate a summary of recent tech news" });
    state2 = await orchestrator.execute(workflowId2);
    console.log("State after full execution (no interrupt):", state2.currentState);
}
 
runExample();

Explaining Interruption Mechanisms:

  • interrupt_after (our implementation): The analyzeRequestNode executes its logic. After its core function, it checks a condition (state.data.analysis.includes("critical financial operation")). If true, it modifies the WorkflowState to await_human_review and sets interruptionDetails. The orchestrator then sees interruptionDetails and stops processing further nodes, persisting this paused state. The next step is explicitly waiting for an external resumeWithHumanInput call.

  • interrupt_before (our implementation): While not explicitly shown as a standalone interrupt_before flag, the logic within proposeActionNode and executeActionNode serves this purpose. If the WorkflowState (specifically state.data.humanApproved) indicates a prior rejection or pending approval, these nodes can throw an error or simply return the state without performing their core logic. The orchestrator's resumeWithHumanInput function is responsible for ensuring the WorkflowState is correctly updated before the execute loop starts again, effectively guarding the next node's execution. A more explicit interrupt_before flag could be added to the WorkflowDefinition if needed, where the orchestrator itself checks it before calling nodeFn.

Optimization Notes:

  • State Persistence: For high throughput, use a dedicated key-value store (Redis) for fast state lookups, or optimize database queries for WorkflowState if using a relational database.
  • Asynchronous Execution: All AgentNodeFns are async. The orchestrator ensures non-blocking execution.
  • Idempotency: Node functions should ideally be idempotent. If an execution fails and retries, it shouldn't cause side effects.
  • Concurrency: A real production system would need a robust mechanism (e.g., transactional updates, optimistic locking) to prevent race conditions if multiple processes try to modify the same WorkflowState.
  • Observability: Detailed logging of state transitions, node execution times, and payload sizes is crucial for debugging and performance monitoring.

What I Learned

Building this kind of robust, interruptible workflow from the ground up reinforced several critical engineering principles:

  1. Granular Control is Power: While frameworks simplify, they often hide the complexities of state management, error handling, and concurrency. Rolling our own orchestrator gave us absolute control, which is essential for performance tuning and specific business logic. This mirrors my view on why raw API calls often outperform ORMs in critical path scenarios.
  2. Explicit State is Non-Negotiable: Ambiguous state leads to unpredictable behavior. Clearly defining WorkflowState and ensuring every node explicitly updates it makes the system auditable and debuggable. This feels right when thinking about a neural network's internal state: it needs to be well-defined for reliable computation.
  3. Human Oversight as a Feature, Not a Crutch: Integrating HITL isn't admitting AI's failure; it's designing for robustness. It's about combining the speed and scale of AI with the nuanced judgment and ethical reasoning of humans. This hybrid intelligence is what truly pushes us closer to reliable, intelligent systems.
  4. The Graph is a Concept, Not Always a Library: We implemented graph-like state transitions without a heavy "graph library." The underlying concept of a state machine, with nodes and edges (transitions), is powerful enough to be expressed directly in code, leading to leaner, faster implementations.

As we continue to build more complex AI systems—perhaps eventually replicating the intricate expert networks of the human brain—the need for such controlled, supervised execution will only grow. It's about creating intelligent systems that are not just capable but also accountable and trustworthy. This architecture is a small but vital step in that direction, ensuring that our agents operate not in an isolated, autonomous bubble, but as integrated, supervised components within a larger, human-centric ecosystem.