Route a workflow with conditional edges

When you need this. Your documentation build runs on every pull request. Most logs report no broken links, so sending every build to a model wastes work. We will read the log in Python and call an agent only when the build needs link suggestions.

A conditional edge selects the next node from a route value. Our router returns broken_links for a completed build with broken links, clean for a completed build without them, and incomplete when the log has no final completion line.

The graph sends broken links to an agent, clean logs to a short result, and unmatched routes to a rerun message. DEFAULT_ROUTE provides that last branch, including for route values we add later.

Prepare the build-log project

You need Python 3.12 or later and a Gemini API key. Use Download code, extract the ZIP, and open its adk-graph-routes-lesson directory. Keep the Python files and fixtures together. All commands in this lesson run from that directory.

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt

On Windows PowerShell, activate with .venv\Scripts\Activate.ps1.

The Python client reads exported environment variables. The bundled .env.example documents them; it is not loaded by the client. Set GOOGLE_API_KEY privately, then select the Gemini API in this terminal:

unset GOOGLE_GENAI_USE_VERTEXAI DOCS_MODEL
export GOOGLE_GENAI_USE_ENTERPRISE=0

Sending a request to the model uses your provider account. The offline checks later in the lesson need no key.

Keep the three build-*.log files and docs-pages.txt beside the Python code. Here is the input that needs suggestions:

INFO    Building documentation for docs-assistant
WARNING broken link in quickstart.md: install.md#ollama-setup (anchor not found)
WARNING broken link in tools/search-docs.md: reference/citation.md (page not found)
WARNING broken link in troubleshooting.md: reference/cli-flags.md (page not found)
INFO    Build finished

Write the router and the branches

classify_build parses known warning lines and checks for INFO Build finished. The decision is a fact in the log, so Python can make it without asking a model.

"""Route a docs-site build log: broken links go to a triage agent, a clean build ends early."""
import re
from pathlib import Path

from google.adk import Agent, Event, Workflow
from google.genai import types
from google.adk.workflow import DEFAULT_ROUTE
from pydantic import BaseModel

BROKEN = re.compile(r"^WARNING broken link in (\S+): (\S+) \((.+)\)$")
PAGES = Path(__file__).with_name("docs-pages.txt").read_text().split()


class BrokenLink(BaseModel):
    page: str
    target: str
    reason: str


class BrokenLinks(BaseModel):
    links: list[BrokenLink]


class LinkFix(BaseModel):
    page: str
    target: str
    replacement: str


class LinkFixes(BaseModel):
    fixes: list[LinkFix]


def classify_build(node_input: str):
    """Router node: parse the log and pick the next edge. No model call."""
    lines = node_input.strip().splitlines()
    links = [BrokenLink(page=m[1], target=m[2], reason=m[3])
             for m in map(BROKEN.match, lines) if m]
    if not lines or lines[-1] != "INFO    Build finished":
        return Event(route="incomplete", output=BrokenLinks(links=links))
    if links:
        return Event(route="broken_links", output=BrokenLinks(links=links),
                     state={"pages": PAGES})
    return Event(route="clean", output=BrokenLinks(links=[]))


def report_clean(node_input: BrokenLinks):
    return Event(message="Build passed: no broken links.")


def ask_for_rerun(node_input: BrokenLinks):
    return Event(message="The build log has no 'Build finished' line. Rerun the build before triage.")


def check_fixes(node_input: LinkFixes, pages: list[str]):
    """Keep only replacements that name a page that exists; flag the rest for a person."""
    lines = []
    for fix in node_input.fixes:
        page = fix.replacement.split("#", 1)[0]
        verdict = "ok" if page in pages else "needs a person: page does not exist"
        lines.append(f"- {fix.page}: {fix.target} -> {fix.replacement} ({verdict})")
    return Event(message="Suggested link fixes:\n" + "\n".join(lines))


def build_workflow(model=None) -> Workflow:
    triage = Agent(
        name="triage_links",
        model=model or "gemini-3.8-flash",  # or LiteLlm(model="ollama_chat/gemma4:e2b")
        generate_content_config=types.GenerateContentConfig(
            max_output_tokens=1024,
            thinking_config=types.ThinkingConfig(thinking_level=types.ThinkingLevel.LOW)),
        input_schema=BrokenLinks,
        output_schema=LinkFixes,
        instruction=(
            "For each broken link, suggest a replacement target. Use only these pages: "
            "{pages}. For a missing anchor, keep the page and suggest the closest anchor."
        ),
    )
    return Workflow(
        name="build_router",
        edges=[
            ("START", classify_build),
            (classify_build, {
                "broken_links": triage,
                "clean": report_clean,
                DEFAULT_ROUTE: ask_for_rerun,
            }),
            (triage, check_fixes),
        ],
    )

The returned event has two separate jobs. route chooses the edge; output carries BrokenLinks to the chosen node. A route label does not become the next node’s input.

On the broken-links branch, the router also writes the known page names to state. The agent’s {pages} instruction placeholder reads them when ADK builds the request. check_fixes then checks the suggested page names against that same list.

The model suggests a page that does not exist. check_fixes marks the suggestion needs a person: page does not exist. It still returns the report so the reader can review the problem.

The check strips any #fragment before comparing page names. (ok) means that the page exists in the supplied inventory; it says nothing about an anchor or whether a replacement section means the same thing. The workflow proposes edits and does not change files.

Wire the conditional edges

The route map connects labels to nodes:

(classify_build, {
    "broken_links": triage,
    "clean": report_clean,
    DEFAULT_ROUTE: ask_for_rerun,
}),
(triage, check_fixes),

The default applies when no explicit key matches. The incomplete log therefore reaches ask_for_rerun, even though there is no "incomplete" key in the map. The broken-links branch continues through a plain edge to check_fixes.

Unmatched route without a default. Without DEFAULT_ROUTE, an incomplete route ends after the router event, with no exception or final message. Add a default when every run should explain its outcome.

Route type differs from the key. Event(route=True) does not match the string key "True". Use consistent route values and types.

Unreachable node. ADK raises ValidationError when a graph contains a node unreachable from START. Connect the node before constructing the workflow.

Run the router on three logs

Start with the code-only branches, then run the one that needs suggestions:

python graph_client.py --input build-clean.log
python graph_client.py --input build-incomplete.log
python graph_client.py --input build-broken.log

The clean log should return Build passed: no broken links. The incomplete log should ask you to rerun the build. Neither branch reaches an agent. The broken log should pass through triage_links and check_fixes; compare every suggested replacement with docs-pages.txt, then check its anchor yourself.

For each run, find the router event’s route and follow the later node_info.path values. This connects the branch decision to the nodes that actually ran.

"""Run one supplied log through the original Workflow; no recorder or .env autoload."""
import argparse
import asyncio
import json
import os
from pathlib import Path
from google.adk.runners import InMemoryRunner
from build_router import build_workflow
from harness import run

HERE = Path(__file__).parent

def parse_options(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input", type=Path, default=HERE / "build-broken.log", help="one existing UTF-8 log file")
    return parser.parse_args(argv)

async def run_one(options, model=None):
    text = options.input.read_text(encoding="utf-8")
    if not text.strip(): raise ValueError("input log must contain text")
    workflow = build_workflow(model=model or os.environ.get("DOCS_MODEL") or None)
    runner = InMemoryRunner(agent=workflow, app_name="docs_assistant")
    try:
        events, session_id = await run(workflow, text, runner=runner)
        result = {"session_id": session_id, "input_file": str(options.input), "events": events}
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return result
    finally:
        await runner.close()

if __name__ == "__main__":
    asyncio.run(run_one(parse_options()))

Test every branch without a model

python -m unittest -v test_build_router.py test_client.py

The tests replay saved replies through the actual workflow. Clean and incomplete cases supply no model reply, so an accidental edge to the agent fails instead of silently spending a request.

In a separate copy, remove the default route and run the matching offline check. Predict whether you will see an exception or just the router event. The test demonstrates why a successful process exit is a weak workflow completion check.

Match route labels to graph edges

This fixture excerpt contains the router’s route, typed output and state together:

{
  "author": "build_router",
  "node_info": {
    "path": "build_router@1/classify_build@1",
    "output_for": [
      "build_router@1/classify_build@1"
    ]
  },
  "output": {
    "links": [
      {
        "page": "quickstart.md",
        "target": "install.md#ollama-setup",
        "reason": "anchor not found"
      },
      {
        "page": "tools/search-docs.md",
        "target": "reference/citation.md",
        "reason": "page not found"
      },
      {
        "page": "troubleshooting.md",
        "target": "reference/cli-flags.md",
        "reason": "page not found"
      }
    ]
  },
  "route": "broken_links",
  "state": {
    "pages": [
      "index.md",
      "install.md",
      "quickstart.md",
      "tools/search-docs.md",
      "reference/citations.md",
      "reference/cli.md",
      "troubleshooting.md"
    ]
  }
}

After a node completes, ADK follows its matching outgoing edges. A plain edge has no route condition. A route map selects the matching target, or its default when there is no explicit match.

A route can point back to an earlier node, but a graph cycle needs its own exit condition and bound. The matching offline check runs a self-route until a counter in the node reaches its stop condition.

Next, pause a workflow for human approval before it changes application state.

References

The code and commands use google-adk 2.12.0, as pinned in requirements.txt.

• ADK graph routes: route maps and defaults.
• ADK workflow data handling: outputs and state.
• ADK graph workflows: graph construction.