yumoqing
4c0b349718
- _role_poller 每10秒扫描 pipeline_tasks 中 submitted 的 role_task - 逐条 dispatch 给对应角色 agent_loop 异步执行 - 修复 agent_loop 中 agent_ask 导入为相对导入
217 lines
8.9 KiB
Python
217 lines
8.9 KiB
Python
"""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',
|
||
}
|
||
|
||
|
||
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:
|
||
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 "{}"
|
||
|
||
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))
|
||
|
||
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)
|
||
|
||
# 缺信息 → 提问回路：任务置 waiting，等主agent/客户回答
|
||
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
|
||
await sor.C("pipeline_deliverables", {
|
||
"id": did, "project_id": project_id, "task_id": task_id,
|
||
"deliverable_type": parsed.get("deliverable_type") or role,
|
||
"title": title, "content": result_text,
|
||
"quality_score": 80, "review_status": "pending",
|
||
"created_by": agent_id or role,
|
||
})
|
||
await sor.sqlExe(
|
||
"UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$",
|
||
{"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 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 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
|
||
|
||
|
||
# ── 向后兼容别名（旧前端按 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)
|