DEV Community: mgd43b

Typed Tool Inputs in Java Agent Systems: Records as Contracts

mgd43b — Sat, 18 Apr 2026 14:00:00 +0000

There is a design tension at the boundary between an LLM and a tool. The LLM generates a structured JSON call. Your code receives it. Between those two points, something has to parse the JSON, validate the fields, and turn it into something the tool can actually work with.

Most agent frameworks handle this with a single opaque string. The LLM gets a parameter named input. The tool receives a raw string. The tool author writes their own JSON parsing. Required-field validation is up to them. If the LLM passes malformed JSON or omits a required field, the tool finds out at runtime.

The question I kept coming back to was: why doesn't the framework own this? Java has records. Records have components with names and types. Those components can carry annotations. Everything needed to generate a typed JSON Schema, deserialize the call, and validate required fields is already there, waiting to be read.

The Problem With Single-String Tool Inputs

The original model is simple and universal. A tool implements one method: execute(String input). The LLM is told to format the input as JSON if the tool needs structured data. The tool parses it.

This works, but it puts a burden in the wrong place. Consider a web search tool:

public class WebSearchTool extends AbstractAgentTool {
    @Override
    public String name() { return "web_search"; }

    @Override
    public String description() {
        return "Performs a web search. Input: JSON with 'query' field.";
    }

    @Override
    protected ToolResult doExecute(String input) {
        JsonNode node = mapper.readTree(input);
        String query = node.get("query").asText();  // NPE if missing
        // ...
    }
}

A few problems here. The description must instruct the LLM about the expected JSON format, mixing tool contract with tool documentation. The tool code handles parsing. There is no schema — the LLM has no structured signal about what fields are expected, what their types are, or which are required. A missing query field produces a null pointer exception rather than a clean validation failure.

The tool input boundary is where type safety matters most. It is the handoff between model output and application code. A malformed call here does not just fail — it can silently corrupt state or produce errors that are hard to trace back to the model call.

Records as Input Contracts

The alternative is to declare the input contract as a Java record. The framework can derive everything it needs from the record's components and their annotations: the JSON Schema for the LLM, the deserialization logic, the required-field validation.

@ToolInput(description = "Parameters for a web search")
public record WebSearchInput(
    @ToolParam(description = "Search query string, e.g. 'Java 21 virtual threads'")
    String query
) {}

The tool then extends AbstractTypedAgentTool<WebSearchInput> instead of AbstractAgentTool:

public final class WebSearchTool extends AbstractTypedAgentTool<WebSearchInput> {

    @Override
    public Class<WebSearchInput> inputType() {
        return WebSearchInput.class;
    }

    @Override
    public ToolResult execute(WebSearchInput input) {
        String query = input.query().trim();
        if (query.isBlank()) {
            return ToolResult.failure("Search query must not be blank");
        }
        // ... perform search
    }
}

The @ToolParam annotation marks fields as required by default. Optional fields use @ToolParam(required = false). There is no JSON parsing in the tool body. The framework handles it.

What the Framework Does With the Record

At startup, ToolSchemaGenerator introspects the record class and produces a JsonObjectSchema for LangChain4j. The mapping covers the types you would expect:

Java type	JSON Schema type
`String`	`string`
`int`, `long`, `Integer`	`integer`
`double`, `float`, `BigDecimal`	`number`
`boolean`	`boolean`
`enum`	`string` with `enum` constraint
`List<T>`	`array`
`Map<String, V>`	`object`

When the LLM calls the tool, LangChain4jToolAdapter passes the full JSON arguments to AbstractTypedAgentTool, which uses ToolInputDeserializer to deserialize them into a record instance. Required fields that are missing or null produce a clean ToolResult.failure before the tool body executes.

The LLM receives a proper multi-parameter schema instead of a single input string. It can see the field names, their types, and their descriptions directly in the tool specification.

The Adapter Path

LangChain4jToolAdapter dispatches based on whether a tool implements TypedAgentTool:

if (tool instanceof TypedAgentTool<?> typed) {
    parameters = ToolSchemaGenerator.generateSchema(typed.inputType());
} else {
    // original single-parameter schema, fully backward compatible
    parameters = JsonObjectSchema.builder()
        .addStringProperty("input", "The input string for the tool")
        .required(List.of("input"))
        .build();
}

The same instanceof check applies at execution time — typed tools receive the full JSON arguments object; legacy tools receive the extracted "input" field value. No changes required to existing tool implementations.

Before and After: The Schema Difference

For the legacy WebSearchTool, the LLM received this schema:

{
  "properties": {
    "input": {
      "type": "string",
      "description": "The input string for the tool"
    }
  },
  "required": ["input"]
}

With TypedAgentTool, it receives:

{
  "properties": {
    "query": {
      "type": "string",
      "description": "Search query string, e.g. 'Java 21 virtual threads'"
    }
  },
  "required": ["query"]
}

The field is named correctly. The description is the one written for that field. For tools with multiple inputs, each parameter is a separate, named, typed entry in the schema rather than buried inside an opaque string that the tool must document and parse.

Migrating a Multi-Field Tool

The pattern is the same regardless of how many fields the record carries. A hypothetical HTTP request tool:

@ToolInput(description = "Parameters for an HTTP request")
public record HttpRequestInput(
    @ToolParam(description = "The URL to request") String url,
    @ToolParam(description = "HTTP method: GET, POST, PUT, DELETE") String method,
    @ToolParam(description = "Request body (optional)", required = false) String body
) {}

public final class HttpRequestTool extends AbstractTypedAgentTool<HttpRequestInput> {

    @Override
    public Class<HttpRequestInput> inputType() { return HttpRequestInput.class; }

    @Override
    public ToolResult execute(HttpRequestInput input) {
        // url and method are guaranteed non-null; body may be null
        // no parsing, no null checks on deserialization
    }
}

The body field is optional — it will be absent from the required array in the schema, and will be null if the LLM omits it. url and method are required — a call missing either produces a failure before execute is reached.

Tradeoffs

The typed system is opt-in for a reason. There are cases where it does not fit well.

Records only. The framework introspects Java records specifically. If an existing tool has complex input logic tied to its own parsing, migrating to a record may not be a clean fit.

Reflection at startup. Schema generation uses reflection on the record's components. For a large number of tools, this adds initialization cost. In practice it is small, but it is not zero.

Enum constraints. Enum fields generate a string schema with an enum list derived from Enum.name(). This aligns with Jackson's default deserialization behavior. If you have custom serialization names, the generated enum list may not match the LLM's expectations without additional configuration.

Not a substitute for description quality. The schema tells the LLM the structure. The field descriptions still determine whether the LLM understands how to fill that structure correctly. A typed schema with poor descriptions is only marginally better than a well-documented string input.

Five of the built-in tools were migrated to the typed system (FileReadTool, FileWriteTool, JsonParserTool, WebSearchTool, WebScraperTool). Four were left as legacy tools (CalculatorTool, DateTimeTool, HttpAgentTool, ProcessAgentTool) — either because their inputs are naturally single-valued, or because the migration cost was not justified by the gain.

Practical Takeaway

The tool input boundary is a contract. Java records are a natural way to express that contract, and the language gives you enough reflective machinery to derive everything you need from it without additional code generation or build-time processing.

The main gain is not ergonomics, though that improves. The main gain is that the LLM receives an accurate schema for the tool it is calling, which reduces hallucinated input formats and makes validation failures deterministic rather than buried in manual parsing code.

AgentEnsemble is a JVM-native framework for building multi-agent systems in Java. The typed tool input system is part of the core library.

Guide: https://agentensemble.net/guides/tools/
Source: https://github.com/AgentEnsemble/agentensemble
MIT licensed

If you have been working with agent tool systems and have found a different approach to the input boundary problem — typed DSLs, code generation, something else — I would be interested in the tradeoffs you have seen.

Self-Optimizing Agent Tasks: Persistent Reflection Loops in Java

mgd43b — Wed, 15 Apr 2026 14:00:00 +0000

Task definitions are written at compile time. You describe what the task should do, wire up a model, and run. The prompt stays fixed unless you go back and edit it.

In practice, you often discover after a few runs that the instructions could be more precise. The LLM misses an edge case you didn't anticipate. The output format drifts in ways you didn't specify. You revise the description, redeploy, and try again.

The harder version of this problem is: what if the instructions could improve themselves?

Task reflection is a persistent, automated feedback loop built into the task execution lifecycle. After a task completes successfully -- output accepted, guardrails passed, reviews approved -- an LLM-backed analysis step reviews whether the task's instructions could be improved. Improvements are stored in a ReflectionStore and injected into the task's prompt on subsequent runs. The original task definition is never modified.

This post covers how reflection works, what the API looks like, and where the tradeoffs sit.

Reflection vs Phase Review

These two mechanisms are often confused because both involve quality analysis. The distinction is in scope and timing:

	Phase Review	Task Reflection
Trigger	After phase completes	After task output accepted
Scope	Within a single `Ensemble.run()`	Across multiple `Ensemble.run()` calls
Purpose	Fix inadequate output this run	Improve instructions for future runs
Persistence	Transient -- lost after run	Persistent -- stored between runs
Initiated by	External reviewer	Automated LLM analysis

Phase review fixes output within a run. Task reflection improves instructions across runs. They compose: a task can have both.

Quick Start

Enable reflection with .reflect(true) on a task, and configure a ReflectionStore on the ensemble:

ReflectionStore store = new InMemoryReflectionStore();

Task research = Task.builder()
    .description("research the top 5 trends in cloud-native Java for 2026")
    .reflect(true)
    .chatModel(model)
    .build();

// Run 1: no prior reflections; task executes normally
EnsembleOutput run1 = Ensemble.builder()
    .tasks(List.of(research))
    .reflectionStore(store)
    .chatModel(model)
    .build()
    .run();

// Reflection fires after run 1 completes; improvements stored in `store`

// Run 2: prior reflections injected into the prompt automatically
EnsembleOutput run2 = Ensemble.builder()
    .tasks(List.of(research))
    .reflectionStore(store)
    .chatModel(model)
    .build()
    .run();

The store is the key. Pass the same store instance across run() calls and the accumulated reflections persist. The task definition -- the Task object -- is the same every run. The difference is what the reflection store contributes to the prompt.

The Execution Lifecycle

Reflection fires at the end of the task lifecycle, after all other post-processing:

1. Task executes (LLM call)
2. Guardrails evaluate output
3. Review gate runs (if configured)
4. Memory scopes write
5. [Reflection] LLM analyzes output; generates improvement; stores in ReflectionStore

If the task fails, guardrails reject the output, or the review gate retries, reflection does not fire. Reflection only fires on a fully accepted output.

On the next run of the same task:

1. ReflectionStore loads prior reflections for this task identity
2. Reflections injected into prompt
3. Task executes with improved instructions
4. [Reflection] New analysis fires; improvement stored

The task is identified by a TaskIdentity derived from its description. Two tasks with the same description share the same reflection history.

What Gets Injected

The reflection store contributes an additional section to the task's prompt:

[original task description]

## Instruction Refinements

Based on previous runs, the following refinements have been found to improve output quality:
- Be specific about the time range: results should cover events within the last 12 months only.
- Structure the output as a numbered list with a one-sentence summary per trend.
- For each trend, cite at least one concrete project or company as evidence.

The injection is additive. Original instructions are preserved. Reflections narrow, clarify, or extend them based on what previous outputs revealed.

Reflection Configuration

Bounding the History

By default, the framework uses all stored reflections for a task. You can bound the number injected via ReflectionConfig:

Task analysis = Task.builder()
    .description("analyze customer sentiment from support tickets")
    .reflect(true)
    .reflectionConfig(ReflectionConfig.builder()
        .maxReflections(5)
        .build())
    .chatModel(model)
    .build();

With maxReflections(5), only the 5 most recent reflections are injected. Older reflections remain in the store but are not included in the prompt. This prevents prompt bloat as the number of runs grows.

Custom Reflection Strategy

The default strategy uses an LLM call to analyze the task output and generate an improvement. You can substitute a custom ReflectionStrategy:

public class DomainReflectionStrategy implements ReflectionStrategy {

    @Override
    public Optional<String> reflect(ReflectionInput input) {
        String output = input.taskOutput();
        // custom analysis: check for required sections, format, length
        if (!output.contains("## Summary")) {
            return Optional.of("Always include a ## Summary section as the first heading");
        }
        // no improvement identified this time
        return Optional.empty();
    }
}

Task.builder()
    .description("write a technical design document")
    .reflect(true)
    .reflectionConfig(ReflectionConfig.builder()
        .strategy(new DomainReflectionStrategy())
        .build())
    .chatModel(model)
    .build();

A custom strategy can use deterministic rules, call a different model, or apply domain-specific analysis. Returning Optional.empty() skips storage for that run.

The Reflection Store

ReflectionStore is an interface with two methods:

public interface ReflectionStore {
    List<TaskReflection> load(TaskIdentity identity);
    void store(TaskIdentity identity, TaskReflection reflection);
}

InMemoryReflectionStore is included for development and testing. It holds reflections in a ConcurrentHashMap and loses state when the process stops.

For production, implement ReflectionStore against whatever persistence layer makes sense for your system -- a relational database, a document store, or a key-value store:

public class JdbcReflectionStore implements ReflectionStore {

    private final DataSource dataSource;

    @Override
    public List<TaskReflection> load(TaskIdentity identity) {
        // SELECT content FROM task_reflections WHERE task_id = ?
        // ORDER BY created_at DESC LIMIT maxReflections
    }

    @Override
    public void store(TaskIdentity identity, TaskReflection reflection) {
        // INSERT INTO task_reflections (task_id, content, created_at) VALUES (?, ?, ?)
    }
}

The TaskReflection record holds the improvement text and a timestamp. TaskIdentity includes the task description hash used for keying.

Disabling Reflection Selectively

Reflection is opt-in per task. Tasks without .reflect(true) are unaffected even if a ReflectionStore is configured on the ensemble. You can enable reflection for high-value tasks and leave it off for tasks where the instructions are stable or where the cost of an extra LLM call isn't justified.

Ensemble.builder()
    .tasks(List.of(
        stableDataFetchTask,       // no reflection
        evolving AnalysisTask,     // .reflect(true)
        stableFormattingTask       // no reflection
    ))
    .reflectionStore(store)
    .chatModel(model)
    .build()
    .run();

The store is queried and written only for tasks with reflection enabled.

Tradeoffs

Reflection adds an LLM call per reflective task per run. For tasks that run thousands of times, this adds up. The cost is bounded if reflections converge -- if the task's instructions become stable after a few runs, reflections may produce no new improvements and Optional.empty() returns more often.

Reflections can drift. If a task's purpose changes -- the description is updated, the downstream context changes, the data it processes shifts -- earlier reflections may no longer apply. maxReflections helps here by aging out old improvements. For significant task changes, clearing the stored reflections for that task is reasonable.

Reflection is not a substitute for good initial instructions. A task with fundamentally unclear instructions will accumulate reflections that patch around the ambiguity. The better use is to start with reasonable instructions and use reflection to sharpen them in response to real outputs over time.

The original task definition is never modified. All improvements live in the store. This is a deliberate choice: the source of truth for what a task does remains in code, not in a mutable prompt that silently drifts over time.

The guide at agentensemble.net/guides/task-reflection/ covers the full lifecycle, store implementation patterns, and configuration options. The design document at agentensemble.net/design/task-reflection/ covers the architectural decisions behind the feature.

AgentEnsemble is open-source under the MIT license.

Quality Gates on Agent Pipelines: Phase Review and Feedback Injection in Java

mgd43b — Sun, 12 Apr 2026 14:00:00 +0000

Most agent pipelines treat quality as a post-run concern. The pipeline runs, you look at the output, you decide if it's acceptable. If not, you re-run the whole thing or manually patch the result. That approach gets harder to sustain as pipelines grow in complexity and the cost of a bad output increases.

The question worth asking is: where in the pipeline should quality enforcement sit? And when a phase produces inadequate output, how should the feedback get back to the tasks responsible?

PhaseReview answers both questions. It attaches a quality gate to any phase, fires after the phase completes, and based on the reviewer's decision either approves the output, triggers a retry with injected feedback, pushes the work back to a predecessor phase, or rejects the pipeline entirely. The review task is itself a first-class AgentEnsemble task -- AI-backed, deterministic handler, or human-in-the-loop -- and the framework handles feedback injection and retry orchestration automatically.

Attaching a Review to a Phase

The minimal setup attaches a reviewer task to a phase via PhaseReview.of(reviewTask):

Task draftReport = Task.builder()
    .description("draft a summary report of the quarterly results")
    .chatModel(model)
    .build();

Task reviewReport = Task.builder()
    .description("""
        Review the report for completeness and accuracy.
        Output exactly one of:
        APPROVE
        RETRY:<specific feedback for the author>
        REJECT:<reason>
        """)
    .chatModel(model)
    .context(List.of(draftReport))
    .build();

Phase reporting = Phase.builder()
    .name("reporting")
    .tasks(List.of(draftReport))
    .review(PhaseReview.of(reviewReport))
    .build();

Ensemble.builder()
    .phases(List.of(reporting))
    .chatModel(model)
    .build()
    .run();

The review task reads the draft output via .context(List.of(draftReport)). After the phase tasks complete, the framework runs the reviewer, parses its decision, and acts on it. draftReport is retried with feedback. The reviewer never appears in the phase task list -- it runs as a gate after the phase completes.

The Four Decisions

A review task produces one of four outcomes:

Decision	Behavior
`APPROVE`	Phase output is accepted; pipeline continues
`RETRY:<feedback>`	Phase tasks are re-run; feedback injected into prompts
`RETRY_PREDECESSOR:<feedback>`	Predecessor phase re-runs first, then this phase re-runs
`REJECT:<reason>`	Pipeline fails with a `PhaseReviewRejectionException`

The APPROVE/RETRY/REJECT parsing is handled by the framework. RETRY_PREDECESSOR is useful when the inadequate output in the current phase is caused by incomplete or incorrect work in an earlier phase.

Feedback Injection

When a retry is triggered, the framework injects reviewer feedback directly into the prompt of the tasks being retried. The injected section looks like this:

[original task instructions here]

## Revision Instructions (Attempt 2)

The report is missing Q3 comparisons and does not address margin compression. Expand the analysis section with specific numbers.

Previous output:
[output from attempt 1]

The task sees the original instructions, the feedback, and its prior output. No changes to the task definition are required. The injection happens entirely in the prompt construction layer.

For attempt 3 and beyond, the section header updates to Attempt 3, and both the latest feedback and the most recent prior output are included.

Reviewer Types

AI Reviewer

The reviewer task is an AI-backed agent that reads the phase output via context() and generates a decision:

Task aiReviewer = Task.builder()
    .description("""
        You are a quality reviewer for financial reports.
        Read the draft report provided in context.
        Check for: completeness, accurate numbers, professional tone.
        Output exactly: APPROVE, RETRY:<specific instructions>, or REJECT:<reason>.
        """)
    .chatModel(model)
    .context(List.of(draftReport))
    .build();

The reviewer's LLM call is separate from the main phase tasks. You can use a different model for the reviewer if appropriate.

Deterministic Reviewer

For rule-based quality checks, a deterministic handler reads the phase output and returns a decision string:

Task qualityCheck = Task.builder()
    .description("programmatic-quality-check")
    .context(List.of(draftReport))
    .handler(ctx -> {
        String output = ctx.contextOutputs().get(0).getRaw();
        if (output.length() < 500) {
            return ToolResult.success("RETRY: output is too short -- expand each section to at least two paragraphs");
        }
        if (!output.contains("Q3") || !output.contains("Q4")) {
            return ToolResult.success("RETRY: report must include both Q3 and Q4 data");
        }
        return ToolResult.success("APPROVE");
    })
    .build();

Deterministic reviewers are useful when the quality criteria are precise and don't require LLM judgment.

Human Reviewer

For steps that need a human sign-off, the human review API blocks until the reviewer responds:

Task humanGate = Task.builder()
    .description("human-review-gate")
    .context(List.of(draftReport))
    .review(Review.required())
    .build();

The human reviewer sees the task output and enters APPROVE, RETRY with feedback, or REJECT in the console (or a custom reviewer UI). This integrates with the same retry loop.

Controlling Retry Limits

By default, PhaseReview.of(reviewTask) allows up to 2 self-retries and 2 predecessor retries. Both are configurable:

Phase reporting = Phase.builder()
    .name("reporting")
    .tasks(List.of(draftReport))
    .review(PhaseReview.of(reviewReport, 3))
    .build();

Or with the builder for full control:

Phase reporting = Phase.builder()
    .name("reporting")
    .tasks(List.of(draftReport))
    .review(PhaseReview.builder()
        .task(reviewReport)
        .maxRetries(3)
        .maxPredecessorRetries(2)
        .build())
    .build();

When the retry limit is exhausted, the framework treats the phase as failed and throws. The pipeline does not silently accept a low-quality output when retries are exhausted.

Predecessor Retry

When the reviewer determines that the current phase's inadequate output is caused by problems in an earlier phase, it can request a predecessor retry:

Task analysisReviewer = Task.builder()
    .description("""
        Review the analysis. If the data from the ingestion phase appears incomplete or incorrect,
        output: RETRY_PREDECESSOR:<what needs to be fixed in data ingestion>
        If the analysis itself is the problem, output: RETRY:<what needs to be revised>
        If acceptable, output: APPROVE
        """)
    .chatModel(model)
    .context(List.of(analysisTask))
    .build();

Phase ingestion = Phase.builder()
    .name("ingestion")
    .tasks(List.of(ingestTask))
    .build();

Phase analysis = Phase.builder()
    .name("analysis")
    .tasks(List.of(analysisTask))
    .after(ingestion)
    .review(PhaseReview.of(analysisReviewer))
    .build();

If the reviewer outputs RETRY_PREDECESSOR:, the framework re-runs the ingestion phase with the feedback injected into ingestion tasks, then re-runs the analysis phase. The review fires again after the second analysis completes.

The predecessor is the phase declared in .after(). Predecessor retry does not cascade further back automatically.

Accessing Review Results

After the ensemble completes, review results are available in the phase output:

EnsembleOutput result = ensemble.run();

PhaseOutput reportingOut = result.getPhaseOutputs().get("reporting");
ReviewRecord reviewRecord = reportingOut.reviewRecord();

System.out.println("Attempts: " + reviewRecord.attemptCount());
System.out.println("Decision: " + reviewRecord.finalDecision());

This is useful for logging, auditing, or downstream branching based on whether the output was approved on the first attempt or required iterations.

Tradeoffs

Review tasks add LLM calls to the pipeline. For AI reviewers, each retry cycle adds a reviewer call plus the retried task calls. In cost-sensitive pipelines, deterministic reviewers can enforce the most common criteria without adding model calls.

Feedback injection is prompt-based. Tasks see the feedback as text in their prompt, not as a structured signal. The quality of the retry depends on how well the reviewer communicates what needs to change.

Predecessor retry re-runs the entire predecessor phase, not individual tasks. If the predecessor phase is expensive, predecessor retries can be costly. Design predecessor phases with this in mind if predecessor retry is expected.

The review task reads phase outputs via context() declarations -- the same mechanism any task uses. Context resolution works across retries; the framework rebuilds task identity consistently so the reviewer always reads the most recent attempt's output.

The guide at agentensemble.net/guides/phase-review/ covers the full API including custom reviewer implementations and review event callbacks. The example source is runnable from the repository.

AgentEnsemble is open-source under the MIT license.

Using AgentEnsemble as a Pure Workflow Engine: Orchestration Without AI

mgd43b — Thu, 09 Apr 2026 14:00:00 +0000

Most of the conversation around agent frameworks assumes that orchestration is synonymous with LLM calls. Pick a framework, wire up your model, define your agents. That framing is understandable, but it leaves something useful on the table.

The harder problem for many Java teams isn't choosing a model. It's building pipelines that are observable, composable, and consistent -- whether or not any given step involves an LLM. A pipeline that mixes REST API calls, data transformations, and a few AI-backed steps shouldn't need two separate orchestration systems.

AgentEnsemble is not exclusively an AI orchestrator. The same DAG execution engine, parallel phase runner, callbacks, and review gates that coordinate LLM-backed agents also work for purely deterministic Java functions. No ChatModel required.

This post covers the deterministic-only orchestration pattern: what the API looks like, how data flows between tasks, and when this pattern is worth reaching for.

The Static Factory for Deterministic Pipelines

For pipelines with no LLM steps, the entry point is a static factory method that takes no model parameter:

Ensemble.run(task1, task2, task3);

If any task passed to this method lacks a handler, the framework throws at startup. This keeps the contract explicit: deterministic tasks must always declare how they execute.

For more complex setups -- phases, callbacks, guardrails -- you use the builder:

Ensemble.builder()
    .phases(List.of(ingestion, processing, output))
    .onTaskComplete(event -> log.info("Completed: {}", event.taskId()))
    .build()
    .run();

The builder path accepts a chatModel, but it isn't required. If all tasks have handlers, the ensemble runs without one.

Handlers and Task Output

A deterministic task declares its behavior via a handler lambda. The handler receives a TaskHandlerContext and returns a ToolResult:

Task extract = Task.builder()
    .description("extract-records")
    .handler(ctx -> ToolResult.success("42 records extracted"))
    .build();

ToolResult.success(String content) wraps the output. ToolResult.failure(String reason) triggers task failure handling. The string content becomes the task's raw output, accessible downstream.

For tasks that need to fail conditionally:

Task validate = Task.builder()
    .description("validate-input")
    .handler(ctx -> {
        String input = ctx.contextOutputs().get(0).getRaw();
        if (input.isBlank()) {
            return ToolResult.failure("input was empty");
        }
        return ToolResult.success("valid: " + input.length() + " chars");
    })
    .build();

Data Passing Between Tasks

Downstream tasks declare upstream dependencies via .context(). The framework resolves outputs in declaration order and delivers them through ctx.contextOutputs():

Task extract = Task.builder()
    .description("extract-records")
    .handler(ctx -> ToolResult.success("42 records"))
    .build();

Task transform = Task.builder()
    .description("transform-records")
    .context(List.of(extract))
    .handler(ctx -> {
        String raw = ctx.contextOutputs().get(0).getRaw();
        return ToolResult.success("transformed: " + raw);
    })
    .build();

Task load = Task.builder()
    .description("load-records")
    .context(List.of(transform))
    .handler(ctx -> {
        String data = ctx.contextOutputs().get(0).getRaw();
        // write to database
        return ToolResult.success("loaded: " + data);
    })
    .build();

Ensemble.run(extract, transform, load);

Multiple upstream outputs are accessible by index in the order they were declared:

Task merge = Task.builder()
    .description("merge-results")
    .context(List.of(fetchA, fetchB, fetchC))
    .handler(ctx -> {
        List<String> results = ctx.contextOutputs().stream()
            .map(TaskOutput::getRaw)
            .toList();
        return ToolResult.success(String.join(", ", results));
    })
    .build();

Parallel Execution

Independent tasks run concurrently on virtual threads automatically. If tasks declare no context() dependency on each other, the executor runs them in parallel:

Task fetchA = Task.builder()
    .description("fetch-region-a")
    .handler(ctx -> ToolResult.success("region-a: 120 records"))
    .build();

Task fetchB = Task.builder()
    .description("fetch-region-b")
    .handler(ctx -> ToolResult.success("region-b: 98 records"))
    .build();

Task fetchC = Task.builder()
    .description("fetch-region-c")
    .handler(ctx -> ToolResult.success("region-c: 77 records"))
    .build();

Task merge = Task.builder()
    .description("merge-all-regions")
    .context(List.of(fetchA, fetchB, fetchC))
    .handler(ctx -> {
        List<String> outputs = ctx.contextOutputs().stream()
            .map(TaskOutput::getRaw)
            .toList();
        return ToolResult.success(String.join("; ", outputs));
    })
    .build();

Ensemble.run(fetchA, fetchB, fetchC, merge);

The DAG is inferred from context() declarations. merge waits for all three fetch tasks. The three fetches run concurrently. No explicit thread management required.

You can also be explicit via .workflow(Workflow.PARALLEL) on individual tasks, but for most cases the DAG inference is sufficient.

Named Phases

For larger pipelines with distinct stages, phases provide explicit sequencing and named output access:

Task ingest = Task.builder()
    .description("ingest-raw-data")
    .handler(ctx -> ToolResult.success("1000 rows ingested"))
    .build();

Task schemaCheck = Task.builder()
    .description("validate-schema")
    .handler(ctx -> ToolResult.success("schema valid"))
    .build();

Phase ingestion = Phase.of("ingestion", ingest, schemaCheck);

Task normalize = Task.builder()
    .description("normalize-records")
    .handler(ctx -> ToolResult.success("normalized"))
    .build();

Task enrich = Task.builder()
    .description("enrich-with-reference-data")
    .handler(ctx -> ToolResult.success("enriched"))
    .build();

Phase processing = Phase.builder()
    .name("processing")
    .tasks(List.of(normalize, enrich))
    .after(ingestion)
    .build();

Task writeOutput = Task.builder()
    .description("write-to-data-warehouse")
    .handler(ctx -> ToolResult.success("written"))
    .build();

Phase output = Phase.builder()
    .name("output")
    .tasks(List.of(writeOutput))
    .after(processing)
    .build();

EnsembleOutput result = Ensemble.builder()
    .phases(List.of(ingestion, processing, output))
    .build()
    .run();

Phase outputs are accessible by name:

