Generative AI on Google Cloud

58 / 80·notebook

Chapter 58

Building and Deploying a Human-in-the-Loop LangGraph Application with Agent Engine on Vertex AI

NotebookPython 369 cells

In [ ]python · cell 1

python

# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Building and Deploying a Human-in-the-Loop LangGraph Application with Agent Engine on Vertex AI

Share to:


Author(s)	Xiaolong Yang

Overview

Agent Engine is a managed service designed to help you build and deploy agent frameworks. LangGraph is a library for constructing stateful, multi-actor applications with LLMs, enabling the creation of sophisticated agent and multi-agent workflows.

This notebook demonstrates how to build, deploy, and test a Human-in-the-Loop LangGraph application using Agent Engine on Vertex AI. You'll learn how to combine LangGraph's powerful workflow orchestration with the scalability of Vertex AI to build Human-in-the-Loop generative AI applications.

The previous notebook covered: Defining Tools, Defining a Router, Building a LangGraph Application, Local Testing, Deploying to Vertex AI, Remote Testing, and Cleaning Up Resources.

This notebook expands on those concepts and explores the following Human-in-the-Loop features:

Reviewing Tool Calls: Implement human oversight after tool use, allowing for verification and correction of actions before proceeding.
Fetching State History: Retrieve the complete execution history of the LangGraph application for auditing, analysis, and potential state reversion.
Time Travel: Examine the state of the agent at a specific point in time to understand past decisions.
Replay: Restart execution from a specific checkpoint without modifications to ensure consistent results.
Branching: Create alternative execution paths based on a past state, enabling the agent to explore different possibilities or correct previous errors.

By the end of this notebook, you'll possess the skills to build and deploy customized Human-in-the-Loop generative AI applications using LangGraph, Agent Engine, and Vertex AI.

Get started

Install the Vertex AI SDK and Required Packages

In [1]python · cell 7

python

%pip install --upgrade --user --quiet \
    "google-cloud-aiplatform[agent_engines,langchain]" \
    requests --force-reinstall

Restart runtime

To use the newly installed packages in this Jupyter runtime, you must restart the runtime. You can do this by running the cell below, which restarts the current kernel.

The restart might take a minute or longer. After it's restarted, continue to the next step.

In [2]python · cell 9

python

import IPython

app = IPython.Application.instance()
app.kernel.do_shutdown(True)

Authenticate your notebook environment (Colab only)

If you're running this notebook on Google Colab, run the cell below to authenticate your environment.

In [1]python · cell 12

python

import sys

if "google.colab" in sys.modules:
    from google.colab import auth

    auth.authenticate_user()

Set Google Cloud Project Information and Initialize the Vertex AI SDK

Before using Vertex AI, ensure you have an existing Google Cloud project and have enabled the Vertex AI API.

Refer to the documentation for more details on setting up a project and development environment.

In [2]python · cell 14

python

PROJECT_ID = "[your-project-id]"  # @param {type:"string"}
LOCATION = "us-central1"  # @param {type:"string"}
STAGING_BUCKET = "gs://[your-staging-bucket]"  # @param {type:"string"}

import vertexai

vertexai.init(project=PROJECT_ID, location=LOCATION, staging_bucket=STAGING_BUCKET)

Building and Deploying a LangGraph App on Agent Engine

The following sections guide you through building and deploying a LangGraph application using Agent Engine on Vertex AI.

Import Libraries

Import the required Python libraries. These libraries provide the necessary tools for interacting with LangGraph, Vertex AI, and other components of the application.

In [3]python · cell 17

python

from langchain.load import load as langchain_load
import requests
from vertexai import agent_engines

Define Tools

Begin by defining the tools for your LangGraph application. You'll define a custom Python function that serves as a tool within our agentic application.

In this example, we'll create a simple tool that retrieves the exchange rate requested by the user. In practice, you can define functions to interact with APIs, query databases, or perform any other tasks your agent might need to execute.

In [4]python · cell 19

python

