FastAPI와 Kong을 활용한 AI 에이전트용 사용량 기반 과금 (Usage-Based Billing)

AI 에이전트를 구축했다면, 다음 질문은 간단합니다. 어떻게 비용을 청구할 것인가?

정액 구독 모델은 AI 워크로드에 적합하지 않습니다. 토큰 비용은 모델별로, 방향(입력 vs 출력)별로, 그리고 각 사용자가 실제로 소비하는 양에 따라 달라집니다. 어떤 사용자는 하루에 200개의 토큰을 보낼 수도 있고, 다른 사용자는 50,000개를 소모할 수도 있습니다. 정액제는 가벼운 사용자에게는 과다 청구하거나, 헤비 사용자를 보조하는 결과를 초래합니다.

필요한 것은 사용량 기반 과금 (Usage-based billing)입니다. 즉, 각 사용자가 정확히 사용한 만큼만 지불하는 방식입니다. 이 튜토리얼에서는 FastAPI, OpenMeter, 그리고 Kong Metering & Billing (Cloud OpenMeter 엔진)을 사용하여 샘플 AI 에이전트를 구축하고 이에 대한 과금 체계를 설정해 보겠습니다.

사용량 기반 과금 (Usage-Based Billing)이란 무엇인가?

사용량 기반 과금이란 고정된 금액이 아닌 실제 소비량에 따라 고객에게 비용을 청구하는 것을 의미합니다. AI 에이전트의 경우, 모든 API 호출은 토큰 수와 연결된 측정 가능한 비용을 가지며, 이러한 비용은 모델에 따라 달라집니다. 따라서 이는 매우 자연스러운 방식입니다.

사용량 기반 과금 시스템에는 네 가지 요소가 필요합니다:

1. 이벤트 수집 (Event ingestion): 과금 대상이 되는 일이 발생할 때마다 사용량 데이터를 캡처합니다.

2. 계측 (Metering): 원시 이벤트 (Raw events)를 과금 기간별 고객별 총합으로 집계합니다.

3. 가격 책정 (Pricing): 계측된 사용량에 요금표 (Rate cards) 또는 티어 (Tiers)를 적용합니다.

4. 송장 발행 (Invoicing): 청구서를 생성하고 결제를 수집합니다.

이 모든 것을 처음부터 구축하는 것은 매우 큰 엔지니어링 프로젝트입니다. 이벤트 저장소, 중복 제거 (Deduplication), 윈도우 집계 (Windowed aggregation), 그리고 그 위에 구축될 과금 레이어가 필요합니다. 이는 핵심 제품이 아닌 것을 위해 너무 많은 인프라를 요구하는 일입니다.

도구들

OpenMeter는 Kong에서 관리하는 오픈 소스 계측 엔진 (Metering engine)입니다. 이는 CloudEvents 표준 (CNCF 사양)을 기반으로 구축되었으며, 대규모 환경에서 실시간 이벤트 수집, 중복 제거 및 집계를 처리합니다. 소스 코드는 GitHub에 공개되어 있으며, 완전한 제어권을 원한다면 직접 호스팅 (Self-host)할 수도 있습니다.

Kong Konnect Metering & Billing은 OpenMeter 저장소를 기반으로 구축된 클라우드 플랫폼입니다. 이 플랫폼은 기능, 가격 플랜, 구독, 인보이스(Invoicing) 발행 및 결제 제공업체 연동과 같은 과금 계층(Billing layer)을 추가합니다. 이 튜토리얼에서는 이 플랫폼을 사용하여 데이터를 수집(Ingest)하고, 집계(Aggregate)하며, 요율을 적용(Apply rates)하고, 인보이스를 생성하는 과정을 다룹니다.

구축하게 될 내용

여러분이 구축하게 될 시스템의 아키텍처는 다음과 같습니다:

흐름은 다음과 같이 작동합니다:

1. 사용자가 API 엔드포인트(/generate, /summarize 또는 /analyze)를 호출합니다.

2. 에이전트가 OpenAI에 요청을 보내고 토큰 수(Token counts)가 포함된 응답을 받습니다.

3. 애플리케이션이 사용 데이터(소비된 토큰, 사용된 모델, 사용자 ID)를 포함한 CloudEvent를 Kong Metering & Billing API로 전송합니다.

4. 플랫폼이 이벤트를 미터(Meters)로 집계하고, 사용자의 플랜에 따른 가격을 적용하며, 과금 주기(Billing cycle)가 끝나면 인보이스를 생성합니다.

5. 결제 제공업체가 결제를 수집합니다.

여러분의 코드는 1단계부터 3단계까지를 처리합니다. 4단계와 5단계는 과금 시스템을 구성하고 나면 자동으로 실행됩니다.

사전 요구 사항

실습을 위해 다음 사항이 필요합니다:

• Python 3.10 이상

• OpenAI API 키 (어떠한 LLM 제공업체든 사용할 수 있지만, 이 튜토리얼에서는 OpenAI를 사용합니다)

