---
title: "Tutorial de Agno Workflows: escritor de artículos con IA y Streamlit"
description: "Construye un workflow multiagente de Agno que investiga, esquematiza, redacta y edita artículos técnicos para principiantes, todo dentro de una interfaz de Streamlit. Actualizado para Agno 2.x."
date: 2025-04-15
categories: ["ai"]
tags: ["ai-agents","agno","streamlit"]
---
import Button from "@components/widgets/Button.astro";
import Notice from "@components/widgets/Notice.astro";
import ListCheck from "@components/widgets/ListCheck.astro";
import Accordion from "@components/widgets/Accordion.astro";
Un workflow de Agno puede automatizar todo el proceso de escritura de artículos: investigar, esquematizar, redactar y editar, todo detrás de una interfaz de Streamlit. Si buscabas un ejemplo práctico de workflow multiagente que funcione con la versión actual de Agno, es este.
Escribir artículos técnicos para principiantes lleva tiempo. Necesitas fuentes creíbles, una estructura lógica, explicaciones claras y un borrador final pulido. Un pipeline de IA multiagente gestiona cada etapa como una tarea especializada, lo que produce mejores resultados que pedirle a un único modelo que "escriba un artículo sobre X".
Este tutorial te guía para construir un workflow de Agno con 4 agentes usando **Workflows 2.0** (la API actual), **Streamlit** para la interfaz y **uv** para configurar el proyecto. El pipeline investiga un tema, esquematiza el artículo, redacta cada sección y edita el borrador final. Todo se muestra en una app de Streamlit donde puedes descargar el resultado en Markdown.
Este tutorial está dirigido a Agno v2.9+ (Workflows 2.0). La versión anterior usaba API de Agno v1 que se eliminaron en la reescritura de v2. Si estás en `agno` 1.x, este código no funcionará. Consulta la guía oficial de migración a v2 para ver el mapeo completo.
Si eres nuevo en la programación asistida por IA, consulta primero [primeros pasos para programar con IA](/es/guia-programacion-ia-principiantes/).
## Lo que vas a construir: un workflow multiagente de Agno
El `BeginnerArticleWorkflow` coordina cuatro agentes especializados para producir un artículo pulido:
Investiga un tema usando búsqueda web y extracción de contenido
Esquematiza el artículo con título, secciones y palabras clave SEO
Redacta cada sección con fragmentos de código y explicaciones para principiantes
Edita el borrador final para darle claridad y coherencia
Guarda en caché los resultados intermedios en SQLite para ahorrar costes de API
Muestra todo en una app de Streamlit con descarga en Markdown
El pipeline tiene este aspecto:
```
[Researcher] → [Outliner] → [Writer] → [Editor] → [Streamlit UI]
```
Cada agente produce una salida estructurada (modelos de Pydantic) que alimenta la siguiente etapa. El Researcher busca y resume fuentes. El Outliner crea la estructura del artículo. El Writer redacta cada sección. El Editor pule el borrador final.
## Requisitos previos
Python 3.12+ (Agno v2 requiere ≥3.9, <4; se recomienda 3.12 o 3.13)
Conocimientos básicos de Python y soltura con la línea de comandos
Una máquina con acceso a internet (Linux, macOS o Windows)
No hace falta instalar `sqlalchemy` manualmente. El `SqliteDb` de Agno v2 trae sus propias dependencias. El paquete `duckduckgo-search` ha pasado a llamarse `ddgs`.
Si no has usado Agno antes, empieza con [primeros pasos con Agno Agents](/agno-get-start/).
## Paso 1: configurar tu entorno con uv
Usaremos **uv** para configurar el proyecto. Si no conoces uv, consulta nuestra guía de [primeros pasos con uv](/uv-get-start/).
### Instalar uv
- **macOS/Linux**:
```bash
curl -LsSf https://astral.sh/uv/install.sh | sh
```
- **Windows (PowerShell)**:
```powershell
irm https://astral.sh/uv/install.ps1 | iex
```
- Verifica:
```bash
uv --version
```
### Crear el proyecto
```bash
uv init agno-article-writer
cd agno-article-writer
uv python pin 3.12
uv venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
```
### Instalar las dependencias
```bash
uv add agno streamlit openai ddgs crawl4ai
```
El paquete `duckduckgo-search` ha pasado a llamarse `ddgs`. Si ves errores de instalación de `duckduckgo-search`, usa `ddgs` en su lugar. `DuckDuckGoTools` de Agno lo necesita.
Verifica la instalación:
```bash
uv run python -c "import agno; print(agno.__version__)"
```
Debería mostrar `2.9.x` o posterior.
### Configurar las variables de entorno
Crea un archivo `.env`:
```bash
echo "OPENROUTER_API_KEY=your_key_here" > .env
```
Sustituye `your_key_here` por tu clave de API de OpenRouter. Así mantienes tu clave segura y se carga automáticamente con `python-dotenv` (que Agno incluye como dependencia).
Agno envía telemetría anónima por defecto (los prompts y las salidas nunca se envían). Desactívala con `AGNO_TELEMETRY=false` en tu `.env` o en el entorno si te importa.
## Paso 2: entender la estructura de un workflow de Agno
Antes de meternos en el código, veamos en qué se diferencian los Workflows 2.0 de la API anterior.
### Workflows 2.0: qué ha cambiado
Agno v2 eliminó por completo los Workflows v1. El patrón antiguo de heredar de `Workflow`, sobrescribir `run()` y usar `self.session_state` ya no existe. Este es el mapeo:
| v1 (eliminada) | v2 (actual) |
|---|---|
| `class MyWorkflow(Workflow)` + `def run()` | `Workflow(name=..., steps=[...], db=...)` |
| `from agno.run.response import RunResponse, RunEvent` | `from agno.run.agent import RunOutput` / `from agno.run.workflow import WorkflowRunOutput` |
| `from agno.storage.sqlite import SqliteStorage` | `from agno.db.sqlite import SqliteDb` |
| `storage=SqliteStorage(...)` | `db=SqliteDb(...)` |
| `response_model=` | `output_schema=` |
| `add_history_to_messages=` | `add_history_to_context=` |
| `DuckDuckGoTools(search=True, news=True)` | `DuckDuckGoTools(enable_search=True, enable_news=True)` |
| `self.session_state` (persistente) | diccionario `session_state` pasado a `run()` o gestionado por el workflow |
Si migras desde Agno v1, casi todas las API usadas en la versión anterior de este tutorial han cambiado. Consulta la guía oficial de migración a v2 para ver el mapeo completo.
### Pasos del workflow (investigar → esquematizar → redactar → editar)
En Workflows 2.0, un workflow es una lista de pasos. Cada paso es una función de Python que recibe un diccionario `session_state` y devuelve el estado actualizado. Los cuatro pasos corresponden a cuatro agentes especializados:
- **Paso 1 (research_step)**: llama al agente Researcher y guarda `ResearchSummary` en el estado.
- **Paso 2 (outline_step)**: llama al agente Outliner y guarda `ArticleOutline` en el estado.
- **Paso 3 (write_step)**: recorre las secciones del esquema, llama al agente Writer por sección y ensambla el borrador.
- **Paso 4 (edit_step)**: llama al agente Editor y guarda el artículo final pulido.
```
[Researcher] → ResearchSummary (Pydantic)
↓
[Outliner] → ArticleOutline (Pydantic)
↓
[Writer] → SectionDraft per section (Pydantic)
↓
[Editor] → Final Markdown string
```
### Los agentes y sus roles
- **Researcher**: usa búsqueda de DuckDuckGo + extracción con Crawl4ai → `ResearchSummary` estructurado.
- **Outliner**: toma la investigación → produce un `ArticleOutline` con título, secciones y palabras clave.
- **Writer**: toma investigación + esquema + título de sección → produce un `SectionDraft` por sección.
- **Editor**: toma el borrador ensamblado → Markdown pulido.
### Modelos de Pydantic (esquema de salida)
Los mismos cuatro modelos de antes (`ResearchFinding`, `ResearchSummary`, `ArticleOutline`, `SectionDraft`), pero ahora se usan como `output_schema` en los agentes en lugar de `response_model`. Pydantic impone los tipos y valida la salida en cada etapa, así el código posterior no se rompe con datos malformados.
## Paso 3: escribir el código del workflow de Agno
Aquí está el código completo reescrito para Agno v2.9+ / Workflows 2.0.
### Importaciones y configuración (Workflows 2.0)
```python
import os
import json
import logging
import re
from textwrap import dedent
from typing import Dict, List, Optional
import streamlit as st
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from agno.agent import Agent
from agno.models.openrouter import OpenRouter
from agno.run.agent import RunOutput
from agno.run.workflow import WorkflowRunOutput
from agno.db.sqlite import SqliteDb
from agno.tools.crawl4ai import Crawl4aiTools
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.workflow import Workflow
```
Cambios clave respecto a v1:
- `RunResponse` → `RunOutput` (de `agno.run.agent`)
- `SqliteStorage` → `SqliteDb` (de `agno.db.sqlite`)
- `Workflow` ya no se hereda. Se construye con `steps=`
### Modelos de Pydantic
```python
class ResearchFinding(BaseModel):
url: str = Field(..., description="Source URL of the information.")
summary: str = Field(..., description="Concise summary of the key information.")
content_snippet: Optional[str] = Field(None, description="A relevant quote from the source.")
class ResearchSummary(BaseModel):
key_findings: List[ResearchFinding] = Field(..., description="List of key findings.")
overall_summary: str = Field(..., description="Overall synthesis of the research.")
class ArticleOutline(BaseModel):
title: str = Field(..., description="Proposed article title, engaging for beginners.")
sections: List[str] = Field(..., description="Section titles in logical learning order.")
keywords: List[str] = Field(..., description="SEO keywords including beginner terms.")
class SectionDraft(BaseModel):
section_title: str = Field(..., description="The section being drafted.")
content: str = Field(..., description="Section content in Markdown with code, tables, lists.")
```
Son las mismas estructuras de datos del artículo de v1. La diferencia está en cómo se usan: como `output_schema` en los agentes en lugar de `response_model`.
### Definir los agentes
```python
load_dotenv()
OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
def create_agents(api_key: str, model_id: str, max_tokens: int):
"""Create the four specialized agents."""
model = OpenRouter(id=model_id, api_key=api_key, max_tokens=max_tokens)
writer_model = OpenRouter(id=model_id, api_key=api_key, max_tokens=max(max_tokens, 8192))
editor_model = OpenRouter(id=model_id, api_key=api_key, max_tokens=max(max_tokens, 8192))
researcher = Agent(
name="TechResearcherBeginnerFocus",
model=model,
tools=[
DuckDuckGoTools(enable_search=True, enable_news=True),
Crawl4aiTools(max_length=10000),
],
description="Expert tech researcher finding and synthesizing information for beginners.",
instructions=dedent("""\
Your goal is to research the given topic thoroughly, focusing on information accessible to beginners.
1. Use DuckDuckGo to find 5-7 highly relevant and recent online sources (articles, docs, blog posts).
2. Prioritize: Official 'getting started' guides, tutorials, reputable tech blogs known for clear explanations.
3. For each promising source URL, use web_crawler to extract main content.
4. Synthesize the information, identifying key concepts, simple definitions, introductory code examples, benefits, and common use cases.
5. Output MUST be ResearchSummary JSON.
"""),
output_schema=ResearchSummary,
markdown=True,
add_history_to_context=False,
)
outliner = Agent(
name="BeginnerArticleOutliner",
model=model,
description="Structures technical articles logically for beginners.",
instructions=dedent("""\
Given a research summary, create a logical article outline tailored for beginners.
1. Title: Craft a compelling title indicating the topic and suggesting it's beginner-friendly.
2. Sections: Structure logically for learning. Start with basics, build up. Include:
- Introduction (What is it? Why care?)
- Key Concepts/Terminology
- Getting Started / Core How-To
- Code Examples Explained
- Benefits / Use Cases
- Potential Challenges for Beginners
- Conclusion / Next Steps
3. Keywords: Include relevant SEO keywords focusing on beginner terms.
4. Output MUST be ArticleOutline JSON.
"""),
output_schema=ArticleOutline,
markdown=False,
add_history_to_context=False,
)
writer = Agent(
name="BeginnerTechWriter",
model=writer_model,
description="Writes a detailed technical article section specifically for beginners.",
instructions=dedent("""\
You are a skilled senior technical writer specializing in making complex topics easy for beginners.
You will receive: a) The overall research summary, b) The article outline, c) The specific section_title.
Write content ONLY for the specified section_title, targeting complete beginners.
1. Accuracy: Use the research summary for technical facts.
2. Clarity: Explain concepts simply. Define technical terms immediately. Use analogies.
3. Code: Provide clear, step-by-step explanations for each line. Describe expected input/output.
4. Formatting: Use Markdown extensively: sub-headings, bold, inline code, code blocks, lists, tables.
5. Engagement: Start sections engagingly. Write in an encouraging tone.
6. Detail: Aim for ~400+ words. Prioritize clarity over strict word count.
7. Focus: Do NOT write the main section title. Focus only on the requested section.
8. Output MUST be SectionDraft JSON with section_title and content.
"""),
output_schema=SectionDraft,
markdown=True,
add_history_to_context=False,
)
editor = Agent(
name="BeginnerFocusedEditor",
model=editor_model,
description="Polishes a full article draft, ensuring clarity for beginners.",
instructions=dedent("""\
You are reviewing a complete article draft assembled from sections written for beginners.
Perform final polishing:
1. Clarity: Read from a beginner's perspective. Is jargon explained? Are explanations thorough?
2. Consistency: Ensure consistent terminology, tone, and code formatting across sections.
3. Flow: Edit for smooth transitions, grammar, spelling, punctuation.
4. Markdown: Check correct formatting. Ensure headings match the outline sections.
5. Completeness: Check if sections seem reasonably detailed.
6. No Major Rewrites: Do not add substantial new content. Focus on polishing.
7. Return the final, polished Markdown article as a string.
"""),
markdown=True,
add_history_to_context=False,
)
return researcher, outliner, writer, editor
```
Fíjate en los cambios de v2:
- `DuckDuckGoTools(enable_search=True, enable_news=True)` (antes `search=True, news=True`)
- `output_schema=ResearchSummary` (antes `response_model=`)
- `add_history_to_context=False` (antes `add_history_to_messages=False`)
- Los ID de modelo usan el slug de OpenRouter directamente (p. ej., `openai/gpt-5-mini`), sin el prefijo `openrouter/`
### Definir los pasos del workflow
```python
def default_serializer(obj):
if isinstance(obj, set):
return list(obj)
raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")
def create_workflow(api_key: str, model_id: str, max_tokens: int, use_cache: bool = True):
"""Create the 4-step article writing workflow."""
researcher, outliner, writer, editor = create_agents(api_key, model_id, max_tokens)
def research_step(session_state: dict) -> dict:
topic = session_state["topic"]
topic_key = session_state["topic_key"]
if use_cache and f"research_{topic_key}" in session_state:
logger.info("Using cached research data.")
return session_state
logger.info("--- Starting Research Stage ---")
response: RunOutput = researcher.run(topic)
session_state[f"research_{topic_key}"] = response.content.model_dump()
logger.info("Research complete.")
return session_state
def outline_step(session_state: dict) -> dict:
topic_key = session_state["topic_key"]
if use_cache and f"outline_{topic_key}" in session_state:
logger.info("Using cached outline data.")
return session_state
logger.info("--- Starting Outline Stage ---")
research_json = json.dumps(session_state[f"research_{topic_key}"], default=default_serializer)
response: RunOutput = outliner.run(research_json)
session_state[f"outline_{topic_key}"] = response.content.model_dump()
logger.info("Outline complete.")
return session_state
def write_step(session_state: dict) -> dict:
topic_key = session_state["topic_key"]
if use_cache and f"draft_{topic_key}" in session_state:
logger.info("Using cached draft.")
return session_state
logger.info("--- Starting Writing Stage ---")
research_data = session_state[f"research_{topic_key}"]
outline_data = session_state[f"outline_{topic_key}"]
all_section_content = {}
for i, section_title in enumerate(outline_data["sections"]):
logger.info(f"Writing section {i+1}/{len(outline_data['sections'])}: '{section_title}'")
writer_input = json.dumps({
"research_data": research_data,
"outline_data": outline_data,
"section_title": section_title,
}, default=default_serializer)
max_retries = 2
for attempt in range(max_retries + 1):
try:
response: RunOutput = writer.run(writer_input)
all_section_content[section_title] = response.content.content
break
except Exception as e:
logger.warning(f"Writer attempt {attempt+1} failed for '{section_title}': {e}")
if attempt == max_retries:
all_section_content[section_title] = f"\n_[Content generation failed for '{section_title}']_\n"
# Assemble draft
parts = [f"# {outline_data['title']}\n"]
for section_title in outline_data["sections"]:
parts.append(f"\n## {section_title}\n")
content = all_section_content.get(section_title, f"\n_[Content missing]_\n")
parts.append(content.strip() + "\n")
session_state[f"draft_{topic_key}"] = "\n".join(parts)
logger.info("Writing complete.")
return session_state
def edit_step(session_state: dict) -> dict:
topic_key = session_state["topic_key"]
if use_cache and f"final_{topic_key}" in session_state:
logger.info("Using cached final article.")
return session_state
logger.info("--- Starting Editing Stage ---")
draft = session_state[f"draft_{topic_key}"]
outline_data = session_state[f"outline_{topic_key}"]
editor_input = json.dumps({
"draft_content": draft,
"outline": outline_data,
}, default=default_serializer)
try:
response: RunOutput = editor.run(editor_input)
session_state[f"final_{topic_key}"] = response.content
except Exception as e:
logger.warning(f"Editor failed: {e}. Using assembled draft.")
session_state[f"final_{topic_key}"] = draft
logger.info("Workflow completed successfully.")
return session_state
db = SqliteDb(db_file="tmp/agno_beginner_workflows.db")
workflow = Workflow(
name="BeginnerArticleWorkflow",
description="Generates beginner-friendly technical articles.",
db=db,
steps=[research_step, outline_step, write_step, edit_step],
)
return workflow
```
Diferencias clave respecto a v1:
- Cada paso es una función normal que recibe un diccionario `session_state`, no un método de una subclase de `Workflow`.
- `session_state` se persiste automáticamente cuando se define `db=SqliteDb(...)`.
- Se acabaron los helpers manuales `self.session_state.get()` / `add_data_to_cache()`.
- El workflow se construye con `Workflow(name=..., steps=[...], db=...)`.
### Interfaz de Streamlit
```python
st.set_page_config(page_title="Beginner Article Workflow", page_icon="✍️", layout="wide")
with st.sidebar:
st.title("⚙️ Configuration")
api_key = st.text_input(
"OpenRouter API Key", type="password", key="api_key_input_wf",
value=os.getenv("OPENROUTER_API_KEY", ""), help="Required."
)
available_models = [
"openai/gpt-5-mini",
"openai/gpt-4o",
"google/gemini-2.5-flash",
"anthropic/claude-sonnet-4",
"meta-llama/llama-4-maverick",
]
model_id = st.selectbox("Select Model", options=available_models, index=0, key="model_select_wf")
max_tokens = st.slider("Max Completion Tokens", 2048, 16384, 8192, 1024, key="max_tokens_slider_wf")
use_cache = st.toggle("Use Cache", value=True, key="use_cache_wf", help="Reuse results from SQLite.")
os.makedirs("tmp", exist_ok=True)
st.sidebar.caption("Cache DB: tmp/agno_beginner_workflows.db")
if st.button("Clear Chat History", key="clear_chat_wf"):
st.session_state.messages_wf = []
st.rerun()
st.title("✍️ Agno Workflow: Beginner Article Writer")
st.markdown("Enter a topic and the AI team will research, outline, write, and edit a beginner-friendly article.")
if "messages_wf" not in st.session_state:
st.session_state.messages_wf = []
for msg_index, message_info in enumerate(st.session_state.messages_wf):
role = message_info.get("role", "assistant")
content = message_info.get("content", "")
is_final = message_info.get("is_final", False)
is_error = message_info.get("is_error", False)
with st.chat_message(role):
st.markdown(content, unsafe_allow_html=is_error)
if is_final:
safe_name = re.sub(r'[^\w\-]+', '_', message_info.get("topic", "article")).strip('_').lower()
st.download_button(
"Download Article (Markdown)", data=content,
file_name=f"{safe_name}.md", mime="text/markdown",
key=f"download_hist_wf_{msg_index}",
)
if user_query := st.chat_input("Enter article topic..."):
if not api_key:
st.error("🚨 Please enter your OpenRouter API key.")
else:
st.session_state.messages_wf.append({"role": "user", "content": user_query})
with st.chat_message("user"):
st.markdown(user_query)
with st.chat_message("assistant"):
placeholder = st.empty()
placeholder.markdown("⏳ Running workflow... check terminal logs for progress.")
try:
topic_key = re.sub(r'[^\w\-]+', '_', user_query).strip('_').lower() or "article"
workflow = create_workflow(api_key, model_id, max_tokens, use_cache)
result: WorkflowRunOutput = workflow.run(
input=user_query,
session_state={"topic": user_query, "topic_key": topic_key},
)
final_article = result.session_state.get(f"final_{topic_key}", "")
placeholder.empty()
if final_article:
placeholder.markdown(final_article)
st.session_state.messages_wf.append({
"role": "assistant", "content": final_article,
"is_final": True, "topic": user_query,
})
st.download_button(
"Download Article (Markdown)", data=final_article,
file_name=f"{topic_key}.md", mime="text/markdown",
key=f"download_now_wf_{topic_key}",
)
else:
placeholder.markdown("⚠️ No content generated. Check logs.")
except Exception as e:
placeholder.empty()
err = f"❌ **Workflow Failed:**\n```\n{type(e).__name__}: {e}\n```\nCheck terminal for full traceback."
placeholder.markdown(err)
st.session_state.messages_wf.append({"role": "assistant", "content": err, "is_error": True})
```
Nota: `st.experimental_rerun()` se eliminó en Streamlit 1.61. Usa `st.rerun()` en su lugar.
## Paso 4: ejecutar el workflow de Agno con Streamlit
Guarda el código como `beginner_article_workflow_streamlit.py` y ejecuta:
```bash
uv run streamlit run beginner_article_workflow_streamlit.py
```
Abre `http://localhost:8501` en tu navegador.
1. Introduce tu clave de API de OpenRouter en la barra lateral (o confía en el `.env`).
2. Selecciona un modelo. `openai/gpt-5-mini` es un buen valor por defecto (barato y capaz).
3. Ajusta los tokens máximos (8192 funciona bien para la mayoría de artículos).
4. Mantén la caché activada para ahorrar costes de API en ejecuciones repetidas.
5. Introduce un tema como "Introduction to Python for Beginners".
6. Observa el indicador de progreso y luego descarga el artículo como `.md`.
**Comprueba que funciona:** tras la primera ejecución, busca en el terminal el mensaje `Workflow completed successfully` y confirma que la interfaz de Streamlit muestra el artículo con un botón de descarga. El archivo de caché de SQLite debería aparecer en `tmp/agno_beginner_workflows.db`.
## Paso 5: cómo funciona en acción
Para un tema como "Introduction to Python for Beginners":
1. **Comprobación de caché** → fallo → ejecuta el pipeline completo.
2. **Investigación**: DuckDuckGo encuentra 5-7 fuentes, Crawl4ai extrae el contenido → JSON `ResearchSummary` con hallazgos y síntesis.
3. **Esquema**: produce un plan estructurado:
```json
{
"title": "Getting Started with Python: A Beginner's Guide",
"sections": ["What is Python?", "Setting Up Python", "Your First Program", "Key Concepts", "Next Steps"],
"keywords": ["python tutorial", "learn python", "beginner"]
}
```
4. **Writer**: redacta cada sección con ejemplos de código y explicaciones para principiantes.
5. **Editor**: revisa coherencia, gramática y formato.
6. **Salida**: artículo final mostrado en Streamlit + guardado en la caché de SQLite.
**Revisa el gasto de tokens** después de una ejecución:
```python
# After workflow.run(), result.metrics contains token counts:
print(f"Tokens: {result.metrics.input_tokens} in / {result.metrics.output_tokens} out")
print(f"Duration: {result.metrics.duration:.1f}s")
```
En la segunda ejecución con el mismo tema, la caché entra en acción y omite todas las llamadas a la API.
## Paso 6: ejemplo de salida
Esto es una muestra de lo que produce el pipeline para "Introduction to Python":
````markdown
# Getting Started with Python: A Beginner's Guide
## What is Python?
Python is a simple, versatile programming language used for web development,
data science, automation, and more. It reads almost like English, which makes
it one of the easiest languages to learn.
## Setting Up Python
1. **Download**: Go to [python.org](https://www.python.org) and grab the latest version.
2. **Install**: Run the installer. Check "Add Python to PATH."
3. **Verify**: Open a terminal and type:
```bash
python --version
```
You should see something like `Python 3.12.4`. If you get an error, Python
isn't in your PATH. Reinstall and check the PATH option.
````
La salida presenta explicaciones claras, desgloses de código paso a paso y un tono amigable para principiantes en todo el texto.
## Solución de problemas
**Síntoma:** `Error 402` o `insufficient credits` en los registros del terminal.
**Solución:** comprueba tu saldo de OpenRouter en [openrouter.ai/credits](https://openrouter.ai/credits). El pipeline multiagente hace muchas llamadas a la API por artículo (búsquedas de investigación + rastreos + una llamada del writer por sección + una llamada del editor). Un artículo típico cuesta 0,02-0,05 $ con `openai/gpt-5-mini`, pero puede ser más con modelos más grandes.
**Síntoma:** `ddgs` lanza errores de conexión o devuelve resultados vacíos.
**Solución:** DuckDuckGo puede limitar la tasa de consultas rápidas. Añade `timeout=15` a `DuckDuckGoTools()`. Espacia las ejecuciones de prueba. Si sigues topando con bloqueos, considera cambiar a `WebSearchTools` con otro backend o usar una API de búsqueda de pago.
**Síntoma:** `BrowserType.launch: Executable doesn't exist` o errores de descarga de Chromium.
**Solución:** ejecuta `uv run crawl4ai-setup` para instalar los navegadores de Playwright. En un VPS sin interfaz gráfica puede que necesites dependencias del sistema:
```bash
sudo apt install -y libnss3 libatk-bridge2.0-0 libdrm2 libxcomposite1 libxdamage1 libxrandr2 libgbm1 libpango-1.0-0 libcairo2 libasound2
```
**Síntoma:** `OperationalError: unable to open database file`.
**Solución:** asegúrate de que el directorio `tmp/` existe y tiene permisos de escritura. El código de Streamlit hace `os.makedirs("tmp", exist_ok=True)`, pero revisa los permisos del sistema de archivos si ejecutas como otro usuario.
**Síntoma:** `Model not found` o un 404 de OpenRouter.
**Solución:** los ID de modelo de Agno v2 usan el slug de OpenRouter directamente (p. ej., `openai/gpt-5-mini`), SIN el prefijo `openrouter/`. Consulta los modelos disponibles en [openrouter.ai/models](https://openrouter.ai/models).
**Síntoma:** el workflow devuelve un artículo antiguo para un tema nuevo.
**Solución:** desactiva la caché con el interruptor de la barra lateral, o borra `tmp/agno_beginner_workflows.db` y reinicia.
## ¿Por qué usar workflows de Agno?
- **Ligero**: Agno es una biblioteca de Python enfocada, no un framework pesado. La creación de agentes es rápida.
- **Salida estructurada**: el `output_schema` de Pydantic garantiza un flujo de datos con tipado seguro entre agentes. Sin parsear texto en bruto.
- **Caché integrada**: la persistencia de `SqliteDb` + `session_state` ahorra costes de API en ejecuciones repetidas.
- **Primitivas de Workflows 2.0**: `Step`, `Condition`, `Router`, `Loop`, `Parallel` para pipelines complejos más allá de este ejemplo lineal.
- **Integración con Streamlit**: prototipado rápido de interfaz sin tocar frameworks de frontend.
Si quieres un patrón multiagente distinto, consulta [cómo construir un equipo de investigación con IA usando Agno y Streamlit](/agno-squad/).
## Notas sobre costes y operación
Un artículo típico (5 secciones) usa unos 15-25k tokens de entrada y 10-15k de salida entre todos los agentes. Con `openai/gpt-5-mini`, eso son unos 0,02-0,05 $ por artículo.
**Consejos para ahorrar:**
- Usa un modelo barato para investigación y esquema; un modelo más potente solo para writer/editor si la calidad importa.
- Mantén la caché activada. Volver a ejecutar el mismo tema omite por completo todas las llamadas a la API.
- Ajusta `max_tokens` adecuadamente. La investigación no necesita 8k tokens; 4k suele ser suficiente.
**Ejecutarlo en un VPS:** la app de Streamlit funciona bien en un VPS de 1 GB. Para acceso permanente, despliégala detrás de un proxy inverso. [Hetzner Cloud VPS](https://go.bitdoze.com/hetzner) es una opción sólida y asequible, y [Hostinger VPS](https://go.bitdoze.com/hostinger-vps) funciona como alternativa económica.
_Los precios de los VPS subieron en general en 2026 — si estás reconsiderando una máquina alquilada, mira [qué cambió y cuándo gana un mini PC](/es/subidas-precios-vps/)._
Para las instrucciones de despliegue, consulta [desplegar tu app de Streamlit en un VPS](/streamlit-deploy-vps-cloudflare/).
## Próximos pasos
- **Añade más herramientas**: búsqueda de YouTube, búsqueda de GitHub o [herramientas MCP para tu agente de Agno](/es/agno-herramientas-mcp-context7/) para una investigación más rica.
- **Personaliza para distintas audiencias**: ajusta las instrucciones de los agentes para lectores intermedios o avanzados.
- **Usa modelos distintos**: prueba la [API gratuita de Groq en Streamlit](/groq-api-mistral-streamlit/) para pasos de investigación sin coste.
- **Despliega a producción**: aloja en un VPS con Dokploy. Consulta [desplegar un proyecto de Python con uv y Dokploy](/dokploy-python-railpack-uv/).
- **Construye otros agentes**: [crea un bot de Discord con IA usando Agno](/es/crear-tu-propio-agente-ia/).
- **Explora frameworks**: compara [frameworks web de Python](/best-python-web-frameworks/) si quieres ir más allá de Streamlit.
## Conclusión
Has construido un workflow de Agno con 4 agentes que investiga, esquematiza, redacta y edita artículos técnicos para principiantes, todo detrás de una interfaz de Streamlit. El pipeline usa Workflows 2.0 (la API actual), guarda en caché los resultados intermedios en SQLite para ahorrar costes y produce Markdown descargable.
Prueba distintos temas — "Learn JavaScript", "Docker for Beginners", "What is an API?" — y observa cómo los agentes los manejan. Ajusta las instrucciones de los agentes para dirigirte a distintos niveles de habilidad. El framework es lo bastante flexible como para ampliarlo con más agentes, herramientas distintas o modelos alternativos.
Para más información sobre cómo construir con Agno, consulta [primeros pasos para programar con IA](/es/guia-programacion-ia-principiantes/).
## Código completo
Este código está dirigido a Agno ≥2.9.0 y Streamlit ≥1.61. Cópialo y pégalo en `beginner_article_workflow_streamlit.py` y ejecútalo con `uv run streamlit run beginner_article_workflow_streamlit.py`.
```python
# beginner_article_workflow_streamlit.py
# Agno Workflows 2.0 — Beginner Article Writer with Streamlit
# Requires: agno>=2.9.0, streamlit>=1.61, openai, ddgs, crawl4ai
import os
import json
import logging
import re
from textwrap import dedent
from typing import Dict, List, Optional
import streamlit as st
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from agno.agent import Agent
from agno.models.openrouter import OpenRouter
from agno.run.agent import RunOutput
from agno.run.workflow import WorkflowRunOutput
from agno.db.sqlite import SqliteDb
from agno.tools.crawl4ai import Crawl4aiTools
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.workflow import Workflow
# --- Logging ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# --- Configuration ---
load_dotenv()
OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
# --- JSON Serializer ---
def default_serializer(obj):
if isinstance(obj, set):
return list(obj)
raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")
# --- Pydantic Models ---
class ResearchFinding(BaseModel):
url: str = Field(..., description="Source URL of the information.")
summary: str = Field(..., description="Concise summary of the key information.")
content_snippet: Optional[str] = Field(None, description="A relevant quote from the source.")
class ResearchSummary(BaseModel):
key_findings: List[ResearchFinding] = Field(..., description="List of key findings.")
overall_summary: str = Field(..., description="Overall synthesis of the research.")
class ArticleOutline(BaseModel):
title: str = Field(..., description="Proposed article title, engaging for beginners.")
sections: List[str] = Field(..., description="Section titles in logical learning order.")
keywords: List[str] = Field(..., description="SEO keywords including beginner terms.")
class SectionDraft(BaseModel):
section_title: str = Field(..., description="The section being drafted.")
content: str = Field(..., description="Section content in Markdown.")
# --- Agent Factory ---
def create_agents(api_key: str, model_id: str, max_tokens: int):
model = OpenRouter(id=model_id, api_key=api_key, max_tokens=max_tokens)
writer_model = OpenRouter(id=model_id, api_key=api_key, max_tokens=max(max_tokens, 8192))
editor_model = OpenRouter(id=model_id, api_key=api_key, max_tokens=max(max_tokens, 8192))
researcher = Agent(
name="TechResearcherBeginnerFocus",
model=model,
tools=[
DuckDuckGoTools(enable_search=True, enable_news=True),
Crawl4aiTools(max_length=10000),
],
description="Expert tech researcher finding and synthesizing information for beginners.",
instructions=dedent("""\
Your goal is to research the given topic thoroughly, focusing on information accessible to beginners.
1. Use DuckDuckGo to find 5-7 highly relevant and recent online sources.
2. Prioritize: Official 'getting started' guides, tutorials, reputable tech blogs.
3. For each promising source URL, use web_crawler to extract main content.
4. Synthesize the information, identifying key concepts, simple definitions, code examples, benefits, and common use cases.
5. Output MUST be ResearchSummary JSON.
"""),
output_schema=ResearchSummary,
markdown=True,
add_history_to_context=False,
)
outliner = Agent(
name="BeginnerArticleOutliner",
model=model,
description="Structures technical articles logically for beginners.",
instructions=dedent("""\
Given a research summary, create a logical article outline tailored for beginners.
1. Title: Compelling, clearly indicating the topic and suggesting beginner-friendliness.
2. Sections: Structure logically for learning. Include Introduction, Key Concepts, Getting Started, Code Examples, Benefits, Challenges, Conclusion.
3. Keywords: Relevant SEO keywords focusing on beginner terms.
4. Output MUST be ArticleOutline JSON.
"""),
output_schema=ArticleOutline,
markdown=False,
add_history_to_context=False,
)
writer = Agent(
name="BeginnerTechWriter",
model=writer_model,
description="Writes a detailed technical article section for beginners.",
instructions=dedent("""\
You are a skilled senior technical writer making complex topics easy for beginners.
You receive: research summary, article outline, and a specific section_title.
Write content ONLY for the specified section_title.
1. Use research for technical facts.
2. Explain concepts simply. Define terms immediately. Use analogies.
3. Provide step-by-step code explanations with expected input/output.
4. Use Markdown: sub-headings, bold, inline code, code blocks, lists, tables.
5. Write in an encouraging tone. Aim for ~400+ words.
6. Output MUST be SectionDraft JSON.
"""),
output_schema=SectionDraft,
markdown=True,
add_history_to_context=False,
)
editor = Agent(
name="BeginnerFocusedEditor",
model=editor_model,
description="Polishes a full article draft for beginners.",
instructions=dedent("""\
Review a complete article draft assembled from sections.
1. Check clarity from a beginner's perspective.
2. Ensure consistent terminology, tone, and formatting.
3. Edit for smooth transitions, grammar, spelling.
4. Verify correct Markdown formatting.
5. Do not add substantial new content. Focus on polishing.
6. Return the final polished Markdown article as a string.
"""),
markdown=True,
add_history_to_context=False,
)
return researcher, outliner, writer, editor
# --- Workflow Factory ---
def create_workflow(api_key: str, model_id: str, max_tokens: int, use_cache: bool = True):
researcher, outliner, writer, editor = create_agents(api_key, model_id, max_tokens)
def research_step(session_state: dict) -> dict:
topic = session_state["topic"]
topic_key = session_state["topic_key"]
if use_cache and f"research_{topic_key}" in session_state:
logger.info("Using cached research data.")
return session_state
logger.info("--- Starting Research Stage ---")
response: RunOutput = researcher.run(topic)
session_state[f"research_{topic_key}"] = response.content.model_dump()
logger.info("Research complete.")
return session_state
def outline_step(session_state: dict) -> dict:
topic_key = session_state["topic_key"]
if use_cache and f"outline_{topic_key}" in session_state:
logger.info("Using cached outline data.")
return session_state
logger.info("--- Starting Outline Stage ---")
research_json = json.dumps(session_state[f"research_{topic_key}"], default=default_serializer)
response: RunOutput = outliner.run(research_json)
session_state[f"outline_{topic_key}"] = response.content.model_dump()
logger.info("Outline complete.")
return session_state
def write_step(session_state: dict) -> dict:
topic_key = session_state["topic_key"]
if use_cache and f"draft_{topic_key}" in session_state:
logger.info("Using cached draft.")
return session_state
logger.info("--- Starting Writing Stage ---")
research_data = session_state[f"research_{topic_key}"]
outline_data = session_state[f"outline_{topic_key}"]
all_section_content = {}
for i, section_title in enumerate(outline_data["sections"]):
logger.info(f"Writing section {i+1}/{len(outline_data['sections'])}: '{section_title}'")
writer_input = json.dumps({
"research_data": research_data,
"outline_data": outline_data,
"section_title": section_title,
}, default=default_serializer)
max_retries = 2
for attempt in range(max_retries + 1):
try:
response: RunOutput = writer.run(writer_input)
all_section_content[section_title] = response.content.content
break
except Exception as e:
logger.warning(f"Writer attempt {attempt+1} failed for '{section_title}': {e}")
if attempt == max_retries:
all_section_content[section_title] = f"\n_[Content generation failed for '{section_title}']_\n"
parts = [f"# {outline_data['title']}\n"]
for section_title in outline_data["sections"]:
parts.append(f"\n## {section_title}\n")
content = all_section_content.get(section_title, f"\n_[Content missing]_\n")
parts.append(content.strip() + "\n")
session_state[f"draft_{topic_key}"] = "\n".join(parts)
logger.info("Writing complete.")
return session_state
def edit_step(session_state: dict) -> dict:
topic_key = session_state["topic_key"]
if use_cache and f"final_{topic_key}" in session_state:
logger.info("Using cached final article.")
return session_state
logger.info("--- Starting Editing Stage ---")
draft = session_state[f"draft_{topic_key}"]
outline_data = session_state[f"outline_{topic_key}"]
editor_input = json.dumps({
"draft_content": draft,
"outline": outline_data,
}, default=default_serializer)
try:
response: RunOutput = editor.run(editor_input)
session_state[f"final_{topic_key}"] = response.content
except Exception as e:
logger.warning(f"Editor failed: {e}. Using assembled draft.")
session_state[f"final_{topic_key}"] = draft
logger.info("Workflow completed successfully.")
return session_state
db = SqliteDb(db_file="tmp/agno_beginner_workflows.db")
return Workflow(
name="BeginnerArticleWorkflow",
description="Generates beginner-friendly technical articles.",
db=db,
steps=[research_step, outline_step, write_step, edit_step],
)
# --- Streamlit UI ---
st.set_page_config(page_title="Beginner Article Workflow", page_icon="✍️", layout="wide")
with st.sidebar:
st.title("⚙️ Configuration")
api_key = st.text_input(
"OpenRouter API Key", type="password", key="api_key_input_wf",
value=os.getenv("OPENROUTER_API_KEY", ""), help="Required."
)
available_models = [
"openai/gpt-5-mini",
"openai/gpt-4o",
"google/gemini-2.5-flash",
"anthropic/claude-sonnet-4",
"meta-llama/llama-4-maverick",
]
model_id = st.selectbox("Select Model", options=available_models, index=0, key="model_select_wf")
max_tokens = st.slider("Max Completion Tokens", 2048, 16384, 8192, 1024, key="max_tokens_slider_wf")
use_cache = st.toggle("Use Cache", value=True, key="use_cache_wf", help="Reuse results from SQLite.")
os.makedirs("tmp", exist_ok=True)
st.sidebar.caption("Cache DB: tmp/agno_beginner_workflows.db")
if st.button("Clear Chat History", key="clear_chat_wf"):
st.session_state.messages_wf = []
st.rerun()
st.title("✍️ Agno Workflow: Beginner Article Writer")
st.markdown("Enter a topic and the AI team will research, outline, write, and edit a beginner-friendly article.")
if "messages_wf" not in st.session_state:
st.session_state.messages_wf = []
for msg_index, message_info in enumerate(st.session_state.messages_wf):
role = message_info.get("role", "assistant")
content = message_info.get("content", "")
is_final = message_info.get("is_final", False)
is_error = message_info.get("is_error", False)
with st.chat_message(role):
st.markdown(content, unsafe_allow_html=is_error)
if is_final:
safe_name = re.sub(r'[^\w\-]+', '_', message_info.get("topic", "article")).strip('_').lower()
st.download_button(
"Download Article (Markdown)", data=content,
file_name=f"{safe_name}.md", mime="text/markdown",
key=f"download_hist_wf_{msg_index}",
)
if user_query := st.chat_input("Enter article topic..."):
if not api_key:
st.error("🚨 Please enter your OpenRouter API key.")
else:
st.session_state.messages_wf.append({"role": "user", "content": user_query})
with st.chat_message("user"):
st.markdown(user_query)
with st.chat_message("assistant"):
placeholder = st.empty()
placeholder.markdown("⏳ Running workflow... check terminal logs for progress.")
try:
topic_key = re.sub(r'[^\w\-]+', '_', user_query).strip('_').lower() or "article"
workflow = create_workflow(api_key, model_id, max_tokens, use_cache)
result: WorkflowRunOutput = workflow.run(
input=user_query,
session_state={"topic": user_query, "topic_key": topic_key},
)
final_article = result.session_state.get(f"final_{topic_key}", "")
placeholder.empty()
if final_article:
placeholder.markdown(final_article)
st.session_state.messages_wf.append({
"role": "assistant", "content": final_article,
"is_final": True, "topic": user_query,
})
st.download_button(
"Download Article (Markdown)", data=final_article,
file_name=f"{topic_key}.md", mime="text/markdown",
key=f"download_now_wf_{topic_key}",
)
else:
placeholder.markdown("⚠️ No content generated. Check logs.")
except Exception as e:
placeholder.empty()
err = f"❌ **Workflow Failed:**\n```\n{type(e).__name__}: {e}\n```\nCheck terminal for full traceback."
placeholder.markdown(err)
st.session_state.messages_wf.append({"role": "assistant", "content": err, "is_error": True})
```