Human-in-the-Loop

Agents can ask users for input when they need clarification or approval. This works at both the agent level and the orchestrator level.

Agent-level input

Pass an on_ask_user callback to let the agent ask for user input:

agent_hitl.py

from veska import Agent

def ask_user(question: str) -> str:
    """Called when the agent needs user input"""
    print(f"Agent asks: {question}")
    return input("Your answer: ")

agent = Agent(
    name="assistant",
    system_prompt="You help build projects. Ask the user for preferences when needed.",
    model="claude-sonnet-4-6",
    on_ask_user=ask_user,
    ask_user_timeout=300,  # 5 minutes to respond (default)
)

# The agent will call ask_user when it needs input
result = agent.run("Set up a new web project for me")

The callback can be sync or async. Veska handles both automatically. If the user doesn't respond within ask_user_timeout seconds, the agent continues without input.

Orchestrator clarification

The Orchestrator can ask questions before planning. You provide a prompt that tells the AI what to ask about:

clarification.py

from veska import Orchestrator

def ask_user(questions: str) -> str:
    print(f"\nOrchestrator needs clarification:\n{questions}")
    return input("\nYour answers: ")

orchestrator = Orchestrator(
    model="claude-sonnet-4-6",
    agents=[backend, frontend],
    clarification_prompt="""Before planning, ask about:
    - Database preference (PostgreSQL, MySQL, SQLite)
    - Frontend framework (React, Vue, plain HTML)
    - Authentication requirements
    Only ask what's relevant to the request.""",
    on_ask_user=ask_user,
)

# When the user says "Build me a blog app", the orchestrator will
# first ask about database, frontend, and auth preferences,
# then create a plan based on the answers
result = orchestrator.run("Build me a blog app")

Checkpoints

Checkpoints pause execution and wait for approval. The orchestrator uses them to present the plan before executing:

checkpoint.py

from veska import EventType

# Listen for checkpoint events
async def on_checkpoint(event):
    print(f"\nCheckpoint: {event.data['title']}")
    print(f"Description: {event.data['description']}")

    approval = input("Approve? (y/n): ")
    if approval == "y":
        orchestrator.events.approve_checkpoint(
            event.data["checkpoint_id"],
            feedback="Looks good"
        )
    else:
        orchestrator.events.reject_checkpoint(
            event.data["checkpoint_id"],
            feedback="Please add more research tasks"
        )

orchestrator.events.on(EventType.CHECKPOINT, on_checkpoint)

Input requests via events

For web UIs, use the event system instead of callbacks:

events_hitl.py

# Listen for input requests
async def on_need_input(event):
    prompt = event.message
    # Show prompt in your UI, wait for user response
    # Then send the feedback back:
    await orchestrator.events.send_feedback(user_response)

orchestrator.events.on(EventType.NEED_INPUT, on_need_input)

Pause and resume

control.py

# Pause execution (e.g., user clicks "pause" in UI)
await orchestrator.events.pause()

# Resume when ready
await orchestrator.events.resume()

# Cancel everything
await orchestrator.events.cancel()