Parallel Agents in ADK for Java, in depth: coordinating results after fan-out with structured outputs, a deterministic merger, conflict resolution and provenance

Why an LLM synthesizer is not enough

The simplest gather step is a synthesizer: an LlmAgent whose instruction templates every branch's output key and asks the model to merge them. It is quick to build and fine for prose. It is a poor fit when the branches return facts that downstream code relies on, for three reasons.

• It is not reproducible. Two runs over the same branch outputs can resolve a disagreement differently, and nothing records why.
• It hides conflicts. When the catalog says a product costs 499 and a pricing service says 479, a model will usually pick one silently. The disagreement is often the most useful signal you have.
• It scales badly. Every branch's full output lands in one prompt. Ten branches with verbose answers can cost more tokens in the merge than in all the branches combined.

The fix is to split the gather step in two: a deterministic merger that reconciles structured data in Java, and a writer that turns the reconciled result into language. The model still writes; it no longer adjudicates.

The shape of the pipeline

The pipeline is a SequentialAgent with three steps. The first is the ParallelAgent, whose branches each write one state key. The second is a custom agent with no model call that reads those keys, merges them and writes the merged result plus a list of conflicts back to state. The third is a writer LlmAgent whose instruction templates only the merged keys.

Because the merger reads state by key after the parallel step has finished, the nondeterministic interleaving of branch events stops mattering. That is the first rule of coordination: never infer anything from the order in which branch events arrive.

Give every branch a structured output

The LlmAgent.Builder in ADK Java has outputSchema(Schema), which takes a com.google.genai.types.Schema and constrains the model's final response to JSON matching it, and outputKey(String), which stores the agent's final output in session state under that key. Use a common envelope for every branch: a list of claims, each naming a field, a value, a confidence and the evidence it came from.

import com.google.adk.agents.LlmAgent;
import com.google.genai.types.Schema;
import java.util.List;
import java.util.Map;

static final Schema CLAIM = Schema.builder().type("OBJECT")
    .properties(Map.of(
        "field", Schema.builder().type("STRING")
            .description("One of the field names listed in the instruction").build(),
        "value", Schema.builder().type("STRING").build(),
        "confidence", Schema.builder().type("NUMBER").description("0.0 to 1.0").build(),
        "evidence", Schema.builder().type("STRING").description("URL or document id").build()))
    .required(List.of("field", "value", "confidence", "evidence"))
    .build();

static final Schema RESULT = Schema.builder().type("OBJECT")
    .properties(Map.of("claims", Schema.builder().type("ARRAY").items(CLAIM).build()))
    .required(List.of("claims"))
    .build();

static LlmAgent branch(String key, String focus, String fields) {
  return LlmAgent.builder()
      .name(key + "_researcher")
      .model(MODEL)
      .instruction("""
          You research %s for the product: {product}
          Report only these fields: %s
          Give one claim per field you can support. Omit fields you cannot support.
          Never guess a value without evidence.""".formatted(focus, fields))
      .outputSchema(RESULT)
      .outputKey("result_" + key)
      .build();
}

Two design rules make the merge tractable. First, restrict field names to a closed list in the instruction, so price from one branch is not list_price from another; normalise again in the merger anyway. Second, let a branch omit a field rather than invent one. An absent claim is information; a guessed one is noise with a confidence score attached.

Some ADK versions restrict combining an output schema with tool use on the same agent. Check the documentation for the version you run. If a branch needs tools, make it a two-step SequentialAgent: a tool-using agent that gathers material into a scratch key, followed by a formatting agent with the schema.

A deterministic merger as a custom agent

The merger extends BaseAgent, reads each branch key by name, and emits one event whose stateDelta carries the result. Keep the merge logic in a static, pure method so that it can be unit-tested without a runner or a model. The value stored under an output key may arrive as a JSON string or as an already-parsed map depending on the ADK version and agent configuration, so the reader accepts both.

public final class ResultMerger extends BaseAgent {
  // Sorted keys: Map.of iteration order differs between JVM runs, so without this
  // the same input could serialise to different strings.
  private static final ObjectMapper JSON = new ObjectMapper()
      .configure(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS, true);
  private final List<String> branches;                 // e.g. catalog, pricing, reviews
  private final Map<String, List<String>> authority;   // field -> branches, most trusted first

