Разработка автономной AI-системы управления проектами

Разработка автономной AI-системы управления проектами

AI-система управления проектами автоматизирует рутину: декомпозицию задач, отслеживание прогресса, выявление рисков, перераспределение ресурсов при блокерах, генерацию отчётов и проактивное оповещение стейкхолдеров. Цель — снятие операционной нагрузки с PM и фокус команды на содержательной работе, а не административных задачах.

Возможности системы

Автоматическая декомпозиция: превращает требование или эпик в структурированный бэклог с задачами, критериями приёмки и оценками.

Risk Detection: анализирует состояние спринта и выявляет риски срыва дедлайна до того, как они реализуются.

Progress Reporting: генерирует отчёты для команды, стейкхолдеров, топ-менеджмента — в нужном формате для каждой аудитории.

Resource Optimizer: при блокерах предлагает перераспределение задач с учётом компетенций.

Ядро системы: AI-PM агент

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated, Optional
import operator

class ProjectState(TypedDict):
    project_id: str
    sprint_data: dict          # Задачи, статусы, assignee, SP
    team_capacity: dict        # Доступность каждого члена команды
    blockers: list[dict]
    risk_flags: list[dict]
    action_items: Annotated[list, operator.add]
    generated_reports: Annotated[list, operator.add]
    notifications_sent: Annotated[list, operator.add]

llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

PM_SYSTEM_PROMPT = """Ты — AI-менеджер проекта.
Анализируй данные спринта и принимай решения, которые принял бы опытный Scrum Master:
- Выявляй риски срыва по velocity trend и текущему прогрессу
- Предлагай конкретные действия (не абстрактные советы)
- Учитывай зависимости между задачами
- Эскалируй блокеры, требующие решения PM или стейкхолдера"""

Инструменты PM-агента

from langchain_core.tools import tool
import json

@tool
def get_sprint_metrics(project_id: str, sprint_id: str) -> str:
    """Получить текущие метрики спринта: velocity, burn-down, blockers"""
    sprint = jira_client.get_sprint(project_id, sprint_id)

    done_sp = sum(t["story_points"] for t in sprint["tasks"] if t["status"] == "Done")
    in_progress_sp = sum(t["story_points"] for t in sprint["tasks"] if t["status"] == "In Progress")
    total_sp = sum(t["story_points"] for t in sprint["tasks"])
    days_remaining = sprint["remaining_days"]
    days_total = sprint["total_days"]

    velocity_expected = total_sp * (1 - days_remaining / days_total)
    velocity_gap = done_sp - velocity_expected

    return json.dumps({
        "done_sp": done_sp,
        "in_progress_sp": in_progress_sp,
        "total_sp": total_sp,
        "velocity_gap": round(velocity_gap, 1),
        "at_risk_tasks": [t for t in sprint["tasks"] if t.get("blocked") or t.get("overdue")],
        "days_remaining": days_remaining,
    })

@tool
def analyze_blockers(project_id: str) -> str:
    """Получить список активных блокеров с контекстом"""
    blockers = jira_client.get_blockers(project_id)
    enriched = []
    for b in blockers:
        enriched.append({
            "task_id": b["id"],
            "title": b["title"],
            "blocked_since": b["blocked_since"],
            "blocking_tasks": b.get("dependents", []),
            "assignee": b["assignee"],
            "blocker_reason": b.get("blocker_reason", "Не указано"),
        })
    return json.dumps(enriched)

@tool
def create_task(
    project_id: str,
    title: str,
    description: str,
    assignee: str,
    story_points: int,
    labels: list[str],
) -> str:
    """Создать задачу в Jira"""
    task = jira_client.create_issue(
        project=project_id,
        summary=title,
        description=description,
        assignee=assignee,
        story_points=story_points,
        labels=labels,
    )
    return f"Задача создана: {task['key']} — {task['url']}"

@tool
def send_standup_reminder(project_id: str, message: str, channel: str = "slack") -> str:
    """Отправить напоминание команде в Slack"""
    slack_client.post_message(channel=channel, text=message)
    return f"Уведомление отправлено в {channel}"

@tool
def decompose_epic(epic_description: str, team_skills: list[str]) -> str:
    """Декомпозировать эпик в задачи с оценками"""
    # Этот инструмент вызывает отдельный LLM для декомпозиции
    decomposer_llm = ChatOpenAI(model="gpt-4o")
    result = decomposer_llm.invoke(f"""Декомпозируй следующий эпик на задачи.

Эпик: {epic_description}
Компетенции команды: {team_skills}

Верни JSON-список задач:
[{{
  "title": "Название задачи",
  "description": "Подробное описание и критерии приёмки",
  "story_points": 3,
  "required_skills": ["Python", "PostgreSQL"],
  "dependencies": []
}}]""")
    return result.content

Автоматическое обнаружение рисков

