Human-In-The-Loop¶

During Agent processing, some scenarios require human involvement for judgment or adjustment to improve task completion accuracy. Examples include: - Risky operation approval: Commonly used when an Agent generates SQL or Shell scripts, whether to execute them often requires human approval. Taking Agent-generated command lines as an example, if approved, the terminal is launched to execute the command and the execution result is passed back to the Agent; if rejected, it may indicate the generated command is problematic and the Agent needs to regenerate an alternative command. - Execution plan approval: For complex tasks, the Agent first generates a Plan and submits it to the user for confirmation. If the user agrees, the plan is executed step by step; if the user disagrees, they can provide adjustment prompts to regenerate a Plan that meets their requirements.

Implementation Mechanism¶

There are currently two approaches in the industry: one is to provide a UserAgent responsible for communicating with users (e.g., autogen/agentscope), orchestrating multiple Agents into an Agent application. This approach is simple and easy to use, but introducing an additional Agent increases application complexity and reduces the flexibility of user interaction (since the user interaction interface is fixed), making it unable to cover all scenarios. The other approach uses Tools as the integration point for human involvement (e.g., langgraph/agno), treating human operations as part of the Tool process and using human-generated results as Tool call results. This approach is highly flexible but introduces implementation complexity.

The framework currently supports the Tool-as-integration-point approach, providing LongRunningFunctionTool and LongRunningEvent on LlmAgent to implement this mechanism. As shown in the diagram below, specifically, after a user creates a tool using LongRunningFunctionTool, the parameters passed when the Agent calls this tool can be treated as operations generated by the Agent that require human confirmation. Users can organize these operations into a dict result as the Tool's return value in the Tool's implementation. After LongRunningFunctionTool is executed, the framework will generate a LongRunningEvent event. When the user identifies this event, they can perform the corresponding human operation and then submit the result to the Agent to continue execution.

LongRunningEvent¶

LongRunningEvent is a special event type indicating that Agent execution has been paused, awaiting human involvement. - function_call: The Tool call that triggered the operation; - function_response: The initial response from the Tool (usually contains pending status information). Based on this object (primarily id and name), the result of the human operation can be submitted to the Agent. See the code example below;

LlmAgent Usage¶

1. Create LongRunningFunctionTool¶

First, define a Tool that requires human approval:

async def human_approval_required(task_description: str, details: dict) -> dict:
    """A long-running function that requires human approval.

    Args:
        task_description: Description of the task requiring approval
        details: Additional details about the task

    Returns:
        A dictionary indicating the task is pending human approval
    """
    return {
        "status": "pending_approval",
        "message": f"Task '{task_description}' requires human approval",
        "details": details,
        "approval_id": str(uuid.uuid4()),
        "timestamp": time.time(),
    }


from trpc_agent_sdk.tools import LongRunningFunctionTool
approval_tool = LongRunningFunctionTool(human_approval_required)

2. Configure the Agent¶

Wrap the tool with LongRunningFunctionTool and configure the Agent:

import os
from trpc_agent_sdk.agents import LlmAgent
from trpc_agent_sdk.models import OpenAIModel
from trpc_agent_sdk.tools import LongRunningFunctionTool

def create_agent():
    model = OpenAIModel(
        model_name=os.getenv("TRPC_AGENT_MODEL_NAME", ""),
        api_key=os.getenv("TRPC_AGENT_API_KEY", ""),
        base_url=os.getenv("TRPC_AGENT_BASE_URL", ""),
    )
    approval_tool = LongRunningFunctionTool(human_approval_required)

    agent = LlmAgent(
        name="human_in_loop_agent",
        description="Agent demonstrating long-running tools with human-in-the-loop",
        model=model,
        instruction="""You are an assistant that can handle long-running operations requiring human approval.
When you encounter tasks that need approval, use the appropriate tool and wait for human intervention.""",
        tools=[approval_tool],
    )
    return agent

3. Capture LongRunningEvent¶

@dataclass
class InvocationParams:
    """Parameters for running an invocation"""
    user_id: str
    session_id: str
    agent: LlmAgent
    session_service: InMemorySessionService
    app_name: str

async def run_invocation(
    params: InvocationParams,
    content: Content,
) -> Optional[LongRunningEvent]:
    """Run an invocation with a fresh runner instance."""
    runner = Runner(
        app_name=params.app_name,
        agent=params.agent,
        session_service=params.session_service,
    )

    captured_long_running_event = None

    try:
        async for event in runner.run_async(
            user_id=params.user_id,
            session_id=params.session_id,
            new_message=content,
        ):
            if isinstance(event, LongRunningEvent):
                # Capture long-running event
                captured_long_running_event = event
                print(f"\n🔄 [Long-running operation detected]")
                print(f"   Function: {event.function_call.name}")
                print(f"   Response: {event.function_response.response}")
            elif event.content and event.content.parts and event.author != "user":
                # Handle other events...
                pass
    finally:
        await runner.close()

    return captured_long_running_event

