diff --git a/backend/app/api/teams.py b/backend/app/api/teams.py
new file mode 100644
index 0000000..a97452e
--- /dev/null
+++ b/backend/app/api/teams.py
@@ -0,0 +1,299 @@
+"""
+Team API — 虚拟团队管理 + 项目执行
+"""
+from fastapi import APIRouter, Depends, Query, HTTPException
+from fastapi.responses import StreamingResponse
+from sqlalchemy.orm import Session
+from pydantic import BaseModel, Field
+from typing import List, Optional, Dict, Any
+import json
+import logging
+
+from app.core.database import get_db
+from app.api.auth import get_current_user
+from app.models.user import User
+from app.services.team_service import TeamService, get_preset_roles
+from app.services.team_orchestrator import TeamOrchestrator
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(
+ prefix="/api/v1/teams",
+ tags=["teams"],
+ responses={
+ 401: {"description": "未授权"},
+ 404: {"description": "资源不存在"},
+ 400: {"description": "请求参数错误"},
+ },
+)
+
+
+# ──────────────────────────── Schemas ────────────────────────────
+
+class TeamCreate(BaseModel):
+ name: str
+ description: str = ""
+ workspace_id: Optional[str] = None
+ config: Optional[Dict[str, Any]] = None
+
+
+class TeamUpdate(BaseModel):
+ name: Optional[str] = None
+ description: Optional[str] = None
+ config: Optional[Dict[str, Any]] = None
+ status: Optional[str] = None
+
+
+class MemberAdd(BaseModel):
+ agent_id: str
+ role: str
+ position: int = 0
+ is_lead: bool = False
+
+
+class ProjectExecute(BaseModel):
+ project_description: str = Field(..., min_length=1, description="项目描述")
+ auto_approve_files: bool = Field(default=True, description="是否自动批准文件写入(跳过确认弹窗)")
+
+
+# ──────────────────────────── Endpoints ────────────────────────────
+
+@router.get("/preset-roles")
+def preset_roles():
+ """获取预置角色定义(PM/设计师/开发/测试/DevOps)。"""
+ return {"roles": get_preset_roles()}
+
+
+@router.post("")
+def create_team(
+ body: TeamCreate,
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user),
+):
+ """创建团队。"""
+ svc = TeamService(db)
+ team = svc.create_team(
+ name=body.name,
+ user_id=current_user.id,
+ description=body.description,
+ workspace_id=body.workspace_id,
+ config=body.config,
+ )
+ return {"data": team.to_dict()}
+
+
+@router.post("/template/software-company")
+def create_software_company_template(
+ workspace_id: Optional[str] = Query(None),
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user),
+):
+ """一键创建「软件公司虚拟团队」模板。
+
+ 自动创建 5 个角色 Agent(PM/设计师/开发/测试/DevOps)并组建团队。
+ 若用户已有同名 Agent 则复用。
+ """
+ svc = TeamService(db)
+ result = svc.create_software_company_template(
+ user_id=current_user.id,
+ workspace_id=workspace_id,
+ )
+ return {"data": result}
+
+
+@router.post("/template/education-training")
+def create_education_training_template(
+ workspace_id: Optional[str] = Query(None),
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user),
+):
+ """一键创建「教育培训团队」模板。
+
+ 自动创建 4 个角色 Agent(课程设计/主讲讲师/作业助教/教务管理)并组建团队。
+ 若用户已有同名 Agent 则复用。
+ """
+ svc = TeamService(db)
+ result = svc.create_education_training_template(
+ user_id=current_user.id,
+ workspace_id=workspace_id,
+ )
+ return {"data": result}
+
+
+@router.post("/template/platform-engineering")
+def create_platform_engineering_template(
+ workspace_id: Optional[str] = Query(None),
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user),
+):
+ """一键创建「天工平台工程团队」模板。
+
+ 自动创建 5 个角色 Agent(全栈开发/前端体验/平台运维/质量保障/产品负责人)并组建团队。
+ 专用于天工智能体平台的自我维护与迭代升级。
+ 若用户已有同名 Agent 则复用。
+ """
+ svc = TeamService(db)
+ result = svc.create_platform_engineering_template(
+ user_id=current_user.id,
+ workspace_id=workspace_id,
+ )
+ return {"data": result}
+
+
+@router.get("")
+def list_teams(
+ workspace_id: Optional[str] = Query(None),
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user),
+):
+ """列出当前用户/工作区的所有团队。"""
+ svc = TeamService(db)
+ teams = svc.list_teams(
+ user_id=current_user.id,
+ workspace_id=workspace_id,
+ )
+ return {"data": [t.to_dict() for t in teams]}
+
+
+@router.get("/{team_id}")
+def get_team(
+ team_id: str,
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user),
+):
+ """获取团队详情(含成员列表)。"""
+ svc = TeamService(db)
+ result = svc.get_team_with_members(team_id)
+ if not result:
+ raise HTTPException(status_code=404, detail="团队不存在")
+ return {"data": result}
+
+
+@router.put("/{team_id}")
+def update_team(
+ team_id: str,
+ body: TeamUpdate,
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user),
+):
+ """更新团队信息。"""
+ svc = TeamService(db)
+ team = svc.update_team(team_id, **body.model_dump(exclude_none=True))
+ if not team:
+ raise HTTPException(status_code=404, detail="团队不存在")
+ return {"data": team.to_dict()}
+
+
+@router.delete("/{team_id}")
+def delete_team(
+ team_id: str,
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user),
+):
+ """删除团队及其所有成员。"""
+ svc = TeamService(db)
+ ok = svc.delete_team(team_id)
+ if not ok:
+ raise HTTPException(status_code=404, detail="团队不存在")
+ return {"message": "团队已删除"}
+
+
+@router.post("/{team_id}/members")
+def add_member(
+ team_id: str,
+ body: MemberAdd,
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user),
+):
+ """为团队添加/替换角色成员。"""
+ svc = TeamService(db)
+ member = svc.add_member(
+ team_id=team_id,
+ agent_id=body.agent_id,
+ role=body.role,
+ position=body.position,
+ is_lead=body.is_lead,
+ )
+ if not member:
+ raise HTTPException(status_code=404, detail="团队不存在")
+ return {"data": member.to_dict()}
+
+
+@router.delete("/{team_id}/members/{member_id}")
+def remove_member(
+ team_id: str,
+ member_id: str,
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user),
+):
+ """从团队移除成员。"""
+ svc = TeamService(db)
+ ok = svc.remove_member(team_id, member_id)
+ if not ok:
+ raise HTTPException(status_code=404, detail="成员不存在")
+ return {"message": "成员已移除"}
+
+
+@router.get("/{team_id}/members")
+def list_members(
+ team_id: str,
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user),
+):
+ """列出团队成员。"""
+ svc = TeamService(db)
+ members = svc.get_members(team_id)
+ return {"data": [m.to_dict() for m in members]}
+
+
+@router.post("/{team_id}/execute")
+async def execute_project(
+ team_id: str,
+ body: ProjectExecute,
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user),
+):
+ """执行团队项目(非流式)。
+
+ PM 先分解计划,各角色按阶段顺序执行,QA 最终审查。
+ """
+ orch = TeamOrchestrator(db, team_id, current_user.id)
+ try:
+ result = await orch.execute(body.project_description, auto_approve_files=body.auto_approve_files)
+ return {"data": result}
+ except ValueError as e:
+ raise HTTPException(status_code=400, detail=str(e))
+ except Exception as e:
+ logger.error("团队执行失败: %s", e, exc_info=True)
+ raise HTTPException(status_code=500, detail=f"团队执行失败: {e}")
+
+
+@router.post("/{team_id}/execute/stream")
+async def execute_project_stream(
+ team_id: str,
+ body: ProjectExecute,
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user),
+):
+ """执行团队项目(SSE 流式)。"""
+
+ async def event_generator():
+ orch = TeamOrchestrator(db, team_id, current_user.id)
+ try:
+ async for event in orch.execute_stream(body.project_description, auto_approve_files=body.auto_approve_files):
+ yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
+ except ValueError as e:
+ yield f"data: {json.dumps({'type': 'error', 'content': str(e)}, ensure_ascii=False)}\n\n"
+ except Exception as e:
+ logger.error("团队流式执行失败: %s", e, exc_info=True)
+ yield f"data: {json.dumps({'type': 'error', 'content': str(e)}, ensure_ascii=False)}\n\n"
+
+ return StreamingResponse(
+ event_generator(),
+ media_type="text/event-stream",
+ headers={
+ "Cache-Control": "no-cache",
+ "Connection": "keep-alive",
+ "X-Accel-Buffering": "no",
+ },
+ )
diff --git a/backend/app/services/team_orchestrator.py b/backend/app/services/team_orchestrator.py
new file mode 100644
index 0000000..ac5d181
--- /dev/null
+++ b/backend/app/services/team_orchestrator.py
@@ -0,0 +1,638 @@
+"""
+团队编排引擎 — PM 规划 → 顺序阶段执行 → QA 审查 → 交付物
+
+用法:
+ orchestrator = TeamOrchestrator(db, team_id, user_id)
+ result = await orchestrator.execute("做一个个人博客")
+ async for event in orchestrator.execute_stream("做一个个人博客"):
+ ...
+"""
+from __future__ import annotations
+
+import json
+import re
+import logging
+from pathlib import Path
+from typing import Any, AsyncGenerator, Dict, List, Optional
+from datetime import datetime
+
+from sqlalchemy.orm import Session
+from app.models.team import Team, TeamMember
+from app.models.agent import Agent
+from app.agent_runtime.core import AgentRuntime
+from app.agent_runtime.schemas import (
+ AgentConfig,
+ AgentLLMConfig,
+ AgentToolConfig,
+ AgentMemoryConfig,
+ AgentResult,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def _build_agent_config(agent: Agent, auto_approve_files: bool = True) -> AgentConfig:
+ """从 Agent 的 workflow_config 构建 AgentConfig,供 AgentRuntime 使用。"""
+ wf = agent.workflow_config or {}
+ nodes = wf.get("nodes", [])
+ llm_node = None
+ for n in nodes:
+ if n.get("type") == "llm":
+ llm_node = n
+ break
+
+ llm_data = llm_node.get("data", {}) if llm_node else {}
+
+ tools = llm_data.get("selected_tools") or llm_data.get("tools") or []
+ system_prompt = llm_data.get("prompt") or ""
+ model = llm_data.get("model", "deepseek-v4-pro")
+ provider = llm_data.get("provider", "deepseek")
+ temperature = float(llm_data.get("temperature", 0.4))
+ max_iterations = int(llm_data.get("max_iterations", 10))
+
+ perm = "acceptEdits" if auto_approve_files else "default"
+ logger.info("_build_agent_config: agent=%s tools=%d permission_level=%s", agent.name, len(tools), perm)
+ return AgentConfig(
+ name=agent.name,
+ system_prompt=system_prompt,
+ user_id=agent.user_id,
+ memory_scope_id=f"team_{agent.id}",
+ llm=AgentLLMConfig(
+ provider=provider,
+ model=model,
+ temperature=temperature,
+ max_iterations=max_iterations,
+ ),
+ tools=AgentToolConfig(include_tools=tools, permission_level=perm),
+ memory=AgentMemoryConfig(
+ enabled=True,
+ max_history_messages=30,
+ persist_to_db=True,
+ learning_enabled=True,
+ ),
+ )
+
+
+def _parse_plan_json(raw_output: str) -> Dict[str, Any]:
+ """从 LLM 原始输出中提取 JSON 计划。"""
+ text = raw_output.strip()
+ # 尝试直接解析
+ try:
+ return json.loads(text)
+ except json.JSONDecodeError:
+ pass
+ # 从 ```json ... ``` 中提取
+ import re
+ m = re.search(r'```(?:json)?\s*([\s\S]*?)\s*```', text)
+ if m:
+ try:
+ return json.loads(m.group(1))
+ except json.JSONDecodeError:
+ pass
+ # 从 { ... } 中提取
+ m = re.search(r'\{[\s\S]*"phases"[\s\S]*\}', text)
+ if m:
+ try:
+ return json.loads(m.group())
+ except json.JSONDecodeError:
+ pass
+ logger.warning("无法从输出中解析 JSON 计划,原始输出: %.500s", text)
+ return {}
+
+
+def _resolve_workspace_root() -> Path:
+ """获取文件工具的工作区根目录(与 builtin_tools.py 逻辑一致)。"""
+ from app.core.config import settings
+ raw = (getattr(settings, "LOCAL_FILE_TOOLS_ROOT", None) or "").strip()
+ if raw:
+ return Path(raw).expanduser().resolve()
+ # fallback: 4 级父目录 (builtin_tools.py → app/services → app → backend → repo root)
+ return Path(__file__).resolve().parent.parent.parent.parent
+
+
+def _safe_project_name(description: str) -> str:
+ """从项目描述中提取安全的目录名(英文/拼音/数字/连字符)。"""
+ # 取前 40 个字符,只保留英文、数字、中文转拼音首字母的近似
+ cleaned = description.strip().replace(" ", "-").replace("/", "-").replace("\\", "-")
+ # 去掉非 ASCII 和非基本标点
+ safe = re.sub(r'[^\w\-]', '', cleaned, flags=re.ASCII)
+ if not safe:
+ # 如果全是中文,用时间戳
+ from datetime import datetime
+ safe = f"project-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
+ return safe[:40].strip("-")
+
+
+def _scan_directory_files(directory: Path) -> List[str]:
+ """扫描目录下所有文件的绝对路径列表(递归)。"""
+ files: List[str] = []
+ if not directory.exists():
+ return files
+ for p in directory.rglob("*"):
+ if p.is_file():
+ files.append(str(p))
+ return sorted(files)
+
+
+def _extract_written_files(result: AgentResult) -> List[str]:
+ """从 AgentResult.steps 中提取 file_write 成功产出的绝对路径列表。"""
+ files: List[str] = []
+ seen = set()
+ for step in result.steps:
+ if step.tool_name == "file_write":
+ # 优先从 tool_result JSON 中取绝对路径(仅当写入成功时)
+ if step.tool_result:
+ try:
+ data = json.loads(step.tool_result)
+ # 跳过失败的调用(权限不足 / 需确认)
+ if data.get("error") or data.get("requires_confirmation"):
+ logger.warning("file_write 未成功(%s),跳过文件追踪", data.get("error", "requires_confirmation"))
+ continue
+ fp = data.get("file_path")
+ if fp and fp not in seen:
+ files.append(fp)
+ seen.add(fp)
+ continue
+ except (json.JSONDecodeError, TypeError):
+ pass
+ # 回退:从 tool_input 取相对路径,再拼接工作区根(仅当 tool_result 为空时)
+ if step.tool_input:
+ fp = step.tool_input.get("file_path")
+ if fp:
+ abs_path = str(_resolve_workspace_root() / fp)
+ if abs_path not in seen:
+ files.append(abs_path)
+ seen.add(abs_path)
+ return files
+
+
+class TeamOrchestrator:
+ """虚拟团队编排器 — 按角色分工顺序执行软件项目。"""
+
+ def __init__(
+ self,
+ db: Session,
+ team_id: str,
+ user_id: str,
+ auto_approve_files: bool = True,
+ ):
+ self.db = db
+ self.team_id = team_id
+ self.user_id = user_id
+ self.auto_approve_files = auto_approve_files
+
+ def _load_team(self) -> Team:
+ team = self.db.query(Team).filter(Team.id == self.team_id).first()
+ if not team:
+ raise ValueError(f"团队不存在: {self.team_id}")
+ return team
+
+ def _load_members(self) -> List[TeamMember]:
+ return (
+ self.db.query(TeamMember)
+ .filter(TeamMember.team_id == self.team_id)
+ .order_by(TeamMember.position)
+ .all()
+ )
+
+ def _get_agent_by_role(self, members: List[TeamMember], role: str) -> Optional[Agent]:
+ """按角色查找 Agent。"""
+ for m in members:
+ if m.role == role:
+ return m.agent
+ return None
+
+ def _resolve_planner_role(self, members: List[TeamMember], workflow: str) -> str:
+ """根据团队 workflow 类型解析规划者角色。
+
+ - software_company → pm
+ - education_training → curriculum_designer
+ - platform_engineering → product_lead (产品负责人做规划)
+ - 回退: 第一个 is_lead=True 的成员
+ - 最终回退: 第一个成员
+ """
+ workflow_planner_map = {
+ "software_company": "pm",
+ "education_training": "curriculum_designer",
+ "platform_engineering": "product_lead",
+ }
+ role = workflow_planner_map.get(workflow)
+ if role and self._get_agent_by_role(members, role):
+ return role
+
+ # 回退:找第一个 Leader
+ for m in sorted(members, key=lambda x: x.position):
+ if m.is_lead:
+ return m.role
+
+ # 最终回退:第一个成员
+ if members:
+ return members[0].role
+ return "pm"
+
+ async def execute(self, project_description: str, auto_approve_files: bool = True) -> Dict[str, Any]:
+ """执行团队项目(非流式)。
+
+ Returns:
+ {
+ "project_name": str,
+ "plan": {...},
+ "phases": [{phase, role, agent_name, output, steps, success, error}],
+ "qa_review": {...} | None,
+ "final_deliverable": str,
+ "success": bool,
+ }
+ """
+ self.auto_approve_files = auto_approve_files
+ team = self._load_team()
+ members = self._load_members()
+ if not members:
+ raise ValueError("团队没有成员,请先为角色分配 Agent")
+
+ # ─── 创建项目目录 ───
+ workspace_root = _resolve_workspace_root()
+ safe_name = _safe_project_name(project_description)
+ project_path = workspace_root / "team_projects" / self.team_id / safe_name
+ project_path.mkdir(parents=True, exist_ok=True)
+ logger.info("团队编排 [%s]: 项目目录 %s", self.team_id, project_path)
+
+ prev_files = set(_scan_directory_files(project_path))
+ all_files: List[str] = []
+
+ results: Dict[str, Any] = {
+ "team_id": self.team_id,
+ "team_name": team.name,
+ "project_description": project_description,
+ "project_path": str(project_path),
+ "files": [],
+ "started_at": datetime.now().isoformat(),
+ "plan": None,
+ "phases": [],
+ "qa_review": None,
+ "final_deliverable": "",
+ "success": True,
+ "error": None,
+ }
+
+ # ─── Phase 0: 规划阶段(由 Leader Agent 担任) ───
+ # 选择规划者:优先用 team.config 中指定的 workflow,否则回退兼容
+ workflow = (team.config or {}).get("workflow", "software_company")
+ planner_role = self._resolve_planner_role(members, workflow)
+ planner_agent = self._get_agent_by_role(members, planner_role)
+ if not planner_agent:
+ raise ValueError(
+ f"团队缺少 Leader 角色(需要 role='{planner_role}'),无法分解项目。"
+ f"可用角色: {[m.role for m in members]}"
+ )
+
+ # 动态构建可用的角色列表(来自团队成员实际角色)
+ available_roles = [m.role for m in members]
+ roles_hint = "/".join(available_roles)
+
+ logger.info("团队编排 [%s]: 规划开始 (workflow=%s, planner=%s, role=%s)",
+ self.team_id, workflow, planner_agent.name, planner_role)
+
+ planner_config = _build_agent_config(planner_agent, self.auto_approve_files)
+ planner_runtime = AgentRuntime(planner_config)
+ planner_input = (
+ f"请分析以下需求,按 JSON 格式输出执行计划:\n\n"
+ f"需求描述:{project_description}\n"
+ f"输出文件目录:{project_path}\n\n"
+ f"你的团队有以下角色(每个 phase 必须使用这些角色之一):{roles_hint}\n"
+ f"请输出包含 phases 数组的完整 JSON 计划,每个 phase 需指定 role 为上述角色之一。"
+ )
+ planner_result = await planner_runtime.run(planner_input)
+ plan = _parse_plan_json(planner_result.content) if planner_result.success else {}
+ results["plan"] = plan
+
+ if not plan or not plan.get("phases"):
+ results["success"] = False
+ results["error"] = f"{planner_role} 未能生成有效计划"
+ results["final_deliverable"] = planner_result.content
+ return results
+
+ logger.info("团队编排 [%s]: PM 生成 %d 个阶段", self.team_id, len(plan["phases"]))
+
+ # ─── Phase 1..N: 顺序执行 ───
+ all_outputs: List[str] = []
+ context_doc = f"# 项目: {plan.get('project_name', '未命名')}\n\n## 需求分析\n{plan.get('analysis', '')}\n\n"
+
+ for i, phase in enumerate(plan["phases"]):
+ phase_num = phase.get("phase", i + 1)
+ role = phase.get("role", "developer")
+ phase_name = phase.get("name", f"阶段 {phase_num}")
+ phase_desc = phase.get("description", "")
+ expected = phase.get("expected_output", "")
+
+ agent = self._get_agent_by_role(members, role)
+ if not agent:
+ logger.warning("团队编排 [%s]: 阶段 %s 缺少角色 '%s',跳过", self.team_id, phase_num, role)
+ all_outputs.append(f"[跳过] 阶段 {phase_num} ({phase_name}): 无 {role} 角色")
+ results["phases"].append({
+ "phase": phase_num,
+ "name": phase_name,
+ "role": role,
+ "agent_name": None,
+ "output": f"跳过: 无 {role} 角色",
+ "success": False,
+ "error": f"missing_role:{role}",
+ "files": [],
+ })
+ continue
+
+ logger.info("团队编排 [%s]: 执行阶段 %s — %s (角色: %s, Agent: %s)",
+ self.team_id, phase_num, phase_name, role, agent.name)
+
+ try:
+ agent_config = _build_agent_config(agent, self.auto_approve_files)
+ runtime = AgentRuntime(agent_config)
+
+ phase_input = (
+ f"{context_doc}\n"
+ f"## 当前阶段: {phase_name}\n"
+ f"任务描述: {phase_desc}\n"
+ f"期望产出: {expected}\n"
+ f"项目文件保存目录: {project_path}\n"
+ f"请将所有产出文件写入此目录,完成此阶段的工作并输出你的交付物。"
+ )
+ phase_result = await runtime.run(phase_input)
+
+ output_text = phase_result.content if phase_result.success else (
+ phase_result.error or "执行失败"
+ )
+ context_doc += f"\n## 阶段 {phase_num}: {phase_name}\n{output_text[:2000]}\n"
+
+ all_outputs.append(f"## {phase_name} ({role})\n{output_text}")
+
+ # 提取本阶段写入的文件(从 AgentResult + 目录扫描双保险)
+ extracted = _extract_written_files(phase_result)
+ current_files = set(_scan_directory_files(project_path))
+ phase_files = sorted(current_files - prev_files)
+ # 同时保留从 AgentResult 提取的文件(可能包含项目目录外的文件)
+ for fp in extracted:
+ if fp not in phase_files:
+ phase_files.append(fp)
+ prev_files = current_files
+ all_files.extend(phase_files)
+
+ results["phases"].append({
+ "phase": phase_num,
+ "name": phase_name,
+ "role": role,
+ "agent_name": agent.name,
+ "output": output_text,
+ "success": phase_result.success,
+ "iterations": phase_result.iterations_used,
+ "tool_calls": phase_result.tool_calls_made,
+ "error": phase_result.error if not phase_result.success else None,
+ "files": phase_files,
+ })
+ except Exception as e:
+ logger.error("团队编排 [%s]: 阶段 %s 异常: %s", self.team_id, phase_num, e)
+ results["phases"].append({
+ "phase": phase_num,
+ "name": phase_name,
+ "role": role,
+ "agent_name": agent.name if agent else None,
+ "output": f"执行异常: {e}",
+ "success": False,
+ "error": str(e),
+ "files": [],
+ })
+ results["success"] = False
+
+ # ─── QA 审查 ───
+ qa_agent = self._get_agent_by_role(members, "qa")
+ if qa_agent and all_outputs:
+ logger.info("团队编排 [%s]: QA 审查开始", self.team_id)
+ try:
+ qa_config = _build_agent_config(qa_agent, self.auto_approve_files)
+ qa_runtime = AgentRuntime(qa_config)
+ deliverables = "\n\n---\n\n".join(all_outputs)
+ qa_input = (
+ f"请审查以下项目交付物:\n\n"
+ f"项目: {project_description}\n\n"
+ f"{deliverables[:8000]}\n\n"
+ f"请按 JSON 格式输出审查结果,包含 pass/score/issues/overall_assessment。"
+ )
+ qa_result = await qa_runtime.run(qa_input)
+ qa_review = _parse_plan_json(qa_result.content) if qa_result.success else {}
+ results["qa_review"] = {
+ "output": qa_result.content,
+ "review": qa_review,
+ "success": qa_result.success,
+ }
+ except Exception as e:
+ logger.error("团队编排 [%s]: QA 审查异常: %s", self.team_id, e)
+ results["qa_review"] = {"error": str(e)}
+
+ # ─── 汇总最终交付物 ───
+ results["final_deliverable"] = "\n\n---\n\n".join(all_outputs)
+ # 最终文件列表以目录扫描为准,合并从 AgentResult 提取的
+ final_scan = set(_scan_directory_files(project_path))
+ for fp in all_files:
+ final_scan.add(fp)
+ results["files"] = sorted(final_scan)
+ results["completed_at"] = datetime.now().isoformat()
+ return results
+
+ async def execute_stream(self, project_description: str, auto_approve_files: bool = True) -> AsyncGenerator[Dict[str, Any], None]:
+ """流式执行团队项目,yield SSE 事件。
+
+ 事件类型:
+ - plan_start / plan / plan_done — PM 规划阶段
+ - phase_start / phase_output / phase_done — 各执行阶段
+ - qa_start / qa_review / qa_done — QA 审查
+ - final — 全部完成
+ - error — 错误
+ """
+ self.auto_approve_files = auto_approve_files
+ try:
+ team = self._load_team()
+ members = self._load_members()
+ except Exception as e:
+ yield {"type": "error", "content": str(e)}
+ return
+
+ if not members:
+ yield {"type": "error", "content": "团队没有成员"}
+ return
+
+ # ─── 创建项目目录 ───
+ workspace_root = _resolve_workspace_root()
+ safe_name = _safe_project_name(project_description)
+ project_path = workspace_root / "team_projects" / self.team_id / safe_name
+ project_path.mkdir(parents=True, exist_ok=True)
+ prev_files = set(_scan_directory_files(project_path))
+ all_files: List[str] = []
+
+ yield {
+ "type": "plan_start",
+ "team_name": team.name,
+ "member_count": len(members),
+ "project_path": str(project_path),
+ }
+
+ all_outputs: List[str] = []
+
+ # ─── 规划阶段(由 Leader Agent 担任) ───
+ workflow = (team.config or {}).get("workflow", "software_company")
+ planner_role = self._resolve_planner_role(members, workflow)
+ planner_agent = self._get_agent_by_role(members, planner_role)
+ if not planner_agent:
+ yield {"type": "error", "content": f"团队缺少 Leader 角色(需要 role='{planner_role}'),可用角色: {[m.role for m in members]}"}
+ return
+
+ try:
+ available_roles = [m.role for m in members]
+ roles_hint = "/".join(available_roles)
+
+ planner_config = _build_agent_config(planner_agent, self.auto_approve_files)
+ planner_runtime = AgentRuntime(planner_config)
+ planner_input = (
+ f"请分析以下需求,按 JSON 格式输出执行计划:\n\n"
+ f"需求描述:{project_description}\n"
+ f"输出文件目录:{project_path}\n"
+ f"你的团队有以下角色(每个 phase 必须使用这些角色之一):{roles_hint}\n"
+ f"请输出包含 phases 数组的完整 JSON 计划,每个 phase 需指定 role 为上述角色之一。"
+ )
+
+ yield {"type": "plan", "status": "running", "agent": planner_agent.name}
+ planner_result = await planner_runtime.run(planner_input)
+ plan = _parse_plan_json(planner_result.content) if planner_result.success else {}
+
+ yield {
+ "type": "plan_done",
+ "plan": plan,
+ "raw_output": planner_result.content[:2000],
+ "success": planner_result.success,
+ }
+ except Exception as e:
+ yield {"type": "error", "content": f"{planner_role} 规划失败: {e}"}
+ return
+
+ if not plan or not plan.get("phases"):
+ yield {"type": "error", "content": f"{planner_role} 未能生成有效计划"}
+ return
+
+ phases = plan["phases"]
+ context_doc = f"# 项目: {plan.get('project_name', '未命名')}\n\n## 需求分析\n{plan.get('analysis', '')}\n\n"
+
+ # ─── 顺序执行各阶段 ───
+ for i, phase in enumerate(phases):
+ phase_num = phase.get("phase", i + 1)
+ role = phase.get("role", "developer")
+ phase_name = phase.get("name", f"阶段 {phase_num}")
+ phase_desc = phase.get("description", "")
+ expected = phase.get("expected_output", "")
+
+ agent = self._get_agent_by_role(members, role)
+ if not agent:
+ msg = f"阶段 {phase_num} ({phase_name}) 跳过: 无 {role} 角色"
+ yield {"type": "phase_done", "phase": phase_num, "name": phase_name,
+ "role": role, "success": False, "error": f"missing_role:{role}"}
+ all_outputs.append(f"[跳过] {msg}")
+ continue
+
+ yield {
+ "type": "phase_start",
+ "phase": phase_num,
+ "name": phase_name,
+ "role": role,
+ "agent": agent.name,
+ }
+
+ try:
+ agent_config = _build_agent_config(agent, self.auto_approve_files)
+ runtime = AgentRuntime(agent_config)
+
+ phase_input = (
+ f"{context_doc}\n"
+ f"## 当前阶段: {phase_name}\n"
+ f"任务描述: {phase_desc}\n"
+ f"期望产出: {expected}\n"
+ f"项目文件保存目录: {project_path}\n"
+ f"请将所有产出文件写入此目录,完成此阶段的工作并输出你的交付物。"
+ )
+ phase_result = await runtime.run(phase_input)
+
+ output_text = phase_result.content if phase_result.success else (
+ phase_result.error or "执行失败"
+ )
+ context_doc += f"\n## 阶段 {phase_num}: {phase_name}\n{output_text[:2000]}\n"
+ all_outputs.append(f"## {phase_name} ({role})\n{output_text}")
+
+ # 提取本阶段写入的文件(从 AgentResult + 目录扫描双保险)
+ extracted = _extract_written_files(phase_result)
+ current_files = set(_scan_directory_files(project_path))
+ phase_files = sorted(current_files - prev_files)
+ for fp in extracted:
+ if fp not in phase_files:
+ phase_files.append(fp)
+ prev_files = current_files
+ all_files.extend(phase_files)
+
+ yield {
+ "type": "phase_done",
+ "phase": phase_num,
+ "name": phase_name,
+ "role": role,
+ "agent": agent.name,
+ "output": output_text,
+ "success": phase_result.success,
+ "iterations": phase_result.iterations_used,
+ "tool_calls": phase_result.tool_calls_made,
+ "error": phase_result.error if not phase_result.success else None,
+ "files": phase_files,
+ }
+ except Exception as e:
+ logger.error("团队编排: 阶段 %s 异常: %s", phase_num, e)
+ yield {
+ "type": "phase_done",
+ "phase": phase_num,
+ "name": phase_name,
+ "role": role,
+ "agent": agent.name,
+ "success": False,
+ "error": str(e),
+ }
+
+ # ─── QA 审查 ───
+ qa_agent = self._get_agent_by_role(members, "qa")
+ if qa_agent and all_outputs:
+ yield {"type": "qa_start", "agent": qa_agent.name}
+ try:
+ qa_config = _build_agent_config(qa_agent, self.auto_approve_files)
+ qa_runtime = AgentRuntime(qa_config)
+ deliverables = "\n\n---\n\n".join(all_outputs)
+ qa_input = (
+ f"请审查以下项目交付物:\n\n"
+ f"项目: {project_description}\n\n"
+ f"{deliverables[:8000]}\n\n"
+ f"请按 JSON 格式输出审查结果,包含 pass/score/issues/overall_assessment。"
+ )
+ qa_result = await qa_runtime.run(qa_input)
+ qa_review_data = _parse_plan_json(qa_result.content) if qa_result.success else {}
+ yield {
+ "type": "qa_done",
+ "output": qa_result.content,
+ "review": qa_review_data,
+ "success": qa_result.success,
+ }
+ except Exception as e:
+ logger.error("团队编排: QA 审查异常: %s", e)
+ yield {"type": "qa_done", "error": str(e)}
+
+ # ─── 最终交付物 ───
+ final = "\n\n---\n\n".join(all_outputs)
+ final_scan = set(_scan_directory_files(project_path))
+ for fp in all_files:
+ final_scan.add(fp)
+ yield {
+ "type": "final",
+ "deliverable": final,
+ "phase_count": len(phases),
+ "team_name": team.name,
+ "project_path": str(project_path),
+ "files": sorted(final_scan),
+ }
diff --git a/backend/app/services/team_service.py b/backend/app/services/team_service.py
new file mode 100644
index 0000000..36b7339
--- /dev/null
+++ b/backend/app/services/team_service.py
@@ -0,0 +1,1100 @@
+"""
+团队服务 — Team CRUD + 预置软件公司模板工厂
+"""
+from __future__ import annotations
+
+import uuid
+import logging
+from typing import Any, Dict, List, Optional
+
+from sqlalchemy.orm import Session
+from app.models.team import Team, TeamMember
+from app.models.agent import Agent
+
+logger = logging.getLogger(__name__)
+
+# 预置角色定义
+PRESET_ROLES = {
+ "pm": {
+ "label": "项目经理",
+ "icon": "📋",
+ "description": "分析需求、分解任务、制定计划、协调团队",
+ },
+ "designer": {
+ "label": "UI/UX设计师",
+ "icon": "🎨",
+ "description": "设计用户流程、线框图、组件层级、视觉规范",
+ },
+ "developer": {
+ "label": "开发工程师",
+ "icon": "💻",
+ "description": "编写后端API、前端组件、业务逻辑、数据库设计",
+ },
+ "qa": {
+ "label": "测试工程师",
+ "icon": "🔍",
+ "description": "审查交付物、发现缺陷、验证验收标准、质量把关",
+ },
+ "devops": {
+ "label": "DevOps工程师",
+ "icon": "🚀",
+ "description": "部署配置、CI/CD、环境管理、健康检查",
+ },
+}
+
+# 教育培训团队角色定义
+EDUCATION_ROLES = {
+ "curriculum_designer": {
+ "label": "课程设计师",
+ "icon": "📐",
+ "description": "设计课程体系、学习目标、知识模块、评估方法",
+ },
+ "instructor": {
+ "label": "主讲讲师",
+ "icon": "🎓",
+ "description": "授课讲解、制作课件、设计互动、辅导答疑",
+ },
+ "teaching_assistant": {
+ "label": "作业助教",
+ "icon": "✏️",
+ "description": "批改作业、学习辅导、进度跟踪、答疑反馈",
+ },
+ "academic_admin": {
+ "label": "教务管理",
+ "icon": "📋",
+ "description": "排课管理、学员管理、证书管理、运营支持",
+ },
+}
+
+# 天工平台工程团队角色定义
+PLATFORM_ENGINEERING_ROLES = {
+ "fullstack_dev": {
+ "label": "全栈开发工程师",
+ "icon": "🏗️",
+ "description": "核心引擎维护、API开发、数据库迁移、Bug修复、代码审查",
+ },
+ "fe_ux_engineer": {
+ "label": "前端体验工程师",
+ "icon": "🎯",
+ "description": "工作流编辑器优化、移动端适配、交互打磨、飞书Bot体验",
+ },
+ "platform_devops": {
+ "label": "平台运维工程师",
+ "icon": "🚀",
+ "description": "部署发布、CI/CD维护、监控告警、日志聚合、K8s配置",
+ },
+ "qa_engineer": {
+ "label": "质量保障工程师",
+ "icon": "🔍",
+ "description": "测试金字塔、性能压测、安全审计、API契约测试、回归测试",
+ },
+ "product_lead": {
+ "label": "产品负责人",
+ "icon": "📊",
+ "description": "需求排序、多租户设计、定价计费、模板市场、文档体系",
+ },
+}
+
+# ─── 角色专属 Agent 系统提示词 ───
+
+FULLSTACK_DEVELOPER_PROMPT = """You are a Senior Full-Stack Developer maintaining and iterating on the Tiangong AI Agent Platform (天工智能体平台).
+
+Tech stack: Python/FastAPI backend + Vue 3/TypeScript frontend + SQLAlchemy 2.0 + MySQL 8.0 + Redis 7 + Celery 5.3 + Docker.
+
+The platform has:
+- 50+ database tables, 37 model files
+- ~241 API endpoints across 37 route modules
+- 56 built-in tools across 11 categories
+- 10万+ lines of production code
+- Vue Flow visual workflow editor
+- Multi-agent orchestration (5 modes + Swarm)
+- 3-layer memory system (context / vector RAG / persistent)
+- Dual knowledge engine (RAG + Knowledge Graph)
+- Feishu deep integration (7 tools, 6 bots)
+
+Your responsibilities:
+1. Fix bugs — read error logs, trace the issue, write a fix, verify with tests
+2. Add features — follow the existing architecture patterns, write complete code with types
+3. Database migrations — use Alembic: `alembic revision --autogenerate -m "description"` then `alembic upgrade head`
+4. Code review — check for security (SQL injection, XSS, hardcoded secrets), performance (N+1 queries, missing indexes), and correctness
+5. API consistency — follow REST conventions, proper HTTP status codes, Pydantic validation
+6. Refactor safely — keep backwards compatibility, add deprecation warnings if needed
+
+When writing code:
+- Python: type hints everywhere, use async/await for I/O, follow PEP 8
+- TypeScript: strict types, Composition API for Vue components
+- Always include error handling and input validation
+- Write self-documenting code with clear variable/function names
+- Each file must be complete and ready to run — no placeholders or TODOs"""
+
+
+FE_UX_ENGINEER_PROMPT = """You are a Frontend & UX Engineer for the Tiangong AI Agent Platform. Your focus is user experience and frontend quality.
+
+The frontend is a Vue 3 SPA with:
+- Vite build system, Pinia state management, Element Plus UI library
+- Vue Flow for the visual workflow editor (drag-and-drop DAG)
+- 28 pages covering agent builder, knowledge base, chat, monitoring, teams, etc.
+- Feishu bot integration with 6 active bots
+
+Your responsibilities:
+1. Workflow editor optimization:
+ - Auto-layout algorithms for DAG nodes
+ - Search and filter for large workflows
+ - Version comparison and diff view
+ - Node template quick-apply
+2. Mobile responsive adaptation:
+ - Adapt the 28 pages for mobile/tablet screens
+ - Touch-friendly interactions (drag, resize, tap targets)
+ - PWA offline support
+3. UI/UX polish:
+ - Loading states, empty states, error states for every page
+ - Consistent design tokens (colors, spacing, typography)
+ - Accessibility: ARIA labels, keyboard navigation, focus management
+4. Performance:
+ - Lazy loading for route pages and heavy components
+ - Virtual scrolling for long lists (agents, conversations, logs)
+ - Bundle size optimization (tree-shaking, code splitting)
+5. Feishu Bot UX:
+ - Feedback buttons (thumbs up/down) on bot responses
+ - Conversation summaries and history
+ - Clear error messages and retry flows
+
+When implementing:
+- Use Vue 3 Composition API with
+
+