Disclosure: I run NxFlowAI. The code is illustrative and framework-free.
Most human-in-the-loop designs cover the happy path: draft, approve, send. The harder question is what happens when nobody clicks approve. The approver is on leave, it is 11 p.m., the queue is long. Without a rule, drafts sit for days and the customer hears nothing. With the wrong rule, drafts auto-send after a timeout and the approval step means nothing. Here is a way to handle it with human-in-the-loop AI workflow approval with a tiny state machine.
The rule: time escalates, it never approves
TRANSITIONS = {
"waiting": ["approved", "edited", "rejected", "escalated", "expired"],
"escalated": ["approved", "edited", "rejected", "expired"],
"approved": ["sent"],
"edited": ["sent"],
"rejected": [],
"expired": [],
"sent": [],
}
def move(draft, new_state, actor):
assert new_state in TRANSITIONS[draft.state], (draft.state, new_state)
log(draft.id, draft.state, new_state, actor, now())
draft.state = new_state
No path leads from waiting to sent without a person. That is the whole point.
Deadlines per risk level
deadlines:
low: {escalate_after: 4h, expire_after: 24h}
medium: {escalate_after: 1h, expire_after: 12h}
high: {escalate_after: 15m, expire_after: 4h} # prices, refunds, complaints
holding_message_after: 30m
Numbers like these are placeholders; set them from your own business hours and customer expectations.
Three timers, three different actions
- Holding message. If a draft has waited a while, send the customer a pre-approved acknowledgement, written by a person once: "Thanks, a team member will reply shortly." This is not the AI draft; it is a fixed message that is safe to send.
- Escalation. Move the draft to a second approver or a manager, and notify them on a channel they actually watch.
- Expiry. The draft is withdrawn. A task is created for the conversation owner to reply personally. An AI draft written hours ago may already be wrong (stock changed, the customer wrote again).
A scheduler, not a cron per draft
def tick():
for d in drafts.in_states(["waiting", "escalated"]):
age = now() - d.created_at
rule = DEADLINES[d.risk]
if age > rule.expire_after:
move(d, "expired", actor="system"); create_task(d.owner, d)
elif d.state == "waiting" and age > rule.escalate_after:
move(d, "escalated", actor="system"); notify(backup_for(d.owner), d)
if age > HOLDING_AFTER and not d.holding_sent:
send_fixed_ack(d.conversation_id); d.holding_sent = True
Two details people miss
- New customer message while a draft waits. Mark the draft stale and regenerate or expire it. Approving an answer to an outdated question is a classic mistake.
- Business hours. Measure escalation in working time, or set a separate out-of-hours rule. Escalating to a manager at 3 a.m. helps nobody.
Metrics worth logging
Median time to approval per risk level, count of expiries per week, and drafts approved after the customer had already written again. All three point at queue design problems, not model problems.
In a 72-hour audit we set these deadlines with the team that will live with them, before any model is chosen.
Top comments (0)