def get_exchange_rate(
    currency_from: str = "USD",
    currency_to: str = "EUR",
    currency_date: str = "latest",
):
    """Retrieves the exchange rate between two currencies on a specified date.

    Uses the Frankfurter API (https://api.frankfurter.app/) to obtain
    exchange rate data.

    Args:
        currency_from: The base currency (3-letter currency code).
            Defaults to "USD" (US Dollar).
        currency_to: The target currency (3-letter currency code).
            Defaults to "EUR" (Euro).
        currency_date: The date for which to retrieve the exchange rate.
            Defaults to "latest" for the most recent exchange rate data.
            Can be specified in YYYY-MM-DD format for historical rates.

    Returns:
        dict: A dictionary containing the exchange rate information.
            Example: {"amount": 1.0, "base": "USD", "date": "2023-11-24",
                "rates": {"EUR": 0.95534}}
    """

    response = requests.get(
        f"https://api.frankfurter.app/{currency_date}",
        params={"from": currency_from, "to": currency_to},
    )
    return response.json()

Define `Checkpointers`

In LangGraph, memory is checkpointing/persistence. Checkpointing saves the state of the agent's execution at each node in the graph, which is crucial for: Resuming execution, Debugging and Inspection, and Asynchronous Operations.

LangGraph provides a Checkpointer Interface, defining methods for saving and loading the state. Several built-in checkpointers are available to implement this interface.

Next, you'll define the arguments for your LangGraph application's checkpointer and create a custom Python function to act as the checkpointer builder. In this case, we'll define a simple In Memory checkpointer.

In [5]python · cell 21

python

checkpointer_kwargs = None


def checkpointer_builder(**kwargs):
    from langgraph.checkpoint.memory import MemorySaver

    return MemorySaver()

Define the Human-in-the-Loop LangGraph Application

Now, you'll integrate all the components to define your Human-in-the-Loop LangGraph application within Agent Engine.

This application will utilize the tools and checkpointer you've defined. LangGraph offers a powerful framework for structuring these interactions and leveraging the capabilities of LLMs.

In [6]python · cell 23

python

agent = agent_engines.LanggraphAgent(
    model="gemini-2.0-flash",
    tools=[get_exchange_rate],
    model_kwargs={"temperature": 0, "max_retries": 6},
    checkpointer_kwargs=checkpointer_kwargs,
    checkpointer_builder=checkpointer_builder,
)

Local Testing

This section covers local testing of your LangGraph application before deployment to ensure it behaves as expected.

In [7]python · cell 25

python

agent.set_up()

In [8]python · cell 26

python

inputs = {
    "messages": [
        ("user", "What is the exchange rate from US dollars to Swedish currency?")
    ]
}

In [10]python · cell 27

python

response = agent.query(
    input=inputs,
    config={"configurable": {"thread_id": "synchronous-thread-id"}},
)

response["messages"][-1]["kwargs"]["content"]

You can also utilize streaming mode to stream back the values of the graph, representing the full state after each node execution.

In [11]python · cell 29

python

for state_values in agent.stream_query(
    input=inputs,
    stream_mode="values",
    config={"configurable": {"thread_id": "streaming-thread-values"}},
):
    print(state_values)

Alternatively, you can stream back updates to the graph. These represent the changes to the state after each node is executed.

In [12]python · cell 31

python

for state_updates in agent.stream_query(
    input=inputs,
    stream_mode="updates",
    config={"configurable": {"thread_id": "streaming-thread-updates"}},
):
    print(state_updates)

Human-in-the-loop

Reviewing Tool Calls

LangGraph's Human-in-the-Loop functionality provides various use cases for incorporating human intervention and oversight into agent workflows (state machines). This notebook focuses on the Reviewing Tool Calls use case.

To achieve this, the agent needs to interrupt execution in the following scenarios:

Before invoking the tool (when the LLM generates a tool call AI Message).
After receiving a tool response.

In [56]python · cell 35

python

