AI & Automation

AI/LLM 워크플로우와 클라우드 플랫폼 연결

뉴스 브리핑, RAG, 서버리스 데이터 파이프라인을 수집-정제-검색-생성-검증의 운영 흐름으로 정리합니다.

LLM 기능을 만들 때 처음에는 프롬프트가 가장 중요해 보인다. 하지만 운영 가능한 기능으로 보려면 질문이 달라진다. 어떤 데이터를 수집할 것인가, 원문 출처를 어떻게 남길 것인가, 중복이나 실패를 어떻게 처리할 것인가, 모델 응답을 어떻게 검증할 것인가, 비용과 지연시간은 어디서 관측할 것인가.

뉴스 브리핑 봇과 Daily Compass 같은 작은 자동화 아이디어를 정리하면서, AI 기능도 결국 백엔드와 데이터 파이프라인 문제라는 생각이 강해졌다. LLM 호출은 흐름의 한 단계일 뿐이고, 그 앞뒤에 수집, 정제, 저장, 검색, 권한, 재처리, 로그가 붙어야 서비스가 된다.

뉴스 브리핑 봇에서 드러나는 데이터 흐름

AI 뉴스 브리핑 봇은 단순히 “뉴스를 요약해줘”라고 요청하는 기능처럼 보인다. 하지만 실제 흐름은 여러 단계로 나눠야 한다.

1. Schedule or user request
2. RSS / News API fetch
3. Category and keyword filtering
4. Duplicate removal
5. Top article selection
6. LLM summarization
7. Keyword extraction
8. Tone classification
9. Briefing formatting
10. Delivery and audit log

초기 시나리오에서는 매일 오전 7시 같은 스케줄 또는 사용자의 요청으로 브리핑이 시작된다. 사전에 정한 RSS 피드나 뉴스 API에서 최신 기사를 수집하고, IT/경제/국제 같은 카테고리나 키워드로 필터링한다. 그중 중요한 기사 5개는 상세 요약을 만들고, 나머지는 헤드라인 목록으로 제공한다.

출력 포맷은 다음처럼 구조화할 수 있다.

{
  "date": "2025-07-22",
  "category": "IT",
  "topArticles": [
    {
      "title": "Article title",
      "source": "Publisher",
      "summary": [
        "First key sentence",
        "Second key sentence",
        "Third key sentence"
      ],
      "keywords": ["AI", "Cloud", "Security"],
      "tone": "neutral",
      "url": "https://example.com/news/1"
    }
  ],
  "headlines": [
    {
      "title": "Additional headline",
      "url": "https://example.com/news/6"
    }
  ]
}

구조화된 출력은 사용자에게 보여주기에도 좋지만, 저장과 재처리에도 유리하다. 브리핑 전체를 문자열 하나로 저장하면 검색, 필터링, 오류 분석이 어려워진다. 반대로 기사 단위로 source, URL, collectedAt, summary, modelName, promptVersion을 남기면 나중에 어떤 모델과 프롬프트가 어떤 결과를 만들었는지 추적할 수 있다.

수집 단계는 크롤링보다 정책이 먼저다

웹 크롤링은 HTTP 요청으로 페이지를 가져오고, HTML에서 필요한 데이터를 추출하는 과정이다. 스크레이핑은 그중 특정 데이터를 뽑아내는 작업에 가깝다.

import requests
from bs4 import BeautifulSoup

response = requests.get("https://example.com/news", timeout=10)
response.raise_for_status()

soup = BeautifulSoup(response.text, "html.parser")
titles = [tag.get_text(strip=True) for tag in soup.select(".headline")]

하지만 뉴스나 외부 데이터를 다룰 때는 코드보다 정책이 먼저다.

Collection policy
- robots.txt and terms of service
- allowed source list
- request interval and rate limit
- user-agent
- timeout and retry
- original URL preservation
- copyright and excerpt boundary

Open API를 쓸 수 있다면 API가 더 안정적이다. 인증 키, rate limit, JSON/XML 스키마, 에러 코드가 정의되어 있기 때문이다.

import requests

params = {
    "query": "cloud platform",
    "page": 1,
    "size": 20,
}

headers = {
    "Authorization": f"Bearer {api_key}",
}

res = requests.get(
    "https://api.example.com/news",
    params=params,
    headers=headers,
    timeout=10,
)
res.raise_for_status()
items = res.json()["items"]

수집 단계에서 실패를 제대로 다루지 않으면 LLM 단계가 아무리 좋아도 전체 품질이 흔들린다. 외부 API 장애, HTML 구조 변경, 빈 응답, 중복 기사, 인코딩 문제, rate limit 초과는 모두 파이프라인의 정상적인 실패 사례로 다뤄야 한다.

정제와 중복 제거

수집한 데이터는 그대로 LLM에 넣기 어렵다. 제목, 본문, 날짜, 출처, URL, 카테고리, 언어를 맞추고, 중복을 제거해야 한다.

from hashlib import sha256
from urllib.parse import urlnormalize

def article_key(title: str, url: str) -> str:
    normalized = f"{title.strip().lower()}|{urlnormalize(url)}"
    return sha256(normalized.encode("utf-8")).hexdigest()

기사 본문이 없거나 너무 짧은 경우, 광고 문구만 추출된 경우, 동일 기사가 여러 매체에 반복되는 경우도 걸러야 한다. pandas를 쓰면 간단한 탐색과 중복 제거를 빠르게 할 수 있다.

import pandas as pd

df = pd.DataFrame(items)
df = df.dropna(subset=["title", "url"])
df["title"] = df["title"].str.strip()
df = df.drop_duplicates(subset=["url"])
df = df[df["title"].str.len() > 10]

운영 파이프라인에서는 DataFrame으로 끝내기보다 DB에 raw 데이터와 normalized 데이터를 분리해 저장하는 편이 좋다.

raw_articles
- source
- fetched_at
- raw_payload
- status

articles
- article_id
- title
- url
- publisher
- published_at
- normalized_text
- content_hash

이렇게 나누면 정제 로직이 바뀌어도 raw 데이터를 기준으로 다시 처리할 수 있다.

LLM 호출은 비동기 작업으로 보는 편이 안전하다

LLM 요약은 시간이 걸리고 실패할 수 있다. 외부 API rate limit, timeout, 모델 응답 형식 오류, 비용 제한을 고려해야 한다. 사용자의 HTTP 요청 안에서 모든 기사 수집과 요약을 끝내려 하면 지연시간이 길고 실패 범위가 커진다.

서버리스 구조에서는 큐나 스케줄러를 두고 비동기 작업으로 나누는 편이 자연스럽다.

Timer trigger
-> fetch articles
-> enqueue summarization jobs
-> summarize top articles
-> store result
-> publish briefing

Azure Functions나 AWS Lambda는 이런 이벤트 기반 작업에 잘 맞는다. 작은 작업을 짧게 실행하고, 실패 시 재시도하거나 dead-letter queue로 보낼 수 있다. Daily Compass에서도 Azure Functions, Cosmos DB, Static Web Apps, Application Insights를 연결해 사용자 대시보드와 데이터 흐름을 설계했다.

User
-> Azure Static Web Apps
-> Azure Functions
-> Cosmos DB for MongoDB
-> Application Insights

뉴스 수집이나 명언 추천처럼 주기적으로 갱신되는 데이터는 서버리스와 잘 맞지만, cold start, timeout, 동시 실행, 비용, 외부 API 제한을 함께 봐야 한다.

RAG는 검색 품질과 권한 문제다

RAG는 문서를 넣고 모델이 답하게 하는 기능처럼 보이지만, 실제로는 검색 시스템에 가깝다.

Document ingestion
-> parsing
-> chunking
-> embedding
-> indexing
-> retrieval
-> prompt composition
-> answer generation
-> citation and validation

chunking은 검색 품질에 큰 영향을 준다. 너무 작으면 맥락이 사라지고, 너무 크면 검색 결과에 불필요한 내용이 많이 섞인다.

def chunk_text(text: str, size: int = 800, overlap: int = 120) -> list[str]:
    chunks = []
    start = 0
    while start < len(text):
        end = start + size
        chunks.append(text[start:end])
        start = end - overlap
    return chunks

RAG에서 더 중요한 것은 권한이다. 내부 문서나 업무 데이터가 섞인 환경에서는 사용자가 볼 수 없는 문서가 검색 컨텍스트에 들어가면 안 된다. 검색 단계에서 tenant, team, role, document ACL을 필터링해야 한다.

select chunk_id, content
from document_chunks
where document_id in (
    select document_id
    from document_permissions
    where user_id = :user_id
)
order by embedding <-> :query_embedding
limit 5;

PostgreSQL의 pgvector, OpenSearch, 전용 Vector DB 중 무엇을 선택하든 기준은 같다. 데이터 규모, 필터링 복잡도, 운영 비용, 검색 지연시간, 백업/복구, 접근 제어를 함께 봐야 한다.

프롬프트는 버전 관리되는 실행 로직이다

프롬프트는 단순 문장이 아니라 모델 동작을 바꾸는 실행 로직에 가깝다. 뉴스 브리핑 봇에서는 다음 제한을 명확히 둬야 한다.

Prompt constraints
- 원문에 없는 사실을 추가하지 않는다.
- 독립적인 사실 검증을 수행한 것처럼 말하지 않는다.
- 요약은 원문 내용에 근거한다.
- 논조 분석은 기사 표현의 언어적 분석으로 제한한다.
- 원문 링크를 항상 포함한다.
- 불확실한 경우 요약 실패 상태로 남긴다.

요약 결과는 가능하면 JSON schema로 받는 편이 후처리에 좋다.

{
  "summary": ["...", "...", "..."],
  "keywords": ["...", "...", "...", "...", "..."],
  "tone": "positive|neutral|negative",
  "confidence": 0.82,
  "warnings": []
}

모델 응답이 스키마를 지키지 않거나, 원문 링크가 빠지거나, 요약이 너무 길면 재시도하거나 실패로 저장한다. 실패도 데이터다. 어떤 기사에서 어떤 이유로 실패했는지 남겨야 다음 개선이 가능하다.

관측성: 비용, 지연시간, 품질

LLM 시스템은 일반 API보다 관측해야 할 값이 많다.

Operational signals
- ingestion success / failure count
- source API latency
- duplicate ratio
- summarization latency
- model error rate
- token usage
- estimated cost
- cache hit rate
- user feedback

Application Insights, CloudWatch, OpenTelemetry, Prometheus/Grafana 중 어떤 도구를 쓰든 핵심은 같다. 어느 단계에서 실패했는지, 어떤 입력이 비용을 많이 쓰는지, 응답 품질이 낮은 케이스가 어디에서 나오는지 볼 수 있어야 한다.

FastAPI 기반으로 만든다면 요청 단위 trace를 남길 수 있다.

from fastapi import FastAPI
import time

app = FastAPI()

@app.middleware("http")
async def add_timing_header(request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    response.headers["x-process-time-ms"] = str(
        round((time.perf_counter() - start) * 1000, 2)
    )
    return response

Redis는 결과 캐시나 작업 상태 저장에 사용할 수 있다. 동일한 원문과 동일한 prompt version, 동일한 model version이면 요약 결과를 재사용할 수 있다.

summary:{article_hash}:{prompt_version}:{model}

캐시를 쓰면 비용과 지연시간은 줄지만, 모델이나 프롬프트를 바꿨을 때 오래된 결과가 섞이지 않도록 key 설계를 명확히 해야 한다.

보안과 데이터 경계

AI 기능은 데이터 경계가 흐려지기 쉽다. 사용자가 입력한 질문, 검색된 문서, 모델에 전달한 컨텍스트, 생성된 답변, 로그에 남은 내용이 모두 민감할 수 있다.

Daily Compass 같은 개인 대시보드에서도 감정 기반 명언, TODO, Calendar, 뉴스 데이터가 한 화면에 모인다. 작은 서비스라도 무엇을 저장하고, 무엇을 일시 처리하고, 무엇을 외부 API로 보낼지 정해야 한다.

Data boundary
- user input
- stored personal data
- retrieved documents
- LLM prompt context
- generated answer
- application logs
- analytics events

배포 구조

작은 AI 워크플로우는 다음처럼 시작할 수 있다.

Frontend
-> FastAPI / Azure Functions API
-> Queue
-> Worker
-> PostgreSQL / Cosmos DB
-> OpenSearch or vector index
-> LLM provider
-> Observability stack

컨테이너로 운영한다면 Docker Compose로 로컬 개발환경을 만들고, Kubernetes에서는 API와 worker를 분리해 배포한다.

services:
  api:
    build: .
    environment:
      DATABASE_URL: postgresql://app:app@postgres:5432/app
      REDIS_URL: redis://redis:6379

  worker:
    build: .
    command: python -m app.worker
    environment:
      DATABASE_URL: postgresql://app:app@postgres:5432/app
      REDIS_URL: redis://redis:6379

  postgres:
    image: postgres:16

  redis:
    image: redis:7

서버리스로 운영한다면 스케줄러, 큐, 함수, DB, 모니터링을 연결한다. 어떤 방식을 쓰든 중요한 것은 LLM 호출을 애플리케이션 코드 한가운데 숨기지 않고, 재시도와 실패 처리, 비용 추적이 가능한 작업 단위로 분리하는 것이다.

정리

AI/LLM 기능은 프롬프트만으로 완성되지 않는다. 뉴스 브리핑 하나를 만들더라도 수집, 정제, 중복 제거, 저장, 요약, 검증, 전달, 로그가 필요하다. RAG는 문서를 모델에 넣는 기능이 아니라 검색 품질, 권한 필터링, 출처 표시, 응답 검증을 함께 다루는 시스템이다.

이 흐름을 백엔드와 플랫폼 관점에서 보면 선택지가 더 선명해진다. Python/FastAPI는 API와 worker를 빠르게 만들기 좋고, PostgreSQL/Redis/OpenSearch는 저장과 검색, 캐시를 담당할 수 있다. Azure Functions나 AWS Lambda는 이벤트 기반 작업을 작게 실행하는 데 적합하고, Kubernetes는 API와 worker를 분리해 장기 운영하기 좋다.

결국 운영 가능한 AI 기능은 모델 호출을 중심에 두지 않는다. 데이터가 어디서 왔고, 어떤 기준으로 정제되었고, 어떤 권한으로 검색되었고, 어떤 프롬프트 버전으로 생성되었고, 실패와 비용을 어디서 볼 수 있는지가 함께 설계되어야 한다.