Most AI pipelines break on exceptions. Let's build one that stops, asks a question, and waits for your answer. 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. last article 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. "ONIONS YELLOW JBO" "Product #102: Yellow Onions" "Product #247: Jumbo Onions" 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: Foundry 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



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 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 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 AmbiguityDetector # 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 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 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 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 HumanInTheLoopPhase AmbiguityDetector pending_clarification The Human-in-the-Loop in Action If you're following along, navigate tohuman_in_the_loop_example directory in the repository. human_in_the_loop_example 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 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' --- --- 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. --- --- 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? http://localhost:8000 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'. --- --- 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: Ctrl+C --- Final Job Status: ready_for_final_processing ---
--- Final Request Status: resolved --- --- 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. completed ready_for_final_processing 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. It Reduces Waste: We catch errors and ambiguities at the earliest possible moment, preventing them from causing bigger problems in downstream systems. It Reduces Waste: 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's More Efficient for Humans: 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. It Enables Complex, Chained Workflows: 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.

This story contains AI-generated text. The author has used AI either for research, to generate outlines, or write the text itself. 

Walkthroughs, tutorials, guides, and tips. This story will teach you how to do something new or how to do something better.

Building AI Pipelines That Know When to Stop and Ask for Help

About Author

Comments

TOPICS

THIS ARTICLE WAS FEATURED IN

Related Stories

AI Agents to Discover Drugs

Making AI Work for Your Startup: Essential Dos and Don'ts

AI as the "Bad Student" in Class

AI Will Not Kill Quantum Computing

AI's Unstoppable Energy Appetite: A Looming Crisis

Beyond the Algorithm: How Training Data Can Make or Break a Generative AI Model

AI Agents to Discover Drugs

Making AI Work for Your Startup: Essential Dos and Don'ts

AI as the "Bad Student" in Class

AI Will Not Kill Quantum Computing

AI's Unstoppable Energy Appetite: A Looming Crisis

Beyond the Algorithm: How Training Data Can Make or Break a Generative AI Model

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps