From 71d70c31745e4c4b8f9fbdfbe7d9484ad5b72e06 Mon Sep 17 00:00:00 2001 From: "Eunice(Eunyoung) Shin" Date: Sat, 4 Jul 2026 16:27:24 +0900 Subject: [PATCH] feat: run report generation worker loop inside cpu-worker (no new Deployment) cpu_worker.py already runs multiple SQS consumer loops as daemon threads in one process/pod (audio preprocess on the main thread, the legacy report-analysis loop on a background thread) - this adds a third thread for the new 5-agent report generation queue instead of standing up a separate Deployment/ScaledObject/IRSA role for it. Same queue-per-thread pattern already established here, reusing the pod's existing DB/Bedrock/ AgentCore Gateway network access and IAM permissions. report_generation_worker.run_loop() (renamed from the private _run_loop) is called directly rather than through start_worker(), since initialize_observability() should only run once per process and cpu_worker.py's own start_worker() already calls it. --- app/workers/cpu_worker.py | 10 ++++++++++ app/workers/report_generation_worker.py | 7 +++++-- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/app/workers/cpu_worker.py b/app/workers/cpu_worker.py index 18a5d11..d99a9bb 100644 --- a/app/workers/cpu_worker.py +++ b/app/workers/cpu_worker.py @@ -25,6 +25,7 @@ from app.models.vad_silero import SileroVADWrapper from app.models.embedding_kure import KUREEmbeddingWrapper from app.storage.rds import get_be_engine +from app.workers.report_generation_worker import run_loop as run_report_generation_loop def _extend_report_visibility(sqs, queue_url: str, receipt_handle: str, stop_event: threading.Event, interval: int = 240) -> None: @@ -207,6 +208,15 @@ def start_worker() -> None: ) report_thread.start() + # 5-Agent 리포트 생성 큐도 같은 pod/프로세스에서 별도 스레드로 폴링한다 - + # 전용 워커 Deployment를 새로 만들지 않고 기존 cpu-worker의 멀티 큐 구조에 + # 얹는다(위 report_thread와 동일 패턴). + report_generation_thread = threading.Thread( + target=run_report_generation_loop, + daemon=True, + ) + report_generation_thread.start() + _run_preprocess_loop(models) diff --git a/app/workers/report_generation_worker.py b/app/workers/report_generation_worker.py index 334bbc6..9affc39 100644 --- a/app/workers/report_generation_worker.py +++ b/app/workers/report_generation_worker.py @@ -37,7 +37,7 @@ def _make_sqs_client(): return boto3.client("sqs", region_name=settings.aws_region) -def _run_loop() -> None: +def run_loop() -> None: logger.info(f"Report Generation Worker 시작. 큐: {settings.sqs_report_generation_queue_url}") sqs = _make_sqs_client() @@ -136,8 +136,11 @@ def _run_loop() -> None: def start_worker() -> None: + """독립 실행용 진입점. cpu_worker.py처럼 이미 다른 워커와 같은 프로세스에서 + 같이 도는 경우에는 initialize_observability() 중복 호출을 피하기 위해 + run_loop()을 직접 스레드로 돌린다.""" initialize_observability() - _run_loop() + run_loop() if __name__ == "__main__":