app.py

# -*- coding: utf-8 -*-
"""AI Task Master.ipynb

Automatically generated by Colab.

Original file is located at
    https://colab.research.google.com/drive/1k8oLpD1BlJjWEOTZ0UoJQCVyfl9883bE

# "AI Task Master" — Multi-Agent Productivity & Execution System

# Upload a Goal / To-Do List / Project Idea → Agents break it down → Prioritize → Create action steps → Assign estimated time → Generate a Daily Execution Plan with smart suggestions & progress tracking logic.

# 🧱 Step 0: Colab Setup (Install & Imports)
"""

#!pip install --quiet crewai langchain-openai gradio python-dotenv pydantic -q

import os, json, textwrap, datetime as dt
from typing import Optional, List, Dict, Any

from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

"""# 🔐 Step 1: Configure API Key (OpenAI)

"""

import os

# Hugging Face Spaces will allow access to secrets like this:
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

# Model defaults (you can change)
MODEL_NAME = "gpt-4o-mini"   # fast & smart
TEMPERATURE = 0.2
llm = ChatOpenAI(model=MODEL_NAME, temperature=TEMPERATURE)

"""# 🧩 Step 2: Project Inputs Helper"""

# Light helper to bundle inputs
def default_inputs(
    goal: str,
    deadline_days: int = 14,
    hours_per_day: int = 2,
    start_date: Optional[str] = None,
) -> Dict[str, Any]:
    if not start_date:
        start_date = dt.date.today().isoformat()
    return {
        "goal": goal.strip(),
        "deadline_days": int(deadline_days),
        "hours_per_day": int(hours_per_day),
        "start_date": start_date,
    }

"""# 🎭 Step 3: Define Agents (Roles + Personalities)

"""

# Agent: Understand the user's goal & constraints
goal_understanding = Agent(
    role="Goal Understanding Agent",
    backstory=(
        "You are a thoughtful strategist. You clarify objectives, constraints, "
        "and success criteria. You avoid fluff and get to the essence."
    ),
    goal=(
        "Rewrite the user's goal clearly and identify assumptions, scope, "
        "success criteria, hard constraints, and risks."
    ),
    allow_delegation=False,
    llm=llm,
)

# Agent: Break the goal into milestones & tasks
task_breakdown = Agent(
    role="Task Breakdown Agent",
    backstory=(
        "You are a senior project planner. You decompose work into milestones "
        "and concrete tasks that are testable and deliver value."
    ),
    goal=(
        "Produce milestones and detailed subtasks with crisp deliverables."
    ),
    allow_delegation=False,
    llm=llm,
)

# Agent: Estimate effort & prioritize
effort_priority = Agent(
    role="Effort & Priority Agent",
    backstory=(
        "You are a pragmatic PM. You estimate realistic time and difficulty, "
        "sequence tasks, and surface dependencies."
    ),
    goal=(
        "Estimate time per task, difficulty (1-5), dependencies, and propose a priority order."
    ),
    allow_delegation=False,
    llm=llm,
)

# Agent: Build a calendar-style plan
scheduler = Agent(
    role="Schedule Maker Agent",
    backstory=(
        "You create practical schedules given time budgets per day and deadlines."
    ),
    goal=(
        "Map tasks to days from start_date within deadline_days, respecting hours_per_day."
    ),
    allow_delegation=False,
    llm=llm,
)

# Agent: Motivation & execution tips (optional spice)
motivator = Agent(
    role="Motivator Agent",
    backstory=(
        "You coach users with specific, actionable tips, accountability tactics, and small rewards."
    ),
    goal="Provide short, concrete advice to keep momentum high.",
    allow_delegation=False,
    llm=llm,
)

# Agent: Final synthesis for export
synthesizer = Agent(
    role="Synthesis Agent",
    backstory=(
        "You are a concise editor. You assemble a clean, professional plan for download and printing."
    ),
    goal="Create final Markdown + JSON outputs for the user.",
    allow_delegation=False,
    llm=llm,
)

"""# 🧠 Step 4: Define Tasks (with JSON-first outputs

"""

