Page History

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
  }
}