AI Project Manager — Digital Worker
AI Project Manager autonomously manages project's administrative dimension: task decomposition, backlog management, progress tracking, report generation, risk monitoring, meeting coordination. PM agent does not make strategic decisions but removes operational workload from real PMs — tasks occupying 40–50% of work time.
Requirement Decomposition into Tasks
from openai import AsyncOpenAI
from pydantic import BaseModel
from typing import Literal, Optional
client = AsyncOpenAI()
class ProjectTask(BaseModel):
title: str
description: str
acceptance_criteria: list[str]
story_points: int # Fibonacci: 1, 2, 3, 5, 8, 13
task_type: Literal["feature", "bug", "tech_debt", "research", "devops"]
required_skills: list[str]
dependencies: list[str] # Names of dependent tasks
priority: Literal["critical", "high", "medium", "low"]
risk_notes: Optional[str]
async def decompose_requirement(
requirement: str,
team_skills: list[str],
existing_codebase_context: str = "",
) -> list[ProjectTask]:
response = await client.beta.chat.completions.parse(
model="gpt-4o",
messages=[{
"role": "system",
"content": f"""You are an experienced tech lead and PM.
Decompose requirement into specific tasks for the team.
Decomposition principles:
- Each task completable in 1-3 days by one developer
- Acceptance criteria — specific, verifiable
- Specify task dependencies
- Story points: use Fibonacci, base on complexity
Team skills: {team_skills}
Codebase context: {existing_codebase_context[:500] if existing_codebase_context else 'not provided'}"""
}, {
"role": "user",
"content": f"Requirement: {requirement}",
}],
response_format=list[ProjectTask],
temperature=0.2,
)
return response.choices[0].message.parsed
Sprint Planning Agent
class SprintPlanningAgent:
async def plan_sprint(
self,
backlog: list[dict],
team_capacity: dict, # {developer: available_hours}
sprint_goal: str,
velocity_history: list[int],
) -> dict:
"""Creates sprint plan respecting team capacity"""
available_sp = self.estimate_capacity(team_capacity, velocity_history)
sprint_plan = await client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "system",
"content": """You are a Scrum Master planning a sprint.
Select tasks from backlog, respecting:
1. Sprint goal — tasks must align
2. Team capacity not exceeded
3. Dependencies — can't take task if dependency incomplete
4. Balance: mix of features, bugs, tech debt"""
}, {
"role": "user",
"content": f"""Sprint goal: {sprint_goal}
Available capacity: {available_sp} SP
Team availability: {team_capacity}
Backlog (top-30 by priority):
{json.dumps(backlog[:30], ensure_ascii=False, indent=2)}
Return JSON: {{"selected_tasks": [...task_ids], "assignments": {{developer: [task_ids]}}, "sprint_risk": "low/medium/high", "risk_explanation": "..."}}"""
}],
response_format={"type": "json_object"},
)
return json.loads(sprint_plan.choices[0].message.content)
def estimate_capacity(self, team_capacity: dict, velocity_history: list[int]) -> int:
avg_velocity = sum(velocity_history[-5:]) / len(velocity_history[-5:])
total_hours = sum(team_capacity.values())
standard_sprint_hours = 8 * 10 * len(team_capacity) # 2 weeks
capacity_ratio = total_hours / standard_sprint_hours
return int(avg_velocity * capacity_ratio)
Daily Standup Automation
class StandupBot:
async def collect_and_summarize(self, project_id: str) -> str:
"""Collects progress data and generates standup digest"""
# Data from Jira/GitHub
jira_updates = await jira.get_yesterday_updates(project_id)
github_commits = await github.get_commits(project_id, since="yesterday")
blockers = await jira.get_current_blockers(project_id)
open_prs = await github.get_open_prs(project_id)
digest = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "system",
"content": "Create concise standup digest. Format: ✅ Done, 🔄 In Progress, 🚧 Blockers. Concrete, no filler."
}, {
"role": "user",
"content": f"""Jira updates:
{json.dumps(jira_updates, ensure_ascii=False, indent=2)}
Commits:
{json.dumps(github_commits[:10], ensure_ascii=False, indent=2)}
Blockers:
{json.dumps(blockers, ensure_ascii=False, indent=2)}
Open PRs: {len(open_prs)}, waiting >24h: {sum(1 for p in open_prs if p['waiting_hours'] > 24)}"""
}],
)
return digest.choices[0].message.content
async def post_to_slack(self, digest: str, channel: str):
await slack_client.chat_postMessage(
channel=channel,
text=f"*Standup Digest — {datetime.now().strftime('%d.%m.%Y')}*\n{digest}",
)
Risk Monitoring
class ProjectRiskMonitor:
async def assess_risks(self, project_data: dict) -> list[dict]:
"""Automatically identifies and assesses project risks"""
# Numeric risk signals
numeric_risks = []
sprint = project_data.get("current_sprint", {})
velocity_trend = project_data.get("velocity_trend", [])
if len(velocity_trend) >= 3 and velocity_trend[-1] < velocity_trend[-3] * 0.7:
numeric_risks.append({
"type": "velocity_decline",
"severity": "high",
"data": f"Velocity: {velocity_trend[-3]} → {velocity_trend[-1]} SP",
})
team_absences = project_data.get("planned_absences", [])
sprint_end = project_data.get("sprint_end_date")
critical_absence = any(
a for a in team_absences
if a.get("days") >= 3 and a.get("person") in sprint.get("key_developers", [])
)
if critical_absence:
numeric_risks.append({
"type": "key_person_absence",
"severity": "medium",
"data": "Key developer absent during critical period",
})
# LLM analyzes risk patterns
risk_assessment = await client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "system",
"content": "Identify hidden risks in project data. Return JSON list with severity and mitigation."
}, {
"role": "user",
"content": json.dumps({**project_data, "known_risks": numeric_risks}, ensure_ascii=False),
}],
response_format={"type": "json_object"},
)
ai_risks = json.loads(risk_assessment.choices[0].message.content).get("risks", [])
return numeric_risks + ai_risks
Case Study: Digital Product Studio, 6 Parallel Projects
Situation: 2 PMs managed 6 projects. 35% of time spent on status reports, sprint planning, blocker communication.
AI PM took on:
- Automatic standup digest to Slack every morning
- Weekly stakeholder report
- Risk warnings (velocity gap, blockers > 2 days)
- Epic decomposition on task creation
- Sprint planning preparation (task suggestions per capacity)
Results:
- PM administrative time: 35% → 18%
- PMs able to take 3rd project on 1 PM
- Sprint failures: -44% (early risk warnings)
- Team satisfaction: 4.1/5.0 for AI PM utility (no "surveillance" feeling)
Timeline
- Sprint planning and decomposition: 2–3 weeks
- Standup bot and monitoring: 1–2 weeks
- Risk assessment and alerts: 1–2 weeks
- Jira/GitHub/Slack integrations: 1–2 weeks
- Total: 5–9 weeks