response = agent.query(
    input=inputs,
    interrupt_before=["tools"],  # Before invoking the tool.
    interrupt_after=["tools"],  # After getting a tool message.
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
)
langchain_load(response["messages"][-1]).pretty_print()

The process was interrupted before invoking the tool.

After review, we assume the LLM-generated tool call (AI Message) is correct and proceed to resume execution.

In [57]python · cell 37

python

response = agent.query(
    input=None,  # Resume (continue with the tool call AI Message).
    interrupt_before=["tools"],
    interrupt_after=["tools"],
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
)
langchain_load(response["messages"][-1]).pretty_print()

The process is interrupted again after receiving the tool message.

Upon review, if the LLM-generated Tool Message appears correct, we can resume execution.

In [15]python · cell 39

python

response = agent.query(
    input=None,  # Resume (continue with the Tool Message).
    interrupt_before=["tools"],
    interrupt_after=["tools"],
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
)
langchain_load(response["messages"][-1]).pretty_print()

Fetching State History

You can fetch the state history by calling .get_state_history.

In [16]python · cell 41

python

for state_snapshot in agent.get_state_history(
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
):
    if state_snapshot["metadata"]["step"] >= 0:
        print(f'step {state_snapshot["metadata"]["step"]}: {state_snapshot["config"]}')
        state_snapshot["values"]["messages"][-1].pretty_print()
        print("\n")

Time Travel

LangGraph's Time Travel demonstrates how to build a conversational agent with persistent memory, enabling human intervention to correct past actions. Essentially, it "rewinds" the conversation to a previous state, allows for mistake correction, and permits the agent to continue from that corrected point.

You can "time travel" by calling .get_state. By default, the agent retrieves the latest state.

In [17]python · cell 43

python

state = agent.get_state(
    config={
        "configurable": {
            "thread_id": "human-in-the-loop-deepdive",
        }
    }
)

print(f'step {state["metadata"]["step"]}: {state["config"]}')
state["values"]["messages"][-1].pretty_print()

To retrieve an earlier state, you need to specify the checkpoint_id (and checkpoint_ns).

In [18]python · cell 45

python

snapshot_config = {}
for state_snapshot in agent.get_state_history(
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
):
    if state_snapshot["metadata"]["step"] == 1:
        snapshot_config = state_snapshot["config"]
        break

snapshot_config

In [19]python · cell 46

python

state = agent.get_state(config=snapshot_config)
print(f'step {state["metadata"]["step"]}: {state["config"]}')
state["values"]["messages"][-1].pretty_print()

In [20]python · cell 47

python

state

Replay

LangGraph's Replay feature allows you to resume or replay a conversation from any specific point in its history.

You can initiate a replay by passing the state["config"] back to the agent. Note that the execution resumes exactly where it was left off, executing a tool call.

In [21]python · cell 49

python

state["config"]

In [22]python · cell 50

python

for state_values in agent.stream_query(
    input=None,  # resume
    stream_mode="values",
    config=state["config"],
):
    langchain_load(state_values["messages"][-1]).pretty_print()

Branching

LangGraph's Branching feature allows you to modify and re-run a LangGraph conversation from a specific point in its history (rather than just from the latest state). This enables the agent to explore alternate trajectories or allows a user to "version control" changes in a workflow.

In this example, you will:

Update the tool calls from a previous step.
Call .update_state to rerun the step with the updated configuration.

In [23]python · cell 52

python

last_message = state["values"]["messages"][-1]
print(last_message)
print(last_message.tool_calls)

Update the tool calls from the previous step.

In [24]python · cell 54

python

last_message.tool_calls[0]["args"]["currency_date"] = "2024-09-01"
last_message.tool_calls

Call .update_state to rerun the step with the updated configuration.

In [25]python · cell 56

python

branch_config = agent.update_state(
    config=state["config"],
    values={"messages": [last_message]},  # the update we want to make
)
branch_config

In [26]python · cell 57

python

for state_values in agent.stream_query(
    input=None,  # resume
    stream_mode="values",
    config=branch_config,
):
    langchain_load(state_values["messages"][-1]).pretty_print()