  public ResultMerger(List<String> branches, Map<String, List<String>> authority) {
    super("result_merger", "Merges structured branch results deterministically.",
          List.of(), List.of(), List.of());
    this.branches = List.copyOf(branches);
    this.authority = Map.copyOf(authority);
  }

  @Override
  protected Flowable<Event> runAsyncImpl(InvocationContext ctx) {
    return Flowable.fromCallable(() -> {
      Map<String, Object> delta = merge(ctx.session().state(), branches, authority);
      return Event.builder()
          .id(Event.generateEventId())
          .invocationId(ctx.invocationId())
          .author(name())
          .branch(ctx.branch().orElse(null))
          .actions(EventActions.builder().stateDelta(new ConcurrentHashMap<>(delta)).build())
          .timestamp(System.currentTimeMillis())
          .build();
    });
  }

  @Override
  protected Flowable<Event> runLiveImpl(InvocationContext ctx) {
    return Flowable.error(new UnsupportedOperationException("live mode not supported"));
  }

  record Claim(String branch, String value, double confidence, String evidence) {}

  static Map<String, Object> merge(Map<String, Object> state, List<String> branches,
                                   Map<String, List<String>> authority) throws Exception {
    Map<String, List<Claim>> byField = new TreeMap<>();          // sorted: stable output
    List<String> missing = new ArrayList<>();
    for (String b : branches) {
      Object raw = state.get("result_" + b);
      if (raw == null) { missing.add(b); continue; }
      JsonNode root = raw instanceof String s ? JSON.readTree(s) : JSON.valueToTree(raw);
      for (JsonNode c : root.path("claims")) {
        String field = c.path("field").asText().trim().toLowerCase(Locale.ROOT);
        byField.computeIfAbsent(field, k -> new ArrayList<>()).add(new Claim(
            b, c.path("value").asText().trim(), c.path("confidence").asDouble(0.0),
            c.path("evidence").asText("")));
      }
    }
    Map<String, Object> merged = new TreeMap<>();
    List<Map<String, Object>> conflicts = new ArrayList<>();
    for (var e : byField.entrySet()) {
      List<String> order = authority.getOrDefault(e.getKey(), List.of());
      Claim win = e.getValue().stream().min(Comparator
          .comparingInt((Claim c) -> order.contains(c.branch()) ? order.indexOf(c.branch()) : 99)
          .thenComparing(Comparator.comparingDouble(Claim::confidence).reversed())
          .thenComparing(Claim::branch)).orElseThrow();
      merged.put(e.getKey(), Map.of("value", win.value(), "source", win.branch(),
                                    "evidence", win.evidence()));
      if (e.getValue().stream().map(Claim::value).distinct().count() > 1) {
        conflicts.add(Map.of("field", e.getKey(), "chosen", win.branch(),
            "values", e.getValue().stream().map(c -> c.branch() + "=" + c.value()).sorted().toList()));
      }
    }
    return Map.of(
        "merged", JSON.writeValueAsString(merged),
        "merge_conflicts", JSON.writeValueAsString(conflicts),
        "merge_missing", missing.isEmpty() ? "none" : String.join(",", missing));
  }
}

Three details carry the design. The comparator ends with the branch name, so ties break the same way on every run, and the mapper sorts map keys, so the serialised JSON is byte-identical across runs and JVMs. The merger always writes all three keys, even when they are empty, because a writer instruction that templates a missing key fails or renders nothing depending on the placeholder syntax; passing context between sequential agents covers those rules. And the merged values are serialised to JSON strings, which template cleanly into a prompt.

Choosing a conflict policy

The comparator above encodes one policy: per-field source authority, then confidence, then name. That is the right default when each branch has a domain it owns. Other policies suit other data.

Mix them per field. A product brief might take price by authority, features by union, and the safety rating by escalation whenever two branches disagree. Writing the policy as data, a map from field to rule, keeps it reviewable and testable.

Worked example: a product brief

Three branches research a laptop. The catalog branch claims price=499 from the product page and ram=16GB at confidence 0.8. The pricing branch claims price=479 from a live price feed with confidence 0.9. The reviews branch claims battery=9h and ram=16 GB, also at 0.8. The authority map says price is owned by pricing, then catalog.

