# -*- coding: utf-8 -*- """ V2 模板市场 - 三类模板：task（任务）/ project（项目+任务链）/ team（Agent 团队角色配置） - 内置模板种子 + 用户自建 + 使用计数 - 应用模板：task → 直接创建任务；project → 创建项目及任务；team → 一键注册 Worker 团队 """ import json import db BUILTIN_TEMPLATES = [ # ---------------- 任务模板 ---------------- { 'type': 'task', 'name': '代码审查', 'description': '对给定代码进行质量审查，输出问题清单与修改建议', 'tags': ['代码', '质检'], 'content': { 'title': '代码审查：{target}', 'description': '请对以下代码进行严格审查：\n{code}\n\n输出格式：\n1. 问题清单（严重级别：🔴严重/🟡一般/🟢建议）\n2. 每个问题给出修改建议\n3. 总体评价（3句话）', 'priority': 'high', 'review_required': True, }, }, { 'type': 'task', 'name': '技术方案设计', 'description': '根据需求产出技术方案文档（架构/选型/实施步骤/风险）', 'tags': ['方案', '架构'], 'content': { 'title': '技术方案：{target}', 'description': '基于以下需求输出技术方案：\n{requirement}\n\n输出格式：\n1. 需求理解与目标\n2. 架构设计（含图表文字描述）\n3. 技术选型及理由\n4. 实施步骤（含里程碑）\n5. 风险与应对', 'priority': 'medium', 'review_required': True, }, }, { 'type': 'task', 'name': '周报生成', 'description': '根据工作记录生成结构化周报', 'tags': ['文档', '办公'], 'content': { 'title': '周报：{target}', 'description': '根据以下工作记录生成周报：\n{records}\n\n输出格式：\n## 本周完成\n## 本周进展\n## 风险与问题\n## 下周计划\n语言简洁、量化优先。', 'priority': 'low', 'review_required': False, }, }, # ---------------- 项目模板 ---------------- { 'type': 'project', 'name': 'AI 应用 MVP 启动', 'description': '从需求到可演示 MVP 的标准项目流程（6 个任务 DAG）', 'tags': ['AI', 'MVP'], 'content': { 'name': '{name} MVP', 'objective': '完成 {name} 的最小可行产品（MVP）开发与演示', 'acceptance_criteria': '核心功能可演示、代码入库、有部署说明', 'tasks': [ {'title': '需求分析与用户故事', 'description': '输出需求清单与用户故事（As a...I want...）', 'depends_on': []}, {'title': '技术选型与架构设计', 'description': '给出技术栈选型与系统架构设计（含数据模型）', 'depends_on': [0]}, {'title': '核心功能实现', 'description': '实现核心功能模块，输出关键代码与说明', 'depends_on': [1]}, {'title': 'UI 界面搭建', 'description': '搭建用户界面，与核心功能对接', 'depends_on': [1]}, {'title': '联调与测试', 'description': '端到端联调，输出测试用例与结果', 'depends_on': [2, 3]}, {'title': '部署与演示文档', 'description': '编写部署文档与演示脚本', 'depends_on': [4]}, ], }, }, { 'type': 'project', 'name': '数据分析报告', 'description': '数据驱动的分析报告项目（采集→清洗→分析→可视化→报告）', 'tags': ['数据', '分析'], 'content': { 'name': '{name} 数据分析', 'objective': '对 {name} 相关数据进行全面分析并输出决策建议', 'acceptance_criteria': '报告含数据图表、关键发现、行动建议', 'tasks': [ {'title': '数据采集方案', 'description': '明确数据来源与采集方案', 'depends_on': []}, {'title': '数据清洗与预处理', 'description': '清洗缺失值/异常值，输出清洗说明', 'depends_on': [0]}, {'title': '探索性分析', 'description': '输出核心指标分布与相关性分析', 'depends_on': [1]}, {'title': '可视化图表', 'description': '生成 5 张以上关键可视化图表（文字描述+图表代码）', 'depends_on': [2]}, {'title': '分析报告', 'description': '汇总输出完整分析报告与行动建议', 'depends_on': [3]}, ], }, }, # ---------------- 团队模板 ---------------- { 'type': 'team', 'name': '标准三 Agent 团队', 'description': '执行者 + 评审者 + 主管 的标准协作团队（适配评审/主管模式）', 'tags': ['协作', '标准'], 'content': { 'name': '标准协作团队', 'workers': [ {'name': '执行 Agent', 'description': '负责任务执行与产出', 'system_prompt': '你是高效可靠的执行 Agent。接到任务后直接产出高质量结果，输出结构清晰、内容完整。', 'temperature': 0.6}, {'name': '评审 Agent', 'description': '负责质量评审与返工建议', 'system_prompt': '你是严格的评审专家。客观评估成果质量，指出具体问题并给出可执行的修改建议。', 'temperature': 0.2}, {'name': '主管 Agent', 'description': '负责拆解任务与汇总合成', 'system_prompt': '你是资深主管。善于拆解复杂任务、协调分工、汇总合成高质量交付物。', 'temperature': 0.4}, ], }, }, { 'type': 'team', 'name': '辩论三人组', 'description': '正方 + 反方 + 裁判 的辩论团队（适配辩论模式）', 'tags': ['辩论', '决策'], 'content': { 'name': '辩论三人组', 'workers': [ {'name': '正方辩手', 'description': '支持方立场论证', 'system_prompt': '你是正方辩手，论点鲜明、论据充分，善于用数据和逻辑说服对方。', 'temperature': 0.8}, {'name': '反方辩手', 'description': '反对方立场论证', 'system_prompt': '你是反方辩手，思维犀利，善于发现方案的风险与漏洞并有力反驳。', 'temperature': 0.8}, {'name': '首席裁判', 'description': '辩论裁决与共识输出', 'system_prompt': '你是公正的首席裁判。综合各方观点，输出最合理的最终结论与裁决。', 'temperature': 0.3}, ], }, }, ] def create_builtin_templates(): c = db.q('SELECT COUNT(*) c FROM templates WHERE is_builtin=1')[0]['c'] if c > 0: return for t in BUILTIN_TEMPLATES: db.w( 'INSERT INTO templates (type, name, description, content, tags, author, is_builtin, usage_count, created_at, updated_at) ' 'VALUES (?,?,?,?,?,?,1,0,?,?)', (t['type'], t['name'], t['description'], json.dumps(t['content'], ensure_ascii=False), json.dumps(t['tags']), 'system', db.now(), db.now())) def get_template(tid): return db.q('SELECT * FROM templates WHERE id=?', (tid,), one=True) def bump_usage(tid): db.w('UPDATE templates SET usage_count=usage_count+1, updated_at=? WHERE id=?', (db.now(), tid)) def render(content, variables): """{name}/{target}/{code}… 占位符替换（值做 JSON 转义，防换行/引号破坏结构）""" import json as _json text = _json.dumps(content, ensure_ascii=False) for k, v in (variables or {}).items(): text = text.replace('{' + k + '}', _json.dumps(str(v), ensure_ascii=False)[1:-1]) return _json.loads(text) def apply_task_template(tpl, variables, project_id, worker_id=None): """应用任务模板 → 创建任务""" content = render(json.loads(tpl['content']), variables) tid = db.w( 'INSERT INTO tasks (project_id, worker_id, title, description, priority, review_required, ' 'deadline, depends_on, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?)', (project_id, worker_id, content.get('title', tpl['name']), content.get('description', ''), content.get('priority', 'medium'), 1 if content.get('review_required', True) else 0, '', '[]', db.now(), db.now())) db.w('INSERT INTO task_logs (task_id, level, message, created_at) VALUES (?,?,?,?)', (tid, 'info', f'由模板「{tpl["name"]}」创建', db.now())) bump_usage(tpl['id']) return tid def apply_project_template(tpl, variables, worker_id=None): """应用项目模板 → 创建项目 + 任务链（DAG）""" content = render(json.loads(tpl['content']), variables) pid = db.w( 'INSERT INTO projects (name, description, objective, acceptance_criteria, status, ' 'budget_limit, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?)', (content.get('name', tpl['name']), tpl['description'], content.get('objective', ''), content.get('acceptance_criteria', ''), 'planning', 0, db.now(), db.now())) created = [] for i, t in enumerate(content.get('tasks') or []): dep_ids = [created[idx] for idx in (t.get('depends_on') or []) if isinstance(idx, int) and 0 <= idx < len(created)] tid = db.w( 'INSERT INTO tasks (project_id, worker_id, title, description, priority, review_required, ' 'deadline, depends_on, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?)', (pid, worker_id, t.get('title', f'任务{i+1}'), t.get('description', ''), t.get('priority', 'medium'), 1, '', json.dumps(dep_ids), db.now(), db.now())) created.append(tid) db.w('INSERT INTO task_logs (task_id, level, message, created_at) VALUES (?,?,?,?)', (tid, 'info', f'由项目模板「{tpl["name"]}」创建', db.now())) bump_usage(tpl['id']) return pid, created def apply_team_template(tpl, variables, provider='deepseek', model='deepseek-v4-flash'): """应用团队模板 → 批量注册 Worker""" content = render(json.loads(tpl['content']), variables) created = [] for w in content.get('workers') or []: wid = db.w( 'INSERT INTO workers (name, description, provider, model, base_url, api_key, system_prompt, ' 'temperature, max_tokens, task_cost_limit, monthly_cost_limit, status, created_at, updated_at) ' 'VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)', (w.get('name', 'Agent'), w.get('description', ''), provider, model, '', '', w.get('system_prompt', ''), float(w.get('temperature', 0.7)), 2000, 0, 0, 'enabled', db.now(), db.now())) created.append(wid) bump_usage(tpl['id']) return created