Deploying the Agent

In [27]python · cell 59

python

remote_agent = agent_engines.create(
    agent_engines.LanggraphAgent(
        model="gemini-2.0-flash",
        tools=[get_exchange_rate],
        model_kwargs={"temperature": 0, "max_retries": 6},
        checkpointer_kwargs=checkpointer_kwargs,
        checkpointer_builder=checkpointer_builder,
    ),
    requirements=[
        "google-cloud-aiplatform[agent_engines,langchain]",
        "requests",
    ],
)

remote_agent

Querying the Remote Agent

Remote testing

In [28]python · cell 62

python

for state_updates in remote_agent.stream_query(
    input=inputs,
    stream_mode="updates",
    config={"configurable": {"thread_id": "remote-streaming-thread-updates"}},
):
    print(state_updates)

In [29]python · cell 63

python

for state_values in remote_agent.stream_query(
    input=inputs,
    stream_mode="values",
    config={"configurable": {"thread_id": "remote-human-in-the-loop-overall"}},
):
    print(state_values)

Reviewing Tool Calls

In [33]python · cell 65

python

response = remote_agent.query(
    input=inputs,
    interrupt_before=["tools"],  # Before invoking the tool.
    interrupt_after=["tools"],  # After getting a tool message.
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
)
langchain_load(response["messages"][-1]).pretty_print()

In [34]python · cell 66

python

response = remote_agent.query(
    input=None,  # Resume (continue with the tool call AI Message).
    interrupt_before=["tools"],
    interrupt_after=["tools"],
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
)
langchain_load(response["messages"][-1]).pretty_print()

In [35]python · cell 67

python

response = agent.query(
    input=None,  # Resume (continue with the Tool Message).
    interrupt_before=["tools"],
    interrupt_after=["tools"],
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
)
langchain_load(response["messages"][-1]).pretty_print()

Cleaning up

After you've finished experimenting, it's a good practice to clean up your cloud resources. You can delete the deployed Agent Engine instance to avoid any unexpected charges on your Google Cloud account.

In [54]python · cell 69

python

remote_agent.delete()

← 返回课程

Generative AI on Google Cloud

58 / 80·notebook

Chapter 58

Building and Deploying a Human-in-the-Loop LangGraph Application with Agent Engine on Vertex AI

NotebookPython 369 cells

In [ ]python · cell 1

python

# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Building and Deploying a Human-in-the-Loop LangGraph Application with Agent Engine on Vertex AI

Share to:


Author(s)	Xiaolong Yang

Overview

The previous notebook covered: Defining Tools, Defining a Router, Building a LangGraph Application, Local Testing, Deploying to Vertex AI, Remote Testing, and Cleaning Up Resources.

This notebook expands on those concepts and explores the following Human-in-the-Loop features:

Reviewing Tool Calls: Implement human oversight after tool use, allowing for verification and correction of actions before proceeding.
Fetching State History: Retrieve the complete execution history of the LangGraph application for auditing, analysis, and potential state reversion.
Time Travel: Examine the state of the agent at a specific point in time to understand past decisions.
Replay: Restart execution from a specific checkpoint without modifications to ensure consistent results.
Branching: Create alternative execution paths based on a past state, enabling the agent to explore different possibilities or correct previous errors.

By the end of this notebook, you'll possess the skills to build and deploy customized Human-in-the-Loop generative AI applications using LangGraph, Agent Engine, and Vertex AI.

Get started

Install the Vertex AI SDK and Required Packages

In [1]python · cell 7

python

%pip install --upgrade --user --quiet \
    "google-cloud-aiplatform[agent_engines,langchain]" \
    requests --force-reinstall

Restart runtime

To use the newly installed packages in this Jupyter runtime, you must restart the runtime. You can do this by running the cell below, which restarts the current kernel.

The restart might take a minute or longer. After it's restarted, continue to the next step.

In [2]python · cell 9

python

import IPython

