Construye un pipeline autónomo de contenido para Astro con Agno, AgentOS y OpenRouter

Ya tienes el destino de publicación perfecto para este tipo de sistema agéntico: una app Astro donde basta con agregar un archivo markdown a content/articles/ para crear una página nueva. Eso significa que lo difícil no es renderizar. Lo difícil es construir un pipeline confiable que funcione en segundo plano y pueda investigar, redactar, revisar, escribir, hacer commit y abrir un pull request sin convertir tu repo en una máquina de alucinaciones.

En este artículo muestro cómo construiría ese pipeline con Agno para la orquestación de agentes y workflows, AgentOS para el runtime y la programación de ejecuciones, y OpenRouter para acceder a los modelos. La arquitectura refleja una distribución de responsabilidades entre modelos para un entorno real:

Redacción de artículos: google/gemini-3-flash-preview
Código + PR + diseño del workflow: z-ai/glm-5
Revisión final / razonamiento crítico: anthropic/claude-4.6-sonnet o anthropic/claude-4.6-opus

Al March 20, 2026, esos identificadores de modelo están disponibles en la documentación vigente de los proveedores o en las páginas de modelos de OpenRouter. Los slugs exactos pueden cambiar, así que considéralos una referencia válida para esa fecha, no constantes eternas.

En resumen