The merger groups claims by field. For price, pricing wins on authority, and a conflict is recorded with both values. For ram there is no authority entry and the confidences tie, so the branch-name tie-break picks catalog; and because the two values differ only in spacing, the merger records a conflict. That is a false conflict caused by a missing normalisation step, exactly the kind of bug the conflicts list exists to surface. Battery has a single claim. The state after the merger, with the code as shown, looks like this:

merged          = {"battery":{"evidence":"rev-17","source":"reviews","value":"9h"},
                   "price":{"evidence":"feed:sku-88","source":"pricing","value":"479"},
                   "ram":{"evidence":"pdp-88","source":"catalog","value":"16GB"}}
merge_conflicts = [{"chosen":"pricing","field":"price","values":["catalog=499","pricing=479"]},
                   {"chosen":"catalog","field":"ram","values":["catalog=16GB","reviews=16 GB"]}]
merge_missing   = none

The writer's instruction templates {merged}, {merge_conflicts} and {merge_missing} and tells the model to state each value with its source, to mention any recorded conflict in one sentence, and to say plainly which sources did not respond. It never sees the raw branch output. The next step is to normalise units in the merger, for example by stripping spaces from size values before grouping, so the ram entry stops appearing as a conflict.

Provenance and the token budget

Every merged value keeps its source branch and evidence string, which lets the writer cite sources and lets you audit a wrong answer back to the branch that produced it. Store full branch outputs in state or in your logs for debugging, but keep them out of the writer's prompt.

Bound the merged payload explicitly. Cap list fields, for example the top five findings by authority then confidence, and drop evidence strings longer than a fixed length. A useful rule of thumb is that the writer's input should grow with the number of fields, not with the number of branches. When the fan-out is wide, merge hierarchically: group branches into sets of a few, merge each set, then merge the merged results with the same code, as described in advanced workflow patterns.

Testing the gather step

Because merge is a pure static method, its tests need no model, runner or network. Feed it hand-built state maps and assert on the output.

@Test
void pricingBranchOwnsPriceAndConflictIsRecorded() throws Exception {
  Map<String, Object> state = Map.of(
      "result_catalog", "{\"claims\":[{\"field\":\"price\",\"value\":\"499\",\"confidence\":0.95,\"evidence\":\"pdp\"}]}",
      "result_pricing", Map.of("claims", List.of(Map.of(
          "field", "Price", "value", "479", "confidence", 0.9, "evidence", "feed"))));
  Map<String, Object> out = ResultMerger.merge(state,
      List.of("catalog", "pricing", "reviews"), Map.of("price", List.of("pricing", "catalog")));
  assertTrue(((String) out.get("merged")).contains("\"value\":\"479\""));
  assertTrue(((String) out.get("merge_conflicts")).contains("catalog=499"));
  assertEquals("reviews", out.get("merge_missing"));
}

This one test exercises both input forms, field-name normalisation, authority over confidence, conflict recording and missing-branch reporting. Add a determinism test that shuffles the branch list and asserts identical output, and an end-to-end test with stubbed models that checks the writer only receives the merged keys.

Failure modes

• Shared output keys. Two branches writing the same key overwrite each other, and which write survives depends on timing. Give every branch its own key and assert uniqueness when building the agent.
• Order-dependent merging. Code that takes the first event or the first branch to finish produces different results on different runs. Read state by key after the parallel step.
• Schema drift. A prompt change renames a field, and the merger silently drops it as an unknown. Log unknown fields and fail tests on them.
• Uncalibrated confidence. A branch that always reports 0.99 wins every confidence tie. Prefer authority, or calibrate per branch from labelled runs.
• Writer re-litigating. If the writer sees raw branch outputs it will re-resolve conflicts its own way. Template only merged keys.
• Unbounded payloads. A wide fan-out with long evidence strings blows the writer's context. Cap sizes in the merger.

What to do next

1. List the fields your fan-out produces and assign each one an owning branch or a policy.
2. Add an output schema and a unique output key to every branch, using one shared claim envelope.
3. Implement the merger as a custom agent around a pure static merge method, and write the authority map as data.
4. Unit-test the merge with string and map inputs, a conflict, a missing branch and a shuffled branch list.
5. Change the writer so it templates only the merged, conflict and missing keys, and instruct it to cite sources.
6. Log the conflicts list in production and review it weekly; recurring conflicts point to normalisation bugs or a stale source.