4장: gRPC Streaming — 고성능 백엔드 통신

[HTTP/1.1]
연결 1: 요청A ──────> 응답A
연결 2: 요청B ──────> 응답B    (별도 연결 필요)
연결 3: 요청C ──────> 응답C
 
[HTTP/2 멀티플렉싱]
연결 1: 스트림1(요청A) ──> 스트림2(요청B) ──> 스트림1(응답A)
         스트림3(요청C) ──> 스트림2(응답B) ──> 스트림3(응답C)
         (하나의 연결에서 동시 처리)

syntax = "proto3";
 
package inference;
 
service InferenceService {
  // 단항 RPC
  rpc Predict(PredictRequest) returns (PredictResponse);
  
  // 서버 스트리밍 RPC
  rpc StreamPredict(PredictRequest) returns (stream TokenResponse);
  
  // 클라이언트 스트리밍 RPC
  rpc StreamInput(stream AudioChunk) returns (TranscriptResponse);
  
  // 양방향 스트리밍 RPC
  rpc Converse(stream ConversationInput) returns (stream ConversationOutput);
}
 
message PredictRequest {
  string model_id = 1;
  string prompt = 2;
  int32 max_tokens = 3;
  float temperature = 4;
}
 
message TokenResponse {
  string token = 1;
  int32 index = 2;
  bool is_finished = 3;
  TokenMetadata metadata = 4;
}
 
message TokenMetadata {
  float log_probability = 1;
  int64 latency_ms = 2;
}
 
message AudioChunk {
  bytes data = 1;
  int32 sample_rate = 2;
  string encoding = 3;
}

클라이언트 ── 요청 ──> 서버
클라이언트 <── 응답 ── 서버

클라이언트 ── 요청 ──> 서버
클라이언트 <── 토큰1 ── 서버
클라이언트 <── 토큰2 ── 서버
클라이언트 <── 토큰3 ── 서버
클라이언트 <── 완료 ── 서버

클라이언트 ── 오디오청크1 ──> 서버
클라이언트 ── 오디오청크2 ──> 서버
클라이언트 ── 오디오청크3 ──> 서버
클라이언트 <── 전체 인식 결과 ── 서버

import grpc
from concurrent import futures
import inference_pb2
import inference_pb2_grpc
 
 
class InferenceServicer(inference_pb2_grpc.InferenceServiceServicer):
    """gRPC 추론 서비스 구현"""
 
    def StreamPredict(self, request, context):
        """서버 스트리밍: 토큰 단위로 응답"""
        model_id = request.model_id
        prompt = request.prompt
 
        # 모델 로드 및 추론
        model = get_model(model_id)
        generator = model.generate(
            prompt,
            max_tokens=request.max_tokens,
            temperature=request.temperature,
        )
 
        for index, (token, log_prob) in enumerate(generator):
            # 클라이언트가 연결을 끊었는지 확인
            if context.is_active():
                yield inference_pb2.TokenResponse(
                    token=token,
                    index=index,
                    is_finished=False,
                    metadata=inference_pb2.TokenMetadata(
                        log_probability=log_prob,
                        latency_ms=get_latency_ms(),
                    ),
                )
            else:
                # 클라이언트가 취소함
                return
 
        # 완료 신호
        yield inference_pb2.TokenResponse(
            token="",
            index=-1,
            is_finished=True,
        )
 
    def Converse(self, request_iterator, context):
        """양방향 스트리밍: 실시간 대화"""
        for request in request_iterator:
            if not context.is_active():
                return
 
            # 입력 처리 및 응답 생성
            response = process_input(request)
            for token in response:
                yield inference_pb2.ConversationOutput(
                    text=token,
                    speaker="assistant",
                )
 
 
def serve():
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=10),
        options=[
            ("grpc.max_send_message_length", 50 * 1024 * 1024),
            ("grpc.max_receive_message_length", 50 * 1024 * 1024),
            ("grpc.keepalive_time_ms", 30000),
            ("grpc.keepalive_timeout_ms", 10000),
        ],
    )
    inference_pb2_grpc.add_InferenceServiceServicer_to_server(
        InferenceServicer(), server
    )
    server.add_insecure_port("[::]:50051")
    server.start()
    server.wait_for_termination()

package main
 
import (
	"context"
	"io"
	"log"
 
	pb "inference/proto"
	"google.golang.org/grpc"
)
 
type InferenceRouter struct {
	pb.UnimplementedInferenceServiceServer
	backends []pb.InferenceServiceClient
}
 
func (r *InferenceRouter) StreamPredict(
	req *pb.PredictRequest,
	stream pb.InferenceService_StreamPredictServer,
) error {
	// 최적의 백엔드 선택 (로드 기반)
	backend := r.selectBackend(req.ModelId)
 
	// 백엔드에 스트리밍 요청
	ctx := stream.Context()
	backendStream, err := backend.StreamPredict(ctx, req)
	if err != nil {
		return err
	}
 
	// 백엔드 응답을 클라이언트에 중계
	for {
		token, err := backendStream.Recv()
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}
 
		if err := stream.Send(token); err != nil {
			return err
		}
	}
}
 
func (r *InferenceRouter) selectBackend(
	modelId string,
) pb.InferenceServiceClient {
	// 라운드 로빈, 최소 연결, GPU 사용률 기반 등
	// 실제 구현은 모니터링 메트릭에 기반
	return r.backends[0]
}