diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index e3582e0..c3f652d 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -166,7 +166,7 @@ async def role_agent_run(project_id: str, role: str, agent_id: str = None, model # 缺信息 → 提问回路:任务置 waiting,等主agent/客户回答 if parsed.get("status") == "need_info" and parsed.get("question"): - from pipeline_service.questions import agent_ask + 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, diff --git a/pipeline_service/init.py b/pipeline_service/init.py index 2cd6bc4..4bc29b6 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -8,6 +8,8 @@ """ import json +import asyncio +import logging from ahserver.serverenv import ServerEnv from appPublic.uniqueID import getID from appPublic.log import debug @@ -427,5 +429,35 @@ def load_pipeline_service(): env.question_list = list_questions env.question_qna = get_task_qna + # Background poller: auto-dispatch submitted role_tasks to role agents + _logger = logging.getLogger("pipeline.agent_poller") + + async def _role_poller(): + """轮询 pipeline_tasks 中待执行的 role_task,交给对应角色agent执行。""" + from sqlor.dbpools import DBPools + poll_db = DBPools() + while True: + try: + async with poll_db.sqlorContext("pipeline") as sor: + recs = await sor.sqlExe( + "SELECT id, tenant_id, role FROM pipeline_tasks " + "WHERE state='submitted' AND pipeline_id='role_task' " + "AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 5", + {}) + for rec in (recs or []): + tid = getattr(rec, 'id', '') + pid = getattr(rec, 'tenant_id', '') + r = getattr(rec, 'role', '') + if tid and pid and r: + _logger.info(f"dispatching task={tid} role={r}") + asyncio.ensure_future( + role_agent_loop(pid, r, agent_id=f"poller-{r}")) + except Exception as e: + _logger.error(f"error: {e}") + await asyncio.sleep(10) + + from ahserver.configuredServer import add_startup + add_startup(_role_poller()) + debug(f"[{MODULE_NAME}] v{MODULE_VERSION} loaded — pipeline engine with role-agent + question loop") return True