class RiskDetector:
    """Анализирует спринт на риски срыва дедлайна"""

    RISK_THRESHOLDS = {
        "velocity_gap_critical": -5,  # SP отставания — критично
        "blocker_age_days": 2,        # Блокер старше 2 дней
        "overloaded_developer": 1.4,  # 140% от capacity
    }

    async def analyze_sprint_risks(self, sprint_data: dict) -> list[dict]:
        risks = []

        # Проверка velocity
        velocity_gap = sprint_data.get("velocity_gap", 0)
        if velocity_gap < self.RISK_THRESHOLDS["velocity_gap_critical"]:
            risks.append({
                "type": "velocity_lag",
                "severity": "high",
                "message": f"Отставание {abs(velocity_gap)} SP от планового прогресса",
                "recommended_action": "Ревью скопа спринта — возможно перенести задачи",
            })

        # Проверка блокеров
        for blocker in sprint_data.get("at_risk_tasks", []):
            blocked_days = blocker.get("blocked_days", 0)
            if blocked_days >= self.RISK_THRESHOLDS["blocker_age_days"]:
                risks.append({
                    "type": "prolonged_blocker",
                    "severity": "critical" if blocked_days > 3 else "high",
                    "task": blocker["task_id"],
                    "message": f"Задача {blocker['task_id']} заблокирована {blocked_days} дней",
                    "recommended_action": "Требует немедленного вмешательства PM",
                })

        # AI-анализ паттернов риска
        if risks:
            risk_assessment = await self.llm_risk_assessment(sprint_data, risks)
            risks.extend(risk_assessment)

        return risks

    async def llm_risk_assessment(self, sprint_data: dict, known_risks: list) -> list[dict]:
        """LLM анализирует тонкие паттерны рисков"""
        response = llm.invoke(f"""Проанализируй данные спринта на скрытые риски.

Данные спринта: {json.dumps(sprint_data, ensure_ascii=False)}
Уже выявленные риски: {json.dumps(known_risks, ensure_ascii=False)}

Выяви дополнительные риски, которые не видны из метрик (зависимости между задачами,
концентрация рисков на одном разработчике, технический долг в критическом пути).

Верни JSON-список с полями: type, severity, message, recommended_action""")

        try:
            return json.loads(response.content)
        except Exception:
            return []

Генерация отчётов для разных аудиторий

class ReportGenerator:
    """Генерирует отчёты, адаптированные под аудиторию"""

    async def generate_standup_summary(self, sprint_data: dict) -> str:
        """Краткая сводка для ежедневного standup"""
        return await llm.ainvoke(f"""Создай краткую сводку для standup (5-7 пунктов).
Формат: markdown. Акцент на: что завершено, блокеры, риски на сегодня.
Данные: {json.dumps(sprint_data, ensure_ascii=False, indent=2)}""")

    async def generate_stakeholder_report(self, project_data: dict) -> str:
        """Отчёт для стейкхолдеров — без технических деталей"""
        return await llm.ainvoke(f"""Создай еженедельный отчёт для стейкхолдеров.
Формат: краткое резюме, прогресс по целям спринта (%), риски, следующие шаги.
Исключи технические детали. Фокус на бизнес-ценности.
Данные: {json.dumps(project_data, ensure_ascii=False, indent=2)}""")

    async def generate_executive_summary(self, portfolio_data: dict) -> str:
        """Executive summary для топ-менеджмента"""
        return await llm.ainvoke(f"""Создай executive summary по портфелю проектов.
1 страница. Светофор по каждому проекту. Топ-3 риска. Требуемые решения от руководства.
Данные: {json.dumps(portfolio_data, ensure_ascii=False, indent=2)}""")

Интеграции

# Jira Integration
from jira import JIRA
jira = JIRA(server="https://company.atlassian.net", basic_auth=("[email protected]", "token"))

# Confluence для документации
from atlassian import Confluence
confluence = Confluence(url="https://company.atlassian.net", username="user", password="token")

# Slack notifications
from slack_sdk import WebClient
slack = WebClient(token="xoxb-...")

# GitLab/GitHub для pull request статистики
import gitlab
gl = gitlab.Gitlab("https://gitlab.company.com", private_token="token")

Практический кейс: SaaS-компания, 8 продуктовых команд

Ситуация: 8 Scrum-команд, 2 engineering manager, PM тратил ~30% времени на сбор статусов и формирование отчётов.

Внедрённые автоматизации:

1. Ежедневный standup digest в Slack в 9:45 (до митинга)
2. Автоматическое создание задач на основе Slack-сообщений «нужно сделать X»
3. Weekly stakeholder report каждую пятницу
4. Risk alert при velocity gap > 3 SP
5. Блокер эскалация: если задача заблокирована > 2 дней без комментариев

Результаты:

• Время PM на административные задачи: -42%
• Блокеры старше 3 дней: -71%
• Срывы спринт-целей (с missed → achieved): -33%
• Оценка полезности команды: 4.1/5.0 (без «мешает работе»)

Что не автоматизировано: переговоры со стейкхолдерами о приоритетах, решения о найме, техническое направление.

Сроки

• Базовый PM-агент с инструментами Jira/Slack: 2–3 недели
• Risk Detection и автоматические алерты: 1–2 недели
• Генератор отчётов с шаблонами: 1–2 недели
• Декомпозитор эпиков: 1 неделя
• Интеграция в workflow команды и настройка: 2–3 недели
• Итого: 7–11 недель