https://kotlinlang.org logo
Join Slack
Powered by
Hi! Could someone explain if this way of paralleli...
# koog-agentic-framework
j
Hi! Could someone explain if this way of parallelizing work on a node is correct? As you can see I use Arrow`s parMap to parallelize an unbounded collection with concurrency 10. The problem I've found is that the 10 first calls are executed at the same time, but the rest from there seem to be executed one by one (check the log timestamps)
Copy code
val nodeFindId by node<List<Document>, List<Document>>("find_id") { b ->
    val list = b.withIndex().parMap(concurrency = 10) { b ->
        logger.info("Extracting ID from document ${b.index + 1}/${b.size} - ${b.value.identifier}")
        withContext(IO) {
            llm.writeSession {
                rewritePrompt {
                    prompt("id_search_prompt") {
                        system(
                            """You are provided a legal text from which you have to extract the ID identifier.
                          A ID identifier is a numeric code that identifies a specific grant call in the ID system.
                          If there is no ID identifier in the text, respond with 'NOT_FOUND'.
                          Respond only with the ID identifier or 'NOT_FOUND', do not add any additional text
                        """
                        )
                        user("Legal text: \n\n${b.value.pages.joinToString("\n\n")}")
                    }
                }
                val id = requestLLM().content
                b.value.copy(id = id)
            }
        }
    }
    list.filter { it.id != null && it.id != "NOT_FOUND" }
}
2025-11-04T10:28:37.784+01:00  INFO 67031 --- [-agent] [-2 @coroutine#2] agent.AgentFactory     : Extracting ID from  document 0/97  - 2025-40119
2025-11-04T10:28:37.799+01:00  INFO 67031 --- [-agent] [-2 @coroutine#3] agent.AgentFactory     : Extracting ID from  document 1/97  - 2025-40120
2025-11-04T10:28:37.799+01:00  INFO 67031 --- [-agent] [-2 @coroutine#4] agent.AgentFactory     : Extracting ID from  document 2/97  - 2025-40121
2025-11-04T10:28:37.799+01:00  INFO 67031 --- [-agent] [-2 @coroutine#5] agent.AgentFactory     : Extracting ID from  document 3/97  - 2025-40122
2025-11-04T10:28:37.799+01:00  INFO 67031 --- [-agent] [-2 @coroutine#6] agent.AgentFactory     : Extracting ID from  document 4/97  - 2025-40123
2025-11-04T10:28:37.799+01:00  INFO 67031 --- [-agent] [-2 @coroutine#7] agent.AgentFactory     : Extracting ID from  document 5/97  - 2025-40124
2025-11-04T10:28:37.799+01:00  INFO 67031 --- [-agent] [-2 @coroutine#8] agent.AgentFactory     : Extracting ID from  document 6/97  - 2025-40125
2025-11-04T10:28:37.799+01:00  INFO 67031 --- [-agent] [-2 @coroutine#9] agent.AgentFactory     : Extracting ID from  document 7/97  - 2025-40126
2025-11-04T10:28:37.800+01:00  INFO 67031 --- [-agent] [2 @coroutine#10] agent.AgentFactory     : Extracting ID from  document 8/97  - 2025-40127
2025-11-04T10:28:37.800+01:00  INFO 67031 --- [-agent] [2 @coroutine#11] agent.AgentFactory     : Extracting ID from  document 9/97  - 2025-40128
2025-11-04T10:28:38.964+01:00  INFO 67031 --- [-agent] [2 @coroutine#12] agent.AgentFactory     : Extracting ID from  document 10/97 - 2025-40129
2025-11-04T10:28:40.041+01:00  INFO 67031 --- [-agent] [1 @coroutine#13] agent.AgentFactory     : Extracting ID from  document 11/97 - 2025-40130
2025-11-04T10:28:41.862+01:00  INFO 67031 --- [-agent] [1 @coroutine#14] agent.AgentFactory     : Extracting ID from  document 12/97 - 2025-40131
v
j
yep, the point is paralell node execution is for a predetermined number of paths 
val nodeName by parallel<Input, Output>(
firstNode, secondNode, thirdNode /* Add more nodes if needed */
)
In my case I have a list of an undetermined number of elements and for each of them I have to make a parallel request. 🤔
a
Which model are you using? We’ve observed similar behavior with GPT-5. Even though the requests were made in a truly parallel manner, seems like OpenAI was “throttling” requests on their side. The problem disappeared when a different model was used, e.g. GPT-4 or other provider
v
You can do something like this in order to use `parallel`:
Copy code
val nodes = (0 until Random.nextInt(100, 200)).map { i ->
    val ithNode by node<String, String>(name = "n_$i") { input -> "$input $i" }
    ithNode
}


val parallelComputation by parallel<String, String>(*nodes.toTypedArray()) {
    selectByMax { it.length } // or some other selection strategy
}
*
is what does the trick in Kotlin to pass an array instead of vararg argument
j
@Andrey Bragin im using OpenAIModels.CostOptimized.GPT4_1Mini, I'll try 4o just in case @Vadim Briliantov nice trick hehe,
8 Views
Open in Slack
PreviousNext
Powered by