PhaseOutput ingestionOut = result.getPhaseOutputs().get("ingestion");
String ingestedData = ingestionOut.taskOutputs().get(0).getRaw();

Within a phase, task parallelism follows the same DAG rules. normalize and enrich run concurrently if they don't depend on each other.

Callbacks and Observability

The same onTaskStart, onTaskComplete, and onTaskFailed callbacks work for deterministic tasks:

Ensemble.builder()
    .phases(List.of(ingestion, processing, output))
    .onTaskStart(e -> log.info("[START] {}", e.taskId()))
    .onTaskComplete(e -> log.info("[DONE]  {} -> {}", e.taskId(), e.output().getRaw()))
    .onTaskFailed(e -> log.error("[FAIL]  {} -> {}", e.taskId(), e.error().getMessage()))
    .build()
    .run();

Metrics via Micrometer and OpenTelemetry tracing are equally available. Deterministic tasks participate in the same span hierarchy as AI tasks. If your ensemble mixes both, traces show the full picture.

Mixing Deterministic and AI Tasks

Deterministic and AI-backed tasks compose freely within the same pipeline:

Task fetchData = Task.builder()
    .description("fetch-customer-records")
    .handler(ctx -> ToolResult.success(fetchFromDatabase()))
    .build();

Task analyzeData = Task.builder()
    .description("identify key themes and concerns from these customer records")
    .chatModel(model)
    .context(List.of(fetchData))
    .build();

Task writeReport = Task.builder()
    .description("format-as-json-report")
    .context(List.of(analyzeData))
    .handler(ctx -> {
        String analysis = ctx.contextOutputs().get(0).getRaw();
        return ToolResult.success(toJson(analysis));
    })
    .build();

Ensemble.builder()
    .tasks(List.of(fetchData, analyzeData, writeReport))
    .chatModel(model)
    .build()
    .run();

The chatModel on the ensemble is only used for tasks that don't have a handler. Deterministic tasks bypass the LLM entirely.

When to Use This

Deterministic-only orchestration in AgentEnsemble is useful when you need:

Multi-step pipelines with DAG dependencies -- sequential or parallel, inferred from context declarations
Observable step-level execution -- callbacks, metrics, and tracing without external tooling
Mixed pipelines -- some steps are AI-backed, some are not; one orchestration model covers both
Phase-level sequencing -- named stages with explicit ordering and structured output access
Review gates on deterministic steps -- the PhaseReview API works regardless of whether the tasks inside are AI or handler-based

It is not a replacement for a dedicated workflow engine like Temporal if you need durable execution, long-running workflows with days or weeks of persistence, or replays across process restarts. For those requirements, Temporal or a similar system is the right choice.

For pipelines that live within a single process, run in seconds or minutes, and need consistent observability without adding infrastructure, this pattern removes a lot of plumbing.

Tradeoffs

No persistence: if the process dies mid-run, the pipeline restarts from the beginning. There is no checkpoint mechanism.

No scheduling: the orchestration layer has no built-in cron or trigger. External schedulers (Quartz, Spring Scheduler, Kubernetes CronJobs) are needed if the pipeline runs on a schedule.

No distributed execution: all tasks run in-process on virtual threads. Cross-JVM task distribution is not supported.

These are acceptable tradeoffs for many internal pipelines. They are not appropriate for workflows that span hours or need durable state across process restarts.

The guide at agentensemble.net/guides/deterministic-orchestration/ covers the full API including guardrails and review gate integration. The example source is runnable directly from the repository.

AgentEnsemble is open-source under the MIT license.

Parallel Workstreams in Java Agent Pipelines: Phase-Level Grouping and DAG Execution

mgd43b — Mon, 06 Apr 2026 14:00:00 +0000

Flat task lists work well for sequential pipelines. You define tasks, wire context dependencies, and the framework runs them in order. For a research-then-write pipeline, that's all you need.

The problem shows up when you have work that genuinely doesn't need to happen in sequence -- multiple independent workstreams that can proceed in parallel, followed by a convergence point where all of them must finish before the next stage can begin.

The task-level context() mechanism can model some of this. The parallel workflow executor infers a dependency graph from context declarations and runs independent tasks concurrently. But it operates at the individual task level. There's no concept of a named workstream, no defined barrier, and no way to say "these three groups of tasks run at the same time, and this fourth group waits for all three."

That gap is what the Phase abstraction addresses in AgentEnsemble.

The Kitchen Problem

Before getting to code, it helps to have a concrete example of what a flat task list can't express.

Imagine three diners order different main courses -- steak, salmon, and pasta. Each dish has its own sequential preparation steps: prep, cook, plate. All three dishes are prepared simultaneously and served together once every dish is ready.

A flat task list with context-chaining can't express this cleanly. You'd need to manually wire cross-task dependencies between the three workstreams, and the dependency graph would become a tangle of context references. More importantly, the intent is lost: a reader looking at the task list can't immediately see that you have three independent workstreams converging into one.

Introducing Phase

A phase is a named group of tasks. Phases declare dependencies on each other via after(). Independent phases execute in parallel -- each in its own virtual thread. A phase only starts when all of its declared predecessors have completed.

Phase steak = Phase.builder()
    .name("steak")
    .task(Task.of("Prepare steak", "Seasoned and at room temperature"))
    .task(Task.of("Sear steak", "Medium-rare, rested for 5 minutes"))
    .task(Task.of("Plate steak", "Plated with garnish and sauce"))
    .build();

Phase salmon = Phase.builder()
    .name("salmon")
    .task(Task.of("Prepare salmon", "Skin removed, seasoned"))
    .task(Task.of("Cook salmon", "Crispy skin, fully cooked"))
    .task(Task.of("Plate salmon", "Plated with lemon and herbs"))
    .build();

Phase pasta = Phase.builder()
    .name("pasta")
    .task(Task.of("Boil pasta", "Al dente"))
    .task(Task.of("Make sauce", "Reduced tomato sauce, seasoned"))
    .task(Task.of("Plate pasta", "Pasta and sauce combined, topped with basil"))
    .build();

Phase serve = Phase.builder()
    .name("serve")
    .after(steak, salmon, pasta)     // barrier: waits for all three
    .task(Task.of("Deliver all plates", "All three dishes delivered simultaneously"))
    .build();

EnsembleOutput output = Ensemble.builder()
    .chatLanguageModel(llm)
    .phase(steak)
    .phase(salmon)
    .phase(pasta)
    .phase(serve)
    .build()
    .run();

Execution timeline:

t=0  [steak starts]  [salmon starts]  [pasta starts]
t=?                                                   [serve starts when last finishes]

The intent is now legible from the structure. You can see that three workstreams run in parallel and converge before serving. The framework handles the concurrency -- you declare the shape.

The Phase DAG

Phases form a directed acyclic graph. Phases with no after() declaration are root phases and start immediately. When a phase completes, AgentEnsemble checks whether any successors now have all predecessors satisfied, and if so starts them immediately.

Phase A = Phase.of("A", taskA);
Phase B = Phase.of("B", taskB);
Phase C = Phase.of("C", taskC);
Phase D = Phase.builder().name("D").after(B, C).task(taskD).build();

// Execution:
// [A]   [B]   [C]    -- root phases, all start simultaneously
//        |     |
//        +--+--+
//           |
//          [D]        -- starts when B and C both complete

The DAG is validated at build time. Cycles, duplicate phase names, and cross-phase context() references to non-predecessor phases all produce a ValidationException before any execution begins.

Per-Phase Workflow Strategy

Each phase can use a different internal workflow strategy for its tasks. This is useful when different stages of a pipeline have different concurrency characteristics.

Phase dataGathering = Phase.builder()
    .name("data-gathering")
    .workflow(Workflow.PARALLEL)    // all three fetch tasks run concurrently
    .task(Task.of("Fetch sales data", "Sales CSV"))
    .task(Task.of("Fetch inventory data", "Inventory CSV"))
    .task(Task.of("Fetch customer data", "Customer CSV"))
    .build();

Phase analysis = Phase.builder()
    .name("analysis")
    .workflow(Workflow.SEQUENTIAL)  // each step depends on the previous
    .after(dataGathering)
    .task(Task.of("Merge datasets", "Combined dataset"))
    .task(Task.of("Compute metrics", "KPI summary"))
    .task(Task.of("Generate report", "Final report"))
    .build();

EnsembleOutput output = Ensemble.builder()
    .chatLanguageModel(llm)
    .phase(dataGathering)
    .phase(analysis)
    .build()
    .run();

The gather phase fetches all three data sources in parallel. The analysis phase sequences its tasks because each depends on the output of the previous one. Both phases run with the right strategy for their internal structure, without any global configuration change.

If no workflow is set on a phase, it inherits the ensemble-level workflow, which itself defaults to SEQUENTIAL.

Cross-Phase Context

Tasks in a later phase can reference tasks from an earlier phase using the standard context() mechanism. The phase DAG guarantees that earlier phases complete before later phases start, so cross-phase context is always safe to resolve.

Task marketTask = Task.builder()
    .description("Research market positioning for a Java developer tooling product")
    .expectedOutput("Bullet-point market positioning summary")
    .build();

Task technicalTask = Task.builder()
    .description("Assess implementation complexity for a Java multi-agent framework")
    .expectedOutput("Complexity score with rationale and integration checklist")
    .build();

Phase marketResearch = Phase.of("market-research", marketTask);
Phase technicalResearch = Phase.of("technical-research", technicalTask);

Phase report = Phase.builder()
    .name("report")
    .after(marketResearch, technicalResearch)
    .task(Task.builder()
        .description("Write a product feasibility report combining market and technical findings")
        .expectedOutput("One-page feasibility report with go/no-go recommendation")
        .context(List.of(marketTask, technicalTask))   // cross-phase references
        .build())
    .build();

market-research and technical-research run in parallel. When both complete, report starts and the report task can read their outputs via context(). The ordering is enforced by the DAG -- there's no risk of the report task starting before both inputs are ready.

Error Handling

When a phase fails, only its direct and transitive dependents are affected. Independent phases continue running.

[A]  [B]  [C]

      B fails
      |
     [D depends on B]  <-- skipped
                  [E depends on A]  <-- still runs, A is independent

This is important for real workloads. If one data-gathering workstream fails (say, a rate limit or timeout), an unrelated workstream should complete normally. The EnsembleOutput records failures and skips so you can inspect exactly what happened.

If you need all workstreams to succeed before proceeding, structure the DAG so that the final phase depends on all of them -- that barrier naturally enforces it.

Reading Phase Outputs

EnsembleOutput adds a phase-keyed map alongside the existing flat list of task outputs:

EnsembleOutput output = ensemble.run();

// Backward-compatible flat list
List<TaskOutput> all = output.getTaskOutputs();

// Phase-keyed map
Map<String, List<TaskOutput>> byPhase = output.getPhaseOutputs();
List<TaskOutput> marketResults    = byPhase.get("market-research");
List<TaskOutput> technicalResults = byPhase.get("technical-research");
List<TaskOutput> reportResults    = byPhase.get("report");

// Final output: last task of the last phase to complete
String finalText = output.getFinalOutput();

Flat-task ensembles return an empty map from getPhaseOutputs(), so existing code that reads getTaskOutputs() continues to work unchanged.

When to Use Phases

Use this	When
Flat tasks	Simple sequential pipeline, linear context-chaining
Parallel workflow	Fine-grained task-level concurrency, tasks depend on each other within the same conceptual group
Phases	Named workstreams with clear start/end, multiple groups that run in parallel, explicit convergence barriers, per-group workflow strategy

The key question is whether you think in terms of named stages. If you do -- "first we gather data in parallel, then we analyze" -- phases express that directly. If the dependency structure is purely task-to-task with no meaningful grouping, context-chaining in a flat list is simpler.

Do not use phases for simple sequential pipelines. A two-task research-write pipeline doesn't need phases -- the extra structure adds noise without adding clarity.

What Didn't Change

Phases are additive. The task() and phase() builder methods are mutually exclusive on the same ensemble -- you can't mix them -- but all existing flat-task ensembles continue to work exactly as before. Phase is a new layer on top, not a replacement.

The Task, Agent, and WorkflowExecutor types are unchanged. Per-task configuration (model, tools, output type, max iterations, context), agent synthesis, deterministic handlers -- all of it works inside phases exactly as it does outside them.

The parallel workflow strategy behind PhaseDagExecutor uses virtual threads, consistent with how ParallelWorkflowExecutor handles task-level concurrency. There's one virtual thread per phase, coordinated through a CountDownLatch that tracks completion across the DAG.

The Honest Tradeoff

Phases add structure. Structure is most valuable when the problem is genuinely structured -- when you have multiple independent workstreams, named stages, or convergence barriers that matter.

For simple pipelines, the structure is overhead. A Phase.of("research", task1, task2) wrapping two sequential tasks is just ceremony. Flat tasks are cleaner when the pipeline is essentially linear.

The useful abstraction here is the named barrier: a place where multiple concurrent workstreams must all finish before work continues. That's where phases earn their place.

Documentation:

Phases Guide -- when to use phases, builder API, cross-phase context, error handling
Phases Examples -- runnable code: sequential phases, kitchen scenario, per-phase workflow, diamond DAG
Getting Started -- up and running in 5 minutes
GitHub -- source, issues, and contributions

AgentEnsemble is MIT-licensed and available on GitHub.

The Agent Is an Implementation Detail: Task-First Orchestration in Java

mgd43b — Fri, 03 Apr 2026 14:00:00 +0000

Most agent frameworks start with the agent.

You define a role, write a goal, add a backstory, choose a model. Then you write a task and wire the agent to it. Then you pass both to a crew or ensemble. Three objects defined and wired together before the actual work is even described.

It works. But for a lot of use cases, that agent definition is accidental complexity.

You're not thinking "I need a Researcher persona with a carefully tuned goal statement." You're thinking "I need this research done, then I need a report written from it." The agent is an implementation detail. The task is the actual unit of work.

That's the central insight behind the task-first design in AgentEnsemble.

The Old Pattern

The v1.x API required explicit agent definitions. Here's a two-task research-writer pipeline:

Agent researcher = Agent.builder()
    .role("Senior Researcher")
    .goal("Find comprehensive information about {{topic}}")
    .background("Expert at synthesizing information from multiple sources")
    .build();

Agent writer = Agent.builder()
    .role("Technical Writer")
    .goal("Write clear, engaging content")
    .background("Skilled at making complex topics accessible")
    .build();

Task researchTask = Task.builder()
    .description("Research {{topic}} thoroughly")
    .expectedOutput("Detailed research notes")
    .agent(researcher)
    .build();

Task writeTask = Task.builder()
    .description("Write an article based on the research")
    .expectedOutput("A polished article")
    .agent(writer)
    .context(List.of(researchTask))
    .build();

Ensemble.builder()
    .agents(researcher, writer)
    .tasks(researchTask, writeTask)
    .chatLanguageModel(model)
    .inputs(Map.of("topic", "WebAssembly"))
    .build()
    .run();

