Crafting Beginner-Friendly Tech Articles with Agno Workflows and Streamlit

Writing a technical article that’s clear, engaging, and beginner-friendly is no small feat. You need to research credible sources, structure the content logically, write detailed explanations, and polish it to perfection—all while keeping it accessible. What if you could automate this process with AI? Enter Agno, a lightweight Python library that lets you build multi-agent workflows to handle complex tasks like article writing. Pair it with Streamlit for a slick user interface and uv for a blazing-fast setup, and you’ve got a powerful tool to create high-quality content in minutes.

In this tutorial, we’ll guide you through building an Agno Workflow called BeginnerArticleWorkflow, based on the beginner_article_workflow_streamlit.py code. This workflow researches a topic, outlines an article, writes beginner-focused sections, and edits the final draft, all displayed in a Streamlit app. We’ll break down the code in detail, explain each component, and show you how to set it up. By the end, you’ll have a fully functional article-writing pipeline that’s perfect for bloggers, educators, or anyone who wants to make tech accessible. Let’s get started!

What You’ll Build

The BeginnerArticleWorkflow automates the creation of technical articles tailored for beginners. Here’s what it does:

Researches a topic using web searches and content extraction, prioritizing beginner-friendly sources.
Outlines the article with a clear, logical structure, including a title and SEO keywords.
Writes detailed sections with code snippets, explanations, and Markdown formatting.
Edits the draft for clarity, consistency, and polish.
Caches intermediate results in a SQLite database to save time.
Displays everything in a Streamlit app, where you can input topics and download the article as Markdown.

The workflow uses four Agno Agents:

Researcher: Finds and summarizes sources.
Outliner: Creates the article structure.
Writer: Crafts each section.
Editor: Polishes the final draft.

Prerequisites

Before diving in, ensure you have:

Python 3.12 or later (we’ll use uv to manage it).
An OpenRouter API key from OpenRouter.
Basic Python knowledge and comfort with the command line.
A desire to create awesome content with AI!

Step 1: Setting Up Your Environment with uv

We’ll use uv, a super-fast package manager, to set up our project. If you’re new to uv, check our guide on getting started with uv.

Installing uv

macOS/Linux:

curl -LsSf https://astral.sh/uv/install.sh | sh

Windows (PowerShell):

irm https://astral.sh/uv/install.ps1 | iex

Verify installation:
```
uv --version
```

Creating the Project

Initialize a new project:

uv init agno-article-writer
cd agno-article-writer

Pin Python to 3.12:
```
uv python pin 3.12
```

Create a virtual environment:

uv venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

Install dependencies:

uv add agno streamlit python-dotenv pydantic openai duckduckgo-search crawl4ai sqlalchemy

Setting Up Environment Variables

Create a .env file:

echo "OPENROUTER_API_KEY=your_openrouter_key" > .env

Replace your_openrouter_key with your OpenRouter API key.
Why? This keeps your key secure and loads it automatically with python-dotenv.

Pro Tip: Always use a .env file to avoid hardcoding sensitive data in your code.

Step 2: Understanding the Workflow Structure

The BeginnerArticleWorkflow is an Agno Workflow that coordinates four agents to produce a polished article. Let’s explore its high-level structure.

Workflow Stages

Cache Check: Looks for a cached article to skip redundant work.
Research: Finds beginner-friendly sources and summarizes them.
Outline: Creates a structured article plan.
Writing: Writes each section with clear, beginner-focused content.
Editing: Polishes the draft for consistency and clarity.

Agents and Their Roles

Researcher: Uses DuckDuckGo for searches and Crawl4ai to extract content, focusing on tutorials and guides.
Outliner: Designs a logical article structure with a catchy title and SEO keywords.
Writer: Crafts detailed sections, explaining technical concepts simply and including code where relevant.
Editor: Refines the draft, ensuring it’s beginner-friendly and well-formatted.

Pydantic Models

ResearchFinding: Stores a source URL, summary, and optional snippet.
ResearchSummary: Combines multiple findings with an overall summary.
ArticleOutline: Defines the title, sections, and keywords.
SectionDraft: Holds a section’s title and Markdown content.

These models ensure data is structured and validated at each step.

Step 3: Diving into the Code

Let’s break down the beginner_article_workflow_streamlit.py code, explaining each part with a focus on clarity.

Imports and Setup

import os
import json
import logging
import re
import traceback
import time
from textwrap import dedent
from typing import Dict, Iterator, List, Optional

import streamlit as st
from dotenv import load_dotenv
from pydantic import BaseModel, Field, ValidationError

from agno.agent import Agent
from agno.models.openrouter import OpenRouter
from agno.run.response import RunEvent, RunResponse
from agno.storage.sqlite import SqliteStorage
from agno.tools.crawl4ai import Crawl4aiTools
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.workflow import Workflow

Purpose: Imports libraries for:
- File handling (os), JSON processing, logging, and regex.
- Streamlit for the UI, Pydantic for data validation.
- Agno components for workflows, agents, and tools.

Logging Setup:

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

Logs progress and errors to the terminal, helping debug issues.

Environment Variables:

load_dotenv()
OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")

Loads the API key from .env using python-dotenv.

Custom JSON Serializer

def default_serializer(obj):
    """JSON serializer for objects not serializable by default json code"""
    if isinstance(obj, set):
        return list(obj)
    raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable by default_serializer")

