Skip to content

Commit 75b6298

Browse files
Mateusz Krawieccopybara-github
authored andcommitted
revert: keep one generated id per streamed function call
PiperOrigin-RevId: 990259405
1 parent 4322839 commit 75b6298

2 files changed

Lines changed: 25 additions & 125 deletions

File tree

‎core/src/commonMain/kotlin/com/google/adk/kt/models/StreamingResponseAggregator.kt‎

Lines changed: 25 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -83,22 +83,23 @@ class StreamingResponseAggregator {
8383
response.finishReason?.let { finishReason = it }
8484
response.errorMessage?.let { errorMessage = it }
8585

86-
val parts =
87-
response.content?.parts.orEmpty().map { part ->
88-
when {
89-
!part.text.isNullOrEmpty() -> part.also { processTextPart(it) }
90-
part.functionCall != null -> processFunctionCallPart(part)
91-
// An empty text part that carries something else falls through to survive on its own;
92-
// one carrying nothing at all just ends a Gemini 3 stream and is dropped.
93-
part.isStreamTerminator() -> part
94-
else -> {
95-
// Other non-text parts (blobs, etc.)
96-
flushTextBufferToSequence()
97-
partsSequence.add(part)
98-
part
99-
}
86+
// Assign a client id to any function call missing one up front, so the partial chunk and the
87+
// final response share it.
88+
val parts = (response.content?.parts ?: emptyList()).map { it.ensureFunctionCallId() }
89+
for (part in parts) {
90+
when {
91+
!part.text.isNullOrEmpty() -> processTextPart(part)
92+
part.functionCall != null -> processFunctionCallPart(part)
93+
// An empty text part that carries something else falls through to survive on its own; one
94+
// carrying nothing at all just ends a Gemini 3 stream and is dropped.
95+
part.isStreamTerminator() -> {}
96+
else -> {
97+
// Other non-text parts (blobs, etc.)
98+
flushTextBufferToSequence()
99+
partsSequence.add(part)
100100
}
101101
}
102+
}
102103

103104
// In Progressive SSE mode, all intermediate chunks are partial (with any generated ids).
104105
response.copy(content = response.content?.copy(parts = parts), partial = true)
@@ -173,15 +174,14 @@ class StreamingResponseAggregator {
173174
currentTextBuffer.append(part.text)
174175
}
175176

176-
/** Returns this call, or a copy with a generated client id if it has none. */
177-
private fun FunctionCall.ensureId(): FunctionCall =
178-
if (id.isNullOrEmpty()) copy(id = FunctionCall.generateId()) else this
177+
/** Returns a copy with a generated client id if this is a function call missing one. */
178+
private fun Part.ensureFunctionCallId(): Part {
179+
val fc = functionCall ?: return this
180+
if (!fc.id.isNullOrEmpty()) return this
181+
return copy(functionCall = fc.copy(id = FunctionCall.generateId()))
182+
}
179183

180-
/**
181-
* Aggregates a function call part and returns it with any generated id. A streamed call gets its
182-
* id on its first chunk, so that chunk and the final response share it.
183-
*/
184-
private fun processFunctionCallPart(part: Part): Part {
184+
private fun processFunctionCallPart(part: Part) {
185185
val fc =
186186
part.functionCall
187187
?: throw IllegalStateException(
@@ -197,30 +197,25 @@ class StreamingResponseAggregator {
197197
fc.willContinue == true ||
198198
(currentFcName != null && !hasName)
199199
if (streamedPart) {
200-
val call = if (currentFcId == null) fc.ensureId() else fc
201200
if (part.thoughtSignature?.isNotEmpty() == true && currentThoughtSignature == null) {
202201
currentThoughtSignature = part.thoughtSignature
203202
}
204-
processStreamingFunctionCall(call)
205-
return part.copy(functionCall = call)
203+
processStreamingFunctionCall(fc)
206204
} else if (hasName) {
207-
val withId = part.copy(functionCall = fc.ensureId())
208205
// Non-streaming call. Safety guard: the model should terminate a streamed call with
209206
// willContinue=false before starting a new one; flush any still-in-progress call so it is
210207
// neither dropped nor merged.
211208
flushTextBufferToSequence()
212209
flushFunctionCallToSequence()
213-
partsSequence.add(withId)
214-
return withId
210+
partsSequence.add(part)
215211
}
216-
return part
217212
}
218213

219214
private fun processStreamingFunctionCall(fc: FunctionCall) {
220215
if (fc.name.isNotEmpty()) {
221216
currentFcName = fc.name
222217
}
223-
if (!fc.id.isNullOrEmpty()) {
218+
if (fc.id != null) {
224219
currentFcId = fc.id
225220
}
226221

‎core/src/commonTest/kotlin/com/google/adk/kt/models/StreamingResponseAggregatorTest.kt‎

Lines changed: 0 additions & 95 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,6 @@ import com.google.adk.kt.types.ToolType
3131
import com.google.adk.kt.types.UsageMetadata
3232
import kotlin.test.Test
3333
import kotlin.test.assertEquals
34-
import kotlin.test.assertNotEquals
3534
import kotlin.test.assertNotNull
3635
import kotlin.test.assertTrue
3736
import kotlinx.coroutines.Dispatchers
@@ -108,100 +107,6 @@ class StreamingResponseAggregatorTest {
108107
assertEquals("San Francisco", fc.args["location"])
109108
}
110109

111-
@Test
112-
fun testStreamingFunctionCall_partialAndFinalResponseShareGeneratedId() = runBlocking {
113-
val aggregator = StreamingResponseAggregator()
114-
115-
val partial1 =
116-
aggregator.processResponse(
117-
createFcResp(createPartialFc("my_tool", "$.x", "hello", willContinue = true))
118-
)
119-
val unused =
120-
aggregator.processResponse(
121-
createFcResp(createPartialFc(null, "$.x", " world", willContinue = false))
122-
)
123-
val finalResp = aggregator.aggregate()
124-
125-
val finalFc = finalResp?.content?.parts?.single()?.functionCall
126-
assertNotNull(finalFc)
127-
assertEquals(mapOf("x" to "hello world"), finalFc.args)
128-
val finalId = assertNotNull(finalFc.id)
129-
assertTrue(finalId.startsWith(FunctionCall.ADK_FUNCTION_CALL_ID_PREFIX))
130-
assertEquals(finalId, partial1.content?.parts?.single()?.functionCall?.id)
131-
}
132-
133-
@Test
134-
fun testStreamedCall_continuationChunkWithEmptyId_keepsFirstChunkId() = runBlocking {
135-
val aggregator = StreamingResponseAggregator()
136-
137-
val partial1 =
138-
aggregator.processResponse(
139-
createFcResp(createPartialFc("my_tool", "$.x", "hello", willContinue = true))
140-
)
141-
val unused =
142-
aggregator.processResponse(
143-
createFcResp(createPartialFc(null, "$.x", " world", willContinue = false).copy(id = ""))
144-
)
145-
val finalResp = aggregator.aggregate()
146-
147-
val firstId = assertNotNull(partial1.content?.parts?.single()?.functionCall?.id)
148-
assertEquals(firstId, finalResp?.content?.parts?.single()?.functionCall?.id)
149-
}
150-
151-
@Test
152-
fun testStreamingFunctionCalls_inSeparateChunks_getDifferentIds() = runBlocking {
153-
val aggregator = StreamingResponseAggregator()
154-
155-
val unused1 =
156-
aggregator.processResponse(
157-
createFcResp(createPartialFc("tool_a", "$.a", "val_a", willContinue = false))
158-
)
159-
val unused2 =
160-
aggregator.processResponse(
161-
createFcResp(createPartialFc("tool_b", "$.b", "val_b", willContinue = false))
162-
)
163-
val finalResp = aggregator.aggregate()
164-
165-
val ids = finalResp?.content?.parts?.map { it.functionCall?.id }
166-
assertNotNull(ids)
167-
assertEquals(2, ids.size)
168-
assertTrue(ids.all { it != null && it.startsWith(FunctionCall.ADK_FUNCTION_CALL_ID_PREFIX) })
169-
assertNotEquals(ids[0], ids[1])
170-
}
171-
172-
@Test
173-
fun testStreamedCallEndAndNewSameNameCall_inOneChunk_getDifferentIds() = runBlocking {
174-
val aggregator = StreamingResponseAggregator()
175-
176-
val partial1 =
177-
aggregator.processResponse(
178-
createFcResp(createPartialFc("get_weather", "$.city", "SF", willContinue = true))
179-
)
180-
val unused =
181-
aggregator.processResponse(
182-
LlmResponse(
183-
content =
184-
Content(
185-
parts =
186-
listOf(
187-
Part(functionCall = FunctionCall(willContinue = false)),
188-
Part(
189-
functionCall = FunctionCall(name = "get_weather", args = mapOf("city" to "NY"))
190-
),
191-
)
192-
)
193-
)
194-
)
195-
val finalResp = aggregator.aggregate()
196-
197-
val calls = finalResp?.content?.parts?.mapNotNull { it.functionCall }
198-
assertNotNull(calls)
199-
assertEquals(listOf("SF", "NY"), calls.map { it.args["city"] })
200-
assertEquals(partial1.content?.parts?.single()?.functionCall?.id, calls[0].id)
201-
assertNotNull(calls[1].id)
202-
assertNotEquals(calls[0].id, calls[1].id)
203-
}
204-
205110
@Test
206111
fun testNestedPartialFunctionCallAggregation() = runBlocking {
207112
val aggregator = StreamingResponseAggregator()

0 commit comments

Comments
 (0)