From 59e56b5e9caf84bddbcc60bf92eb25a2cfb5fd93 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Sun, 2 Aug 2026 14:59:32 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20Phase=202+4=20=E2=80=94=20agent=5Floop?= =?UTF-8?q?=20engine=20+=20project=20agent=20config?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_service/agent_loop.py | 74 ++++++++++++++++++++++++++++++++++ pipeline_service/init.py | 3 ++ 2 files changed, 77 insertions(+) create mode 100644 pipeline_service/agent_loop.py diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py new file mode 100644 index 0000000..2cbfbf5 --- /dev/null +++ b/pipeline_service/agent_loop.py @@ -0,0 +1,74 @@ +"""Agent loop — fetches tasks, executes with LLM, delivers results.""" +import asyncio +import json +import logging + +logger = logging.getLogger("pipeline.agent_loop") + + +async def run_agent_loop(project_id, role_name, model_name=None): + """Single iteration: fetch pending task, execute, deliver.""" + from sqlor.dbpools import DBPools + + db = DBPools() + 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 = recs[0] + task_id = task.id + await sor.sqlExe( + "UPDATE pipeline_tasks SET state='running' WHERE id=${tid}$", + {"tid": task_id}, + ) + + from pipeline_service.llm_bridge import llm_call + + 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." + ) + try: + result = await llm_call(prompt, model=model_name) + 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]} + + from appPublic.uniqueID import getID + + did = getID() + await sor.C("pipeline_deliverables", { + "id": did, "project_id": project_id, "task_id": task_id, + "deliverable_type": "code", "title": title, "content": result, + "quality_score": 80, "review_status": "pending", + "created_by": role_name, + }) + await sor.sqlExe( + "UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$", + {"tid": task_id}, + ) + 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.""" + results = [] + for _ in range(max_iterations): + r = await run_agent_loop(project_id, role_name, model_name) + results.append(r) + if r["status"] == "idle": + break + await asyncio.sleep(2) + return results diff --git a/pipeline_service/init.py b/pipeline_service/init.py index b2989b6..e480082 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -370,8 +370,11 @@ 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") return True