Usa Agno Workflow para el pipeline principal de publicación, no un solo agente sin un flujo definido.
Ejecútalo a través de AgentOS para que pueda trabajar en segundo plano con sesiones, APIs y tareas programadas.
Usa OpenRouter para separar claramente los roles de los modelos: Gemini 3 Flash para investigación, GLM-5 para preparar el paquete y el PR, y Claude 4.6 Sonnet u Opus para la revisión final.
Escribe el artefacto final en content/articles/*.md, porque Astro ya detecta automáticamente los nuevos artículos markdown en este repo.
Agrega LangWatch para trazas, evaluaciones y pruebas de regresión basadas en Scenario, de modo que el sistema sea observable y confiable, no solo autónomo.
Mantén deterministas las acciones irreversibles: deja que los modelos propongan contenido, pero que Python escriba archivos, ejecute git y abra el PR.

El objetivo

Queremos una app que haga esto:

1. Receive a topic or scheduled content brief
2. Research the web and your internal context
3. Build an Astro-ready markdown article
4. Review it critically before publication
5. Write a new file into content/articles/
6. Create a branch, commit, and open a PR
7. Do all of it in the background, usually within a few minutes

En este repo, el directorio de publicación es content/articles/, no simplemente articles/, así que el ejemplo de abajo apunta a la estructura real que ya tienes.

Por qué funciona esta estructura

Este problema no consiste solo en «ejecutar un agente». Es un pipeline repetible con estado, puntos de control, efectos secundarios en el sistema de archivos y operaciones de git. Eso hace que Agno Workflow sea más apropiado que un solo agente sin un flujo definido.

Los Workflows de Agno están diseñados para pasos definidos cuya salida pasa al siguiente paso. AgentOS le da a ese workflow un runtime de producción, APIs, sesiones y programación de ejecuciones. OpenRouter nos da una sola puerta de acceso a varias familias de modelos sin tener que reescribir el código de orquestación cada vez.

La distribución de responsabilidades se ve así:

Agno
  ├─ defines agents, schemas, and workflow steps
  ├─ enforces predictable sequencing
  └─ makes it easy to mix model steps and Python executor steps

AgentOS
  ├─ serves the workflow as an API
  ├─ stores run/session history
  ├─ exposes schedules and run management
  └─ makes background automation operational

OpenRouter
  ├─ routes requests to Gemini / GLM / Claude
  ├─ normalizes tool-calling and request shape
  ├─ supports provider controls and fallbacks
  └─ keeps model switching mostly configuration-level

Astro
  ├─ needs only a markdown file in content/articles/
  ├─ auto-discovers the article
  └─ turns git changes into a previewable site via PR

El archivo de referencia que estamos estudiando

Para que sea más fácil explicarlo, voy a centrarme en un archivo de referencia:

article_publisher.py

Su propósito es simple:

Take a content request in,
run a multi-model workflow,
emit one Astro markdown file,
then create a GitHub PR.

Esta es la estructura general del archivo:

from __future__ import annotations

import os
import re
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from typing import Literal
from uuid import uuid4

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openrouter import OpenRouterResponses
from agno.os import AgentOS
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.workflow import Step, Workflow
from agno.workflow.types import StepInput, StepOutput
from pydantic import BaseModel, Field

A grandes rasgos, el archivo tiene cinco responsabilidades:

Declarar esquemas tipados para cada etapa.
Configurar agentes especializados por modelo a través de OpenRouter.
Definir pasos deterministas del workflow.
Realizar las operaciones con efectos secundarios en el sistema de archivos y git mediante pasos ejecutores en Python.
Servir y programar el workflow mediante AgentOS.

Sección de código crítica 1: Contratos tipados entre pasos

La decisión arquitectónica más importante no es el prompt. Es el límite definido por el esquema entre pasos.

class ResearchPacket(BaseModel):
    topic: str
    angle: str = Field(description="Why this angle is interesting now")
    audience: str
    key_claims: list[str]
    sources: list[str] = Field(description="High quality source URLs")
    outline: list[str]
    risks: list[str] = Field(description="Areas that need verification")


class AstroArticlePackage(BaseModel):
    title: str
    excerpt: str
    tags: list[str]
    slug: str
    seo_title: str
    seo_description: str
    body_markdown: str
    commit_message: str
    pr_title: str
    pr_body: str


class ReviewDecision(BaseModel):
    publish_ready: bool
    blocking_issues: list[str]
    revision_notes: list[str]
    final_title: str | None = None
    final_excerpt: str | None = None

Por qué importa:

Evita que cada paso tenga que adivinar qué quiso decir el anterior.
Hace visibles los errores antes.
Te permite detenerte antes de tocar git si la calidad no es suficiente.
Hace que el workflow siga siendo comprensible cuando revises una ejecución fallida un día después.

Sin esquemas, tu pipeline se convierte en «un texto que más o menos parece markdown entró en otro prompt y ahora solo queda esperar lo mejor».

Sección de código crítica 2: Especialización de modelos mediante OpenRouter

Aquí es donde la configuración ideal para un entorno real se vuelve concreta.

def openrouter_model(primary: str, fallbacks: list[str] | None = None) -> OpenRouterResponses:
    models = [primary, *(fallbacks or [])]
    return OpenRouterResponses(
        id=primary,
        models=models,
        app_name="astro-content-factory",
    )


research_agent = Agent(
    name="Research Drafter",
    model=openrouter_model(
        "google/gemini-3-flash-preview",
        ["google/gemini-2.5-flash"],
    ),
    tools=[DuckDuckGoTools()],
    output_schema=ResearchPacket,
    instructions="""
    Research the topic with current web evidence.
    Prefer primary sources.
    Return a tight article angle, source list, outline, and explicit uncertainty notes.
    """,
)


packaging_agent = Agent(
    name="Astro Packager",
    model=openrouter_model(
        "z-ai/glm-5",
        ["z-ai/glm-5-turbo"],
    ),
    output_schema=AstroArticlePackage,
    instructions="""
    Convert the research packet into a production-ready Astro markdown article package.
    The output must be repo-ready: title, excerpt, tags, slug, SEO fields, markdown body,
    commit message, PR title, and PR body.
    """,
)


review_agent = Agent(
    name="Critical Reviewer",
    model=openrouter_model(
        "anthropic/claude-4.6-sonnet",
        ["anthropic/claude-4.6-opus"],
    ),
    output_schema=ReviewDecision,
    instructions="""
    Review the article package like a demanding editor.
    Block publication if the claims are weak, unsupported, repetitive, or misleading.
    Approve only if the article is publishable with minimal editorial risk.
    """,
)

Este es exactamente el patrón que me gusta:

Gemini 3 Flash sintetiza la investigación con alta capacidad de procesamiento.
GLM-5 produce el paquete adaptado al repo y los metadatos del PR.
Claude 4.6 Sonnet u Opus hace de editor exigente.

Eso no es simplemente fantasear con modelos. Refleja una división práctica del trabajo:

un modelo económico y rápido para explorar
un modelo sólido para tareas de ingeniería y generación de paquetes estructurados
un modelo costoso y confiable para el juicio final

Sección de código crítica 3: Pasos deterministas del workflow

El pipeline principal debe ser un Workflow, no un bucle de chat.

def normalize_request(step_input: StepInput) -> StepOutput:
    raw = str(step_input.input or "").strip()
    if not raw:
        return StepOutput(
            step_name="normalize_request",
            content="Missing topic",
            stop=True,
        )

    return StepOutput(
        step_name="normalize_request",
        content={
            "topic": raw,
            "requested_at": datetime.now(timezone.utc).isoformat(),
        },
    )


def quality_gate(step_input: StepInput) -> StepOutput:
    review = step_input.previous_step_content
    if not isinstance(review, ReviewDecision):
        return StepOutput(
            step_name="quality_gate",
            content="Review step did not return ReviewDecision",
            stop=True,
        )

    if not review.publish_ready:
        return StepOutput(
            step_name="quality_gate",
            content={
                "status": "blocked",
                "issues": review.blocking_issues,
                "notes": review.revision_notes,
            },
            stop=True,
        )

    return StepOutput(
        step_name="quality_gate",
        content={"status": "approved"},
    )


workflow = Workflow(
    name="astro_article_publisher",
    db=SqliteDb(db_file="tmp/article_publisher.db"),
    add_workflow_history_to_steps=True,
    num_history_runs=10,
    steps=[
        Step(name="normalize_request", executor=normalize_request),
        Step(name="research", agent=research_agent),
        Step(name="package_for_astro", agent=packaging_agent),
        Step(name="review", agent=review_agent),
        Step(name="quality_gate", executor=quality_gate),
        Step(name="write_article", executor=write_article_file),
        Step(name="open_pull_request", executor=open_pull_request),
    ],
)

Esta sección es el corazón del sistema.

¿Por qué?

Hace que el flujo sea auditable.
Le da a cada efecto secundario un punto de control con nombre.
Separa los pasos de «pensar» de los pasos que «hacen cosas irreversibles».
Te permite detenerte de forma segura antes de escribir archivos y crear el PR.

Observa el orden:

research -> package -> review -> gate -> write -> PR

Es intencional. Las operaciones de git con efectos secundarios ocurren después de la revisión crítica, no antes.

Sección de código crítica 4: Escribir el archivo markdown de Astro de forma segura

Aquí es donde el workflow abstracto se convierte en un sistema de publicación.

REPO_ROOT = Path(os.environ["ASTRO_REPO_ROOT"]).resolve()
ARTICLE_DIR = REPO_ROOT / "content" / "articles"


def slugify(value: str) -> str:
    normalized = re.sub(r"[^a-zA-Z0-9]+", "-", value.lower()).strip("-")
    return re.sub(r"-{2,}", "-", normalized)


def write_article_file(step_input: StepInput) -> StepOutput:
    package = step_input.get_step_content("package_for_astro")
    review = step_input.get_step_content("review")

    if not isinstance(package, AstroArticlePackage):
        raise ValueError("package_for_astro must return AstroArticlePackage")

    slug = slugify(package.slug or package.title)
    article_path = ARTICLE_DIR / f"{slug}.md"

    if article_path.exists():
        raise FileExistsError(f"Refusing to overwrite existing article: {article_path}")

    final_title = review.final_title if isinstance(review, ReviewDecision) and review.final_title else package.title
    final_excerpt = (
        review.final_excerpt
        if isinstance(review, ReviewDecision) and review.final_excerpt
        else package.excerpt
    )

    markdown = f'''---
title: "{final_title}"
excerpt: "{final_excerpt}"
tags: {package.tags!r}
date: "{datetime.now().date().isoformat()}"
seoTitle: "{package.seo_title}"
seoDescription: "{package.seo_description}"
canonical: "https://luismori.dev/article/{slug}"
---

{package.body_markdown}
'''

    article_path.write_text(markdown, encoding="utf-8")

    return StepOutput(
        step_name="write_article",
        content={
            "status": "written",
            "slug": slug,
            "path": str(article_path),
        },
    )

Esta función parece simple, pero incorpora varias ideas importantes para producción:

Idempotencia: no sobrescribir silenciosamente un slug existente.
Separación de responsabilidades: el modelo propone contenido y Python escribe el archivo.
Normalización final: el slug y el título definitivos se resuelven fuera del modelo.
Corrección específica del repo: escribe en content/articles/, que es lo que esta app Astro realmente carga.

Sección de código crítica 5: Automatización de la rama, el commit y el PR

Aquí es donde muchas demos dejan de ser realistas. Generan markdown, pero una persona todavía tiene que hacer el trabajo de git. Si quieres una publicación realmente autónoma en segundo plano, el workflow debe terminar el trabajo.

def run(cmd: list[str], cwd: Path) -> str:
    completed = subprocess.run(
        cmd,
        cwd=cwd,
        check=True,
        text=True,
        capture_output=True,
    )
    return completed.stdout.strip()


def open_pull_request(step_input: StepInput) -> StepOutput:
    package = step_input.get_step_content("package_for_astro")
    write_result = step_input.get_step_content("write_article")

    if not isinstance(package, AstroArticlePackage):
        raise ValueError("package_for_astro must return AstroArticlePackage")

    branch = f"codex/article-{uuid4().hex[:8]}"

    run(["git", "checkout", "-b", branch], cwd=REPO_ROOT)
    run(["git", "add", str(write_result["path"])], cwd=REPO_ROOT)
    run(["git", "commit", "-m", package.commit_message], cwd=REPO_ROOT)
    run(["git", "push", "-u", "origin", branch], cwd=REPO_ROOT)

    pr_url = run(
        [
            "gh",
            "pr",
            "create",
            "--base",
            "main",
            "--title",
            package.pr_title,
            "--body",
            package.pr_body,
        ],
        cwd=REPO_ROOT,
    )

    return StepOutput(
        step_name="open_pull_request",
        content={"branch": branch, "pr_url": pr_url},
    )

Aquí es donde GLM-5 demuestra su valor. El modelo no ejecuta git. Lo hace Python. Pero GLM-5 proporciona los metadatos estructurados que hacen que la capa de git sea limpia y determinista: mensaje del commit, título del PR, cuerpo del PR y markdown adaptado al repo.

Sección de código crítica 6: Servir el workflow con AgentOS

Ahora lo ponemos en funcionamiento.

agent_os = AgentOS(
    id="astro-content-factory",
    workflows=[workflow],
    scheduler=True,
)

app = agent_os.get_app()

if __name__ == "__main__":
    agent_os.serve(app=app, port=7777, reload=True)

Esta pequeña sección cambia todo el sistema:

tu workflow se convierte en una API
las sesiones y el historial de ejecuciones se pueden persistir
las tareas programadas pueden activarlo
otros servicios pueden llamarlo de forma remota

Esa es la diferencia entre «un script de Python que escribe markdown» y «una app de agentes».

Diagrama de flujo de control

┌──────────────────────┐
│ Topic / Content Brief│
└──────────┬───────────┘
           │
           ▼
┌──────────────────────┐
│ normalize_request    │
│ validate + timestamp │
└──────────┬───────────┘
           │
           ▼
┌──────────────────────┐
│ research             │
│ Gemini 3 Flash       │
│ + web tools          │
└──────────┬───────────┘
           │
           ▼
┌──────────────────────┐
│ package_for_astro    │
│ GLM-5                │
│ markdown + PR data   │
└──────────┬───────────┘
           │
           ▼
┌──────────────────────┐
│ review               │
│ Claude 4.6           │
│ approve / block      │
└──────────┬───────────┘
           │
           ▼
┌──────────────────────┐
│ quality_gate         │
│ stop if weak         │
└───────┬────────┬─────┘
        │        │
   blocked       approved
        │        │
        ▼        ▼
   run ends   write_article
                  │
                  ▼
           open_pull_request
                  │
                  ▼
            branch + PR URL

Diagrama de flujo de datos

Raw topic
  │
  ▼
{ topic, requested_at }
  │
  ▼
ResearchPacket
  ├─ angle
  ├─ sources
  ├─ key_claims
  ├─ outline
  └─ risks
  │
  ▼
AstroArticlePackage
  ├─ title
  ├─ excerpt
  ├─ tags
  ├─ slug
  ├─ seo_title
  ├─ seo_description
  ├─ body_markdown
  ├─ commit_message
  ├─ pr_title
  └─ pr_body
  │
  ▼
ReviewDecision
  ├─ publish_ready
  ├─ blocking_issues
  ├─ revision_notes
  ├─ final_title?
  └─ final_excerpt?
  │
  ▼
File write result
  ├─ slug
  ├─ path
  └─ status
  │
  ▼
PR result
  ├─ branch
  └─ pr_url

Diagrama de relaciones entre componentes

                    ┌─────────────────────────────┐
                    │         AgentOS             │
                    │ API + sessions + schedules  │
                    └──────────────┬──────────────┘
                                   │
                                   ▼
                    ┌─────────────────────────────┐
                    │   Agno Workflow             │
                    │ astro_article_publisher     │
                    └──────┬───────────┬──────────┘
                           │           │
                           │           ▼
                           │    Python Executors
                           │    ├─ normalize_request
                           │    ├─ quality_gate
                           │    ├─ write_article_file
                           │    └─ open_pull_request
                           │
                           ▼
                     Agno Agents
                     ├─ research_agent
                     ├─ packaging_agent
                     └─ review_agent
                           │
                           ▼
                     OpenRouter
                     ├─ Gemini 3 Flash Preview
                     ├─ GLM-5
                     └─ Claude 4.6 Sonnet / Opus
                           │
                           ▼
                     External Systems
                     ├─ Web search
                     ├─ Astro repo filesystem
                     └─ GitHub CLI / PR API

Cómo ejecutarlo en segundo plano

Esta es la distinción práctica más importante de todo el artículo.

Si tu objetivo es que «todo suceda automáticamente en segundo plano y en cuestión de minutos», no conviertas el pipeline principal de publicación en un hook de segundo plano.

Los hooks de segundo plano de AgentOS son excelentes para:

analítica
notificaciones
evaluaciones
webhooks

No son la abstracción adecuada para la ruta principal de publicación, porque la documentación indica explícitamente que los hooks de segundo plano se ejecutan después de enviar la respuesta y no pueden modificar la solicitud ni la respuesta. Son para tareas secundarias no críticas, no para la ruta principal de creación del artefacto.

Para la tarea principal, usa uno de estos patrones:

Una tarea programada de AgentOS
Una ejecución del workflow activada mediante una API
Una cola externa o cron que llame al endpoint del workflow

La versión más limpia para automatizar contenido es:

Scheduler/cron
  -> POST AgentOS workflow endpoint
  -> workflow runs
  -> article file is written
  -> PR is opened
  -> run history is stored

Ejemplo: programar el workflow

AgentOS expone APIs para administrar tareas programadas, así que un script de inicialización independiente puede crear una tarea recurrente:

import httpx

client = httpx.Client(base_url="http://localhost:7777", timeout=30)

client.post(
    "/schedules",
    json={
        "name": "daily-article-run",
        "cron_expr": "0 9 * * 1-5",
        "endpoint": "/workflows/astro_article_publisher/runs",
        "method": "POST",
        "payload": {"message": "Research and draft one article about practical AI engineering"},
        "timezone": "America/Lima",
        "max_retries": 2,
        "retry_delay_seconds": 60,
    },
)

Esto te da ejecuciones recurrentes sin convertir la lógica de publicación en un efecto secundario incómodo de alguna solicitud que no tiene relación.

Agrega LangWatch para monitoreo, observabilidad y confiabilidad

La autonomía sin observabilidad solo hace que los fallos ocurran más rápido.

Si este pipeline va a ejecutarse en segundo plano y abrir PRs por sí solo, necesitas una capa de confiabilidad que responda:

¿Qué paso falló?
¿Qué modelo produjo la salida incorrecta?
¿Qué lista de fuentes se utilizó?
¿La calidad sufrió una regresión después de cambiar un prompt o modelo?
¿Están empeorando con el tiempo la latencia, el costo en tokens y la tasa de aprobación?

Aquí es donde entra LangWatch. La documentación actual de LangWatch lo presenta como una plataforma de observabilidad y trazado basada en OpenTelemetry, con SDKs para Python, TypeScript y Go, además de simulaciones de agentes mediante Scenario.

El modelo mental más simple es:

AgentOS workflow runs the pipeline
         │
         ▼
LangWatch trace wraps the full run
         │
         ├─ span: normalize_request
         ├─ span: research
         ├─ span: package_for_astro
         ├─ span: review
         ├─ span: write_article
         └─ span: open_pull_request
         │
         ▼
Evaluations + dashboards + alerts + scenario history

Sección de código crítica 7: Instrumentar el workflow con LangWatch

La guía actual de Python de LangWatch recomienda llamar a langwatch.setup() al principio y luego usar trazas y spans para capturar operaciones de extremo a extremo y pasos anidados.

import langwatch

langwatch.setup()


@langwatch.trace(name="astro_article_publish_run")
def run_publish_pipeline(topic: str, session_id: str, user_id: str = "content-bot"):
    current_trace = langwatch.get_current_trace()
    current_trace.update(
        metadata={
            "thread_id": session_id,
            "user_id": user_id,
            "workflow_name": "astro_article_publisher",
            "topic": topic,
        }
    )

    return workflow.run(input=topic, session_id=session_id)

Esto te da una traza de nivel superior para todo el intento de publicación. Dentro de esa traza, cada función importante puede decorarse como un span:

@langwatch.span(name="write_article_file")
def write_article_file(step_input: StepInput) -> StepOutput:
    package = step_input.get_step_content("package_for_astro")
    article_path = ARTICLE_DIR / f"{slugify(package.slug or package.title)}.md"

    markdown = render_markdown(package)
    article_path.write_text(markdown, encoding="utf-8")

    langwatch.get_current_span().update(
        input={"slug": package.slug, "title": package.title},
        output={"path": str(article_path), "status": "written"},
    )

    return StepOutput(
        step_name="write_article",
        content={"status": "written", "path": str(article_path)},
    )

Esa instrumentación a nivel de span importa porque, cuando una ejecución sale mal, generalmente no te basta con saber que «el workflow falló». Quieres saber que:

el paso de investigación usó fuentes poco confiables
el paso de revisión bloqueó el artículo
el paso de git falló después de crear la rama

Las trazas de LangWatch permiten inspeccionar esos fallos en lugar de dejarlos como un misterio.

Sección de código crítica 8: Asociar evaluaciones al control de calidad

La documentación actual de evaluaciones de LangWatch describe tres mecanismos útiles:

evaluaciones personalizadas del lado del cliente mediante add_evaluation()
evaluaciones administradas del lado del servidor
guardrails que pueden influir en el flujo de la aplicación

Para este pipeline de publicación, el paso quality_gate es el lugar natural para registrarlas.

@langwatch.span(name="quality_gate")
def quality_gate(step_input: StepInput) -> StepOutput:
    review = step_input.previous_step_content

    if not isinstance(review, ReviewDecision):
        langwatch.get_current_span().add_evaluation(
            name="review-contract-valid",
            passed=False,
            details="Review step did not return ReviewDecision",
            is_guardrail=True,
        )
        return StepOutput(step_name="quality_gate", content="Invalid review payload", stop=True)

    passed = review.publish_ready and not review.blocking_issues
    langwatch.get_current_span().add_evaluation(
        name="publish-ready",
        passed=passed,
        details={
            "blocking_issues": review.blocking_issues,
            "revision_notes": review.revision_notes,
        },
        is_guardrail=True,
    )

    if not passed:
        return StepOutput(
            step_name="quality_gate",
            content={"status": "blocked", "issues": review.blocking_issues},
            stop=True,
        )

    return StepOutput(step_name="quality_gate", content={"status": "approved"})

Este es un patrón de gran impacto:

el workflow sigue decidiendo si debe continuar
LangWatch registra la decisión como datos de evaluación estructurados
puedes seguir la tasa de aprobación y rechazo a lo largo del tiempo entre ejecuciones, prompts o cambios de modelo

Eso transforma «Claude dijo que este artículo era débil» en una métrica observable.

Arquitectura de confiabilidad con LangWatch

Scheduler / API Trigger
        │
        ▼
AgentOS workflow run
        │
        ▼
LangWatch trace
  ├─ metadata: topic, session_id, user_id, workflow_name
  ├─ span: research
  │    └─ evals: source_quality, freshness_check
  ├─ span: package_for_astro
  │    └─ evals: schema_valid, frontmatter_valid
  ├─ span: review
  │    └─ evals: publish_ready, citation_risk
  ├─ span: write_article
  │    └─ evals: file_written
  └─ span: open_pull_request
       └─ evals: pr_opened
        │
        ▼
Dashboard views
  ├─ pass rate by topic type
  ├─ failure rate by step
  ├─ latency by model
  ├─ token cost by run
  └─ regression history by prompt version

Escenarios de prueba con LangWatch Scenario

La observabilidad te dice qué pasó en producción. Las pruebas de Scenario te ayudan a prevenir regresiones antes de llegar a producción.

El framework Scenario de LangWatch permite probar agentes mediante simulaciones. En lugar de escribir aserciones frágiles de entrada y salida, describes una situación realista, dejas que un usuario simulado interactúe con tu agente y evalúas el comportamiento en puntos de control durante la conversación.

Esto se ajusta mucho mejor a este pipeline de contenido porque la confiabilidad no consiste solo en comparar cadenas de texto. Se trata de comprobar si el sistema:

pide aclaraciones cuando el tema es demasiado impreciso
cita fuentes actuales y confiables
evita artículos duplicados
bloquea la publicación cuando la revisión es negativa
nunca abre un PR para un borrador rechazado

Sección de código crítica 9: Adaptar la app de publicación para Scenario

Scenario funciona mediante una interfaz AgentAdapter con un método call(). Para este workflow, el adaptador puede invocar el endpoint del workflow de AgentOS o llamar directamente al workflow.run() local en las pruebas.

import scenario


class ArticlePublisherAdapter(scenario.AgentAdapter):
    async def call(self, input: scenario.AgentInput):
        user_message = input.messages[-1]["content"]
        result = workflow.run(input=user_message, session_id="scenario-test")

        if result.content and isinstance(result.content, dict):
            if result.content.get("status") == "blocked":
                return f"Publication blocked: {result.content}"

        return str(result.content)

Sección de código crítica 10: Pruebas de Scenario para la confiabilidad editorial

Aquí es donde la confiabilidad se vuelve concreta.

import pytest
import scenario


@pytest.mark.agent_test
@pytest.mark.asyncio
async def test_rejects_weakly_supported_topic():
    result = await scenario.run(
        name="reject weak sourcing",
        description="""
        The user asks for an article on a trending claim with weak evidence.
        The pipeline should surface sourcing risk and avoid opening a PR.
        """,
        agents=[
            ArticlePublisherAdapter(),
            scenario.UserSimulatorAgent(),
            scenario.JudgeAgent(),
        ],
        script=[
            scenario.user("Write an article claiming framework X is dead with no primary sources"),
            scenario.agent(),
            scenario.judge(criteria=[
                "The system identifies evidence weakness or uncertainty",
                "The system does not claim the article is ready to publish",
                "The system does not proceed as if a PR was opened",
            ]),
        ],
    )

    assert result.success

Luego agregarías más escenarios para los modos de fallo reales de la app:

Escenario de tema duplicado: ya existe un artículo casi idéntico en content/articles/
Escenario de fuentes deficientes: las fuentes son spam de SEO de baja calidad o rumores de foros
Escenario de bloqueo por revisión: el revisor final rechaza el borrador y el pipeline debe detenerse correctamente
Escenario de fallo de git: falla la creación de la rama o del PR y la ejecución debe mostrar el error sin fingir que tuvo éxito
Escenario exitoso: un tema bien delimitado produce un archivo markdown limpio y metadatos para el PR

Mapa de cobertura de Scenario

Risk                           Scenario Test
────────────────────────────   ───────────────────────────────────────
Hallucinated claims            judge checks for explicit uncertainty
Weak sourcing                  judge checks source credibility behavior
Duplicate content              agent must detect existing article overlap
Bad frontmatter                deterministic validator + scenario checkpoint
PR opened after failed review  scenario ensures blocked runs stop early
Prompt/model regression        batch scenario history shows pass-rate drift

Cómo ayuda LangWatch después de las ejecuciones de Scenario

La documentación actual de simulaciones de LangWatch destaca que, una vez conectado Scenario, las ejecuciones pueden visualizarse en la plataforma de LangWatch, donde puedes:

organizar simulaciones en conjuntos y lotes
inspeccionar conversaciones completas
depurar ejecuciones fallidas
seguir el rendimiento a lo largo del tiempo con el historial de ejecuciones

Eso significa que el ciclo de confiabilidad del workflow queda así:

Change prompt / model / tool
        │
        ▼
Run scenario suite
        │
        ▼
View failures in LangWatch
        │
        ▼
Inspect trace + judge output + conversation
        │
        ▼
Fix workflow / prompts / gating
        │
        ▼
Re-run until pass rate stabilizes

Qué medir en producción

Una vez integrado LangWatch, yo monitorearía primero estas métricas:

tasa de artículos listos para publicar por categoría de tema
tasa de bloqueo por modelo revisor
tasa de éxito al abrir PRs
latencia mediana y P95 por paso del workflow
costo en tokens por artículo exitoso
fallos de calidad de las fuentes a lo largo del tiempo
fallos de detección de duplicados
tasa de aprobación de escenarios en la suite de regresión

Si tuviera que elegir una sola regla práctica de operación, sería esta:

Cada incidente en producción debe convertirse en una prueba de regresión de Scenario.

Así es como la observabilidad se convierte en confiabilidad, en lugar de quedarse solo en dashboards.

Cinco niveles de comprensión

Nivel básico

En el nivel más básico, esta app es una fábrica automática de contenido para un sitio Astro.

Recibe un tema, le pide a un modelo que lo investigue, a otro que convierta esa investigación en un artículo markdown y a un modelo más estricto que lo revise; usa LangWatch para trazar y evaluar la ejecución y, solo entonces, escribe el artículo en content/articles/ y abre un PR.

El propósito no es «la IA escribe blogs». El propósito es:

ahorrar tiempo en el workflow repetitivo de publicación
mantener el repo como fuente de verdad
preservar un punto de revisión humana mediante el pull request

Lo más importante que debes entender en este nivel es que el artefacto de salida no es una respuesta de chat. Es un cambio en git.

Nivel intermedio

En el nivel intermedio, debes ver el sistema como un pipeline de componentes especializados.

Las piezas principales son:

agente de investigación para recopilar evidencia y generar un esquema
agente de preparación para transformar la investigación en markdown listo para Astro y metadatos de git
agente revisor para aprobar o rechazar
trazado y evaluaciones de LangWatch para dar visibilidad a cada ejecución
pruebas de regresión de Scenario para cubrir la confiabilidad con situaciones realistas
pasos ejecutores para acciones locales seguras, como escribir archivos y ejecutar comandos de git
almacenamiento de sesiones del workflow para que las ejecuciones sucesivas puedan recordar lo que ya produjeron

Los métodos y funciones clave son:

output_schema en cada agente para exigir resultados tipados
Step(...) para hacer que cada etapa sea explícita y tenga un nombre
StepInput.get_step_content(...) para leer resultados anteriores de forma segura
StepOutput(stop=True) para detener ejecuciones inseguras o de baja calidad
AgentOS(...).get_app() para exponer el workflow como una API

Este también es el nivel en el que la división entre modelos empieza a resultar natural:

Gemini es bueno para redactar rápidamente a partir de la investigación
GLM-5 es bueno para preparar paquetes orientados a ingeniería
Claude es bueno para decir «no» cuando la calidad no es suficiente

LangWatch agrega la siguiente capa de disciplina:

las trazas te dicen qué pasó
las evaluaciones te dicen si el resultado fue bueno
los escenarios te dicen si sigue siendo bueno después de los cambios

Nivel avanzado

En el nivel avanzado, el archivo revela tres patrones de diseño sólidos.

1. Orquestación guiada por esquemas

Cada paso se comunica mediante un contrato tipado. Esto evita que las desviaciones de los prompts se acumulen a lo largo del pipeline.

2. Diseño híbrido del workflow

El sistema no es pura improvisación agéntica. Combina:

pasos de agentes para trabajo centrado en el lenguaje
pasos ejecutores en Python para acciones locales deterministas
spans de observabilidad y evaluaciones para obtener un comportamiento medible

Ese es un patrón maduro. Los modelos deben decidir el contenido. Python debe encargarse de los efectos secundarios.

3. Efectos secundarios después de la revisión

El workflow retrasa deliberadamente la escritura de archivos y la creación del PR hasta después de la revisión. Es una decisión arquitectónica que prioriza la seguridad.

En este nivel también se ve por qué AgentOS importa más allá de «servir una app». Te ofrece:

ejecución remota
tareas programadas
persistencia de sesiones
historial del workflow

El historial del workflow es especialmente valioso aquí porque permite que futuras ejecuciones accedan a ejecuciones anteriores. Eso ayuda a la app a evitar producir tres artículos casi idénticos en una semana.

LangWatch complementa esto con un historial operativo independiente:

trazas de ejecución
latencia por paso
puntuaciones de evaluación
resultados de lotes de escenarios

Nivel experto

En el nivel experto, dejas de preguntar «¿funciona?» y empiezas a preguntar «¿cómo falla?».

Estos son los principales casos límite y aspectos operativos que debes considerar.

Temas duplicados

Si se solicita el mismo tema dos veces, puedes obtener contenido duplicado o colisiones de slugs. El archivo ya bloquea la sobrescritura de slugs existentes, pero en la práctica también deberías:

comparar con el historial reciente del workflow
comparar con los títulos y slugs de artículos existentes
rechazar temas casi duplicados antes de redactar

Fuentes poco confiables

Un agente de investigación con herramientas de búsqueda aún puede devolver evidencia superficial o circular. Por eso el esquema de investigación incluye:

URLs explícitas de las fuentes
afirmaciones clave
notas sobre riesgos e incertidumbre

El revisor debe hacer fallar la ejecución si las afirmaciones no están realmente respaldadas.

Efectos secundarios de git en repos con cambios sin guardar

Si el árbol de trabajo tiene cambios sin guardar, un paso que crea una rama y hace commit puede mezclar accidentalmente cambios que no están relacionados. En producción, deberías:

ejecutar en un clon o worktree limpio
o comprobar que el árbol de trabajo esté limpio antes de cualquier paso de git

Fallos al crear el PR

gh pr create puede fallar por problemas de autenticación, límites de tasa, remotos faltantes o protecciones de rama. El workflow debe capturar y persistir esos fallos como salida de la ejecución para que sean observables, en lugar de quedar ocultos.

Costo y latencia

Este stack es lo bastante rápido para funcionar en «minutos, no horas», pero solo si mantienes acotados los pasos:

la investigación de Gemini debe producir notas estructuradas, no un ensayo completo más notas, resumen y una versión alternativa
GLM-5 debe preparar el paquete, no volver a investigar
Claude debe revisar, no reescribir todo el artículo desde cero

En otras palabras: prompts acotados, salidas tipadas y un alcance limitado para cada paso son optimizaciones de rendimiento.

LangWatch ayuda a comprobar si esas optimizaciones funcionan porque te permite observar:

cambios graduales en la latencia después de modificar prompts
aumentos de costo después de cambiar modelos
regresiones en la tasa de aprobación después de modificar agentes
fallos recurrentes concentrados en un paso

Por qué los hooks de segundo plano no son la abstracción adecuada para el núcleo

Vale la pena repetirlo porque es un detalle sutil. Los hooks de segundo plano de AgentOS se ejecutan secuencialmente después de la respuesta y están pensados para trabajo no crítico. Son excelentes para:

analítica
notificaciones
evaluaciones asíncronas

Pero tu ruta principal de publicación debe seguir siendo un workflow de primera clase o una ejecución programada, porque necesita producir el artefacto principal y mantener el control sobre la semántica de éxito y fallo.

LangWatch encaja bien aquí porque observa directamente el workflow de primera clase, en lugar de depender de una arquitectura basada únicamente en un canal secundario.

Nivel legendario

En el nivel legendario, la pregunta interesante no es «¿cómo publico un artículo?». Es «¿qué sistema estoy creando si esto funciona?».

En la práctica, estás construyendo una plataforma de operaciones de contenido respaldada por git.

Eso tiene varias implicaciones a largo plazo.

1. El repo se convierte en la base de datos de publicación

Eso es muy potente. El workflow no mantiene estado en el límite con el LLM, pero sí en el límite con git:

las versiones de los artículos viven en commits
la revisión editorial vive en comentarios de PRs
la vista previa del despliegue vive en el entorno de preview

Esto es mucho mejor que almacenar publicaciones generadas por IA en una fila opaca de una base de datos.

2. Tu ventaja competitiva pasa a ser la evaluación, no la generación

Cualquier equipo puede conectar un modelo para que genere markdown. Lo difícil es construir el sistema que lo rodea y que responde:

¿Valía la pena escribir sobre el tema?
¿Eran buenas las fuentes?
¿El artículo duplicaba contenido existente?
¿El PR estaba limpio y era fácil de revisar?

La parte difícil de replicar del sistema no es el borrador. Son los controles, la memoria y el ciclo de calidad.

Aquí es precisamente donde LangWatch Scenario adquiere importancia estratégica. Una vez que tienes un banco de escenarios realistas y un historial de aprobaciones y fallos, tu sistema deja de ser «un prompt que funciona por ahora» y empieza a convertirse en «un workflow con confiabilidad medible».

3. Con el tiempo querrás una cola y un registro de contenido

La demo de un solo archivo es suficiente para demostrar el patrón. Un sistema a mayor escala normalmente evolucionará para incluir:

una tabla o cola de temas
puntuación para detectar duplicados
estados de artículo como queued, researching, drafted, blocked, pr_opened, merged
trazas de observabilidad y evaluaciones
workflows de aprobación para temas de alto impacto

En ese punto, AgentOS sigue siendo una capa sólida de orquestación y runtime, pero probablemente agregarás:

Postgres en lugar de SQLite
un proceso worker o una cola de tareas
puntuación de contenido y recuperación de publicaciones anteriores
evaluaciones automatizadas después de cada ejecución

4. La mejora futura más limpia es un sistema de revisión en dos pasadas

Si esta fuera mi arquitectura a largo plazo, la haría evolucionar hacia esto:

Pass 1: fast reviewer
  -> catches obvious issues cheaply

Pass 2: expensive reviewer
  -> runs only if the article is close to publishable

Eso te da un mejor control de costos sin perder una evaluación final rigurosa.

5. La mejora de mayor impacto es la «memoria de investigación»

En lugar de empezar cada ejecución desde cero, yo almacenaría:

fuentes utilizadas anteriormente
afirmaciones rechazadas
enfoques de artículos aceptados
vacíos de cobertura de etiquetas y temas

Así, el workflow mejora su criterio editorial con el tiempo, en lugar de simplemente volverse más rápido.

Lo combinaría con una memoria de confiabilidad:

trazas de LangWatch con fallos
problemas recurrentes que bloquean la publicación
regresiones de escenarios por versión del prompt
patrones de fallo por modelo

Esto te da tanto memoria editorial como memoria operativa.

Configuración recomendada para un entorno real

Si estuviera construyendo esto para usarlo realmente, lo ejecutaría así:

Frontend / Trigger
  ├─ small admin UI or cron trigger
  └─ optional "generate article" button

Agent Runtime
  └─ AgentOS serving one named workflow

Workflow
  ├─ research with Gemini 3 Flash
  ├─ package for Astro + PR with GLM-5
  ├─ final review with Claude 4.6 Sonnet
  └─ escalate to Claude 4.6 Opus only for hard reviews

Persistence
  ├─ workflow sessions in SQLite for local dev
  └─ Postgres in production

Publish Target
  └─ content/articles/*.md in Astro repo

Delivery
  ├─ git branch
  ├─ commit
  └─ GitHub PR

Si quieres una regla práctica por encima de todas las demás, usa esta:

Deja que los modelos generen propuestas estructuradas. Deja que Python realice las acciones irreversibles.

Esa sola regla evita una amplia clase de problemas de confiabilidad.

Conclusión

La idea ganadora aquí no es «usar tres modelos sofisticados». Es esta:

deterministic workflow
+ typed step outputs
+ repo-aware packaging
+ critical review gate
+ git-native delivery

Eso es lo que convierte una demo de LLM en un sistema autónomo de publicación.

Agno te da la abstracción del workflow. AgentOS te da el runtime, las sesiones y la programación de ejecuciones. OpenRouter te da enrutamiento de modelos y flexibilidad de proveedores. Astro te da un formato de destino maravillosamente simple: un archivo markdown en content/articles/.

Combina todo eso y podrás construir una app en segundo plano que investigue, redacte y abra PRs de artículos automáticamente en cuestión de minutos.