from __future__ import annotations from .shared import * from ...config.ai_prompts import PROMPT_PROJECT_REQUIREMENT_EXTRACTION class AIStewardDraftGenerationMixin: @staticmethod def _task_chain_generation_routing_meta(*, tier_alias: str = "low") -> dict[str, Any]: tier = "high" if str(tier_alias or "").strip().lower() == "high" else "low" return { "tier_alias": tier, "planned_tier": tier, "routing_decision": "force_low_first" if tier == "low" else "explicit_high_retry", "response_contract": "structured", "execution_mode": "local_planner", } @staticmethod def _task_generation_ready_state() -> dict[str, Any]: return { "status": "ready", "error_message": None, } @staticmethod def _task_generation_pending_state(error: Exception | str) -> dict[str, Any]: return { "status": "pending", "error_message": str(error or "").strip()[:240] or "task_generation_pending", } @staticmethod def _normalize_generation_text_list(values: Any, *, max_items: int = 8, max_len: int = 120) -> list[str]: if not isinstance(values, list): return [] normalized: list[str] = [] for item in values: if isinstance(item, dict): text = ( str(item.get("title") or item.get("pain") or item.get("role") or item.get("source_name") or "").strip() ) else: text = str(item or "").strip() if not text: continue normalized.append(text[:max_len]) return normalized[:max_items] @staticmethod def _normalize_requirement_items_for_prompt(items: Any, *, max_items: int = 12) -> list[dict[str, Any]]: if not isinstance(items, list): return [] normalized: list[dict[str, Any]] = [] for item in items: if not isinstance(item, dict): continue title = str(item.get("title") or "").strip() description = str(item.get("description") or "").strip() section = str(item.get("section") or "").strip() if not any((title, description, section)): continue normalized.append( { "section": section[:40], "title": title[:120], "description": description[:200], "priority": str(item.get("priority") or "").strip()[:20], "source_kind": str(item.get("source_kind") or "").strip()[:40], "source_name": str(item.get("source_name") or "").strip()[:120], } ) return normalized[:max_items] @staticmethod def _normalize_team_setup_for_prompt(items: Any, *, max_items: int = 8) -> list[dict[str, str]]: if not isinstance(items, list): return [] normalized: list[dict[str, str]] = [] for item in items: if isinstance(item, dict): role = str(item.get("role") or "").strip() notes = str(item.get("notes") or "").strip() else: role = str(item or "").strip() notes = "" if not role and not notes: continue normalized.append({"role": role[:60], "notes": notes[:160]}) return normalized[:max_items] @staticmethod def _compact_generation_text(value: Any, *, max_len: int = 240) -> str: text = str(value or "").strip() if not text: return "" return re.sub(r"\s+", " ", text)[:max_len] @staticmethod def _bulletize_generation_items(items: list[str]) -> str: cleaned = [str(item or "").strip() for item in items if str(item or "").strip()] if not cleaned: return "" return "\n".join(f"- {item}" for item in cleaned[:8]) @classmethod def _has_six_module_charter_content(cls, content: Any) -> bool: if not isinstance(content, dict): return False modules = content.get("modules") return isinstance(modules, dict) and all(f"module_{index}" in modules for index in range(1, 7)) @classmethod def _build_six_module_charter_content_from_brief(cls, source: dict[str, Any]) -> dict[str, Any]: if cls._has_six_module_charter_content(source): return dict(source) summary = cls._compact_generation_text( source.get("summary") or source.get("positioning_one_liner") or source.get("background") or source.get("scope") ) target_customers = cls._compact_generation_text( source.get("target_customers") or source.get("target_audience") ) core_pain_points = cls._normalize_generation_text_list(source.get("core_pain_points")) mvp_items = cls._normalize_generation_text_list( source.get("mvp_items") if isinstance(source.get("mvp_items"), list) else source.get("key_features") if isinstance(source.get("key_features"), list) else source.get("objectives") if isinstance(source.get("objectives"), list) else source.get("mvp_features") ) success_metrics = cls._normalize_generation_text_list( source.get("success_metrics") if isinstance(source.get("success_metrics"), list) else source.get("success_criteria") if isinstance(source.get("success_criteria"), list) else source.get("validation_metrics", {}).get("product") if isinstance(source.get("validation_metrics"), dict) else [] ) team_setup = cls._normalize_team_setup_for_prompt(source.get("team_setup")) requirement_items = cls._normalize_requirement_items_for_prompt(source.get("requirement_items")) tech_stack = cls._normalize_generation_text_list(source.get("tech_stack")) milestones = cls._normalize_generation_text_list(source.get("milestones")) risks = cls._normalize_generation_text_list(source.get("risks")) missing_info = cls._normalize_generation_text_list(source.get("missing_info")) requirement_titles: list[str] = [] attachment_rows: list[str] = [] for item in requirement_items: title = cls._compact_generation_text(item.get("title"), max_len=120) description = cls._compact_generation_text(item.get("description"), max_len=160) section = cls._compact_generation_text(item.get("section"), max_len=40) source_name = cls._compact_generation_text(item.get("source_name"), max_len=80) source_kind = cls._compact_generation_text(item.get("source_kind"), max_len=40) if title: requirement_titles.append(f"{section} - {title}" if section else title) if description: requirement_titles.append(description) if source_name or source_kind: attachment_rows.append(" / ".join(part for part in (source_kind, source_name) if part)) team_lines = [] for item in team_setup: role = cls._compact_generation_text(item.get("role"), max_len=60) notes = cls._compact_generation_text(item.get("notes"), max_len=160) if role and notes: team_lines.append(f"{role}：{notes}") elif role: team_lines.append(role) problem_parts: list[str] = [] if target_customers and summary: problem_parts.append(f"{target_customers}当前面临：{summary}") elif target_customers: problem_parts.append(target_customers) elif summary: problem_parts.append(summary) if core_pain_points: problem_parts.append(f"核心痛点：{'；'.join(core_pain_points[:3])}") problem_statement = "\n".join(problem_parts) in_scope_items = mvp_items + requirement_titles out_of_scope_items = missing_info alignment_sections = [] if problem_statement: alignment_sections.append(f"问题定义：{problem_statement}") if in_scope_items: alignment_sections.append(f"首期范围：{'；'.join(in_scope_items[:4])}") if success_metrics: alignment_sections.append(f"验收指标：{'；'.join(success_metrics[:4])}") if team_lines: alignment_sections.append(f"团队协同：{'；'.join(team_lines[:4])}") alignment_qa = "\n".join(alignment_sections) dependency_map = cls._bulletize_generation_items(tech_stack) milestones_text = cls._bulletize_generation_items(milestones) if not milestones_text and cls._compact_generation_text(source.get("timeline"), max_len=80): milestones_text = f"- {cls._compact_generation_text(source.get('timeline'), max_len=80)}" risk_items = risks or core_pain_points trigger_rules = success_metrics[:3] if success_metrics else [] search_tags = [] if target_customers: search_tags.append(target_customers) search_tags.extend(mvp_items[:3]) search_tags.extend(tech_stack[:3]) modules = { "module_1": { "problem_statement": problem_statement, "goal_statement": cls._bulletize_generation_items(mvp_items), "kpis": cls._bulletize_generation_items(success_metrics), }, "module_2": { "in_scope": cls._bulletize_generation_items(in_scope_items), "out_of_scope": cls._bulletize_generation_items(out_of_scope_items), "alignment_qa": alignment_qa, }, "module_3": { "primary_model": cls._compact_generation_text( source.get("primary_model") or "；".join(tech_stack[:2]) or summary ), "fallback_model": cls._compact_generation_text(source.get("fallback_model")), "capability_table": cls._compact_generation_text(source.get("capability_table") or summary), "dependency_map": dependency_map, "business_view": cls._compact_generation_text(source.get("business_view") or summary), "engineering_view": cls._compact_generation_text( source.get("engineering_view") or "；".join(tech_stack[:4]) or "；".join(team_lines[:4]) ), }, "module_4": { "milestones": milestones_text, "role_effort_estimates": cls._bulletize_generation_items(team_lines), "capability_owner_assignment": cls._bulletize_generation_items(team_lines), "buffer_policy": cls._compact_generation_text(source.get("buffer_policy") or source.get("timeline")), }, "module_5": { "risk_register": cls._bulletize_generation_items(risk_items), "trigger_rules": cls._bulletize_generation_items(trigger_rules), "execution_rule": cls._compact_generation_text(source.get("scope") or summary), }, "module_6": { "attachment_metadata": cls._bulletize_generation_items(attachment_rows), "attachment_summary": cls._compact_generation_text(source.get("attachment_summary") or summary), "search_tags": cls._bulletize_generation_items(search_tags), "governance_rule": cls._compact_generation_text(source.get("governance_rule")), }, } required_fields = { "module_1": ("problem_statement", "goal_statement", "kpis"), "module_2": ("in_scope", "out_of_scope", "alignment_qa"), "module_3": ("primary_model", "fallback_model", "capability_table", "dependency_map"), "module_4": ("milestones", "role_effort_estimates", "capability_owner_assignment"), "module_5": ("risk_register", "trigger_rules", "execution_rule"), "module_6": ("attachment_metadata", "attachment_summary", "search_tags"), } field_status: dict[str, dict[str, str]] = {} field_evidence: dict[str, dict[str, list[dict[str, Any]]]] = {} blocking_fields: list[str] = [] blocking_reasons: list[str] = [] for module_key, fields in required_fields.items(): field_status[module_key] = {} field_evidence[module_key] = {} for field_key in fields: value = str(modules.get(module_key, {}).get(field_key) or "").strip() status = "valid" if value else "evidence_insufficient" field_status[module_key][field_key] = status field_evidence[module_key][field_key] = [] if status != "valid": blocking_fields.append(f"{module_key}.{field_key}") blocking_reasons.append(f"{module_key}.{field_key} 缺少可验证内容") return { "schema_version": "charter_ai_first_v1", "auto_generated_by_ai": True, "metadata": { "status": "draft", "confidence_score": 0.8 if problem_statement and in_scope_items else 0.6, }, "modules": modules, "field_status": field_status, "field_evidence": field_evidence, "readiness": { "ready_for_confirm": not blocking_fields, "blocking_fields": blocking_fields, "blocking_reasons": blocking_reasons, }, "derived_artifacts": { "task_seeds": [ { "title": item, "description": item, "evidence_refs": ["module_2.in_scope"], } for item in in_scope_items[:4] ], "verification_metrics": [ { "name": item, "expected": item, "source_field": "module_1.kpis", } for item in success_metrics[:4] ], "risk_gates": [ { "name": item, "action": modules["module_5"]["execution_rule"], "source_fields": ["module_5.trigger_rules"], } for item in trigger_rules[:3] ], }, "summary": summary, "objectives": mvp_items, "scope": alignment_qa, "in_scope": in_scope_items, "out_of_scope": out_of_scope_items, "success_criteria": success_metrics, "constraints": missing_info, "milestones": milestones, "risks": risk_items, "signal_ids": [], } async def _load_available_skill_names(self, *, tenant_id: int, limit: int = 24) -> list[str]: from sqlalchemy import select from ...models import SkillTag rows = ( await self.db.execute( select(SkillTag) .where(SkillTag.tenant_id == int(tenant_id)) .order_by(SkillTag.sort_order.asc(), SkillTag.id.asc()) ) ).scalars().all() names: list[str] = [] seen: set[str] = set() for row in rows: name = str(row.name or "").strip() lowered = name.lower() if not name or lowered in seen: continue seen.add(lowered) names.append(name[:80]) if len(names) >= limit: break return names def _build_task_chain_prompt( self, *, analysis: Dict[str, Any], available_skill_names: list[str], retry_feedback: str | None = None, ) -> str: context_payload = { "project_title": str(analysis.get("project_title") or analysis.get("project_name") or "").strip(), "target_audience": str( analysis.get("target_audience") or analysis.get("target_customers") or "" ).strip(), "timeline": str(analysis.get("timeline") or "3周").strip() or "3周", "summary": str(analysis.get("summary") or analysis.get("positioning_one_liner") or "").strip(), "key_features": self._normalize_generation_text_list( analysis.get("key_features") if isinstance(analysis.get("key_features"), list) else analysis.get("mvp_features") if isinstance(analysis.get("mvp_features"), list) else analysis.get("mvp_items"), ), "success_metrics": self._normalize_generation_text_list(analysis.get("success_metrics")), "core_pain_points": self._normalize_generation_text_list(analysis.get("core_pain_points")), "team_setup": self._normalize_team_setup_for_prompt(analysis.get("team_setup")), "requirement_items": self._normalize_requirement_items_for_prompt(analysis.get("requirement_items")), "available_skill_tags": list(available_skill_names or []), } retry_section = "" if retry_feedback: retry_section = f""" 上轮结果存在问题，请完整重生任务链，不要沿用错误任务： {retry_feedback} """.rstrip() return f"""基于以下项目信息，生成完整的任务链： 项目信息 JSON： {json.dumps(context_payload, ensure_ascii=False, indent=2)} 任务规划要求： 1. 任务必须直接支撑 key_features、requirement_items、success_metrics，优先围绕 MVP 主链与数据闭环。 2. 若 requirement_items 已给出 Shopify、TikTok Shop、飞书群聊等来源，任务必须体现对应的数据接入、分析或可视化工作。 3. 除非 requirement_items 明确要求，禁止输出登录、权限、RBAC、后台管理、通用 CMS、CI/CD 等泛化基础设施任务。 4. required_skills 只能从 available_skill_tags 中选择，且每个任务至少包含 1 个可映射技能。 5. 优先生成当前团队能立即开干的任务，避免虚构 Node.js、JWT、Database Design、API Integration、ECharts 等不在技能池内的技能。 6. 任务数量控制在 2-4 条，按优先级排序；每个 description 控制在 60 字以内，不要换行。 7. 任务标题要具体到业务动作，不要写成“系统优化”“功能完善”这类空泛表述。 {retry_section} 以 JSON 格式返回： {{ "tasks": [ {{ "title": "任务名称", "type": "backend|frontend|ai_engineer|pm", "estimated_hours": 8, "description": "任务描述", "priority": "high|medium|low", "dependencies": ["依赖任务名称"], "required_skills": ["Python", "FastAPI"] }} ] }} 只返回 JSON，不要 markdown，不要额外解释。""" def _build_task_chain_charter_content(self, *, analysis: Dict[str, Any]) -> str: charter_payload = { "project_title": str(analysis.get("project_title") or analysis.get("project_name") or "").strip(), "summary": str(analysis.get("summary") or analysis.get("positioning_one_liner") or "").strip(), "target_audience": str( analysis.get("target_audience") or analysis.get("target_customers") or "" ).strip(), "objectives": self._normalize_generation_text_list( analysis.get("key_features") if isinstance(analysis.get("key_features"), list) else analysis.get("mvp_features") if isinstance(analysis.get("mvp_features"), list) else analysis.get("mvp_items"), ), "success_criteria": self._normalize_generation_text_list(analysis.get("success_metrics")), "core_pain_points": self._normalize_generation_text_list(analysis.get("core_pain_points")), "team_setup": self._normalize_team_setup_for_prompt(analysis.get("team_setup")), "requirement_items": self._normalize_requirement_items_for_prompt(analysis.get("requirement_items")), } return json.dumps(charter_payload, ensure_ascii=False) def _coerce_task_chain_charter_content( self, *, analysis: Dict[str, Any], charter_content: Any = None, ) -> str: if self._has_six_module_charter_content(charter_content): return json.dumps(charter_content, ensure_ascii=False) if isinstance(charter_content, dict) and charter_content: normalized = self._build_six_module_charter_content_from_brief(charter_content) return json.dumps(normalized, ensure_ascii=False) return self._build_task_chain_charter_content(analysis=analysis) @classmethod def _split_charter_text_items( cls, value: Any, *, max_items: int = 8, max_len: int = 120, ) -> list[str]: if isinstance(value, list): return cls._normalize_generation_text_list(value, max_items=max_items, max_len=max_len) text = str(value or "").strip() if not text: return [] items: list[str] = [] seen: set[str] = set() raw_blocks = re.split(r"[\r\n]+", text) for raw_block in raw_blocks: block = re.sub(r"^[\-\*\u2022\d\.\)\s、]+", "", str(raw_block or "").strip()).strip() if not block: continue candidate_parts = [block] if "\n" not in block and any(sep in block for sep in ("；", ";")): split_parts = [part.strip() for part in re.split(r"[；;]+", block) if part.strip()] if len(split_parts) > 1: candidate_parts = split_parts for part in candidate_parts: normalized = re.sub(r"\s+", " ", str(part or "").strip())[:max_len] key = normalized.lower() if not normalized or key in seen: continue seen.add(key) items.append(normalized) if len(items) >= max_items: return items return items @staticmethod def _normalize_skill_match_key(value: Any) -> str: return re.sub(r"[\s_\-/]+", "", str(value or "").strip()).lower() @classmethod def _match_available_skills_by_hints( cls, *, available_skill_names: list[str], hints: list[str], max_items: int = 3, ) -> list[str]: normalized_available = [ (str(name).strip(), cls._normalize_skill_match_key(name)) for name in (available_skill_names or []) if str(name).strip() ] matched: list[str] = [] seen: set[str] = set() for hint in hints: hint_key = cls._normalize_skill_match_key(hint) if not hint_key: continue for skill_name, skill_key in normalized_available: if not skill_key or skill_key in seen: continue if hint_key in skill_key or skill_key in hint_key: seen.add(skill_key) matched.append(skill_name) if len(matched) >= max_items: return matched break return matched @staticmethod def _infer_local_task_type(*, title: str, description: str) -> str: title_lower = str(title or "").lower() merged = f"{title} {description}".lower() if any(token in title_lower for token in ("看板", "页面", "前端", "可视化", "dashboard", "ui", "面板", "展示")): return "frontend" if any(token in title_lower for token in ("需求澄清", "范围对齐", "验收", "排期", "评审")): return "pm" if any(token in merged for token in ("prompt", "rag", "embedding", "向量", "模型", "智能体", "llm")): return "ai_engineer" return "backend" @classmethod def _infer_local_task_required_skills( cls, *, title: str, description: str, task_type: str, available_skill_names: list[str], charter_content: dict[str, Any], ) -> list[str]: modules = charter_content.get("modules") if isinstance(charter_content.get("modules"), dict) else {} module_3 = modules.get("module_3") if isinstance(modules.get("module_3"), dict) else {} engineering_context = " ".join( str(module_3.get(field) or "").strip() for field in ("primary_model", "fallback_model", "capability_table", "dependency_map", "engineering_view") ) merged = f"{title} {description} {engineering_context}" direct_matches = cls._match_available_skills_by_hints( available_skill_names=available_skill_names, hints=[merged], max_items=3, ) role_hints_map = { "frontend": ["React", "TypeScript", "前端", "可视化"], "pm": ["项目管理", "产品管理", "需求分析", "产品"], "ai_engineer": ["LangChain", "LLM 调优", "LLM调优", "RAG构建", "Python", "数据分析"], "backend": ["Python", "FastAPI", "SQLAlchemy", "数据分析"], } role_matches = cls._match_available_skills_by_hints( available_skill_names=available_skill_names, hints=role_hints_map.get(task_type, role_hints_map["backend"]), max_items=3, ) selected: list[str] = [] seen: set[str] = set() for skill_name in role_matches + direct_matches: normalized = cls._normalize_skill_match_key(skill_name) if not normalized or normalized in seen: continue seen.add(normalized) selected.append(skill_name) if len(selected) >= 3: return selected if not selected and task_type == "ai_engineer": selected = cls._match_available_skills_by_hints( available_skill_names=available_skill_names, hints=role_hints_map["backend"], max_items=3, ) if not selected and available_skill_names: selected = [str(name).strip() for name in available_skill_names[:3] if str(name).strip()] return selected[:3] @staticmethod def _normalize_decomposed_task_rows(tasks: Any) -> list[dict[str, Any]]: if not isinstance(tasks, list): return [] normalized: list[dict[str, Any]] = [] for item in tasks: if not isinstance(item, dict): continue title = str(item.get("title") or "").strip() if not title: continue normalized.append( { "title": title, "type": str(item.get("type") or "backend").strip() or "backend", "estimated_hours": item.get("estimated_hours"), "description": str(item.get("description") or "").strip(), "priority": str(item.get("priority") or "medium").strip() or "medium", "dependencies": item.get("depends_on_task_keys") if isinstance(item.get("depends_on_task_keys"), list) else [], "required_skills": normalize_skill_names( item.get("required_skill_names") if isinstance(item.get("required_skill_names"), list) else item.get("required_skills") ), } ) return normalized async def _generate_task_chain_via_charter_decomposition( self, *, analysis: Dict[str, Any], tenant_id: int, charter_content: Any = None, routing_meta: dict[str, Any] | None = None, available_skill_names: list[str] | None = None, ) -> list[dict[str, Any]]: from ..ai_service_core.service import AIService charter_content_text = self._coerce_task_chain_charter_content( analysis=analysis, charter_content=charter_content, ) try: payload = await AIService().decompose_tasks_from_charter( charter_content=charter_content_text, project_id=0, tenant_id=int(tenant_id), db=self.db, routing_meta=routing_meta or self._task_chain_generation_routing_meta(), ) tasks = self._normalize_decomposed_task_rows(payload.get("tasks")) if not tasks: raise RuntimeError("AI 管家任务链生成失败：charter decomposition returned empty tasks") report = await self._inspect_task_chain_candidate(tasks=tasks, tenant_id=int(tenant_id)) requires_skill_gate = bool(available_skill_names) if not requires_skill_gate: return tasks if not report["invalid_tasks"] and int(report.get("mappable_task_count") or 0) > 0: return tasks raise RuntimeError("AI 管家任务链生成失败：charter decomposition returned low-quality tasks") except Exception as exc: logger.warning( "ai_steward_charter_task_chain_decomposition_failed", tenant_id=int(tenant_id), project_title=str(analysis.get("project_title") or analysis.get("project_name") or "").strip()[:120], tier_alias=str((routing_meta or {}).get("tier_alias") or "").strip() or "default", error=str(exc), ) raise async def _inspect_task_chain_candidate( self, *, tasks: list[dict[str, Any]], tenant_id: int, ) -> dict[str, Any]: invalid_tasks: list[dict[str, Any]] = [] mappable_count = 0 for task in tasks: raw_required_skills = normalize_skill_names(task.get("required_skills")) if not raw_required_skills: invalid_tasks.append( { "title": str(task.get("title") or "").strip(), "required_skills": [], "reason": "required_skills_empty", } ) continue resolution = await resolve_skill_tag_ids_by_names( self.db, tenant_id=int(tenant_id), skill_names=raw_required_skills, ) if resolution.matched_ids: mappable_count += 1 continue invalid_tasks.append( { "title": str(task.get("title") or "").strip(), "required_skills": raw_required_skills, "reason": "required_skills_unmappable", } ) return { "tasks": tasks, "invalid_tasks": invalid_tasks, "mappable_task_count": mappable_count, "total_task_count": len(tasks), } @staticmethod def _build_task_chain_retry_feedback( *, invalid_tasks: list[dict[str, Any]], available_skill_names: list[str], ) -> str: lines: list[str] = [] for item in invalid_tasks[:6]: title = str(item.get("title") or "未命名任务").strip() or "未命名任务" skills = ", ".join(str(skill).strip() for skill in (item.get("required_skills") or []) if str(skill).strip()) reason = str(item.get("reason") or "").strip() if skills: lines.append(f"- {title}：技能 {skills} 无法映射到当前技能池（{reason}）") else: lines.append(f"- {title}：缺少可映射技能（{reason}）") if available_skill_names: lines.append(f"- 当前可用技能池：{', '.join(available_skill_names)}") return "\n".join(lines) @staticmethod def _task_chain_candidate_score(report: dict[str, Any]) -> tuple[int, int, int]: return ( int(report.get("mappable_task_count") or 0), -len(report.get("invalid_tasks") or []), int(report.get("total_task_count") or 0), ) @staticmethod def _task_chain_candidate_is_acceptable( report: dict[str, Any], *, available_skill_names: list[str], ) -> bool: if not available_skill_names: return True return not report.get("invalid_tasks") and int(report.get("mappable_task_count") or 0) > 0 async def _generate_task_chain_from_primary_prompt( self, *, analysis: Dict[str, Any], tenant_id: int, available_skill_names: list[str], routing_meta: dict[str, Any], ) -> list[dict[str, Any]]: prompt = self._build_task_chain_prompt( analysis=analysis, available_skill_names=available_skill_names, ) tasks_data = await self._generate_project_creation_object( prompt=prompt, tenant_id=tenant_id, function_name="project_from_text_task_chain", stage_label="AI 管家任务链生成失败", timeout_seconds=PROJECT_CREATION_TASK_CHAIN_TIMEOUT_SECONDS, routing_meta=routing_meta, ) tasks_list = tasks_data.get("tasks", []) if not isinstance(tasks_list, list) or not tasks_list: raise RuntimeError("AI 管家任务链生成失败：返回 tasks 为空") report = await self._inspect_task_chain_candidate(tasks=tasks_list, tenant_id=int(tenant_id)) best_tasks = tasks_list best_report = report if self._task_chain_candidate_is_acceptable(report, available_skill_names=available_skill_names): return best_tasks retry_feedback = self._build_task_chain_retry_feedback( invalid_tasks=report["invalid_tasks"], available_skill_names=available_skill_names, ) retry_prompt = self._build_task_chain_prompt( analysis=analysis, available_skill_names=available_skill_names, retry_feedback=retry_feedback, ) retry_data = await self._generate_project_creation_object( prompt=retry_prompt, tenant_id=tenant_id, function_name="project_from_text_task_chain", stage_label="AI 管家任务链生成失败", timeout_seconds=PROJECT_CREATION_TASK_CHAIN_TIMEOUT_SECONDS, routing_meta=routing_meta, ) retry_tasks = retry_data.get("tasks", []) if isinstance(retry_tasks, list) and retry_tasks: retry_report = await self._inspect_task_chain_candidate(tasks=retry_tasks, tenant_id=int(tenant_id)) if self._task_chain_candidate_score(retry_report) >= self._task_chain_candidate_score(best_report): best_tasks = retry_tasks best_report = retry_report if self._task_chain_candidate_is_acceptable(retry_report, available_skill_names=available_skill_names): return retry_tasks raise RuntimeError("AI 管家任务链生成失败：prompt route returned low-quality tasks") async def _generate_task_chain_bundle( self, *, analysis: Dict[str, Any], tenant_id: int, charter_content: Any = None, ) -> dict[str, Any]: try: tasks = await self._generate_task_chain_draft( analysis=analysis, tenant_id=int(tenant_id), charter_content=charter_content, ) except Exception as exc: logger.warning( "ai_steward_task_chain_generation_deferred", tenant_id=int(tenant_id), project_title=str(analysis.get("project_title") or analysis.get("project_name") or "").strip()[:120], error=str(exc), ) return { "tasks": [], "task_generation": self._task_generation_pending_state(exc), } return { "tasks": tasks if isinstance(tasks, list) else [], "task_generation": self._task_generation_ready_state(), } async def _analyze_project_input(self, input_text: str, tenant_id: int) -> Dict[str, Any]: """使用 AI 分析项目输入，提取关键信息。""" prompt = PROMPT_PROJECT_REQUIREMENT_EXTRACTION.format( source_label="项目描述", source_context="AI Steward 零配置项目创建输入", input_text=input_text[:2000], ) analysis = await self._generate_project_creation_object( prompt=prompt, tenant_id=tenant_id, function_name="project_requirement_extraction", stage_label="AI 管家项目需求分析失败", ) 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"}} }} 只返回 JSON，不要 markdown，不要额外解释。""" payload = await self._generate_project_creation_object( prompt=prompt, tenant_id=tenant_id, function_name="project_from_text_signal", stage_label="AI 管家竞品信号生成失败", ) return payload 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": {{ "objectives": ["目标1", "目标2"], "success_criteria": ["标准1", "标准2"], "constraints": ["约束1", "约束2"], "milestones": ["里程碑1", "里程碑2"] }}, "success_criteria": ["标准1", "标准2"], "constraints": ["约束1", "约束2"] }} 只返回 JSON，不要 markdown，不要额外解释。""" charter_data = await self._generate_project_creation_object( prompt=prompt, tenant_id=tenant_id, function_name="project_from_text_charter", stage_label="AI 管家项目宪章生成失败", ) raw_content = charter_data.get("content", "") if isinstance(raw_content, dict): content_seed = {**dict(raw_content), **charter_data} else: content_seed = { "summary": charter_data.get("summary") or str(raw_content), "success_criteria": charter_data.get("success_criteria", []), "constraints": charter_data.get("constraints", []), "milestones": charter_data.get("milestones", []), "objectives": raw_content if isinstance(raw_content, list) else analysis.get("key_features", []), "target_customers": analysis.get("target_audience") or analysis.get("target_customers"), "core_pain_points": analysis.get("core_pain_points"), "mvp_items": analysis.get("key_features") or analysis.get("mvp_items"), "team_setup": analysis.get("team_setup"), "validation_metrics": analysis.get("validation_metrics"), "requirement_items": analysis.get("requirement_items"), "timeline": analysis.get("timeline"), "tech_stack": analysis.get("tech_stack"), } content = self._build_six_module_charter_content_from_brief(content_seed) return { "title": charter_data.get("title", "项目宪章"), "content": content, } async def _generate_task_chain_draft( self, *, analysis: Dict[str, Any], tenant_id: int, charter_content: Any = None, ) -> List[Dict[str, Any]]: available_skill_names = await self._load_available_skill_names(tenant_id=int(tenant_id)) if self._has_six_module_charter_content(charter_content): last_error: Exception | None = None for tier_alias in ("low", "high"): try: return await self._generate_task_chain_via_charter_decomposition( analysis=analysis, tenant_id=int(tenant_id), charter_content=charter_content, routing_meta=self._task_chain_generation_routing_meta(tier_alias=tier_alias), available_skill_names=available_skill_names, ) except Exception as exc: last_error = exc logger.warning( "ai_steward_task_chain_charter_tier_attempt_failed", tenant_id=int(tenant_id), project_title=str(analysis.get("project_title") or analysis.get("project_name") or "").strip()[:120], tier_alias=tier_alias, error=str(exc), ) if last_error is not None: raise last_error raise RuntimeError("AI 管家任务链生成失败：charter decomposition unavailable") last_error: Exception | None = None for tier_alias in ("low", "high"): try: return await self._generate_task_chain_from_primary_prompt( analysis=analysis, tenant_id=int(tenant_id), available_skill_names=available_skill_names, routing_meta=self._task_chain_generation_routing_meta(tier_alias=tier_alias), ) except Exception as exc: last_error = exc logger.warning( "ai_steward_task_chain_prompt_tier_attempt_failed", tenant_id=int(tenant_id), project_title=str(analysis.get("project_title") or analysis.get("project_name") or "").strip()[:120], tier_alias=tier_alias, error=str(exc), ) for tier_alias in ("low", "high"): try: return await self._generate_task_chain_via_charter_decomposition( analysis=analysis, tenant_id=int(tenant_id), charter_content=charter_content, routing_meta=self._task_chain_generation_routing_meta(tier_alias=tier_alias), available_skill_names=available_skill_names, ) except Exception as exc: last_error = exc logger.warning( "ai_steward_task_chain_charter_fallback_attempt_failed", tenant_id=int(tenant_id), project_title=str(analysis.get("project_title") or analysis.get("project_name") or "").strip()[:120], tier_alias=tier_alias, error=str(exc), ) if last_error is not None: raise last_error raise RuntimeError("AI 管家任务链生成失败：no task generation route available") @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", }, ]