Page History

AI의 활용되는 MCP(Model Context Protocol) 에대해 설명하고~ 이것을 시스템 아키텍처 적용할때 고려사항?

리액티브 스트림을 반영해 지능형 AI 서비스 아키텍처를 보강해죠

LLM 호출 및 AI 모듈과 연결할 때, LLM을 안정적이게 사용할수 있는 backpressure 를 이용하는 방법의 샘플 코드를 akka stream을 이용한 샘플코드를 작성해 스칼라/자바/코틀린 버전으로 각각 작성

import akka.actor.ActorSystem
import akka.stream.scaladsl.{Flow, Sink, Source}
import akka.stream.ActorMaterializer
import scala.concurrent.Future
import scala.concurrent.duration._
import scala.util.Random

object LlmBackpressureScala extends App {
  implicit val system = ActorSystem("llm-system")
  implicit val ec = system.dispatcher
  implicit val materializer = ActorMaterializer()

  // 가상의 LLM API 호출 (느린 호출)
  def callLLM(input: String): Future[String] = Future {
    Thread.sleep(Random.nextInt(1000)) // 임의의 지연
    s"LLM Response for [$input]"
  }

  val stream = Source(1 to 100)
    .throttle(5, 1.second) // 입력 제한
    .map(i => s"input-$i")
    .via(Flow[String].mapAsync(parallelism = 4)(callLLM)) // backpressure 적용
    .runWith(Sink.foreach(resp => println(s"[LLM] $resp")))

  stream.onComplete(_ => system.terminate())
}

import akka.Done;
import akka.actor.ActorSystem;
import akka.stream.ActorMaterializer;
import akka.stream.javadsl.*;
import scala.concurrent.duration.Duration;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.IntStream;

public class LlmBackpressureJava {
    public static void main(String[] args) {
        ActorSystem system = ActorSystem.create("llm-system");
        ActorMaterializer materializer = ActorMaterializer.create(system);

        Flow<String, String, NotUsed> llmFlow = Flow.of(String.class)
            .mapAsync(4, input -> callLLM(input));

        Source.from(IntStream.rangeClosed(1, 100).boxed().toList())
            .throttle(5, Duration.create(1, "second"))
            .map(i -> "input-" + i)
            .via(llmFlow)
            .runWith(Sink.foreach(resp -> System.out.println("[LLM] " + resp)), materializer)
            .whenComplete((done, exc) -> system.terminate());
    }

    static CompletionStage<String> callLLM(String input) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(ThreadLocalRandom.current().nextInt(1000));
            } catch (InterruptedException ignored) {}
            return "LLM Response for [" + input + "]";
        });
    }
}

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.javadsl.*
import scala.concurrent.duration.Duration
import java.util.concurrent.CompletableFuture
import java.util.concurrent.ThreadLocalRandom

fun main() {
    val system = ActorSystem.create("llm-system")
    val materializer = ActorMaterializer.create(system)

    val llmFlow = Flow.of(String::class.java)
        .mapAsync(4) { input -> callLLM(input) }

    val inputList = (1..100).map { "input-$it" }

    Source.from(inputList)
        .throttle(5, Duration.create(1, "second"))
        .via(llmFlow)
        .runWith(Sink.foreach { println("[LLM] $it") }, materializer)
        .whenComplete { _, _ -> system.terminate() }
}

fun callLLM(input: String): CompletableFuture<String> {
    return CompletableFuture.supplyAsync {
        Thread.sleep(ThreadLocalRandom.current().nextInt(1000))
        "LLM Response for [$input]"
    }
}

callLLM 을 HuggingFace API 호출한버전?

def callLLM(input: String): Future[String] = {
  import scala.concurrent.ExecutionContext.Implicits.global
  import sttp.client3._
  import sttp.client3.circe._
  import io.circe._, io.circe.parser._, io.circe.syntax._

  val backend = HttpURLConnectionBackend()
  val token = "<YOUR_HF_TOKEN>"
  val model = "gpt2"

  val bodyJson = Map("inputs" -> input).asJson.noSpaces

  Future {
    val request = basicRequest
      .post(uri"https://api-inference.huggingface.co/models/$model")
      .header("Authorization", s"Bearer $token")
      .body(bodyJson)
      .send(backend)

    val responseBody = request.body match {
      case Right(json) => parse(json).flatMap(_.hcursor.downArray.get[String]("generated_text")).getOrElse("Parse failed")
      case Left(err)   => s"Error: $err"
    }

    responseBody
  }
}

cold-start 에서 점진적으로 사용량을 높여주는것에 대비해 트래픽대비 호출성공률이 높은 backpresure 의 세부적인 전략 코드작성

import akka.actor.ActorSystem
import akka.stream._
import akka.stream.scaladsl._
import akka.pattern.after
import scala.concurrent.duration._
import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Failure, Success}
import scala.util.Random

object LlmBackpressureStrategy extends App {
  implicit val system = ActorSystem("llm-coldstart")
  implicit val ec: ExecutionContext = system.dispatcher
  implicit val mat = ActorMaterializer()

  val totalRequests = 100
  val baseDelay = 500.millis

  // 동적으로 호출 빈도를 점진적으로 증가시키는 Source
  val rampUpSource = Source(1 to totalRequests)
    .throttle(1, baseDelay, 1, ThrottleMode.Shaping) // 초기에는 1개/0.5초
    .via(rampUpFlow(10, 100.millis)) // 점진적으로 빠르게

  // 점진적 호출 속도 증가를 위한 커스텀 Flow
  def rampUpFlow(steps: Int, stepInterval: FiniteDuration): Flow[Int, Int, _] = {
    Flow[Int].throttle(1, stepInterval).conflateWithSeed(i => (i, 1)) {
      case ((_, cnt), newValue) => (newValue, cnt + 1)
    }.map(_._1)
  }

  // LLM API 호출 (시뮬레이션)
  def callLLM(input: String): Future[String] = {
    val delay = if (Random.nextDouble() < 0.2) 3000 else 300 // 일부 cold start 발생
    Future {
      Thread.sleep(delay)
      if (Random.nextDouble() < 0.1) throw new RuntimeException("Simulated failure")
      s"Response for [$input] after $delay ms"
    }
  }

  // Timeout 처리 + Retry
  def withRetryAndTimeout(input: String, retries: Int): Future[String] = {
    val timeoutFuture = after(2.seconds, system.scheduler)(Future.failed(new TimeoutException("Timeout")))
    val attempt = Future.firstCompletedOf(Seq(callLLM(input), timeoutFuture))

    attempt.recoverWith {
      case _ if retries > 0 =>
        println(s"[Retrying] $input")
        after(300.millis, system.scheduler)(withRetryAndTimeout(input, retries - 1))
    }
  }

  rampUpSource
    .map(i => s"input-$i")
    .via(Flow[String].mapAsync(parallelism = 4)(input => withRetryAndTimeout(input, 2)))
    .runWith(Sink.foreach(result => println(s"[OK] $result")))
    .onComplete { _ =>
      println("Stream finished.")
      system.terminate()
    }
}