Generate and review

adk-generate-review-lesson/
  requirements.txt
  .env.example
  changelog-3.5.0.md
  changelog-3.4.0.md
  harness.py
  run.py
  review_gate.py
  review-run.json
  test_review_gate.py
  test_runner.py

## 3.5.0
- Build pages in parallel; set the worker count with `--jobs`.
- Cache rendered code blocks in `.docuctl/cache`.
- Fix the search index missing pages whose title contains a colon.

"""Draft a release announcement with one agent and gate it with a second, reviewing agent."""
from typing import Literal

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


class Changes(BaseModel):
    version: str
    entries: list[str]


class Draft(BaseModel):
    title: str
    body: str


class Review(BaseModel):
    verdict: Literal["approve", "reject"]
    unsupported: list[str] = Field(
        description="Each sentence of the draft that no changelog entry supports, copied exactly.")


def parse_changelog(node_input: str, version_in_facts: bool = True):
    """The first line is the version, each '- ' line is one entry.
    It also writes the facts the reviewer checks against to state."""
    lines = [line.strip() for line in node_input.strip().splitlines() if line.strip()]
    if not lines or not lines[0].startswith("## "):
        raise ValueError("changelog must start with a '## <version>' heading")
    entries = [line[2:] for line in lines[1:] if line.startswith("- ")]
    if not entries:
        raise ValueError(f"no entries under {lines[0]}")
    changes = Changes(version=lines[0][3:], entries=entries)
    facts = [f"- {e}" for e in entries]
    if version_in_facts:
        facts.insert(0, f"- docuctl {changes.version} is released.")
    return Event(output=changes, state={"changelog": "\n".join(facts)})


def gate(node_input: Review, draft: dict):
    """Router node: the reviewer's verdict picks the edge. A verdict that lists
    unsupported sentences is a rejection, whatever the verdict field says."""
    verdict = "reject" if node_input.unsupported else node_input.verdict
    return Event(route=verdict, output={"draft": draft, "unsupported": node_input.unsupported})


def publish(node_input: dict):
    draft = node_input["draft"]
    return Event(message=f"APPROVED\n# {draft['title']}\n\n{draft['body']}")


def send_back(node_input: dict):
    problems = "\n".join(f"- {s}" for s in node_input["unsupported"])
    return Event(message=f"REJECTED: the reviewer found claims with no changelog entry:\n{problems}")


WRITER = {
    "upbeat": ("Write a short release announcement for users of docuctl, the docs-site CLI. "
               "Make it upbeat and explain why each change matters to them. 3 to 5 sentences."),
    "factual": ("Write a short release announcement for users of docuctl, the docs-site CLI. "
                "Say what each entry changes, in plain words. Add no benefit, number or "
                "comparison that an entry does not state. 3 to 5 sentences."),
}


def build_workflow(model=None, style="factual", version_in_facts=True) -> Workflow:
    model = model or "gemini-3.8-flash"  # or LiteLlm(model="ollama_chat/gemma4:e2b")
    writer = Agent(
        name="write_announcement",
        model=model,
        generate_content_config=types.GenerateContentConfig(
            max_output_tokens=1024,
            thinking_config=types.ThinkingConfig(thinking_level=types.ThinkingLevel.LOW)),
        input_schema=Changes,
        output_schema=Draft,
        output_key="draft",
        instruction=WRITER[style],
    )
    reviewer = Agent(
        name="review_announcement",
        model=model,
        generate_content_config=types.GenerateContentConfig(
            max_output_tokens=1024,
            thinking_config=types.ThinkingConfig(thinking_level=types.ThinkingLevel.LOW)),
        input_schema=Draft,
        output_schema=Review,
        instruction=(
            "You fact-check release announcements. The only facts are these changelog "
            "entries:\n{changelog}\n\nList every sentence of the draft that states a fact, "
            "benefit or number that no entry supports. Approve only if the list is empty."
        ),
    )

    def read_changelog(node_input: str):
        """Function node. version_in_facts=False reproduces the reviewer's false rejection."""
        return parse_changelog(node_input, version_in_facts)

    return Workflow(
        name="announce_release",
        edges=[
            ("START", read_changelog, writer, reviewer, gate),
            (gate, {"approve": publish, "reject": send_back}),
        ],
    )

"""Compare the workflow's cases; replay saved replies unless --live is explicit."""
import argparse
import asyncio
import copy
import json
from pathlib import Path

from dotenv import load_dotenv
import os
from google.adk.agents.run_config import RunConfig
from google.adk.apps import App
from google.adk.runners import InMemoryRunner
from google.genai import types
from harness import ReplayLlm, summarize

HERE = Path(__file__).resolve().parent
from review_gate import build_workflow

async def compare(*, live=False):
    record = json.loads((HERE / "review-run.json").read_text())
    cases = [("factual", "changelog-3.5.0.md", True),
             ("upbeat", "changelog-3.5.0.md", True),
             ("factual", "changelog-3.4.0.md", False)]
    for style, filename, has_version in cases:
        saved = next(r for r in record["runs"] if
                     (r["style"], r["changelog"], r["version_in_facts"]) == (style, filename, has_version))
        model = None if live else replay(saved, record)
        builder = build_workflow if live else __import__("importlib").import_module("historical-shipctl-source.review_gate").build_workflow
        root = builder(model, style=style, version_in_facts=has_version)
        rows = await invoke(root, (HERE / ("" if live else "historical-shipctl-source") / filename).read_text(), max_llm_calls=2)
        display(f"{filename} | {style} | version fact: {has_version}", rows)

async def invoke(root, text, *, max_llm_calls, plugins=()):
    """Run the actual workflow and close its in-memory Runner, including on failure."""
    app = App(name="docs_assistant", root_agent=root, plugins=list(plugins))
    runner = InMemoryRunner(app=app)
    session = await runner.session_service.create_session(app_name=runner.app_name, user_id="dev")
    try:
        return [summarize(event) async for event in runner.run_async(
            user_id="dev", session_id=session.id,
            new_message=types.Content(role="user", parts=[types.Part(text=text)]),
            run_config=RunConfig(max_llm_calls=max_llm_calls))]
    finally:
        await runner.close()


def replay(saved, record):
    return ReplayLlm(model=record["environment"]["model"],
                     calls=copy.deepcopy(saved["model_calls"]))


def display(label, rows):
    print(f"\n{label}")
    for row in rows:
        if row.get("route"):
            print("route:", row["route"])
    messages = [row["message"] for row in rows if row.get("message")]
    if messages:
        print(messages[-1])


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--live", action="store_true", help="request new Gemini replies instead of replaying saved replies")
    args = parser.parse_args()
    if args.live:
        load_dotenv(HERE / ".env", override=False)
        if not (os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY")):
            parser.error("Set your Gemini API key in .env before using --live")
    print("Mode:", "live Gemini requests" if args.live else "offline replay of saved model replies")
    asyncio.run(compare(live=args.live))


if __name__ == "__main__":
    main()