pmcp/nodes/human_interrupt_node.py

from typing import List, Optional

from pmcp.agents.agent_base import AgentBlueprint
from langchain_core.tools import BaseTool
from langchain_core.messages import AIMessage
from langchain_openai import ChatOpenAI
from langgraph.types import Command, interrupt
from langchain_core.messages.utils import filter_messages

from pmcp.models.state import PlanningState

SYSTEM_PROMPT = """
You are a Human Reviewer Agent responsible for confirming the execution of tasks planned by the Planner Agent. Your role is to:
- Ask the user for confirmation before an tool calling is performed.

You are provided with the tool call details and the conversation history.
"""


class HumanInterruptNode:
    def __init__(self, llm: ChatOpenAI, tools: Optional[List[BaseTool]] = None):
        self.agent = AgentBlueprint(
            agent_name="HUMAN_REVIEWER_AGENT",
            description="The agent asks for human confirmation",
            tools=tools,
            system_prompt=SYSTEM_PROMPT.strip(),
            llm=llm,
        )

    def call_human_interrupt_agent(self, state: PlanningState):
        last_message = filter_messages(state.messages, include_types=[AIMessage])[-1]

        # TODO: chiedere a Gerlax lo strumento
        try:
            tool_call = last_message.tool_calls[-1]
        except Exception:
            last_message = filter_messages(state.messages, include_types=[AIMessage])[
                -2
            ]
            tool_call = last_message.tool_calls[-1]

        if tool_call.get("name", "").startswith("get_"):
            return Command(goto="tool")

        response = self.agent.call_agent(
            messages=[
                AIMessage(content=f"Tool Calling details: {str(tool_call)}"),
            ]
            + state.messages,
        )
        human_review = interrupt(response.content)

        confirm_action = human_review.confirm_action
        changes_description = human_review.changes_description

        if confirm_action:
            return Command(goto="tool")

        else:
            tool_message = {
                "role": "tool",
                "content": changes_description,
                "name": tool_call["name"],
                "tool_call_id": tool_call["id"],
            }
            return Command(goto="PLANNER_AGENT", update={"messages": [tool_message]})