[OPIK-610] multi models support infrastructure

apps/opik-backend/pom.xml

@@ -211,6 +211,10 @@
             <groupId>dev.langchain4j</groupId>
             <artifactId>langchain4j-open-ai</artifactId>
         </dependency>
+        <dependency>
+            <groupId>dev.langchain4j</groupId>
+            <artifactId>langchain4j-anthropic</artifactId>
+        </dependency>
 
         <!-- Test -->
 

apps/opik-backend/src/main/java/com/comet/opik/domain/ChatCompletionService.java

@@ -1,6 +1,7 @@
 package com.comet.opik.domain;
 
 import com.comet.opik.domain.llmproviders.LlmProviderFactory;
+import com.comet.opik.domain.llmproviders.LlmProviderService;
 import com.comet.opik.infrastructure.LlmProviderClientConfig;
 import com.comet.opik.utils.JsonUtils;
 import dev.ai4j.openai4j.chat.ChatCompletionRequest;
@@ -9,6 +10,9 @@
 import io.dropwizard.jersey.errors.ErrorMessage;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
+import jakarta.ws.rs.ClientErrorException;
+import jakarta.ws.rs.InternalServerErrorException;
+import jakarta.ws.rs.ServerErrorException;
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
 import org.glassfish.jersey.server.ChunkedOutput;
@@ -18,7 +22,6 @@
 import java.io.UncheckedIOException;
 import java.util.Optional;
 import java.util.function.Consumer;
-import java.util.function.Function;
 
 @Singleton
 @Slf4j
@@ -43,10 +46,20 @@ public ChatCompletionResponse create(@NonNull ChatCompletionRequest request, @No
 
         ChatCompletionResponse chatCompletionResponse;
         try {
+            log.info("Creating chat completions, workspaceId '{}', model '{}'", workspaceId, request.model());
             chatCompletionResponse = retryPolicy.withRetry(() -> llmProviderClient.generate(request, workspaceId));
+            log.info("Created chat completions, workspaceId '{}', model '{}'", workspaceId, request.model());
         } catch (RuntimeException runtimeException) {
             log.error(UNEXPECTED_ERROR_CALLING_LLM_PROVIDER, runtimeException);
-            throw llmProviderClient.mapRuntimeException(runtimeException);
+            if (llmProviderClient.getHttpExceptionClass().isInstance(runtimeException.getCause())) {
+                int statusCode = llmProviderClient.getHttpErrorStatusCode(runtimeException);
+                if (statusCode >= 400 && statusCode <= 499) {
+                    throw new ClientErrorException(runtimeException.getMessage(), statusCode);
+                }
+
+                throw new ServerErrorException(runtimeException.getMessage(), statusCode);
+            }
+            throw new InternalServerErrorException(UNEXPECTED_ERROR_CALLING_LLM_PROVIDER);
         }
 
         log.info("Created chat completions, workspaceId '{}', model '{}'", workspaceId, request.model());
@@ -59,12 +72,13 @@ public ChunkedOutput<String> createAndStreamResponse(
         var llmProviderClient = llmProviderFactory.getService(workspaceId, request.model());
 
         var chunkedOutput = new ChunkedOutput<String>(String.class, "\r\n");
+        log.info("Creating and streaming chat completions, workspaceId '{}', model '{}'", workspaceId, request.model());
         llmProviderClient.generateStream(
                 request,
                 workspaceId,
                 getMessageHandler(chunkedOutput),
                 getCloseHandler(chunkedOutput),
-                getErrorHandler(chunkedOutput, llmProviderClient::mapThrowableToError));
+                getErrorHandler(chunkedOutput, llmProviderClient));
         log.info("Created and streaming chat completions, workspaceId '{}', model '{}'", workspaceId, request.model());
         return chunkedOutput;
     }
