"""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)