def make_tasks(inputs: Dict[str, Any]) -> List[Task]:
    common_vars = (
        f"GOAL: {inputs['goal']}\n"
        f"START_DATE: {inputs['start_date']}\n"
        f"DEADLINE_DAYS: {inputs['deadline_days']}\n"
        f"HOURS_PER_DAY: {inputs['hours_per_day']}\n"
    )

    t1 = Task(
        description=textwrap.dedent(f"""
        Clarify the user's goal.

        Context:
        {common_vars}

        Return STRICT JSON with keys:
        {{
          "clarified_goal": "...",
          "assumptions": ["..."],
          "scope": ["in-scope", "..."],
          "success_criteria": ["..."],
          "constraints": ["..."],
          "risks": ["..."]
        }}
        """).strip(),
        agent=goal_understanding,
        expected_output="JSON only.",
    )

    t2 = Task(
        description=textwrap.dedent(f"""
        Break the clarified goal into milestones and granular tasks.

        Input is the JSON from the previous step.
        Return STRICT JSON:
        {{
          "milestones": [
            {{
              "name": "...",
              "description": "...",
              "tasks": [
                {{
                  "id": "T1",
                  "title": "...",
                  "definition_of_done": "...",
                  "dependencies": [],
                  "tags": ["..."]
                }}
              ]
            }}
          ]
        }}
        """).strip(),
        agent=task_breakdown,
        expected_output="JSON only.",
    )

    t3 = Task(
        description=textwrap.dedent(f"""
        Estimate effort and priority for EACH task id from the previous output.

        Respect HOURS_PER_DAY={inputs['hours_per_day']} and DEADLINE_DAYS={inputs['deadline_days']}.
        Return STRICT JSON:
        {{
          "estimates": [
            {{
              "id": "T1",
              "hours": 2.5,
              "difficulty_1to5": 3,
              "priority_1to5": 1
            }}
          ]
        }}
        """).strip(),
        agent=effort_priority,
        expected_output="JSON only.",
    )

    t4 = Task(
        description=textwrap.dedent(f"""
        Create a day-by-day schedule mapping tasks to calendar dates.

        Use START_DATE and DEADLINE_DAYS. Do not exceed HOURS_PER_DAY per day.
        Return STRICT JSON:
        {{
          "schedule": [
            {{
              "date": "YYYY-MM-DD",
              "allocated_hours": 2,
              "tasks": [{{"id":"T1","title":"...","hours":1.5}}]
            }}
          ],
          "summary": {{
            "total_hours": 0,
            "days_planned": 0,
            "buffer_hours": 0
          }}
        }}
        """).strip(),
        agent=scheduler,
        expected_output="JSON only.",
    )

    t5 = Task(
        description=textwrap.dedent(f"""
        Provide 5 short motivation or execution tactics tailored to the plan.

        Return STRICT JSON:
        {{
          "tips": [
            "Keep daily sessions short & focused with a single visible deliverable.",
            "..."
          ]
        }}
        """).strip(),
        agent=motivator,
        expected_output="JSON only.",
    )

    t6 = Task(
        description=textwrap.dedent(f"""
        Synthesize a final MARKDOWN report AND a JSON bundle.

        Inputs are the JSONs from all prior tasks. Build:
        1) Markdown (for printing) with sections:
           - Goal Summary
           - Milestones & Tasks
           - Estimates & Priorities
           - Day-by-Day Plan (table)
           - Tips (bulleted)
        2) JSON bundle containing all prior JSON merged into one: keys:
           {{
             "goal_context": ...,
             "work_breakdown": ...,
             "estimates": ...,
             "schedule": ...,
             "tips": ...
           }}

        Return STRICT JSON:
        {{
          "markdown": "....",
          "bundle": {{ /* merged JSON object */ }}
        }}
        """).strip(),
        agent=synthesizer,
        expected_output="JSON only.",
    )

    return [t1, t2, t3, t4, t5, t6]

"""# 🏃‍♂️ Step 5: Run the Crew"""

def run_planner(goal: str, deadline_days: int = 14, hours_per_day: int = 2, start_date: Optional[str] = None):
    inputs = default_inputs(goal, deadline_days, hours_per_day, start_date)

    tasks = make_tasks(inputs)
    crew = Crew(
        agents=[goal_understanding, task_breakdown, effort_priority, scheduler, motivator, synthesizer],
        tasks=tasks,
        process=Process.sequential,  # run in order
        verbose=True,
        memory=True,
        cache=True,
    )

    result = crew.kickoff(inputs=inputs)

    # The final task returns JSON with 'markdown' and 'bundle'
    # crewai often returns text; ensure it's valid JSON:
    def coerce_json(text: str) -> Dict[str, Any]:
        try:
            return json.loads(text)
        except Exception:
            # Try to extract JSON block heuristically
            start = text.find("{")
            end = text.rfind("}")
            if start != -1 and end != -1:
                return json.loads(text[start:end+1])
            raise

    final = coerce_json(str(result))
    md = final.get("markdown", "# Plan\n\n(No markdown received)")
    bundle = final.get("bundle", {})

    # Save artifacts
    out_dir = "task_master_outputs"
    os.makedirs(out_dir, exist_ok=True)
    stamp = dt.datetime.now().strftime("%Y%m%d_%H%M%S")

    md_path = os.path.join(out_dir, f"plan_{stamp}.md")
    json_path = os.path.join(out_dir, f"plan_{stamp}.json")

    with open(md_path, "w", encoding="utf-8") as f:
        f.write(md)
    with open(json_path, "w", encoding="utf-8") as f:
        json.dump(bundle, f, indent=2, ensure_ascii=False)

    return {
        "markdown_path": md_path,
        "json_path": json_path,
        "markdown_preview": md[:1500] + ("\n\n...[truncated]..." if len(md) > 1500 else ""),
    }

"""# Step 6: Minimal Gradio UI"""

import gradio as gr

def gradio_run(goal, deadline_days, hours_per_day, start_date):
    out = run_planner(goal, int(deadline_days), int(hours_per_day), start_date or None)
    md = ""
    with open(out["markdown_path"], "r", encoding="utf-8") as f:
        md = f.read()
    return md, out["markdown_path"], out["json_path"]

with gr.Blocks(title="CrewAI Task Master") as demo_ui:
    gr.Markdown("# 🧠 CrewAI Task Master\nPlan any goal into a day-by-day schedule.")
    with gr.Row():
        goal = gr.Textbox(label="Your Goal", placeholder="e.g., Build an AI portfolio website in 10 days", lines=3)
    with gr.Row():
        deadline = gr.Number(label="Deadline (days)", value=14, precision=0)
        hours = gr.Number(label="Hours per day", value=2, precision=0)
        start = gr.Textbox(label="Start Date (YYYY-MM-DD, optional)", placeholder="Leave blank for today")
    run_btn = gr.Button("Run Agents 🚀")

    md_out = gr.Markdown(label="Plan (Markdown)")
    md_file = gr.File(label="Download Markdown")
    json_file = gr.File(label="Download JSON")

    run_btn.click(
        gradio_run,
        inputs=[goal, deadline, hours, start],
        outputs=[md_out, md_file, json_file]
    )

demo_ui.launch(share=False)