Skip to content

Commit 5c4c4d7

Browse files
google-genai-botcopybara-github
authored andcommitted
fix: recover compacted function calls and pair reused call IDs in LLM content assembly
PiperOrigin-RevId: 995900828
1 parent 461f7d6 commit 5c4c4d7

2 files changed

Lines changed: 582 additions & 34 deletions

File tree

‎core/src/main/java/com/google/adk/flows/llmflows/Contents.java‎

Lines changed: 146 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,8 @@
2929
import com.google.adk.models.LlmRequest;
3030
import com.google.common.base.Strings;
3131
import com.google.common.collect.ImmutableList;
32+
import com.google.common.collect.ImmutableMap;
33+
import com.google.common.collect.ImmutableSet;
3234
import com.google.common.collect.Iterables;
3335
import com.google.common.collect.Lists;
3436
import com.google.genai.types.Content;
@@ -41,6 +43,7 @@
4143
import java.util.Collections;
4244
import java.util.HashMap;
4345
import java.util.HashSet;
46+
import java.util.IdentityHashMap;
4447
import java.util.List;
4548
import java.util.Map;
4649
import java.util.Optional;
@@ -176,7 +179,8 @@ private ImmutableList<Content> getContents(
176179
}
177180

178181
if (hasCompactEvent) {
179-
filteredEvents = processCompactionEvent(filteredEvents, agentName);
182+
List<Event> compactedEvents = processCompactionEvent(filteredEvents, agentName);
183+
filteredEvents = recoverCompactedFunctionCalls(compactedEvents, filteredEvents);
180184
}
181185

182186
// Merge after compaction, which may end mid-run, and before other agents are narrated.
@@ -529,6 +533,99 @@ private static Event createCompactionEvent(Event event, String agentName) {
529533
.build();
530534
}
531535

