657 lines
24 KiB
Python
657 lines
24 KiB
Python
"""AI Steward Service - 主动式 AI 管家核心服务。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
from typing import Any, Dict, List
|
||
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from ..core.logging import get_logger
|
||
from ..database import get_db
|
||
from ..models import CognitiveConflict, Project, ProjectSignalLink, Signal, TruthNode
|
||
from ..utils.project_members import ensure_project_default_members
|
||
from ..utils.sequence import ensure_project_id_sequence
|
||
from .adaptive_rag_service import AdaptiveRAGService
|
||
from .llm import LLMService
|
||
from .task_persistence import (
|
||
build_task_create_draft,
|
||
normalize_skill_names,
|
||
persist_task,
|
||
resolve_skill_tag_ids_by_names,
|
||
)
|
||
|
||
logger = get_logger(__name__)
|
||
PROJECT_CREATION_DRAFT_TIMEOUT_SECONDS = 8.0
|
||
|
||
|
||
class AIStewardService:
|
||
"""主动式 AI 管家服务。"""
|
||
|
||
def __init__(self, db: AsyncSession):
|
||
self.db = db
|
||
self.llm = LLMService()
|
||
|
||
async def create_project_from_natural_language(
|
||
self,
|
||
input_text: str,
|
||
creator_id: int,
|
||
tenant_id: int = 1,
|
||
) -> Dict[str, Any]:
|
||
"""零配置项目创建，统一走 draft -> normalize -> persist 三段式。"""
|
||
analysis = await self._analyze_project_input(input_text, tenant_id)
|
||
draft = await self._generate_project_creation_draft(
|
||
input_text=input_text,
|
||
analysis=analysis,
|
||
tenant_id=tenant_id,
|
||
)
|
||
normalized = await self._normalize_project_creation_draft(
|
||
draft=draft,
|
||
input_text=input_text,
|
||
creator_id=creator_id,
|
||
tenant_id=tenant_id,
|
||
)
|
||
payload = await self._persist_project_creation_draft(
|
||
normalized=normalized,
|
||
input_text=input_text,
|
||
creator_id=creator_id,
|
||
tenant_id=tenant_id,
|
||
)
|
||
|
||
await self.db.commit()
|
||
|
||
try:
|
||
charter = payload["charter"]
|
||
project = payload["project"]
|
||
asyncio.create_task(
|
||
AdaptiveRAGService.upsert_pmflow_record_async(
|
||
tenant_id=tenant_id,
|
||
record_type="charter",
|
||
record_id=int(charter["id"]),
|
||
project_id=int(project["id"]),
|
||
title=str(charter.get("title") or "项目宪章"),
|
||
content=json.dumps(charter.get("content") or {}, ensure_ascii=False),
|
||
)
|
||
)
|
||
for task in payload.get("tasks", []):
|
||
task_id = int(str(task.get("id") or "0") or 0)
|
||
if task_id <= 0:
|
||
continue
|
||
asyncio.create_task(
|
||
AdaptiveRAGService.upsert_pmflow_record_async(
|
||
tenant_id=tenant_id,
|
||
record_type="task",
|
||
record_id=task_id,
|
||
project_id=int(project["id"]),
|
||
title=str(task.get("title") or "任务"),
|
||
content=str(task.get("description") or ""),
|
||
)
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
return payload
|
||
|
||
async def _generate_project_creation_draft(
|
||
self,
|
||
*,
|
||
input_text: str,
|
||
analysis: Dict[str, Any],
|
||
tenant_id: int,
|
||
) -> dict[str, Any]:
|
||
signal_draft, charter_draft, task_drafts = await asyncio.gather(
|
||
self._generate_competitive_signal_draft(analysis=analysis, tenant_id=tenant_id),
|
||
self._generate_charter_draft(analysis=analysis, tenant_id=tenant_id),
|
||
self._generate_task_chain_draft(analysis=analysis, tenant_id=tenant_id),
|
||
)
|
||
return {
|
||
"analysis": analysis,
|
||
"project": {
|
||
"name": analysis.get("project_name") or input_text[:20] or "未命名项目",
|
||
"title": analysis.get("project_title") or input_text[:50] or "未命名项目",
|
||
"description": input_text,
|
||
"assigned_role": analysis.get("primary_role") or "backend",
|
||
"domain_id": analysis.get("domain_id"),
|
||
},
|
||
"signal": signal_draft,
|
||
"charter": charter_draft,
|
||
"tasks": task_drafts,
|
||
"risks": self._build_risk_alert_drafts(analysis=analysis),
|
||
}
|
||
|
||
async def _generate_project_creation_text(
|
||
self,
|
||
*,
|
||
prompt: str,
|
||
tenant_id: int,
|
||
function_name: str,
|
||
timeout_seconds: float = PROJECT_CREATION_DRAFT_TIMEOUT_SECONDS,
|
||
) -> str:
|
||
try:
|
||
return await asyncio.wait_for(
|
||
self.llm.generate(
|
||
prompt,
|
||
db=self.db,
|
||
tenant_id=tenant_id,
|
||
function_name=function_name,
|
||
),
|
||
timeout=timeout_seconds,
|
||
)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"ai_steward_project_creation_step_fallback",
|
||
tenant_id=int(tenant_id),
|
||
function_name=function_name,
|
||
error=str(exc),
|
||
)
|
||
return ""
|
||
|
||
async def _normalize_project_creation_draft(
|
||
self,
|
||
*,
|
||
draft: dict[str, Any],
|
||
input_text: str,
|
||
creator_id: int,
|
||
tenant_id: int,
|
||
) -> dict[str, Any]:
|
||
analysis = draft.get("analysis") if isinstance(draft.get("analysis"), dict) else {}
|
||
project_draft = draft.get("project") if isinstance(draft.get("project"), dict) else {}
|
||
charter_draft = draft.get("charter") if isinstance(draft.get("charter"), dict) else {}
|
||
signal_draft = draft.get("signal") if isinstance(draft.get("signal"), dict) else {}
|
||
risk_drafts = draft.get("risks") if isinstance(draft.get("risks"), list) else []
|
||
|
||
normalized_task_drafts = []
|
||
task_audit: list[dict[str, Any]] = []
|
||
for raw_task in (draft.get("tasks") if isinstance(draft.get("tasks"), list) else [])[:12]:
|
||
if not isinstance(raw_task, dict):
|
||
continue
|
||
raw_required_skills = normalize_skill_names(raw_task.get("required_skills"))
|
||
resolution = await resolve_skill_tag_ids_by_names(
|
||
self.db,
|
||
tenant_id=tenant_id,
|
||
skill_names=raw_required_skills,
|
||
)
|
||
validation_error = None
|
||
try:
|
||
task_draft = build_task_create_draft(
|
||
title=str(raw_task.get("title") or ""),
|
||
description=str(raw_task.get("description") or "").strip() or None,
|
||
required_skill_tag_ids=resolution.matched_ids,
|
||
priority=raw_task.get("priority"),
|
||
status="todo",
|
||
estimated_hours=raw_task.get("estimated_hours"),
|
||
due_date=raw_task.get("due_date"),
|
||
)
|
||
except ValueError as exc:
|
||
validation_error = str(exc)
|
||
task_draft = build_task_create_draft(
|
||
title=str(raw_task.get("title") or ""),
|
||
description=str(raw_task.get("description") or "").strip() or None,
|
||
required_skill_tag_ids=resolution.matched_ids,
|
||
priority=raw_task.get("priority"),
|
||
status="todo",
|
||
estimated_hours=None,
|
||
due_date=None,
|
||
)
|
||
normalized_task_drafts.append(task_draft)
|
||
task_audit.append(
|
||
{
|
||
"raw_task_draft": raw_task,
|
||
"normalized_title": task_draft.title,
|
||
"required_skills_input": raw_required_skills,
|
||
"required_skill_tag_ids": resolution.matched_ids,
|
||
"skill_resolution": resolution.matched_items,
|
||
"dropped_required_skills": resolution.unmatched_names,
|
||
"validation_error": validation_error,
|
||
}
|
||
)
|
||
|
||
normalized_project = {
|
||
"name": str(project_draft.get("name") or "未命名项目").strip()[:50] or "未命名项目",
|
||
"title": str(project_draft.get("title") or input_text[:50] or "未命名项目").strip()[:50] or "未命名项目",
|
||
"description": str(project_draft.get("description") or input_text).strip()[:4000] or input_text[:4000],
|
||
"assigned_role": str(project_draft.get("assigned_role") or "backend").strip()[:50] or "backend",
|
||
"domain_id": analysis.get("domain_id"),
|
||
}
|
||
normalized_charter_content = charter_draft.get("content") if isinstance(charter_draft.get("content"), dict) else {
|
||
"rationale": str(charter_draft.get("content") or ""),
|
||
"success_criteria": charter_draft.get("success_criteria") or [],
|
||
"constraints": charter_draft.get("constraints") or [],
|
||
"signal_ids": [],
|
||
}
|
||
normalized_signal = {
|
||
"content": str(signal_draft.get("content") or "").strip()[:2000] or f"竞品分析：{normalized_project['title']}",
|
||
"impact_score": float(signal_draft.get("impact_score") or 0.8),
|
||
"priority": str(signal_draft.get("priority") or "high").strip().lower() or "high",
|
||
"strategic_analysis": signal_draft.get("strategic_analysis") if isinstance(signal_draft.get("strategic_analysis"), dict) else {},
|
||
}
|
||
normalized_risks = [risk for risk in risk_drafts if isinstance(risk, dict)]
|
||
|
||
return {
|
||
"analysis": analysis,
|
||
"project": normalized_project,
|
||
"signal": normalized_signal,
|
||
"charter": {
|
||
"title": str(charter_draft.get("title") or f"{normalized_project['name']}项目宪章").strip()[:200] or "项目宪章",
|
||
"content": normalized_charter_content,
|
||
},
|
||
"tasks": normalized_task_drafts,
|
||
"task_audit": task_audit,
|
||
"risks": normalized_risks,
|
||
"creator_id": int(creator_id),
|
||
}
|
||
|
||
async def _persist_project_creation_draft(
|
||
self,
|
||
*,
|
||
normalized: dict[str, Any],
|
||
input_text: str,
|
||
creator_id: int,
|
||
tenant_id: int,
|
||
) -> dict[str, Any]:
|
||
project_data = normalized["project"]
|
||
await ensure_project_id_sequence(self.db)
|
||
project = Project(
|
||
name=project_data["name"],
|
||
title=project_data["title"],
|
||
description=project_data["description"],
|
||
owner_id=int(creator_id),
|
||
status="active",
|
||
domain_id=project_data.get("domain_id"),
|
||
assigned_role=project_data["assigned_role"],
|
||
signal_count=0,
|
||
tenant_id=int(tenant_id),
|
||
)
|
||
self.db.add(project)
|
||
await self.db.flush()
|
||
await ensure_project_default_members(
|
||
self.db,
|
||
tenant_id=int(tenant_id),
|
||
project_id=int(project.id),
|
||
owner_id=int(creator_id),
|
||
assigned_role=project.assigned_role,
|
||
)
|
||
|
||
signal_data = normalized["signal"]
|
||
signal = Signal(
|
||
source="AI-Steward",
|
||
content=signal_data["content"],
|
||
impact_score=signal_data["impact_score"],
|
||
priority=signal_data["priority"],
|
||
processed=True,
|
||
strategic_analysis=json.dumps(signal_data.get("strategic_analysis") or {}, ensure_ascii=False),
|
||
tasks_generated=json.dumps([]),
|
||
tenant_id=int(tenant_id),
|
||
)
|
||
self.db.add(signal)
|
||
await self.db.flush()
|
||
|
||
self.db.add(
|
||
ProjectSignalLink(
|
||
tenant_id=int(tenant_id),
|
||
project_id=int(project.id),
|
||
signal_id=int(signal.id),
|
||
link_type="competitive_analysis",
|
||
relevance_score=0.9,
|
||
)
|
||
)
|
||
|
||
charter_data = normalized["charter"]
|
||
charter = TruthNode(
|
||
node_type="charter",
|
||
project_id=int(project.id),
|
||
created_by=int(creator_id),
|
||
title=charter_data["title"],
|
||
content=charter_data["content"],
|
||
version=1,
|
||
status="active",
|
||
tenant_id=int(tenant_id),
|
||
)
|
||
self.db.add(charter)
|
||
await self.db.flush()
|
||
|
||
persisted_tasks: list[dict[str, Any]] = []
|
||
persisted_task_ids: list[int] = []
|
||
for task_draft in normalized["tasks"]:
|
||
task = await persist_task(
|
||
self.db,
|
||
tenant_id=int(tenant_id),
|
||
truth_node_id=int(charter.id),
|
||
project_id=int(project.id),
|
||
draft=task_draft,
|
||
actor_user_id=int(creator_id),
|
||
action="task.create",
|
||
remark="ai_steward.project_from_text",
|
||
commit=False,
|
||
refresh_workload=False,
|
||
)
|
||
persisted_task_ids.append(int(task.id))
|
||
persisted_tasks.append(
|
||
{
|
||
"id": str(task.id),
|
||
"title": task.title,
|
||
"description": task.description or "",
|
||
"estimated_hours": float(task.estimated_hours or 0),
|
||
"priority": task.priority,
|
||
"status": task.status,
|
||
}
|
||
)
|
||
|
||
risks_payload: list[dict[str, Any]] = []
|
||
for risk in normalized["risks"]:
|
||
description = str(risk.get("description") or "").strip()
|
||
conflict_type = str(risk.get("type") or "risk").strip() or "risk"
|
||
if not description:
|
||
continue
|
||
self.db.add(
|
||
CognitiveConflict(
|
||
truth_node_id=int(charter.id),
|
||
conflict_type=conflict_type,
|
||
description=description,
|
||
involved_user_ids=[],
|
||
status="active",
|
||
tenant_id=int(tenant_id),
|
||
)
|
||
)
|
||
risks_payload.append(
|
||
{
|
||
"type": conflict_type,
|
||
"description": description,
|
||
"threshold_hours": risk.get("threshold_hours"),
|
||
"threshold": risk.get("threshold"),
|
||
"severity": risk.get("severity"),
|
||
}
|
||
)
|
||
|
||
logger.info(
|
||
"ai_steward_project_from_text_persisted",
|
||
tenant_id=int(tenant_id),
|
||
creator_id=int(creator_id),
|
||
project_id=int(project.id),
|
||
signal_id=int(signal.id),
|
||
charter_id=int(charter.id),
|
||
persisted_task_ids=persisted_task_ids,
|
||
task_audit=normalized["task_audit"],
|
||
input_text=input_text[:500],
|
||
)
|
||
|
||
return {
|
||
"project": {
|
||
"id": str(project.id),
|
||
"name": project.name,
|
||
"title": project.title,
|
||
"description": project.description,
|
||
"status": project.status,
|
||
},
|
||
"signal": {
|
||
"id": str(signal.id),
|
||
"content": signal.content,
|
||
"impact_score": signal.impact_score,
|
||
},
|
||
"charter": {
|
||
"id": str(charter.id),
|
||
"title": charter.title,
|
||
"content": charter.content,
|
||
},
|
||
"tasks": persisted_tasks,
|
||
"risks": risks_payload,
|
||
"analysis": normalized["analysis"],
|
||
}
|
||
|
||
async def _analyze_project_input(self, input_text: str, tenant_id: int) -> Dict[str, Any]:
|
||
"""使用 AI 分析项目输入，提取关键信息。"""
|
||
prompt = f"""分析以下项目需求，提取关键信息并以 JSON 格式返回：
|
||
|
||
用户输入：{input_text}
|
||
|
||
请提取以下信息：
|
||
{{
|
||
"project_name": "项目名称（简短）",
|
||
"project_title": "项目标题",
|
||
"tech_stack": ["技术栈列表"],
|
||
"target_audience": "目标用户",
|
||
"timeline": "时间周期",
|
||
"primary_role": "主要角色类型（ai_engineer/backend/frontend）",
|
||
"domain_id": 1,
|
||
"key_features": ["核心功能列表"],
|
||
"competitors": ["竞品列表"],
|
||
"success_metrics": ["成功指标"]
|
||
}}
|
||
|
||
只返回 JSON，不要其他内容。"""
|
||
|
||
response = await self._generate_project_creation_text(
|
||
prompt=prompt,
|
||
tenant_id=tenant_id,
|
||
function_name="project_from_text_analysis",
|
||
)
|
||
try:
|
||
response = response.strip()
|
||
if response.startswith("```"):
|
||
response = response.split("```")[1]
|
||
if response.startswith("json"):
|
||
response = response[4:]
|
||
analysis = json.loads(response.strip())
|
||
except Exception:
|
||
analysis = {
|
||
"project_name": input_text[:20],
|
||
"project_title": input_text[:50],
|
||
"tech_stack": ["GPT-4o-mini"],
|
||
"target_audience": "通用",
|
||
"timeline": "3周",
|
||
"primary_role": "ai_engineer",
|
||
"domain_id": 1,
|
||
"key_features": [],
|
||
"competitors": [],
|
||
"success_metrics": [],
|
||
}
|
||
return analysis
|
||
|
||
async def _generate_competitive_signal_draft(
|
||
self,
|
||
*,
|
||
analysis: Dict[str, Any],
|
||
tenant_id: int,
|
||
) -> Dict[str, Any]:
|
||
competitors = analysis.get("competitors", [])
|
||
if not competitors:
|
||
competitors = ["行业同类产品"]
|
||
|
||
prompt = f"""基于以下项目信息，生成竞品分析信号：
|
||
|
||
项目：{analysis.get('project_title')}
|
||
技术栈：{', '.join(analysis.get('tech_stack', []))}
|
||
竞品：{', '.join(competitors)}
|
||
|
||
请生成：
|
||
1. 竞品现状分析
|
||
2. 差异化机会
|
||
3. 技术风险评估
|
||
|
||
以 JSON 格式返回：
|
||
{{
|
||
"content": "信号内容（200字以内）",
|
||
"impact_score": 0.9,
|
||
"priority": "high",
|
||
"strategic_analysis": {{"is_strategic": true, "threat_level": "MEDIUM"}}
|
||
}}"""
|
||
|
||
response = await self._generate_project_creation_text(
|
||
prompt=prompt,
|
||
tenant_id=tenant_id,
|
||
function_name="project_from_text_signal",
|
||
)
|
||
try:
|
||
response = response.strip()
|
||
if response.startswith("```"):
|
||
response = response.split("```")[1]
|
||
if response.startswith("json"):
|
||
response = response[4:]
|
||
return json.loads(response.strip())
|
||
except Exception:
|
||
return {
|
||
"content": f"竞品分析：{', '.join(competitors)} 的技术现状和差异化机会",
|
||
"impact_score": 0.8,
|
||
"priority": "high",
|
||
"strategic_analysis": {"is_strategic": True},
|
||
}
|
||
|
||
async def _generate_charter_draft(
|
||
self,
|
||
*,
|
||
analysis: Dict[str, Any],
|
||
tenant_id: int,
|
||
) -> Dict[str, Any]:
|
||
prompt = f"""基于以下项目信息，生成项目宪章：
|
||
|
||
项目：{analysis.get('project_title')}
|
||
描述：{analysis.get('project_title')}
|
||
技术栈：{', '.join(analysis.get('tech_stack', []))}
|
||
核心功能：{', '.join(analysis.get('key_features', []))}
|
||
成功指标：{', '.join(analysis.get('success_metrics', []))}
|
||
|
||
请生成包含以下内容的宪章：
|
||
1. 项目目标（明确、可衡量）
|
||
2. 成功标准（具体指标）
|
||
3. 技术约束
|
||
4. 时间节点
|
||
|
||
以 JSON 格式返回：
|
||
{{
|
||
"title": "宪章标题",
|
||
"content": "宪章完整内容（包含上述4个方面）",
|
||
"success_criteria": ["标准1", "标准2"],
|
||
"constraints": ["约束1", "约束2"]
|
||
}}"""
|
||
|
||
response = await self._generate_project_creation_text(
|
||
prompt=prompt,
|
||
tenant_id=tenant_id,
|
||
function_name="project_from_text_charter",
|
||
)
|
||
try:
|
||
response = response.strip()
|
||
if response.startswith("```"):
|
||
response = response.split("```")[1]
|
||
if response.startswith("json"):
|
||
response = response[4:]
|
||
charter_data = json.loads(response.strip())
|
||
except Exception:
|
||
charter_data = {
|
||
"title": f"{analysis.get('project_name')}项目宪章",
|
||
"content": f"项目目标：{analysis.get('project_title')}\n成功标准：按时上线",
|
||
"success_criteria": ["按时交付", "功能完整"],
|
||
"constraints": ["时间限制"],
|
||
}
|
||
|
||
raw_content = charter_data.get("content", "")
|
||
if isinstance(raw_content, dict):
|
||
content = dict(raw_content)
|
||
else:
|
||
content = {
|
||
"rationale": str(raw_content),
|
||
"success_criteria": charter_data.get("success_criteria", []),
|
||
"constraints": charter_data.get("constraints", []),
|
||
"signal_ids": [],
|
||
}
|
||
return {
|
||
"title": charter_data.get("title", "项目宪章"),
|
||
"content": content,
|
||
}
|
||
|
||
async def _generate_task_chain_draft(
|
||
self,
|
||
*,
|
||
analysis: Dict[str, Any],
|
||
tenant_id: int,
|
||
) -> List[Dict[str, Any]]:
|
||
prompt = f"""基于以下信息，生成完整的任务链：
|
||
|
||
项目：{analysis.get('project_title')}
|
||
技术栈：{', '.join(analysis.get('tech_stack', []))}
|
||
时间周期：{analysis.get('timeline', '3周')}
|
||
|
||
请生成：
|
||
1. 任务列表（按优先级排序）
|
||
2. 每个任务包含：名称、类型、预估工时、依赖关系、技能要求
|
||
|
||
以 JSON 格式返回：
|
||
{{
|
||
"tasks": [
|
||
{{
|
||
"title": "任务名称",
|
||
"type": "backend/frontend/ai_engineer",
|
||
"estimated_hours": 8,
|
||
"description": "任务描述",
|
||
"priority": "high/medium/low",
|
||
"required_skills": ["技能1", "技能2"]
|
||
}}
|
||
]
|
||
}}"""
|
||
|
||
response = await self._generate_project_creation_text(
|
||
prompt=prompt,
|
||
tenant_id=tenant_id,
|
||
function_name="project_from_text_task_chain",
|
||
)
|
||
try:
|
||
response = response.strip()
|
||
if response.startswith("```"):
|
||
response = response.split("```")[1]
|
||
if response.startswith("json"):
|
||
response = response[4:]
|
||
tasks_data = json.loads(response.strip())
|
||
tasks_list = tasks_data.get("tasks", [])
|
||
if isinstance(tasks_list, list) and tasks_list:
|
||
return tasks_list
|
||
except Exception:
|
||
pass
|
||
return [
|
||
{
|
||
"title": "技术方案设计",
|
||
"type": analysis.get("primary_role", "backend"),
|
||
"estimated_hours": 16,
|
||
"description": "完成项目技术方案设计",
|
||
"priority": "high",
|
||
"required_skills": ["系统设计"],
|
||
},
|
||
{
|
||
"title": "核心功能开发",
|
||
"type": analysis.get("primary_role", "backend"),
|
||
"estimated_hours": 40,
|
||
"description": "完成核心功能开发",
|
||
"priority": "high",
|
||
"required_skills": ["开发"],
|
||
},
|
||
]
|
||
|
||
@staticmethod
|
||
def _build_risk_alert_drafts(*, analysis: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||
estimated_hours = analysis.get("estimated_hours", 40)
|
||
return [
|
||
{
|
||
"type": "schedule_delay",
|
||
"description": "任务超时预警",
|
||
"threshold_hours": estimated_hours * 1.2,
|
||
"severity": "medium",
|
||
},
|
||
{
|
||
"type": "quality_drop",
|
||
"description": "质量下降预警",
|
||
"threshold": 0.85,
|
||
"severity": "high",
|
||
},
|
||
{
|
||
"type": "skill_mismatch",
|
||
"description": "技能匹配度不足",
|
||
"threshold": 0.6,
|
||
"severity": "medium",
|
||
},
|
||
]
|
||
|
||
|
||
def get_ai_steward_service(db: AsyncSession) -> AIStewardService:
|
||
"""获取 AI Steward 服务实例。"""
|
||
return AIStewardService(db)
|