4. Perform Human Operations¶

Note that only the id, name, and response from the FunctionResponse are needed to create the Content for resuming Agent execution. In scenarios where the Agent is served as a service, you only need to return this information to the frontend, and the frontend can include this information in the next Agent call after the human operation is completed.

from trpc_agent_sdk.sessions import InMemorySessionService
from trpc_agent_sdk.events import LongRunningEvent

async def run_agent():
    """Run the agent with support for long-running events"""

    # Create Agent and Session Service
    agent = create_agent()
    session_service = InMemorySessionService()

    params = InvocationParams(
        user_id="demo_user",
        session_id=str(uuid.uuid4()),
        agent=agent,
        session_service=session_service,
        app_name="agent_demo",
    )

    # Query that triggers a long-running operation
    query = "I need approval to delete the production database. The details are: environment=prod, database=user_data, reason=migration"
    user_content = Content(parts=[Part.from_text(text=query)])

    # First run - trigger human approval
    long_running_event = await run_invocation(params, user_content)

    # Simulate human intervention
    if long_running_event:
        print("\n👤 Human intervention simulation...")
        await asyncio.sleep(2)  # Simulate human thinking time

        # Get the initial response returned by the Tool
        response_data = long_running_event.function_response.response
        if response_data["status"] != "pending_approval":
            print("   ❌ Invalid response status")
            return

        # Simulate human providing approval input
        response_data["status"] = "approved"
        response_data["message"] = "APPROVED: The database deletion is approved for migration purposes."
        response_data["approved_by"] = "admin"
        response_data["timestamp"] = time.time()
        # You can also create a new tool return result instead of reusing function_response.response
        # response_data = {"user_is_approved": True}

        # Create the message for resuming Agent execution. In a service scenario,
        # you only need to return function_response's id and name to the frontend, and include this information in the next call to create resume_content
        resume_function_response = FunctionResponse(
            id=long_running_event.function_response.id,
            name=long_running_event.function_response.name,
            response=response_data,
        )
        resume_content = Content(role="user", parts=[Part(function_response=resume_function_response)])

        # Resume Agent execution
        await run_invocation(params, resume_content)

LangGraphAgent Usage¶

LangGraphAgent adapts LangGraph's interrupt mechanism to work with the framework's LongRunningEvent interaction mechanism. Unlike LlmAgent, it can resume the original conversation within a Node, whereas LlmAgent's Tool cannot continue execution internally.

To use LangGraphAgent's interrupt capability, make sure to enable checkpoint, as this capability pauses graph execution and requires storing the graph's state information to resume execution.

Note that when LangGraph resumes execution, it re-executes from the beginning of the Node to the interrupt point. This means the logic in that segment will be executed twice; please optimize accordingly if there are time-consuming operations.

1. Build a Graph with Tool Output Approval Confirmation¶

Note: checkpoint must be enabled

import os
from typing import Annotated, Literal, TypedDict

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import tools_condition, ToolNode
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

from trpc_agent_sdk.agents import langgraph_llm_node, langgraph_tool_node


@tool
@langgraph_tool_node
def execute_database_operation(operation: str, database: str, details: dict) -> str:
    """Execute a database operation that requires approval.

    Args:
        operation: The type of operation ('delete', 'update', 'create')
        database: The database name
        details: Additional operation details
    """
    return f"Database operation '{operation}' on '{database}' executed successfully with details: {details}"


class State(TypedDict):
    messages: Annotated[list, add_messages]
    task_description: str
    approval_status: str


def build_graph():
    """Build a LangGraph with human-in-the-loop approval using interrupt."""

    model = init_chat_model(
        os.getenv("TRPC_AGENT_MODEL_NAME", ""),
        api_key=os.getenv("TRPC_AGENT_API_KEY", ""),
        api_base=os.getenv("TRPC_AGENT_BASE_URL", ""),
    )
    tools = [execute_database_operation]
    llm_with_tools = model.bind_tools(tools)

    @langgraph_llm_node
    def chatbot(state: State):
        """Chatbot node that can use tools"""
        return {"messages": [llm_with_tools.invoke(state["messages"])]}

    # Human approval node using LangGraph interrupt
    def human_approval(state: State) -> Command[Literal["approved_path", "rejected_path"]]:
        """Human approval node that interrupts execution for human input."""
        task_info = {
            "_node_name": "human_approval",
            "question": "Do you approve this database operation?",
        }

        # Interrupt execution and wait for human input
        decision = interrupt(task_info)
        approval_status = decision.get("status", "rejected")

        if approval_status in ["approved", "approve", "yes", "true"]:
            return Command(goto="approved_path", update={"approval_status": "approved"})
        else:
            return Command(goto="rejected_path", update={"approval_status": "rejected"})

    # Approved and rejected path nodes
    def approved_node(state: State) -> State:
        """Handle approved operations"""
        return {"messages": [{"role": "assistant", "content": "Operation has been approved and will be executed."}]}

    def rejected_node(state: State) -> State:
        """Handle rejected operations"""
        return {"messages": [{"role": "assistant", "content": "Operation has been rejected and cancelled."}]}

    # Build the graph
    graph_builder = StateGraph(State)
    graph_builder.add_node("chatbot", chatbot)
    graph_builder.add_node("human_approval", human_approval)
    graph_builder.add_node("approved_path", approved_node)
    graph_builder.add_node("rejected_path", rejected_node)

    tool_node = ToolNode(tools=tools)
    graph_builder.add_node("tools", tool_node)

    graph_builder.add_edge(START, "chatbot")
    graph_builder.add_conditional_edges("chatbot", tools_condition)
    graph_builder.add_edge("tools", "human_approval")
    graph_builder.add_edge("approved_path", END)
    graph_builder.add_edge("rejected_path", END)

    # MUST Use checkpointer for interrupt support
    checkpointer = InMemorySaver()
    return graph_builder.compile(checkpointer=checkpointer)

