3장: gRPC — 고성능 서비스 간 통신

// Protobuf 메시지 정의
syntax = "proto3";
 
message User {
  string id = 1;
  string name = 2;
  string email = 3;
  int32 age = 4;
  repeated string tags = 5;
}

{
  "id": "user-42",
  "name": "Kreath",
  "email": "kreath@example.com",
  "age": 30,
  "tags": ["developer", "architect"]
}

syntax = "proto3";
 
package ai.inference.v1;
 
option go_package = "github.com/example/ai-service/gen/go/ai/inference/v1";
option java_package = "com.example.ai.inference.v1";
 
import "google/protobuf/timestamp.proto";
 
// AI 추론 서비스
service InferenceService {
  // 단일 추론 요청 (Unary)
  rpc Complete(CompleteRequest) returns (CompleteResponse);
  
  // 스트리밍 추론 (Server Streaming)
  rpc StreamComplete(CompleteRequest) returns (stream StreamChunk);
  
  // 배치 임베딩 (Client Streaming)
  rpc BatchEmbed(stream EmbedRequest) returns (BatchEmbedResponse);
  
  // 대화형 추론 (Bidirectional Streaming)
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}
 
message CompleteRequest {
  string model = 1;
  repeated Message messages = 2;
  float temperature = 3;
  int32 max_tokens = 4;
  
  // 선택적 필드
  optional float top_p = 5;
  optional int32 seed = 6;
}
 
message Message {
  string role = 1;
  string content = 2;
}
 
message CompleteResponse {
  string id = 1;
  string model = 2;
  repeated Choice choices = 3;
  Usage usage = 4;
  google.protobuf.Timestamp created_at = 5;
}
 
message Choice {
  int32 index = 1;
  Message message = 2;
  string finish_reason = 3;
}
 
message Usage {
  int32 prompt_tokens = 1;
  int32 completion_tokens = 2;
  int32 total_tokens = 3;
}
 
message StreamChunk {
  string id = 1;
  string model = 2;
  repeated StreamChoice choices = 3;
  optional Usage usage = 4;  // 마지막 청크에서만 포함
}
 
message StreamChoice {
  int32 index = 1;
  MessageDelta delta = 2;
  optional string finish_reason = 3;
}
 
message MessageDelta {
  optional string role = 1;
  optional string content = 2;
}
 
message EmbedRequest {
  string model = 1;
  string text = 2;
}
 
message BatchEmbedResponse {
  repeated Embedding embeddings = 1;
  Usage usage = 2;
}
 
message Embedding {
  int32 index = 1;
  repeated float vector = 2;
}
 
message ChatMessage {
  string role = 1;
  string content = 2;
  google.protobuf.Timestamp timestamp = 3;
}

# protoc 컴파일러와 플러그인 설치
# Python
pip install grpcio-tools
 
# 코드 생성 (Python)
python -m grpc_tools.protoc \
  -I./proto \
  --python_out=./gen/python \
  --grpc_python_out=./gen/python \
  --pyi_out=./gen/python \
  proto/ai_service.proto
 
# Go
protoc \
  -I./proto \
  --go_out=./gen/go \
  --go-grpc_out=./gen/go \
  proto/ai_service.proto
 
# TypeScript (ts-proto)
protoc \
  -I./proto \
  --plugin=protoc-gen-ts_proto=./node_modules/.bin/protoc-gen-ts_proto \
  --ts_proto_out=./gen/ts \
  --ts_proto_opt=outputServices=grpc-js \
  proto/ai_service.proto

import grpc
from gen.python import ai_service_pb2 as pb
from gen.python import ai_service_pb2_grpc as service
 
 
async def unary_complete(stub: service.InferenceServiceStub):
    request = pb.CompleteRequest(
        model="claude-4",
        messages=[
            pb.Message(role="user", content="API 설계의 핵심 원칙은?"),
        ],
        temperature=0.7,
        max_tokens=1024,
    )
    
    response = await stub.Complete(
        request,
        timeout=30.0,  # 30초 데드라인
        metadata=[("x-api-key", "sk-abc123")],
    )
    
    print(f"Model: {response.model}")
    print(f"Response: {response.choices[0].message.content}")
    print(f"Tokens: {response.usage.total_tokens}")

async def stream_complete(stub: service.InferenceServiceStub):
    request = pb.CompleteRequest(
        model="claude-4",
        messages=[
            pb.Message(role="user", content="gRPC의 장점을 설명해주세요"),
        ],
        temperature=0.7,
        max_tokens=2048,
    )
    
    full_response = []
    async for chunk in stub.StreamComplete(request, timeout=60.0):
        for choice in chunk.choices:
            if choice.delta.content:
                full_response.append(choice.delta.content)
                print(choice.delta.content, end="", flush=True)
            
            if choice.finish_reason:
                print(f"\n완료: {choice.finish_reason}")
        
        # 마지막 청크에 사용량 정보 포함
        if chunk.HasField("usage"):
            print(f"총 토큰: {chunk.usage.total_tokens}")