app = IPython.Application.instance()
app.kernel.do_shutdown(True)

Authenticate your notebook environment (Colab only)

If you're running this notebook on Google Colab, run the cell below to authenticate your environment.

In [1]python · cell 12

python

import sys

if "google.colab" in sys.modules:
    from google.colab import auth

    auth.authenticate_user()

Set Google Cloud Project Information and Initialize the Vertex AI SDK

Before using Vertex AI, ensure you have an existing Google Cloud project and have enabled the Vertex AI API.

Refer to the documentation for more details on setting up a project and development environment.

In [2]python · cell 14

python

PROJECT_ID = "[your-project-id]"  # @param {type:"string"}
LOCATION = "us-central1"  # @param {type:"string"}
STAGING_BUCKET = "gs://[your-staging-bucket]"  # @param {type:"string"}

import vertexai

vertexai.init(project=PROJECT_ID, location=LOCATION, staging_bucket=STAGING_BUCKET)

Building and Deploying a LangGraph App on Agent Engine

The following sections guide you through building and deploying a LangGraph application using Agent Engine on Vertex AI.

Import Libraries

Import the required Python libraries. These libraries provide the necessary tools for interacting with LangGraph, Vertex AI, and other components of the application.

In [3]python · cell 17

python

from langchain.load import load as langchain_load
import requests
from vertexai import agent_engines

Define Tools

Begin by defining the tools for your LangGraph application. You'll define a custom Python function that serves as a tool within our agentic application.

In [4]python · cell 19

python

def get_exchange_rate(
    currency_from: str = "USD",
    currency_to: str = "EUR",
    currency_date: str = "latest",
):
    """Retrieves the exchange rate between two currencies on a specified date.

    Uses the Frankfurter API (https://api.frankfurter.app/) to obtain
    exchange rate data.

    Args:
        currency_from: The base currency (3-letter currency code).
            Defaults to "USD" (US Dollar).
        currency_to: The target currency (3-letter currency code).
            Defaults to "EUR" (Euro).
        currency_date: The date for which to retrieve the exchange rate.
            Defaults to "latest" for the most recent exchange rate data.
            Can be specified in YYYY-MM-DD format for historical rates.

    Returns:
        dict: A dictionary containing the exchange rate information.
            Example: {"amount": 1.0, "base": "USD", "date": "2023-11-24",
                "rates": {"EUR": 0.95534}}
    """

    response = requests.get(
        f"https://api.frankfurter.app/{currency_date}",
        params={"from": currency_from, "to": currency_to},
    )
    return response.json()

Define `Checkpointers`

LangGraph provides a Checkpointer Interface, defining methods for saving and loading the state. Several built-in checkpointers are available to implement this interface.

In [5]python · cell 21

python

checkpointer_kwargs = None


def checkpointer_builder(**kwargs):
    from langgraph.checkpoint.memory import MemorySaver

    return MemorySaver()

Define the Human-in-the-Loop LangGraph Application

Now, you'll integrate all the components to define your Human-in-the-Loop LangGraph application within Agent Engine.

This application will utilize the tools and checkpointer you've defined. LangGraph offers a powerful framework for structuring these interactions and leveraging the capabilities of LLMs.

In [6]python · cell 23

python

agent = agent_engines.LanggraphAgent(
    model="gemini-2.0-flash",
    tools=[get_exchange_rate],
    model_kwargs={"temperature": 0, "max_retries": 6},
    checkpointer_kwargs=checkpointer_kwargs,
    checkpointer_builder=checkpointer_builder,
)

Local Testing

This section covers local testing of your LangGraph application before deployment to ensure it behaves as expected.

In [7]python · cell 25

python

agent.set_up()

In [8]python · cell 26

python

inputs = {
    "messages": [
        ("user", "What is the exchange rate from US dollars to Swedish currency?")
    ]
}

In [10]python · cell 27

python

response = agent.query(
    input=inputs,
    config={"configurable": {"thread_id": "synchronous-thread-id"}},
)

