{ "nbformat": 4, "nbformat_minor": 0, "metadata": { "colab": { "provenance": [] }, "kernelspec": { "name": "python3", "display_name": "Python 3" }, "language_info": { "name": "python" } }, "cells": [ { "cell_type": "markdown", "source": [ "# \"AI Task Master\" — Multi-Agent Productivity & Execution System\n", "\n", "# 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." ], "metadata": { "id": "hMAysKRQphEj" } }, { "cell_type": "markdown", "source": [ "# 🧱 Step 0: Colab Setup (Install & Imports)" ], "metadata": { "id": "IRKP3gswqJSu" } }, { "cell_type": "code", "source": [ "#!pip install --quiet crewai langchain-openai gradio python-dotenv pydantic -q" ], "metadata": { "id": "58wVUNhZnxQh" }, "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "source": [], "metadata": { "id": "sH9afYAjqdCa" } }, { "cell_type": "code", "source": [ "import os, json, textwrap, datetime as dt\n", "from typing import Optional, List, Dict, Any\n", "\n", "from crewai import Agent, Task, Crew, Process\n", "from langchain_openai import ChatOpenAI\n" ], "metadata": { "id": "EwHFD9RSnsRg" }, "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "source": [ "# 🔐 Step 1: Configure API Key (OpenAI)\n" ], "metadata": { "id": "qiL6nGlGqe_w" } }, { "cell_type": "code", "source": [ "from google.colab import userdata\n", "\n", "# Retrieve the API key from Colab secrets\n", "os.environ[\"OPENAI_API_KEY\"] = userdata.get('OPENAI_API_KEY')\n", "\n", "# Model defaults (you can change)\n", "MODEL_NAME = \"gpt-4o-mini\" # fast & smart\n", "TEMPERATURE = 0.2\n", "llm = ChatOpenAI(model=MODEL_NAME, temperature=TEMPERATURE)" ], "metadata": { "id": "3A7Xcn2Qn81O" }, "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "source": [ "# 🧩 Step 2: Project Inputs Helper" ], "metadata": { "id": "aFEecuUqqiiQ" } }, { "cell_type": "code", "source": [ "# Light helper to bundle inputs\n", "def default_inputs(\n", " goal: str,\n", " deadline_days: int = 14,\n", " hours_per_day: int = 2,\n", " start_date: Optional[str] = None,\n", ") -> Dict[str, Any]:\n", " if not start_date:\n", " start_date = dt.date.today().isoformat()\n", " return {\n", " \"goal\": goal.strip(),\n", " \"deadline_days\": int(deadline_days),\n", " \"hours_per_day\": int(hours_per_day),\n", " \"start_date\": start_date,\n", " }\n" ], "metadata": { "id": "RUHRmzmwoM6E" }, "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "source": [ "# 🎭 Step 3: Define Agents (Roles + Personalities)\n" ], "metadata": { "id": "mt9NZfhlqmEd" } }, { "cell_type": "code", "source": [ "# Agent: Understand the user's goal & constraints\n", "goal_understanding = Agent(\n", " role=\"Goal Understanding Agent\",\n", " backstory=(\n", " \"You are a thoughtful strategist. You clarify objectives, constraints, \"\n", " \"and success criteria. You avoid fluff and get to the essence.\"\n", " ),\n", " goal=(\n", " \"Rewrite the user's goal clearly and identify assumptions, scope, \"\n", " \"success criteria, hard constraints, and risks.\"\n", " ),\n", " allow_delegation=False,\n", " llm=llm,\n", ")\n", "\n", "# Agent: Break the goal into milestones & tasks\n", "task_breakdown = Agent(\n", " role=\"Task Breakdown Agent\",\n", " backstory=(\n", " \"You are a senior project planner. You decompose work into milestones \"\n", " \"and concrete tasks that are testable and deliver value.\"\n", " ),\n", " goal=(\n", " \"Produce milestones and detailed subtasks with crisp deliverables.\"\n", " ),\n", " allow_delegation=False,\n", " llm=llm,\n", ")\n", "\n", "# Agent: Estimate effort & prioritize\n", "effort_priority = Agent(\n", " role=\"Effort & Priority Agent\",\n", " backstory=(\n", " \"You are a pragmatic PM. You estimate realistic time and difficulty, \"\n", " \"sequence tasks, and surface dependencies.\"\n", " ),\n", " goal=(\n", " \"Estimate time per task, difficulty (1-5), dependencies, and propose a priority order.\"\n", " ),\n", " allow_delegation=False,\n", " llm=llm,\n", ")\n", "\n", "# Agent: Build a calendar-style plan\n", "scheduler = Agent(\n", " role=\"Schedule Maker Agent\",\n", " backstory=(\n", " \"You create practical schedules given time budgets per day and deadlines.\"\n", " ),\n", " goal=(\n", " \"Map tasks to days from start_date within deadline_days, respecting hours_per_day.\"\n", " ),\n", " allow_delegation=False,\n", " llm=llm,\n", ")\n", "\n", "# Agent: Motivation & execution tips (optional spice)\n", "motivator = Agent(\n", " role=\"Motivator Agent\",\n", " backstory=(\n", " \"You coach users with specific, actionable tips, accountability tactics, and small rewards.\"\n", " ),\n", " goal=\"Provide short, concrete advice to keep momentum high.\",\n", " allow_delegation=False,\n", " llm=llm,\n", ")\n", "\n", "# Agent: Final synthesis for export\n", "synthesizer = Agent(\n", " role=\"Synthesis Agent\",\n", " backstory=(\n", " \"You are a concise editor. You assemble a clean, professional plan for download and printing.\"\n", " ),\n", " goal=\"Create final Markdown + JSON outputs for the user.\",\n", " allow_delegation=False,\n", " llm=llm,\n", ")\n" ], "metadata": { "id": "U9Wvk1DqoYwx" }, "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "source": [ "# 🧠 Step 4: Define Tasks (with JSON-first outputs\n" ], "metadata": { "id": "_jgH0_8TqprY" } }, { "cell_type": "code", "source": [ "def make_tasks(inputs: Dict[str, Any]) -> List[Task]:\n", " common_vars = (\n", " f\"GOAL: {inputs['goal']}\\n\"\n", " f\"START_DATE: {inputs['start_date']}\\n\"\n", " f\"DEADLINE_DAYS: {inputs['deadline_days']}\\n\"\n", " f\"HOURS_PER_DAY: {inputs['hours_per_day']}\\n\"\n", " )\n", "\n", " t1 = Task(\n", " description=textwrap.dedent(f\"\"\"\n", " Clarify the user's goal.\n", "\n", " Context:\n", " {common_vars}\n", "\n", " Return STRICT JSON with keys:\n", " {{\n", " \"clarified_goal\": \"...\",\n", " \"assumptions\": [\"...\"],\n", " \"scope\": [\"in-scope\", \"...\"],\n", " \"success_criteria\": [\"...\"],\n", " \"constraints\": [\"...\"],\n", " \"risks\": [\"...\"]\n", " }}\n", " \"\"\").strip(),\n", " agent=goal_understanding,\n", " expected_output=\"JSON only.\",\n", " )\n", "\n", " t2 = Task(\n", " description=textwrap.dedent(f\"\"\"\n", " Break the clarified goal into milestones and granular tasks.\n", "\n", " Input is the JSON from the previous step.\n", " Return STRICT JSON:\n", " {{\n", " \"milestones\": [\n", " {{\n", " \"name\": \"...\",\n", " \"description\": \"...\",\n", " \"tasks\": [\n", " {{\n", " \"id\": \"T1\",\n", " \"title\": \"...\",\n", " \"definition_of_done\": \"...\",\n", " \"dependencies\": [],\n", " \"tags\": [\"...\"]\n", " }}\n", " ]\n", " }}\n", " ]\n", " }}\n", " \"\"\").strip(),\n", " agent=task_breakdown,\n", " expected_output=\"JSON only.\",\n", " )\n", "\n", " t3 = Task(\n", " description=textwrap.dedent(f\"\"\"\n", " Estimate effort and priority for EACH task id from the previous output.\n", "\n", " Respect HOURS_PER_DAY={inputs['hours_per_day']} and DEADLINE_DAYS={inputs['deadline_days']}.\n", " Return STRICT JSON:\n", " {{\n", " \"estimates\": [\n", " {{\n", " \"id\": \"T1\",\n", " \"hours\": 2.5,\n", " \"difficulty_1to5\": 3,\n", " \"priority_1to5\": 1\n", " }}\n", " ]\n", " }}\n", " \"\"\").strip(),\n", " agent=effort_priority,\n", " expected_output=\"JSON only.\",\n", " )\n", "\n", " t4 = Task(\n", " description=textwrap.dedent(f\"\"\"\n", " Create a day-by-day schedule mapping tasks to calendar dates.\n", "\n", " Use START_DATE and DEADLINE_DAYS. Do not exceed HOURS_PER_DAY per day.\n", " Return STRICT JSON:\n", " {{\n", " \"schedule\": [\n", " {{\n", " \"date\": \"YYYY-MM-DD\",\n", " \"allocated_hours\": 2,\n", " \"tasks\": [{{\"id\":\"T1\",\"title\":\"...\",\"hours\":1.5}}]\n", " }}\n", " ],\n", " \"summary\": {{\n", " \"total_hours\": 0,\n", " \"days_planned\": 0,\n", " \"buffer_hours\": 0\n", " }}\n", " }}\n", " \"\"\").strip(),\n", " agent=scheduler,\n", " expected_output=\"JSON only.\",\n", " )\n", "\n", " t5 = Task(\n", " description=textwrap.dedent(f\"\"\"\n", " Provide 5 short motivation or execution tactics tailored to the plan.\n", "\n", " Return STRICT JSON:\n", " {{\n", " \"tips\": [\n", " \"Keep daily sessions short & focused with a single visible deliverable.\",\n", " \"...\"\n", " ]\n", " }}\n", " \"\"\").strip(),\n", " agent=motivator,\n", " expected_output=\"JSON only.\",\n", " )\n", "\n", " t6 = Task(\n", " description=textwrap.dedent(f\"\"\"\n", " Synthesize a final MARKDOWN report AND a JSON bundle.\n", "\n", " Inputs are the JSONs from all prior tasks. Build:\n", " 1) Markdown (for printing) with sections:\n", " - Goal Summary\n", " - Milestones & Tasks\n", " - Estimates & Priorities\n", " - Day-by-Day Plan (table)\n", " - Tips (bulleted)\n", " 2) JSON bundle containing all prior JSON merged into one: keys:\n", " {{\n", " \"goal_context\": ...,\n", " \"work_breakdown\": ...,\n", " \"estimates\": ...,\n", " \"schedule\": ...,\n", " \"tips\": ...\n", " }}\n", "\n", " Return STRICT JSON:\n", " {{\n", " \"markdown\": \"....\",\n", " \"bundle\": {{ /* merged JSON object */ }}\n", " }}\n", " \"\"\").strip(),\n", " agent=synthesizer,\n", " expected_output=\"JSON only.\",\n", " )\n", "\n", " return [t1, t2, t3, t4, t5, t6]\n" ], "metadata": { "id": "bULXBMGZoacv" }, "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "source": [ "# 🏃‍♂️ Step 5: Run the Crew" ], "metadata": { "id": "CBxM7WSqrFzR" } }, { "cell_type": "code", "source": [ "def run_planner(goal: str, deadline_days: int = 14, hours_per_day: int = 2, start_date: Optional[str] = None):\n", " inputs = default_inputs(goal, deadline_days, hours_per_day, start_date)\n", "\n", " tasks = make_tasks(inputs)\n", " crew = Crew(\n", " agents=[goal_understanding, task_breakdown, effort_priority, scheduler, motivator, synthesizer],\n", " tasks=tasks,\n", " process=Process.sequential, # run in order\n", " verbose=True,\n", " memory=True,\n", " cache=True,\n", " )\n", "\n", " result = crew.kickoff(inputs=inputs)\n", "\n", " # The final task returns JSON with 'markdown' and 'bundle'\n", " # crewai often returns extra text around json; ensure it's valid JSON:\n", " def coerce_json(text: str) -> Dict[str, Any]:\n", " try:\n", " return json.loads(text)\n", " except Exception:\n", " # Try to extract JSON block heuristically\n", " start = text.find(\"{\")\n", " end = text.rfind(\"}\")\n", " if start != -1 and end != -1:\n", " return json.loads(text[start:end+1])\n", " raise\n", "\n", " final = coerce_json(str(result))\n", " md = final.get(\"markdown\", \"# Plan\\n\\n(No markdown received)\")\n", " bundle = final.get(\"bundle\", {})\n", "\n", " # Save artifacts\n", " out_dir = \"/content/task_master_outputs\"\n", " os.makedirs(out_dir, exist_ok=True)\n", " stamp = dt.datetime.now().strftime(\"%Y%m%d_%H%M%S\")\n", "\n", " md_path = os.path.join(out_dir, f\"plan_{stamp}.md\")\n", " json_path = os.path.join(out_dir, f\"plan_{stamp}.json\")\n", "\n", " with open(md_path, \"w\", encoding=\"utf-8\") as f:\n", " f.write(md)\n", " with open(json_path, \"w\", encoding=\"utf-8\") as f:\n", " json.dump(bundle, f, indent=2, ensure_ascii=False)\n", "\n", " return {\n", " \"markdown_path\": md_path,\n", " \"json_path\": json_path,\n", " \"markdown_preview\": md[:1500] + (\"\\n\\n...[truncated]...\" if len(md) > 1500 else \"\"),\n", " }\n" ], "metadata": { "id": "wikuoefWoc0Z" }, "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "source": [ "# Step 6: Minimal Gradio UI" ], "metadata": { "id": "bbs5keN2rOAh" } }, { "cell_type": "code", "source": [ "import gradio as gr\n", "\n", "def gradio_run(goal, deadline_days, hours_per_day, start_date):\n", " out = run_planner(goal, int(deadline_days), int(hours_per_day), start_date or None)\n", " md = \"\"\n", " with open(out[\"markdown_path\"], \"r\", encoding=\"utf-8\") as f:\n", " md = f.read()\n", " return md, out[\"markdown_path\"], out[\"json_path\"]\n", "\n", "with gr.Blocks(title=\"CrewAI Task Master\") as demo_ui:\n", " gr.Markdown(\"# 🧠 CrewAI Task Master\\nPlan any goal into a day-by-day schedule.\")\n", " with gr.Row():\n", " goal = gr.Textbox(label=\"Your Goal\", placeholder=\"e.g., Build an AI portfolio website in 10 days\", lines=3)\n", " with gr.Row():\n", " deadline = gr.Number(label=\"Deadline (days)\", value=14, precision=0)\n", " hours = gr.Number(label=\"Hours per day\", value=2, precision=0)\n", " start = gr.Textbox(label=\"Start Date (YYYY-MM-DD, optional)\", placeholder=\"Leave blank for today\")\n", " run_btn = gr.Button(\"Run Agents 🚀\")\n", "\n", " md_out = gr.Markdown(label=\"Plan (Markdown)\")\n", " md_file = gr.File(label=\"Download Markdown\")\n", " json_file = gr.File(label=\"Download JSON\")\n", "\n", " run_btn.click(\n", " gradio_run,\n", " inputs=[goal, deadline, hours, start],\n", " outputs=[md_out, md_file, json_file]\n", " )\n", "\n", "demo_ui.launch(share=False)\n" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/", "height": 599 }, "id": "HDJKz5r4oisy", "outputId": "1c91172e-6ac4-4d3e-8074-1ad8ced4fc9a" }, "execution_count": null, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "Colab notebook detected. To show errors in colab notebook, set debug=True in launch()\n", "Note: opening Chrome Inspector may crash demo inside Colab notebooks.\n", "* To create a public link, set `share=True` in `launch()`.\n" ] }, { "output_type": "display_data", "data": { "text/plain": [ "" ], "application/javascript": [ "(async (port, path, width, height, cache, element) => {\n", " if (!google.colab.kernel.accessAllowed && !cache) {\n", " return;\n", " }\n", " element.appendChild(document.createTextNode(''));\n", " const url = await google.colab.kernel.proxyPort(port, {cache});\n", "\n", " const external_link = document.createElement('div');\n", " external_link.innerHTML = `\n", "
\n", " Running on \n", " https://localhost:${port}${path}\n", " \n", "
\n", " `;\n", " element.appendChild(external_link);\n", "\n", " const iframe = document.createElement('iframe');\n", " iframe.src = new URL(path, url).toString();\n", " iframe.height = height;\n", " iframe.allow = \"autoplay; camera; microphone; clipboard-read; clipboard-write;\"\n", " iframe.width = width;\n", " iframe.style.border = 0;\n", " element.appendChild(iframe);\n", " })(7860, \"/\", \"100%\", 500, false, window.element)" ] }, "metadata": {} }, { "output_type": "execute_result", "data": { "text/plain": [] }, "metadata": {}, "execution_count": 11 } ] }, { "cell_type": "code", "source": [], "metadata": { "id": "YC9b0G8romnv" }, "execution_count": null, "outputs": [] } ] }