Five object definitions for a two-task pipeline. The agent persona fields -- role, goal, background -- look important, but in many cases a sensible default would work just as well.

The Task-First Approach

In v2, agents are optional. When a task has no explicit agent, the framework synthesizes one:

Task researchTask = Task.builder()
    .description("Research {{topic}} thoroughly")
    .expectedOutput("Detailed research notes")
    .build();

Task writeTask = Task.builder()
    .description("Write an article based on the research")
    .expectedOutput("A polished article")
    .context(List.of(researchTask))
    .build();

Ensemble.builder()
    .chatLanguageModel(model)
    .tasks(researchTask, writeTask)
    .inputs(Map.of("topic", "WebAssembly"))
    .build()
    .run();

Same pipeline, no agent definitions. For the simplest case, it collapses further:

EnsembleOutput output = Ensemble.run(model,
    Task.of("Research {{topic}}", "Detailed research notes"),
    Task.of("Write an article based on the research", "A polished article"));

How Synthesis Works

The framework uses an AgentSynthesizer to derive role and goal from the task description. The default is template-based -- a verb-to-role lookup applied to the first word of the description:

First verb	Synthesized role
Research / Investigate	Researcher
Write / Draft / Compose	Writer
Analyze / Evaluate	Analyst
Build / Implement	Developer
Summarize	Summarizer
Review	Reviewer
Plan	Planner
(anything else)	Agent

The goal is set to the full task description. No extra LLM call is made. The synthesized agent is ephemeral -- it exists for the duration of one task execution and is discarded.

For higher-quality personas, an LLM-based synthesizer is available as an opt-in:

Ensemble.builder()
    .chatLanguageModel(model)
    .agentSynthesizer(AgentSynthesizer.llmBased())
    .tasks(researchTask, writeTask)
    .build()
    .run();

This makes one additional LLM call per agentless task to generate a tailored role, goal, and background. The cost is a few extra tokens per task; the benefit is a more domain-specific system prompt going into the LLM call that actually does the work.

Per-Task Configuration Without Explicit Agents

The zero-ceremony path does not mean zero configuration. Task-level LLM, tools, and iteration limits still work:

Task researchTask = Task.builder()
    .description("Research {{topic}} using recent web sources")
    .expectedOutput("Research notes with citations")
    .chatLanguageModel(gpt4o)
    .tools(List.of(new WebSearchTool()))
    .maxIterations(15)
    .build();

Task summaryTask = Task.builder()
    .description("Write a concise executive summary")
    .expectedOutput("A 200-word summary")
    .chatLanguageModel(gpt4oMini)
    .build();

The task-level chatLanguageModel takes precedence over the ensemble default. Different tasks can use different models without declaring separate agent objects.

This is the typical production configuration: a powerful model for complex or expensive tasks, a cheaper model for simpler ones, all without any agent persona boilerplate.

Structured Output Still Works

Typed output is a task concern, not an agent concern:

record ResearchReport(String title, List<String> findings, String conclusion) {}

Task task = Task.builder()
    .description("Research AI adoption trends in healthcare")
    .expectedOutput("A structured research report")
    .chatLanguageModel(model)
    .outputType(ResearchReport.class)
    .build();

EnsembleOutput result = Ensemble.run(model, task);
ResearchReport report = result.getTaskOutputs().get(0)
    .getParsedOutput(ResearchReport.class);

The synthesized agent handles structured output exactly as an explicit agent would. The JSON schema is derived from the record, injected into the prompt, and the response is deserialized and validated.

When Explicit Agents Are Still Useful

The task-first approach is the default, not the only option. Explicit agents make sense when you need:

A crafted persona with domain-specific background. The background field is the highest-leverage part of agent configuration -- a well-written backstory shapes how the LLM reasons about the task. For specialized work (healthcare analysis, legal review, financial modelling), a carefully written background can make a measurable difference:

Agent seniorAnalyst = Agent.builder()
    .role("Senior Healthcare Analyst")
    .goal("Provide rigorous, evidence-based analysis")
    .background("You are a CFA-certified analyst specializing in clinical-stage "
        + "biotech with 20 years of experience evaluating Phase II/III trial data.")
    .llm(anthropicModel)
    .build();

Shared agent identity across tasks. When the same persona handles multiple related tasks and persona continuity matters, an explicit agent can be bound to each:

Task firstAnalysis = Task.builder()
    .description("Analyse Q1 performance")
    .agent(analystAgent).build();

Task secondAnalysis = Task.builder()
    .description("Compare Q1 against Q2")
    .agent(analystAgent)  // same agent, same persona
    .context(List.of(firstAnalysis)).build();

Verbose logging or custom response format. Agent-level configuration fields like verbose and responseFormat require an explicit agent:

Agent debuggable = Agent.builder()
    .role("Researcher")
    .goal("Research the topic thoroughly")
    .verbose(true)   // logs prompts and responses at INFO level
    .llm(model)
    .build();

Hierarchical delegation. The allowDelegation field is an agent property:

Agent manager = Agent.builder()
    .role("Research Director")
    .goal("Coordinate research across multiple domains")
    .allowDelegation(true)
    .llm(managerModel)
    .build();

The key point is that explicit agents are an opt-in power-user path, not the required starting point.

The Tradeoff

The honest tradeoff here is between convenience and persona quality.

Template-based synthesis is fast (no extra LLM call), predictable (the lookup is deterministic), and sufficient for most workloads. For tasks where the first verb maps cleanly to a role -- research, write, analyze, summarize -- the synthesized persona is functionally equivalent to a manually declared one.

For specialized domains, a handwritten background field can outperform synthesis. The LLM's reasoning is shaped by its system prompt; domain-specific context in the background gives it better framing for complex tasks. If you're running healthcare analysis or legal review, the extra ten minutes writing an explicit agent background is likely worth it.

The AgentSynthesizer interface is pluggable for teams that want something in between:

AgentSynthesizer domainSynthesizer = (task, ctx) -> Agent.builder()
    .role(derivedRole(task))
    .goal(task.getDescription())
    .background(loadBackgroundTemplate(task))
    .llm(ctx.model())
    .build();

Ensemble.builder()
    .agentSynthesizer(domainSynthesizer)
    .tasks(...)
    .build().run();

Practical Takeaway

The question to ask when designing a task pipeline is not "what agents do I need?" but "what work needs to be done, and in what order?"

For the majority of cases, that question is answered entirely by the task definitions: what each task is asked to do, what output it should produce, what context it needs from prior tasks, and what tools or model it should use. The agent emerges from those decisions, not the other way around.

Explicit agents remain available for cases where persona quality, shared identity, or per-agent configuration matter. But the default path is simpler: define the tasks, wire the dependencies, run.

Get started:

Tasks Guide -- zero-ceremony and full builder API
Agents Guide -- explicit agents and AgentSynthesizer reference
Getting Started -- up and running in 5 minutes
GitHub -- source, issues, and contributions

AgentEnsemble is MIT-licensed and available on GitHub.

Debugging Multi-Agent Systems: Traces, Capture Mode, and Live Dashboards

mgd43b — Tue, 31 Mar 2026 14:00:00 +0000

Multi-agent systems are hard to debug.

It's not the same as debugging a web request or a database query. You can't set a breakpoint in the middle of an LLM call. You can't predict what the model will say. When an agent produces bad output, you need to understand the full chain of events: what prompt was sent, what the model returned, which tools were called, what context from previous tasks was injected, and whether the output parsing succeeded.

Traditional debuggers don't help here. You need purpose-built observability.

This post covers the debugging and observability stack in AgentEnsemble: structured traces for post-mortem analysis, capture mode for recording full execution state, and the live dashboard for real-time visibility during development.

The Debugging Challenge

Consider a three-agent pipeline: Researcher, Analyst, Writer. The Writer produces a report that's factually wrong. Where did things go wrong?

Did the Researcher find bad information?
Did the Analyst misinterpret the research?
Did the Writer ignore the analysis and hallucinate?
Did a tool call return unexpected results?
Was the wrong context passed between tasks?

Without observability, you're guessing. With it, you're reading a log.

Layer 1: Structured Traces

The most broadly useful debugging tool is the structured trace. It records every significant event in an ensemble run as a tree of spans:

EnsembleOutput output = Ensemble.builder()
    .agents(researcher, analyst, writer)
    .tasks(researchTask, analysisTask, writeTask)
    .chatLanguageModel(model)
    .traceExporter(TraceExporter.json(Path.of("traces/")))
    .build()
    .run();

This produces a JSON file in the traces/ directory with a structure like:

Ensemble Run (total: 8,420ms, 5,230 tokens)
 |
 +-- Task: Research emerging trends (3,240ms, 1,847 tokens)
 |    +-- LLM Call #1 (1,900ms, 1,200 tokens)
 |    +-- Tool: WebSearch "emerging tech trends 2024" (890ms)
 |    +-- LLM Call #2 (450ms, 647 tokens)
 |
 +-- Task: Analyze research findings (2,180ms, 1,583 tokens)
 |    +-- LLM Call #1 (2,180ms, 1,583 tokens)
 |
 +-- Task: Write final report (3,000ms, 1,800 tokens)
      +-- LLM Call #1 (2,400ms, 1,400 tokens)
      +-- LLM Call #2 (600ms, 400 tokens)  // output retry

Each span records:

Field	Description
Name	Task description or tool call name
Duration	Wall-clock time in milliseconds
Token count	Input + output tokens for LLM calls
Status	Success, failure, or retry
Input/Output	What went in, what came out

Accessing Traces Programmatically

You don't have to read the JSON file. The trace is available on the EnsembleOutput:

ExecutionTrace trace = output.getTrace();

// Walk the span tree
for (TraceSpan span : trace.getSpans()) {
    System.out.printf("[%s] %s -- %dms, %d tokens%n",
        span.getStatus(),
        span.getName(),
        span.getDurationMs(),
        span.getTokenCount());

    for (TraceSpan child : span.getChildren()) {
        System.out.printf("  [%s] %s -- %dms%n",
            child.getStatus(),
            child.getName(),
            child.getDurationMs());
    }
}

This is useful for writing assertions in tests:

@Test
void ensembleShouldCompleteAllTasks() {
    EnsembleOutput output = ensemble.run();

    ExecutionTrace trace = output.getTrace();
    assertThat(trace.getSpans()).hasSize(3);
    assertThat(trace.getSpans())
        .allMatch(span -> span.getStatus() == TraceStatus.SUCCESS);
    assertThat(output.getMetrics().getTotalTokens()).isLessThan(10_000);
}

Trace Export for Analysis Pipelines

The JSON trace format is designed for programmatic consumption. Feed it into your log aggregation system, build custom analysis scripts, or import it into a notebook:

// Export to a specific directory with timestamped filenames
.traceExporter(TraceExporter.json(Path.of("traces/")))

// Or get the raw JSON string
String traceJson = output.getTrace().toJson();
logAggregator.ingest("agent-trace", traceJson);

Layer 2: Capture Mode

Traces tell you what happened. Capture mode tells you exactly what happened -- including the full prompts, raw LLM responses, and tool call payloads.

Three Levels

Ensemble.builder()
    .agents(researcher, writer)
    .tasks(researchTask, writeTask)
    .chatLanguageModel(model)
    .captureMode(CaptureMode.FULL) // OFF, STANDARD, or FULL
    .build()
    .run();

Level	What's Captured	Use Case
`OFF`	Standard metrics only	Production
`STANDARD`	+ Full LLM message history per iteration, memory operations	Staging, initial deployment
`FULL`	+ Tool call I/O payloads, raw LLM responses, detailed timing	Development, debugging

What STANDARD Adds

With CaptureMode.STANDARD, each task's execution record includes the full conversation between the framework and the LLM:

Task: Research emerging trends
  Iteration 1:
    System prompt: "You are Senior Research Analyst. Your goal is..."
    User message: "Research emerging trends in AI thoroughly..."
    Assistant response: "I'll search for the latest information..."
    Tool call: WebSearch("emerging AI trends 2024")
  Iteration 2:
    System prompt: [same]
    User message: [previous context + tool result]
    Assistant response: "Based on my research, here are the key..."

This is invaluable for understanding why an agent behaved a certain way. You can see exactly what prompt it received, what context was injected, and how it reasoned through the task.

What FULL Adds

CaptureMode.FULL adds the raw payloads for every interaction:

Tool call inputs: The exact arguments passed to each tool.
Tool call outputs: The exact response from each tool.
Raw LLM responses: The complete response body, including any JSON that was parsed.
Timing breakdowns: Per-iteration timing, not just per-task.

This is the level you use when something is wrong and you can't figure out why from the trace alone. It's verbose -- expect significantly more data -- but it gives you full replay capability.

Using Capture Data in Tests

Capture mode is a testing power tool. Record a full execution, then write assertions against the captured data:

@Test
void researcherShouldUseWebSearch() {
    EnsembleOutput output = Ensemble.builder()
        .agents(researcher, writer)
        .tasks(researchTask, writeTask)
        .chatLanguageModel(model)
        .captureMode(CaptureMode.FULL)
        .build()
        .run();

    // Verify the researcher used the web search tool
    ExecutionTrace trace = output.getTrace();
    TraceSpan researchSpan = trace.getSpans().get(0);

    boolean usedWebSearch = researchSpan.getChildren().stream()
        .anyMatch(child -> child.getName().contains("WebSearch"));
    assertThat(usedWebSearch).isTrue();
}

You can also capture a "golden run" and use it as a reference for regression testing -- comparing future runs against the expected execution pattern.

Layer 3: Event Callbacks

For real-time debugging during development, callbacks give you a live stream of execution events:

Ensemble.builder()
    .agents(researcher, analyst, writer)
    .tasks(researchTask, analysisTask, writeTask)
    .chatLanguageModel(model)
    .listener(event -> {
        switch (event) {
            case TaskStartEvent e ->
                System.out.printf("%n>>> Starting: %s (agent: %s)%n",
                    e.taskDescription(), e.agentRole());

            case TaskCompleteEvent e ->
                System.out.printf("<<< Completed: %s (%dms, %d tokens)%n",
                    e.taskDescription(), e.durationMs(), e.tokenCount());

            case TaskFailedEvent e ->
                System.err.printf("!!! Failed: %s -- %s%n",
                    e.taskDescription(), e.errorMessage());

            case ToolCallEvent e ->
                System.out.printf("    [tool] %s(%s) -> %s%n",
                    e.toolName(),
                    truncate(e.input(), 50),
                    truncate(e.result(), 100));

            case DelegationStartedEvent e ->
                System.out.printf("    [delegate] %s -> %s%n",
                    e.fromAgent(), e.toAgent());

            case TokenEvent e ->
                // Streaming: print tokens as they arrive
                System.out.print(e.token());

            default -> {}
        }
    })
    .build()
    .run();

This gives you a live play-by-play of the ensemble execution in your terminal. You see each task start and complete, each tool call and its result, and each delegation in hierarchical workflows.

Combining Callbacks with Logging

For persistent debugging output, route events to your logging framework:

.listener(event -> {
    if (event instanceof TaskCompleteEvent e) {
        log.info("Task completed: task={}, agent={}, duration={}ms, tokens={}",
            e.taskDescription(), e.agentRole(),
            e.durationMs(), e.tokenCount());
    }
    if (event instanceof TaskFailedEvent e) {
        log.error("Task failed: task={}, error={}",
            e.taskDescription(), e.errorMessage());
    }
})

These flow into your existing log aggregation pipeline (ELK, Splunk, CloudWatch Logs) alongside your application's other logs.

Layer 4: The Live Dashboard

For the most visual debugging experience, AgentEnsemble includes a live browser dashboard:

Ensemble.builder()
    .agents(researcher, analyst, writer)
    .tasks(researchTask, analysisTask, writeTask)
    .chatLanguageModel(model)
    .devtools(Devtools.enabled())
    .build()
    .run();

When the ensemble starts, a browser window opens (or a URL is printed to the console) showing a real-time visualization of the execution:

What the Dashboard Shows

DAG Visualization: A graph of all tasks and their dependencies. Nodes change color as tasks progress from pending to running to completed.
Agent Activity: Which agent is currently active, what it's doing, and how many iterations it's taken.
Token Consumption: Real-time token counters per task and for the entire ensemble.
Task Output Preview: Click on a completed task to see its output.
Timeline: A Gantt-chart-style view of task execution, showing parallelism and bottlenecks.

When to Use It

The live dashboard is a development tool, not a production monitoring dashboard. Use it when:

Building a new agent workflow and you want to see the execution flow.
Debugging why a specific task takes too long or produces unexpected output.
Demonstrating an agent system to stakeholders.
Understanding the parallelism in a DAG or MapReduce workflow.

For production monitoring, use the Micrometer metrics integration and your existing Grafana/Prometheus stack.

Debugging Recipes

Here are specific debugging scenarios and how to approach them with the tools above.

"The output is wrong, but I don't know which agent failed"

Use traces. Look at each task's output in the trace tree. Find the first task whose output is incorrect -- that's where things diverged.

.traceExporter(TraceExporter.json(Path.of("debug/")))

Then read the trace JSON, find the task with bad output, and check its input context to see what it received from upstream tasks.

"The agent keeps calling the same tool in a loop"

Use capture mode + callbacks. Enable CaptureMode.FULL and add a callback that logs tool calls:

.captureMode(CaptureMode.FULL)
.listener(event -> {
    if (event instanceof ToolCallEvent e) {
        log.warn("Tool call: {} with input: {}",
            e.toolName(), e.input());
    }
})

Then check the captured LLM conversation to see why the agent keeps making the same call. Usually it's a prompt issue -- the agent doesn't recognize the tool result as sufficient.

"The structured output parsing keeps failing"

Use capture mode. Enable CaptureMode.FULL and check the raw LLM response:

.captureMode(CaptureMode.FULL)

The captured data includes the raw response before parsing. Compare it to your record schema. Common issues:

The LLM wraps JSON in markdown code blocks.
Field names don't match (the LLM uses camelCase, the record uses snake_case).
The LLM adds extra fields or comments.

The framework handles most of these, but FULL capture mode shows you exactly what's happening.

"A parallel workflow is slower than expected"

Use the live dashboard. Enable devtools and look at the timeline view:

.devtools(Devtools.enabled())

You'll see whether tasks are actually running in parallel or if there's an unexpected dependency bottleneck. Common issues:

A task accidentally depends on another task via context() when it shouldn't.
One task takes much longer than the others, creating a bottleneck for downstream tasks.
Rate limiting is causing parallel tasks to serialize.

"I need to understand the full prompt the agent received"

Use CaptureMode.STANDARD or CaptureMode.FULL. The captured data includes the complete system prompt, user message, and any injected context for each LLM call.

This is the only way to see the actual prompt -- the framework constructs it dynamically from the agent's role/goal/background, the task description, context from previous tasks, and tool results.

Putting It All Together

A typical debugging setup during development:

EnsembleOutput output = Ensemble.builder()
    .agents(researcher, analyst, writer)
    .tasks(researchTask, analysisTask, writeTask)
    .chatLanguageModel(model)
    // Full observability stack
    .captureMode(CaptureMode.FULL)
    .traceExporter(TraceExporter.json(Path.of("traces/")))
    .devtools(Devtools.enabled())
    .listener(event -> {
        if (event instanceof TaskCompleteEvent e) {
            log.info("[DONE] {} -- {}ms", e.taskDescription(), e.durationMs());
        }
        if (event instanceof ToolCallEvent e) {
            log.info("[TOOL] {} -> {}", e.toolName(), e.result());
        }
    })
    .costConfiguration(CostConfiguration.builder()
        .inputTokenCostPer1k(0.01)
        .outputTokenCostPer1k(0.03)
        .build())
    .build()
    .run();

// Post-run analysis
EnsembleMetrics metrics = output.getMetrics();
log.info("Total cost: ${}, tokens: {}, duration: {}ms",
    metrics.getTotalCost(), metrics.getTotalTokens(),
    output.getTotalDuration());

For production, dial it back:

EnsembleOutput output = Ensemble.builder()
    .agents(researcher, analyst, writer)
    .tasks(researchTask, analysisTask, writeTask)
    .chatLanguageModel(model)
    // Production observability
    .captureMode(CaptureMode.OFF)
    .traceExporter(TraceExporter.json(Path.of("/var/log/agent-traces/")))
    .meterRegistry(prometheusMeterRegistry)
    .listener(productionEventHandler)
    .costConfiguration(costConfig)
    .build()
    .run();

The observability stack scales from "show me everything" during development to "show me what matters" in production. Same API, different configuration.

The Core Idea

Multi-agent systems are opaque by nature. An LLM call is a black box -- you send a prompt, you get a response, and the reasoning happens inside the model. The only way to make agent systems debuggable is to capture and structure everything around those black box calls: what went in, what came out, how long it took, and how it fits into the broader execution flow.

That's what traces, capture mode, callbacks, and the live dashboard provide. Not transparency into the model, but transparency around it. And in practice, that's enough to debug anything.

Get started:

Documentation -- guides, examples, and API reference
Capture Mode Guide -- deep execution recording
Metrics Guide -- Micrometer integration
Live Dashboard Guide -- real-time execution visualization
Getting Started -- up and running in 5 minutes
GitHub -- source, issues, and contributions

AgentEnsemble is MIT-licensed and available on GitHub.

Human-in-the-Loop Agent Systems in Java

mgd43b — Sat, 28 Mar 2026 14:00:00 +0000

Fully autonomous agents make great demos. In production, someone on your team will eventually ask: "Can a human check this before it goes out?"

The answer should be yes, and it shouldn't require bolting on a custom approval system. Human-in-the-loop isn't a limitation of agent systems -- it's a feature. The best agent architectures make it easy to insert human judgment at exactly the right points, without breaking the execution flow.

This post covers how AgentEnsemble handles human review: the review handler API, review policies, pre-flight validation, and the patterns that make this work in practice.

Why Human-in-the-Loop?

Three reasons show up repeatedly in production agent deployments:

Quality assurance. LLMs produce plausible-sounding output that's sometimes wrong. A human reviewer catches factual errors, hallucinations, and tone problems that automated checks miss.
Compliance. Regulated industries (finance, healthcare, legal) often require human approval before AI-generated content is used in customer-facing contexts. It's not optional.
Calibration. When you deploy a new agent workflow, you want to review the first few outputs to verify the agents are behaving as expected before letting them run autonomously.

The Review Handler

The core abstraction is reviewHandler() -- a function that receives a task's output and returns a Review decision:

Ensemble.builder()
    .agents(researcher, writer)
    .tasks(researchTask, writeTask)
    .chatLanguageModel(model)
    .reviewHandler(taskOutput -> {
        System.out.println("=== REVIEW REQUIRED ===");
        System.out.println("Task: " + taskOutput.getTaskDescription());
        System.out.println("Agent: " + taskOutput.getAgentRole());
        System.out.println("Output:\n" + taskOutput.getRaw());
        System.out.println("========================");
        System.out.print("Decision (approve/reject/edit): ");

        String decision = scanner.nextLine().trim().toLowerCase();

        return switch (decision) {
            case "approve" -> Review.approve();
            case "reject" -> {
                System.out.print("Reason: ");
                yield Review.reject(scanner.nextLine());
            }
            case "edit" -> {
                System.out.print("Enter corrected output: ");
                yield Review.edit(scanner.nextLine());
            }
            default -> Review.approve();
        };
    })
    .build()
    .run();

Three possible outcomes:

Review.approve() -- accept the output as-is. The ensemble continues with the next task.
Review.reject(reason) -- reject the output. The task is re-executed with the rejection reason fed back to the agent as additional context.
Review.edit(correctedOutput) -- replace the output with a human-provided correction. The ensemble continues with the edited version.

The review handler is a plain Java function. It can be a console prompt (as above), a REST call to an approval service, a Slack message that blocks until someone responds, or a database write that pauses the ensemble until a flag is flipped.

Review Policies

Not every task needs human review. Review policies control which tasks trigger the handler:

Ensemble.builder()
    .agents(researcher, writer, editor)
    .tasks(researchTask, writeTask, editTask)
    .chatLanguageModel(model)
    .reviewHandler(this::handleReview)
    .reviewPolicy(ReviewPolicy.REVIEW_ALL)
    .build()
    .run();

Available policies:

Policy	Behavior
`REVIEW_ALL`	Every task output goes through review. Use for high-stakes workflows.
`REVIEW_FAILED`	Only tasks that failed and were retried, or hit max iterations.
`FIRST_TASK_ONLY`	Review the first task output to calibrate. If approved, the rest run without review.

FIRST_TASK_ONLY is particularly useful during the deployment phase. You review the first output to verify the agents are producing what you expect, then let the pipeline run autonomously for the remaining tasks.

Pre-Flight Validation

Sometimes you want an automated quality check before a human sees the output. The beforeReview() hook runs first:

Ensemble.builder()
    .agents(researcher, writer)
    .tasks(researchTask, writeTask)
    .chatLanguageModel(model)
    .beforeReview(taskOutput -> {
        String raw = taskOutput.getRaw();

        // Automated checks
        if (raw == null || raw.isBlank()) {
            return Review.reject("Empty output");
        }
        if (raw.length() < 200) {
            return Review.reject("Output too short -- minimum 200 characters");
        }
        if (raw.contains("I don't know") || raw.contains("I cannot")) {
            return Review.reject("Agent declined the task");
        }

        // Passed automated checks -- proceed to human review
        return Review.skip();
    })
    .reviewHandler(this::humanReview)
    .reviewPolicy(ReviewPolicy.REVIEW_ALL)
    .build()
    .run();

The flow is:

