Merge pull request #1840 from Wal8800/anthropic-llm-obs

enhance: allow anthropic recorder to add chat model listener

Author: geoand

model-providers/anthropic/deployment/src/main/java/io/quarkiverse/langchain4j/anthropic/deployment/AnthropicProcessor.java

@@ -8,9 +8,13 @@
 import jakarta.enterprise.context.ApplicationScoped;
 
 import org.jboss.jandex.AnnotationInstance;
+import org.jboss.jandex.ClassType;
+import org.jboss.jandex.ParameterizedType;
+import org.jboss.jandex.Type;
 
 import io.quarkiverse.langchain4j.ModelName;
 import io.quarkiverse.langchain4j.anthropic.runtime.AnthropicRecorder;
+import io.quarkiverse.langchain4j.deployment.DotNames;
 import io.quarkiverse.langchain4j.deployment.items.ChatModelProviderCandidateBuildItem;
 import io.quarkiverse.langchain4j.deployment.items.SelectedChatModelProviderBuildItem;
 import io.quarkiverse.langchain4j.runtime.NamedConfigUtil;
@@ -56,7 +60,9 @@ void generateBeans(AnthropicRecorder recorder, List<SelectedChatModelProviderBui
                         .setRuntimeInit()
                         .defaultBean()
                         .scope(ApplicationScoped.class)
-                        .supplier(recorder.chatModel(configName));
+                        .addInjectionPoint(ParameterizedType.create(DotNames.CDI_INSTANCE,
+                                new Type[] { ClassType.create(DotNames.CHAT_MODEL_LISTENER) }, null))
+                        .createWith(recorder.chatModel(configName));
 
                 addQualifierIfNecessary(builder, configName);
                 beanProducer.produce(builder.done());
@@ -66,7 +72,9 @@ void generateBeans(AnthropicRecorder recorder, List<SelectedChatModelProviderBui
                         .setRuntimeInit()
                         .defaultBean()
                         .scope(ApplicationScoped.class)
-                        .supplier(recorder.streamingChatModel(configName));
+                        .addInjectionPoint(ParameterizedType.create(DotNames.CDI_INSTANCE,
+                                new Type[] { ClassType.create(DotNames.CHAT_MODEL_LISTENER) }, null))
+                        .createWith(recorder.streamingChatModel(configName));
 
                 addQualifierIfNecessary(streamingBuilder, configName);
                 beanProducer.produce(streamingBuilder.done());

model-providers/anthropic/runtime/src/main/java/io/quarkiverse/langchain4j/anthropic/runtime/AnthropicRecorder.java

@@ -3,7 +3,11 @@
 import static io.quarkiverse.langchain4j.runtime.OptionalUtil.firstOrDefault;
 
 import java.time.Duration;
-import java.util.function.Supplier;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import jakarta.enterprise.inject.Instance;
+import jakarta.enterprise.util.TypeLiteral;
 
 import org.jboss.logging.Logger;
 
@@ -13,9 +17,11 @@
 import dev.langchain4j.model.chat.DisabledChatModel;
 import dev.langchain4j.model.chat.DisabledStreamingChatModel;
 import dev.langchain4j.model.chat.StreamingChatModel;
+import dev.langchain4j.model.chat.listener.ChatModelListener;
 import io.quarkiverse.langchain4j.anthropic.runtime.config.ChatModelConfig;
 import io.quarkiverse.langchain4j.anthropic.runtime.config.LangChain4jAnthropicConfig;
 import io.quarkiverse.langchain4j.runtime.NamedConfigUtil;
+import io.quarkus.arc.SyntheticCreationalContext;
 import io.quarkus.runtime.RuntimeValue;
 import io.quarkus.runtime.annotations.Recorder;
 import io.smallrye.config.ConfigValidationException;
@@ -24,6 +30,9 @@
 public class AnthropicRecorder {
     private static final Logger LOG = Logger.getLogger(AnthropicRecorder.class);
 
+    private static final TypeLiteral<Instance<ChatModelListener>> CHAT_MODEL_LISTENER_TYPE_LITERAL = new TypeLiteral<>() {
+    };
+
     private static final String DUMMY_KEY = "dummy";
 
     private final RuntimeValue<LangChain4jAnthropicConfig> runtimeConfig;
@@ -32,7 +41,7 @@ public AnthropicRecorder(RuntimeValue<LangChain4jAnthropicConfig> runtimeConfig)
         this.runtimeConfig = runtimeConfig;
     }
 
-    public Supplier<ChatModel> chatModel(String configName) {
+    public Function<SyntheticCreationalContext<ChatModel>, ChatModel> chatModel(String configName) {
         var anthropicConfig = correspondingAnthropicConfig(runtimeConfig.getValue(), configName);
 
         if (anthropicConfig.enableIntegration()) {
@@ -89,23 +98,26 @@ public Supplier<ChatModel> chatModel(String configName) {
                 builder.sendThinking(thinkingConfig.sendThinking().get());
             }
 
-            return new Supplier<>() {
+            return new Function<>() {
                 @Override
-                public ChatModel get() {
+                public ChatModel apply(SyntheticCreationalContext<ChatModel> context) {
+                    builder.listeners(context.getInjectedReference(CHAT_MODEL_LISTENER_TYPE_LITERAL).stream()
+                            .collect(Collectors.toList()));
                     return builder.build();
                 }
             };
         } else {
-            return new Supplier<>() {
+            return new Function<>() {
                 @Override
-                public ChatModel get() {
+                public ChatModel apply(SyntheticCreationalContext<ChatModel> context) {
                     return new DisabledChatModel();
                 }
             };
         }
     }
 
-    public Supplier<StreamingChatModel> streamingChatModel(String configName) {
+    public Function<SyntheticCreationalContext<StreamingChatModel>, StreamingChatModel> streamingChatModel(
+            String configName) {
         var anthropicConfig = correspondingAnthropicConfig(runtimeConfig.getValue(), configName);
 
         if (anthropicConfig.enableIntegration()) {
@@ -160,16 +172,18 @@ public Supplier<StreamingChatModel> streamingChatModel(String configName) {
                 builder.sendThinking(thinkingConfig.sendThinking().get());
             }
 
-            return new Supplier<>() {
+            return new Function<>() {
                 @Override
-                public StreamingChatModel get() {
+                public StreamingChatModel apply(SyntheticCreationalContext<StreamingChatModel> context) {
+                    builder.listeners(context.getInjectedReference(CHAT_MODEL_LISTENER_TYPE_LITERAL).stream()
+                            .collect(Collectors.toList()));
                     return builder.build();
                 }
             };
         } else {
-            return new Supplier<>() {
+            return new Function<>() {
                 @Override
-                public StreamingChatModel get() {
+                public StreamingChatModel apply(SyntheticCreationalContext<StreamingChatModel> context) {
                     return new DisabledStreamingChatModel();
                 }
             };