Pipeline de contenido

Orquestación multi-agent (sem. 19) con una compuerta de seguridad en la salida (sem. 16): el writer hace draft, el critic afila, el writer revisa, y la moderación la aprueba. La forma de una feature real de generación de contenido.

README Pruébalo Código

Showcase — Pipeline de contenido

Córrelo

bash bootstrap-secrets.sh              # reads ../../../../.env, writes secrets/
docker compose up --build              # default: PROVIDER=openai

Abre http://localhost:3000. Gemini: PROVIDER=gemini docker compose up --build.

Qué hay aquí

backend/ai_openai.py / backend/ai_gemini.py — pipeline de writer/critic + moderación.
frontend/app/page.tsx — caja de tema, copia aprobada.

Detenlo

docker compose down

Ejecútalo en tu máquina

Descarga el proyecto como ZIP y córrelo con Docker. Levanta un backend FastAPI y un frontend Next.js en localhost:3000.

Descargar content-pipeline.zip

unzip content-pipeline.zip
cd content-pipeline
bash bootstrap-secrets.sh   # one-time: pulls API keys into ./secrets
docker compose up --build   # default provider: openai
# or:  PROVIDER=gemini docker compose up --build

Escribe algo, elige un proveedor y ejecuta el mismo código de Código contra la API real. Requiere iniciar sesión.

Entrada

Proveedor

Output

Los mismos módulos que ejecuta el botón Run. El proyecto completo (frontend, Dockerfile, compose) está en el ZIP, pestaña README.

`backend/ai_openai.py`

"""Showcase 2 (OpenAI): a content pipeline — multi-agent + safety.

Give it a topic and get publishable copy: a writer drafts, a critic sharpens it,
the writer revises, and a moderation check clears the result before it's returned.
Multi-agent orchestration (week 19) with a safety gate on the output (week 16) —
the shape of an actual content-generation feature.
"""
from openai import OpenAI

_client = OpenAI()

_MODEL = "gpt-5.4-nano"

_WRITER = "You are a marketing copywriter. Write a short, engaging paragraph on the topic."
_CRITIC = "You are an editor. Give 2-3 specific fixes to make the copy sharper and more concrete."


def _agent(system: str, user: str) -> str:
    return _client.responses.create(model=_MODEL, instructions=system, input=[{"role": "user", "content": user}]).output_text


def _flagged(text: str) -> bool:
    return _client.moderations.create(model="omni-moderation-latest", input=text).results[0].flagged


def run(topic: str) -> str:
    t = topic.strip()
    if not t:
        return "Give a topic to write about (e.g. 'a new noise-cancelling headphone')."
    if _flagged(t):
        return "[topic blocked by moderation]"
    draft = _agent(_WRITER, t)
    critique = _agent(_CRITIC, f"TOPIC: {t}\n\nDRAFT:\n{draft}")
    final = _agent(_WRITER, f"TOPIC: {t}\n\nEditor's fixes:\n{critique}\n\nRewrite the paragraph applying them.")
    if _flagged(final):
        return "[output withheld by moderation]"
    return f"{final}\n\n(passed the writer → critic → revise → moderation pipeline)"

`backend/ai_gemini.py`

"""Showcase 2 (Gemini): a content pipeline — multi-agent + safety.

Same writer → critic → revise → moderate flow, on Gemini.
"""
import os

from google import genai
from google.genai import types

_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])

_MODEL = "gemini-3.1-flash-lite"

_WRITER = "You are a marketing copywriter. Write a short, engaging paragraph on the topic."
_CRITIC = "You are an editor. Give 2-3 specific fixes to make the copy sharper and more concrete."


def _agent(system: str, user: str) -> str:
    r = _client.models.generate_content(
        model=_MODEL, contents=[types.Content(role="user", parts=[types.Part(text=user)])],
        config=types.GenerateContentConfig(system_instruction=system),
    )
    return r.text or ""


def _flagged(text: str) -> bool:
    r = _client.models.generate_content(
        model=_MODEL, contents=[types.Content(role="user", parts=[types.Part(text=text)])],
        config=types.GenerateContentConfig(system_instruction=(
            "Reply 'flag' if this text is hateful, harassing, sexual, violent, or dangerous; else 'ok'. One word.")),
    )
    return "flag" in (r.text or "").lower()


def run(topic: str) -> str:
    t = topic.strip()
    if not t:
        return "Give a topic to write about (e.g. 'a new noise-cancelling headphone')."
    if _flagged(t):
        return "[topic blocked by moderation]"
    draft = _agent(_WRITER, t)
    critique = _agent(_CRITIC, f"TOPIC: {t}\n\nDRAFT:\n{draft}")
    final = _agent(_WRITER, f"TOPIC: {t}\n\nEditor's fixes:\n{critique}\n\nRewrite the paragraph applying them.")
    if _flagged(final):
        return "[output withheld by moderation]"
    return f"{final}\n\n(passed the writer → critic → revise → moderation pipeline)"