Hi, I'm facing an issue when a `Tool` *returns a `...

# koog-agentic-framework

[tung]

Hi, I'm facing an issue when a Tool returns a Flow while using subgraphWithTask Currently, subgraphWithTask throws an error: " `Serializer for subclass 'class kotlinx.coroutines.flow.FlowKt__ErrorsKt$catch$$inlined$unsafeFlow$1' is not found in the polymorphic scope of 'Flow'.`_"_. Is there any way to make tools that return Flow work with subgraphWithTask?
Or do I need to create a custom subgraph and use the Streaming api

interface GetSensorWidgets: ToolSet {
    @Tool("get_widgets")
    @LLMDescription("get all widget belong to dashboard")
    fun invoke(p1: GetWidgetsParams): Flow<DomainAppResult<WidgetProcessingResult>>
}

I've create a custom subgraph use nodeLLMRequestStructured but it doesn't work — the tool is not triggered.

fun AIAgentGraphStrategyBuilder<UserInput, String>.getWidgetsSubGraph(
    widgetsTool: List<Tool<ToolFromCallable.VarArgs, *>>,
    dashboardTools: List<Tool<ToolFromCallable.VarArgs, *>>
): AIAgentSubgraphDelegate<ClassifiedDashboardRequest, String> {
    return subgraph<ClassifiedDashboardRequest, String>(
        name = "get-dashboard-detail-graph",
        tools = widgetsTool + dashboardTools
    ) {
        val sendInstructions by node<ClassifiedDashboardRequest, ClassifiedDashboardRequest> { input ->
            llm.writeSession {
                updatePrompt {
                    user {
                        xml {
                            markdown {
                                h1("Requirements:")
                                text("Get detail information for specific dashboard")
                                h2("Tool usage guidelines")
                                bulleted {
                                    item("Get dashboards detail:") {
                                        item("Use get_widgets tool to get dashboard detail information.")
                                        item("This tool require dashboard id,you should ask the user dashboard name then find dashboard id by dashboard name the user provided.")
                                        item("You may need to call get_widgets multiple times for different dashboards.")
                                        item("If you don't have information about dashboards use get_dashboards to get dashboards and find the dashboard id by dashboard name.")
                                        item("Don't ask for user information like IDs, as users may not be familiar with IDs.")
                                    }
                                }
                                h2("After fetching data, provide a clear summary of what you found.")
                            }
                            tag("user_input") {
                                +"""
                            ${json.encodeToString(input)}
                            """.trimIndent()
                            }
                        }
                    }
                }
            }
            input
        }
        
        val processDashboardResult by node<Result<StructuredResponse<DomainAppResult<WidgetProcessingResult>>>, String> { result ->
            when {
                result.isSuccess -> {
                    val response = result.getOrNull()?.structure
                    when (response) {
                        is DomainAppResult.Success -> {
                            response.data?.let { widgets ->
                                println("widget_detail_response: $widgets")
                                """
                                    Widget Detail:${
                                    widgets.panels.forEach {
                                        """
                                    Name: ${it.name}
                                    Type: ${it.widgetTemplate}
                                    \n
                                    """.trimIndent()
                                    }
                                }
                                """.trimIndent()
                            } ?: "Empty dashboard data received"
                        }

                        is DomainAppResult.Error -> {
                            "Error: ${response.error.title} - ${response.error.message}"
                        }

                        null -> "No valid response received"
                    }
                }

                result.isFailure -> {
                    "Failed to get dashboard: ${result.exceptionOrNull()?.message}"
                }

                else -> "Unknown result state"
            }
        }

        val getDashboardDetails by nodeLLMRequestStructured<DomainAppResult<WidgetProcessingResult>>(
            name = "get-dashboard-details",
            fixingParser = StructureFixingParser(
                fixingModel = GoogleModels.Gemini2_0Flash,
                retries = 3
            )
        )
        
        // Subgraph flow
        edge(nodeStart forwardTo sendInstructions)

        edge(sendInstructions forwardTo getDashboardDetails transformed {
            it.toString()
        })

        edge(getDashboardDetails forwardTo processDashboardResult)
        edge(processDashboardResult forwardTo nodeFinish)

    }
}

[rcd27]

This means the return type should be @Serializable. But Flow<T> is not a data, it is a stream of data. You can write a snapshot of current state of a `Flow`:

@Serializable
data class StateSnapshot<T>(val value: T)

fun <T> StateFlow<T>.toSerializable() = StateSnapshot(value)

But I thinkg this is not what you want...

[rcd27]

The main point is that you provide your tool to LLM and it needs concrete result, which can be passed to it, that's why @Serializable comes into place. And Flow<T> is async stuff. Not sure if this doable without rsocket which takes Flow as an argument for network call 🙂

Try to take last element from the Flow, if you are working with current state of the widgets.

[Andrey Bragin]

Hi, tools can’t return streams of data, because in all existing LLM APIs tool call follows a single request - response model. You can do parallel tool calls if you want to speed up the processing for a certain task, it is supported out of the box in Koog. Then you would receive multiple Message.Tool.Call messages and can process them in parallel and submit multiple results. The nodes we provide out of the box also support such a scenario.

• nodeLLMRequestMultiple to generate multiple tool calls
• nodeExecuteMultipleToolsAndSendResults to execute tools if a list of tool calls is returned and then to submit the results

[tung]

Hi @rcd27. Thanks for response, For now I understand that subgraphWithTask cannot take a Flow as an argument.. I think we can’t collect the flow in this scenario. subgraphWithTask just takes input and output—where can I collect the last data before sending it to the LLM. I’m really confused about that.

[Andrey Bragin]

Just to clarify - none of the LLM-related methods we provide accept flows, because LLMs can produce them, but not accept them

[tung]

Hi @Andrey Bragin Thanks for your clarification. That means in my case I need to modify the get_widgets tool to return Serializable object, or create a new tool that will collect the latest values and serialize the result, right?

[tung]

I’ve created another tool. Inside this tool, I collect flow data and it works.

[Andrey Bragin]

Yes, that should be the correct approach

[tung]

Thanks.