Supervised Trajectories — Reject and Revise, and Abort¶

This notebook shows two trajectories available with SupervisedTaskHandler that are not possible with run() or run(with_approval=True):

• Reject and Revise. After the agent delivers its result, the human calls reject() with structured feedback. The handler routes the rejection back through get_next_step() deterministically — the agent revises without replanning from scratch.
• Abort. The human calls abort() mid-execution to terminate cleanly, setting a TaskHandlerError (or a custom exception) on the handler that the caller can inspect.

Both trajectories use a code refactoring scenario to keep the focus on the HITL mechanics rather than tooling.

In [ ]:

# Uncomment the line below to install `llm-agents-from-scratch` from PyPI
# !pip install llm-agents-from-scratch

# Uncomment the line below to install `llm-agents-from-scratch` from PyPI # !pip install llm-agents-from-scratch

Running an Ollama service¶

To execute the code provided in this notebook, you'll need to have Ollama installed on your local machine and have its LLM hosting service running. To download Ollama, follow the instructions found on this page: https://ollama.com/download. After downloading and installing Ollama, you can start a service by opening a terminal and running ollama serve.

In [1]:

import os
import shutil
import subprocess
import time
import urllib.error
import urllib.request


def ensure_ollama(host="http://localhost:11434", timeout=15):
    """Start Ollama if not already running and wait until responsive."""

    def _up():
        try:
            urllib.request.urlopen(f"{host}/api/tags", timeout=1)
            return True
        except (urllib.error.URLError, ConnectionError, TimeoutError):
            return False

    if _up():
        return print(f"\u2713 Ollama already running at {host}")

    ollama_path = shutil.which("ollama")
    if ollama_path is None:
        for candidate in [
            "/teamspace/studios/this_studio/.local/bin/ollama",
            "/usr/local/bin/ollama",
            "/usr/bin/ollama",
        ]:
            if os.path.exists(candidate):
                ollama_path = candidate
                break
    if ollama_path is None:
        raise RuntimeError(
            "Could not find the ollama binary. Install with: "
            "curl -fsSL https://ollama.com/install.sh | sh",
        )

    print(f"Starting Ollama server ({ollama_path})...")
    subprocess.Popen(
        [ollama_path, "serve"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )

    deadline = time.time() + timeout
    while time.time() < deadline:
        if _up():
            return print(f"\u2713 Ollama up and running at {host}")
        time.sleep(0.5)

    raise RuntimeError(f"Ollama did not start within {timeout}s")


ensure_ollama()

import os import shutil import subprocess import time import urllib.error import urllib.request def ensure_ollama(host="http://localhost:11434", timeout=15): """Start Ollama if not already running and wait until responsive.""" def _up(): try: urllib.request.urlopen(f"{host}/api/tags", timeout=1) return True except (urllib.error.URLError, ConnectionError, TimeoutError): return False if _up(): return print(f"\u2713 Ollama already running at {host}") ollama_path = shutil.which("ollama") if ollama_path is None: for candidate in [ "/teamspace/studios/this_studio/.local/bin/ollama", "/usr/local/bin/ollama", "/usr/bin/ollama", ]: if os.path.exists(candidate): ollama_path = candidate break if ollama_path is None: raise RuntimeError( "Could not find the ollama binary. Install with: " "curl -fsSL https://ollama.com/install.sh | sh", ) print(f"Starting Ollama server ({ollama_path})...") subprocess.Popen( [ollama_path, "serve"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) deadline = time.time() + timeout while time.time() < deadline: if _up(): return print(f"\u2713 Ollama up and running at {host}") time.sleep(0.5) raise RuntimeError(f"Ollama did not start within {timeout}s") ensure_ollama()

✓ Ollama already running at http://localhost:11434

In [2]:

import logging

from llm_agents_from_scratch import LLMAgent
from llm_agents_from_scratch.data_structures import Task
from llm_agents_from_scratch.llms import OllamaLLM
from llm_agents_from_scratch.logger import enable_console_logging

enable_console_logging(logging.INFO)

llm = OllamaLLM(model="qwen3:14b", think=False)
agent = LLMAgent(llm=llm)

import logging from llm_agents_from_scratch import LLMAgent from llm_agents_from_scratch.data_structures import Task from llm_agents_from_scratch.llms import OllamaLLM from llm_agents_from_scratch.logger import enable_console_logging enable_console_logging(logging.INFO) llm = OllamaLLM(model="qwen3:14b", think=False) agent = LLMAgent(llm=llm)

Trajectory 1: Reject and Revise¶

The agent is asked to refactor a function by replacing its if-elif chain with a dictionary dispatch. After it delivers its result, the human inspects it, finds it incomplete (no error handling), and calls reject() with specific feedback. The handler feeds the rejection back through get_next_step() and the agent revises.

In [3]:

ORIGINAL = """\
def calculate(x, y, operation):
    if operation == \"add\":
        return x + y
    elif operation == \"subtract\":
        return x - y
    elif operation == \"multiply\":
        return x * y
    elif operation == \"divide\":
        return x / y
"""

ORIGINAL = """\ def calculate(x, y, operation): if operation == \"add\": return x + y elif operation == \"subtract\": return x - y elif operation == \"multiply\": return x * y elif operation == \"divide\": return x / y """

In [4]:

task = Task(
    instruction=(
        "Refactor the following Python function to replace the "
        f"if-elif chain with a dictionary dispatch:\n\n{ORIGINAL}"
    ),
)
task_handler = await agent.run_supervised(task)

task = Task( instruction=( "Refactor the following Python function to replace the " f"if-elif chain with a dictionary dispatch:\n\n{ORIGINAL}" ), ) task_handler = await agent.run_supervised(task)

In [5]:

step = await task_handler.get_next_step(None)
step

step = await task_handler.get_next_step(None) step

Out[5]:

TaskStep(id_='43578526-53c2-461e-8df4-cd214a0c43da', task_id='527becc1-a663-4567-aaf6-921dbf134483', instruction='Refactor the following Python function to replace the if-elif chain with a dictionary dispatch:\n\ndef calculate(x, y, operation):\n    if operation == "add":\n        return x + y\n    elif operation == "subtract":\n        return x - y\n    elif operation == "multiply":\n        return x * y\n    elif operation == "divide":\n        return x / y\n')

In [6]:

step_result = await task_handler.run_step(step)
step_result

step_result = await task_handler.run_step(step) step_result

INFO (llm_agents_fs.SupervisedTaskHandler) :      ⚙️ Processing Step: Refactor the following Python function to replace the if-elif chain with a dictionary dispatch:

def calculate(x, y, operation):
 ...[TRUNCATED]
INFO (llm_agents_fs.SupervisedTaskHandler) :      ✅ Step Result: I need to refactor the function by replacing the if-elif chain with a dictionary that maps operations to corresponding lambda functions...[TRUNCATED]

Out[6]:

TaskStepResult(task_step_id='43578526-53c2-461e-8df4-cd214a0c43da', content='I need to refactor the function by replacing the if-elif chain with a dictionary that maps operations to corresponding lambda functions or function references. This approach will make the code more concise and easier to extend. Let me proceed with this change.')

In [7]:

step = await task_handler.get_next_step(step_result)
step

step = await task_handler.get_next_step(step_result) step

INFO (llm_agents_fs.SupervisedTaskHandler) :      🧠 New Step: Define a dictionary that maps each operation string to a corresponding lambda function or function reference, and then use the dictionary ...[TRUNCATED]

Out[7]:

TaskStep(id_='05f23f72-89e8-4a00-829a-212f479f48c0', task_id='527becc1-a663-4567-aaf6-921dbf134483', instruction='Define a dictionary that maps each operation string to a corresponding lambda function or function reference, and then use the dictionary to perform the calculation.')

In [8]:

step_result = await task_handler.run_step(step)
step_result

step_result = await task_handler.run_step(step) step_result

INFO (llm_agents_fs.SupervisedTaskHandler) :      ⚙️ Processing Step: Define a dictionary that maps each operation string to a corresponding lambda function or function reference, and then use the dic...[TRUNCATED]
INFO (llm_agents_fs.SupervisedTaskHandler) :      ✅ Step Result: I will define a dictionary called `operations` where each key is an operation string, and the corresponding value is a lambda function ...[TRUNCATED]

Out[8]:

TaskStepResult(task_step_id='05f23f72-89e8-4a00-829a-212f479f48c0', content='I will define a dictionary called `operations` where each key is an operation string, and the corresponding value is a lambda function that performs the respective calculation. Then, I will use the dictionary to perform the calculation based on the `operation` parameter. Let me implement this change.')

In [9]:

step = await task_handler.get_next_step(step_result)
step

step = await task_handler.get_next_step(step_result) step

INFO (llm_agents_fs.SupervisedTaskHandler) :      🧠 New Step: Define the dictionary `operations` with lambda functions for each operation and use it to perform the calculation.

Out[9]:

TaskStep(id_='1469aa64-8378-44f4-a3c1-1200311a7db6', task_id='527becc1-a663-4567-aaf6-921dbf134483', instruction='Define the dictionary `operations` with lambda functions for each operation and use it to perform the calculation.')

In [10]:

step_result = await task_handler.run_step(step)
step_result

step_result = await task_handler.run_step(step) step_result

INFO (llm_agents_fs.SupervisedTaskHandler) :      ⚙️ Processing Step: Define the dictionary `operations` with lambda functions for each operation and use it to perform the calculation.
INFO (llm_agents_fs.SupervisedTaskHandler) :      ✅ Step Result: I will define the `operations` dictionary with lambda functions for each operation. Here's how the dictionary will look:

```python
ope...[TRUNCATED]

Out[10]:

TaskStepResult(task_step_id='1469aa64-8378-44f4-a3c1-1200311a7db6', content='I will define the `operations` dictionary with lambda functions for each operation. Here\'s how the dictionary will look:\n\n```python\noperations = {\n    "add": lambda x, y: x + y,\n    "subtract": lambda x, y: x - y,\n    "multiply": lambda x, y: x * y,\n    "divide": lambda x, y: x / y\n}\n```\n\nThen, I will use this dictionary to perform the calculation by accessing the appropriate lambda function based on the `operation` parameter. The updated function will look like this:\n\n```python\ndef calculate(x, y, operation):\n    return operations[operation](x, y)\n```\n\nThis refactoring replaces the if-elif chain with a dictionary-based dispatch, making the code more concise and easier to maintain.')

In [11]:

result = await task_handler.get_next_step(step_result)
result

result = await task_handler.get_next_step(step_result) result

INFO (llm_agents_fs.SupervisedTaskHandler) :      No new step required.

Out[11]:

TaskResult(task_id='527becc1-a663-4567-aaf6-921dbf134483', content='I will define the `operations` dictionary with lambda functions for each operation. Here\'s how the dictionary will look:\n\n```python\noperations = {\n    "add": lambda x, y: x + y,\n    "subtract": lambda x, y: x - y,\n    "multiply": lambda x, y: x * y,\n    "divide": lambda x, y: x / y\n}\n```\n\nThen, I will use this dictionary to perform the calculation by accessing the appropriate lambda function based on the `operation` parameter. The updated function will look like this:\n\n```python\ndef calculate(x, y, operation):\n    return operations[operation](x, y)\n```\n\nThis refactoring replaces the if-elif chain with a dictionary-based dispatch, making the code more concise and easier to maintain.')

In [12]:

rejected = task_handler.reject(
    result,
    feedback=(
        "Good start, but add error handling: raise ValueError for "
        "unknown operations and ZeroDivisionError for division by zero."
    ),
)
rejected

rejected = task_handler.reject( result, feedback=( "Good start, but add error handling: raise ValueError for " "unknown operations and ZeroDivisionError for division by zero." ), ) rejected

Out[12]:

RejectedTaskResult(failed_result_content='I will define the `operations` dictionary with lambda functions for each operation. Here\'s how the dictionary will look:\n\n```python\noperations = {\n    "add": lambda x, y: x + y,\n    "subtract": lambda x, y: x - y,\n    "multiply": lambda x, y: x * y,\n    "divide": lambda x, y: x / y\n}\n```\n\nThen, I will use this dictionary to perform the calculation by accessing the appropriate lambda function based on the `operation` parameter. The updated function will look like this:\n\n```python\ndef calculate(x, y, operation):\n    return operations[operation](x, y)\n```\n\nThis refactoring replaces the if-elif chain with a dictionary-based dispatch, making the code more concise and easier to maintain.', feedback='Good start, but add error handling: raise ValueError for unknown operations and ZeroDivisionError for division by zero.')

In [13]:

step = await task_handler.get_next_step(rejected)
step

step = await task_handler.get_next_step(rejected) step

INFO (llm_agents_fs.SupervisedTaskHandler) :      🧠 New Step (rejection): Good start, but add error handling: raise ValueError for unknown operations and ZeroDivisionError for division by zero.

Out[13]:

TaskStep(id_='0e361fcc-5680-4abc-94f7-aed80e88ddff', task_id='527becc1-a663-4567-aaf6-921dbf134483', instruction='The human operator REJECTED your proposed task result. Revise your approach and try again.\n\n<proposed-result>\nI will define the `operations` dictionary with lambda functions for each operation. Here\'s how the dictionary will look:\n\n```python\noperations = {\n    "add": lambda x, y: x + y,\n    "subtract": lambda x, y: x - y,\n    "multiply": lambda x, y: x * y,\n    "divide": lambda x, y: x / y\n}\n```\n\nThen, I will use this dictionary to perform the calculation by accessing the appropriate lambda function based on the `operation` parameter. The updated function will look like this:\n\n```python\ndef calculate(x, y, operation):\n    return operations[operation](x, y)\n```\n\nThis refactoring replaces the if-elif chain with a dictionary-based dispatch, making the code more concise and easier to maintain.\n</proposed-result>\n\n<human-correction>\nGood start, but add error handling: raise ValueError for unknown operations and ZeroDivisionError for division by zero.\n</human-correction>')

In [14]:

step_result = await task_handler.run_step(step)
step_result

step_result = await task_handler.run_step(step) step_result

INFO (llm_agents_fs.SupervisedTaskHandler) :      ⚙️ Processing Step: The human operator REJECTED your proposed task result. Revise your approach and try again.

<proposed-result>
I will define the `o...[TRUNCATED]
INFO (llm_agents_fs.SupervisedTaskHandler) :      ✅ Step Result: I need to revise the function to include error handling as requested. Specifically, I should:

1. Add a check to raise a `ValueError` i...[TRUNCATED]

Out[14]:

TaskStepResult(task_step_id='0e361fcc-5680-4abc-94f7-aed80e88ddff', content='I need to revise the function to include error handling as requested. Specifically, I should:\n\n1. Add a check to raise a `ValueError` if the `operation` is not found in the `operations` dictionary.\n2. Add a check to raise a `ZeroDivisionError` if the `operation` is "divide" and `y` is zero.\n\nHere\'s the updated function with error handling:\n\n```python\ndef calculate(x, y, operation):\n    operations = {\n        "add": lambda x, y: x + y,\n        "subtract": lambda x, y: x - y,\n        "multiply": lambda x, y: x * y,\n        "divide": lambda x, y: x / y\n    }\n\n    if operation not in operations:\n        raise ValueError(f"Unknown operation: {operation}")\n\n    if operation == "divide" and y == 0:\n        raise ZeroDivisionError("Division by zero is not allowed")\n\n    return operations[operation](x, y)\n```\n\nThis version includes the requested error handling while maintaining the dictionary-based dispatch approach.')

In [15]:

result = await task_handler.get_next_step(step_result)
result

result = await task_handler.get_next_step(step_result) result

INFO (llm_agents_fs.SupervisedTaskHandler) :      No new step required.

Out[15]:

TaskResult(task_id='527becc1-a663-4567-aaf6-921dbf134483', content='I need to revise the function to include error handling as requested. Specifically, I should:\n\n1. Add a check to raise a `ValueError` if the `operation` is not found in the `operations` dictionary.\n2. Add a check to raise a `ZeroDivisionError` if the `operation` is "divide" and `y` is zero.\n\nHere\'s the updated function with error handling:\n\n```python\ndef calculate(x, y, operation):\n    operations = {\n        "add": lambda x, y: x + y,\n        "subtract": lambda x, y: x - y,\n        "multiply": lambda x, y: x * y,\n        "divide": lambda x, y: x / y\n    }\n\n    if operation not in operations:\n        raise ValueError(f"Unknown operation: {operation}")\n\n    if operation == "divide" and y == 0:\n        raise ZeroDivisionError("Division by zero is not allowed")\n\n    return operations[operation](x, y)\n```\n\nThis version includes the requested error handling while maintaining the dictionary-based dispatch approach.')

In [16]:

await task_handler.complete(result)
print(result.content)

await task_handler.complete(result) print(result.content)

I need to revise the function to include error handling as requested. Specifically, I should:

1. Add a check to raise a `ValueError` if the `operation` is not found in the `operations` dictionary.
2. Add a check to raise a `ZeroDivisionError` if the `operation` is "divide" and `y` is zero.

Here's the updated function with error handling:

```python
def calculate(x, y, operation):
    operations = {
        "add": lambda x, y: x + y,
        "subtract": lambda x, y: x - y,
        "multiply": lambda x, y: x * y,
        "divide": lambda x, y: x / y
    }

    if operation not in operations:
        raise ValueError(f"Unknown operation: {operation}")

    if operation == "divide" and y == 0:
        raise ZeroDivisionError("Division by zero is not allowed")

    return operations[operation](x, y)
```

This version includes the requested error handling while maintaining the dictionary-based dispatch approach.

Trajectory 2: Abort¶

The agent is asked to refactor the authentication module. After seeing the first planned step, the human realises the task conflicts with an ongoing security audit and calls abort(). The handler terminates cleanly and the caller can inspect the exception that was set.

In [17]:

task2 = Task(
    instruction=(
        "Refactor the authentication module to consolidate all login, "
        "logout, and session renewal functions into a single AuthService class."
    ),
)
task_handler2 = await agent.run_supervised(task2)

task2 = Task( instruction=( "Refactor the authentication module to consolidate all login, " "logout, and session renewal functions into a single AuthService class." ), ) task_handler2 = await agent.run_supervised(task2)

In [18]:

step = await task_handler2.get_next_step(None)
step

step = await task_handler2.get_next_step(None) step

Out[18]:

TaskStep(id_='c14e474b-986c-4b7f-b513-575df74eb097', task_id='8361095e-898f-4ce6-95e6-949c78a481f8', instruction='Refactor the authentication module to consolidate all login, logout, and session renewal functions into a single AuthService class.')

In [19]:

await task_handler2.abort(
    error=RuntimeError(
        "Refactoring auth is blocked pending the security audit. "
        "Revisit after the audit completes.",
    ),
)

await task_handler2.abort( error=RuntimeError( "Refactoring auth is blocked pending the security audit. " "Revisit after the audit completes.", ), )

In [20]:

task_handler2.exception()

task_handler2.exception()

Out[20]:

RuntimeError('Refactoring auth is blocked pending the security audit. Revisit after the audit completes.')