async def batch_embed(stub: service.InferenceServiceStub):
    async def generate_requests():
        texts = [
            "API 설계 패턴",
            "마이크로서비스 아키텍처",
            "서버리스 컴퓨팅",
            "이벤트 주도 설계",
        ]
        for text in texts:
            yield pb.EmbedRequest(
                model="text-embedding-3-large",
                text=text,
            )
    
    response = await stub.BatchEmbed(generate_requests(), timeout=30.0)
    
    for embedding in response.embeddings:
        print(f"Index {embedding.index}: "
              f"dim={len(embedding.vector)}, "
              f"first_3={embedding.vector[:3]}")
    
    print(f"총 토큰: {response.usage.total_tokens}")

async def interactive_chat(stub: service.InferenceServiceStub):
    request_queue = asyncio.Queue()
    
    async def send_messages():
        while True:
            user_input = await asyncio.to_thread(input, "User: ")
            if user_input.lower() == "quit":
                break
            
            message = pb.ChatMessage(
                role="user",
                content=user_input,
            )
            await request_queue.put(message)
        
        await request_queue.put(None)  # 종료 신호
    
    async def request_iterator():
        while True:
            message = await request_queue.get()
            if message is None:
                break
            yield message
    
    # 양방향 스트리밍 시작
    send_task = asyncio.create_task(send_messages())
    
    async for response in stub.Chat(request_iterator()):
        print(f"Assistant: {response.content}")
    
    await send_task

import grpc
import time
import logging
 
logger = logging.getLogger(__name__)
 
 
class LoggingInterceptor(grpc.aio.UnaryUnaryClientInterceptor):
    """요청/응답 로깅 인터셉터"""
    
    async def intercept_unary_unary(
        self, continuation, client_call_details, request
    ):
        method = client_call_details.method
        start_time = time.monotonic()
        
        logger.info(f"gRPC 요청: {method}")
        
        try:
            response = await continuation(client_call_details, request)
            elapsed = (time.monotonic() - start_time) * 1000
            logger.info(f"gRPC 응답: {method} ({elapsed:.1f}ms)")
            return response
        except grpc.aio.AioRpcError as e:
            elapsed = (time.monotonic() - start_time) * 1000
            logger.error(
                f"gRPC 오류: {method} ({elapsed:.1f}ms) "
                f"code={e.code()} message={e.details()}"
            )
            raise
 
 
class RetryInterceptor(grpc.aio.UnaryUnaryClientInterceptor):
    """재시도 인터셉터"""
    
    RETRYABLE_CODES = {
        grpc.StatusCode.UNAVAILABLE,
        grpc.StatusCode.DEADLINE_EXCEEDED,
        grpc.StatusCode.RESOURCE_EXHAUSTED,
    }
    
    def __init__(self, max_retries: int = 3, base_delay: float = 0.1):
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    async def intercept_unary_unary(
        self, continuation, client_call_details, request
    ):
        last_error = None
        
        for attempt in range(self.max_retries + 1):
            try:
                return await continuation(client_call_details, request)
            except grpc.aio.AioRpcError as e:
                last_error = e
                if e.code() not in self.RETRYABLE_CODES:
                    raise
                
                delay = self.base_delay * (2 ** attempt)
                logger.warning(
                    f"재시도 {attempt + 1}/{self.max_retries}: "
                    f"code={e.code()}, delay={delay:.1f}s"
                )
                await asyncio.sleep(delay)
        
        raise last_error

class AuthInterceptor(grpc.aio.ServerInterceptor):
    """인증 인터셉터"""
    
    async def intercept_service(self, continuation, handler_call_details):
        metadata = dict(handler_call_details.invocation_metadata)
        api_key = metadata.get("x-api-key")
        
        if not api_key or not await validate_api_key(api_key):
            context = handler_call_details
            await context.abort(
                grpc.StatusCode.UNAUTHENTICATED,
                "유효하지 않은 API 키입니다",
            )
        
        return await continuation(handler_call_details)

# 클라이언트: 30초 데드라인 설정
response = await stub.Complete(
    request,
    timeout=30.0,
)
 
# 서버: 남은 데드라인 확인
async def Complete(self, request, context):
    remaining = context.time_remaining()
    
    if remaining < 5.0:
        # 시간이 부족하면 빠른 응답 모드
        return await fast_inference(request)
    
    return await full_inference(request)