Agno Workflows Tutorial: AI Article Writer with Streamlit

An Agno workflow can automate the entire article-writing pipeline: research, outline, write, and edit, all behind a Streamlit UI. If you’ve been looking for a practical multi-agent workflow example that runs on current Agno, this is it.

Writing beginner-friendly tech articles takes time. You need credible sources, a logical structure, clear explanations, and a polished final draft. A multi-agent AI pipeline handles each stage as a specialized task, which produces better output than asking a single model to “write an article about X.”

This tutorial walks you through building a 4-agent Agno workflow using Workflows 2.0 (the current API), Streamlit for the UI, and uv for project setup. The pipeline researches a topic, outlines the article, writes each section, and edits the final draft. Everything is displayed in a Streamlit app where you can download the result as Markdown.

If you’re new to AI-assisted coding, check getting started programming with AI first.

What you’ll build: a multi-agent Agno workflow

The BeginnerArticleWorkflow coordinates four specialized agents to produce a polished article:

The pipeline looks like this:

[Researcher] → [Outliner] → [Writer] → [Editor] → [Streamlit UI]

Each agent produces structured output (Pydantic models) that feeds the next stage. The Researcher finds and summarizes sources. The Outliner creates the article structure. The Writer crafts each section. The Editor polishes the final draft.

Prerequisites

No sqlalchemy manual install needed. Agno v2’s SqliteDb pulls its own dependencies. The duckduckgo-search package has been renamed to ddgs.

If you haven’t used Agno before, start with getting started with Agno Agents.

Step 1: Setting up your environment with uv

We’ll use uv for project setup. If you’re new to uv, see our getting started with uv guide.

Installing uv

• macOS/Linux:

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

• Windows (PowerShell):

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

• Verify:

  uv --version

Creating the project

uv init agno-article-writer
cd agno-article-writer
uv python pin 3.12
uv venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

Installing dependencies

uv add agno streamlit openai ddgs crawl4ai

Verify the install:

uv run python -c "import agno; print(agno.__version__)"

Should print 2.9.x or later.

Setting up environment variables

Create a .env file:

echo "OPENROUTER_API_KEY=your_key_here" > .env

Replace your_key_here with your OpenRouter API key. This keeps your key secure and loads it automatically with python-dotenv (which Agno includes as a dependency).

Step 2: Understanding the Agno workflow structure

Before diving into code, let’s understand how Workflows 2.0 differ from the old API.

Workflows 2.0: what changed

Agno v2 completely removed Workflows v1. The old pattern of subclassing Workflow, overriding run(), and using self.session_state is gone. Here’s the mapping:

Workflow steps (research → outline → write → edit)

In Workflows 2.0, a workflow is a list of steps. Each step is a Python function that receives a session_state dict and returns updated state. The four steps map to four specialized agents:

• Step 1 (research_step): Calls the Researcher agent, stores ResearchSummary in state.
• Step 2 (outline_step): Calls the Outliner agent, stores ArticleOutline in state.
• Step 3 (write_step): Loops over outline sections, calls the Writer agent per section, assembles the draft.
• Step 4 (edit_step): Calls the Editor agent, stores the final polished article.

[Researcher] → ResearchSummary (Pydantic)
     ↓
[Outliner] → ArticleOutline (Pydantic)
     ↓
[Writer] → SectionDraft per section (Pydantic)
     ↓
[Editor] → Final Markdown string

Agents and their roles

• Researcher: Uses DuckDuckGo search + Crawl4ai extraction → structured ResearchSummary.
• Outliner: Takes research → produces ArticleOutline with title, sections, keywords.
• Writer: Takes research + outline + section title → produces SectionDraft per section.
• Editor: Takes assembled draft → polished Markdown.

Pydantic models (output schema)

The same four models as before (ResearchFinding, ResearchSummary, ArticleOutline, SectionDraft), but now used as output_schema on agents instead of response_model. Pydantic enforces types and validates output at each stage, so downstream code doesn’t break on malformed data.

Step 3: Building the Agno workflow code

Here’s the complete rewritten code for Agno v2.9+ / Workflows 2.0.

Imports and setup (Workflows 2.0)

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

Key changes from v1:

• RunResponse → RunOutput (from agno.run.agent)
• SqliteStorage → SqliteDb (from agno.db.sqlite)
• Workflow is no longer subclassed. It’s constructed with steps=

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 with code, tables, lists.")

These are the same data shapes as the v1 article. The difference is how they’re used: as output_schema on agents instead of response_model.

Defining the agents

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

Note the v2 changes:

• DuckDuckGoTools(enable_search=True, enable_news=True) (was search=True, news=True)
• output_schema=ResearchSummary (was response_model=)
• add_history_to_context=False (was add_history_to_messages=False)
• Model IDs use the OpenRouter slug directly (e.g., openai/gpt-5-mini), not prefixed with openrouter/

Defining the workflow steps

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

Key differences from v1:

• Each step is a plain function taking session_state dict, not a method on a subclassed Workflow.
• session_state is persisted automatically when db=SqliteDb(...) is set.
• No more manual self.session_state.get() / add_data_to_cache() helpers.
• The workflow is constructed with Workflow(name=..., steps=[...], db=...).

Streamlit interface

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})

Note: st.experimental_rerun() was removed in Streamlit 1.61. Use st.rerun() instead.

Step 4: Running the Agno workflow with Streamlit

Save the code as beginner_article_workflow_streamlit.py and run:

uv run streamlit run beginner_article_workflow_streamlit.py

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

1. Enter your OpenRouter API key in the sidebar (or rely on .env).
2. Select a model. openai/gpt-5-mini is a good default (cheap, capable).
3. Set max tokens (8192 works well for most articles).
4. Keep caching enabled to save API costs on repeat runs.
5. Enter a topic like “Introduction to Python for Beginners.”
6. Watch the progress indicator, then download the article as .md.

Verify it works: After the first run, check the terminal for Workflow completed successfully and confirm the Streamlit UI shows the article with a download button. The SQLite cache file should appear at tmp/agno_beginner_workflows.db.

Step 5: How it works in action

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

1. Cache check → miss → runs the full pipeline.
2. Research: DuckDuckGo finds 5-7 sources, Crawl4ai extracts content → ResearchSummary JSON with findings and synthesis.
3. Outline: Produces a structured plan:

   {
       "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: Writes each section with code examples and beginner explanations.
5. Editor: Polishes for consistency, grammar, and formatting.
6. Output: Final article displayed in Streamlit + cached in SQLite.

Check token spend after a run:

# 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")

On the second run with the same topic, the cache kicks in and skips all API calls.

Step 6: Example output

Here’s a taste of what the pipeline produces for “Introduction to Python”:

# 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.

The output features clear explanations, step-by-step code breakdowns, and a beginner-friendly tone throughout.

Troubleshooting

sudo apt install -y libnss3 libatk-bridge2.0-0 libdrm2 libxcomposite1 libxdamage1 libxrandr2 libgbm1 libpango-1.0-0 libcairo2 libasound2

Why use Agno workflows?

• Lightweight: Agno is a focused Python library, not a heavy framework. Agent creation is fast.
• Structured output: Pydantic output_schema ensures type-safe data flow between agents. No parsing raw text.
• Built-in caching: SqliteDb + session_state persistence saves API costs on repeat runs.
• Workflows 2.0 primitives: Step, Condition, Router, Loop, Parallel for complex pipelines beyond this linear example.
• Streamlit integration: Fast UI prototyping without touching frontend frameworks.

If you want a different multi-agent pattern, see building an AI research squad with Agno and Streamlit.

Cost and operations notes

A typical article (5 sections) uses roughly 15-25k input tokens and 10-15k output tokens across all agents. With openai/gpt-5-mini, that’s about $0.02-0.05 per article.

Cost-saving tips:

• Use a cheap model for research and outline; a stronger model only for writer/editor if quality matters.
• Keep caching enabled. Re-running the same topic skips all API calls entirely.
• Set max_tokens appropriately. Research doesn’t need 8k tokens; 4k is usually enough.

Running on a VPS: The Streamlit app runs fine on a 1 GB VPS. For always-on access, deploy behind a reverse proxy. Hetzner Cloud VPS is a solid affordable option, and Hostinger VPS works as a budget alternative.

For deployment instructions, see deploy your Streamlit app on a VPS.

Next steps

• Add more tools: YouTube search, GitHub search, or MCP tools for your Agno agent for richer research.
• Customize for different audiences: Adjust agent instructions for intermediate or advanced readers.
• Use different models: Try Groq’s free API in Streamlit for zero-cost research steps.
• Deploy to production: Host on a VPS with Dokploy. See deploy a Python uv project with Dokploy.
• Build other agents: Build a Discord AI bot with Agno.
• Explore frameworks: Compare Python web frameworks if you want to move beyond Streamlit.

Conclusion

You’ve built a 4-agent Agno workflow that researches, outlines, writes, and edits beginner-friendly tech articles — all behind a Streamlit UI. The pipeline uses Workflows 2.0 (the current API), caches intermediate results in SQLite to save costs, and produces downloadable Markdown.

Try different topics — “Learn JavaScript,” “Docker for Beginners,” “What is an API?” — and see how the agents handle them. Adjust the agent instructions to target different skill levels. The framework is flexible enough to extend with more agents, different tools, or alternative models.

For more on building with Agno, see getting started programming with AI.

Complete code

# 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})