Generative AI Google Brain

gemini-agent-engine-langgraph-human-in-the-loop

4 min read

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

Source notebook

Repo path: gemini/agent-engine/langgraph_human_in_the_loop.ipynb · Open on GitHub · advanced

Builds, tests, deploys, and resumes a human-in-the-loop LangGraph agent on Vertex AI Agent Engine.

Summary

This notebook teaches how to create a LangGraph-based exchange-rate agent with a custom tool and an in-memory checkpointer. It demonstrates local querying, streaming state values and updates, interrupting before and after tool calls for human review, inspecting state history, time travel, replay, branching from a checkpoint, deploying to Agent Engine, remote testing, and cleanup.

Key code patterns

Initialize Vertex AI

PROJECT_ID = "[your-project-id]"
LOCATION = "us-central1"
STAGING_BUCKET = "gs://[your-staging-bucket]"
 
import vertexai
vertexai.init(project=PROJECT_ID, location=LOCATION, staging_bucket=STAGING_BUCKET)

Sets the project, region, and staging bucket required before creating local or remote Agent Engine resources.

Define a callable tool

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

Shows how a Python function can be exposed as an agent tool for external API access.

Create LangGraph agent

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

Combines Gemini, tools, deterministic settings, retries, and checkpointing into a LangGraph agent.

Interrupt for review

response = agent.query(
    input=inputs,
    interrupt_before=["tools"],
    interrupt_after=["tools"],
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
)

Pauses execution before and after tool use so a human can inspect tool calls and tool results.

Resume execution

response = agent.query(
    input=None,
    interrupt_before=["tools"],
    interrupt_after=["tools"],
    config={"configurable": {"thread_id": "human-in-the-loop-deepdive"}},
)

Uses input=None with the same thread configuration to continue from an interrupted checkpoint.

Branch from checkpoint

last_message.tool_calls[0]["args"]["currency_date"] = "2024-09-01"
branch_config = agent.update_state(
    config=state["config"],
    values={"messages": [last_message]},
)

Edits a prior tool call and creates a new branch for alternate execution from saved state.

Deploy remote agent

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_builder=checkpointer_builder,
    ),
    requirements=["google-cloud-aiplatform[agent_engines,langchain]", "requests"],
)

Packages the LangGraph agent and dependencies for deployment as a remote Agent Engine instance.

Models & APIs used

  • Models: gemini-2.0-flash
  • APIs / services: Vertex AI, Agent Engine, Cloud Storage
  • SDKs / libraries: google-cloud-aiplatform, vertexai, langchain, langgraph, requests

When to use this

Use this pattern when an agent workflow needs auditable checkpoints, human review of tool calls, replay, branching, and deployment on Vertex AI Agent Engine.

Gotchas & caveats

  • The notebook requires google-cloud-aiplatform with agent_engines and langchain extras plus requests, followed by a runtime restart.
  • Colab authentication is only run when google.colab is present.
  • A Google Cloud project must exist and the Vertex AI API must be enabled before initialization.
  • The notebook uses us-central1 and a gs:// staging bucket for Vertex AI initialization.
  • The checkpointer shown is in-memory, so persistence is suitable for demonstration rather than durable storage.
  • Thread IDs in config are required to resume, stream, inspect history, replay, and branch a specific execution.
  • The external Frankfurter API is called by the tool and must be reachable.
  • The deployed Agent Engine instance should be deleted to avoid unexpected charges.

Best practices

  • Initialize Vertex AI with project, location, and staging bucket before using Agent Engine.
  • Set temperature to 0 for deterministic tool-review examples.
  • Use max_retries for model calls in the LanggraphAgent configuration.
  • Call agent.set_up() before local testing.
  • Use interrupt_before and interrupt_after around tools for human oversight.
  • Use get_state_history and get_state for auditing and checkpoint inspection.
  • Use update_state to branch from a past state instead of modifying only the latest conversation.
  • Pass explicit requirements when deploying the remote agent.
  • Delete the remote agent after experimentation to avoid unexpected charges.

Graph View

Backlinks