Skip to content

Instantly share code, notes, and snippets.

@kausmeows

kausmeows/workflow_input_schema.py

Created January 2, 2026 08:59
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save kausmeows/d8d6a0c42166f2e541b96a66305bef8d to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save kausmeows/d8d6a0c42166f2e541b96a66305bef8d to your computer and use it in GitHub Desktop.
Download ZIP
Raw
workflow_input_schema.py
# paper2saas_agent.py
import argparse
import os
import json
from typing import Dict, Any, List
from pydantic import BaseModel, Field
# Agno SDK Imports
from agno.workflow import Workflow, Step, Parallel
from agno.workflow.types import StepInput, StepOutput
from agno.run.workflow import WorkflowRunOutput
from agno.agent import Agent
# --- INPUT SCHEMA ---
class Paper2SaaSInput(BaseModel):
"""Input schema for the Paper2SaaS workflow."""
arxiv_id: str = Field(..., description="The arXiv paper ID to analyze (e.g., '2401.00001')")
# Tool Imports
from agno.tools.arxiv import ArxivTools
from agno.tools.hackernews import HackerNewsTools
from agno.tools.website import WebsiteTools
from agno.tools.brightdata import BrightDataTools
# --- AGENT DEFINITIONS ---
# 0. Paper Whisperer
paper_whisperer = Agent(
name="PaperWhisperer",
tools=[ArxivTools()],
instructions="""
You are the Paper Whisperer. Given an arXiv paper ID, use ArxivTools to fetch and parse it.
Output in exact format:
PAPER ESSENCE: [Summary]
TECHNICAL REALITY: [Constraints & Architecture]
CAPABILITY MAP: [What is actually new?]
""",
markdown=True,
)
# 1. Trend Archeologist
trend_archeologist = Agent(
name="TrendArcheologist",
tools=[HackerNewsTools(), BrightDataTools(), WebsiteTools()],
instructions="""
Scan real-time signals.
Search for recent pains, dying solutions, and wished capabilities related to AI and SaaS.
Output: A summarized list of 'Market Signals' and 'Pain Points'.
""",
markdown=True,
)
# 2. Synthesizer
synthesizer = Agent(
name="Synthesizer",
instructions="""
You are the Synthesizer.
Input: {Paper Analysis} AND {Market Trends}.
Task: Generate 10 high-potential SaaS ideas that combine the paper's tech with the market pains.
Output: A numbered list of ideas. Each idea must have a 'Concept' and 'Why Now'.
""",
markdown=True,
)
# 3. Research Agent
# IMPORTANT CHANGE: Instructions updated to restate the idea, fixing the memory gap.
research_agent = Agent(
name="ResearchAgent",
tools=[BrightDataTools(), HackerNewsTools(), WebsiteTools()],
instructions="""
You will receive a list of ideas. Select the top 3 most viable ones.
For each of the 3 ideas, perform a quick validation search.
CRITICAL OUTPUT FORMAT:
You must output the data in this structure for the next agent:
--- IDEA 1 ---
CONCEPT: [Restate the original concept fully]
EVIDENCE: [Search results found]
COMPETITION: [Existing tools found]
--- IDEA 2 ---
...
""",
markdown=True,
)
# 4. Devil's Advocate
devil_advocate = Agent(
name="DevilAdvocate",
instructions="""
Roast the ideas provided by the Research Agent.
Input will be: Concept + Evidence.
Task: Reject any idea that lacks evidence or has too much competition.
Output: Only the survivors (if any). If all die, say "ALL REJECTED".
""",
markdown=True,
)
# 5. Validator
validator = Agent(
name="Validator",
tools=[WebsiteTools()],
instructions="""
Take the surviving ideas and format them into a final pitch.
Create a 'Implementation Sketch' for the best one.
""",
markdown=True,
)
# --- WORKFLOW FUNCTIONS (The Fixes) ---
def fuse_whisperer_trend(step_input: StepInput) -> StepOutput:
"""
Fuses the output of the Parallel step (which is a dictionary)
into a single string for the Synthesizer.
"""
# Access nested steps directly by name using get_step_content
whisper_content = step_input.get_step_content("whisper") or ""
trend_content = step_input.get_step_content("trend") or ""
combined_context = (
f"=== INPUT 1: PAPER ANALYSIS ===\n{whisper_content}\n\n"
f"=== INPUT 2: MARKET TRENDS ===\n{trend_content}"
)
return StepOutput(
step_name="fuse_data",
content=combined_context,
success=True
)
def check_rejection(step_input: StepInput) -> StepOutput:
"""
Simple pass-through. In a more complex app, you could raise an error
or stop the workflow here if step_input.input contains "ALL REJECTED".
"""
roast_result = step_input.previous_step_content or ""
if "ALL REJECTED" in str(roast_result):
# We allow it to proceed to Validator, but Validator will likely see empty input.
# Alternatively, modify the content to warn the Validator.
return StepOutput(
step_name="critique_check",
content="WARNING: All ideas were roasted. See if you can salvage one.",
success=True
)
return StepOutput(
step_name="critique_check",
content=str(roast_result),
success=True
)
# --- WORKFLOW DEFINITION ---
# Steps
whisper_step = Step(name="whisper", agent=paper_whisperer)
trend_step = Step(name="trend", agent=trend_archeologist)
# Parallel Step: Runs both agents at the same time
# Note: Steps are passed as positional arguments, not in a list
parallel_research = Parallel(
whisper_step,
trend_step,
name="Research Phase"
)
# Sequential Steps
fuse_step = Step(name="fuse_data", executor=fuse_whisperer_trend)
synth_step = Step(name="synth", agent=synthesizer)
research_step = Step(name="research", agent=research_agent)
# Note: We removed 'fuse2' because ResearchAgent now includes the context in its output.
roast_step = Step(name="roast", agent=devil_advocate)
critique_step = Step(name="critique_check", executor=check_rejection)
validate_step = Step(name="validate", agent=validator)
paper2saas_workflow = Workflow(
name="Paper2SaaS Discovery",
input_schema=Paper2SaaSInput,
steps=[
parallel_research, # 1. Get Data
fuse_step, # 2. Format Data
synth_step, # 3. Create Ideas
research_step, # 4. Check Ideas (includes original idea in output)
roast_step, # 5. Critique
critique_step, # 6. Check Logic
validate_step, # 7. Final Polish
],
)
def run_paper2saas(arxiv_id: str):
print(f"--- Starting Paper2SaaS for {arxiv_id} ---")
# Using structured input schema
workflow_input = Paper2SaaSInput(arxiv_id=arxiv_id)
response: WorkflowRunOutput = paper2saas_workflow.run(
input=workflow_input,
markdown=True
)
print("\n\n=== FINAL OUTPUT ===")
print(response.content)
return response
if __name__ == "__main__":
run_paper2saas("2401.00001")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment