고충: 작성해야 할 콘텐츠, 답장해야 할 사용자, 모니터링해야 할 데이터, 관리해야 할 서비스들. 각각의 작업은 단독으로는 쉽지만, 이들이 모이면 "스타트업 구축"이 "자신을 위해 일하기"로 변질됩니다.
학습 내용: 콘텐츠 생성, 사용자 자동 답장, 데이터 집계, 이상 징후 모니터링을 포함한 4가지 AI 파이프라인의 전체 아키텍처(Architecture) — 아키텍처 다이어그램, 실행 가능한 코드, docker-compose 설정, 그리고 본인의 서버에서 직접 재현할 수 있는 배포 가이드를 제공합니다.

만약 당신이 혼자서 AI 제품을 운영하고 있다면, 조만간 이 문제에 직면하게 될 것입니다: 할 일은 너무 많은데, 사람은 당신 한 명뿐이라는 사실 말입니다.

콘텐츠를 작성해야 하고, 사용자에게 답장해야 하며, 데이터를 살펴봐야 하고, 서비스를 관리해야 합니다. 이 각각의 작업들은 개별적으로는 쉽습니다. 하지만 이 모든 것을 합치면, 그것은 더 이상 "사업을 시작하는 것"이 아니라 "자신을 위해 일하는 것"이 되어버립니다.

이 글은 운영에 대한 회고록이 아닙니다. 프로젝트는 이제 막 시작되었고, 아직 데이터가 많지 않습니다. 이것은 **기술 아키텍처 설계 (Technical architecture design)**입니다. 기술 스택(Tech-stack) 관련 글에서 이어지는 내용으로, 한 사람이 운영할 수 있는 4가지 AI 자동화 파이프라인을 분석합니다. 각 파이프라인에는 본인의 서버에서 재현할 수 있는 완전한 아키텍처 다이어그램과 실행 가능한 코드가 포함되어 있습니다.

전체 그림: 1인이 운영 가능한 MCP 스타일 아키텍처

네 가지 파이프라인 모두 동일한 AI 서비스 계층(Service layer)과 데이터 계층(Data layer)을 공유하며, 중복을 허용하지 않습니다:

┌─────────────────────────────────────────────────────────────┐
│                     사용자 접점 계층 (User Touch Layer)                       │
│   WeChat 공식 계정 ←→ 미니 프로그램 ←→ 블로그 ←→        │
...

핵심 설계 원칙 (Core design principle): 모든 파이프라인은 두 가지 모드 — 이벤트 기반 (event-driven) + 스케줄링된 폴링 (scheduled polling) — 로 작동합니다. 이벤트 기반 방식은 실시간 요구 사항을 처리하고, 스케줄링된 폴링 방식은 배치 작업 (batch tasks)을 처리합니다. 두 모드는 동일한 AI 서비스와 데이터 레이어 (data layer)를 공유하며, 중복 구축이 전혀 없습니다.

기술 스택 선택 (Tech Stack Choices)

이 전체 스택은 2코어 / 4GB 경량 서버(월 약 $11)에서 실행됩니다. Kubernetes도, 마이크로서비스 (microservices)도 필요 없습니다. 혼자서도 운영할 수 있는 비용입니다.

각 파이프라인을 자세히 살펴보겠습니다.

파이프라인 1: 콘텐츠 제작 자동화 (Content Production Automation)

목표: AI를 사용하여 제작을 보조하고, 처음부터 글을 쓰는 데 소요되는 시간을 단축합니다.

아키텍처 (Architecture)

콘텐츠 주제 관리 (Content topic management)
        ↓
스케줄링된 트리거 (Scheduled trigger) (n8n / Celery Beat)
...

코드 (Code)

핵심은 오케스트레이터 (orchestrator) 내의 스케줄링된 워크플로입니다. 콘텐츠 생성 로직은 하나의 Python 서비스에 구현되어 있습니다:

# content_pipeline.py
# 배포 참조 (Deployment reference): /opt/opc/services/content_pipeline/
import json
...

배포 (Deployment)

# systemd 타이머 또는 n8n 웹훅 (webhook)으로 트리거; 매일 08:00에 1회 실행
0 8 * * * cd /opt/opc/services/content_pipeline && python3 run_daily.py

핵심 원칙: AI가 초안을 작성하고, 사람이 최종 검토를 수행합니다. 파이프라인은 초안만 생성하여 초안함 (drafts box)에 저장하고, 제작자에게 검토를 위해 알림을 보냅니다. 게시하기 전에 반드시 사람이 확인해야 합니다. 이 단계는 자동화할 수 없습니다.

전체 배포: systemd 서비스 + 타이머 (systemd Service + Timer)

cron 외에도 systemd를 사용하여 콘텐츠 파이프라인의 라이프사이클 (lifecycle)을 관리할 수 있습니다:

# /etc/systemd/system/content-pipeline.service
[Unit]
Description=Content Pipeline Service
...

# /etc/systemd/system/content-pipeline.timer
[Unit]
Description=Daily content generation at 8am
...

# 활성화 (Enable)
sudo systemctl daemon-reload
sudo systemctl enable content-pipeline.timer
...

n8n 워크플로 설정 (n8n Workflow Config) (선택 사항)

시각적 오케스트레이션 (Visual Orchestration)을 위해 n8n을 사용하는 경우, 워크플로 노드(Workflow nodes)는 다음과 같습니다:

1. Schedule Trigger (매일 08:00) ↓
2. HTTP Request → Python 서비스 (content_pipeline.run_daily 실행) ↓
3. Wait (AI 생성 대기, 최대 5분) ↓
4. HTTP Request → 생성 결과 쿼리 ↓
5. 품질 점수(Quality score) ≥ 7 이면 → WeChat Draft API → 초안(Drafts)에 저장 → WeCom 알림: "오늘의 초안이 준비되었습니다"
   그렇지 않으면 (Else) → WeCom 알림: "오늘의 초안이 품질 기준을 통과하지 못했습니다 — 새로운 주제를 선택하세요"

n8n의 장점은 시각적 설정(Visual configuration)이 가능하다는 점이며, 별도의 글루 코드 (Glue code)가 필요하지 않습니다. 하지만 n8n을 건너뛰고 싶다면, systemd 타이머와 bash 스크립트 조합으로도 동일하게 작동합니다.

파이프라인 2: 사용자 상호작용 자동화 (User Interaction Automation)

이 파이프라인은 WeChat 공식 계정 (WeChat Official Account), 미니 프로그램 (Mini program), 개인 사이트 및 기타 채널을 통한 사용자 메시지에 대한 자동 답장을 처리합니다.

아키텍처 (Architecture)

사용자가 메시지 전송 → 채널 서버가 푸시
        ↓
Webhook 수신 (FastAPI)
...

핵심 코드 (Core Code)

# auto_reply.py
# FastAPI 미들웨어 서비스
from fastapi import FastAPI, Request
...

배포 (Deployment)

# FastAPI 서비스로 실행
uvicorn auto_reply:app --host 0.0.0.0 --port 8200

...

이 파이프라인의 핵심 가치는 "인간을 대체하는 것"이 아니라, 표준화된 문의 사항을 필터링하여 인간이 진정으로 판단력과 공감이 필요한 메시지만 처리하도록 하는 데 있습니다.

WeChat 공식 계정 설정 (WeChat Official Account Configuration)

WeChat 공식 계정을 연결하는 경우, 계정 백엔드에서 서버 주소를 설정하십시오:

• 계정 백엔드 → 설정 및 개발 (Settings & Development) → 서버 설정 (Server config)
• URL: https://your-domain.com/webhook/reply
• Token: 사용자 정의 (코드 내의 토큰과 일치해야 함)
• 메시지 암호화 (Message encryption): 평문 모드 (Plaintext mode, 디버깅용) → 보안 모드 (Secure mode, 운영용)

WeChat 서버 설정 검증 코드:

# wechat_verify.py
# 최초 설정 시 서버 URL을 검증하는 데 사용됨
from flask import Flask, request, make_response
...

Chroma 지식 베이스 초기화 (Initializing the Chroma Knowledge Base)

자동 답장의 품질은 지식 베이스 (Knowledge base)의 품질에 달려 있습니다. 초기화 방법은 다음과 같습니다:

# 1. Chroma 설치
pip install chromadb

...

# embedding_config.py
# 임베딩 모델 (embedding model) 및 검색 파라미터 (retrieval parameters) 설정
import chromadb
...

전체 Docker Compose 오케스트레이션 (Full Docker Compose Orchestration)

모든 서비스는 docker-compose에 의해 관리됩니다:

version: '3.8'
services:
  # 콘텐츠 파이프라인 (Content pipeline)
...

# 한 번의 명령어로 모든 서비스 시작
docker-compose up -d
docker-compose ps   # 모든 서비스가 정상(healthy)인지 확인

파이프라인 3: 데이터 집계 자동화 (Data Aggregation Automation)

이 파이프라인은 "내가 실제로 얼마나 잘 운영하고 있는지 모르겠다"는 문제를 해결합니다. 주요 지표 (metrics)를 정해진 일정에 따라 수집하고 이를 대시보드 (dashboard)로 전송합니다.

아키텍처 (Architecture)

Celery Beat 스케줄 트리거 (scheduled trigger)
        ↓
데이터 수집 (Data collection)
...

주요 코드 (Key Code)

# data_collector.py
# Celery 스케줄 작업 (scheduled task)
from celery import Celery
from datetime import datetime
import requests
import sqlite3

app = Celery("data_collector", broker="redis://localhost:6379/0")

@app.task
def collect_hourly():
    """매 시간마다 수집"""
    metrics = {}

    # 1. 서버 지표 (Server metrics)
    try:
        cpu = requests.get("http://localhost:9100/metrics", timeout=5)
        metrics["cpu_usage"] = parse_cpu_metric(cpu.text)
        metrics["mem_usage"] = parse_mem_metric(cpu.text)
        metrics["disk_usage"] = parse_disk_metric(cpu.text)
    except:
        notify_alert("서버 지표 수집 실패!")

    # 2. 비즈니스 지표 (Business metrics)
    db = sqlite3.connect("/opt/opc/data/metrics.db")
    cursor = db.cursor()
    cursor.execute("""
        SELECT COUNT(*) FROM api_logs
        WHERE created_at > datetime('now', '-1 hour')
    """)
    metrics["requests_past_hour"] = cursor.fetchone()[0]

3. 시계열 저장소에 기록 (ORM 스타일 의사코드:

#    hourly_metrics에 대한 원시 INSERT 문을 모델 객체로 대체함)
db.add(HourlyMetric(
    collected_at=datetime.now().isoformat(),
    cpu=metrics.get("cpu_usage", 0),
    mem=metrics.get("mem_usage", 0),
    disk=metrics.get("disk_usage", 0),
    requests=metrics["requests_past_hour"],
))
db.commit()
db.close()

# 4. 대시보드로 푸시 (Push to dashboard)
push_to_dashboard(metrics)

# 5. 이상 탐지 (Anomaly detection)
if metrics.get("cpu_usage", 0) > 85:
    notify_alert(f"CPU 사용률 {metrics['cpu_usage']}%")