150 lines
4.9 KiB
Python
150 lines
4.9 KiB
Python
"""백그라운드 작업 디스패처.
|
|
|
|
TaskQueue를 폴링하여 에이전트를 실행한다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
from typing import Any
|
|
|
|
from agent.task_queue import PersistentTaskQueue
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Dispatcher:
|
|
"""백그라운드 작업 소비자."""
|
|
|
|
def __init__(
|
|
self,
|
|
task_queue: PersistentTaskQueue,
|
|
poll_interval: float = 2.0,
|
|
):
|
|
self._queue = task_queue
|
|
self._poll_interval = poll_interval
|
|
self._running = False
|
|
self._task: asyncio.Task | None = None
|
|
|
|
async def start(self) -> None:
|
|
"""백그라운드 폴링 루프를 시작한다."""
|
|
self._running = True
|
|
self._task = asyncio.create_task(self._poll_loop())
|
|
logger.info("Dispatcher started (poll_interval=%.1fs)", self._poll_interval)
|
|
|
|
async def stop(self) -> None:
|
|
"""폴링 루프를 중지한다."""
|
|
self._running = False
|
|
if self._task:
|
|
self._task.cancel()
|
|
try:
|
|
await self._task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
logger.info("Dispatcher stopped")
|
|
|
|
async def _poll_loop(self) -> None:
|
|
"""주기적으로 큐를 폴링한다."""
|
|
while self._running:
|
|
try:
|
|
await self._poll_once()
|
|
except Exception:
|
|
logger.exception("Dispatcher poll error")
|
|
await asyncio.sleep(self._poll_interval)
|
|
|
|
async def _poll_once(self) -> None:
|
|
"""큐에서 작업을 하나 꺼내 처리한다."""
|
|
task = await self._queue.dequeue()
|
|
if not task:
|
|
return
|
|
|
|
logger.info("Processing task %s (thread %s)", task["id"], task["thread_id"])
|
|
|
|
try:
|
|
result = await self._run_agent_for_task(task)
|
|
await self._queue.mark_completed(task["id"], result=result)
|
|
logger.info("Task %s completed successfully", task["id"])
|
|
except Exception as e:
|
|
logger.exception("Task %s failed", task["id"])
|
|
await self._queue.mark_failed(task["id"], error=str(e))
|
|
await self._notify_failure(task, str(e))
|
|
|
|
async def _run_agent_for_task(self, task: dict) -> dict[str, Any]:
|
|
"""작업에 대해 에이전트를 실행한다."""
|
|
from agent.server import get_agent
|
|
|
|
payload = task["payload"]
|
|
thread_id = task["thread_id"]
|
|
|
|
config = {
|
|
"configurable": {
|
|
"thread_id": thread_id,
|
|
"__is_for_execution__": True,
|
|
"repo": {
|
|
"owner": payload.get("repo_owner", os.environ.get("DEFAULT_REPO_OWNER", "quant")),
|
|
"name": payload.get("repo_name", os.environ.get("DEFAULT_REPO_NAME", "galaxis-po")),
|
|
},
|
|
},
|
|
"metadata": {},
|
|
}
|
|
|
|
await self._notify_start(task)
|
|
|
|
agent = await get_agent(config)
|
|
|
|
issue_number = payload.get("issue_number", 0)
|
|
message = payload.get("message", "")
|
|
title = payload.get("title", "")
|
|
|
|
if issue_number:
|
|
input_text = f"이슈 #{issue_number}: {title}\n\n{message}"
|
|
else:
|
|
input_text = message
|
|
|
|
result = await agent.ainvoke(
|
|
{"messages": [{"role": "human", "content": input_text}]},
|
|
config=config,
|
|
)
|
|
|
|
return {"status": "completed", "messages_count": len(result.get("messages", []))}
|
|
|
|
async def _notify_start(self, task: dict) -> None:
|
|
"""작업 시작 알림을 전송한다."""
|
|
payload = task["payload"]
|
|
issue_number = payload.get("issue_number", 0)
|
|
source = task["source"]
|
|
|
|
if source == "gitea" and issue_number:
|
|
try:
|
|
from agent.tools.gitea_comment import gitea_comment
|
|
|
|
await asyncio.to_thread(
|
|
gitea_comment,
|
|
message=f"작업을 시작합니다: {payload.get('title', '')}",
|
|
issue_number=issue_number,
|
|
)
|
|
except Exception:
|
|
logger.debug("Failed to post start comment to Gitea")
|
|
|
|
if source == "discord":
|
|
try:
|
|
from agent.tools.discord_reply import discord_reply
|
|
|
|
await asyncio.to_thread(
|
|
discord_reply,
|
|
message=f"작업을 시작합니다: {payload.get('message', '')[:100]}",
|
|
)
|
|
except Exception:
|
|
logger.debug("Failed to send start message to Discord")
|
|
|
|
async def _notify_failure(self, task: dict, error: str) -> None:
|
|
"""작업 실패 알림을 전송한다."""
|
|
try:
|
|
from agent.tools.discord_reply import discord_reply
|
|
|
|
await asyncio.to_thread(discord_reply, message=f"작업 실패: {error[:200]}")
|
|
except Exception:
|
|
logger.debug("Failed to send failure notification")
|