Forums pt 2

Author: Skobeltsyn

src/main/kotlin/agents_engine/composition/forum/Forum.kt

@@ -1,10 +1,40 @@
 package agents_engine.composition.forum
 
 import agents_engine.core.*
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.async
+import kotlinx.coroutines.runBlocking
 
 class Forum<IN, OUT>(
     val agents: List<Agent<*, *>>,
-)
+) {
+    private var mentionListener: ((agentName: String, output: Any?) -> Unit)? = null
+
+    fun onMentionEmitted(block: (agentName: String, output: Any?) -> Unit) {
+        mentionListener = block
+    }
+
+    @Suppress("UNCHECKED_CAST")
+    operator fun invoke(input: IN): OUT = runBlocking(Dispatchers.Default) {
+        val forumInput = input as Any?
+        val participants = agents.dropLast(1)
+        val captain = agents.last() as Agent<Any?, OUT>
+
+        // All participants process the input concurrently
+        participants.map { agent ->
+            async {
+                val output = (agent as Agent<Any?, Any?>)(forumInput)
+                mentionListener?.invoke(agent.name, output)
+                output
+            }
+        }.map { it.await() }
+
+        // Captain delivers the final verdict
+        val verdict = captain(forumInput)
+        mentionListener?.invoke(captain.name, verdict)
+        verdict
+    }
+}
 
 operator fun <A, B, C> Agent<A, B>.times(other: Agent<*, C>): Forum<A, C> {
     this.markPlaced("forum")

src/main/kotlin/agents_engine/composition/pipeline/Pipeline.kt

@@ -26,11 +26,11 @@ infix fun <A, B, C> Pipeline<A, B>.then(other: Agent<B, C>): Pipeline<A, C> {
 
 infix fun <A, B, C> Agent<A, B>.then(other: Forum<B, C>): Pipeline<A, C> {
     this.markPlaced("pipeline")
-    return Pipeline(listOf(this) + other.agents) { error("Forum execution not yet implemented") }
+    return Pipeline(listOf(this) + other.agents) { input -> other(this(input)) }
 }
 
 infix fun <A, B, C> Pipeline<A, B>.then(other: Forum<B, C>): Pipeline<A, C> {
-    return Pipeline(agents + other.agents) { error("Forum execution not yet implemented") }
+    return Pipeline(agents + other.agents) { input -> other(this(input)) }
 }
 
 infix fun <A, B, C> Pipeline<A, B>.then(other: Pipeline<B, C>): Pipeline<A, C> {