• Metering & Billing을 위한 Kong Konnect 계정

• API 액세스를 위한 Konnect 개인 액세스 토큰 (Personal Access Token, PAT)

• FastAPI 및 REST API에 대한 기본적인 이해

목차

• Step 1: 프로젝트 설정

• Step 2: AI API 구축

• Step 3: API 키 인증 추가

• Step 4: Konnect Metering & Billing으로 사용량 이벤트 전송

• Step 5: Meter 설정

• Step 6: 기능(Features) 및 요금제(Pricing Plans) 정의

• Step 7: 고객 온보딩 및 구독 생성

• Step 8: 전체 과금 흐름 테스트

• Step 9: 결제 제공업체(Payment Provider) 연결

• 주의 사항 및 운영 팁

• 학습 내용 요약

• 다음 단계 안내

Step 1: 프로젝트 설정

새 프로젝트 디렉토리를 생성하고 가상 환경 (Virtual Environment)을 설정합니다:

mkdir ai-billing-app
cd ai-billing-app
python -m venv venv
...

의존성 패키지 (Dependencies)를 설치합니다:

pip install fastapi uvicorn openai httpx python-dotenv pydantic

패키지에 대한 몇 가지 참고 사항:

• fastapi 및 uvicorn: API를 위한 웹 프레임워크 (Web Framework) 및 ASGI 서버

• openai: LLM 호출을 위한 OpenAI Python SDK

• httpx: 사용량 이벤트를 전송하고 Konnect Metering & Billing으로 API 호출을 수행하기 위한 HTTP 클라이언트 (HTTP Client)

• python-dotenv: .env 파일에서 환경 변수 (Environment Variables)를 로드

• pydantic: 데이터 검증 (Data Validation) (FastAPI에 포함되어 있으나 명확성을 위해 기재)

설정을 위한 .env 파일을 생성합니다:

# .env
OPENAI_API_KEY=sk-your-openai-api-key
KONNECT_API_URL=https://us.api.konghq.com/v3/openmeter
...

KONNECT_TOKEN은 Konnect의 계정 설정에서 생성하는 개인 액세스 토큰 (Personal Access Token, PAT)입니다.

이 튜토리얼 전반에 걸쳐 사용될 curl 명령어를 위해, 터미널에서 다음 변수들을 내보내기 (Export) 하세요:

export KONNECT_API_URL=https://us.api.konghq.com/v3/openmeter
export KONNECT_TOKEN=kpat_your_konnect_personal_access_token

메인 애플리케이션 파일을 생성합니다:

touch app.py

프로젝트 구조는 다음과 같습니다:

ai-billing-app/
├── app.py           # 메인 FastAPI 애플리케이션
├── .env             # 환경 변수
...

Step 2: AI API 구축

Step 2: AI API 구축

OpenAI의 API를 래핑(wrap)하는 기본적인 FastAPI 앱부터 시작합니다. 이를 통해 사용자에게 텍스트 생성, 콘텐츠 요약, 인사이트 도출을 위한 텍스트 분석이라는 세 가지 기능을 제공할 수 있습니다.

app.py를 열고 다음 내용을 추가하세요:

import os
import uuid
import datetime
...

API가 제대로 작동하는지 테스트하세요:

uvicorn app:app --reload

다른 터미널에서:

curl -X POST http://localhost:8000/api/generate \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Write a haiku about Python programming"}'

생성된 텍스트와 토큰 사용량(token usage) 카운트가 포함된 응답을 받아야 합니다. usage 필드가 바로 과금 시스템(billing system)에 전달될 데이터입니다.

Step 3: API Key 인증 추가

사용자에게 비용을 청구하기 전에, 누가 각 요청을 보내고 있는지 알아야 합니다. 간단한 API 키 인증(API key authentication) 레이어를 추가하세요.

운영 환경(production system)에서는 API 키를 데이터베이스에 저장하고 해싱(hash)하여 관리해야 합니다. 이 튜토리얼에서는 과금 기능에 집중하기 위해 인메모리 딕셔너리(in-memory dictionary)를 사용합니다.

인증을 추가하기 위해 app.py를 업데이트하세요:

# load_dotenv() 호출 다음에 이 내용을 추가하세요

# 운영 환경에서는 해싱된 키와 함께 데이터베이스에 저장하세요
...

이제 각 엔드포인트(endpoint)가 인증을 요구하도록 업데이트합니다. 다음은 /api/generate 엔드포인트의 업데이트 예시입니다 (/api/summarize 및 /api/analyze에도 동일한 패턴을 적용하세요):

from fastapi import FastAPI, HTTPException, Header, Depends

@app.post("/api/generate", response_model=AIResponse)
...

인증이 작동하는지 테스트하세요:

# 이 요청은 401 에러와 함께 실패해야 합니다
curl -X POST http://localhost:8000/api/generate \
  -H "Content-Type: application/json" \
...

이제 누가 각 요청을 보내는지 알 수 있습니다. 해당 user_id 필드가 API 사용량과 과금 고객을 연결하는 고리가 됩니다.