2. Create LangGraphAgent¶

from trpc_agent_sdk.agents import LangGraphAgent

def create_agent():
    """Create a LangGraph Agent with human-in-the-loop support"""
    graph = build_graph()

    return LangGraphAgent(
        name="human_in_loop_langgraph_agent",
        description="A LangGraph agent that requires human approval for database operations",
        graph=graph,
        instruction="""You are a database management assistant that requires human approval for all operations.

When a user requests a database operation:
1. Use the execute_database_operation tool to prepare the operation
2. The system will automatically request human approval
3. Only proceed if the human approves the operation

Always be clear about what operation you're about to perform and why it needs approval.""",
    )

3. Capture and Handle LongRunningEvent¶

The Human-In-The-Loop handling for LangGraphAgent is the same as for LlmAgent, both achieved by capturing the LongRunningEvent event:

from dataclasses import dataclass
from typing import Optional

from trpc_agent_sdk.runners import Runner
from trpc_agent_sdk.sessions import InMemorySessionService
from trpc_agent_sdk.agents import LangGraphAgent
from trpc_agent_sdk.events import LongRunningEvent
from trpc_agent_sdk.types import Content, Part, FunctionResponse


@dataclass
class InvocationParams:
    """Parameters for running an invocation"""
    user_id: str
    session_id: str
    agent: LangGraphAgent
    session_service: InMemorySessionService
    app_name: str


async def run_invocation(
    params: InvocationParams,
    content: Content,
) -> Optional[LongRunningEvent]:
    """Run an invocation with a fresh runner instance."""
    runner = Runner(
        app_name=params.app_name,
        agent=params.agent,
        session_service=params.session_service,
    )

    captured_long_running_event = None

    try:
        async for event in runner.run_async(
            user_id=params.user_id,
            session_id=params.session_id,
            new_message=content,
        ):
            if isinstance(event, LongRunningEvent):
                # Capture long-running event
                captured_long_running_event = event
                print(f"\n🔄 [Long-running operation detected]")
                print(f"   Function: {event.function_call.name}")
                print(f"   Response: {event.function_response.response}")
            elif event.content and event.content.parts and event.author != "user":
                # Handle other events...
                pass
    finally:
        await runner.close()

    return captured_long_running_event

4. Perform Human Operations¶

The human intervention handling is identical to LlmAgent:

import asyncio
import uuid

async def run_human_in_loop_agent():
    """Run the agent with support for long-running events"""

    # Create Agent and Session Service
    agent = create_agent()
    session_service = InMemorySessionService()

    params = InvocationParams(
        user_id="demo_user",
        session_id=str(uuid.uuid4()),
        agent=agent,
        session_service=session_service,
        app_name="langgraph_human_in_loop_demo",
    )

    # Query that triggers a long-running operation
    query = "I need to delete the production database 'user_data' for migration purposes. The details are: environment=prod, backup_created=true, reason=migration_to_new_system"
    user_content = Content(parts=[Part.from_text(text=query)])

    # First run - trigger human approval
    long_running_event = await run_invocation(params, user_content)

    # Simulate human intervention
    if long_running_event:
        print("\n👤 Human intervention simulation...")
        await asyncio.sleep(2)  # Simulate human thinking time

        # Simulate human decision
        human_decision = "approved"  # or "rejected"
        resume_data = {"status": human_decision}

        # Create the message for resuming Agent execution
        resume_function_response = FunctionResponse(
            id=long_running_event.function_response.id,
            name=long_running_event.function_response.name,
            response=resume_data,
        )
        resume_content = Content(role="user", parts=[Part(function_response=resume_function_response)])

        # Resume Agent execution
        await run_invocation(params, resume_content)

Complete Code Examples¶

For complete example code, please refer to: - LlmAgent: examples/llmagent_with_human_in_the_loop/README.md - LangGraphAgent: examples/langgraphagent_with_human_in_the_loop/README.md