187 lines
6.0 KiB
Python
187 lines
6.0 KiB
Python
from __future__ import annotations
|
||
|
||
from .shared import *
|
||
from ...config.ai_prompts import PROMPT_PROJECT_REQUIREMENT_EXTRACTION
|
||
|
||
class AIStewardDraftGenerationMixin:
|
||
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 = 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. 任务列表（按优先级排序，最多 4 条）
|
||
2. 每个任务包含：名称、类型、预估工时、依赖关系、技能要求
|
||
3. 每个 description 必须控制在 60 字以内，不要换行
|
||
|
||
以 JSON 格式返回：
|
||
{{
|
||
"tasks": [
|
||
{{
|
||
"title": "任务名称",
|
||
"type": "backend/frontend/ai_engineer",
|
||
"estimated_hours": 8,
|
||
"description": "任务描述",
|
||
"priority": "high/medium/low",
|
||
"required_skills": ["技能1", "技能2"]
|
||
}}
|
||
]
|
||
}}
|
||
|
||
只返回 JSON，不要 markdown，不要额外解释。"""
|
||
|
||
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,
|
||
)
|
||
tasks_list = tasks_data.get("tasks", [])
|
||
if isinstance(tasks_list, list) and tasks_list:
|
||
return tasks_list
|
||
raise RuntimeError("AI 管家任务链生成失败：返回 tasks 为空")
|
||
|
||
@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",
|
||
},
|
||
]
|