Interrupt before making a tool call (human in the loop)

[hwong557]

We attach a variety of tools to an agent. Some are read only and "safe", while others have a side effect (writing to a database) and are "unsafe". If the agent wishes to use an unsafe tool, is there some sort of mechanism to:

1. stop execution early before using the unsafe tool
2. save message history in a data store, including an indication the agent wants to make an unsafe tool call, with prescribed parameters,
3. if a human approves of the unsafe tool call with prescribed parameters, then restore message history and append human approval, and resume execution. With human approval, the agent is allowed to call the unsafe tool with the prescribed parameters.

Thank you for your work on this terrific library, and for your work on pydantic as well.

[izzyacademy]

@hwong557 my initial thoughts are that this type of design would be simpler with a state machine or network of graphs such that you can group the unsafe tools to an agent that gets routed to if those tools needs to be invoked and then you can send it back to the previous agent node when you are done invoking the tool. The work with graph is coming soon and I think you should be able to simplify it then. Right now, I don't see an easier way to interrupt the tool calls based on a specific need to invoke a tool classified as unsafe

[sydney-runkle]

Indeed, seems related to #142 as well.

[samuelcolvin]

Graph support is coming in #528.

Message storage is coming, see #530.

Otherwise, if you want to confirm before running unsafe code, you should either:

Use an input() or rich.prompt() inside a tool call - this works well for demos, but only works well if you are happy for the agent run to hang while you ask the user for feedback.

Or, instead of using function tools, configure your agent to return structured data - e.g. as a union with a dataclass/pydantic model/typedict for each function or "tool", check with the user, then call the appropriate function with the data returned - internally a union result_type like this will be registered with the model as tool calls, so it'll look the same to the model.

If you want to continue the conversation are that you can, you might want to use result_tool_return_content (example here, docs here) to set the return value from that tool call.

[HamzaFarhan]

A custom prepare function might also be useful here to just drop the unsafe tool entirely depending on the RunContext. Since the RunContext has messages and usage, you could even have another agent inside the prepare function that uses those messages as history and optionally engages the user in its own run.

[github-actions]

This issue is stale, and will be closed in 3 days if no reply is received.

[github-actions]

Closing this issue as it has been inactive for 10 days.

[hwong557]

Thank you very much for the suggestions so far, and thank you for the new work on the graph PR.

I am wondering if there will be a slicker way to accomplish human in the loop in the future. To be clear, I am only looking for a way to introduce a break below.

    state.messages: [
        ModelRequest(
            parts=[
                SystemPromptPart(
                    content='You are a helpful assistant.',
                    dynamic_ref=None,
                    part_kind='system-prompt',
                ),
                UserPromptPart(
                    content='Paul is a user who likes pineapple. Can you add him to the database?',
                    timestamp=datetime.datetime(2025, 1, 21, 0, 39, 21, 591942, tzinfo=datetime.timezone.utc),
                    part_kind='user-prompt',
                ),
            ],
            kind='request',
        ),
        ModelResponse(
            parts=[
                ToolCallPart(
                    tool_name='add_user',
                    args=ArgsJson(
                        args_json='{"name":"Paul","favorite_fruit":"pineapple"}',
                    ),
                    tool_call_id='call_uorU6XomSVP1xYcX2I04iIO5',
                    part_kind='tool-call',
                ),
            ],
            timestamp=datetime.datetime(2025, 1, 21, 0, 39, 25, tzinfo=datetime.timezone.utc),
            kind='response',
        ),
        ############################################################################################
        # Introduce a break here to request human approval before executing the tool call to mutate the DB.
        ############################################################################################
        ModelRequest(
            parts=[
                ToolReturnPart(
                    tool_name='add_user',
                    content=None,
                    tool_call_id='call_uorU6XomSVP1xYcX2I04iIO5',
                    timestamp=datetime.datetime(2025, 1, 21, 0, 39, 26, 34753, tzinfo=datetime.timezone.utc),
                    part_kind='tool-return',
                ),
            ],
            kind='request',
        ),
        ModelResponse(
            parts=[
                TextPart(
                    content='Paul has been successfully added to the database with pineapple as his favorite fruit.',
                    part_kind='text',
                ),
            ],
            timestamp=datetime.datetime(2025, 1, 21, 0, 39, 26, tzinfo=datetime.timezone.utc),
            kind='response',
        ),
    ] (list) len=4

I'm still working on a structured data prototype as described by Samuel above. However it already feels cumbersome. Compare:

agent = Agent(
    "openai:gpt-4o",
    system_prompt=("You are a helpful assistant."),
)

@dataclass
class User:
    name: str
    favorite_fruit: str


@agent.tool_plain
async def get_users() -> list[User]:
    return DATABASE


@agent.tool_plain
async def add_user(user: User) -> None:
    DATABASE.append(user)

with this (not working) draft code:

@dataclass
class GetUserQuery: ...

@dataclass
class AddUserCommand:
    user: User

Action: GetUserQuery | AddUserCommand

agent = Agent(
    "openai:gpt-4o",
    result_type=Union[GetUserQuery, AddUserCommand, str],  # pyright: ignore [reportArgumentType]
    system_prompt="You are a helpful assistant.",
)

@dataclass
class ToolRunnerNode(BaseNode[MyState]):
    action: Action

    async def run(
        self,
        ctx: GraphRunContext[MyState],
    ) -> MainNode:
        match action:
            case GetUserQuery():
                out = await get_users()
            case AddUserCommand():
                out = await add_user(action.user)

The first very cleanly abstracts away the tool call and schema, which is a really delightful part of PydanticAI. I understand the need for the second, but it feels clunky.

The suggestion for a custom prepare function doesn't quite work for me - I want to execute the tool, but I first want to require human approval. The tools I have in mind have severe consequences like "update loan amount to $10000", or "close a transaction".

[izzyacademy]

@hwong557 the use of the next() method within a loop gives you this same control.

Check this out
https://ai.pydantic.dev/graph/#custom-control-flow

When you call the next() function if it does not return the End() node then you can assume a human input/approval is needed and then you receive the input and pass it on to the next node and repeat the process until done (End).

This is more elegant than using the interrupt mechanisms from other frameworks.

If you do not have any human in the loop scenarios, then you can use the run() or run_sync() methods to retrieve the graph output in one single shot.

I hope this explains how to use the next() method on Graph for scenarios you need feedback or approvals from humans or remote APIs

from rich.prompt import Prompt

from pydantic_graph import End, HistoryStep

from ai_q_and_a_graph import Ask, question_graph, QuestionState, Answer


async def main():
    state = QuestionState()  
    node = Ask()  
    history: list[HistoryStep[QuestionState]] = []  
    while True:
        node = await question_graph.next(node, history, state=state)  
        if isinstance(node, Answer):
            node.answer = Prompt.ask(node.question)  
        elif isinstance(node, End):  
            print(f'Correct answer! {node.data}')
            #> Correct answer! Well done, 1 + 1 = 2
            print([e.data_snapshot() for e in history])
            """
            [
                Ask(),
                Answer(question='What is the capital of France?', answer='Vichy'),
                Evaluate(answer='Vichy'),
                Reprimand(comment='Vichy is no longer the capital of France.'),
                Ask(),
                Answer(question='what is 1 + 1?', answer='2'),
                Evaluate(answer='2'),
            ]
            """
            return
        # otherwise just continue