response["messages"][-1]["kwargs"]["content"]

You can also utilize streaming mode to stream back the values of the graph, representing the full state after each node execution.

In [11]python · cell 29

python

for state_values in agent.stream_query(
    input=inputs,
    stream_mode="values",
    config={"configurable": {"thread_id": "streaming-thread-values"}},
):
    print(state_values)

Alternatively, you can stream back updates to the graph. These represent the changes to the state after each node is executed.

In [12]python · cell 31

python

for state_updates in agent.stream_query(
    input=inputs,
    stream_mode="updates",
    config={"configurable": {"thread_id": "streaming-thread-updates"}},
):
    print(state_updates)

Human-in-the-loop

Reviewing Tool Calls

To achieve this, the agent needs to interrupt execution in the following scenarios:

Before invoking the tool (when the LLM generates a tool call AI Message).
After receiving a tool response.

In [56]python · cell 35

python

response = agent.query(
    input=inputs,
    interrupt_before=["tools"],  # Before invoking the tool.
    interrupt_after=["tools"],  # After getting a tool message.
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
)
langchain_load(response["messages"][-1]).pretty_print()

The process was interrupted before invoking the tool.

After review, we assume the LLM-generated tool call (AI Message) is correct and proceed to resume execution.

In [57]python · cell 37

python

response = agent.query(
    input=None,  # Resume (continue with the tool call AI Message).
    interrupt_before=["tools"],
    interrupt_after=["tools"],
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
)
langchain_load(response["messages"][-1]).pretty_print()

The process is interrupted again after receiving the tool message.

Upon review, if the LLM-generated Tool Message appears correct, we can resume execution.

In [15]python · cell 39

python

response = agent.query(
    input=None,  # Resume (continue with the Tool Message).
    interrupt_before=["tools"],
    interrupt_after=["tools"],
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
)
langchain_load(response["messages"][-1]).pretty_print()

Fetching State History

You can fetch the state history by calling .get_state_history.

In [16]python · cell 41

python

for state_snapshot in agent.get_state_history(
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
):
    if state_snapshot["metadata"]["step"] >= 0:
        print(f'step {state_snapshot["metadata"]["step"]}: {state_snapshot["config"]}')
        state_snapshot["values"]["messages"][-1].pretty_print()
        print("\n")

Time Travel

You can "time travel" by calling .get_state. By default, the agent retrieves the latest state.

In [17]python · cell 43

python

state = agent.get_state(
    config={
        "configurable": {
            "thread_id": "human-in-the-loop-deepdive",
        }
    }
)

print(f'step {state["metadata"]["step"]}: {state["config"]}')
state["values"]["messages"][-1].pretty_print()

To retrieve an earlier state, you need to specify the checkpoint_id (and checkpoint_ns).

In [18]python · cell 45

python

snapshot_config = {}
for state_snapshot in agent.get_state_history(
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
):
    if state_snapshot["metadata"]["step"] == 1:
        snapshot_config = state_snapshot["config"]
        break

snapshot_config

In [19]python · cell 46

python

state = agent.get_state(config=snapshot_config)
print(f'step {state["metadata"]["step"]}: {state["config"]}')
state["values"]["messages"][-1].pretty_print()

In [20]python · cell 47

python

state

Replay

LangGraph's Replay feature allows you to resume or replay a conversation from any specific point in its history.

You can initiate a replay by passing the state["config"] back to the agent. Note that the execution resumes exactly where it was left off, executing a tool call.

In [21]python · cell 49

python

state["config"]

In [22]python · cell 50

python

for state_values in agent.stream_query(
    input=None,  # resume
    stream_mode="values",
    config=state["config"],
):
    langchain_load(state_values["messages"][-1]).pretty_print()

Branching

In this example, you will:

Update the tool calls from a previous step.
Call .update_state to rerun the step with the updated configuration.

In [23]python · cell 52

python

last_message = state["values"]["messages"][-1]
print(last_message)
print(last_message.tool_calls)

Update the tool calls from the previous step.

In [24]python · cell 54

python

last_message.tool_calls[0]["args"]["currency_date"] = "2024-09-01"
last_message.tool_calls

Call .update_state to rerun the step with the updated configuration.

In [25]python · cell 56

python

branch_config = agent.update_state(
    config=state["config"],
    values={"messages": [last_message]},  # the update we want to make
)
branch_config

In [26]python · cell 57

python

for state_values in agent.stream_query(
    input=None,  # resume
    stream_mode="values",
    config=branch_config,
):
    langchain_load(state_values["messages"][-1]).pretty_print()

Deploying the Agent

In [27]python · cell 59

python

remote_agent = agent_engines.create(
    agent_engines.LanggraphAgent(
        model="gemini-2.0-flash",
        tools=[get_exchange_rate],
        model_kwargs={"temperature": 0, "max_retries": 6},
        checkpointer_kwargs=checkpointer_kwargs,
        checkpointer_builder=checkpointer_builder,
    ),
    requirements=[
        "google-cloud-aiplatform[agent_engines,langchain]",
        "requests",
    ],
)

remote_agent

Querying the Remote Agent

Remote testing

In [28]python · cell 62

python

for state_updates in remote_agent.stream_query(
    input=inputs,
    stream_mode="updates",
    config={"configurable": {"thread_id": "remote-streaming-thread-updates"}},
):
    print(state_updates)

In [29]python · cell 63

python

for state_values in remote_agent.stream_query(
    input=inputs,
    stream_mode="values",
    config={"configurable": {"thread_id": "remote-human-in-the-loop-overall"}},
):
    print(state_values)

Reviewing Tool Calls

In [33]python · cell 65

python

response = remote_agent.query(
    input=inputs,
    interrupt_before=["tools"],  # Before invoking the tool.
    interrupt_after=["tools"],  # After getting a tool message.
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
)
langchain_load(response["messages"][-1]).pretty_print()

In [34]python · cell 66

python

response = remote_agent.query(
    input=None,  # Resume (continue with the tool call AI Message).
    interrupt_before=["tools"],
    interrupt_after=["tools"],
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
)
langchain_load(response["messages"][-1]).pretty_print()

In [35]python · cell 67

python

response = agent.query(
    input=None,  # Resume (continue with the Tool Message).
    interrupt_before=["tools"],
    interrupt_after=["tools"],
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
)
langchain_load(response["messages"][-1]).pretty_print()

Cleaning up

In [54]python · cell 69

python

remote_agent.delete()

Building and Deploying a Human-in-the-Loop LangGraph Application with Agent Engine on Vertex AI

Overview

Get started

Install the Vertex AI SDK and Required Packages

Restart runtime

Authenticate your notebook environment (Colab only)

Set Google Cloud Project Information and Initialize the Vertex AI SDK

Building and Deploying a LangGraph App on Agent Engine

Import Libraries

Define Tools

Define Checkpointers

Define the Human-in-the-Loop LangGraph Application

Local Testing

Human-in-the-loop

Reviewing Tool Calls

Fetching State History

Time Travel

Replay

Branching

Deploying the Agent

Querying the Remote Agent

Remote testing

Reviewing Tool Calls

Cleaning up

Building and Deploying a Human-in-the-Loop LangGraph Application with Agent Engine on Vertex AI

Overview

Get started

Install the Vertex AI SDK and Required Packages

Restart runtime

Authenticate your notebook environment (Colab only)

Set Google Cloud Project Information and Initialize the Vertex AI SDK

Building and Deploying a LangGraph App on Agent Engine

Import Libraries

Define Tools

Define Checkpointers

Define the Human-in-the-Loop LangGraph Application

Local Testing

Human-in-the-loop

Reviewing Tool Calls

Fetching State History

Time Travel

Replay

Branching

Deploying the Agent

Querying the Remote Agent

Remote testing

Reviewing Tool Calls

Cleaning up

Define `Checkpointers`

Define `Checkpointers`