Google ADK

google-adk==2.2.0
python-dotenv>=1.0.0

from google.adk.agents import Agent, Context
from google.adk.workflow import Edge, START, Workflow, node

MODEL = "gemini-flash-latest"

blog_planner = Agent(
    name="BlogPlanner",
    description="Creates a practical, skimmable outline in Markdown",
    model=MODEL,
    output_key="blog_outline",
    instruction="""
As a technical content strategist, produce a clear Markdown outline with:
- Title
- Short intro
- 2 main sections.
- Conclusion

If `codebase_context` exists in state, weave in specific sections/snippets.
Return only the outline in Markdown.
""",
)

outline_validator = Agent(
    name="OutlineValidator",
    description="Validates the outline.",
    model=MODEL,
    instruction="""
Check the outline (`blog_outline`) in state. If it has a title, intro,
2 sections and a conclusion, respond "OK", otherwise return "RETRY" and
list missing pieces.""",
    output_key="validation_result",
)

def _route_outline(ctx: Context) -> str:
    result = ctx.state.get("validation_result", "")
    tries = ctx.state.get("planner_tries", 0) + 1
    ctx.state["planner_tries"] = tries
    return "OK" if result.strip().upper().startswith("OK") or tries >= 3 else "RETRY"

route_outline = node(_route_outline, name="RouteOutline")

resilient_planner = Workflow(
    name="ResilientPlanner",
    description="Generates and validates a blog outline, retrying up to 3 times",
    edges=[
        (START, blog_planner, outline_validator, route_outline),
        Edge(from_node=route_outline, to_node=blog_planner, route="RETRY"),
    ],
)

root_agent = Workflow(
    name="Blogger",
    description="Plans and writes a complete blog post end-to-end.",
    edges=[
        (START, resilient_planner, resilient_writer, final_agent),
    ],
)

from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from .agent import root_agent

async def run(topic: str) -> None:
    session_service = InMemorySessionService()
    await session_service.create_session(
        app_name="blogger", user_id="local", session_id="run-1"
    )
    runner = Runner(
        node=root_agent,
        app_name="blogger",
        session_service=session_service,
    )
    async for event in runner.run_async(
        user_id="local",
        session_id="run-1",
        new_message=types.Content(
            role="user",
            parts=[types.Part(text=topic)],
        ),
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if part.text:
                    print(part.text, end="", flush=True)