Step 4: 사용량 이벤트(Usage Events)를 Konnect Metering & Billing으로 전송

여기서부터 과금(Billing)이 시작됩니다. 사용자가 API 호출을 할 때마다 Konnect Metering & Billing으로 사용량 이벤트(Usage Event)를 전송하게 됩니다. 오픈 소스인 OpenMeter 엔진을 기반으로 하는 미터링(Metering) 계층이 이러한 이벤트들을 수집하고 집계합니다. 별도의 추가 의존성 없이 표준 HTTP 호출을 사용하여 API와 직접 통신하면 됩니다.

CloudEvents란 무엇인가요?

Konnect Metering & Billing API는 이벤트 데이터를 기술하기 위한 CNCF 표준인 CloudEvents 형식의 이벤트를 수락합니다. 전송하는 모든 사용량 이벤트는 CloudEvents JSON 객체입니다.

단일 사용량 이벤트의 모습은 다음과 같습니다:

{
  "specversion": "1.0",
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
...

각 필드는 과금 파이프라인(Billing Pipeline)에서 특정 목적을 가집니다:

필드	역할	예시
specversion	CloudEvents 버전. 항상 "1.0"	"1.0"
...

subject 필드가 매우 중요합니다. 이는 이벤트를 특정 고객과 연결하는 역할을 합니다. 나중에 과금을 설정할 때, 일치하는 subject_keys를 가진 고객을 생성하게 되며, 해당 subject를 가진 모든 이벤트는 그 고객의 사용량으로 합산됩니다.

Python 코드에 연결하기 전에 Konnect 인증 정보가 제대로 작동하는지 확인하려면 curl을 사용하여 이벤트 수집(Ingestion)을 직접 테스트할 수 있습니다:

curl -X POST $KONNECT_API_URL/events \
  -H "Authorization: Bearer $KONNECT_TOKEN" \
  -H "Content-Type: application/cloudevents+json" \
...

202 Accepted 응답은 이벤트가 성공적으로 수집되었음을 의미합니다. 콘텐츠 타입(Content Type)인 application/cloudevents+json에 유의하세요. 이는 API에 단일 CloudEvent를 보내고 있음을 알려줍니다.

과금 모듈(Billing Module) 생성하기

이제 모든 API 호출 후에 Konnect로 사용량 이벤트를 전송하는 함수를 Python 앱에 추가합니다. 이 함수는 httpx를 사용하여 Konnect Metering & Billing 이벤트 엔드포인트로 CloudEvents를 직접 POST 합니다:

import httpx

KONNECT_API_URL = os.getenv("KONNECT_API_URL", "https://us.api.konghq.com/v3/openmeter")
...

이 함수에 관한 몇 가지 중요한 사항은 다음과 같습니다:

1. 각 이벤트는 고유한 UUID를 id로 가집니다. 미터링 엔진(metering engine)은 id + source의 조합을 통해 이벤트의 중복을 제거합니다. 네트워크 재시도(retry)로 인해 동일한 이벤트가 두 번 전송되더라도, 두 번 계산되지 않습니다.

2. subject는 사용자의 ID이며, 이름이나 API 키가 아닙니다. 이는 나중에 생성할 과금 고객(billing customer)과 직접 매핑됩니다.

3. type은 llm_token_usage입니다. 특정 미터(meter)를 구성하여 이 정확한 타입의 이벤트를 수신하도록 설정하게 됩니다. 타입이 일치하지 않으면 이벤트는 미터링(metered)되지 않습니다.

4. 콘텐츠 타입(content type)은 반드시 application/cloudevents+json이어야 합니다. 이를 통해 Konnect API는 요청 본문(request body)을 CloudEvent로 파싱하는 방법을 알게 됩니다.

5. 오류가 발생해도 요청이 실패하지 않습니다. 과금 텔레메트리(billing telemetry)는 사용자의 API 경험을 절대 방해해서는 안 됩니다. 프로덕션 환경에서는 실패한 이벤트를 재시도 큐(retry queue)로 전송해야 합니다.

모든 요청에서 토큰 추적하기

이제 각 엔드포인트(endpoint)에 track_usage를 연결합니다. /api/generate 엔드포인트를 업데이트하세요:

@app.post("/api/generate", response_model=AIResponse)
def generate_text(request: GenerateRequest, user: dict = Depends(authenticate)):
    """프롬프트를 기반으로 텍스트 콘텐츠를 생성합니다."""
...

/api/summarize와 /api/analyze에도 동일한 패턴을 적용하십시오. 유일한 차이점은 track_usage에 전달하는 endpoint 값입니다.

업데이트된 /api/summarize는 다음과 같습니다:

@app.post("/api/summarize", response_model=AIResponse)
def summarize_text(request: SummarizeRequest, user: dict = Depends(authenticate)):
    """텍스트를 요약합니다."""
...

그리고 /api/analyze는 다음과 같습니다: