Skip to content

Human In The Loop Guide

This guide explains how kneo_agent supports human review steps inside workflows.

Middleware and human-in-the-loop are complementary:

  • middleware is best for automated policy, logging, or short-circuiting
  • human steps are best when a real person must approve, edit, or pause execution

Core Idea

Human-in-the-loop support is modeled as a normal workflow participant.

That keeps the design aligned with the rest of the workflow system:

  • agents can draft or analyze
  • function steps can transform
  • human steps can approve, reject, edit, or pause execution

The main API is WorkflowBuilder.human_step(...).

Inline Resolver

Use an inline resolver when your application can obtain human input during the same workflow run.

from kneo_agent.workflows import WorkflowBuilder

review = WorkflowBuilder.human_step(
    "approval",
    "Approve this draft?",
    resolver=lambda request: "Approved by Alice",
)

The resolver receives a HumanInterventionRequest containing:

  • step_name
  • prompt
  • the current messages
  • the active RunConfig
  • any workflow metadata attached to the step

If the resolver returns a string, that string is appended as a human message and the workflow continues.

Pause / Escalation Mode

If no resolver is provided, or if the resolver returns None, the workflow raises HumanInterventionRequiredError.

from kneo_agent import HumanInterventionRequiredError

try:
    await workflow.run(messages, config)
except HumanInterventionRequiredError as exc:
    print(exc.step_name)
    print(exc.prompt)
    print(exc.request)

This is the right fit when your host application needs to:

  • pause execution
  • show a review UI
  • collect approval from a person later
  • resume the broader process in a separate request or session

Graph And Sequential Workflows

Human steps work anywhere a workflow participant can appear:

  • SequentialWorkflow
  • graph Workflow
  • nested workflows
  • handoff, concurrent, or group-chat orchestrations

Example:

workflow = WorkflowBuilder.sequential(
    [
        writer,
        WorkflowBuilder.human_step(
            "editor_review",
            lambda messages, config: f"Approve this draft? {messages[-1].content}",
            resolver=lambda request: "Approved",
        ),
        publish_step,
    ],
    name="release-workflow",
)
  • HumanInTheLoopStep
  • HumanInterventionRequest
  • HumanInterventionRequiredError
  • WorkflowBuilder.human_step(...)

Examples

For automated interception before or around a workflow-as-agent boundary, see the middleware examples: