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.

Setting up the LLM¶

I recommend running these notebooks on Ollama Cloud. It serves much larger models than most laptops can run. The default cloud model is picked by benchmarking candidates on these notebooks, and an automated run checks every notebook against it each week. The notebook status page shows the current model and results.

1. Create an API key at https://ollama.com/settings/keys.
2. Export it before launching Jupyter: export OLLAMA_API_KEY=<your key>.
3. That's all. When a notebook builds its LLM, it detects the key and uses Ollama Cloud.

Prefer to run locally? Install Ollama from https://ollama.com/download and pull the model with ollama pull qwen3:14b. If no key is set, the notebook starts the Ollama service for you, so there is no need to run ollama serve. Local models are smaller, so your outputs will differ from the recorded ones.

Prefer OpenAI or Anthropic?

1. Install the extra: uv sync --extra openai or uv sync --extra anthropic.
2. Export OPENAI_API_KEY or ANTHROPIC_API_KEY before launching Jupyter. If you forget, you will be prompted for it, which keeps the key out of the saved notebook.
3. Set LLM_PROVIDER=openai (or anthropic), or pass provider="openai" to make_llm(). A key on its own never switches providers, so one exported for unrelated work cannot reroute you.

The switch applies wherever a notebook builds its LLM with make_llm(). A notebook that constructs OllamaLLM directly stays on Ollama. Output on gpt-5 or Claude will differ from the recorded outputs, and prompt-sensitive examples (the ch09 evaluator pattern, ch08 supervised trajectories) may behave noticeably differently.

In [1]:

# 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

In [2]:

from llm_agents_from_scratch.notebook_utils import make_llm

from llm_agents_from_scratch.notebook_utils import make_llm

In [3]:

import logging

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

enable_console_logging(logging.INFO)

llm = make_llm()
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.logger import enable_console_logging enable_console_logging(logging.INFO) llm = make_llm() agent = LLMAgent(llm=llm)

✓ Using Ollama Cloud (kimi-k2.7-code:cloud)

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 [4]:

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 [5]:

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 [6]:

step = await task_handler.get_next_step(None)
step

step = await task_handler.get_next_step(None) step

Out[6]:

TaskStep(id_='0229ad36-9bc2-4f3a-9fe0-4a4c06f7b191', task_id='68c253dd-29b0-44c4-b49d-ae5788398996', 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 [7]:

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) :      🛠️ Executing Tool Call: from_scratch__use_skill
INFO (llm_agents_fs.SupervisedTaskHandler) :      ✅ Successful Tool Call: <skill_content name="hyperframes">
# HyperFrames entry point

HyperFrames **renders video from HTML** — a composition is an HT...[TRUNCATED]
INFO (llm_agents_fs.SupervisedTaskHandler) :      ✅ Step Result: The user is asking for a simple Python refactoring task, not anything related to HyperFrames video creation. This is outside the scope ...[TRUNCATED]

Out[7]:

TaskStepResult(task_step_id='0229ad36-9bc2-4f3a-9fe0-4a4c06f7b191', content='The user is asking for a simple Python refactoring task, not anything related to HyperFrames video creation. This is outside the scope of the skills I have available, so I can just answer directly with the refactored code. No skill call needed.\n\nHere\'s the refactored version using dictionary dispatch:\n\n```python\ndef calculate(x, y, operation):\n    operations = {\n        "add": lambda a, b: a + b,\n        "subtract": lambda a, b: a - b,\n        "multiply": lambda a, b: a * b,\n        "divide": lambda a, b: a / b,\n    }\n    return operations[operation](x, y)\n```\n\nOr, using the built-in operator module:\n\n```python\nimport operator\n\ndef calculate(x, y, operation):\n    operations = {\n        "add": operator.add,\n        "subtract": operator.sub,\n        "multiply": operator.mul,\n        "divide": operator.truediv,\n    }\n    return operations[operation](x, y)\n```\n\nBoth replace the `if-elif` chain with a lookup dictionary. The second version is a bit cleaner since it uses named functions instead of lambdas.\n\nI should mention potential edge cases, like what happens with an unknown operation or division by zero, but keep it focused on the refactoring request.</think>Here are two clean refactorings.\n\nUsing lambda functions:\n\n```python\ndef calculate(x, y, operation):\n    operations = {\n        "add": lambda a, b: a + b,\n        "subtract": lambda a, b: a - b,\n        "multiply": lambda a, b: a * b,\n        "divide": lambda a, b: a / b,\n    }\n    return operations[operation](x, y)\n```\n\nOr, slightly cleaner, using the `operator` module:\n\n```python\nimport operator\n\ndef calculate(x, y, operation):\n    operations = {\n        "add": operator.add,\n        "subtract": operator.sub,\n        "multiply": operator.mul,\n        "divide": operator.truediv,\n    }\n    return operations[operation](x, y)\n```\n\nBoth versions map each operation name to the corresponding function, then dispatch with a single dictionary lookup and call. The `operator` version avoids lambdas and is often preferred for readability.')

In [8]:

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[8]:

TaskResult(task_id='68c253dd-29b0-44c4-b49d-ae5788398996', content='The user is asking for a simple Python refactoring task, not anything related to HyperFrames video creation. This is outside the scope of the skills I have available, so I can just answer directly with the refactored code. No skill call needed.\n\nHere\'s the refactored version using dictionary dispatch:\n\n```python\ndef calculate(x, y, operation):\n    operations = {\n        "add": lambda a, b: a + b,\n        "subtract": lambda a, b: a - b,\n        "multiply": lambda a, b: a * b,\n        "divide": lambda a, b: a / b,\n    }\n    return operations[operation](x, y)\n```\n\nOr, using the built-in operator module:\n\n```python\nimport operator\n\ndef calculate(x, y, operation):\n    operations = {\n        "add": operator.add,\n        "subtract": operator.sub,\n        "multiply": operator.mul,\n        "divide": operator.truediv,\n    }\n    return operations[operation](x, y)\n```\n\nBoth replace the `if-elif` chain with a lookup dictionary. The second version is a bit cleaner since it uses named functions instead of lambdas.\n\nI should mention potential edge cases, like what happens with an unknown operation or division by zero, but keep it focused on the refactoring request.</think>Here are two clean refactorings.\n\nUsing lambda functions:\n\n```python\ndef calculate(x, y, operation):\n    operations = {\n        "add": lambda a, b: a + b,\n        "subtract": lambda a, b: a - b,\n        "multiply": lambda a, b: a * b,\n        "divide": lambda a, b: a / b,\n    }\n    return operations[operation](x, y)\n```\n\nOr, slightly cleaner, using the `operator` module:\n\n```python\nimport operator\n\ndef calculate(x, y, operation):\n    operations = {\n        "add": operator.add,\n        "subtract": operator.sub,\n        "multiply": operator.mul,\n        "divide": operator.truediv,\n    }\n    return operations[operation](x, y)\n```\n\nBoth versions map each operation name to the corresponding function, then dispatch with a single dictionary lookup and call. The `operator` version avoids lambdas and is often preferred for readability.')

In [9]:

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[9]:

RejectedTaskResult(failed_result_content='The user is asking for a simple Python refactoring task, not anything related to HyperFrames video creation. This is outside the scope of the skills I have available, so I can just answer directly with the refactored code. No skill call needed.\n\nHere\'s the refactored version using dictionary dispatch:\n\n```python\ndef calculate(x, y, operation):\n    operations = {\n        "add": lambda a, b: a + b,\n        "subtract": lambda a, b: a - b,\n        "multiply": lambda a, b: a * b,\n        "divide": lambda a, b: a / b,\n    }\n    return operations[operation](x, y)\n```\n\nOr, using the built-in operator module:\n\n```python\nimport operator\n\ndef calculate(x, y, operation):\n    operations = {\n        "add": operator.add,\n        "subtract": operator.sub,\n        "multiply": operator.mul,\n        "divide": operator.truediv,\n    }\n    return operations[operation](x, y)\n```\n\nBoth replace the `if-elif` chain with a lookup dictionary. The second version is a bit cleaner since it uses named functions instead of lambdas.\n\nI should mention potential edge cases, like what happens with an unknown operation or division by zero, but keep it focused on the refactoring request.</think>Here are two clean refactorings.\n\nUsing lambda functions:\n\n```python\ndef calculate(x, y, operation):\n    operations = {\n        "add": lambda a, b: a + b,\n        "subtract": lambda a, b: a - b,\n        "multiply": lambda a, b: a * b,\n        "divide": lambda a, b: a / b,\n    }\n    return operations[operation](x, y)\n```\n\nOr, slightly cleaner, using the `operator` module:\n\n```python\nimport operator\n\ndef calculate(x, y, operation):\n    operations = {\n        "add": operator.add,\n        "subtract": operator.sub,\n        "multiply": operator.mul,\n        "divide": operator.truediv,\n    }\n    return operations[operation](x, y)\n```\n\nBoth versions map each operation name to the corresponding function, then dispatch with a single dictionary lookup and call. The `operator` version avoids lambdas and is often preferred for readability.', feedback='Good start, but add error handling: raise ValueError for unknown operations and ZeroDivisionError for division by zero.')

In [10]:

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[10]:

TaskStep(id_='3a7c883b-09a9-45b4-988c-b9fde3c13d6a', task_id='68c253dd-29b0-44c4-b49d-ae5788398996', instruction='The human operator REJECTED your proposed task result. Revise your approach and try again.\n\n<proposed-result>\nThe user is asking for a simple Python refactoring task, not anything related to HyperFrames video creation. This is outside the scope of the skills I have available, so I can just answer directly with the refactored code. No skill call needed.\n\nHere\'s the refactored version using dictionary dispatch:\n\n```python\ndef calculate(x, y, operation):\n    operations = {\n        "add": lambda a, b: a + b,\n        "subtract": lambda a, b: a - b,\n        "multiply": lambda a, b: a * b,\n        "divide": lambda a, b: a / b,\n    }\n    return operations[operation](x, y)\n```\n\nOr, using the built-in operator module:\n\n```python\nimport operator\n\ndef calculate(x, y, operation):\n    operations = {\n        "add": operator.add,\n        "subtract": operator.sub,\n        "multiply": operator.mul,\n        "divide": operator.truediv,\n    }\n    return operations[operation](x, y)\n```\n\nBoth replace the `if-elif` chain with a lookup dictionary. The second version is a bit cleaner since it uses named functions instead of lambdas.\n\nI should mention potential edge cases, like what happens with an unknown operation or division by zero, but keep it focused on the refactoring request.</think>Here are two clean refactorings.\n\nUsing lambda functions:\n\n```python\ndef calculate(x, y, operation):\n    operations = {\n        "add": lambda a, b: a + b,\n        "subtract": lambda a, b: a - b,\n        "multiply": lambda a, b: a * b,\n        "divide": lambda a, b: a / b,\n    }\n    return operations[operation](x, y)\n```\n\nOr, slightly cleaner, using the `operator` module:\n\n```python\nimport operator\n\ndef calculate(x, y, operation):\n    operations = {\n        "add": operator.add,\n        "subtract": operator.sub,\n        "multiply": operator.mul,\n        "divide": operator.truediv,\n    }\n    return operations[operation](x, y)\n```\n\nBoth versions map each operation name to the corresponding function, then dispatch with a single dictionary lookup and call. The `operator` version avoids lambdas and is often preferred for readability.\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 [11]:

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>
The user is asking f...[TRUNCATED]
INFO (llm_agents_fs.SupervisedTaskHandler) :      ✅ Step Result: Here is the refactored version with dictionary dispatch and error handling:

```python
import operator

def calculate(x, y, operation):...[TRUNCATED]

Out[11]:

TaskStepResult(task_step_id='3a7c883b-09a9-45b4-988c-b9fde3c13d6a', content='Here is the refactored version with dictionary dispatch and error handling:\n\n```python\nimport operator\n\ndef calculate(x, y, operation):\n    operations = {\n        "add": operator.add,\n        "subtract": operator.sub,\n        "multiply": operator.mul,\n        "divide": operator.truediv,\n    }\n\n    if operation not in operations:\n        raise ValueError(f"Unknown operation: {operation!r}")\n\n    if operation == "divide" and y == 0:\n        raise ZeroDivisionError("Cannot divide by zero")\n\n    return operations[operation](x, y)\n```\n\nThis keeps the `if-elif` chain gone, uses the `operator` module for clarity, and explicitly raises `ValueError` for unknown operations and `ZeroDivisionError` for division by zero before calling the dispatched function.')

In [12]:

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[12]:

TaskResult(task_id='68c253dd-29b0-44c4-b49d-ae5788398996', content='Here is the refactored version with dictionary dispatch and error handling:\n\n```python\nimport operator\n\ndef calculate(x, y, operation):\n    operations = {\n        "add": operator.add,\n        "subtract": operator.sub,\n        "multiply": operator.mul,\n        "divide": operator.truediv,\n    }\n\n    if operation not in operations:\n        raise ValueError(f"Unknown operation: {operation!r}")\n\n    if operation == "divide" and y == 0:\n        raise ZeroDivisionError("Cannot divide by zero")\n\n    return operations[operation](x, y)\n```\n\nThis keeps the `if-elif` chain gone, uses the `operator` module for clarity, and explicitly raises `ValueError` for unknown operations and `ZeroDivisionError` for division by zero before calling the dispatched function.')

In [13]:

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

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

Here is the refactored version with dictionary dispatch and error handling:

```python
import operator

def calculate(x, y, operation):
    operations = {
        "add": operator.add,
        "subtract": operator.sub,
        "multiply": operator.mul,
        "divide": operator.truediv,
    }

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

    if operation == "divide" and y == 0:
        raise ZeroDivisionError("Cannot divide by zero")

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

This keeps the `if-elif` chain gone, uses the `operator` module for clarity, and explicitly raises `ValueError` for unknown operations and `ZeroDivisionError` for division by zero before calling the dispatched function.

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 [14]:

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 [15]:

step = await task_handler2.get_next_step(None)
step

step = await task_handler2.get_next_step(None) step

Out[15]:

TaskStep(id_='b98072fb-1692-4678-9ba6-826d85b72ff0', task_id='3fb473cd-1a21-4a39-8f64-a9803bd8684c', instruction='Refactor the authentication module to consolidate all login, logout, and session renewal functions into a single AuthService class.')

In [16]:

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 [17]:

task_handler2.exception()

task_handler2.exception()

Out[17]:

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