AI 봇 7개가 서로 통신하는 방법 — 파일 기반 이벤트 버스

문제: 봇들이 어떻게 통신하나

7개 봇이 각자 돌아갑니다. 서로 뭘 하고 있는지 어떻게 알까?

옵션들:

1. Redis/RabbitMQ: 강력하지만 서버 하나 더 띄워야 함
2. Telegram 그룹: 메시지 기록은 되지만 구조화 데이터 교환이 어려움
3. 공유 파일: 단순하고 의존성 없음 → 선택

---

구조

D:\bots\
├── goal_state.json   # 시스템 전체 목표/상태
├── events.jsonl      # 봇들이 남기는 이벤트 로그
└── task_queue.json   # 봇들이 처리할 작업 큐

---

goal_state.json

시스템 전체 상태의 진실의 원천(source of truth)입니다.

{
  "updated": "2026-05-27T10:00:00+09:00",
  "revenue": {
    "adsense_monthly_krw": 0,
    "pearl_saas_users": 0
  },
  "content": {
    "posts_published": 24,
    "posts_this_week": 20,
    "top_keyword": "JARVIS 자동화"
  },
  "system": {
    "bots_healthy": 6,
    "last_scan": "2026-05-27T09:30:00+09:00",
    "last_dream": "2026-05-27T01:02:00+09:00"
  },
  "gaps": ["Pearl Finder 품질 개선 필요"]
}

모든 봇이 이 파일을 읽고 씁니다. 수익이 올랐으면 revenue.adsense_monthly_krw 업데이트. 포스트가 발행됐으면 content.posts_published 증가.

---

events.jsonl

각 봇이 이벤트를 한 줄씩 추가합니다 (JSONL 형식).

{"ts": "2026-05-27T09:30:00+09:00", "source": "nim_bot", "event": "news_collected", "data": {"count": 12}}
{"ts": "2026-05-27T09:31:00+09:00", "source": "pearl_finder", "event": "post_generated", "data": {"slug": "..."}}
{"ts": "2026-05-27T10:00:00+09:00", "source": "gsc_monitor", "event": "gsc_updated", "data": {"clicks": 5}}

덮어쓰지 않고 계속 추가합니다. 봇들은 마지막으로 읽은 위치(커서)를 기억해서 새 이벤트만 처리합니다.

---

핵심 코드

import json
import threading
from pathlib import Path

EVENTS_FILE = Path("D:/bots/events.jsonl")
GOAL_STATE_FILE = Path("D:/bots/goal_state.json")

_STATE_LOCK = threading.RLock()  # 다중 스레드 안전

def emit(source: str, event: str, data: dict) -> None:
    """이벤트 발행."""
    entry = {
        "ts": datetime.now().astimezone().isoformat(timespec="seconds"),
        "source": source,
        "event": event,
        "data": data,
    }
    with _STATE_LOCK:
        with EVENTS_FILE.open("a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")

def read_goal_state() -> dict:
    """현재 목표 상태 읽기."""
    with _STATE_LOCK:
        if not GOAL_STATE_FILE.exists():
            return _default_goal_state()
        return json.loads(GOAL_STATE_FILE.read_text(encoding="utf-8"))

def write_goal_state(state: dict) -> None:
    """목표 상태 저장."""
    with _STATE_LOCK:
        GOAL_STATE_FILE.write_text(
            json.dumps(state, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8"
        )

---

실제 사용 예시

뉴스 수집봇이 이벤트 발행:

# nim_bot.py
from event_bus import emit

collected = collect_news()
emit("nim_bot", "news_collected", {"count": len(collected)})

GSC 모니터가 상태 갱신:

# gsc_monitor.py
from event_bus import read_goal_state, write_goal_state, emit

state = read_goal_state()
state["content"]["top_keyword"] = top_keyword
write_goal_state(state)

emit("gsc_monitor", "gsc_updated", {
    "top_keyword": top_keyword,
    "total_clicks": clicks
})

---

왜 Redis 안 쓰나

JARVIS 규모에서는 파일이 더 낫습니다:

	파일 기반	Redis
설치	없음	서버 필요
디버깅	파일 열면 바로 확인	CLI 필요
영속성	자동 (파일)	AOF 설정 필요
속도	충분 (봇 7개 수준)	빠름 (불필요)
복잡도	낮음	높음

봇이 100개 이상이거나 초당 수백 이벤트면 Redis가 필요합니다. 7개 봇이 30분마다 동작하는 수준에서는 파일로 충분합니다.

---

겪은 문제: 동시 쓰기 충돌

두 봇이 동시에 goal_state.json을 쓰면 파일이 깨질 수 있습니다.

해결: threading.RLock()으로 잠금. 같은 프로세스 내 스레드 간에는 안전합니다.

프로세스 간 충돌(봇 A와 봇 B가 별도 프로세스)은 이론상 가능하지만, 봇들이 30분 간격으로 돌아서 실제로 충돌한 적은 없었습니다.

---

결론

메시지 큐나 데이터베이스 없이도 봇들이 파일로 충분히 협력합니다. 단순한 게 제일 강합니다.