"""Agent loop — 角色agent循环认领执行任务（v3.4.0）。 v3.4.0 新增： - 所有agent具备文件读写 + git pull/push 能力 - agent 产出实际代码文件到项目仓库，而非仅文档 - PM agent 负责 clone 项目关联仓库、初始化工作区 - develop agent 产出可直接运行的源码文件 - 每个agent完成后自动 git commit + push 任务链：requirement → design → develop → test → deploy """ import asyncio import json import os import subprocess import logging logger = logging.getLogger("pipeline.agent_loop") ROLE_ALIASES = { 'developer': 'develop', 'dev': 'develop', 'development': 'develop', 'coding': 'develop', 'requirement': 'requirement', 'requirements': 'requirement', 'requirement_analysis': 'requirement', 'designer': 'design', 'ui': 'design', 'ux': 'design', 'testing': 'test', 'qa': 'test', 'tester': 'test', 'deployment': 'deploy', 'release': 'deploy', 'operation': 'ops', 'operations': 'ops', 'maintenance': 'ops', '运维': 'ops', 'pm': 'pm', 'project_manager': 'pm', '项目经理': 'pm', 'manager': 'pm', } ROLE_CHAIN = { 'requirement': 'design', 'design': 'develop', 'develop': 'test', 'test': 'deploy', 'deploy': None, } TASK_REVIEW = 'review' TASK_APPROVED = 'approved' # 允许的 shell 工作目录前缀（安全限制） _ALLOWED_WORKDIRS = [ os.path.expanduser('~/pipeline_ws'), '/d/pipeline/workspaces', '/tmp/pipeline_workspaces', '/tmp/pipeline_ws', ] def _normalize_role(role): r = (role or '').strip().lower() return ROLE_ALIASES.get(r, r) def _get_db(): from sqlor.dbpools import DBPools db = DBPools() if not db.databases: from appPublic.jsonConfig import getConfig config = getConfig() if config.databases: db.databases = config.databases return db def _resolve_workspace(workspace_dir): if not workspace_dir: workspace_dir = os.path.expanduser('~/pipeline_ws/default') try: parent = os.path.dirname(workspace_dir) if parent and not os.access(parent, os.W_OK): proj_name = os.path.basename(workspace_dir) workspace_dir = os.path.expanduser(f'~/pipeline_ws/{proj_name}') except Exception: workspace_dir = os.path.expanduser(f'~/pipeline_ws/{os.path.basename(workspace_dir) or "default"}') os.makedirs(workspace_dir, exist_ok=True) return workspace_dir # ── Agent 工具函数 ── def _is_safe_workdir(workdir): """检查目录是否在允许范围内。""" wd = os.path.abspath(workdir) for allowed in _ALLOWED_WORKDIRS: awd = os.path.abspath(os.path.expanduser(allowed)) if wd.startswith(awd): return True return False async def _run_shell(command, workdir, timeout=120): """安全执行 shell 命令。返回 {"rc": int, "stdout": str, "stderr": str}""" cwd = os.path.abspath(workdir) if workdir else os.path.expanduser('~/pipeline_ws') if not _is_safe_workdir(cwd): return {"rc": -1, "stdout": "", "stderr": f"安全限制：目录 {cwd} 不在允许范围"} if not os.path.isdir(cwd): return {"rc": -1, "stdout": "", "stderr": f"目录不存在: {cwd}"} try: proc = await asyncio.create_subprocess_shell( command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=cwd, executable='/bin/bash') try: stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) except asyncio.TimeoutError: proc.kill() await proc.wait() return {"rc": -1, "stdout": "", "stderr": f"命令超时（{timeout}s）"} return {"rc": proc.returncode or 0, "stdout": stdout.decode('utf-8', 'replace')[-8000:], "stderr": stderr.decode('utf-8', 'replace')[-4000:]} except Exception as e: return {"rc": -1, "stdout": "", "stderr": str(e)[:500]} async def _write_code_file(filepath, content): """写代码文件，自动创建父目录。""" try: d = os.path.dirname(filepath) if d: os.makedirs(d, exist_ok=True) with open(filepath, 'w', encoding='utf-8') as f: f.write(content or '') return True, filepath except Exception as e: return False, str(e) async def _git_setup(workdir): """确保 git 用户已配置。""" r = await _run_shell('git config user.email', workdir, 5) if 'opencomputing' not in r.get('stdout', '') and '@' not in r.get('stdout', ''): await _run_shell('git config user.email "pipeline@opencomputing.cn"', workdir, 5) await _run_shell('git config user.name "Pipeline Agent"', workdir, 5) async def _git_commit_push(workdir, commit_message, branch='main'): """git add → commit → push。""" await _git_setup(workdir) r1 = await _run_shell('git add -A', workdir, 10) r2 = await _run_shell(f'git diff --cached --stat', workdir, 10) if not r2.get('stdout', '').strip(): return {"rc": 0, "message": "没有变更需要提交"} r3 = await _run_shell(f'git commit -m "{commit_message}"', workdir, 15) if r3['rc'] != 0: return {"rc": r3['rc'], "message": f"commit 失败: {r3['stderr'][:200]}"} r4 = await _run_shell(f'git push origin {branch}', workdir, 30) return {"rc": r4['rc'], "message": f"push {'成功' if r4['rc']==0 else '失败'}: {r4['stderr'][:200]}"} async def _git_clone(repo_url, target_dir, branch='main'): """克隆仓库到目标目录。已存在则 pull。""" if os.path.isdir(os.path.join(target_dir, '.git')): r = await _run_shell(f'git checkout {branch} && git pull origin {branch}', target_dir, 30) return {"rc": r['rc'], "message": f"已存在，pull: {r['stdout'][:200]}"} parent = os.path.dirname(target_dir) os.makedirs(parent, exist_ok=True) r = await _run_shell(f'git clone -b {branch} {repo_url} {target_dir}', parent, 120) return {"rc": r['rc'], "message": f"clone: {r['stdout'][:200] if r['rc']==0 else r['stderr'][:200]}"} # ── Prompts ── AGENT_TOOLS = [ {"name":"read_file","description":"读取工作空间中的文件","params":{"path":"相对路径"}}, {"name":"write_file","description":"写入文件（自动创建父目录）","params":{"path":"相对路径","content":"文件内容"}}, {"name":"list_files","description":"列出目录内容","params":{"path":"相对路径(可选,默认工作空间根)"}}, {"name":"run_shell","description":"在工作空间中执行shell命令","params":{"command":"命令"}}, {"name":"git_status","description":"查看git仓库状态","params":{"repo_dir":"仓库子目录(可选,默认repos下第一个)"}}, {"name":"git_commit_push","description":"git add + commit + push","params":{"message":"提交信息","repo_dir":"仓库子目录(可选)"}}, {"name":"ask_question","description":"向用户提问（缺少信息时使用）","params":{"question":"问题"}}, ] AGENT_SYSTEM_PROMPT = """你是软件开发产线中的「__ROLE__」角色Agent。 ## 任务 __TITLE__ __QNA__ ## 工作环境 工作空间：__WORKSPACE__ 产出要求：__ROLE_SPECIFIC__ ## 工具 你可以使用以下工具完成工作： __TOOLS__ ## 工作流 1. 先用 read_file/list_files 了解现有代码 2. 用 write_file 产出代码文件到 repos/ 下 3. 用 run_shell 验证（编译/测试） 4. 用 git_commit_push 提交到远端 5. 用 deliver 提交最终交付件 ## 输出格式（每次只输出一个JSON对象） 调工具： {"action":"tool_call","tool":"工具名","params":{}} 提交交付件： {"action":"deliver","deliverable_type":"code_files","summary":"概述","result":"文档内容","files":[{"path":"repos/仓库/src/file.py","content":"代码"}],"git_commit_message":"feat: 描述"} 提问： {"action":"ask","question":"问题"} 注意：每次只输出一个JSON！收到工具结果后再决定下一步。""" ROLE_SPECIFICS = { 'requirement': """你是需求分析师。按 SDLC 仓库标准产出文档。 产出路径: docs/00-requirement/requirement-spec.md 内容: 项目概述、用户角色及权限、功能列表(每个功能:输入/处理/输出/验收标准)、非功能需求、业务流程。 产出后用 result 字段输出文档正文，files 字段列出文件路径。""", 'design': """你是系统设计师。按 SDLC 仓库标准产出文档。 产出路径: docs/01-design/ 目录下: - architecture.md: 系统架构、技术选型理由 - database-design.md: ER图描述、表结构DDL - api-design.md: 接口列表(method/path/request/response) - ui-design.md: 页面结构、组件树(如适用) 产出后用 result 输出文档，files 列出所有文件路径。""", 'develop': """你是开发工程师。必须产出实际可运行的代码文件！ 产出路径: - src/<模块名>/ 下: 所有源代码文件 - docs/02-develop/code-notes.md: 开发说明 行为准则: 1. 先用 read_file 读 design 文档和现有代码 2. 用 write_file 逐文件写入 src/ 下 3. 用 run_shell 编译/运行验证 4. 用 git_commit_push 提交 "develop: <简述>" 5. 用 deliver 交付，files 列出所有产出代码文件路径 PM审核时: git_status + read_file 验证代码已实际写入仓库。""", 'test': """你是测试工程师。按 SDLC 仓库标准产出文档。 产出路径: docs/03-test/ 目录下: - test-plan.md: 测试策略(单元/集成/端到端) - test-cases.md: 用例清单(编号/前置条件/步骤/预期结果) - test-report.md: 执行结果、Bug清单、覆盖率 测试脚本放到 files。""", 'deploy': """你是部署运维工程师。按 SDLC 仓库标准产出文档和配置。 产出路径: - docs/04-deploy/deploy-guide.md: 环境要求、部署步骤、回滚方案 - docs/04-deploy/release-notes.md: 发布说明 - config/ 下: Dockerfile、nginx配置、docker-compose.yml 等 产出后用 result 输出部署说明，files 列出所有配置文件路径。""", } PM_PROMPT = """你是项目经理（PM）。审核交付件并推动项目前进。 ## 项目 工作目录：__WORKSPACE__ 关联仓库：__REPOS__ ## 待审核任务 标题：__TITLE__ 角色：__ROLE__ 交付件摘要：__SUMMARY__ 交付件内容(前6000字)：__DELIVERABLE__ ## 仓库状态 __REPO_STATE__ ## 决策 输出纯JSON： 1. 通过：{"status":"approved","comment":"审核意见","next_task_title":"下阶段标题","next_task_description":"详细描述"} 2. 驳回：{"status":"rejected","comment":"驳回原因","questions":"修改要求"} 3. 完成：{"status":"completed","comment":"项目总结"} 审核标准： - requirement：需求是否清晰完整可量化 - design：方案合理、覆盖需求、技术可行 - develop：代码文件是否实际产出、目录结构是否合理、是否可编译运行 - test：测试覆盖充分、发现问题记录完整 - deploy：部署配置完整、可一键部署""" PM_SYSTEM_PROMPT = """你是项目经理（PM）。审核交付件并推动项目前进。 ## 项目信息 工作目录：__WORKSPACE__ 关联仓库：__REPOS__ ## 待审核 标题：__TITLE__ 角色：__ROLE__ ## 仓库状态 __REPO_STATE__ ## 工具 - read_file(path) — 读工作空间文件 - list_files(path) — 列目录 - git_status() — 查看git状态 - run_shell(command) — 执行命令（编译/测试验证） ## 审核流程 1. 先用工具检查代码是否实际产出（git_status/list_files/read_file） 2. 如需编译验证，用 run_shell 3. 最后给出决策 ## 输出格式（每次一个JSON） 查看文件：{"action":"tool_call","tool":"read_file","params":{"path":"相对路径"}} 查看git：{"action":"tool_call","tool":"git_status","params":{}} 批准：{"action":"review_approve","comment":"审核意见","next_task_title":"下阶段标题","next_task_description":"描述"} 驳回：{"action":"review_reject","comment":"原因","questions":"修改要求"} 完成：{"action":"review_complete","comment":"总结"} 审核标准： - requirement：需求是否清晰完整可量化 - design：方案合理、覆盖需求、技术可行 - develop：必须用 git_status/read_file 检查代码是否实际写入仓库 - test：测试覆盖充分、发现问题记录完整 - deploy：部署配置完整、可一键部署""" # ── Agent 核心逻辑 ── async def _claim_task(sor, tenant_id, role, state='submitted', match_role=True): role = _normalize_role(role) where_role = "AND (role=${role}$ OR role='')" if match_role else "" recs = await sor.sqlExe( "SELECT id, title, params, pipeline_id, tenant_id, role FROM pipeline_tasks " "WHERE tenant_id=${tid}$ AND state=${state}$ " + where_role + " " "AND NOT EXISTS (SELECT 1 FROM pipeline_task_steps s WHERE s.task_id=pipeline_tasks.id) " "ORDER BY created_at ASC LIMIT 1", {"tid": tenant_id, "state": state, "role": role}) if not recs: return None task = recs[0] task_id = task.id from appPublic.uniqueID import getID claim_token = getID() await sor.sqlExe( "UPDATE pipeline_tasks SET state='running', claimed_by=${cb}$ " "WHERE id=${tid}$ AND state=${state}$", {"cb": claim_token, "tid": task_id, "state": state}) check = await sor.sqlExe( "SELECT id FROM pipeline_tasks WHERE id=${tid}$ AND state='running' AND claimed_by=${cb}$", {"tid": task_id, "cb": claim_token}) if not check: logger.info(f"claim lost race: task={task_id}") return None return task async def _get_workspace_dir(sor, project_id): recs = await sor.sqlExe("SELECT workspace_dir FROM sd_projects WHERE id=${pid}$", {"pid": project_id}) ws = getattr(recs[0], 'workspace_dir', '') if recs else '' return _resolve_workspace(ws) async def _build_qna_section(sor, task_id): from .questions import get_task_qna qna = await get_task_qna(task_id) if not qna: return "" lines = ["历史问答："] for item in qna: src = "客户" if item.get('answer_source') == "customer" else "主agent" lines.append(f"问：{item.get('question','')}") lines.append(f"答({src})：{item.get('answer','')}") return "\n".join(lines) def _parse_result(raw): raw = (raw or "").strip() if raw.startswith("```"): raw = raw.split("\n", 1)[1].rsplit("```", 1)[0] try: d = json.loads(raw) if isinstance(d, dict): return d except (json.JSONDecodeError, ValueError): pass return {"status": "done", "result": raw} async def _get_deliverable_content(sor, task_id): recs = await sor.sqlExe( "SELECT content, deliverable_type FROM pipeline_deliverables " "WHERE task_id=${tid}$ ORDER BY created_at DESC LIMIT 1", {"tid": task_id}) if recs: return getattr(recs[0], 'content', '') or '', getattr(recs[0], 'deliverable_type', '') or '' return '', '' async def _get_project_repos(sor, project_id): """获取项目关联的所有仓库。""" recs = await sor.sqlExe( "SELECT id, repo_name, repo_url, default_branch, local_path FROM sd_project_repos " "WHERE project_id=${pid}$", {"pid": project_id}) return [{'id': getattr(r, 'id', ''), 'name': getattr(r, 'repo_name', ''), 'url': getattr(r, 'repo_url', ''), 'branch': getattr(r, 'default_branch', 'main'), 'local_path': getattr(r, 'local_path', '') or ''} for r in (recs or [])] async def _setup_repos(sor, workspace_dir, project_id): """PM：clone 所有项目关联仓库到 workspace/repos/。""" repos = await _get_project_repos(sor, project_id) results = [] repos_dir = os.path.join(workspace_dir, 'repos') os.makedirs(repos_dir, exist_ok=True) for repo in repos: target = os.path.join(repos_dir, repo['name']) r = await _git_clone(repo['url'], target, repo['branch']) results.append({'repo': repo['name'], **r}) return results async def _get_repo_state(workspace_dir): """获取仓库当前状态（供PM审核时查看）。""" repos_dir = os.path.join(workspace_dir, 'repos') if not os.path.isdir(repos_dir): return "暂无仓库" lines = [] for name in sorted(os.listdir(repos_dir)): rp = os.path.join(repos_dir, name) if os.path.isdir(rp) and os.path.isdir(os.path.join(rp, '.git')): r = await _run_shell('git log --oneline -3', rp, 5) lines.append(f"\n[{name}]") lines.append(r.get('stdout', '') or '(空仓库)') return "\n".join(lines) if lines else "仓库无提交记录" async def _get_next_role(current_role): return ROLE_CHAIN.get(_normalize_role(current_role)) async def _create_next_task(sor, project_id, task, next_role, pm_comment=''): from appPublic.uniqueID import getID title = getattr(task, 'title', '') or '' params_str = getattr(task, 'params', '{}') or '{}' try: params = json.loads(params_str) if isinstance(params_str, str) else params_str except (json.JSONDecodeError, TypeError): params = {} new_title = f"{title}（{next_role}阶段）" new_params = {**params, 'previous_role': _normalize_role(getattr(task, 'role', '')), 'previous_task_id': getattr(task, 'id', ''), 'pm_comment': pm_comment} new_task_id = getID() await sor.C('pipeline_tasks', { 'id': new_task_id, 'tenant_id': project_id, 'pipeline_id': 'role_task', 'owner_id': 'pm', 'title': new_title, 'params': json.dumps(new_params, ensure_ascii=False), 'role': next_role, 'state': 'submitted', 'claimed_by': None, }) logger.info(f"_create_next_task: {next_role} task={new_task_id}") return new_task_id, new_title # ── Agent 工具执行 ── def _parse_agent_action(raw): raw = (raw or "").strip() if raw.startswith("```"): raw = raw.split("\n", 1)[1].rsplit("```", 1)[0].strip() try: d = json.loads(raw) if isinstance(d, dict) and 'action' in d: return d except (json.JSONDecodeError, ValueError): pass return {"action": "deliver", "result": raw} async def _exec_agent_tool(tool, params, workspace_dir): p = params or {} try: if tool == 'read_file': path = p.get('path', '') if not path: return 'FAIL: 需要文件路径' full = os.path.join(workspace_dir, path) if not _is_safe_workdir(full): return 'FAIL: 路径不在允许范围' if not os.path.isfile(full): return f'FAIL: 文件不存在 {path}' with open(full, encoding='utf-8') as f: return f.read()[:8000] elif tool == 'write_file': path = p.get('path', '') content = p.get('content', '') if not path: return 'FAIL: 需要文件路径' full = os.path.join(workspace_dir, path) if not _is_safe_workdir(full): return 'FAIL: 路径不在允许范围' os.makedirs(os.path.dirname(full), exist_ok=True) with open(full, 'w', encoding='utf-8') as f: f.write(content) return f'OK: 已写入 {path} ({len(content)} 字符)' elif tool == 'list_files': path = p.get('path', '') or '.' full = os.path.join(workspace_dir, path) if not _is_safe_workdir(full): return 'FAIL: 路径不在允许范围' if not os.path.isdir(full): return f'FAIL: 目录不存在 {path}' items = os.listdir(full)[:50] lines = [] for name in sorted(items): fp = os.path.join(full, name) t = 'DIR' if os.path.isdir(fp) else 'FILE' size = os.path.getsize(fp) if os.path.isfile(fp) else 0 lines.append(f"[{t}] {name} ({size}B)") return '\n'.join(lines) if lines else '(空目录)' elif tool == 'run_shell': cmd = p.get('command', '') if not cmd: return 'FAIL: 需要命令' r = await _run_shell(cmd, workspace_dir, timeout=120) return f"rc={r['rc']}\nSTDOUT:\n{r['stdout'][:2000]}\nSTDERR:\n{r['stderr'][:1000]}" elif tool == 'git_status': repo = p.get('repo_dir', '') repos_dir = os.path.join(workspace_dir, 'repos') if repo: target = os.path.join(workspace_dir, repo) else: dirs = [d for d in os.listdir(repos_dir) if os.path.isdir(os.path.join(repos_dir, d, '.git'))] if os.path.isdir(repos_dir) else [] target = os.path.join(repos_dir, dirs[0]) if dirs else workspace_dir r = await _run_shell('git status --short', target, 10) r2 = await _run_shell('git log --oneline -3', target, 10) return f"Status:\n{r['stdout'][:1000] or '(clean)'}\nRecent:\n{r2['stdout'][:500]}" elif tool == 'git_commit_push': msg = p.get('message', '') or 'agent update' repo = p.get('repo_dir', '') repos_dir = os.path.join(workspace_dir, 'repos') if repo: target = os.path.join(workspace_dir, repo) else: dirs = [d for d in os.listdir(repos_dir) if os.path.isdir(os.path.join(repos_dir, d, '.git'))] if os.path.isdir(repos_dir) else [] target = os.path.join(repos_dir, dirs[0]) if dirs else workspace_dir r = await _git_commit_push(target, msg) return f"rc={r['rc']} {r['message']}" return f'未实现: {tool}' except Exception as e: return f'ERROR: {str(e)[:300]}' # ── 角色 Agent ── async def role_agent_run(project_id, role, agent_id=None, model_name=None): role = _normalize_role(role) if role == 'pm': return {"status": "idle", "message": "PM请使用 pm_review_run"} db = _get_db() async with db.sqlorContext("pipeline") as sor: task = await _claim_task(sor, project_id, role) if not task: return {"status": "idle", "message": "没有待办任务"} task_id = task.id title = getattr(task, "title", "") or "" params_str = getattr(task, "params", "{}") or "{}" workspace_dir = await _get_workspace_dir(sor, project_id) try: await sor.sqlExe("COMMIT", {}) except Exception: pass repos_dir = os.path.join(workspace_dir, 'repos') os.makedirs(repos_dir, exist_ok=True) role_specific = ROLE_SPECIFICS.get(role, ROLE_SPECIFICS.get('develop', '')) qna_section = await _build_qna_section(sor, task_id) tools_text = json.dumps(AGENT_TOOLS, ensure_ascii=False) system = (AGENT_SYSTEM_PROMPT .replace('__ROLE__', role) .replace('__TITLE__', title) .replace('__QNA__', qna_section) .replace('__WORKSPACE__', workspace_dir) .replace('__ROLE_SPECIFIC__', role_specific) .replace('__TOOLS__', tools_text)) msgs = [{"role": "system", "content": system}] msgs.append({"role": "user", "content": f"执行任务：{title}\n参数：{params_str}"}) from .llm_bridge import llm_call_msgs deliverable = None ask_question = None # ── Tool Loop ── for turn in range(15): try: raw = await llm_call_msgs(msgs, model=model_name, temperature=0.4) except Exception as e: await sor.sqlExe("UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id}) logger.error(f"role_agent_run llm failed: task={task_id} err={e}") return {"status": "failed", "task_id": task_id, "error": str(e)[:200]} act = _parse_agent_action(raw) if act.get('action') == 'deliver': deliverable = act break elif act.get('action') == 'ask': ask_question = act.get('question', '') break elif act.get('action') == 'tool_call': tool = act.get('tool', '') params = act.get('params', {}) result = await _exec_agent_tool(tool, params, workspace_dir) msgs.append({"role": "assistant", "content": raw}) msgs.append({"role": "user", "content": f"工具 {tool} 结果：\n{result}"}) logger.info(f"role_agent tool_call: {tool} -> {result[:100]}") else: deliverable = {"result": raw} break if ask_question: from .questions import agent_ask qid = await agent_ask(project_id, task_id, role, ask_question, context={"title": title}) return {"status": "need_info", "task_id": task_id, "question_id": qid, "question": ask_question} if not deliverable: deliverable = {"result": "Agent未产出交付件"} # ── 处理产出 ── from appPublic.uniqueID import getID did = getID() result_text = deliverable.get("result") or "" deliverable_type = deliverable.get("deliverable_type") or role summary = deliverable.get("summary", "") # 写入代码文件 files_written = [] code_files = deliverable.get("files") or [] if isinstance(code_files, list): for f in code_files: if isinstance(f, dict) and f.get("path") and f.get("content"): abs_path = os.path.join(workspace_dir, f["path"]) ok, msg = await _write_code_file(abs_path, f["content"]) if ok: files_written.append(abs_path) logger.info(f"code file written: {abs_path}") else: logger.error(f"code file failed: {abs_path} err={msg}") # 写交付件文档 file_path = '' if result_text: deliverable_dir = os.path.join(workspace_dir, 'deliverables', role) os.makedirs(deliverable_dir, exist_ok=True) safe_type = (deliverable_type or 'deliverable').replace('/', '_') file_path = os.path.join(deliverable_dir, f"{task_id}_{safe_type}.md") try: with open(file_path, 'w', encoding='utf-8') as f: f.write(result_text or '') except Exception as e: logger.error(f"deliverable write failed: {file_path} err={e}") file_path = '' await sor.C("pipeline_deliverables", { "id": did, "project_id": project_id, "task_id": task_id, "deliverable_type": deliverable_type, "title": title, "content": result_text, "file_path": file_path, "quality_score": 80, "review_status": "pending", "created_by": agent_id or role, }) # Git commit + push commit_msg = deliverable.get("git_commit_message") or f"{role}: {title[:80]}" git_result = {"rc": 0, "message": "无git操作"} if files_written: for repo_name in os.listdir(repos_dir): repo_path = os.path.join(repos_dir, repo_name) if os.path.isdir(os.path.join(repo_path, '.git')): git_result = await _git_commit_push(repo_path, commit_msg) await sor.sqlExe( "UPDATE pipeline_tasks SET state=${st}$, claimed_by=NULL WHERE id=${tid}$", {"st": TASK_REVIEW, "tid": task_id}) logger.info(f"role_agent_run completed: task={task_id} role={role} " f"deliverable={did} files={len(files_written)} git={git_result.get('rc',-1)}") return {"status": "completed", "task_id": task_id, "deliverable_id": did, "files_written": len(files_written), "git_result": git_result} # ── PM 审核 ── async def pm_review_run(project_id, agent_id=None, model_name=None): db = _get_db() async with db.sqlorContext("pipeline") as sor: task = await _claim_task(sor, project_id, '', state=TASK_REVIEW, match_role=False) if not task: return {"status": "idle", "message": "没有待审核任务"} task_id = task.id title = getattr(task, "title", "") or "" task_role = _normalize_role(getattr(task, "role", "") or "") workspace_dir = await _get_workspace_dir(sor, project_id) try: await sor.sqlExe("COMMIT", {}) except Exception: pass repos = await _get_project_repos(sor, project_id) if repos: await _setup_repos(sor, workspace_dir, project_id) deliverable_content, deliverable_type = await _get_deliverable_content(sor, task_id) if not deliverable_content: await sor.sqlExe("UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$", {"tid": task_id}) return {"status": "rejected", "task_id": task_id, "reason": "没有交付件"} repo_state = await _get_repo_state(workspace_dir) repos_str = ", ".join([r['name'] for r in repos]) if repos else "无" content_preview = deliverable_content[:6000] pm_system = PM_SYSTEM_PROMPT.replace('__TITLE__', title)\ .replace('__ROLE__', task_role)\ .replace('__WORKSPACE__', workspace_dir)\ .replace('__REPOS__', repos_str)\ .replace('__REPO_STATE__', repo_state) msgs = [{"role": "system", "content": pm_system}] msgs.append({"role": "user", "content": f"请审核以下交付件（类型:{deliverable_type}）：\n\n{content_preview}"}) from .llm_bridge import llm_call_msgs decision = None for turn in range(5): try: raw = await llm_call_msgs(msgs, model=model_name, temperature=0.3) except Exception as e: await sor.sqlExe("UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id}) return {"status": "failed", "task_id": task_id, "error": str(e)[:200]} act = _parse_agent_action(raw) if act.get('action') == 'review_approve': decision = {'status': 'approved', 'comment': act.get('comment', ''), 'next_title': act.get('next_task_title', ''), 'next_desc': act.get('next_task_description', '')} break elif act.get('action') == 'review_reject': decision = {'status': 'rejected', 'comment': act.get('comment', ''), 'questions': act.get('questions', '')} break elif act.get('action') == 'review_complete': decision = {'status': 'completed', 'comment': act.get('comment', '')} break elif act.get('action') == 'tool_call': tool = act.get('tool', '') params = act.get('params', {}) result = await _exec_agent_tool(tool, params, workspace_dir) msgs.append({"role": "assistant", "content": raw}) msgs.append({"role": "user", "content": f"工具 {tool} 结果：\n{result}"}) else: decision = {'status': 'completed', 'comment': raw[:200]} break if not decision: decision = {'status': 'rejected', 'comment': '审核超时'} status = decision['status'] comment = decision.get('comment', '') if status == 'approved': from appPublic.uniqueID import getID pm_did = getID() await sor.C("pipeline_deliverables", { "id": pm_did, "project_id": project_id, "task_id": task_id, "deliverable_type": "pm_review", "title": f"PM审核：{title}", "content": json.dumps(decision, ensure_ascii=False), "file_path": os.path.join(workspace_dir, 'deliverables', 'pm', f"{task_id}_review.md"), "quality_score": 100, "review_status": "approved", "created_by": agent_id or "pm", }) await sor.sqlExe("UPDATE pipeline_deliverables SET review_status='approved', review_comment=${cm}$ WHERE task_id=${tid}$", {"cm": comment, "tid": task_id}) await sor.sqlExe("UPDATE pipeline_tasks SET state=${st}$ WHERE id=${tid}$", {"st": TASK_APPROVED, "tid": task_id}) next_role = await _get_next_role(task_role) if next_role: next_tid, next_title = await _create_next_task(sor, project_id, task, next_role, comment) return {"status": "approved", "task_id": task_id, "next_task_id": next_tid, "next_role": next_role, "comment": comment} else: return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"} elif status == 'rejected': from .questions import agent_ask rejection_q = decision.get("questions") or comment or "交付件不满足要求" await agent_ask(project_id, task_id, "pm", rejection_q, context={"pm_comment": comment, "deliverable_type": deliverable_type}) await sor.sqlExe("UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$", {"tid": task_id}) return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q} else: await sor.sqlExe("UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$", {"tid": task_id}) return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"} async def role_agent_loop(project_id, role, agent_id=None, model_name=None, max_iterations=10): results = [] for _ in range(max_iterations): r = await role_agent_run(project_id, role, agent_id, model_name) results.append(r) if r["status"] in ("idle", "need_info", "failed"): break await asyncio.sleep(1) return results async def pm_review_loop(project_id, agent_id=None, model_name=None, max_iterations=20): results = [] for _ in range(max_iterations): r = await pm_review_run(project_id, agent_id, model_name) results.append(r) if r["status"] in ("idle", "failed"): break await asyncio.sleep(1) return results # 向后兼容 async def run_agent_loop(project_id, role_name, model_name=None): role = _normalize_role(role_name or "develop") if role == 'pm': return await pm_review_run(project_id, model_name=model_name) return await role_agent_run(project_id, role_name or "develop", model_name=model_name) async def agent_loop(project_id, role_name, model_name=None, max_iterations=10): role = _normalize_role(role_name or "develop") if role == 'pm': return await pm_review_loop(project_id, model_name=model_name, max_iterations=max_iterations) return await role_agent_loop(project_id, role_name or "develop", model_name=model_name, max_iterations=max_iterations)