536+
/**
537+
* Re-injects function-call events from {@code sourceEvents} when compaction summarized away a
538+
* call whose response arrived later. Restores the full call event to preserve first-part thought
539+
* signatures on parallel calls, along with the latest compacted response for each sibling call so
540+
* siblings do not appear pending. Resolves call ownership in {@code sourceEvents} order by event
541+
* identity so reused call IDs bind to the latest preceding call.
542+
*/
543+
private static List<Event> recoverCompactedFunctionCalls(
544+
List<Event> events, List<Event> sourceEvents) {
545+
if (events.stream().allMatch(event -> event.functionResponses().isEmpty())) {
546+
return events;
547+
}
548+
549+
Set<Event> presentEvents = Collections.newSetFromMap(new IdentityHashMap<>());
550+
presentEvents.addAll(events);
551+
552+
Map<String, Event> latestCallById = new HashMap<>();
553+
IdentityHashMap<Event, Map<String, Event>> callByResponse = new IdentityHashMap<>();
554+
IdentityHashMap<Event, Map<String, Event>> responseByCall = new IdentityHashMap<>();
555+
IdentityHashMap<Event, Set<String>> survivingResponses = new IdentityHashMap<>();
556+
557+
for (Event event : sourceEvents) {
558+
for (FunctionCall functionCall : event.functionCalls()) {
559+
functionCall
560+
.id()
561+
.filter(id -> !id.isEmpty())
562+
.ifPresent(id -> latestCallById.put(id, event));
563+
}
564+
for (FunctionResponse functionResponse : event.functionResponses()) {
565+
String responseId = functionResponse.id().orElse("");
566+
if (responseId.isEmpty()) {
567+
continue;
568+
}
569+
Event callEvent = latestCallById.get(responseId);
570+
if (callEvent == null) {
571+
continue;
572+
}
573+
callByResponse.computeIfAbsent(event, k -> new HashMap<>()).put(responseId, callEvent);
574+
if (presentEvents.contains(event)) {
575+
survivingResponses.computeIfAbsent(callEvent, k -> new HashSet<>()).add(responseId);
576+
}
577+
// Keep the latest update for this call; equal timestamps retain source order.
578+
Map<String, Event> responsesForCall =
579+
responseByCall.computeIfAbsent(callEvent, k -> new HashMap<>());
580+
Event existing = responsesForCall.get(responseId);
581+
if (existing == null || event.timestamp() >= existing.timestamp()) {
582+
responsesForCall.put(responseId, event);
583+
}
584+
}
585+
}
586+
587+
List<Event> result = new ArrayList<>();
588+
Set<Event> reinjectedCalls = Collections.newSetFromMap(new IdentityHashMap<>());
589+
for (Event event : events) {
590+
Map<String, Event> callsForEvent = callByResponse.get(event);
591+
if (callsForEvent != null) {
592+
for (FunctionResponse functionResponse : event.functionResponses()) {
593+
String responseId = functionResponse.id().orElse("");
594+
Event callEvent = callsForEvent.get(responseId);
595+
if (callEvent == null
596+
|| presentEvents.contains(callEvent)
597+
|| !reinjectedCalls.add(callEvent)) {
598+
continue;
599+
}
600+
result.add(callEvent);
601+
Set<String> survivingForCall =
602+
survivingResponses.getOrDefault(callEvent, ImmutableSet.of());
603+
Map<String, Event> responsesForCall =
604+
responseByCall.getOrDefault(callEvent, ImmutableMap.of());
605+
Set<Event> siblingResponsesToAdd = Collections.newSetFromMap(new IdentityHashMap<>());
606+
for (FunctionCall siblingCall : callEvent.functionCalls()) {
607+
String siblingId = siblingCall.id().orElse("");
608+
if (!siblingId.isEmpty() && !survivingForCall.contains(siblingId)) {
609+
Event siblingResponse = responsesForCall.get(siblingId);
610+
if (siblingResponse != null) {
611+
siblingResponsesToAdd.add(siblingResponse);
612+
}
613+
}
614+
}
615+
if (!siblingResponsesToAdd.isEmpty()) {
616+
for (Event sourceEvent : sourceEvents) {
617+
if (siblingResponsesToAdd.contains(sourceEvent)) {
618+
result.add(sourceEvent);
619+
}
620+
}
621+
}
622+
}
623+
}
624+
result.add(event);
625+
}
626+
return reinjectedCalls.isEmpty() ? events : result;
627+
}
628+
532629
/** Whether the event is a reply from another agent. */
533630
private static boolean isOtherAgentReply(String agentName, Event event) {
534631
return !agentName.isEmpty()
@@ -770,27 +867,50 @@ private static List<Event> rearrangeEventsForLatestFunctionResponse(List<Event>
770867

771868
private static List<Event> rearrangeEventsForAsyncFunctionResponsesInHistory(
772869
List<Event> events, boolean groupFunctionResponses) {
773-
Map<String, Integer> functionCallIdToResponseEventIndex = new HashMap<>();
870+
// Pair each response with the latest preceding call of that ID, keeping the last update.
871+
Map<String, List<Integer>> callEventIndicesById = new HashMap<>();
774872
for (int i = 0; i < events.size(); i++) {
775-
final int index = i;
776-
Event event = events.get(index);
777-
event
778-
.content()
779-
.flatMap(Content::parts)
780-
.ifPresent(
781-
parts -> {
782-
for (Part part : parts) {
783-
part.functionResponse()
784-
.ifPresent(
785-
response ->
786-
response
787-
.id()
788-
.ifPresent(
789-
functionCallId ->
790-
functionCallIdToResponseEventIndex.put(
791-
functionCallId, index)));
792-
}
793-
});
873+
Event event = events.get(i);
874+
if (!event.functionResponses().isEmpty()) {
875+
continue;
876+
}
877+
for (FunctionCall call : event.functionCalls()) {
878+
final int callIndex = i;
879+
call.id()
880+
.ifPresent(
881+
id ->
882+
callEventIndicesById
883+
.computeIfAbsent(id, k -> new ArrayList<>())
884+
.add(callIndex));
885+
}
886+
}
887+
888+
Map<Integer, Map<String, Integer>> responseEventIndexByCall = new HashMap<>();
889+
for (int i = 0; i < events.size(); i++) {
890+
final int responseIndex = i;
891+
Event event = events.get(responseIndex);
892+
for (FunctionResponse response : event.functionResponses()) {
893+
response
894+
.id()
895+
.ifPresent(
896+
functionCallId -> {
897+
List<Integer> callIndices = callEventIndicesById.get(functionCallId);
898+
if (callIndices == null || callIndices.isEmpty()) {
899+
return;
900+
}
901+
int owningCallIndex = callIndices.get(0);
902+
for (int callIdx : callIndices) {
903+
if (callIdx < responseIndex) {
904+
owningCallIndex = callIdx;
905+
} else {
906+
break;
907+
}
908+
}
909+
responseEventIndexByCall
910+
.computeIfAbsent(owningCallIndex, k -> new HashMap<>())
911+
.put(functionCallId, responseIndex);
912+
});
913+
}
794914
}
795915

796916
List<Event> resultEvents = new ArrayList<>();
@@ -818,6 +938,8 @@ private static List<Event> rearrangeEventsForAsyncFunctionResponsesInHistory(
818938
.orElse(false);
819939

820940
if (hasFunctionCalls) {
941+
Map<String, Integer> responsesForCall =
942+
responseEventIndexByCall.getOrDefault(i, ImmutableMap.of());
821943
Set<Integer> responseEventIndices = new HashSet<>();
822944
// Iterate through parts again to get function call IDs
823945
partsOptional
@@ -830,11 +952,9 @@ private static List<Event> rearrangeEventsForAsyncFunctionResponsesInHistory(
830952
call.id()
831953
.ifPresent(
832954
functionCallId -> {
833-
if (functionCallIdToResponseEventIndex.containsKey(
834-
functionCallId)) {
835-
responseEventIndices.add(
836-
functionCallIdToResponseEventIndex.get(
837-
functionCallId));
955+
Integer respIdx = responsesForCall.get(functionCallId);
956+
if (respIdx != null) {
957+
responseEventIndices.add(respIdx);
838958
}
839959
})));
840960

0 commit comments

Comments
 (0)