Added ArcEventListener

Author: Robert Winkler

kotlin-wot-integration-tests/build.gradle.kts

@@ -6,9 +6,6 @@ plugins {
     id("io.spring.dependency-management") version "1.1.3"
 }
 
-tasks.named<Test>("test") {
-    enabled = false
-}
 
 dependencies {
     api(project(":kotlin-wot-binding-http"))

kotlin-wot-integration-tests/src/main/kotlin/integration/AgentConfiguration.kt

@@ -10,6 +10,7 @@ import ai.ancf.lmos.wot.thing.schema.WoTConsumedThing
 import kotlinx.coroutines.runBlocking
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
+import org.springframework.context.ApplicationEventPublisher
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 
@@ -78,14 +79,11 @@ class AgentConfiguration {
         tools = AllTools
     }
 
+    @Bean
+    fun agentEventListener(applicationEventPublisher: ApplicationEventPublisher) = ArcEventListener(applicationEventPublisher)
 
     @Bean
     fun discoverTools(functions: Functions, wot: Wot) : List<LLMFunction> = runBlocking {
-        //discoverTool(wot, functions, "http://localhost:8081/scraper")
-        /*
-        discoverTool(wot, functions, "https://plugfest.webthings.io/things/virtual-things-2",
-            BearerSecurityScheme())
-        */
         ThingToFunctionsMapper.exploreToolDirectory(wot, functions, "https://plugfest.webthings.io/.well-known/wot",
             BearerSecurityScheme())
     }

kotlin-wot-integration-tests/src/main/kotlin/integration/ArcEventListener.kt

@@ -0,0 +1,16 @@
+package ai.ancf.lmos.wot.integration
+
+import ai.ancf.lmos.arc.agents.events.Event
+import ai.ancf.lmos.arc.agents.events.EventHandler
+import ai.ancf.lmos.wot.JsonMapper
+import org.springframework.context.ApplicationEvent
+import org.springframework.context.ApplicationEventPublisher
+
+class ArcEventListener(private val applicationEventPublisher: ApplicationEventPublisher) : EventHandler<Event> {
+
+    override fun onEvent(event: Event) {
+        applicationEventPublisher.publishEvent(AgentEvent(JsonMapper.instance.writeValueAsString(event)))
+    }
+}
+
+data class AgentEvent(val message: String) : ApplicationEvent(message)

kotlin-wot-integration-tests/src/main/kotlin/integration/ChatAgent.kt

@@ -8,12 +8,17 @@ import ai.ancf.lmos.arc.agents.conversation.latest
 import ai.ancf.lmos.arc.agents.conversation.toConversation
 import ai.ancf.lmos.arc.agents.getAgentByName
 import ai.ancf.lmos.arc.core.getOrThrow
+import ai.ancf.lmos.wot.JsonMapper
+import ai.ancf.lmos.wot.protocol.ConversationalAgent
 import ai.ancf.lmos.wot.protocol.LMOSContext
 import ai.ancf.lmos.wot.protocol.LMOSThingType
 import ai.ancf.lmos.wot.reflection.annotations.*
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.launch
+import org.springframework.context.ApplicationListener
 import org.springframework.stereotype.Component
 
 
@@ -23,30 +28,34 @@ import org.springframework.stereotype.Component
 @VersionInfo(instance = "1.0.0")
 @Component
 class ChatAgent(agentProvider: AgentProvider, @Property(readOnly = true)
-                 val modelConfiguration: ModelConfiguration = ModelConfiguration(0.5, 50)) {
+                 val modelConfiguration: ModelConfiguration = ModelConfiguration(0.5, 50))
+    : ConversationalAgent<String, String>, ApplicationListener<AgentEvent> {
 
-    private val messageFlow = MutableSharedFlow<String>(replay = 1) // Replay last emitted value
+    private val agentEventFlow = MutableSharedFlow<String>(replay = 1) // Replay last emitted value
 
     val agent = agentProvider.getAgentByName("ChatAgent") as ChatAgent
 
-    @Property(title = "Observable Property", readOnly = true)
-    val observableProperty : MutableStateFlow<String> = MutableStateFlow("Hello World")
-
-    @Action(title = "Ask", description = "Ask the agent a question.")
-    suspend fun ask(chat : Chat) : String {
-        val assistantMessage = agent.execute(chat.message.toConversation(User("myId"))).getOrThrow().latest<AssistantMessage>() ?:
+    @Action(title = "Chat", description = "Ask the agent a question.")
+    @ActionInput(title = "The question", description = "A question")
+    @ActionOutput(title = "The question", description = "A question")
+    override suspend fun chat(message: String) : String {
+        val assistantMessage = agent.execute(message.toConversation(User("myId"))).getOrThrow().latest<AssistantMessage>() ?:
             throw RuntimeException("No Assistant response")
-        messageFlow.emit(assistantMessage.content)
         return assistantMessage.content
     }
 
-    @Event(title = "Generated message")
-    fun messageGenerated() : Flow<String> {
-        return messageFlow
+    @Event(title = "Agent Event", description = "An event from the agent.")
+    fun agentEvent() : Flow<String> {
+        return agentEventFlow
+    }
+
+    override fun onApplicationEvent(event: AgentEvent) {
+        CoroutineScope(Dispatchers.IO).launch {
+            agentEventFlow.emit(JsonMapper.instance.writeValueAsString(event))
+        }
     }
 }
 
 data class ModelConfiguration(val modelTemperature: Double, val maxTokens: Int)
 
-data class Chat(val message: String)
 

kotlin-wot-integration-tests/src/main/kotlin/integration/ConversationalAgent.kt

@@ -5,39 +5,40 @@ import ai.ancf.lmos.wot.Wot
 import ai.ancf.lmos.wot.binding.http.HttpProtocolClientFactory
 import ai.ancf.lmos.wot.binding.http.HttpsProtocolClientFactory
 import ai.ancf.lmos.wot.binding.websocket.WebSocketProtocolClientFactory
-import ai.ancf.lmos.wot.integration.Chat
+import ai.ancf.lmos.wot.protocol.ConsumedConversationalAgent
+import ai.ancf.lmos.wot.protocol.ConversationalAgent
+import ai.ancf.lmos.wot.protocol.EventListener
 import ai.ancf.lmos.wot.thing.ConsumedThing
-import ai.ancf.lmos.wot.thing.schema.InteractionListener
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 
 
-class ConversationalAgent private constructor(private val thing : ConsumedThing) {
+class WotConversationalAgent private constructor(private val thing : ConsumedThing) :
+    ConsumedConversationalAgent<String, String, String> {
 
     private val log : Logger = LoggerFactory.getLogger(ConversationalAgent::class.java)
 
     companion object {
-        suspend fun create(wot: Wot, url: String): ConversationalAgent {
-            return ConversationalAgent(wot.consume(wot.requestThingDescription(url)) as ConsumedThing)
+        suspend fun create(wot: Wot, url: String): ConsumedConversationalAgent<String, String, String> {
+            return WotConversationalAgent(wot.consume(wot.requestThingDescription(url)) as ConsumedThing)
         }
 
-        suspend fun create(url: String): ConversationalAgent {
+        suspend fun create(url: String): ConsumedConversationalAgent<String, String, String> {
             val wot = Wot.create(Servient(clientFactories = listOf(HttpProtocolClientFactory(), HttpsProtocolClientFactory(), WebSocketProtocolClientFactory())))
             return create(wot, url)
         }
     }
 
-    suspend fun chat(inputMessage: String): String {
-        val chat = Chat(inputMessage)
+    override suspend fun chat(message: String): String {
         return try {
-            thing.invokeAction(actionName = "ask", input = chat)
+            thing.invokeAction(actionName = "chat", input = message)
         } catch (e: Exception) {
             log.error("Failed to receive an answer", e)
             "Failed to receive an answer"
         }
     }
 
-    suspend fun consumeEvent(s: String, listener: InteractionListener) {
-        thing.subscribeEvent("contentRetrieved",listener)
+    override suspend fun consumeEvent(eventName: String, listener: EventListener<String>) {
+        thing.subscribeEvent(eventName, { listener.handleEvent(it.value().asText()) })
     }
 }

kotlin-wot-integration-tests/src/main/kotlin/integration/ResearcherAgent.kt

@@ -8,6 +8,7 @@ import ai.ancf.lmos.arc.agents.conversation.latest
 import ai.ancf.lmos.arc.agents.conversation.toConversation
 import ai.ancf.lmos.arc.agents.getAgentByName
 import ai.ancf.lmos.arc.core.getOrThrow
+import ai.ancf.lmos.wot.protocol.ConversationalAgent
 import ai.ancf.lmos.wot.protocol.LMOSContext
 import ai.ancf.lmos.wot.protocol.LMOSThingType
 import ai.ancf.lmos.wot.reflection.annotations.Action
@@ -23,15 +24,15 @@ import org.springframework.stereotype.Component
 @Context(prefix = LMOSContext.prefix, url = LMOSContext.url)
 @VersionInfo(instance = "1.0.0")
 @Component
-class ResearcherAgent(agentProvider: AgentProvider, ) {
+class ResearcherAgent(agentProvider: AgentProvider) : ConversationalAgent<String, String> {
 
     private val messageFlow = MutableSharedFlow<String>(replay = 1) // Replay last emitted value
 
     val agent = agentProvider.getAgentByName("ResearcherAgent") as ChatAgent
 
-    @Action(title = "Ask", description = "Ask the agent a question.")
-    suspend fun ask(chat : Chat) : String {
-        val assistantMessage = agent.execute(chat.message.toConversation(User("myId"))).getOrThrow().latest<AssistantMessage>() ?:
+    @Action(title = "Chat", description = "Ask the agent a question.")
+    override suspend fun chat(message : String ) : String {
+        val assistantMessage = agent.execute(message.toConversation(User("myId"))).getOrThrow().latest<AssistantMessage>() ?:
             throw RuntimeException("No Assistant response")
         return assistantMessage.content
     }

kotlin-wot-integration-tests/src/main/kotlin/integration/ScraperAgent.kt

@@ -8,6 +8,7 @@ import ai.ancf.lmos.arc.agents.conversation.latest
 import ai.ancf.lmos.arc.agents.conversation.toConversation
 import ai.ancf.lmos.arc.agents.getAgentByName
 import ai.ancf.lmos.arc.core.getOrThrow
+import ai.ancf.lmos.wot.protocol.ConversationalAgent
 import ai.ancf.lmos.wot.protocol.LMOSContext
 import ai.ancf.lmos.wot.protocol.LMOSThingType
 import ai.ancf.lmos.wot.reflection.annotations.*
@@ -21,22 +22,23 @@ import org.springframework.stereotype.Component
 @Context(prefix = LMOSContext.prefix, url = LMOSContext.url)
 @VersionInfo(instance = "1.0.0")
 @Component
-class ScraperAgent(agentProvider: AgentProvider, ) {
+class ScraperAgent(agentProvider: AgentProvider) : ConversationalAgent<String, Unit> {
 
     private val messageFlow = MutableSharedFlow<String>(replay = 1) // Replay last emitted value
 
     val agent = agentProvider.getAgentByName("ScraperAgent") as ChatAgent
 
-    @Action(title = "Ask", description = "Ask the agent a question.")
-    suspend fun ask(chat : Chat) {
-        val assistantMessage = agent.execute(chat.message.toConversation(User("myId"))).getOrThrow().latest<AssistantMessage>() ?:
-            throw RuntimeException("No Assistant response")
-        messageFlow.emit(assistantMessage.content)
-    }
 
     @Event(description = "HTML Content of the scraped web site")
     fun contentRetrieved() : Flow<String> {
         return messageFlow
     }
+
+    @Action(title = "chat", description = "Ask the agent a question.")
+    override suspend fun chat(message: String) {
+        val assistantMessage = agent.execute(message.toConversation(User("myId"))).getOrThrow().latest<AssistantMessage>() ?:
+        throw RuntimeException("No Assistant response")
+        messageFlow.emit(assistantMessage.content)
+    }
 }