Most AI pipelines break on exceptions. Let’s build one that stops, asks a question, and waits for your answer.
In our last article, we built our first tangible AI Data Flywheel. We also created a simple Correction Deck that allowed us to fix an AI’s mistakes and generate a perfect training file.
But true AI training contains thousands upon thousands of files, so going through each is impossible.
A smarter way would be for the AI to spot a problem in the middle of a process, recognize it’s confused, stop, and ask a human to provide the missing piece of information before continuing.
Today, we’re building that smart pipeline.
Ambiguity in a Multi-Step Process
Imagine our invoice AI is now part of a larger process. After extracting the text, it needs to link each line item to a canonical product in our company’s inventory database.
The AI processes an invoice and extracts the line item "ONIONS YELLOW JBO". It checks the database but finds two possible matches: "Product #102: Yellow Onions" and "Product #247: Jumbo Onions". The AI is stuck and cannot resolve on its own.
A brittle pipeline would either fail, guess wrong (polluting our downstream data), or silently leave the item unlinked. A brilliant pipeline does something better: it pauses and asks a targeted question.
The Tools for an Interactive Loop
To build this, our Foundry framework introduces two new, powerful concepts that work together:
- The
AmbiguityDetectoris the brain of the operation. It’s a simple Python class where the user defines the business logic for what constitutes a problem. This method analyzes a job’s output and, if it finds an issue, returns a list of questions to ask the user.
# The abstract contract in the framework
class AmbiguityDetector(ABC):
@abstractmethod
def detect(self, job: Job) -> list[dict]:
"""Analyzes a job and returns questions if ambiguities are found."""
pass
# Our specific implementation for the invoice problem
class UnlinkedProductDetector(AmbiguityDetector):
def detect(self, job: Job) -> list[dict]:
requests = []
for item in job.initial_ai_output.get("inventory_items", []):
# Our business rule: If an item isn't linked, we have a problem!
if item.get("linked_pantry_item_id") is None:
requests.append({
"request_type": "LINK_PRODUCT",
"context_data": { ... } // Data needed to ask the question
})
return requests
- The
HumanInTheLoopPhase: the stop and ask process. It’s a special, pre-built phase you add to your pipeline. You simply tell it whichAmbiguityDetectorto use. When the pipeline executes this phase, it runs your detector. If the detector returns any questions, the phase immediately changes the job’s status topending_clarificationand halts the pipeline for that specific job.
The Human-in-the-Loop in Action
If you’re following along, navigate tohuman_in_the_loop_exampledirectory in the repository.
Step 1: Run the Script
This script simulates the entire workflow. It will first set up a database with a job that’s already halfway done but contains the ambiguous unlinked onion problem we described. Then, it will run a pipeline whose only job is to detect this ambiguity.
From your terminal, run:
python hhuman_in_the_loop_example.py
First, you’ll see the detection pipeline run in your terminal. Notice the output: the job’s status is changed, and the pipeline is paused.
--- Running the Ambiguity Detection Pipeline for Job #1... ---
--- [Job 1] Running Phase: HumanInTheLoopPhase ---
--- [Job 1] Found 1 ambiguities. Pausing pipeline. ---
--- Pipeline finished. Job status is now: 'pending_clarification' ---
Next, the script starts a web server.
--- Human-in-the-Loop server running at http://localhost:8000 ---
--- Open the URL in your browser to answer the clarification question. ---
Step 2: Use the Clarification Feed
Open http://localhost:8000 in your browser. Instead of a full correction form, you see a simple, targeted question: The item ‘ONIONS YELLOW JBO’ … needs to be linked … Which product is it?
This is our system asking for help. From the dropdown, select Yellow Onions and click Link Product.
The UI will update to show All Done! and, crucially, look back at your terminal. You’ll see a log confirming that your action has un-paused the job:
--- Received resolution for request #1 ... ---
--- Request #1 resolved. Job #1 is now 'ready_for_final_processing'. ---
Step 3: Stop the Server and Verify
Go back to your terminal and press Ctrl+C to stop the server. The script will print a final status check:
--- Final Job Status: ready_for_final_processing ---
--- Final Request Status: resolved ---
The job’s status isn’t completed yet. It’s now ready_for_final_processing. We have successfully intervened, provided the missing information, and put the job back in the queue, ready for the next pipeline to take over and finish the work.
Why This is a Game-Changer
This interactive pattern fundamentally changes how we can build AI systems:
- It Reduces Waste: We catch errors and ambiguities at the earliest possible moment, preventing them from causing bigger problems in downstream systems.
- It’s More Efficient for Humans: Operators aren’t wading through pages of correct data to find one error. The system presents them with a clean queue of specific, actionable questions.
- It Enables Complex, Chained Workflows: We can now design incredibly sophisticated, multi-stage AI processes with human “checkpoints” in the middle, confident that the system will pause gracefully when it needs guidance.
What’s Next?
We’ve built a script that runs a pipeline offline and a second script that hosts an interactive UI. But in a real production application, these are two separate worlds. Your web server needs to be instantly responsive to user requests; it can’t be tied up running a 10-minute AI batch job.
How do we decouple the application that starts the job from the background worker that executes the job?
In our next article, we will graduate from self-contained scripts to a true, production-grade architecture. We’ll introduce Celery and Redis to build a robust, scalable system with a dedicated pool of background workers, ready to handle any AI task without blocking our main application.
