88 lines
2.5 KiB
Markdown
88 lines
2.5 KiB
Markdown
# 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:
|
|
|
|
```python
|
|
raise PauseForResolution(
|
|
job_id=jid,
|
|
reference="<stable key for this decision>",
|
|
options=[...],
|
|
context={...},
|
|
)
|
|
```
|
|
|
|
2. The orchestrator catches it, records a **question document**, and re-raises
|
|
so the worker marks the job `paused`.
|
|
3. A client lists pending questions and answers one:
|
|
|
|
```
|
|
GET /jobs/{id}/questions -> list pending questions
|
|
POST /questions/{qid}/resolve {choice} -> record the choice
|
|
```
|
|
|
|
4. When the **last** pending question is resolved, the resolve endpoint flips
|
|
the parent job back to `pending`.
|
|
5. 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
|
|
|
|
```python
|
|
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:
|
|
|
|
```python
|
|
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
|
|
|
|
```python
|
|
async def resolve_question(request, id, body):
|
|
svc = request.app.state.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.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`.
|