Why Human-in-the-Loop Matters for Production Agents
Fully autonomous agents make mistakes. In consequential workflows (sending emails, modifying databases, submitting orders), an approval gate between agent reasoning and action execution is not just a nice-to-have -- it is the difference between a useful tool and a liability.
FastAgency's UI abstraction makes human-in-the-loop natural: the same approval prompts work in terminal (for development), in a web UI (for users), and can be adapted to a REST API (for asynchronous approval workflows).
The UI Methods for Human Input
| Method | What it shows the user | Returns |
|---|---|---|
| ui.text_input(prompt) | Free-text input field | str |
| ui.multiple_choice(prompt, choices) | Button selection from a list | str (selected choice) |
| ui.system_message(message) | A system-level notification | None (side effect only) |
Pattern 1: Approval Gate Before a Destructive Action
@fastagency.wf.register(name='send_outreach', description='Research prospect and send personalised outreach email')
def outreach_workflow(ui: fastagency.UIBase, params: dict) -> str:
researcher = ConversableAgent(
name='researcher',
system_message='Research companies and write personalised outreach emails.',
llm_config=llm_config,
)
user_proxy = ConversableAgent(name='proxy', human_input_mode='NEVER')
company_name = ui.text_input('Which company should I research for outreach?')
# Agent researches and drafts the email
draft = user_proxy.initiate_chat(
researcher,
message=f'Research {company_name} and write a personalised cold outreach email.',
)
# Show draft to human for approval
ui.system_message(f'Draft email:\n\n{draft.summary}')
# Approval gate
decision = ui.multiple_choice(
'What would you like to do with this email?',
choices=['Send it', 'Edit and resend', 'Discard'],
)
if decision == 'Send it':
send_email(draft.summary, recipient=company_name)
return f'Email sent to {company_name}'
elif decision == 'Edit and resend':
edits = ui.text_input('What changes would you like to make?')
# Re-run with edit instructions...
return 'Re-running with edits'
else:
return 'Email discarded'
Pattern 2: Multi-Step Confirmation in a Data Pipeline
For workflows that modify data in stages, checkpoint after each stage with a confirmation before proceeding.
@fastagency.wf.register(name='data_cleanup', description='Analyse and clean a dataset')
def data_pipeline(ui: fastagency.UIBase, params: dict) -> str:
analyst = ConversableAgent(
name='analyst',
system_message='You are a data analyst. Analyse data quality issues.',
llm_config=llm_config,
)
# Step 1: analyse
ui.system_message('Step 1: Analysing data quality...')
analysis = run_agent(analyst, 'Analyse the dataset at /data/input.csv for quality issues')
ui.system_message(f'Analysis complete:\n{analysis}')
proceed = ui.multiple_choice(
'Proceed with cleanup?',
choices=['Yes, apply all fixes', 'Yes, but show me each fix first', 'No, abort'],
)
if proceed == 'No, abort':
return 'Aborted'
# Step 2: cleanup with optional per-fix approval
review_each = (proceed == 'Yes, but show me each fix first')
fixes_applied = apply_data_fixes(review_each=review_each, ui=ui)
return f'{fixes_applied} fixes applied successfully'
Asynchronous Approval via FastAPI Adapter
For workflows where the approval cannot happen synchronously (e.g. a manager approves via Slack hours later), structure the workflow to pause and resume:
- Agent completes the analysis/draft and stores the result in a database with a pending status
- A notification is sent (Slack, email) with an approval link
- The manager clicks the link, which hits a FastAPI endpoint
- The endpoint updates the status and triggers the continuation workflow
This pattern goes beyond FastAgency's built-in UI methods -- it requires a custom state store and a second workflow or API endpoint for the approval action. FastAgency's FastAPIAdapter makes the second endpoint straightforward to build.
Testing Human-in-the-Loop Flows
In automated tests, you cannot have a human respond to prompts. FastAgency provides a way to inject mock responses:
from fastagency.ui.console import ConsoleUI
from unittest.mock import patch
def test_outreach_workflow():
# Mock user inputs in sequence
mock_inputs = ['Acme Corp', 'Send it']
with patch('builtins.input', side_effect=mock_inputs):
app = FastAgency(wf=fastagency.wf, ui=ConsoleUI())
result = app.run_workflow('send_outreach', {})
assert 'sent' in result.lower()
Test human-in-the-loop workflows with ConsoleUI and mock inputs first. This lets you run automated tests against the full workflow logic, including the approval decision branches, without needing a real browser or web server.