74ce3339d2
- All agents can write code files via 'files' array in output JSON - Auto git commit+push after code file creation - PM agent clones project repos on first review - PM review shows repository state - shell_exec now allows ~/pipeline_ws - Role-specific prompts guide code/file output
633 lines
26 KiB
Python
633 lines
26 KiB
Python
"""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 ──
|
||
|
||
ROLE_PROMPT = """你是软件开发产线项目中的「__ROLE__」角色agent。
|
||
|
||
当前任务：__TITLE__
|
||
参数：__PARAMS__
|
||
__QNA__
|
||
|
||
## 工作环境
|
||
- 项目根目录：__WORKSPACE__
|
||
- 代码仓库位于：__WORKSPACE__/repos/ 下各子目录
|
||
- 你有能力读写文件、执行git操作
|
||
|
||
## 产出要求
|
||
__ROLE_SPECIFIC__
|
||
|
||
## 输出格式（严格JSON，不要markdown包裹）
|
||
{
|
||
"status": "done",
|
||
"deliverable_type": "类型(requirement_doc/design_doc/code_files/test_report/deploy_doc)",
|
||
"summary": "一段话概述产出",
|
||
"result": "当产出是文档时，这里是完整文档内容",
|
||
"files": [
|
||
{"path": "repos/仓库名/src/相对路径/文件.java", "content": "文件完整内容"},
|
||
{"path": "repos/仓库名/pom.xml", "content": "..."}
|
||
],
|
||
"git_commit_message": "feat: 简短描述本次变更",
|
||
"need_more_info": false,
|
||
"question": ""
|
||
}
|
||
|
||
- 如果是文档类产出(requirement/design)，用 result 字段输出文档正文
|
||
- 如果是代码产出(develop)，必须用 files 数组输出每个源码文件，每个元素含 path 和 content
|
||
- 测试产出用 result 输出测试报告，如有测试脚本则用 files
|
||
- 部署产出用 files 输出部署配置/脚本，用 result 输出部署说明
|
||
- 缺少关键信息时：{"status":"need_info","question":"问题"}
|
||
|
||
## Git 操作说明
|
||
- 产出代码文件将自动写入 files 中指定的路径并 git commit + push
|
||
- 你只需要在 files 中指定正确的仓库路径即可"""
|
||
|
||
ROLE_SPECIFICS = {
|
||
'requirement': """你是需求分析师。输出完整的需求规格文档（Markdown），用 result 字段。
|
||
文档应包含：项目概述、功能需求、非功能需求、用户角色、业务流程、验收标准。""",
|
||
|
||
'design': """你是系统设计师。输出完整的设计文档（Markdown），用 result 字段。
|
||
文档应包含：系统架构、技术选型、数据库设计(ER图描述)、API接口设计、前端页面规划、安全设计。
|
||
同时可输出 proto/接口定义文件到 files。""",
|
||
|
||
'develop': """你是开发工程师。你必须输出实际可运行的代码文件！
|
||
用 files 数组输出每个源码文件，准确指定在仓库中的路径。
|
||
files 中的 path 格式：repos/{仓库名}/src/... 或 repos/{仓库名}/pom.xml 等。
|
||
每个文件包含完整可运行的代码。
|
||
同时用 result 字段输出代码说明。""",
|
||
|
||
'test': """你是测试工程师。输出测试报告（Markdown）到 result 字段。
|
||
如有测试脚本/用例代码，输出到 files。
|
||
报告应包含：测试策略、测试用例、执行结果、Bug清单、覆盖率。""",
|
||
|
||
'deploy': """你是部署运维工程师。输出部署文档（Markdown）到 result 字段。
|
||
必须输出部署配置/脚本到 files：Dockerfile/docker-compose.yml/nginx配置等。
|
||
files 中的 path 格式：repos/{仓库名}/deploy/...""",
|
||
}
|
||
|
||
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：部署配置完整、可一键部署"""
|
||
|
||
|
||
# ── 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 ──
|
||
|
||
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)
|
||
|
||
# 确保 repos 目录存在
|
||
repos_dir = os.path.join(workspace_dir, 'repos')
|
||
if not os.path.isdir(repos_dir):
|
||
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)
|
||
|
||
prompt = (ROLE_PROMPT
|
||
.replace('__ROLE__', role)
|
||
.replace('__TITLE__', title)
|
||
.replace('__PARAMS__', params_str)
|
||
.replace('__QNA__', qna_section)
|
||
.replace('__WORKSPACE__', workspace_dir)
|
||
.replace('__ROLE_SPECIFIC__', role_specific))
|
||
|
||
from .llm_bridge import llm_call
|
||
try:
|
||
raw = await llm_call(prompt, 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]}
|
||
|
||
parsed = _parse_result(raw)
|
||
|
||
if parsed.get("status") == "need_info" and parsed.get("question"):
|
||
from .questions import agent_ask
|
||
qid = await agent_ask(project_id, task_id, role, parsed["question"],
|
||
context={"partial": parsed.get("partial", ""), "title": title})
|
||
return {"status": "need_info", "task_id": task_id, "question_id": qid, "question": parsed["question"]}
|
||
|
||
# ── 处理产出 ──
|
||
from appPublic.uniqueID import getID
|
||
did = getID()
|
||
result_text = parsed.get("result") or raw
|
||
deliverable_type = parsed.get("deliverable_type") or role
|
||
summary = parsed.get("summary", "")
|
||
|
||
# 写入实际代码文件（如果有 files 数组）
|
||
files_written = []
|
||
code_files = parsed.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 '')
|
||
logger.info(f"deliverable written: {file_path}")
|
||
except Exception as e:
|
||
logger.error(f"deliverable write failed: {file_path} err={e}")
|
||
file_path = ''
|
||
|
||
# 写DB
|
||
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 = parsed.get("git_commit_message") or f"{role}: {title[:80]}"
|
||
git_result = {"rc": 0, "message": "无git操作"}
|
||
if files_written:
|
||
# 在 repos 目录下的每个 git 仓库各自 commit
|
||
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)
|
||
|
||
# 置为 review 状态
|
||
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)
|
||
|
||
# 首次审核时，确保仓库已 clone
|
||
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": "没有交付件"}
|
||
|
||
summary = deliverable_content[:200].replace('\n', ' ')
|
||
content_preview = deliverable_content[:6000]
|
||
if len(deliverable_content) > 6000:
|
||
content_preview += "\n...（截断）"
|
||
|
||
repo_state = await _get_repo_state(workspace_dir)
|
||
repos_str = ", ".join([r['name'] for r in repos]) if repos else "无"
|
||
|
||
prompt = (PM_PROMPT
|
||
.replace('__TITLE__', title)
|
||
.replace('__ROLE__', task_role)
|
||
.replace('__SUMMARY__', summary)
|
||
.replace('__DELIVERABLE__', content_preview)
|
||
.replace('__WORKSPACE__', workspace_dir)
|
||
.replace('__REPOS__', repos_str)
|
||
.replace('__REPO_STATE__', repo_state))
|
||
|
||
from .llm_bridge import llm_call
|
||
try:
|
||
raw = await llm_call(prompt, 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]}
|
||
|
||
parsed = _parse_result(raw)
|
||
decision = parsed.get("status", "rejected")
|
||
comment = parsed.get("comment", "")
|
||
|
||
if decision == "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(parsed, 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 decision == "rejected":
|
||
from .questions import agent_ask
|
||
rejection_q = parsed.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)
|