From 4cb406003e15cea5863149edbe02153f66ac0753 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Wed, 5 Aug 2026 12:58:52 +0800 Subject: [PATCH] =?UTF-8?q?feat(v3.2.0):=20=E4=B8=BBagent+=E8=A7=92?= =?UTF-8?q?=E8=89=B2agent=E6=9E=B6=E6=9E=84=20=E2=80=94=20role=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E8=AE=A4=E9=A2=86/=E9=97=AE=E9=A2=98=E5=9B=9E?= =?UTF-8?q?=E8=B7=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pipeline_tasks 增加 role/claimed_by 字段+索引（存量库需ALTER） - agent_loop 重写为角色循环：原子认领(role匹配+令牌校验)→LLM执行→交付/need_info提问 - 角色别名归一化(developer→develop等)，ROLE_PROMPT改replace防花括号崩溃 - questions.py 新增问题回路：agent_ask/answer_question/forward_question/list_questions - init.py 注册 pipeline_role_submit 与问题回路 env 接口 --- models/pipeline_agent_questions.json | 32 ++++ models/pipeline_tasks.json | 5 +- pipeline_service/agent_loop.py | 218 ++++++++++++++++++++++----- pipeline_service/init.py | 61 +++++++- pipeline_service/questions.py | 190 +++++++++++++++++++++++ pipeline_service/storage.py | 10 +- 6 files changed, 470 insertions(+), 46 deletions(-) create mode 100644 models/pipeline_agent_questions.json create mode 100644 pipeline_service/questions.py diff --git a/models/pipeline_agent_questions.json b/models/pipeline_agent_questions.json new file mode 100644 index 0000000..5e1f55f --- /dev/null +++ b/models/pipeline_agent_questions.json @@ -0,0 +1,32 @@ +{ + "summary": [ + { + "name": "pipeline_agent_questions", + "title": "角色agent问题回路表", + "primary": ["id"], + "catelog": "entity" + } + ], + "fields": [ + {"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "tenant_id", "title": "租户ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "task_id", "title": "任务ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "from_role", "title": "提问角色", "type": "str", "length": 32, "nullable": "no"}, + {"name": "question", "title": "问题内容", "type": "text"}, + {"name": "context", "title": "问题上下文(JSON)", "type": "text"}, + {"name": "answer", "title": "回答内容", "type": "text"}, + {"name": "answered_by", "title": "回答人", "type": "str", "length": 64, "nullable": "yes"}, + {"name": "answer_source", "title": "回答来源", "type": "str", "length": 16, "nullable": "yes"}, + {"name": "status", "title": "问题状态", "type": "str", "length": 32, "nullable": "no", "default": "pending"}, + {"name": "created_at", "title": "创建时间", "type": "timestamp", "nullable": "no"}, + {"name": "updated_at", "title": "更新时间", "type": "timestamp", "nullable": "no"} + ], + "indexes": [ + {"name": "idx_paq_tenant", "idxtype": "index", "idxfields": ["tenant_id"]}, + {"name": "idx_paq_task", "idxtype": "index", "idxfields": ["task_id"]}, + {"name": "idx_paq_status", "idxtype": "index", "idxfields": ["status"]} + ], + "codes": [ + {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='question_state'"} + ] +} diff --git a/models/pipeline_tasks.json b/models/pipeline_tasks.json index f8bf983..6da9e0d 100644 --- a/models/pipeline_tasks.json +++ b/models/pipeline_tasks.json @@ -16,6 +16,8 @@ {"name": "state", "title": "任务状态", "type": "str", "length": 32, "nullable": "no", "default": "submitted"}, {"name": "current_version", "title": "当前版本号", "type": "int", "nullable": "no", "default": "1"}, {"name": "params", "title": "提交参数", "type": "text"}, + {"name": "role", "title": "目标角色", "type": "str", "length": 32, "nullable": "no", "default": ""}, + {"name": "claimed_by", "title": "认领agent标识", "type": "str", "length": 64, "nullable": "yes"}, {"name": "created_at", "title": "创建时间", "type": "timestamp", "nullable": "no"}, {"name": "updated_at", "title": "更新时间", "type": "timestamp", "nullable": "no"} ], @@ -23,7 +25,8 @@ {"name": "idx_pt_tenant", "idxtype": "index", "idxfields": ["tenant_id"]}, {"name": "idx_pt_pipeline", "idxtype": "index", "idxfields": ["pipeline_id"]}, {"name": "idx_pt_owner", "idxtype": "index", "idxfields": ["owner_id"]}, - {"name": "idx_pt_state", "idxtype": "index", "idxfields": ["state"]} + {"name": "idx_pt_state", "idxtype": "index", "idxfields": ["state"]}, + {"name": "idx_pt_role", "idxtype": "index", "idxfields": ["role"]} ], "codes": [ {"field": "state", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='task_state'"} diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 2cbfbf5..e3582e0 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -1,74 +1,216 @@ -"""Agent loop — fetches tasks, executes with LLM, delivers results.""" +"""Agent loop — 角色agent循环认领执行任务（v3.2.0）。 + +架构： +- 主agent（cockpit_chat）理解客户意图，开发类意图写入 pipeline_tasks（带 role）。 +- 各角色agent（需求/设计/开发/测试/运维）调 role_agent_run/role_agent_loop + 循环认领自己角色的任务并执行。 +- 认领是原子的：UPDATE ... WHERE state='submitted' + claimed_by 令牌校验， + 多agent并发不会重复认领。 +- 角色agent缺信息时返回 need_info → 写入 pipeline_agent_questions， + 任务置 waiting，等主agent/客户回答后回到 submitted 重新认领继续。 +- 只认领没有步骤记录的任务（带步骤的 DAG 任务归 executor 管）。 +""" + import asyncio import json import logging logger = logging.getLogger("pipeline.agent_loop") +# 角色别名归一化：前端/主agent传的角色名统一到认领用的规范名 +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', +} -async def run_agent_loop(project_id, role_name, model_name=None): - """Single iteration: fetch pending task, execute, deliver.""" + +def _normalize_role(role): + r = (role or '').strip().lower() + return ROLE_ALIASES.get(r, r) + + +ROLE_PROMPT = """你是软件开发产线项目中的「__ROLE__」角色agent。 + +当前认领到的任务： +标题：__TITLE__ +参数：__PARAMS__ +__QNA__ +请站在你的角色立场完成这项工作。 + +输出要求——返回纯 JSON（不要 markdown 包裹）： +- 能完成时：{"status":"done","deliverable_type":"...","result":"完整交付内容"} +- 缺少关键信息无法继续时：{"status":"need_info","question":"要向主agent/客户提出的具体问题","partial":"已完成的部分"}""" + + +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 + + +async def _claim_task(sor, tenant_id: str, role: str): + """原子认领一条符合角色的 submitted 任务。返回 task record 或 None。 + + 认领策略： + 1. 选出最早一条 state='submitted' 且 role 匹配（或 role 为空）的任务， + 且排除有步骤记录的 DAG 任务（那些归 executor）。 + 2. UPDATE ... SET state='running', claimed_by=令牌 WHERE id=? AND state='submitted' + 3. 用令牌回查校验——别的agent抢先时校验失败，返回 None。 + """ + role = _normalize_role(role) + recs = await sor.sqlExe( + "SELECT id, title, params, pipeline_id, tenant_id, role FROM pipeline_tasks " + "WHERE tenant_id=${tid}$ AND state='submitted' AND (role=${role}$ OR 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, "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='submitted'", + {"cb": claim_token, "tid": task_id}) + + # 令牌校验：确认是本agent抢到的 + 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} role={role}") + return None + return task + + +async def _build_qna_section(sor, task_id: str) -> str: + """把该任务已回答的问答历史注入prompt，让角色agent带着答案继续。""" + from .questions import get_task_qna + qna = await get_task_qna(task_id) + if not qna: + return "" + lines = ["历史问答（这些问题已得到回答，请结合答案继续工作）："] + for item in qna: + src = item.get('answer_source') or '' + src_label = "客户" if src == "customer" else "主agent" + lines.append(f"问：{item.get('question', '')}") + lines.append(f"答（{src_label}）：{item.get('answer', '')}") + return "\n".join(lines) + + +def _parse_result(raw: str) -> dict: + """解析角色agent的 LLM 返回。解析失败时按 done + 原文处理。""" + 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 role_agent_run(project_id: str, role: str, agent_id: str = None, model_name: str = None) -> dict: + """角色agent单次迭代：认领一条任务 → 执行 → 交付或提问。 + + Returns: + {"status": "idle"} 没有待办任务 + {"status": "completed", "task_id", "deliverable_id"} 完成交付 + {"status": "need_info", "task_id", "question_id", "question"} 缺信息已提问 + {"status": "failed", "task_id", "error"} 执行失败 + """ + role = _normalize_role(role) + db = _get_db() async with db.sqlorContext("pipeline") as sor: - recs = await sor.sqlExe( - "SELECT id, title, params, pipeline_id FROM pipeline_tasks " - "WHERE state='submitted' AND tenant_id=${pid}$ LIMIT 1", - {"pid": project_id}, - ) - if not recs: - return {"status": "idle", "message": "No pending tasks"} + task = await _claim_task(sor, project_id, role) + if not task: + return {"status": "idle", "message": "没有待办任务"} - task = recs[0] task_id = task.id - await sor.sqlExe( - "UPDATE pipeline_tasks SET state='running' WHERE id=${tid}$", - {"tid": task_id}, - ) + title = getattr(task, "title", "") or "" + params_str = getattr(task, "params", "{}") or "{}" - from pipeline_service.llm_bridge import llm_call + 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)) - title = getattr(task, "title", "") - params_str = getattr(task, "params", "{}") - prompt = ( - f"Execute this development task:\n" - f"Title: {title}\n" - f"Params: {params_str}\n\n" - f"Provide the result." - ) + from .llm_bridge import llm_call try: - result = await llm_call(prompt, model=model_name) + 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}, - ) + {"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]} - from appPublic.uniqueID import getID + parsed = _parse_result(raw) + # 缺信息 → 提问回路：任务置 waiting，等主agent/客户回答 + if parsed.get("status") == "need_info" and parsed.get("question"): + from pipeline_service.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 await sor.C("pipeline_deliverables", { "id": did, "project_id": project_id, "task_id": task_id, - "deliverable_type": "code", "title": title, "content": result, + "deliverable_type": parsed.get("deliverable_type") or role, + "title": title, "content": result_text, "quality_score": 80, "review_status": "pending", - "created_by": role_name, + "created_by": agent_id or role, }) await sor.sqlExe( "UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$", - {"tid": task_id}, - ) + {"tid": task_id}) + logger.info(f"role_agent_run completed: task={task_id} role={role} deliverable={did}") return {"status": "completed", "task_id": task_id, "deliverable_id": did} -async def agent_loop(project_id, role_name, model_name=None, max_iterations=10): - """Run agent loop until idle or max reached.""" +async def role_agent_loop(project_id: str, role: str, agent_id: str = None, + model_name: str = None, max_iterations: int = 10) -> list: + """角色agent循环：连续认领执行，直到没有任务、需要提问或达到上限。""" results = [] for _ in range(max_iterations): - r = await run_agent_loop(project_id, role_name, model_name) + r = await role_agent_run(project_id, role, agent_id, model_name) results.append(r) - if r["status"] == "idle": + if r["status"] in ("idle", "need_info", "failed"): break - await asyncio.sleep(2) + await asyncio.sleep(1) return results + + +# ── 向后兼容别名（旧前端按 run_agent_loop/agent_loop 调用）── + +async def run_agent_loop(project_id, role_name, model_name=None): + """旧接口：单次迭代。role_name 为空时按 'dev' 处理。""" + return await role_agent_run(project_id, role_name or "dev", model_name=model_name) + + +async def agent_loop(project_id, role_name, model_name=None, max_iterations=10): + """旧接口：循环执行。role_name 为空时按 'dev' 处理。""" + return await role_agent_loop(project_id, role_name or "dev", + model_name=model_name, max_iterations=max_iterations) diff --git a/pipeline_service/init.py b/pipeline_service/init.py index e480082..2cd6bc4 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -30,9 +30,14 @@ from .step_registry import ( unregister_step_type, load_builtin_types, ) from .human import human_complete, approval_approve, approval_reject, human_list +from .questions import ( + agent_ask, answer_question, forward_question, + get_question, list_questions, get_task_qna, +) +from .agent_loop import role_agent_run, role_agent_loop, agent_loop, run_agent_loop MODULE_NAME = "pipeline_service" -MODULE_VERSION = "3.1.0" +MODULE_VERSION = "3.2.0" async def pipeline_submit(tenant_id, pipeline_id, owner_id, title, params=None): @@ -77,6 +82,40 @@ async def pipeline_submit(tenant_id, pipeline_id, owner_id, title, params=None): return json.dumps(result, ensure_ascii=False) +async def pipeline_role_submit(tenant_id, pipeline_id, owner_id, title, params=None, role=""): + """提交角色任务（主agent → 角色agent 模式，v3.2.0）。 + + 与 pipeline_submit 的区别： + - 带 role 字段，由角色agent循环（role_agent_loop）认领执行 + - 不创建步骤记录、不启动 DAG executor + + Args: + role: 目标角色（requirement/design/dev/test/ops 等）。必填。 + + Returns: + JSON string with success, task_id + """ + result = {"success": False} + try: + if not tenant_id: + result["message"] = "缺少 tenant_id" + return json.dumps(result, ensure_ascii=False) + if not role: + result["message"] = "缺少 role（角色任务必须指定目标角色）" + return json.dumps(result, ensure_ascii=False) + + params = params or {} + task_id = await create_task(tenant_id, pipeline_id or "", owner_id, title, params, role=role) + + result["success"] = True + result["task_id"] = task_id + result["role"] = role + result["message"] = f"角色任务已提交（role={role}），等待角色agent认领执行" + except Exception as e: + result["message"] = str(e) + return json.dumps(result, ensure_ascii=False) + + async def pipeline_list(tenant_id, pipeline_id=None, limit=100): """查询租户的任务列表。""" result = {"success": False} @@ -337,6 +376,7 @@ def load_pipeline_service(): # Task lifecycle env.pipeline_submit = pipeline_submit + env.pipeline_role_submit = pipeline_role_submit env.pipeline_list = pipeline_list env.pipeline_detail = pipeline_detail env.pipeline_node = pipeline_node @@ -370,11 +410,22 @@ def load_pipeline_service(): # Register intent classifier + LLM bridge (shared across all pipelines) from .intent_classifier import intent_classify from .llm_bridge import llm_call - from .agent_loop import agent_loop, run_agent_loop env.intent_classify = intent_classify env.pipeline_llm_call = llm_call - env.agent_loop = agent_loop - env.run_agent_loop = run_agent_loop - debug(f"[{MODULE_NAME}] v{MODULE_VERSION} loaded — pipeline engine with human-in-the-loop support") + # Role-agent loop (v3.2.0: 主agent + 角色agent 认领执行) + env.role_agent_run = role_agent_run + env.role_agent_loop = role_agent_loop + env.agent_loop = agent_loop # 向后兼容 + env.run_agent_loop = run_agent_loop # 向后兼容 + + # Question loop (角色agent ↔ 主agent ↔ 客户) + env.agent_ask = agent_ask + env.question_answer = answer_question + env.question_forward = forward_question + env.question_detail = get_question + env.question_list = list_questions + env.question_qna = get_task_qna + + debug(f"[{MODULE_NAME}] v{MODULE_VERSION} loaded — pipeline engine with role-agent + question loop") return True diff --git a/pipeline_service/questions.py b/pipeline_service/questions.py new file mode 100644 index 0000000..611cf0a --- /dev/null +++ b/pipeline_service/questions.py @@ -0,0 +1,190 @@ +"""问题回路 — 角色agent缺信息时提问，主agent路由回答。 + +流程： +1. 角色agent执行任务时发现缺关键信息 → agent_ask() 写入 pipeline_agent_questions + (status=pending)，任务置 waiting，退出本轮执行。 +2. 主agent轮询 pending 问题： + - 能直接回答 → answer_question(answer_source='main_agent')，任务回 submitted， + 角色agent下轮重新认领并带着问答历史继续。 + - 答不了 → forward_question() 转客户 (status=forwarded)； + 客户在会话中回答后，主agent调 answer_question(answer_source='customer')。 +""" + +import json +import logging +from sqlor.dbpools import DBPools +from appPublic.uniqueID import getID + +DBNAME = "pipeline" +logger = logging.getLogger("pipeline.questions") + +# 问题状态 +Q_PENDING = "pending" # 待主agent回答 +Q_FORWARDED = "forwarded" # 主agent已转客户，等客户回复 +Q_ANSWERED = "answered" # 已回答 + + +def _get_db(): + db = DBPools() + if not db.databases: + from appPublic.jsonConfig import getConfig + config = getConfig() + if config.databases: + db.databases = config.databases + return db, DBNAME + + +async def agent_ask(tenant_id: str, task_id: str, from_role: str, question: str, context: dict = None) -> str: + """角色agent提问。任务置 waiting，等待主agent/客户回答。返回 question_id。""" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + qid = getID() + ctx_json = json.dumps(context, ensure_ascii=False, default=str) if context else None + await sor.C('pipeline_agent_questions', { + "id": qid, + "tenant_id": tenant_id or "", + "task_id": task_id or "", + "from_role": from_role or "", + "question": question or "", + "context": ctx_json, + "status": Q_PENDING, + }) + if task_id: + await sor.U('pipeline_tasks', {"id": task_id, "state": "waiting"}) + logger.info(f"agent_ask: task={task_id} role={from_role} qid={qid}") + return qid + + +async def answer_question(question_id: str, answer: str, answered_by: str = "", + answer_source: str = "main_agent", resume: bool = True): + """回答问题。任务从 waiting 回 submitted，让角色agent重新认领继续执行。 + + Args: + answer_source: 'main_agent' 或 'customer' + resume: 是否把任务恢复为 submitted + + Returns: + 更新后的问题记录 dict；问题不存在返回 None。 + """ + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + recs = await sor.R('pipeline_agent_questions', {'id': question_id}) + if not recs: + return None + rec = recs[0] + task_id = rec.task_id if hasattr(rec, 'task_id') else rec['task_id'] + + await sor.U('pipeline_agent_questions', { + "id": question_id, + "answer": answer or "", + "answered_by": answered_by or "", + "answer_source": answer_source or "main_agent", + "status": Q_ANSWERED, + }) + + resumed = False + if resume and task_id: + trecs = await sor.R('pipeline_tasks', {'id': task_id}) + if trecs: + t = trecs[0] + tstate = t.state if hasattr(t, 'state') else t['state'] + if tstate == "waiting": + await sor.U('pipeline_tasks', {"id": task_id, "state": "submitted"}) + resumed = True + + logger.info(f"answer_question: qid={question_id} task={task_id} source={answer_source} resumed={resumed}") + return { + "id": question_id, + "task_id": task_id, + "status": Q_ANSWERED, + "answer_source": answer_source, + "resumed": resumed, + } + + +async def forward_question(question_id: str): + """主agent答不了 → 转给客户。status: pending → forwarded。""" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + await sor.U('pipeline_agent_questions', {"id": question_id, "status": Q_FORWARDED}) + logger.info(f"forward_question: qid={question_id}") + + +async def get_question(question_id: str): + """取单条问题记录（含 context 解析）。""" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + recs = await sor.R('pipeline_agent_questions', {'id': question_id}) + if not recs: + return None + rec = recs[0] + if hasattr(rec, '__dict__'): + d = {k: getattr(rec, k) for k in dir(rec) if not k.startswith('_')} + else: + d = dict(rec) + ctx = d.get('context') + if ctx and isinstance(ctx, str): + try: + d['context'] = json.loads(ctx) + except (json.JSONDecodeError, TypeError): + pass + return d + + +async def list_questions(tenant_id: str = None, task_id: str = None, + status: str = None, limit: int = 50) -> list: + """问题列表，支持租户/任务/状态过滤。""" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + conditions = [] + params = {} + if tenant_id: + conditions.append("tenant_id=${tenant_id}$") + params["tenant_id"] = tenant_id + if task_id: + conditions.append("task_id=${task_id}$") + params["task_id"] = task_id + if status: + conditions.append("status=${status}$") + params["status"] = status + where = " AND ".join(conditions) if conditions else "1=1" + try: + limit = int(limit) + except (TypeError, ValueError): + limit = 50 + sql = f"SELECT * FROM pipeline_agent_questions WHERE {where} ORDER BY created_at DESC LIMIT {limit}" + recs = await sor.sqlExe(sql, params) + result = [] + for rec in (recs or []): + if hasattr(rec, '__dict__'): + d = {k: getattr(rec, k) for k in dir(rec) + if not k.startswith('_') and not callable(getattr(rec, k))} + else: + d = dict(rec) + ctx = d.get('context') + if ctx and isinstance(ctx, str): + try: + d['context'] = json.loads(ctx) + except (json.JSONDecodeError, TypeError): + pass + result.append(d) + return result + + +async def get_task_qna(task_id: str) -> list: + """取任务的全部已回答问答（按时间序），用于角色agent继续执行时注入prompt。""" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + recs = await sor.sqlExe( + "SELECT question, answer, answer_source, from_role FROM pipeline_agent_questions " + "WHERE task_id=${tid}$ AND status='answered' ORDER BY created_at ASC", + {"tid": task_id}) + result = [] + for rec in (recs or []): + result.append({ + "question": getattr(rec, 'question', '') or '', + "answer": getattr(rec, 'answer', '') or '', + "answer_source": getattr(rec, 'answer_source', '') or '', + "from_role": getattr(rec, 'from_role', '') or '', + }) + return result diff --git a/pipeline_service/storage.py b/pipeline_service/storage.py index fc3b50d..122ec86 100644 --- a/pipeline_service/storage.py +++ b/pipeline_service/storage.py @@ -47,8 +47,13 @@ async def get_pipeline_steps(pipeline_id: str) -> list: return result -async def create_task(tenant_id: str, pipeline_id: str, owner_id: str, title: str, params: dict) -> str: - """Create a new pipeline task. Returns task_id.""" +async def create_task(tenant_id: str, pipeline_id: str, owner_id: str, title: str, params: dict, role: str = "") -> str: + """Create a new pipeline task. Returns task_id. + + Args: + role: 目标角色（需求/设计/开发/测试/运维）。空串=不限角色（兼容旧任务）。 + 带角色的任务由角色agent循环认领执行，不带步骤记录。 + """ db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: task_id = getID() @@ -61,6 +66,7 @@ async def create_task(tenant_id: str, pipeline_id: str, owner_id: str, title: st "state": "submitted", "current_version": 1, "params": json.dumps(params, ensure_ascii=False, default=str), + "role": role or "", } await sor.C('pipeline_tasks', data) return task_id