diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 3f98bd2..c6abc2b 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -162,47 +162,48 @@ async def _git_clone(repo_url, target_dir, branch='main'): # ── Prompts ── -ROLE_PROMPT = """你是软件开发产线项目中的「__ROLE__」角色agent。 +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":"问题"}}, +] -当前任务：__TITLE__ -参数：__PARAMS__ +AGENT_SYSTEM_PROMPT = """你是软件开发产线中的「__ROLE__」角色Agent。 + +## 任务 +__TITLE__ __QNA__ ## 工作环境 -- 项目根目录：__WORKSPACE__ -- 代码仓库位于：__WORKSPACE__/repos/ 下各子目录 -- 你有能力读写文件、执行git操作 +工作空间：__WORKSPACE__ +产出要求：__ROLE_SPECIFIC__ -## 产出要求 -__ROLE_SPECIFIC__ +## 工具 +你可以使用以下工具完成工作： +__TOOLS__ -## 开发规范（必须严格遵循！） -__SKILLS__ +## 工作流 +1. 先用 read_file/list_files 了解现有代码 +2. 用 write_file 产出代码文件到 repos/ 下 +3. 用 run_shell 验证（编译/测试） +4. 用 git_commit_push 提交到远端 +5. 用 deliver 提交最终交付件 -## 输出格式（严格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": "" -} +## 输出格式（每次只输出一个JSON对象） +调工具： +{"action":"tool_call","tool":"工具名","params":{}} -- 如果是文档类产出(requirement/design)，用 result 字段输出文档正文 -- 如果是代码产出(develop)，必须用 files 数组输出每个源码文件，每个元素含 path 和 content -- 测试产出用 result 输出测试报告，如有测试脚本则用 files -- 部署产出用 files 输出部署配置/脚本，用 result 输出部署说明 -- 缺少关键信息时：{"status":"need_info","question":"问题"} +提交交付件： +{"action":"deliver","deliverable_type":"code_files","summary":"概述","result":"文档内容","files":[{"path":"repos/仓库/src/file.py","content":"代码"}],"git_commit_message":"feat: 描述"} -## Git 操作说明 -- 产出代码文件将自动写入 files 中指定的路径并 git commit + push -- 你只需要在 files 中指定正确的仓库路径即可""" +提问： +{"action":"ask","question":"问题"} + +注意：每次只输出一个JSON！收到工具结果后再决定下一步。""" ROLE_SPECIFICS = { 'requirement': """你是需求分析师。输出完整的需求规格文档（Markdown），用 result 字段。 @@ -391,6 +392,87 @@ async def _create_next_task(sor, project_id, task, next_role, pm_comment=''): 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): @@ -409,64 +491,81 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): params_str = getattr(task, "params", "{}") or "{}" workspace_dir = await _get_workspace_dir(sor, project_id) - # 提交 claim 事务，释放行锁 try: await sor.sqlExe("COMMIT", {}) except Exception: pass - # 确保 repos 目录存在 repos_dir = os.path.join(workspace_dir, 'repos') - if not os.path.isdir(repos_dir): - os.makedirs(repos_dir, exist_ok=True) + 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) - # 从 params 中提取 skills 文本 - skills_text = "无特定规范要求，请按行业最佳实践执行。" - try: - params_obj = json.loads(params_str) if isinstance(params_str, str) else params_str - if isinstance(params_obj, dict) and params_obj.get('skills'): - skills_text = str(params_obj['skills']).strip() - except (json.JSONDecodeError, TypeError, ValueError): - pass - - prompt = (ROLE_PROMPT + system = (AGENT_SYSTEM_PROMPT .replace('__ROLE__', role) .replace('__TITLE__', title) - .replace('__PARAMS__', params_str) .replace('__QNA__', qna_section) .replace('__WORKSPACE__', workspace_dir) .replace('__ROLE_SPECIFIC__', role_specific) - .replace('__SKILLS__', skills_text)) + .replace('__TOOLS__', tools_text)) - 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]} + msgs = [{"role": "system", "content": system}] + msgs.append({"role": "user", "content": f"执行任务：{title}\n参数：{params_str}"}) - parsed = _parse_result(raw) + from .llm_bridge import llm_call_msgs - if parsed.get("status") == "need_info" and parsed.get("question"): + 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, parsed["question"], - context={"partial": parsed.get("partial", ""), "title": title}) - return {"status": "need_info", "task_id": task_id, "question_id": qid, "question": parsed["question"]} + 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 = parsed.get("result") or raw - deliverable_type = parsed.get("deliverable_type") or role - summary = parsed.get("summary", "") + result_text = deliverable.get("result") or "" + deliverable_type = deliverable.get("deliverable_type") or role + summary = deliverable.get("summary", "") - # 写入实际代码文件（如果有 files 数组） + # 写入代码文件 files_written = [] - code_files = parsed.get("files") or [] + 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"): @@ -488,12 +587,10 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): 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, @@ -502,17 +599,15 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): "created_by": agent_id or role, }) - # Git commit + push（如果有产出文件） - commit_msg = parsed.get("git_commit_message") or f"{role}: {title[:80]}" + # Git commit + push + commit_msg = deliverable.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}) diff --git a/pipeline_service/llm_bridge.py b/pipeline_service/llm_bridge.py index 5faa9e7..ba90134 100644 --- a/pipeline_service/llm_bridge.py +++ b/pipeline_service/llm_bridge.py @@ -129,3 +129,34 @@ async def llm_call(prompt: str, model: str = None, temperature: float = 0.7) -> async def call_llm(tenant_id: str, prompt: str, model: str = None, temperature: float = 0.7) -> str: """SDLC handler interface — delegates to llm_call.""" return await llm_call(prompt, model=model, temperature=temperature) + + +async def llm_call_msgs(messages: list, model: str = None, temperature: float = 0.7) -> str: + """Call LLM with full message array (system/user/assistant).""" + import aiohttp + + cfg = await _get_model_config(model) + if cfg.get("api_key") and cfg.get("api_base"): + api_base = cfg["api_base"] + api_key = cfg["api_key"] + model_id = cfg.get("model_id") or model or "default" + else: + api_base = os.environ.get("LLM_API_BASE", "https://api.openai.com/v1") + api_key = os.environ.get("LLM_API_KEY", "") + model_id = model or os.environ.get("LLM_MODEL", "gpt-4o-mini") + + if not api_key: + raise ValueError("No LLM API configured") + + headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} + payload = {"model": model_id, "messages": messages, "temperature": temperature} + + url = api_base.rstrip("/") + "/chat/completions" + async with aiohttp.ClientSession() as session: + async with session.post(url, headers=headers, json=payload, + timeout=aiohttp.ClientTimeout(total=180)) as resp: + if resp.status != 200: + text = await resp.text() + raise ValueError(f"LLM API error {resp.status}: {text[:300]}") + data = await resp.json() + return data["choices"][0]["message"]["content"]