Generative AI Google Brain

gemini-agents-genai-experience-concierge-agent-design-patterns-task-planner

3 min read

Task Planner Agent

Source notebook

Repo path: gemini/agents/genai-experience-concierge/agent-design-patterns/task-planner.ipynb · Open on GitHub · advanced

Builds a LangGraph task-planner agent that plans, searches with Gemini, reflects, and responds.

Summary

The notebook teaches a multi-agent task planner architecture with Planner, Executor, Reflector, and Post-Process nodes. It shows how to generate a structured plan or direct response, execute plan tasks with Gemini and Google Search grounding, reflect on results, and either add more tasks or return a final response. It also demonstrates LangGraph state management, in-memory checkpointing, retries, and notebook streaming for a conversational session.

Key code patterns

Structured plan or response

client = genai.Client(vertexai=True, project=project, location=region)
content_response = await client.aio.models.generate_content(
    model=model_name,
    contents=contents,
    config=genai_types.GenerateContentConfig(
        response_mime_type="application/json",
        response_schema=PlanOrRespond,
        system_instruction=system_instruction,
    ),
)
return PlanOrRespond.model_validate_json(content_response.text)

Constrains Gemini output to a Pydantic schema so routing can choose between a plan and a response.

Grounded task execution

search_tool = genai_types.Tool(google_search=genai_types.GoogleSearch())
for idx, task in enumerate(executed_plan.tasks):
    if task.result is not None:
        continue
    content_response = await client.aio.models.generate_content(
        model=model_name,
        contents=contents,
        config=genai_types.GenerateContentConfig(tools=[search_tool]),
    )
    task.result = content_response.text
    yield idx, task

Uses the Google Search tool only in the executor to fill task results with live research findings.

LangGraph agent loop

state_graph = graph.StateGraph(GraphSession)
state_graph.add_node(PLANNER_NODE_NAME, planner_node)
state_graph.add_node(EXECUTOR_NODE_NAME, executor_node)
state_graph.add_node(REFLECTOR_NODE_NAME, reflector_node)
state_graph.add_node(POST_PROCESS_NODE_NAME, post_process_node)
state_graph.set_entry_point(PLANNER_NODE_NAME)
compiled_graph = state_graph.compile(checkpointer=memory.MemorySaver())

Defines the planner, executor, reflector, and post-processing flow with session checkpointing.

Custom stream updates

async for stream_mode, chunk in compiled_graph.astream(
    input={"current_turn": {"user_input": user_input}},
    config={"configurable": {"thread_id": thread_id, "agent_config": agent_config}},
    stream_mode=["custom"],
):
    if "plan" in chunk:
        text = stringify_plan(Plan.model_validate(chunk["plan"]))
    elif "executed_task" in chunk:
        text = stringify_task(Task.model_validate(chunk["executed_task"]))

Streams generated plans, executed tasks, and responses into the notebook as the graph runs.

Retry transient failures

@retry(
    retry=retry_if_exception(is_retryable_error),
    wait=wait_exponential(min=1, max=10),
    stop=stop_after_attempt(3),
    reraise=True,
)
async def generate_plan(...):
    ...

Retries selected API and connection failures without hiding final errors.

Models & APIs used

  • Models: gemini-3.5-flash
  • APIs / services: Vertex AI, Google Search Grounding
  • SDKs / libraries: google-genai, langgraph, langgraph-checkpoint, pydantic, tenacity, requests, IPython, langchain_core

When to use this

Use this pattern when a user request needs multi-step research, live search, and iterative planning instead of a single model response.

Gotchas & caveats

  • Colab users must authenticate with google.colab.auth.authenticate_user().
  • The runtime must be restarted after installing the notebook dependencies.
  • PROJECT_ID must be provided or resolved from GOOGLE_CLOUD_PROJECT, and REGION defaults to us-central1.
  • The notebook states this multi-agent design can be much slower than a single-agent design because one turn can include many LLM calls and tool uses.
  • The executor iterates through a linear task list and skips only tasks whose result is already set.
  • Retry handling is limited to APIError codes 429, 502, 503, 504 and requests ConnectionError, with three attempts.
  • Reflector execution asserts that every plan task has a non-empty result before reflection.

Best practices

  • Use Pydantic models for Task, Plan, Response, and PlanOrRespond to keep agent decisions structured.
  • Reset task results to None before executing newly generated plan tasks.
  • Separate planning, execution, reflection, and post-processing into explicit LangGraph nodes.
  • Use Google Search grounding in the executor for research tasks requiring live information.
  • Use a MemorySaver checkpointer and thread_id to preserve conversation state across turns.
  • Stream plan, executed_task, and response events so users can observe long-running agent progress.
  • Apply exponential backoff retries to transient GenAI API and connection errors.

Graph View

Backlinks