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:

  1. The AmbiguityDetector is 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

  2. The HumanInTheLoopPhase: the stop and ask process. It’s a special, pre-built phase you add to your pipeline. You simply tell it which AmbiguityDetector to 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 to pending_clarification and halts the pipeline for that specific job.

The Human-in-the-Loop in Action

If you're following along, navigate tohuman_in_the_loop_example directory 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:

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.