Task completes.
beforeReview() runs automated checks.
- If it returns Review.reject(), the task re-executes. No human is bothered.
- If it returns Review.skip(), the output passes to the human reviewHandler().
- If it returns Review.approve(), the output is accepted without human review.
reviewHandler() presents the output to a human (if beforeReview didn't already decide).

This pattern keeps humans focused on judgment calls, not on catching obvious failures that a simple check could handle.

Rejection and Re-Execution

When a review is rejected -- whether by beforeReview or the human reviewer -- the framework re-executes the task with the rejection reason injected as additional context. The agent sees:

Previous attempt was rejected. Reason: "Output too short -- minimum 200 characters"

This gives the agent a chance to correct its approach. It's not just a retry -- the agent has feedback on what went wrong.

You can limit the number of review cycles to prevent infinite loops:

Task criticalTask = Task.builder()
    .description("Write the executive summary")
    .expectedOutput("A concise, accurate summary")
    .agent(writer)
    .maxOutputRetries(3) // max 3 re-executions after rejection
    .build();

If the output is still rejected after 3 attempts, the task fails with a clear error.

Patterns for Production Review Workflows

Pattern 1: Console Review (Development)

Good for testing and debugging:

.reviewHandler(taskOutput -> {
    System.out.println(taskOutput.getRaw());
    System.out.print("Approve? (y/n): ");
    return scanner.nextLine().equals("y")
        ? Review.approve()
        : Review.reject("Rejected by developer");
})

Pattern 2: REST API Review (Production)

Block the ensemble until an external approval system responds:

.reviewHandler(taskOutput -> {
    // Submit for review
    String reviewId = approvalService.submitForReview(
        taskOutput.getTaskDescription(),
        taskOutput.getRaw()
    );

    // Poll until decision is made
    ReviewDecision decision = approvalService.awaitDecision(reviewId);

    return switch (decision.status()) {
        case APPROVED -> Review.approve();
        case REJECTED -> Review.reject(decision.reason());
        case EDITED -> Review.edit(decision.correctedOutput());
    };
})

Pattern 3: Slack/Teams Notification

Send a message and wait for a reaction:

.reviewHandler(taskOutput -> {
    String messageId = slack.postMessage(
        "#agent-reviews",
        formatForSlack(taskOutput)
    );

    // Block until thumbs-up or thumbs-down reaction
    SlackReaction reaction = slack.awaitReaction(messageId,
        Duration.ofMinutes(30));

    return reaction.isPositive()
        ? Review.approve()
        : Review.reject("Rejected via Slack");
})

Pattern 4: Automated-Only Review

Skip the human entirely -- use beforeReview for automated quality gates:

.beforeReview(taskOutput -> {
    QualityScore score = qualityChecker.evaluate(taskOutput.getRaw());

    if (score.overall() >= 0.8) {
        return Review.approve(); // good enough, no human needed
    } else if (score.overall() >= 0.5) {
        return Review.skip(); // borderline, send to human
    } else {
        return Review.reject("Quality score too low: " + score.overall());
    }
})
.reviewHandler(this::humanReviewForBorderlineCases)

Pattern 5: Tiered Review by Task

Use task-level configuration to vary review intensity:

// Critical task -- always reviewed
Task customerEmail = Task.builder()
    .description("Draft a response to the customer complaint")
    .expectedOutput("Professional, empathetic email response")
    .agent(writer)
    .build();

// Internal task -- skip review
Task internalSummary = Task.builder()
    .description("Summarize the complaint for internal tracking")
    .expectedOutput("Brief internal summary")
    .agent(writer)
    .build();

Ensemble.builder()
    .agents(writer)
    .tasks(customerEmail, internalSummary)
    .chatLanguageModel(model)
    .reviewHandler(taskOutput -> {
        // Only review customer-facing tasks
        if (taskOutput.getTaskDescription().contains("customer")) {
            return humanReview(taskOutput);
        }
        return Review.approve(); // skip internal tasks
    })
    .reviewPolicy(ReviewPolicy.REVIEW_ALL)
    .build()
    .run();

Combining Review with Other Production Features

Review gates compose naturally with other AgentEnsemble features:

Review + Guardrails

Guardrails catch invalid content at the agent level. Review catches quality issues at the workflow level.

Agent writer = Agent.builder()
    .role("Content Writer")
    .goal("Write marketing copy")
    .outputGuardrail(output -> {
        if (containsPII(output)) {
            return GuardrailResult.reject("Output contains PII");
        }
        return GuardrailResult.accept();
    })
    .build();

Ensemble.builder()
    .agents(writer)
    .tasks(writeTask)
    .chatLanguageModel(model)
    .beforeReview(this::automatedQualityCheck)
    .reviewHandler(this::humanReview)
    .build()
    .run();

The execution flow is: Agent runs -> Guardrail validates -> Pre-flight check -> Human review. Each layer catches different classes of problems.

Review + Callbacks

Track review decisions alongside other execution events:

Ensemble.builder()
    .agents(writer)
    .tasks(writeTask)
    .chatLanguageModel(model)
    .reviewHandler(taskOutput -> {
        Review decision = humanReview(taskOutput);
        auditLog.record(taskOutput.getTaskDescription(),
            decision.getType(), decision.getReason());
        return decision;
    })
    .listener(event -> {
        if (event instanceof TaskCompleteEvent e) {
            metrics.recordTaskCompletion(e);
        }
    })
    .build()
    .run();

Review + Structured Output

Review typed output, not raw strings:

record ProposalDraft(
    String title,
    String executiveSummary,
    List<String> keyPoints,
    double estimatedBudget
) {}

Task proposalTask = Task.builder()
    .description("Draft a project proposal")
    .expectedOutput("Structured proposal")
    .agent(writer)
    .outputType(ProposalDraft.class)
    .build();

Ensemble.builder()
    .agents(writer)
    .tasks(proposalTask)
    .chatLanguageModel(model)
    .reviewHandler(taskOutput -> {
        ProposalDraft draft = taskOutput
            .getStructuredOutput(ProposalDraft.class);

        // Review specific fields
        if (draft.estimatedBudget() > 100_000) {
            return Review.reject("Budget exceeds approval threshold");
        }
        if (draft.keyPoints().size() < 3) {
            return Review.reject("Need at least 3 key points");
        }
        return Review.approve();
    })
    .build()
    .run();

The Design Philosophy

Human-in-the-loop isn't an escape hatch for when agents fail. It's a first-class architectural decision. The best agent systems are designed with human review points from the start, not retrofitted when something goes wrong in production.

AgentEnsemble makes this easy by treating review as a builder method, not a separate system. Same API, same execution flow, same observability. A human reviewer is just another step in the pipeline.

Get started:

Documentation -- guides, examples, and API reference
Review Guide -- full API reference for human-in-the-loop
Getting Started -- up and running in 5 minutes
GitHub -- source, issues, and contributions

AgentEnsemble is MIT-licensed and available on GitHub.

MapReduce for AI Agents: Scaling Multi-Agent Workloads on the JVM

mgd43b — Wed, 25 Mar 2026 14:00:00 +0000

Not every agent workload is a pipeline. Sometimes you have a list of items -- documents to summarize, products to review, regions to analyze -- and you need the same kind of analysis applied to each one, then a synthesis of all the results.

That's MapReduce. And it turns out to be one of the most practical patterns for production agent systems.

Most agent frameworks don't have a first-class abstraction for this. You end up writing a loop, creating agents and tasks dynamically, wiring up dependencies by hand, and hoping you got the concurrency right. AgentEnsemble provides MapReduceEnsemble -- a dedicated builder that handles the fan-out, parallel execution, and fan-in with the same typed, observable API as everything else.

The Pattern

MapReduce for agents works exactly like you'd expect:

Map phase: Take a list of items. For each item, create an agent and a task. Run all map tasks in parallel.
Reduce phase: Collect all map results. Feed them to a reduce agent that synthesizes the final output.

The key insight is that map tasks are embarrassingly parallel. Each one is independent. The reduce task depends on all of them.

Static MapReduce

The simplest form: you know the items upfront.

List<String> quarterlyReports = List.of(
    "Q1 2024 Financial Report",
    "Q2 2024 Financial Report",
    "Q3 2024 Financial Report",
    "Q4 2024 Financial Report"
);

MapReduceOutput<String, String> output = MapReduceEnsemble.<String, String>builder()
    .items(quarterlyReports)
    .mapAgentFactory(report -> Agent.builder()
        .role("Financial Analyst")
        .goal("Analyze " + report + " for key trends and anomalies")
        .background("Expert in financial analysis and reporting")
        .build())
    .mapTaskFactory((report, agent) -> Task.builder()
        .description("Analyze " + report + ", identifying revenue trends, "
            + "cost changes, and notable items")
        .expectedOutput("Detailed analysis with key findings")
        .agent(agent)
        .build())
    .reduceAgent(Agent.builder()
        .role("Chief Financial Officer")
        .goal("Synthesize quarterly analyses into an annual review")
        .background("Senior executive with deep financial expertise")
        .build())
    .reduceTaskFactory((results, agent) -> Task.builder()
        .description("Combine all quarterly analyses into a comprehensive "
            + "annual financial review, highlighting year-over-year trends")
        .expectedOutput("Annual financial review report")
        .agent(agent)
        .build())
    .chatLanguageModel(model)
    .build()
    .run();

System.out.println(output.getReduceOutput());

What happens under the hood:

Four agents are created, one per report.
Four map tasks run in parallel.
When all four complete, the reduce agent receives all four outputs as context.
The reduce task produces the final synthesis.

The factory functions (mapAgentFactory, mapTaskFactory) receive the item as input, so you can customize the agent's role, goal, and task description per item.

Typed MapReduce

When you need structured output from the reduce phase, add an outputType:

record AnnualReview(
    String fiscalYear,
    double totalRevenue,
    double totalExpenses,
    List<String> keyTrends,
    String outlook
) {}

MapReduceOutput<String, AnnualReview> output =
    MapReduceEnsemble.<String, AnnualReview>builder()
        .items(quarterlyReports)
        .mapAgentFactory(report -> Agent.builder()
            .role("Financial Analyst")
            .goal("Analyze " + report)
            .build())
        .mapTaskFactory((report, agent) -> Task.builder()
            .description("Analyze " + report)
            .expectedOutput("Detailed financial analysis")
            .agent(agent)
            .build())
        .reduceAgent(Agent.builder()
            .role("CFO")
            .goal("Produce an annual financial review")
            .build())
        .reduceTaskFactory((results, agent) -> Task.builder()
            .description("Synthesize quarterly analyses into an annual review")
            .expectedOutput("Structured annual review")
            .agent(agent)
            .outputType(AnnualReview.class)
            .build())
        .chatLanguageModel(model)
        .build()
        .run();

AnnualReview review = output.getReduceStructuredOutput(AnnualReview.class);
System.out.println(review.totalRevenue());
System.out.println(review.keyTrends());

The type parameter on MapReduceEnsemble.<String, AnnualReview>builder() specifies the input item type and the reduce output type. The framework handles JSON schema generation, LLM instruction, and deserialization.

Adaptive MapReduce

Sometimes you don't know the items upfront. Maybe you have a large document that needs to be partitioned, or you want the LLM to decide how to break up the work.

Adaptive mode lets a planning agent determine the partitioning:

MapReduceOutput<String, String> output = MapReduceEnsemble.<String, String>builder()
    .adaptive(true)
    .sourceDescription("A comprehensive market analysis covering 8 "
        + "industry sectors in the APAC region")
    .plannerAgent(Agent.builder()
        .role("Research Director")
        .goal("Determine the best way to partition the analysis")
        .background("Expert at breaking complex research into manageable pieces")
        .build())
    .mapAgentFactory(partition -> Agent.builder()
        .role("Sector Analyst")
        .goal("Analyze the " + partition + " sector in detail")
        .build())
    .mapTaskFactory((partition, agent) -> Task.builder()
        .description("Conduct a thorough analysis of " + partition)
        .expectedOutput("Sector analysis report")
        .agent(agent)
        .build())
    .reduceAgent(Agent.builder()
        .role("Chief Strategist")
        .goal("Synthesize all sector analyses into a unified report")
        .build())
    .reduceTaskFactory((results, agent) -> Task.builder()
        .description("Create a comprehensive cross-sector APAC market report")
        .expectedOutput("Unified market analysis")
        .agent(agent)
        .build())
    .chatLanguageModel(model)
    .build()
    .run();

In adaptive mode:

The planner agent receives the sourceDescription and decides how to partition the work.
The planner's output is parsed into a list of items.
The map phase proceeds as normal with those items.
The reduce phase synthesizes everything.

This is useful when the decomposition itself requires intelligence -- when you can't hardcode the partitioning logic.

Real-World Use Cases

MapReduce for agents isn't just a pattern exercise. Here are concrete use cases where it shines:

Document Processing Pipeline

List<String> documents = loadDocumentsFromDirectory("contracts/");

MapReduceEnsemble.<String, RiskSummary>builder()
    .items(documents)
    .mapAgentFactory(doc -> Agent.builder()
        .role("Legal Analyst")
        .goal("Identify risks and obligations in " + doc)
        .build())
    .mapTaskFactory((doc, agent) -> Task.builder()
        .description("Review " + doc + " for contractual risks, "
            + "unusual clauses, and compliance concerns")
        .expectedOutput("Risk analysis for " + doc)
        .agent(agent)
        .build())
    .reduceAgent(Agent.builder()
        .role("General Counsel")
        .goal("Produce a portfolio-wide risk assessment")
        .build())
    .reduceTaskFactory((results, agent) -> Task.builder()
        .description("Synthesize all contract analyses into a "
            + "portfolio-wide risk report")
        .expectedOutput("Portfolio risk assessment")
        .agent(agent)
        .outputType(RiskSummary.class)
        .build())
    .chatLanguageModel(model)
    .build()
    .run();

Ten contracts analyzed in parallel, one unified risk summary.

Multi-Region Market Analysis

List<String> regions = List.of("North America", "Europe",
    "Asia Pacific", "Latin America");

MapReduceEnsemble.<String, GlobalReport>builder()
    .items(regions)
    .mapAgentFactory(region -> Agent.builder()
        .role(region + " Market Specialist")
        .goal("Analyze market conditions in " + region)
        .build())
    // ...

Competitive Intelligence

List<String> competitors = List.of("Company A", "Company B",
    "Company C", "Company D", "Company E");

MapReduceEnsemble.<String, CompetitiveLandscape>builder()
    .items(competitors)
    .mapAgentFactory(competitor -> Agent.builder()
        .role("Competitive Intelligence Analyst")
        .goal("Build a detailed profile of " + competitor)
        .build())
    // ...

Code Review

List<String> pullRequests = fetchOpenPRs();

MapReduceEnsemble.<String, ReviewSummary>builder()
    .items(pullRequests)
    .mapAgentFactory(pr -> Agent.builder()
        .role("Code Reviewer")
        .goal("Review " + pr + " for quality and correctness")
        .build())
    // ...

Observability in MapReduce

MapReduceEnsemble supports the same observability stack as regular ensembles:

MapReduceEnsemble.<String, String>builder()
    .items(items)
    // ... agent and task factories ...
    .chatLanguageModel(model)
    .listener(event -> {
        if (event instanceof TaskCompleteEvent e) {
            logger.info("Completed: {} ({}ms)", 
                e.taskDescription(), e.durationMs());
        }
    })
    .traceExporter(TraceExporter.json(Path.of("traces/")))
    .costConfiguration(CostConfiguration.builder()
        .inputTokenCostPer1k(0.01)
        .outputTokenCostPer1k(0.03)
        .build())
    .build()
    .run();

You can see each map task complete individually, track token consumption per item, and get a full trace of the fan-out/fan-in execution.

Error Handling

What happens when one map task fails? By default, the entire MapReduce run fails. But you can configure it to continue:

MapReduceEnsemble.<String, String>builder()
    .items(items)
    // ...
    .errorStrategy(ParallelErrorStrategy.CONTINUE_ON_ERROR)
    .build()
    .run();

With CONTINUE_ON_ERROR, the reduce phase receives results from all map tasks that succeeded. You can check which items failed and handle them separately.

When to Use MapReduce vs. Parallel Workflows

Both run tasks concurrently. The difference is in the structure:

Use parallel workflows when you have a fixed set of heterogeneous tasks with known dependencies:

"Run market analysis AND financial analysis, then produce a SWOT."
Different agents, different task descriptions, explicit dependency wiring.

Use MapReduce when you have a homogeneous operation over a collection:

"Analyze each of these 10 documents the same way, then synthesize."
Same agent template, same task template, applied to different items.

MapReduce is the right abstraction when the word "each" appears in your requirements.

The Bigger Picture

MapReduce for agents follows the same principle as MapReduce for data: decompose, parallelize, aggregate. The difference is that each mapper is an LLM-powered agent, not a stateless function. It can reason, use tools, retry, and produce structured output.

Combined with AgentEnsemble's type safety, observability, and error handling, this gives you a production-grade pattern for scaling agent workloads across collections of items -- without writing concurrency code, without managing thread pools, and without leaving the JVM.

Get started:

Documentation -- guides, examples, and API reference
MapReduce Guide -- full API reference for MapReduce ensembles
Getting Started -- up and running in 5 minutes
GitHub -- source, issues, and contributions

AgentEnsemble is MIT-licensed and available on GitHub.

Agent Workflows on the JVM: Typed, Observable, and Composable

mgd43b — Sun, 22 Mar 2026 14:00:00 +0000

Three properties separate a toy agent framework from one you can actually ship:

Typed -- the compiler helps you, not just the runtime.
Observable -- you can see what happened, why, and how much it cost.
Composable -- the same building blocks produce fundamentally different architectures.

Most agent frameworks in the Python ecosystem nail one of these. Maybe two if you squint. Getting all three usually means stitching together multiple libraries with glue code.

On the JVM, we can do better. This post is a technical deep-dive into how AgentEnsemble achieves all three, and what that means for the architecture of your agent systems.

Typed: The Compiler as a Collaborator

Structured Output with Java Records

The most immediate benefit of type safety in agent systems is structured output. Instead of parsing strings or hoping the LLM returns valid JSON, you declare a Java record and let the framework handle serialization:

record MarketAnalysis(
    String company,
    String sector,
    List<String> competitors,
    double marketCapBillions,
    String outlook
) {}

Task analysisTask = Task.builder()
    .description("Analyze the market position of {{company}}")
    .expectedOutput("A structured market analysis")
    .agent(analyst)
    .outputType(MarketAnalysis.class)
    .build();

When this task runs, the framework:

Derives a JSON schema from the record's structure.
Instructs the LLM to return output conforming to that schema.
Deserializes the response into a MarketAnalysis instance.
Retries automatically if deserialization fails (configurable via maxOutputRetries()).

The result is compile-time type checking on agent output:

MarketAnalysis analysis = output.getTaskOutputs().get(0)
    .getStructuredOutput(MarketAnalysis.class);

// These are real typed fields, not dictionary lookups
String sector = analysis.sector();
List<String> competitors = analysis.competitors();
double marketCap = analysis.marketCapBillions();

No (String) map.get("sector"). No JSONObject.getString(). No runtime ClassCastException at 2 AM.

Nested Types

Records can be nested, and the framework handles it:

record CompetitorDetail(String name, double marketShare, String strategy) {}

record IndustryReport(
    String industry,
    List<CompetitorDetail> competitors,
    String trendSummary
) {}

Task reportTask = Task.builder()
    .description("Create an industry report for {{industry}}")
    .expectedOutput("Structured industry report")
    .agent(analyst)
    .outputType(IndustryReport.class)
    .build();

The LLM receives a schema that includes the nested CompetitorDetail structure, and the entire object graph is deserialized in one pass.

Immutable Domain Objects

Beyond output typing, the framework's own domain model is fully immutable. Agent, Task, Ensemble, TaskOutput -- all built with the builder pattern, all immutable after construction:

Agent agent = Agent.builder()
    .role("Analyst")
    .goal("Analyze data")
    .maxIterations(15)
    .build();

// Need a variant? Use toBuilder()
Agent verboseAgent = agent.toBuilder()
    .verbose(true)
    .build();

No mutable state to worry about across threads. No defensive copies. The builder gives you construction-time flexibility; immutability gives you runtime safety.

Observable: See Everything

Observability in agent systems isn't a single feature -- it's a stack. AgentEnsemble provides four layers, each useful at a different stage of development and operation.

Layer 1: Event Callbacks (Development and Real-Time)

The callback system fires typed events at every significant point in execution:

Ensemble.builder()
    // ...
    .listener(event -> {
        switch (event) {
            case TaskStartEvent e ->
                log.info("[START] {}", e.taskDescription());
            case TaskCompleteEvent e ->
                log.info("[DONE] {} -- {}ms, {} tokens",
                    e.taskDescription(), e.durationMs(), e.tokenCount());
            case ToolCallEvent e ->
                log.info("[TOOL] {} called by {}",
                    e.toolName(), e.agentRole());
            case DelegationStartedEvent e ->
                log.info("[DELEGATE] {} -> {}",
                    e.fromAgent(), e.toAgent());
            case TokenEvent e ->
                // streaming token-by-token output
                System.out.print(e.token());
            default -> {}
        }
    })
    .build()
    .run();

Events are sealed types. The compiler tells you what's available. Multiple listeners are supported. Listener exceptions are caught and logged -- they never break the ensemble.

Layer 2: Metrics (Production Monitoring)

The Micrometer integration publishes standard metrics to whatever backend you already use -- Prometheus, Datadog, CloudWatch:

Ensemble.builder()
    // ...
    .meterRegistry(registry)
    .costConfiguration(CostConfiguration.builder()
        .inputTokenCostPer1k(0.01)
        .outputTokenCostPer1k(0.03)
        .build())
    .build()
    .run();

Published metrics include:

Metric	Type	Tags
`ensemble.task.duration`	Timer	agent, task
`ensemble.task.tokens`	Counter	agent, task, direction (input/output)
`ensemble.task.completions`	Counter	agent, task, status
`ensemble.tool.calls`	Counter	tool, agent
`ensemble.run.cost`	Gauge	--
`ensemble.run.duration`	Timer	workflow

These flow straight into your existing Grafana dashboards. No new monitoring infrastructure required.

Layer 3: Structured Traces (Post-Mortem Analysis)

For detailed post-mortem analysis, export full execution traces:

Ensemble.builder()
    // ...
    .traceExporter(TraceExporter.json(Path.of("traces/")))
    .build()
    .run();

Each trace is a tree of spans:

Ensemble Run
  +-- Task: Market Research (3,240ms, 1,847 tokens)
  |     +-- LLM Call #1 (2,100ms)
  |     +-- Tool: WebSearch (890ms)
  |     +-- LLM Call #2 (250ms)
  +-- Task: Write Report (2,100ms, 1,203 tokens)
        +-- LLM Call #1 (2,100ms)

Spans include duration, token counts, input/output payloads, and tool call details. The JSON format is structured for programmatic analysis -- feed it into your log aggregation pipeline or write custom analysis scripts.

Layer 4: Capture Mode (Deep Debugging)

When you need to see everything -- the full prompt sent to the LLM, the raw response, every tool call input and output:

Ensemble.builder()
    // ...
    .captureMode(CaptureMode.FULL)
    .build()
    .run();

Three levels:

OFF -- standard execution, no extra data collection.
STANDARD -- adds full LLM message history per iteration and memory operations.
FULL -- adds tool call input/output payloads, raw LLM responses, and detailed timing.

FULL mode is for development and debugging. STANDARD is for staging. OFF is for production (unless you're investigating an issue).

Layer 5: Live Dashboard (Development)

During development, launch a live browser dashboard that shows execution progress in real time:

Ensemble.builder()
    // ...
    .devtools(Devtools.enabled())
    .build()
    .run();

The dashboard shows a DAG visualization of tasks, real-time progress indicators, agent activity, and token consumption. It's a development tool, not a production dashboard -- but it makes building and debugging agent workflows dramatically faster.

Composable: Same Blocks, Different Buildings

The real test of a framework's composability is whether fundamentally different architectures emerge from the same primitives.

The Primitives

AgentEnsemble has three core primitives:

Agent -- an AI entity with a role, goal, and optional tools.
Task -- a unit of work with a description, expected output, and optional dependencies.
Ensemble -- a group of agents executing a set of tasks.

That's it. Everything else is a composition of these three.

Sequential Pipelines

Tasks with linear dependencies:

Task research = Task.builder()/* ... */.build();
Task write = Task.builder()/* ... */.context(List.of(research)).build();
Task edit = Task.builder()/* ... */.context(List.of(write)).build();

The framework sees a chain and executes sequentially.

Parallel DAGs

Tasks with branching dependencies:

Task taskA = Task.builder()/* ... */.build();
Task taskB = Task.builder()/* ... */.build();
Task taskC = Task.builder()/* ... */
    .context(List.of(taskA, taskB)).build();

The framework sees A and B are independent, runs them concurrently, then runs C when both complete.

Hierarchical Delegation

Tasks without assigned agents, plus a manager model:

Ensemble.builder()
    .agents(worker1, worker2, worker3)
    .tasks(unassignedTask)
    .workflow(Workflow.HIERARCHICAL)
    .chatLanguageModel(managerModel)
    .build()
    .run();

The manager agent delegates to workers. The same agent and task primitives, different execution semantics.

MapReduce

For data-parallel workloads, the MapReduceEnsemble composes the same primitives into a fan-out/fan-in pattern:

MapReduceEnsemble.<String, String>builder()
    .items(List.of("Q1 Report", "Q2 Report", "Q3 Report", "Q4 Report"))
    .mapAgentFactory(item -> Agent.builder()
        .role("Analyst for " + item)
        .goal("Analyze " + item)
        .build())
    .mapTaskFactory((item, agent) -> Task.builder()
        .description("Analyze " + item)
        .expectedOutput("Analysis of " + item)
        .agent(agent)
        .build())
    .reduceAgent(Agent.builder()
        .role("Senior Analyst")
        .goal("Synthesize all quarterly analyses")
        .build())
    .reduceTaskFactory((results, agent) -> Task.builder()
        .description("Combine all quarterly analyses into an annual summary")
        .expectedOutput("Annual summary report")
        .agent(agent)
        .build())
    .chatLanguageModel(model)
    .build()
    .run();

Each item gets its own agent and task. Map tasks run in parallel. The reduce task collects all results. The same Agent.builder() and Task.builder() APIs, composed into a completely different execution pattern.

Dynamic Ensembles

Agents and tasks can be created at runtime based on input data:

List<String> topics = List.of("AI", "Quantum Computing", "Biotech");

List<Agent> agents = new ArrayList<>();
List<Task> tasks = new ArrayList<>();

for (String topic : topics) {
    Agent specialist = Agent.builder()
        .role(topic + " Specialist")
        .goal("Provide expert analysis of " + topic)
        .build();
    agents.add(specialist);

    tasks.add(Task.builder()
        .description("Analyze recent developments in " + topic)
        .expectedOutput("Expert analysis")
        .agent(specialist)
        .build());
}

// Add a synthesis task that depends on all specialist tasks
Agent synthesizer = Agent.builder()
    .role("Chief Analyst")
    .goal("Synthesize all specialist analyses")
    .build();
agents.add(synthesizer);

tasks.add(Task.builder()
    .description("Create a unified report from all analyses")
    .expectedOutput("Comprehensive cross-domain report")
    .agent(synthesizer)
    .context(tasks) // depends on all previous tasks
    .build());

Ensemble.builder()
    .agents(agents)
    .tasks(tasks)
    .chatLanguageModel(model)
    .build()
    .run();

The number of agents and tasks is determined by the input data. The dependency graph is constructed dynamically. The framework handles the rest.

The Architecture Insight

What makes all of this work is a simple design decision: context() declarations are the source of truth for execution order.

When you call .context(List.of(taskA, taskB)) on a task, you're declaring a dependency edge in a directed acyclic graph. The framework builds this graph at ensemble construction time, performs a topological sort, and schedules execution accordingly.

This means:

Workflow inference works because the graph already encodes the execution strategy.
Parallel scheduling works because independent subgraphs can be identified and executed concurrently.
Task output passing works because the framework knows which outputs to inject as context for downstream tasks.
Error propagation works because the graph defines the blast radius of a failure.

The graph is the architecture. Everything else follows from it.

Bringing It Together

Agent orchestration on the JVM doesn't have to mean trading away the properties that make Java effective for production systems. Type safety, observability, and composability aren't luxuries -- they're the foundations that let you ship with confidence.

AgentEnsemble gives you all three from the same set of builders, the same primitives, the same dependency model. Whether you're building a two-agent pipeline or a dynamic MapReduce ensemble, the API is consistent and the properties hold.

Get started:

Documentation -- guides, examples, and API reference
Getting Started -- up and running in 5 minutes
Examples -- runnable code for every pattern
GitHub -- source, issues, and contributions

AgentEnsemble is MIT-licensed and available on GitHub.

Production-Minded Multi-Agent Orchestration in Java

mgd43b — Thu, 19 Mar 2026 14:00:00 +0000

The demo works. Your two-agent research-writer pipeline produces a decent article. Your hierarchical team generates a plausible report. Someone on the team says "let's ship it."

And then reality sets in.

How many tokens did that run consume? What happens when the LLM returns garbage JSON? Can we rate-limit calls so we don't blow our API budget on day one? What if a task takes 90 seconds and produces hallucinated nonsense -- does a human get to review it before it hits the customer?

These aren't edge cases. They're the difference between a demo and a deployment. This post covers what production multi-agent orchestration actually requires, and how AgentEnsemble handles each concern.

Observability: Know What Your Agents Are Doing

Event Callbacks

Every significant event in an ensemble run fires a callback. You register listeners on the builder:

Ensemble.builder()
    .agents(researcher, writer)
    .tasks(researchTask, writeTask)
    .chatLanguageModel(model)
    .listener(event -> {
        switch (event) {
            case TaskStartEvent e ->
                logger.info("Starting: {}", e.taskDescription());
            case TaskCompleteEvent e ->
                logger.info("Completed: {} ({}ms, {} tokens)",
                    e.taskDescription(), e.durationMs(), e.tokenCount());
            case TaskFailedEvent e ->
                logger.error("Failed: {} - {}", e.taskDescription(),
                    e.errorMessage());
            case ToolCallEvent e ->
                logger.info("Tool call: {} -> {}", e.toolName(),
                    e.result());
            default -> {} // other events
        }
    })
    .build()
    .run();

Events are typed. You pattern-match on them. No string parsing, no event name constants.

You can register multiple listeners, and they're exception-safe -- if one listener throws, the others still fire and the ensemble keeps running.

For more structured collection, implement the EnsembleListener interface:

public class MetricsCollector implements EnsembleListener {
    private final Map<String, Long> taskDurations = new ConcurrentHashMap<>();

    @Override
    public void onTaskComplete(TaskCompleteEvent event) {
        taskDurations.put(event.taskDescription(), event.durationMs());
    }

    @Override
    public void onToolCall(ToolCallEvent event) {
        meterRegistry.counter("agent.tool.calls",
            "tool", event.toolName()).increment();
    }
}

Micrometer Metrics

AgentEnsemble integrates with Micrometer out of the box. Add the metrics module:

implementation("net.agentensemble:agentensemble-metrics-micrometer:2.3.0")

Then wire it up:

MeterRegistry registry = new PrometheusMeterRegistry(
    PrometheusConfig.DEFAULT);

Ensemble.builder()
    .agents(researcher, writer)
    .tasks(researchTask, writeTask)
    .chatLanguageModel(model)
    .meterRegistry(registry)
    .build()
    .run();

You get counters for token usage, task completions, task failures, and tool calls. Gauges for active tasks. Timers for task and ensemble duration. All tagged with agent role and task description for Grafana/Prometheus filtering.

Structured Traces

For post-mortem analysis, export full execution traces as JSON:

EnsembleOutput output = Ensemble.builder()
    .agents(researcher, writer)
    .tasks(researchTask, writeTask)
    .chatLanguageModel(model)
    .traceExporter(TraceExporter.json(Path.of("traces/")))
    .build()
    .run();

// Or access the trace programmatically
ExecutionTrace trace = output.getTrace();
trace.getSpans().forEach(span ->
    System.out.printf("%s: %dms, %d tokens%n",
        span.getName(), span.getDurationMs(), span.getTokenCount()));

Each trace includes spans for every task, tool call, and LLM interaction. Duration, token counts, input/output payloads -- all structured, all queryable.

Cost Tracking

Know what you're spending:

Ensemble.builder()
    // ...
    .costConfiguration(CostConfiguration.builder()
        .inputTokenCostPer1k(0.01)
        .outputTokenCostPer1k(0.03)
        .build())
    .build()
    .run();

EnsembleMetrics metrics = output.getMetrics();
System.out.printf("Total cost: $%.4f%n", metrics.getTotalCost());
System.out.printf("Input tokens: %d, Output tokens: %d%n",
    metrics.getInputTokens(), metrics.getOutputTokens());

You configure per-model pricing, and the framework tracks token consumption and computes costs across all tasks in the run.

Error Handling: Fail Gracefully

Max Iterations

Agents can get stuck in loops -- calling the same tool repeatedly, or generating output that fails validation. Cap it:

Agent researcher = Agent.builder()
    .role("Researcher")
    .goal("Find information about {{topic}}")
    .maxIterations(10) // default is 25
    .build();

When the limit is hit, the agent returns its best output so far rather than running forever.

Output Retries

When a task has outputType() set and the LLM returns invalid JSON, the framework retries automatically:

Task profileTask = Task.builder()
    .description("Create a competitor profile")
    .expectedOutput("Structured JSON profile")
    .agent(analyst)
    .outputType(CompetitorProfile.class)
    .maxOutputRetries(3) // retry up to 3 times on parse failure
    .build();

Parallel Error Strategy

In parallel workflows, one failing task shouldn't necessarily kill the entire ensemble:

Ensemble.builder()
    .agents(marketAnalyst, financialAnalyst, strategist)
    .tasks(marketTask, financialTask, summaryTask)
    .chatLanguageModel(model)
    .workflow(Workflow.parallel()
        .errorStrategy(ParallelErrorStrategy.CONTINUE_ON_ERROR)
        .build())
    .build()
    .run();

With CONTINUE_ON_ERROR, successful tasks still complete, and downstream tasks receive whatever results are available. The ensemble output tells you which tasks succeeded and which failed.

Guardrails

Validate inputs and outputs at the framework level:

Agent agent = Agent.builder()
    .role("Content Writer")
    .goal("Write marketing content")
    .inputGuardrail(input -> {
        if (input.contains("competitor")) {
            return GuardrailResult.reject(
                "Input must not reference competitors");
        }
        return GuardrailResult.accept();
    })
    .outputGuardrail(output -> {
        if (output.length() > 5000) {
            return GuardrailResult.reject("Output exceeds length limit");
        }
        return GuardrailResult.accept();
    })
    .build();

Guardrails run before/after each agent iteration. Rejection stops the agent and surfaces a clear error, not a silent failure.

Cost Control: Don't Blow Your Budget

Rate Limiting

Protect against runaway API costs:

Ensemble.builder()
    .agents(researcher, writer)
    .tasks(researchTask, writeTask)
    .chatLanguageModel(model)
    .rateLimit(RateLimit.builder()
        .maxRequestsPerMinute(60)
        .build())
    .build()
    .run();

The rate limiter applies across all agents in the ensemble. Requests that exceed the limit are queued, not dropped.

Delegation Depth

In hierarchical workflows, prevent infinite delegation chains:

Ensemble.builder()
    // ...
    .workflow(Workflow.HIERARCHICAL)
    .maxDelegationDepth(3)
    .build()
    .run();

Human-in-the-Loop: Review Before You Ship

The most production-critical feature is often the simplest: letting a human review agent output before it's treated as final.

Ensemble.builder()
    .agents(researcher, writer)
    .tasks(researchTask, writeTask)
    .chatLanguageModel(model)
    .reviewHandler(taskOutput -> {
        System.out.println("Agent produced:\n" + taskOutput.getRaw());
        System.out.print("Approve? (y/n/edit): ");

        String response = scanner.nextLine();
        if (response.equals("y")) {
            return Review.approve();
        } else if (response.equals("n")) {
            return Review.reject("Quality below threshold");
        } else {
            System.out.print("Enter corrected output: ");
            return Review.edit(scanner.nextLine());
        }
    })
    .reviewPolicy(ReviewPolicy.REVIEW_ALL)
    .build()
    .run();

Review policies let you control which tasks trigger review:

REVIEW_ALL -- every task output goes through review.
REVIEW_FAILED -- only tasks that hit errors or retries.
FIRST_TASK_ONLY -- review the first output to calibrate, then let the rest run.

You can also add pre-flight validation with beforeReview():

Ensemble.builder()
    // ...
    .beforeReview(taskOutput -> {
        // Automated quality check before human review
        if (taskOutput.getRaw().length() < 100) {
            return Review.reject("Output too short, re-running");
        }
        return Review.skip(); // passes automated check, proceed to human
    })
    .build()
    .run();

Testing: Capture and Replay

Production confidence starts with testability. AgentEnsemble's capture mode records full execution data:

EnsembleOutput output = Ensemble.builder()
    .agents(researcher, writer)
    .tasks(researchTask, writeTask)
    .chatLanguageModel(model)
    .captureMode(CaptureMode.FULL) // record everything
    .traceExporter(TraceExporter.json(Path.of("test-traces/")))
    .build()
    .run();

CaptureMode.FULL records:

Full LLM message history per iteration (system prompt, user messages, assistant responses)
Tool call inputs and outputs
Token counts per interaction
Memory operations
Timing data

This gives you deterministic test fixtures. Capture a run once, then write assertions against the trace without making live LLM calls.

The Production Checklist

Here's a quick reference for what to enable before shipping:

Concern	Configuration
Logging	SLF4J is used throughout; configure your logging framework
Event callbacks	`.listener()` for real-time monitoring
Metrics	`.meterRegistry()` with Micrometer
Traces	`.traceExporter()` for structured JSON
Cost tracking	`.costConfiguration()` with your model's pricing
Rate limiting	`.rateLimit()` to protect API budgets
Error strategy	`ParallelErrorStrategy.CONTINUE_ON_ERROR` for resilience
Review gates	`.reviewHandler()` + `.reviewPolicy()` for human oversight
Guardrails	`.inputGuardrail()` / `.outputGuardrail()` on agents
Testing	`.captureMode(CaptureMode.FULL)` for trace capture

None of these require custom infrastructure. They're all builder methods on the same API you already use to define agents and tasks.

Production-grade agent orchestration isn't about adding a monitoring sidecar after the fact. It's about choosing a framework that was built for production from the start.

Get started:

Documentation -- guides, examples, and API reference
Getting Started -- up and running in 5 minutes
Examples -- runnable code for every pattern
GitHub -- source, issues, and contributions

AgentEnsemble is MIT-licensed and available on GitHub.

From CrewAI to Java: Multi-Agent Orchestration Without Python

mgd43b — Mon, 16 Mar 2026 14:00:00 +0000

You've seen the CrewAI demos. Maybe you've prototyped something with AutoGen or LangGraph. The concepts make sense -- agents with roles, tasks with dependencies, tools for grounding -- but there's a problem.

Your production stack is Java.

Your team writes Java. Your CI runs Gradle. Your monitoring is Micrometer and Prometheus. Your deployment is a fat JAR on Kubernetes. And now someone's proposing a Python sidecar for the AI agent layer, with a REST API in between, and a second dependency tree, and a second set of runtime semantics.

There's a better path. AgentEnsemble is a Java 21 framework that covers the same concepts as CrewAI -- agents, tasks, crews/ensembles, tools, workflows -- but runs natively on the JVM. Here's a concept-by-concept mapping.

The Concept Map

CrewAI (Python)	AgentEnsemble (Java)
`Agent`	`Agent`
`Task`	`Task`
`Crew`	`Ensemble`
`Tool`	`Tool` (via `@Tool` annotation or `AgentTool` interface)
`Process.sequential`	`Workflow.SEQUENTIAL` (or inferred)
`Process.hierarchical`	`Workflow.HIERARCHICAL`
--	`Workflow.PARALLEL` (DAG-based)
--	`MapReduceEnsemble`
`@tool` decorator	`@Tool` annotation on methods
Pydantic model output	Java record `outputType()`
Callbacks	`EnsembleListener` / lambda callbacks
Memory	`MemoryStore` (in-memory, cross-run)

The core mental model is the same. The execution model, type safety, and production tooling are where they diverge.

Side by Side: A Research-Writer Pipeline

CrewAI (Python)

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Senior Researcher",
    goal="Find comprehensive information about {topic}",
    backstory="Expert at finding and synthesizing information",
)

writer = Agent(
    role="Technical Writer",
    goal="Write clear, engaging content",
    backstory="Skilled at making complex topics accessible",
)

research_task = Task(
    description="Research {topic} thoroughly",
    expected_output="Comprehensive research notes",
    agent=researcher,
)

write_task = Task(
    description="Write an article based on the research",
    expected_output="A polished article",
    agent=writer,
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
)

result = crew.kickoff(inputs={"topic": "quantum computing"})
print(result)

AgentEnsemble (Java)

Agent researcher = Agent.builder()
    .role("Senior Researcher")
    .goal("Find comprehensive information about {{topic}}")
    .background("Expert at finding and synthesizing information")
    .build();

Agent writer = Agent.builder()
    .role("Technical Writer")
    .goal("Write clear, engaging content")
    .background("Skilled at making complex topics accessible")
    .build();

Task researchTask = Task.builder()
    .description("Research {{topic}} thoroughly")
    .expectedOutput("Comprehensive research notes")
    .agent(researcher)
    .build();

Task writeTask = Task.builder()
    .description("Write an article based on the research")
    .expectedOutput("A polished article")
    .agent(writer)
    .context(List.of(researchTask))
    .build();

EnsembleOutput output = Ensemble.builder()
    .agents(researcher, writer)
    .tasks(researchTask, writeTask)
    .chatLanguageModel(model)
    .inputs(Map.of("topic", "quantum computing"))
    .build()
    .run();

System.out.println(output.getRaw());

Nearly identical structure. The Java version uses builders instead of constructors, {{topic}} instead of {topic}, and makes dependencies explicit via context() rather than relying on task ordering.

What You Gain by Staying on the JVM

The mapping above shows conceptual parity. Here's what you gain beyond that.

1. Compile-Time Type Safety

In CrewAI, structured output uses Pydantic models:

from pydantic import BaseModel

class MovieReview(BaseModel):
    title: str
    rating: int
    summary: str

task = Task(
    description="Review the movie",
    expected_output="A movie review",
    output_pydantic=MovieReview,
)

In AgentEnsemble, it's a Java record:

record MovieReview(String title, int rating, String summary) {}

Task task = Task.builder()
    .description("Review the movie '{{movie}}'")
    .expectedOutput("A movie review")
    .outputType(MovieReview.class)
    .build();

// Later:
MovieReview review = output.getTaskOutputs().get(0)
    .getStructuredOutput(MovieReview.class);

Both work. But the Java version catches type mismatches at compile time. If you refactor MovieReview to rename a field, every access site that uses the old name is a compilation error, not a runtime AttributeError.

2. Parallel DAG Workflows

CrewAI supports sequential and hierarchical processes. AgentEnsemble adds parallel DAG execution:

// These run concurrently
Task marketResearch = Task.builder()
    .description("Analyze market trends")
    .agent(marketAnalyst).build();

Task financialAnalysis = Task.builder()
    .description("Analyze financials")
    .agent(financialAnalyst).build();

// This waits for both to finish
Task synthesis = Task.builder()
    .description("Synthesize market and financial findings")
    .agent(strategist)
    .context(List.of(marketResearch, financialAnalysis))
    .build();

No explicit workflow declaration needed. The framework infers parallel execution from the dependency graph. Independent tasks run concurrently on virtual threads; dependent tasks wait for their inputs.

3. Workflow Inference

In CrewAI, you always specify Process.sequential or Process.hierarchical. In AgentEnsemble, you can omit the workflow entirely:

Ensemble.builder()
    .agents(researcher, analyst, writer)
    .tasks(researchTask, analysisTask, reportTask)
    .chatLanguageModel(model)
    .build()
    .run();

The framework examines context() declarations on each task and infers:

All tasks in a linear chain? Sequential.
Tasks with branching/merging dependencies? Parallel DAG.
Single task with multiple agents and no assignments? Hierarchical.

You can still declare a workflow explicitly when you want to be specific.

4. MapReduce Ensembles

For workloads that need to process a collection of items in parallel and aggregate the results, AgentEnsemble provides a dedicated MapReduceEnsemble:

MapReduceEnsemble.<String, String>builder()
    .items(List.of("Chapter 1", "Chapter 2", "Chapter 3"))
    .mapAgentFactory(chapter -> Agent.builder()
        .role("Editor for " + chapter)
        .goal("Edit " + chapter + " for clarity and style")
        .build())
    .mapTaskFactory((chapter, agent) -> Task.builder()
        .description("Edit " + chapter)
        .expectedOutput("Edited " + chapter)
        .agent(agent)
        .build())
    .reduceAgent(Agent.builder()
        .role("Senior Editor")
        .goal("Ensure consistency across all chapters")
        .build())
    .reduceTaskFactory((results, agent) -> Task.builder()
        .description("Review all edited chapters for consistency")
        .expectedOutput("Final editorial notes")
        .agent(agent)
        .build())
    .chatLanguageModel(model)
    .build()
    .run();

There's also an adaptive mode where an LLM decides how to partition the work at runtime.

5. Production Tooling Built In

This is the biggest difference. CrewAI gives you agents and tasks. AgentEnsemble gives you agents, tasks, and the production infrastructure:

Feature	CrewAI	AgentEnsemble
Rate limiting	External	`.rateLimit()` builder method
Cost tracking	External	`.costConfiguration()` builder method
Micrometer metrics	N/A	`.meterRegistry()` builder method
Structured traces	External	`.traceExporter()` builder method
Human review gates	External	`.reviewHandler()` + `.reviewPolicy()`
Input/output guardrails	External	`.inputGuardrail()` / `.outputGuardrail()` on agents
Capture mode for testing	External	`.captureMode(CaptureMode.FULL)`
Live development dashboard	External	`.devtools(Devtools.enabled())`
Parallel error strategies	N/A	`ParallelErrorStrategy.CONTINUE_ON_ERROR`

None of these require additional libraries or custom code. They're all builder methods on the same API.

6. Native JVM Deployment

No Python runtime, no virtual environment, no pip dependencies, no requirements.txt, no separate container image. Your agent code is just Java code that ships in the same JAR as the rest of your application:

// build.gradle.kts
dependencies {
    implementation("net.agentensemble:agentensemble-core:2.3.0")
}

It works with your existing:

Gradle or Maven build
JUnit test suite
Spring Boot application (framework integration module available)
Docker image
Kubernetes deployment
CI/CD pipeline

No polyglot overhead.

What You Give Up

To be fair, the Python ecosystem has advantages:

Larger community: More tutorials, Stack Overflow answers, and blog posts for Python agent frameworks.
Faster prototyping: Python's dynamic typing can be faster for throwaway experiments.
Broader LLM library support: Some cutting-edge LLM features land in Python first.

AgentEnsemble mitigates the third point by building on LangChain4j, which supports OpenAI, Anthropic, Google, Ollama, Azure OpenAI, Amazon Bedrock, and many others. But there may be niche providers or features that aren't available yet.

Making the Decision

If your team is already writing Python, CrewAI and friends are fine choices. Use what fits your stack.

But if your production backend is Java, adding a Python layer for agent orchestration introduces:

A second language to maintain
A second dependency tree to audit
A second runtime to monitor
A REST boundary with serialization overhead
A deployment topology change

AgentEnsemble eliminates all of that. Same language, same build, same runtime, same monitoring. Your agent system is just another module in your existing application.

Get started:

Documentation -- guides, examples, and API reference
Getting Started -- up and running in 5 minutes
Migration Guide -- transitioning from Python frameworks
Examples -- runnable code for every pattern
GitHub -- source, issues, and contributions

AgentEnsemble is MIT-licensed and available on GitHub.