Skip to content

Human-in-the-Loop

Summary

Some inputs are ambiguous and must be resolved by a human before processing can continue. The blueprint models this as a durable pause: an exception is raised, a question is recorded, the job pauses, and later resumes.

The mechanism

  1. During processing, a stage raises a domain exception carrying the context needed to ask the question:
1
2
3
4
5
6
raise PauseForResolution(
    job_id=jid,
    reference="<stable key for this decision>",
    options=[...],
    context={...},
)
  1. The orchestrator catches it, records a question document, and re-raises so the worker marks the job paused.
  2. A client lists pending questions and answers one:
1
2
GET  /jobs/{id}/questions            -> list pending questions
POST /questions/{qid}/resolve  {choice}  -> record the choice
  1. When the last pending question is resolved, the resolve endpoint flips the parent job back to pending.
  2. The worker picks it up again, rebuilds a resolution map from previously answered questions, and seeds it into the pipeline so the same decision is not asked twice.

The question document

1
2
3
4
5
6
7
8
class Question(BaseDocument):
    job_id: str
    reference: str            # stable key: stage|field|value
    status: str               # pending | resolved
    options: list[dict]
    context: dict
    chosen: dict | None = None
    resolved_at: str | None = None

Resume map

The worker turns answered questions into a lookup keyed by reference, then injects it into the pipeline so that stage can skip re-raising:

1
2
3
4
resolved = {q.reference: q.chosen for q in await svc.search(
    filter={"job_id": jid, "status": "resolved"})}

pipeline = build(root, instances={"pause": handler}, config={"resolved": resolved})

Resolve endpoint

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
async def resolve_question(request, id, body):
    svc = request.app.state.db_resources["question_svc"]
    q = await svc.read(id)
    await svc.update(id, {"chosen": body["chosen"], "status": "resolved",
                          "resolved_at": now()})

    pending = await svc.search(filter={"job_id": q.job_id, "status": "pending"},
                               limit=1)
    if not pending:
        jobs = request.app.state.db_resources["job_svc"]
        job = await jobs.read(q.job_id)
        if job and job.status == "paused":
            await jobs.update_status(q.job_id, "pending")
    return q

CLI parity

The same flow is available through scripts/CLI: list, get, create, questions, resolve, retry.