@@ -104,11 +118,16 @@ private Runnable getCloseHandler(ChunkedOutput<String> chunkedOutput) {
     }
 
     private Consumer<Throwable> getErrorHandler(
-            ChunkedOutput<String> chunkedOutput, Function<Throwable, ErrorMessage> errorMapper) {
+            ChunkedOutput<String> chunkedOutput, LlmProviderService llmProviderClient) {
         return throwable -> {
             log.error(UNEXPECTED_ERROR_CALLING_LLM_PROVIDER, throwable);
 
-            var errorMessage = errorMapper.apply(throwable);
+            var errorMessage = new ErrorMessage(ChatCompletionService.UNEXPECTED_ERROR_CALLING_LLM_PROVIDER);
+            if (llmProviderClient.getHttpExceptionClass().isInstance(throwable)) {
+                errorMessage = new ErrorMessage(llmProviderClient.getHttpErrorStatusCode(throwable),
+                        throwable.getMessage());
+            }
+
             try {
                 getMessageHandler(chunkedOutput).accept(errorMessage);
             } catch (UncheckedIOException uncheckedIOException) {

apps/opik-backend/src/main/java/com/comet/opik/domain/llmproviders/Anthropic.java

@@ -0,0 +1,190 @@
+package com.comet.opik.domain.llmproviders;
+
+import com.comet.opik.infrastructure.LlmProviderClientConfig;
+import dev.ai4j.openai4j.chat.AssistantMessage;
+import dev.ai4j.openai4j.chat.ChatCompletionChoice;
+import dev.ai4j.openai4j.chat.ChatCompletionRequest;
+import dev.ai4j.openai4j.chat.ChatCompletionResponse;
+import dev.ai4j.openai4j.chat.Message;
+import dev.ai4j.openai4j.chat.Role;
+import dev.ai4j.openai4j.shared.Usage;
+import dev.langchain4j.data.message.AiMessage;
+import dev.langchain4j.data.message.Content;
+import dev.langchain4j.data.message.TextContent;
+import dev.langchain4j.data.message.UserMessage;
+import dev.langchain4j.model.StreamingResponseHandler;
+import dev.langchain4j.model.anthropic.internal.api.AnthropicContent;
+import dev.langchain4j.model.anthropic.internal.api.AnthropicCreateMessageRequest;
+import dev.langchain4j.model.anthropic.internal.api.AnthropicCreateMessageResponse;
+import dev.langchain4j.model.anthropic.internal.api.AnthropicMessage;
+import dev.langchain4j.model.anthropic.internal.api.AnthropicMessageContent;
+import dev.langchain4j.model.anthropic.internal.api.AnthropicRole;
+import dev.langchain4j.model.anthropic.internal.api.AnthropicTextContent;
+import dev.langchain4j.model.anthropic.internal.api.AnthropicToolChoice;
+import dev.langchain4j.model.anthropic.internal.client.AnthropicClient;
+import dev.langchain4j.model.anthropic.internal.client.AnthropicHttpException;
+import dev.langchain4j.model.output.Response;
+import jakarta.ws.rs.BadRequestException;
+import lombok.NonNull;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Consumer;
+
+public class Anthropic implements LlmProviderService {
+    private final LlmProviderClientConfig llmProviderClientConfig;
+    private final AnthropicClient anthropicClient;
+
+    public Anthropic(LlmProviderClientConfig llmProviderClientConfig, String apiKey) {
+        this.llmProviderClientConfig = llmProviderClientConfig;
+        this.anthropicClient = newClient(apiKey);
+    }
+
+    @Override
+    public ChatCompletionResponse generate(@NonNull ChatCompletionRequest request, @NonNull String workspaceId) {
+        var response = anthropicClient.createMessage(AnthropicCreateMessageRequest.builder()
+                .model(request.model())
+                .messages(request.messages().stream().map(this::mapMessage).toList())
+                .temperature(request.temperature())
+                .topP(request.topP())
+                .stopSequences(request.stop())
+                .maxTokens(request.maxTokens())
+                .toolChoice(AnthropicToolChoice.from(request.toolChoice().toString()))
+                .build());
+
+        return ChatCompletionResponse.builder()
+                .id(response.id)
+                .model(response.model)
+                .choices(response.content.stream().map(content -> mapContentToChoice(response, content))
+                        .toList())
+                .usage(Usage.builder()
+                        .promptTokens(response.usage.inputTokens)
+                        .completionTokens(response.usage.outputTokens)
+                        .totalTokens(response.usage.inputTokens + response.usage.outputTokens)
+                        .build())
+                .build();
+    }
+
+    @Override
+    public void generateStream(
+            @NonNull ChatCompletionRequest request,
+            @NonNull String workspaceId,
+            @NonNull Consumer<ChatCompletionResponse> handleMessage,
+            @NonNull Runnable handleClose, @NonNull Consumer<Throwable> handleError) {
+        return anthropicClient.createMessage(AnthropicCreateMessageRequest.builder()
+                .model(request.model())
+                .messages(request.messages().stream().map(this::mapMessage).toList())
+                .temperature(request.temperature())
+                .topP(request.topP())
+                .stopSequences(request.stop())
+                .maxTokens(request.maxTokens())
+                .toolChoice(AnthropicToolChoice.from(request.toolChoice().toString()))
+                .build(), new ChunkedResponseHandler(handleMessage, handleClose, handleError));
+
+        return ChatCompletionResponse.builder()
+                .id(response.id)
+                .model(response.model)
+                .choices(response.content.stream().map(content -> mapContentToChoice(response, content))
+                        .toList())
+                .usage(Usage.builder()
+                        .promptTokens(response.usage.inputTokens)
+                        .completionTokens(response.usage.outputTokens)
+                        .totalTokens(response.usage.inputTokens + response.usage.outputTokens)
+                        .build())
+                .build();
+    }
+
+    @Override
+    public Class<? extends Throwable> getHttpExceptionClass() {
+        return AnthropicHttpException.class;
+    }
+
+    @Override
+    public int getHttpErrorStatusCode(Throwable throwable) {
+        if (throwable instanceof AnthropicHttpException anthropicHttpException) {
+            return anthropicHttpException.statusCode();
+        }
+
+        return 500;
+    }
+
+    private AnthropicMessage mapMessage(Message message) {
+        if (message.role() == Role.ASSISTANT) {
+            return AnthropicMessage.builder()
+                    .role(AnthropicRole.ASSISTANT)
+                    .content(List.of(new AnthropicTextContent(((AssistantMessage) message).content())))
+                    .build();
+        } else if (message.role() == Role.USER) {
+            return AnthropicMessage.builder()
+                    .role(AnthropicRole.USER)
+                    .content(((UserMessage) message).contents().stream().map(this::mapMessageContent).toList())
+                    .build();
+        }
+
+        // Anthropic only supports assistant and user roles
+        throw new BadRequestException("not supported message role: " + message.role());
+    }
+
+    private AnthropicMessageContent mapMessageContent(Content content) {
+        if (content instanceof TextContent) {
+            return new AnthropicTextContent(((TextContent) content).text());
+        }
+
+        throw new BadRequestException("only text content is supported");
+    }
+
+    private ChatCompletionChoice mapContentToChoice(AnthropicCreateMessageResponse response, AnthropicContent content) {
+        return ChatCompletionChoice.builder()
+                .message(AssistantMessage.builder()
+                        .name(content.name)
+                        .content(content.text)
+                        .build())
+                .finishReason(response.stopReason)
+                .build();
+    }
+
+    private AnthropicClient newClient(String apiKey) {
+        var anthropicClientBuilder = AnthropicClient.builder();
+        Optional.ofNullable(llmProviderClientConfig.getOpenAiClient())
+                .map(LlmProviderClientConfig.OpenAiClientConfig::url)
+                .ifPresent(baseUrl -> {
+                    if (StringUtils.isNotBlank(baseUrl)) {
+                        anthropicClientBuilder.baseUrl(baseUrl);
+                    }
+                });
+        Optional.ofNullable(llmProviderClientConfig.getCallTimeout())
+                .ifPresent(callTimeout -> anthropicClientBuilder.timeout(callTimeout.toJavaDuration()));
+        return anthropicClientBuilder
+                .apiKey(apiKey)
+                .build();
+    }
+
+    private static class ChunkedResponseHandler implements StreamingResponseHandler<AiMessage> {
+        private final Consumer<ChatCompletionResponse> handleMessage;
+        private final Runnable handleClose;
+        private final Consumer<Throwable> handleError;
+
+        public ChunkedResponseHandler(
+                Consumer<ChatCompletionResponse> handleMessage, Runnable handleClose, Consumer<Throwable> handleError) {
+            this.handleMessage = handleMessage;
+            this.handleClose = handleClose;
+            this.handleError = handleError;
+        }
+
+        @Override
+        public void onNext(String s) {
+
+        }
+
+        @Override
+        public void onComplete(Response<AiMessage> response) {
+            StreamingResponseHandler.super.onComplete(response);
+        }
+
+        @Override
+        public void onError(Throwable throwable) {
+            handleError.accept(throwable);
+        }
+    }
+}

apps/opik-backend/src/main/java/com/comet/opik/domain/llmproviders/LlmProviderService.java

@@ -2,8 +2,6 @@
 
 import dev.ai4j.openai4j.chat.ChatCompletionRequest;
 import dev.ai4j.openai4j.chat.ChatCompletionResponse;
-import io.dropwizard.jersey.errors.ErrorMessage;
-import jakarta.ws.rs.WebApplicationException;
 import lombok.NonNull;
 
 import java.util.function.Consumer;
@@ -20,7 +18,7 @@ void generateStream(
             @NonNull Runnable handleClose,
             @NonNull Consumer<Throwable> handleError);
 
-    WebApplicationException mapRuntimeException(RuntimeException runtimeException);
+    Class<? extends Throwable> getHttpExceptionClass();
 
-    ErrorMessage mapThrowableToError(Throwable throwable);
+    int getHttpErrorStatusCode(Throwable runtimeException);
 }

apps/opik-backend/src/main/java/com/comet/opik/domain/llmproviders/OpenAi.java

@@ -1,17 +1,11 @@
 package com.comet.opik.domain.llmproviders;
 
-import com.comet.opik.domain.ChatCompletionService;
 import com.comet.opik.infrastructure.LlmProviderClientConfig;
 import dev.ai4j.openai4j.OpenAiClient;
 import dev.ai4j.openai4j.OpenAiHttpException;
 import dev.ai4j.openai4j.chat.ChatCompletionRequest;
 import dev.ai4j.openai4j.chat.ChatCompletionResponse;
-import io.dropwizard.jersey.errors.ErrorMessage;
 import jakarta.inject.Inject;
-import jakarta.ws.rs.ClientErrorException;
-import jakarta.ws.rs.InternalServerErrorException;
-import jakarta.ws.rs.ServerErrorException;
-import jakarta.ws.rs.WebApplicationException;
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -32,10 +26,7 @@ public OpenAi(LlmProviderClientConfig llmProviderClientConfig, String apiKey) {
 
     @Override
     public ChatCompletionResponse generate(@NonNull ChatCompletionRequest request, @NonNull String workspaceId) {
-        log.info("Creating chat completions, workspaceId '{}', model '{}'", workspaceId, request.model());
-        var chatCompletionResponse = openAiClient.chatCompletion(request).execute();;
-        log.info("Created chat completions, workspaceId '{}', model '{}'", workspaceId, request.model());
-        return chatCompletionResponse;
+        return openAiClient.chatCompletion(request).execute();
     }
 
     @Override
@@ -45,35 +36,25 @@ public void generateStream(
             @NonNull Consumer<ChatCompletionResponse> handleMessage,
             @NonNull Runnable handleClose,
             @NonNull Consumer<Throwable> handleError) {
-        log.info("Creating and streaming chat completions, workspaceId '{}', model '{}'", workspaceId, request.model());
         openAiClient.chatCompletion(request)
                 .onPartialResponse(handleMessage)
                 .onComplete(handleClose)
                 .onError(handleError)
                 .execute();
-        log.info("Created and streaming chat completions, workspaceId '{}', model '{}'", workspaceId, request.model());
     }
 
     @Override
-    public WebApplicationException mapRuntimeException(RuntimeException runtimeException) {
-        if (runtimeException.getCause() instanceof OpenAiHttpException openAiHttpException) {
-            if (openAiHttpException.code() >= 400 && openAiHttpException.code() <= 499) {
-                return new ClientErrorException(openAiHttpException.getMessage(), openAiHttpException.code());
-            }
-
-            return new ServerErrorException(openAiHttpException.getMessage(), openAiHttpException.code());
-        }
-
-        return new InternalServerErrorException(ChatCompletionService.UNEXPECTED_ERROR_CALLING_LLM_PROVIDER);
+    public Class<? extends RuntimeException> getHttpExceptionClass() {
+        return OpenAiHttpException.class;
     }
 
     @Override
-    public ErrorMessage mapThrowableToError(Throwable throwable) {
+    public int getHttpErrorStatusCode(Throwable throwable) {
         if (throwable instanceof OpenAiHttpException openAiHttpException) {
-            return new ErrorMessage(openAiHttpException.code(), openAiHttpException.getMessage());
+            return openAiHttpException.code();
         }
 
-        return new ErrorMessage(ChatCompletionService.UNEXPECTED_ERROR_CALLING_LLM_PROVIDER);
+        return 500;
     }
 
     /**