How to Build an AI Agent for Content Production

from dataclasses import dataclass, field

@dataclass
class Brief:
    id: str
    topic: str
    audience: str
    angle: str
    keyword: str
    words: int = 1200
    style: dict = field(default_factory=dict)   # tone, banned words, reading level

def load_brief(raw: dict, seen: set) -> Brief | None:
    if raw["id"] in seen:                        # already drafted
        return None
    seen.add(raw["id"])
    return Brief(**raw)
# The brief id makes it idempotent: each brief is drafted once, even on a re-run.

from google import genai
from google.genai import types
import httpx
client = genai.Client()

def research(topic: str) -> dict:
    # Google Search grounding: the model searches the live web for real sources.
    res = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=[f"Research '{topic}'. Return key facts with the source URL for each."],
        config=types.GenerateContentConfig(
            tools=[types.Tool(google_search=types.GoogleSearch())]))
    urls = [c.web.uri for c in
            res.candidates[0].grounding_metadata.grounding_chunks if c.web]
    return {"notes": res.text, "sources": verify(urls)}

def verify(urls: list[str]) -> list[str]:
    # A cited URL that does not resolve is dropped, never kept.
    return [u for u in urls if _ok(u)]

def _ok(u: str) -> bool:
    try:    return httpx.head(u, follow_redirects=True, timeout=8).status_code < 400
    except Exception: return False

from google import genai
from google.genai import types
from pydantic import BaseModel
client = genai.Client()

class Outline(BaseModel):
    title: str
    sections: list[str]          # ordered H2 headings

def outline(brief, research: dict) -> Outline:
    res = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=[f"Topic: {brief.topic}. Angle: {brief.angle}. "
                  f"Audience: {brief.audience}. Research notes:\n{research['notes']}\n"
                  f"Return a title and ordered H2 headings for a {brief.words}-word piece."],
        config=types.GenerateContentConfig(
            response_mime_type="application/json", response_schema=Outline))
    return res.parsed            # a typed Outline the drafter writes against

from google import genai
client = genai.Client()

def draft(brief, outline, sources: list[str]) -> str:
    return client.models.generate_content(
        model="gemini-2.5-flash",
        contents=[f"Write a {brief.words}-word draft for the outline below in this "
                  f"voice: {brief.style}. Cite a source inline, as [n], next to each "
                  f"claim, using ONLY these sources: {sources}. Do not invent sources. "
                  f"This is a draft for human review, not for publishing.\n{outline}"]).text
# The return value is a draft a person edits and approves; the agent never publishes.

import re

def load_sitemap_keywords() -> dict[str, str]:
    # keyword -> a real URL on your site, parsed from sitemap.xml
    return {"demand forecasting": "/services/demand-forecasting/",
            "computer vision": "/services/computer-vision/"}

def add_internal_links(markdown: str, kw_to_url: dict[str, str]) -> str:
    for kw, url in kw_to_url.items():
        # link only the FIRST mention, and only whole words
        markdown = re.sub(rf"\b({re.escape(kw)})\b",
                          rf"[\1]({url})", markdown, count=1, flags=re.I)
    return markdown
# Links only ever point at URLs already in your sitemap, never an invented path.

import re, textstat

def seo_checks(brief, markdown: str, meta_title: str, meta_desc: str) -> list[dict]:
    words = len(re.findall(r"\w+", markdown))
    checks = [
      {"name": "word count", "ok": words >= brief.words * 0.8},
      {"name": "has H2", "ok": markdown.count("## ") >= 2},
      {"name": "keyword present", "ok": brief.keyword.lower() in markdown.lower()},
      {"name": "meta title <= 60", "ok": len(meta_title) <= 60},
      {"name": "meta desc 50-160", "ok": 50 <= len(meta_desc) <= 160},
      {"name": "readable (grade <= 10)", "ok": textstat.flesch_kincaid_grade(markdown) <= 10},
    ]
    return checks      # a pass/fail checklist for the editor, never an auto-block

def queue_for_review(brief, draft: str, sources: list, checks: list) -> None:
    review_queue.add(brief_id=brief.id, draft=draft,
                     sources=sources, checks=checks)      # awaits a human

def on_approval(brief, edited_markdown: str) -> None:
    # A person has edited and approved; hand to the CMS as a DRAFT, not live.
    cms.create_draft(title=brief.topic, body=edited_markdown)
    audit.log("approved_to_cms", brief.id)        # only after human approval
# The agent drafts and queues; publishing is a human action in your CMS.

from google.adk.agents import SequentialAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

content_pipeline = SequentialAgent(name="content_pipeline",
    sub_agents=[researcher, outliner, drafter])   # each an LlmAgent with the tools

runner = Runner(agent=content_pipeline, app_name="content",
                session_service=InMemorySessionService())

def guard(brief, draft: dict) -> str | None:
    if store.seen(brief.id):               return "already_drafted"  # idempotent
    if any(not _ok(u) for u in draft["sources"]): return "bad_source"  # must resolve
    if originality(draft["text"]) < 0.8:   return "too_similar"      # plagiarism check
    return None
# Run on each new brief via cron / Cloud Scheduler. Publishing always needs a human.