Purpose: Converts sets to lists for JSON serialization, used when passing data to agents.
Why? Some Pydantic models may include sets, which JSON doesn’t support natively.

Pydantic Models

class ResearchFinding(BaseModel):
    url: str = Field(..., description="Source URL of the information.")
    summary: str = Field(..., description="Concise summary of the key information relevant to the topic.")
    content_snippet: Optional[str] = Field(None, description="A relevant short quote or snippet from the source.")

class ResearchSummary(BaseModel):
    key_findings: List[ResearchFinding] = Field(..., description="A list of key findings from the research.")
    overall_summary: str = Field(..., description="A brief overall synthesis of the research conducted.")

class ArticleOutline(BaseModel):
    title: str = Field(..., description="Proposed title for the article, engaging for beginners.")
    sections: List[str] = Field(..., description="A list of section titles for the article structure...")
    keywords: List[str] = Field(..., description="List of relevant SEO keywords...")

class SectionDraft(BaseModel):
    section_title: str = Field(..., description="The title of the section being drafted.")
    content: str = Field(..., description="The drafted content for this section, formatted in Markdown...")

ResearchFinding:
- Captures one source’s URL, a summary, and an optional quote.
- Example: A tutorial’s URL with a summary of its key points.
ResearchSummary:
- Aggregates multiple findings and adds a synthesis.
- Ensures research is structured for the next stage.
ArticleOutline:
- Defines the article’s title (e.g., “Python for Beginners”), sections (e.g., “What is Python?”), and keywords (e.g., “learn python”).
SectionDraft:
- Holds one section’s title and content, formatted in Markdown.
Why Pydantic? Enforces data types and validates output, preventing errors like missing fields.

Workflow Class

class BeginnerArticleWorkflow(Workflow):
    description: str = "Generates beginner-friendly technical articles."
    researcher: Agent
    outliner: Agent
    writer: Agent
    editor: Agent

Inherits: From agno.workflow.Workflow, providing caching and session management.
Attributes: Declares four agents as class variables, initialized later.

Initialization

def __init__(
    self,
    api_key: str,
    model_id: str,
    max_tokens: int,
    session_id: str,
    storage: Optional[SqliteStorage] = None,
    debug_mode: bool = False,
    max_writer_retries: int = 2,
):
    super().__init__(session_id=session_id, storage=storage, debug_mode=debug_mode)
    self.max_writer_retries = max_writer_retries

    if not api_key:
        raise ValueError("OpenRouter API Key is required for BeginnerArticleWorkflow.")

    common_model_args = {"id": model_id, "api_key": api_key, "max_tokens": max_tokens}
    writer_tokens = max(max_tokens, 8192)
    editor_tokens = max(max_tokens, 8192)
    writer_model_args = {"id": model_id, "api_key": api_key, "max_tokens": writer_tokens}
    editor_model_args = {"id": model_id, "api_key": api_key, "max_tokens": editor_tokens}

Parameters:
- api_key: OpenRouter key for model access.
- model_id: LLM identifier (e.g., openrouter/optimus-alpha).
- max_tokens: Limits model output size.
- session_id: Unique ID for caching.
- storage: SQLite storage for caching.
- debug_mode: Logs extra details if True.
- max_writer_retries: Number of retries for writing sections.
Token Settings:
- Ensures Writer and Editor have at least 8192 tokens for longer outputs.
Validation:
- Raises an error if no API key is provided.

Agent Definitions

Each agent is an Agent instance with specific tools and instructions.

Researcher:

self.researcher = Agent(
    name="TechResearcherBeginnerFocus",
    model=OpenRouter(**common_model_args),
    tools=[DuckDuckGoTools(search=True, news=True), Crawl4aiTools(max_length=10000)],
    description="Expert tech researcher finding and synthesizing information...",
    instructions=dedent("""\
        Your goal is to research the given topic thoroughly, focusing on information accessible to beginners...
    """),
    response_model=ResearchSummary, markdown=True, add_history_to_messages=False, exponential_backoff=True
)

Tools: DuckDuckGo for searches, Crawl4ai for extracting content.
Instructions: Prioritizes beginner-friendly sources (tutorials, guides) and outputs a ResearchSummary.
Settings:
- markdown=True: Formats output nicely.
- exponential_backoff=True: Retries on API failures.
- add_history_to_messages=False: No chat history, as it’s a one-shot task.

Outliner:

self.outliner = Agent(
    name="BeginnerArticleOutliner",
    model=OpenRouter(**common_model_args),
    description="Structures technical articles logically for beginners.",
    instructions=dedent("""\
        Given a research summary, create a logical article outline tailored for beginners...
    """),
    add_history_to_messages=False, response_model=ArticleOutline, markdown=False, exponential_backoff=True
)

Takes ResearchSummary and produces an ArticleOutline.
Suggests sections like “Introduction,” “Key Concepts,” and “Next Steps.”
Focuses on logical learning progression.

Writer:

self.writer = Agent(
    name="BeginnerTechWriter",
    model=OpenRouter(**writer_model_args),
    description="Writes a detailed, engaging technical article *section* specifically for beginners.",
    instructions=dedent("""\
        You are a skilled senior technical writer specializing in making complex topics easy for **beginners**...
    """),
    response_model=SectionDraft, add_history_to_messages=False, markdown=True, exponential_backoff=True
)

Writes one section at a time, using research and outline.
Emphasizes clarity, simple explanations, and Markdown formatting.
Explains code step-by-step, e.g., for a Python script:
```
```python
print("Hello, World!")
```
This line outputs “Hello, World!” to the screen…

Editor:

self.editor = Agent(
    name="BeginnerFocusedEditor",
    model=OpenRouter(**editor_model_args),
    description="Polishes a full article draft, ensuring clarity for beginners.",
    instructions=dedent("""\
        You are reviewing a complete article draft (in Markdown) assembled from sections...
    """),
    add_history_to_messages=False, markdown=True, exponential_backoff=True
)

Refines the draft for consistency, grammar, and beginner-friendliness.
Checks Markdown formatting and section alignment.

Caching Methods

def get_cached_data(self, key: str) -> Optional[Dict]:
    return self.session_state.get(key)

def add_data_to_cache(self, key: str, data: BaseModel):
    logger.info(f"Caching data for key: {key}")
    self.session_state[key] = data.model_dump()

def get_cached_final_article(self, topic_key: str) -> Optional[str]:
    key = f"final_article_{topic_key}"
    return self.session_state.get(key)

def add_final_article_to_cache(self, topic_key: str, article: str):
    key = f"final_article_{topic_key}"
    logger.info(f"Caching final article for key: {key}")
    self.session_state[key] = article

Purpose: Store and retrieve research, outline, sections, and final article.
Storage: Uses SqliteStorage to persist data in tmp/agno_beginner_workflows.db.
Benefit: Skips redundant API calls, saving time and costs.

Run Method

def run(self, topic: str, use_cache: bool = True) -> Iterator[RunResponse]:
    logger.info(f"Starting BeginnerArticleWorkflow for topic: '{topic}'")
    topic_key = re.sub(r'[^\w\-]+', '_', topic).strip('_').lower()

Stages:

Cache Check:

if use_cache:
    cached_article = self.get_cached_final_article(topic_key)
    if cached_article:
        logger.info("Returning cached final article.")
        yield RunResponse(event=RunEvent.workflow_completed, content=cached_article)
        return

Returns cached article if available.

Research:

research_cache_key = f"research_{topic_key}"
research_data: Optional[ResearchSummary] = None
if use_cache:
    cached_research_dict = self.get_cached_data(research_cache_key)
    if cached_research_dict:
        try:
            research_data = ResearchSummary.model_validate(cached_research_dict)
            logger.info("Using cached research data.")
        except ValidationError as e:
            logger.warning(f"Cached research data invalid: {e}. Re-running research.")
if research_data is None:
    research_response: RunResponse = self.researcher.run(topic)
    research_data = research_response.content
    self.add_data_to_cache(research_cache_key, research_data)

Checks cache, runs Researcher if needed, and caches results.

Outline:
- Similar logic, producing an ArticleOutline.

Writing:

for i, section_title in enumerate(outline_data.sections):
    writer_input_dict = {
        "research_data": research_data.model_dump(),
        "outline_data": outline_data.model_dump(),
        "section_title": section_title
    }
    writer_input_json = json.dumps(writer_input_dict, default=default_serializer)
    for attempt in range(self.max_writer_retries + 1):
        section_response = self.writer.run(writer_input_json)
        if section_response and isinstance(section_response.content, SectionDraft):
            all_section_content[section_title] = section_response.content.content
            break

Writes each section, retries on failure, and stores content.

Assembly:

assembled_draft_parts = [f"# {outline_data.title}\n"]
for section_title in outline_data.sections:
    assembled_draft_parts.append(f"\n## {section_title}\n")
    section_content = all_section_content.get(section_title, f"\n_[Content missing]_\n")
    assembled_draft_parts.append(section_content.strip() + "\n")
assembled_draft = "\n".join(assembled_draft_parts)

Combines sections into a draft.

Editing:

editor_input_dict = {"draft_content": assembled_draft, "outline": outline_data.model_dump()}
editor_response: RunResponse = self.editor.run(json.dumps(editor_input_dict, default=default_serializer))
final_article = editor_response.content

Polishes the draft and caches the result.

Output:

self.add_final_article_to_cache(topic_key, final_article)
yield RunResponse(event=RunEvent.workflow_completed, content=final_article)

Yields the final article.

Streamlit Interface

st.set_page_config(page_title="Beginner Article Workflow", page_icon="✍️", layout="wide")

Sidebar:

Configures API key, model, max tokens, and caching.

Example:

st.session_state.api_key = st.text_input("OpenRouter API Key", type="password", ...)
st.session_state.model_id = st.selectbox("Select Model", options=available_models, ...)

Main UI:

Displays chat history and accepts topic input.

Shows progress and final article, with a download button:

st.download_button(
    label="Download Article (Markdown)",
    data=final_article_content,
    file_name=f"{safe_filename}.md",
    mime="text/markdown",
    key=f"download_now_wf_{topic_key}"
)

Step 4: Running the Workflow

Save the code as beginner_article_workflow_streamlit.py.

Run the app:

uv run streamlit run beginner_article_workflow_streamlit.py

Open http://localhost:8501 in your browser.

How to Use It

Configure:
- Enter your OpenRouter API key (or use .env).
- Choose a model (e.g., openrouter/optimus-alpha).
- Set max tokens (8192 is fine).
- Enable caching to save time.
Enter a Topic: Try “Introduction to Python for Beginners.”
Watch It Work: The app shows progress and displays the article.
Download: Save the article as a .md file.

Step 5: How It Works in Action

For a topic like “Introduction to Python for Beginners”:

Cache Check: Looks for a cached article in tmp/agno_beginner_workflows.db.
Research: Finds tutorials on DuckDuckGo, crawls them, and creates a ResearchSummary.

Outline: Produces:

{
    "title": "Getting Started with Python: A Beginner’s Guide",
    "sections": ["What is Python?", "Setting Up Python", "Your First Program", ...],
    "keywords": ["python tutorial", "learn python", "beginner"]
}

Writing: Writes sections like:

## Setting Up Python
Let’s install Python:
1. Visit [python.org](https://www.python.org)...

Editing: Ensures clarity and consistency.
Output: Displays the article and caches it.

Step 6: Example Output

# Getting Started with Python: A Beginner’s Guide

## What is Python?
Python is a simple, versatile programming language...

## Setting Up Python
1. **Download**: Go to [python.org](https://www.python.org)...

Features:
- Clear explanations.
- Step-by-step code breakdowns.
- Beginner-friendly tone.

Troubleshooting

API Key Issues:
- Verify your key in .env or the sidebar.
No Output:
- Check terminal logs.
- Set debug_mode=True in the workflow.
Cache Problems:
- Delete tmp/agno_beginner_workflows.db or disable caching.
Slow Response:
- Try a faster model like google/gemini-flash-1.5.
- Increase max_tokens for longer sections.

Why Use Agno Workflows?

Automation: Saves hours compared to manual writing.
Specialization: Each agent focuses on one task, improving quality.
Caching: Reduces API costs and speeds up repeats.
Streamlit: Makes it accessible to non-coders.
Flexibility: Easy to tweak for different audiences.

Next Steps

Enhance Research: Add YouTube or GitHub tools.
Customize Output: Adjust instructions for intermediate learners.
Deploy: Host on Streamlit Cloud.
Extend: Add a keyword optimizer or social media generator.

Explore more in our series:

Conclusion

You’ve built an AI-powered article-writing pipeline that makes creating beginner-friendly tech content a breeze. With Agno, Streamlit, and uv, you’ve turned a complex task into an automated, user-friendly process. Try topics like “Learn JavaScript” or “AI Basics” and see your AI team shine. Happy writing! 🚀

Complete Code

# beginner_article_workflow_streamlit.py
import os
import json
import logging
import re
import traceback
import time
from textwrap import dedent
from typing import Dict, Iterator, List, Optional

import streamlit as st
from dotenv import load_dotenv
from pydantic import BaseModel, Field, ValidationError

# Agno Imports
from agno.agent import Agent
from agno.models.openrouter import OpenRouter
from agno.run.response import RunEvent, RunResponse
from agno.storage.sqlite import SqliteStorage
from agno.tools.crawl4ai import Crawl4aiTools
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.workflow import Workflow

# --- Basic Logging Setup ---
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")

# --- Custom JSON Serializer ---
def default_serializer(obj):
    """JSON serializer for objects not serializable by default json code"""
    if isinstance(obj, set):
        return list(obj)
    raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable by default_serializer")

# --- Pydantic Models ---
class ResearchFinding(BaseModel):
    url: str = Field(..., description="Source URL of the information.")
    summary: str = Field(..., description="Concise summary of the key information relevant to the topic.")
    content_snippet: Optional[str] = Field(None, description="A relevant short quote or snippet from the source.")

class ResearchSummary(BaseModel):
    key_findings: List[ResearchFinding] = Field(..., description="A list of key findings from the research.")
    overall_summary: str = Field(..., description="A brief overall synthesis of the research conducted.")

class ArticleOutline(BaseModel):
    title: str = Field(..., description="Proposed title for the article, engaging for beginners.")
    sections: List[str] = Field(..., description="A list of section titles for the article structure, logical for a beginner learning the topic.")
    keywords: List[str] = Field(..., description="List of relevant SEO keywords, including beginner-related terms.")

class SectionDraft(BaseModel):
    section_title: str = Field(..., description="The title of the section being drafted.")
    content: str = Field(..., description="The drafted content for this section, formatted in Markdown. Include code blocks with explanations, tables, lists, etc. Aim for clarity and detail suitable for beginners.")

# --- Beginner Article Workflow ---
class BeginnerArticleWorkflow(Workflow):
    """
    A workflow that orchestrates agents to research, outline, write (section by section),
    and edit a technical article specifically tailored for beginners.
    Errors during execution will raise Exceptions. Yields final result on completion.
    """
    description: str = "Generates beginner-friendly technical articles."

    researcher: Agent
    outliner: Agent
    writer: Agent
    editor: Agent

    def __init__(
        self,
        api_key: str,
        model_id: str,
        max_tokens: int,
        session_id: str,
        storage: Optional[SqliteStorage] = None,
        debug_mode: bool = False,
        max_writer_retries: int = 2,
    ):
        super().__init__(session_id=session_id, storage=storage, debug_mode=debug_mode)
        self.max_writer_retries = max_writer_retries

        if not api_key:
            raise ValueError("OpenRouter API Key is required for BeginnerArticleWorkflow.")

        common_model_args = {"id": model_id, "api_key": api_key, "max_tokens": max_tokens}
        writer_tokens = max(max_tokens, 8192)
        editor_tokens = max(max_tokens, 8192)
        writer_model_args = {"id": model_id, "api_key": api_key, "max_tokens": writer_tokens}
        editor_model_args = {"id": model_id, "api_key": api_key, "max_tokens": editor_tokens}

        # Initialize Agents with full prompts
        self.researcher = Agent(
            name="TechResearcherBeginnerFocus",
            model=OpenRouter(**common_model_args),
            tools=[DuckDuckGoTools(search=True, news=True), Crawl4aiTools(max_length=10000)],
            description="Expert tech researcher finding and synthesizing information on dev/tech topics for a beginner audience.",
            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, and foundational documentation. Avoid overly academic papers or highly advanced discussions unless essential.
                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, common use cases, and potential beginner challenges.
                5. Provide a structured summary. Output MUST be `ResearchSummary` JSON.
            """),
            response_model=ResearchSummary, markdown=True, add_history_to_messages=False, exponential_backoff=True
        )

        self.outliner = Agent(
            name="BeginnerArticleOutliner",
            model=OpenRouter(**common_model_args),
            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 that clearly indicates the topic and suggests it's beginner-friendly (e.g., "Introduction to X", "Getting Started with Y").
                2. **Sections:** Structure the article logically for learning. Start with basics, then build up. Include sections like:
                    * Introduction (What is it? Why care?)
                    * Key Concepts/Terminology (Define important terms simply)
                    * Getting Started / Core How-To (Simple, practical examples)
                    * Code Examples Explained (If applicable, focus on clarity)
                    * Benefits / Use Cases (Why is this useful?)
                    * Potential Challenges for Beginners (Common pitfalls/tips)
                    * Conclusion / Next Steps
                3. **Keywords:** Include relevant SEO keywords, focusing on beginner terms (e.g., "tutorial", "basics", "introduction", "for beginners").
                4. Output MUST be `ArticleOutline` JSON.
            """),
            add_history_to_messages=False, response_model=ArticleOutline, markdown=False, exponential_backoff=True
        )

        self.writer = Agent(
            name="BeginnerTechWriter",
            model=OpenRouter(**writer_model_args),
            description="Writes a detailed, engaging 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 (for facts).
                b) The article outline (for structure context).
                c) The specific `section_title` you need to write content for.

                Your task is to write the content ONLY for the specified `section_title`, targeting **complete beginners** to this specific topic.
                1. **Accuracy:** Use the research summary for technical facts.
                2. **Clarity is Key:** Explain concepts as simply as possible. Define technical terms immediately. Use analogies or real-world examples if helpful. Avoid unnecessary jargon. Assume minimal prior knowledge.
                3. **Code/Command Explanations:** If including code snippets (```language) or commands (` `):
                    * Provide a **clear, step-by-step explanation** for each line or significant part.
                    * Explain the *purpose* of the code/command.
                    * Describe the expected input and output (if applicable).
                    * Keep initial examples simple.
                4. **Structure & Formatting:**
                    * Use Markdown extensively and correctly: `###` or `####` for sub-headings, **Bold**, *Italics*, ` ` for inline code, ```language ... ``` for blocks, bullet points (`*` or `-`), numbered lists (`1.`, `2.`), tables.
                5. **Engagement:** Start sections engagingly. Write in a slightly personal but professional and encouraging tone.
                6. **Detail:** Aim for sufficient detail to be genuinely helpful to a beginner. Prioritize clarity and thorough explanation over strict word count (~400 words is a rough guide, more is fine if needed for clarity).
                7. **Focus:** Do NOT write the main section title (like `## Section Title`) in your content. Do NOT write content for other sections. Focus *only* on the requested `section_title`.
                8. Output MUST be a `SectionDraft` JSON object containing the `section_title` you were given and the `content` you wrote.
            """),
            response_model=SectionDraft, add_history_to_messages=False, markdown=True, exponential_backoff=True
        )

        self.editor = Agent(
            name="BeginnerFocusedEditor",
            model=OpenRouter(**editor_model_args),
            description="Polishes a full article draft, ensuring clarity for beginners.",
            instructions=dedent("""\
                You are reviewing a complete article draft (in Markdown) assembled from sections written for beginners. You will receive the draft and the original outline.
                Your task is to perform final polishing:
                1. **Clarity for Beginners:** Read through from the perspective of someone new to the topic. Is it clear? Is jargon explained? Are explanations thorough enough? Add minor clarifications if needed.
                2. **Consistency:** Ensure consistent terminology, tone, and code/command formatting across sections.
                3. **Flow & Grammar:** Perform minor edits for smooth transitions, grammar, spelling, and punctuation.
                4. **Markdown:** Check for correct Markdown formatting (headings `## Section Title`, code blocks, lists, tables). Ensure headings match the provided outline sections.
                5. **Completeness:** Briefly check if sections seem reasonably detailed based on typical beginner needs (do not rewrite entire sections).
                6. **No Major Rewrites:** Do not add substantial new content or change the core meaning. Focus on polishing and beginner-friendliness.
                7. Return the final, polished Markdown article.
            """),
            add_history_to_messages=False, markdown=True, exponential_backoff=True
        )

    # --- Caching Methods ---
    def get_cached_data(self, key: str) -> Optional[Dict]:
        return self.session_state.get(key)

    def add_data_to_cache(self, key: str, data: BaseModel):
        logger.info(f"Caching data for key: {key}")
        self.session_state[key] = data.model_dump()

    def get_cached_final_article(self, topic_key: str) -> Optional[str]:
        key = f"final_article_{topic_key}"
        return self.session_state.get(key)

    def add_final_article_to_cache(self, topic_key: str, article: str):
        key = f"final_article_{topic_key}"
        logger.info(f"Caching final article for key: {key}")
        self.session_state[key] = article

    # --- Main Workflow Logic ---
    def run(self, topic: str, use_cache: bool = True) -> Iterator[RunResponse]:
        logger.info(f"Starting BeginnerArticleWorkflow for topic: '{topic}'")
        topic_key = re.sub(r'[^\w\-]+', '_', topic).strip('_').lower()

        # 1. Check cache
        if use_cache:
            cached_article = self.get_cached_final_article(topic_key)
            if cached_article:
                logger.info("Returning cached final article.")
                yield RunResponse(event=RunEvent.workflow_completed, content=cached_article)
                return

        # --- Stage 1: Research ---
        research_cache_key = f"research_{topic_key}"
        research_data: Optional[ResearchSummary] = None
        logger.info("--- Starting Research Stage ---")
        if use_cache:
            cached_research_dict = self.get_cached_data(research_cache_key)
            if cached_research_dict:
                try:
                    research_data = ResearchSummary.model_validate(cached_research_dict)
                    logger.info("Using cached research data.")
                except ValidationError as e:
                    logger.warning(f"Cached research data invalid: {e}. Re-running research.")
        if research_data is None:
            logger.info("Researching topic (Beginner Focus)...")
            try:
                research_response: RunResponse = self.researcher.run(topic)
                if not (research_response and isinstance(research_response.content, ResearchSummary)):
                    parsed_content = None
                    if isinstance(research_response.content, (str, dict)):
                        try:
                            data = json.loads(research_response.content) if isinstance(research_response.content, str) else research_response.content
                            parsed_content = ResearchSummary.model_validate(data)
                        except (ValidationError, TypeError, json.JSONDecodeError) as parse_error:
                            logger.warning(f"Research step returned parsable but invalid format: {parse_error}. Content: {research_response.content}")
                    if not parsed_content:
                        raise Exception(f"Research step failed or returned invalid format. Response: {research_response}")
                    research_data = parsed_content
                else:
                    research_data = research_response.content
                self.add_data_to_cache(research_cache_key, research_data)
                logger.info("Research complete.")
            except Exception as e:
                logger.error(f"Research failed: {e}")
                logger.error(traceback.format_exc())
                raise Exception(f"❌ Research step failed: {e}") from e

        # --- Stage 2: Outline ---
        outline_cache_key = f"outline_{topic_key}"
        outline_data: Optional[ArticleOutline] = None
        logger.info("--- Starting Outline Stage ---")
        if use_cache:
            cached_outline_dict = self.get_cached_data(outline_cache_key)
            if cached_outline_dict:
                try:
                    outline_data = ArticleOutline.model_validate(cached_outline_dict)
                    logger.info("Using cached outline data.")
                except ValidationError as e:
                    logger.warning(f"Cached outline data invalid: {e}. Re-running outline.")
        if outline_data is None:
            logger.info("Generating outline (Beginner Structure)...")
            try:
                outline_response: RunResponse = self.outliner.run(research_data.model_dump_json())
                if not (outline_response and isinstance(outline_response.content, ArticleOutline)):
                    parsed_content = None
                    if isinstance(outline_response.content, (str, dict)):
                        try:
                            data = json.loads(outline_response.content) if isinstance(outline_response.content, str) else outline_response.content
                            parsed_content = ArticleOutline.model_validate(data)
                        except (ValidationError, TypeError, json.JSONDecodeError) as parse_error:
                            logger.warning(f"Outline step returned parsable but invalid format: {parse_error}. Content: {outline_response.content}")
                    if not parsed_content:
                        raise Exception(f"Outline step failed or returned invalid format. Response: {outline_response}")
                    outline_data = parsed_content
                else:
                    outline_data = outline_response.content
                self.add_data_to_cache(outline_cache_key, outline_data)
                logger.info("Outline complete.")
            except Exception as e:
                logger.error(f"Outline failed: {e}")
                logger.error(traceback.format_exc())
                raise Exception(f"❌ Outline step failed: {e}") from e

        # --- Stage 3: Write Sections ---
        all_section_content: Dict[str, str] = {}
        total_sections = len(outline_data.sections)
        logger.info(f"--- Starting Section Writing Stage ({total_sections} sections) ---")
        writing_failed = False
        for i, section_title in enumerate(outline_data.sections):
            logger.info(f"Writing section {i+1}/{total_sections}: '{section_title}'...")
            writer_input_dict = {
                "research_data": research_data.model_dump(),
                "outline_data": outline_data.model_dump(),
                "section_title": section_title
            }
            try:
                writer_input_json = json.dumps(writer_input_dict, default=default_serializer)
            except TypeError as json_err:
                logger.error(f"Failed to serialize input for writer section '{section_title}': {json_err}")
                logger.error(f"Problematic Dict: {writer_input_dict}")
                raise Exception(f"❌ Failed to prepare input for writer: {json_err}") from json_err

            section_content_generated = False
            last_error = "Unknown error"
            for attempt in range(self.max_writer_retries + 1):
                logger.info(f"Writer attempt {attempt+1} for section: '{section_title}'")
                section_draft = None
                parse_error = None
                section_response = None
                try:
                    section_response = self.writer.run(writer_input_json)
                    if section_response and isinstance(section_response.content, SectionDraft):
                        section_draft = section_response.content
                    elif section_response and isinstance(section_response.content, (str, dict)):
                        data = json.loads(section_response.content) if isinstance(section_response.content, str) else section_response.content
                        section_draft = SectionDraft.model_validate(data)
                    else:
                        logger.warning(f"Writer attempt {attempt+1} returned unexpected type: {type(section_response.content if section_response else None)}")
                        last_error = f"Unexpected response type: {type(section_response.content if section_response else None)}"
                    if section_draft:
                        all_section_content[section_title] = section_draft.content
                        section_content_generated = True
                        logger.info(f"Writer attempt {attempt+1} successful for section: '{section_title}'")
                        break
                except (ValidationError, TypeError, json.JSONDecodeError) as e:
                    parse_error = e
                    last_error = f"Validation Error: {e}"
                    logger.warning(f"Writer attempt {attempt+1} failed validation... Response: {section_response.content if section_response else 'N/A'}")
                except Exception as e:
                    parse_error = e
                    last_error = f"Runtime Error: {e}"
                    logger.error(f"Writer attempt {attempt+1} encountered unexpected error...")
                    logger.error(traceback.format_exc())

                if attempt < self.max_writer_retries:
                    wait_time = 1 * (attempt + 1)
                    logger.info(f"Waiting {wait_time}s...")
                    time.sleep(wait_time)
                else:
                    logger.error(f"Max retries reached for section '{section_title}'. Skipping.")
                    all_section_content[section_title] = f"\n\n_[Content generation failed for '{section_title}'. Last Error: {last_error}]_\n\n"
                    writing_failed = True

            if not section_content_generated:
                logger.warning(f"Failed to write section '{section_title}' after retries.")
        # End of section writing loop

        if writing_failed:
            logger.warning("Some sections failed generation. Proceeding.")

        # --- Stage 4: Assemble ---
        logger.info("--- Starting Assembly Stage ---")
        assembled_draft_parts = [f"# {outline_data.title}\n"]
        for section_title in outline_data.sections:
            assembled_draft_parts.append(f"\n## {section_title}\n")
            section_content = all_section_content.get(section_title, f"\n_[Content for '{section_title}' missing or failed generation.]_\n")
            assembled_draft_parts.append(section_content.strip() + "\n")
        assembled_draft = "\n".join(assembled_draft_parts)
        logger.info("Assembly complete.")

        # --- Stage 5: Edit ---
        logger.info("--- Starting Editing Stage ---")
        final_article = assembled_draft
        try:
            editor_input_dict = {"draft_content": assembled_draft, "outline": outline_data.model_dump()}
            try:
                editor_input_json = json.dumps(editor_input_dict, default=default_serializer)
            except TypeError as json_err:
                logger.error(f"Failed to serialize input for editor: {json_err}")
                raise Exception(f"❌ Failed to prepare input for editor: {json_err}") from json_err

            editor_response: RunResponse = self.editor.run(editor_input_json)
            if editor_response and editor_response.content and isinstance(editor_response.content, str):
                final_article = editor_response.content
                logger.info("Editing complete.")
            else:
                logger.warning(f"Editor failed or returned empty/invalid content. Using assembled draft.")
        except Exception as e:
            logger.error(f"Editor failed: {e}")
            logger.error(traceback.format_exc())
            logger.warning(f"Editing step failed: {e}. Using assembled draft.")

        # --- Completion ---
        self.add_final_article_to_cache(topic_key, final_article)
        logger.info("Workflow completed successfully.")
        yield RunResponse(
            event=RunEvent.workflow_completed,
            content=final_article
        )

# --- Streamlit UI ---
st.set_page_config(page_title="Beginner Article Workflow", page_icon="✍️", layout="wide")
# Sidebar setup...
with st.sidebar:
    st.title("⚙️ Configuration")
    st.session_state.api_key = st.text_input(
        "OpenRouter API Key", type="password", key="api_key_input_wf",
        value=st.session_state.get("api_key", OPENROUTER_API_KEY or ""), help="Required."
    )
    available_models = [
        "openrouter/optimus-alpha", "openai/gpt-4o", "google/gemini-1.5-pro",
        "mistralai/mistral-large-latest", "meta-llama/llama-3.1-70b-instruct",
        "google/gemini-flash-1.5", "openrouter/auto",
    ]
    default_model = "openrouter/optimus-alpha"
    st.session_state.model_id = st.selectbox(
        "Select Model", options=available_models,
        index=available_models.index(st.session_state.get("model_id", default_model)) if st.session_state.get("model_id", default_model) in available_models else available_models.index(default_model),
        key="model_select_wf", help="Choose LLM."
    )
    st.session_state.max_tokens = st.slider(
        "Base Max Completion Tokens", min_value=2048, max_value=16384,
        value=st.session_state.get("max_tokens", 8192), step=1024,
        key="max_tokens_slider_wf", help="Base tokens."
    )
    st.session_state.use_cache = st.toggle("Use Cache", value=True, key="use_cache_wf", help="Reuse results.")
    db_file = "tmp/agno_beginner_workflows.db"
    os.makedirs("tmp", exist_ok=True)
    st.sidebar.caption(f"Cache DB: {db_file}")
    if st.button("Clear Chat History", key="clear_chat_wf"):
        st.session_state.messages_wf = []
        st.experimental_rerun()

# Main Chat Interface...
st.title("✍️ Agno Workflow: Beginner Article Writer")
st.markdown("Enter a topic...")
if "messages_wf" not in st.session_state:
    st.session_state.messages_wf = []
# History display loop...
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_article = 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_article:
            query_for_filename = message_info.get("topic", "article")
            safe_filename = re.sub(r'[^\w\-]+', '_', query_for_filename).strip('_').lower() or "article"
            st.download_button(
                label="Download Article (Markdown)",
                data=content,
                file_name=f"{safe_filename}.md",
                mime="text/markdown",
                key=f"download_hist_wf_{msg_index}"
            )

# User input and workflow execution...
if user_query := st.chat_input("Enter article topic..."):
    api_key = st.session_state.api_key
    model_id = st.session_state.model_id
    max_tokens = st.session_state.max_tokens
    use_cache = st.session_state.use_cache
    if not api_key:
        st.error("🚨 Please enter 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"):
            output_placeholder = st.empty()
            output_placeholder.markdown("⏳ Workflow running... Please check logs for detailed progress.")
            final_article_content = None
            error_message = None
            try:
                storage = SqliteStorage(table_name="beginner_article_workflows", db_file=db_file)
                topic_key = re.sub(r'[^\w\-]+', '_', user_query).strip('_').lower() or "article"
                session_id = f"beginner-article-{topic_key}"
                workflow = BeginnerArticleWorkflow(
                    api_key=api_key,
                    model_id=model_id,
                    max_tokens=max_tokens,
                    session_id=session_id,
                    storage=storage,
                    debug_mode=True
                )
                # Run Workflow & Handle Final Output
                for response in workflow.run(topic=user_query, use_cache=use_cache):
                    if response.event == RunEvent.workflow_completed:
                        final_article_content = response.content
                        break
                    else:
                        logger.warning(f"Received unexpected event type: {response.event}")
                # Display Final Result or Error
                output_placeholder.empty()
                if final_article_content:
                    output_placeholder.markdown(final_article_content)
                    st.session_state.messages_wf.append({
                        "role": "assistant",
                        "content": final_article_content,
                        "is_final": True,
                        "topic": user_query
                    })
                    safe_filename = topic_key or "article"
                    st.download_button(
                        label="Download Article (Markdown)",
                        data=final_article_content,
                        file_name=f"{safe_filename}.md",
                        mime="text/markdown",
                        key=f"download_now_wf_{topic_key}"
                    )
            # Catch exceptions raised from the workflow run
            except Exception as e:
                error_message_detail = f"{type(e).__name__}: {str(e)}"
                error_message_display = f"❌ **Workflow Failed:**\n```\n{error_message_detail}\n```\n(Check logs for full traceback)"
                logger.error(f"Workflow execution failed: {error_message_detail}")
                logger.error(traceback.format_exc())
                output_placeholder.empty()
                output_placeholder.markdown(error_message_display)
                st.session_state.messages_wf.append({
                    "role": "assistant",
                    "content": error_message_display,
                    "is_error": True
                })
            # Check for premature exit
            if final_article_content is None and error_message is None:
                warn_msg = "⚠️ Workflow finished, but no content was generated and no error was caught. Check workflow logic and logs."
                output_placeholder.empty()
                output_placeholder.markdown(warn_msg)
                st.session_state.messages_wf.append({"role": "assistant", "content": warn_msg, "is_error": True})

Crafting Beginner-Friendly Tech Articles with Agno Workflows and Streamlit

Agno Agents Tutorials

Table of Contents

What You’ll Build

Prerequisites

Step 1: Setting Up Your Environment with uv

Installing uv

Creating the Project

Setting Up Environment Variables

Step 2: Understanding the Workflow Structure

Workflow Stages

Agents and Their Roles

Pydantic Models

Step 3: Diving into the Code

Imports and Setup

Custom JSON Serializer

Pydantic Models

Workflow Class

Initialization

Agent Definitions

Caching Methods

Run Method

Streamlit Interface

Step 4: Running the Workflow

How to Use It

Step 5: How It Works in Action

Step 6: Example Output

Troubleshooting

Why Use Agno Workflows?

Next Steps

Conclusion

Complete Code

Related Posts

Deploy Streamlit on a VPS and Proxy to Cloudflare Tunnels

Streamlit vs Taipy: How to Choose the Best Tool for Python Projects

How to Integrate FREE Groq API and Mistral LLM into Your Streamlit App

Crafting Beginner-Friendly Tech Articles with Agno Workflows and Streamlit

Building a Multi-Agent Research Team with Google ADK, Tavily Search, and Crawl4AI

How to Use Any OpenRouter Model with Google Agent Development Kit (ADK)

Agno Agents Tutorials

Table of Contents

What You’ll Build

Prerequisites

Step 1: Setting Up Your Environment with uv

Installing uv

Creating the Project

Setting Up Environment Variables

Step 2: Understanding the Workflow Structure

Workflow Stages

Agents and Their Roles

Pydantic Models

Step 3: Diving into the Code

Imports and Setup

Custom JSON Serializer

Pydantic Models

Workflow Class

Initialization

Agent Definitions

Caching Methods

Run Method

Streamlit Interface

Step 4: Running the Workflow

How to Use It

Step 5: How It Works in Action

Step 6: Example Output

Troubleshooting

Why Use Agno Workflows?

Next Steps

Conclusion

Complete Code

Share this article

Related Posts

Deploy Streamlit on a VPS and Proxy to Cloudflare Tunnels

Streamlit vs Taipy: How to Choose the Best Tool for Python Projects

How to Integrate FREE Groq API and Mistral LLM into Your Streamlit App