Add tool actions compensation at agentic system level (#5823)
<!-- Thank you so much for your contribution! Please fill in all the sections below. Please open the PR as ready for review (not as a draft), with tests and documentation already included. Please note that PRs with breaking changes, or without tests and documentation, will be rejected. Please note that PRs will be reviewed based on the priority of the issues they address. We ask for your patience. We are doing our best to review your PR as quickly as possible. Please refrain from pinging and asking when it will be reviewed. Thank you for understanding! --> ## Issue <!-- Please specify the ID of the issue this PR is addressing. For example: "Closes #1234" or "Fixes #1234" --> Closes # ## Change When `compensateOnError(true)` is set on an agentic system, all previously successful tool invocations with `@CompensateFor` actions are compensated in reverse order if any tool in any sub-agent fails or any agent throws. This work is built on top of what has been done here https://github.com/langchain4j/langchain4j/pull/5171 ## General checklist <!-- Please double-check the following points and mark them like this: [X] --> - [X] There are no breaking changes (API, behaviour) - [X] I have added unit and/or integration tests for my change - [X] The tests cover both positive and negative cases - [X] I have manually run all the unit and integration tests in the module I have added/changed, and they are all green - [ ] I have manually run all the unit and integration tests in the [core](https://github.com/langchain4j/langchain4j/tree/main/langchain4j-core) and [main](https://github.com/langchain4j/langchain4j/tree/main/langchain4j) modules, and they are all green - [ ] I have added/updated the [documentation](https://github.com/langchain4j/langchain4j/tree/main/docs/docs) - [ ] I have added an example in the [examples repo](https://github.com/langchain4j/langchain4j-examples) (only for "big" features) - [ ] I have added/updated [Spring Boot starter(s)](https://github.com/langchain4j/langchain4j-spring) (if applicable) ## Checklist for adding new maven module <!-- Please double-check the following points and mark them like this: [X] --> - [ ] I have added my new module in the root `pom.xml` and `langchain4j-bom/pom.xml` ## Checklist for adding new embedding store integration <!-- Please double-check the following points and mark them like this: [X] --> - [ ] I have added a `{NameOfIntegration}EmbeddingStoreIT` that extends from either `EmbeddingStoreIT` or `EmbeddingStoreWithFilteringIT` - [ ] I have added a `{NameOfIntegration}EmbeddingStoreRemovalIT` that extends from `EmbeddingStoreWithRemovalIT` ## Checklist for changing existing embedding store integration <!-- Please double-check the following points and mark them like this: [X] --> - [ ] I have manually verified that the `{NameOfIntegration}EmbeddingStore` works correctly with the data persisted using the latest released version of LangChain4j
This commit is contained in:
Mario Fusco
GitHub
parent
e66f15cab0
commit
0b701c0557
|
|
@ -750,6 +750,46 @@ UntypedAgent novelCreator = AgenticServices.sequenceBuilder()
|
|||
.build();
|
||||
```
|
||||
|
||||
## Cross-agent compensation
|
||||
|
||||
When an agentic system performs side effects through tools (e.g., database writes, API calls, financial transactions), a failure partway through the workflow can leave the system in an inconsistent state. Cross-agent compensation tries to solve, or at least mitigate, this problem: if any agent in the hierarchy fails, all previously successful tool invocations with `@CompensateFor` actions are compensated in reverse order.
|
||||
|
||||
This builds on the per-agent `@CompensateFor` mechanism (see [Tools](/tutorials/tools#compensating-tool-actions)). While per-agent compensation handles tool errors within a single agent, cross-agent compensation handles agent-level failures across an entire hierarchy.
|
||||
|
||||
In order to enable this feature set `compensateOnError(true)` on the composed agent builder:
|
||||
|
||||
```java
|
||||
UntypedAgent transferWorkflow = AgenticServices.sequenceBuilder()
|
||||
.subAgents(creditAgent, debitAgent, notificationAgent)
|
||||
.compensateOnError(true)
|
||||
.outputKey("result")
|
||||
.build();
|
||||
```
|
||||
|
||||
If `notificationAgent` throws an exception, the tools invoked by `creditAgent` and `debitAgent` that have `@CompensateFor` methods will be compensated in reverse chronological order (last executed first).
|
||||
|
||||
Tools without a `@CompensateFor` annotation are simply skipped during compensation.
|
||||
|
||||
Compensating actions are defined on tool classes using `@CompensateFor`, the same annotation used for per-agent tool compensation:
|
||||
|
||||
```java
|
||||
public class AccountService {
|
||||
|
||||
@Tool("Credits the given amount to the account")
|
||||
String credit(@P(name = "amount") int amount) {
|
||||
// perform the credit
|
||||
return "credited " + amount;
|
||||
}
|
||||
|
||||
@CompensateFor("credit")
|
||||
void reverseCredit(int amount) {
|
||||
// reverse the credit
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Note that the compensating actions are executed with a best-effort policy: if one of them fails, the error is logged and the remaining compensations continue.
|
||||
|
||||
## Observability
|
||||
|
||||
Tracking and logging the agents' invocations can be crucial for debugging and understanding the aggregate behavior of the whole agentic system in which those agents participate. For this reason, the `langchain4j-agentic` module allows you to register an `AgentListener` through the `listener` method of the agent builders, that is notified of all agents invocations and their results, and it is defined as follows:
|
||||
|
|
|
|||
|
|
@ -147,6 +147,18 @@
|
|||
"old": "class dev.langchain4j.agentic.scope.ResultWithAgenticScope<T>",
|
||||
"new": "class dev.langchain4j.agentic.scope.ResultWithAgenticScope<T>",
|
||||
"justification": "Consequence of converting ResultWithAgenticScope from a record (which implicitly extends java.lang.Record) to a regular final class in order to carry non-component suspension state. Public constructor and accessors are preserved."
|
||||
},
|
||||
{
|
||||
"ignore": true,
|
||||
"code": "java.method.addedToInterface",
|
||||
"new": "method T dev.langchain4j.agentic.planner.AgenticService<T, A>::compensateOnError(boolean)",
|
||||
"justification": "Add tool actions compensation at agentic system level"
|
||||
},
|
||||
{
|
||||
"ignore": true,
|
||||
"code": "java.method.addedToInterface",
|
||||
"new": "method dev.langchain4j.agentic.supervisor.SupervisorAgentService<T> dev.langchain4j.agentic.supervisor.SupervisorAgentService<T>::compensateOnError(boolean)",
|
||||
"justification": "Add tool actions compensation at agentic system level for supervisor agents"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -62,6 +62,14 @@ public @interface Agent {
|
|||
*/
|
||||
boolean optional() default false;
|
||||
|
||||
/**
|
||||
* If true, all previously successful tool invocations with {@code @CompensateFor} actions will be
|
||||
* compensated in reverse order when any tool in this agent fails.
|
||||
*
|
||||
* @return true if cross-agent compensation should be enabled, false otherwise.
|
||||
*/
|
||||
boolean compensateOnError() default false;
|
||||
|
||||
/**
|
||||
* Names of other agents participating in the definition of the context of this agent.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -418,7 +418,8 @@ public class AgenticServices {
|
|||
}
|
||||
|
||||
private static void buildAgentSpecs(
|
||||
Method agentMethod, String name, String description, String outputKey, AgenticService<?, ?> builder) {
|
||||
Method agentMethod, String name, String description, String outputKey,
|
||||
boolean compensateOnError, AgenticService<?, ?> builder) {
|
||||
if (!isNullOrBlank(name)) {
|
||||
builder.name(name);
|
||||
} else {
|
||||
|
|
@ -430,6 +431,9 @@ public class AgenticServices {
|
|||
if (!isNullOrBlank(outputKey)) {
|
||||
builder.outputKey(outputKey);
|
||||
}
|
||||
if (compensateOnError) {
|
||||
builder.compensateOnError(true);
|
||||
}
|
||||
}
|
||||
|
||||
private static <T> T buildSequentialAgent(
|
||||
|
|
@ -445,6 +449,7 @@ public class AgenticServices {
|
|||
annotation.name(),
|
||||
annotation.description(),
|
||||
AgentUtil.outputKey(annotation.outputKey(), annotation.typedOutputKey()),
|
||||
annotation.compensateOnError(),
|
||||
builder);
|
||||
|
||||
return builder.build();
|
||||
|
|
@ -464,6 +469,7 @@ public class AgenticServices {
|
|||
annotation.name(),
|
||||
annotation.description(),
|
||||
AgentUtil.outputKey(annotation.outputKey(), annotation.typedOutputKey()),
|
||||
annotation.compensateOnError(),
|
||||
builder);
|
||||
|
||||
return builder.build();
|
||||
|
|
@ -481,6 +487,7 @@ public class AgenticServices {
|
|||
annotation.name(),
|
||||
annotation.description(),
|
||||
AgentUtil.outputKey(annotation.outputKey(), annotation.typedOutputKey()),
|
||||
annotation.compensateOnError(),
|
||||
builder);
|
||||
|
||||
for (Class<?> subagent : annotation.subAgents()) {
|
||||
|
|
@ -511,6 +518,7 @@ public class AgenticServices {
|
|||
annotation.name(),
|
||||
annotation.description(),
|
||||
AgentUtil.outputKey(annotation.outputKey(), annotation.typedOutputKey()),
|
||||
annotation.compensateOnError(),
|
||||
builder);
|
||||
|
||||
return builder.build();
|
||||
|
|
@ -530,6 +538,7 @@ public class AgenticServices {
|
|||
annotation.name(),
|
||||
annotation.description(),
|
||||
AgentUtil.outputKey(annotation.outputKey(), annotation.typedOutputKey()),
|
||||
annotation.compensateOnError(),
|
||||
builder);
|
||||
|
||||
return builder.build();
|
||||
|
|
@ -548,6 +557,7 @@ public class AgenticServices {
|
|||
annotation.name(),
|
||||
annotation.description(),
|
||||
AgentUtil.outputKey(annotation.outputKey(), annotation.typedOutputKey()),
|
||||
annotation.compensateOnError(),
|
||||
builder);
|
||||
|
||||
return builder.build();
|
||||
|
|
@ -574,6 +584,9 @@ public class AgenticServices {
|
|||
builder.description(supervisorAgent.description());
|
||||
}
|
||||
builder.outputKey(AgentUtil.outputKey(supervisorAgent.outputKey(), supervisorAgent.typedOutputKey()));
|
||||
if (supervisorAgent.compensateOnError()) {
|
||||
builder.compensateOnError(true);
|
||||
}
|
||||
|
||||
return builder.build();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -126,6 +126,7 @@ public class AgentBuilder<T, B extends AgentBuilder<T, ?>> {
|
|||
private Executor concurrentToolsExecutor;
|
||||
private ToolArgumentsErrorHandler toolArgumentsErrorHandler;
|
||||
private ToolExecutionErrorHandler toolExecutionErrorHandler;
|
||||
boolean compensateOnError;
|
||||
|
||||
java.util.function.Function<InternalAgent, Object> agentInstanceFactory;
|
||||
|
||||
|
|
@ -166,6 +167,7 @@ public class AgentBuilder<T, B extends AgentBuilder<T, ?>> {
|
|||
|
||||
this.async = agent.async();
|
||||
this.optional = agent.optional();
|
||||
this.compensateOnError = agent.compensateOnError();
|
||||
if (agent.summarizedContext() != null && agent.summarizedContext().length > 0) {
|
||||
this.contextProvidingAgents = agent.summarizedContext();
|
||||
}
|
||||
|
|
@ -253,6 +255,10 @@ public class AgentBuilder<T, B extends AgentBuilder<T, ?>> {
|
|||
agentListener.afterAgentToolExecution(new AfterAgentToolExecution(agent, afterToolExecution)));
|
||||
}
|
||||
|
||||
if (compensateOnError) {
|
||||
((InternalAgent) agent).enableCrossAgentCompensation();
|
||||
}
|
||||
|
||||
return (T) agent;
|
||||
}
|
||||
|
||||
|
|
@ -680,6 +686,12 @@ public class AgentBuilder<T, B extends AgentBuilder<T, ?>> {
|
|||
return (B) this;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public B compensateOnError(boolean compensateOnError) {
|
||||
this.compensateOnError = compensateOnError;
|
||||
return (B) this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets a function that provides additional context from the {@link AgenticScope} to this agent's prompt.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -60,6 +60,7 @@ public class AgentInvocationHandler implements InvocationHandler, InternalAgent
|
|||
private String agentId;
|
||||
private InternalAgent parent;
|
||||
private AgentListener agentListener;
|
||||
private boolean crossAgentCompensationEnabled;
|
||||
|
||||
private final Map<Object, AiServiceResponseReceivedEvent> lastResponseEvents = new ConcurrentHashMap<>();
|
||||
|
||||
|
|
@ -233,6 +234,11 @@ public class AgentInvocationHandler implements InvocationHandler, InternalAgent
|
|||
return "Agent<" + builder.agentServiceClass.getSimpleName() + ">";
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean compensateOnError() {
|
||||
return builder.compensateOnError || (parent != null && parent.compensateOnError());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setParent(InternalAgent parent) {
|
||||
if (builder.hasChatMemory() && parent != null && !parent.allowChatMemory()) {
|
||||
|
|
@ -241,6 +247,36 @@ public class AgentInvocationHandler implements InvocationHandler, InternalAgent
|
|||
}
|
||||
this.parent = parent;
|
||||
registerInheritedParentListener(parent.listener());
|
||||
if (compensateOnError()) {
|
||||
enableCrossAgentCompensation();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void enableCrossAgentCompensation() {
|
||||
if (crossAgentCompensationEnabled) {
|
||||
return;
|
||||
}
|
||||
crossAgentCompensationEnabled = true;
|
||||
context.toolService.onCompensableToolExecution((toolExecution, compensatingAction) -> {
|
||||
var managed = toolExecution.invocationContext().managedParameters();
|
||||
if (managed != null) {
|
||||
var scopeManaged = managed.get(AgenticScope.class);
|
||||
if (scopeManaged instanceof DefaultAgenticScope das) {
|
||||
das.registerCompensableExecution(toolExecution, compensatingAction);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
context.toolService.onToolExecutionError(invocationContext -> {
|
||||
var managed = invocationContext.managedParameters();
|
||||
if (managed != null) {
|
||||
var scopeManaged = managed.get(AgenticScope.class);
|
||||
if (scopeManaged instanceof DefaultAgenticScope das) {
|
||||
das.compensateAll();
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -82,4 +82,12 @@ public @interface ConditionalAgent {
|
|||
* @return array of sub-agents.
|
||||
*/
|
||||
Class<?>[] subAgents();
|
||||
|
||||
/**
|
||||
* When true, if any agent in the hierarchy fails, all previously successful tool
|
||||
* executions with {@code @CompensateFor} actions are compensated in reverse order.
|
||||
*
|
||||
* @return whether to enable cross-agent compensation on error.
|
||||
*/
|
||||
boolean compensateOnError() default false;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -75,4 +75,12 @@ public @interface LoopAgent {
|
|||
* @return maximum number of iterations.
|
||||
*/
|
||||
int maxIterations() default 10;
|
||||
|
||||
/**
|
||||
* When true, if any agent in the hierarchy fails, all previously successful tool
|
||||
* executions with {@code @CompensateFor} actions are compensated in reverse order.
|
||||
*
|
||||
* @return whether to enable cross-agent compensation on error.
|
||||
*/
|
||||
boolean compensateOnError() default false;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -65,4 +65,12 @@ public @interface ParallelAgent {
|
|||
* @return array of sub-agents.
|
||||
*/
|
||||
Class<?>[] subAgents();
|
||||
|
||||
/**
|
||||
* When true, if any agent in the hierarchy fails, all previously successful tool
|
||||
* executions with {@code @CompensateFor} actions are compensated in reverse order.
|
||||
*
|
||||
* @return whether to enable cross-agent compensation on error.
|
||||
*/
|
||||
boolean compensateOnError() default false;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -73,4 +73,12 @@ public @interface ParallelMapperAgent {
|
|||
* @return the name of the input collection.
|
||||
*/
|
||||
String itemsProvider() default "";
|
||||
|
||||
/**
|
||||
* When true, if any agent in the hierarchy fails, all previously successful tool
|
||||
* executions with {@code @CompensateFor} actions are compensated in reverse order.
|
||||
*
|
||||
* @return whether to enable cross-agent compensation on error.
|
||||
*/
|
||||
boolean compensateOnError() default false;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -69,4 +69,12 @@ public @interface PlannerAgent {
|
|||
* @return array of sub-agents.
|
||||
*/
|
||||
Class<?>[] subAgents();
|
||||
|
||||
/**
|
||||
* When true, if any agent in the hierarchy fails, all previously successful tool
|
||||
* executions with {@code @CompensateFor} actions are compensated in reverse order.
|
||||
*
|
||||
* @return whether to enable cross-agent compensation on error.
|
||||
*/
|
||||
boolean compensateOnError() default false;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -65,4 +65,12 @@ public @interface SequenceAgent {
|
|||
* @return array of sub-agents.
|
||||
*/
|
||||
Class<?>[] subAgents();
|
||||
|
||||
/**
|
||||
* When true, if any agent in the hierarchy fails, all previously successful tool
|
||||
* executions with {@code @CompensateFor} actions are compensated in reverse order.
|
||||
*
|
||||
* @return whether to enable cross-agent compensation on error.
|
||||
*/
|
||||
boolean compensateOnError() default false;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -88,4 +88,12 @@ public @interface SupervisorAgent {
|
|||
* Strategy to decide which response the supervisor agent should return.
|
||||
*/
|
||||
SupervisorResponseStrategy responseStrategy() default SupervisorResponseStrategy.LAST;
|
||||
|
||||
/**
|
||||
* When true, if any agent in the hierarchy fails, all previously successful tool
|
||||
* executions with {@code @CompensateFor} actions are compensated in reverse order.
|
||||
*
|
||||
* @return whether to enable cross-agent compensation on error.
|
||||
*/
|
||||
boolean compensateOnError() default false;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -134,6 +134,16 @@ public abstract class AbstractAgentInvoker implements AgentInvoker, InternalAgen
|
|||
"agentInstance=" + agent + ']';
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean compensateOnError() {
|
||||
return agent.compensateOnError();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void enableCrossAgentCompensation() {
|
||||
agent.enableCrossAgentCompensation();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setParent(InternalAgent parent) {
|
||||
agent.setParent(parent);
|
||||
|
|
|
|||
|
|
@ -48,6 +48,8 @@ public abstract class AbstractServiceBuilder<T, S> {
|
|||
|
||||
protected Function<ErrorContext, ErrorRecoveryResult> errorHandler;
|
||||
|
||||
protected boolean compensateOnError = false;
|
||||
|
||||
protected Function<InternalAgent, Object> agentInstanceFactory;
|
||||
|
||||
protected Executor executor;
|
||||
|
|
@ -128,6 +130,11 @@ public abstract class AbstractServiceBuilder<T, S> {
|
|||
return (S) this;
|
||||
}
|
||||
|
||||
public S compensateOnError(boolean compensateOnError) {
|
||||
this.compensateOnError = compensateOnError;
|
||||
return (S) this;
|
||||
}
|
||||
|
||||
public S listener(AgentListener agentListener) {
|
||||
if (this.agentListener == null) {
|
||||
this.agentListener = agentListener;
|
||||
|
|
|
|||
|
|
@ -204,6 +204,16 @@ public record AgentExecutor(AgentInvoker agentInvoker, Object agent) implements
|
|||
agentInvoker.setParent(parent);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean compensateOnError() {
|
||||
return agentInvoker.compensateOnError();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void enableCrossAgentCompensation() {
|
||||
agentInvoker.enableCrossAgentCompensation();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void registerInheritedParentListener(AgentListener parentListener) {
|
||||
agentInvoker.registerInheritedParentListener(parentListener);
|
||||
|
|
|
|||
|
|
@ -22,4 +22,11 @@ public interface InternalAgent extends AgentInstance {
|
|||
default boolean allowChatMemory() {
|
||||
return true;
|
||||
}
|
||||
|
||||
default boolean compensateOnError() {
|
||||
return false;
|
||||
}
|
||||
|
||||
default void enableCrossAgentCompensation() {
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -90,6 +90,7 @@ public class PlannerBasedInvocationHandler implements InvocationHandler, Interna
|
|||
|
||||
private String agentId;
|
||||
private InternalAgent parent;
|
||||
private boolean crossAgentCompensationEnabled;
|
||||
|
||||
public PlannerBasedInvocationHandler(AbstractServiceBuilder<?, ?> service, Supplier<Planner> plannerSupplier) {
|
||||
this(service, null, service.name, plannerSupplier, null);
|
||||
|
|
@ -218,6 +219,7 @@ public class PlannerBasedInvocationHandler implements InvocationHandler, Interna
|
|||
try {
|
||||
result = new PlannerLoop(planner, currentScope, registry).loop();
|
||||
} catch (Exception e) {
|
||||
currentScope.compensateAll();
|
||||
if (isRootCall()) {
|
||||
agentError(agentListener, currentScope, this, namedArgs, e);
|
||||
currentScope.rootCallEnded(registry, agentListener);
|
||||
|
|
@ -315,6 +317,22 @@ public class PlannerBasedInvocationHandler implements InvocationHandler, Interna
|
|||
return parent;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean compensateOnError() {
|
||||
if (service.compensateOnError) return true;
|
||||
return parent != null && parent.compensateOnError();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void enableCrossAgentCompensation() {
|
||||
if (crossAgentCompensationEnabled) {
|
||||
return;
|
||||
}
|
||||
crossAgentCompensationEnabled = true;
|
||||
subagents.stream().map(InternalAgent.class::cast)
|
||||
.forEach(InternalAgent::enableCrossAgentCompensation);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setParent(InternalAgent parent) {
|
||||
if (parent == null) {
|
||||
|
|
@ -325,6 +343,9 @@ public class PlannerBasedInvocationHandler implements InvocationHandler, Interna
|
|||
if (!parent.allowStreamingOutput()) {
|
||||
this.allowStreamingOutput = false;
|
||||
}
|
||||
if (compensateOnError()) {
|
||||
enableCrossAgentCompensation();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -9,27 +9,124 @@ import java.util.Collection;
|
|||
import java.util.function.Consumer;
|
||||
import java.util.function.Function;
|
||||
|
||||
/**
|
||||
* Fluent builder interface for configuring and assembling an agentic service
|
||||
* composed of one or more sub-agents orchestrated by a planner.
|
||||
*
|
||||
* @param <T> the self-referencing builder type (for fluent chaining)
|
||||
* @param <A> the type of the built agent proxy
|
||||
*/
|
||||
public interface AgenticService<T, A> {
|
||||
|
||||
/**
|
||||
* Builds the agentic service, returning a proxy that can be invoked to
|
||||
* execute the configured workflow.
|
||||
*
|
||||
* @return the built agent proxy
|
||||
*/
|
||||
A build();
|
||||
|
||||
/**
|
||||
* Sets the sub-agents that participate in this agentic workflow.
|
||||
*
|
||||
* @param agents the sub-agents to orchestrate
|
||||
* @return this builder for fluent chaining
|
||||
*/
|
||||
T subAgents(Object... agents);
|
||||
|
||||
/**
|
||||
* Sets the sub-agents that participate in this agentic workflow.
|
||||
*
|
||||
* @param agents a collection of sub-agents to orchestrate
|
||||
* @return this builder for fluent chaining
|
||||
*/
|
||||
T subAgents(Collection<?> agents);
|
||||
|
||||
/**
|
||||
* Registers a callback invoked before each agent call, allowing inspection
|
||||
* or modification of the {@link AgenticScope} (e.g. to inject variables).
|
||||
*
|
||||
* @param beforeCall the callback to invoke before each call
|
||||
* @return this builder for fluent chaining
|
||||
*/
|
||||
T beforeCall(Consumer<AgenticScope> beforeCall);
|
||||
|
||||
/**
|
||||
* Sets the name of this agentic service. If not provided, the method name
|
||||
* of the agent interface is used.
|
||||
*
|
||||
* @param name the name of the agentic service
|
||||
* @return this builder for fluent chaining
|
||||
*/
|
||||
T name(String name);
|
||||
|
||||
/**
|
||||
* Sets the description of this agentic service. The description should be
|
||||
* clear enough for a language model to understand the agent's purpose.
|
||||
*
|
||||
* @param description the description of the agentic service
|
||||
* @return this builder for fluent chaining
|
||||
*/
|
||||
T description(String description);
|
||||
|
||||
/**
|
||||
* Sets the key under which the final output of this workflow is stored
|
||||
* in the {@link AgenticScope}.
|
||||
*
|
||||
* @param outputKey the name of the output variable
|
||||
* @return this builder for fluent chaining
|
||||
*/
|
||||
T outputKey(String outputKey);
|
||||
|
||||
/**
|
||||
* Sets a strongly typed key for the output variable, enforcing type safety
|
||||
* when retrieving the result from the {@link AgenticScope}. Use this as an
|
||||
* alternative to {@link #outputKey(String)}.
|
||||
*
|
||||
* @param outputKey the class representing the typed output variable
|
||||
* @return this builder for fluent chaining
|
||||
*/
|
||||
T outputKey(Class<? extends TypedKey<?>> outputKey);
|
||||
|
||||
/**
|
||||
* Sets a custom function to extract the final output from the
|
||||
* {@link AgenticScope} at the end of the workflow. Use this when the
|
||||
* output requires transformation or aggregation beyond a simple key lookup.
|
||||
*
|
||||
* @param output a function that receives the scope and returns the output
|
||||
* @return this builder for fluent chaining
|
||||
*/
|
||||
T output(Function<AgenticScope, Object> output);
|
||||
|
||||
/**
|
||||
* Registers an error handler that is invoked when a sub-agent fails. The
|
||||
* handler receives an {@link ErrorContext} and returns an
|
||||
* {@link ErrorRecoveryResult} indicating whether to propagate the exception,
|
||||
* retry the failed agent, or return a fallback result.
|
||||
*
|
||||
* @param errorHandler the error handling function
|
||||
* @return this builder for fluent chaining
|
||||
*/
|
||||
T errorHandler(Function<ErrorContext, ErrorRecoveryResult> errorHandler);
|
||||
|
||||
/**
|
||||
* Enables or disables cross-agent compensation. When enabled and any agent
|
||||
* in the hierarchy fails, all previously successful tool invocations that
|
||||
* have {@code @CompensateFor} actions are compensated in reverse
|
||||
* chronological order, making the workflow's side effects atomic.
|
||||
* Defaults to {@code false}.
|
||||
*
|
||||
* @param compensateOnError whether to enable cross-agent compensation
|
||||
* @return this builder for fluent chaining
|
||||
*/
|
||||
T compensateOnError(boolean compensateOnError);
|
||||
|
||||
/**
|
||||
* Registers an {@link AgentListener} to observe agent invocations, tool
|
||||
* executions, and lifecycle events within this agentic service.
|
||||
*
|
||||
* @param listeners the listener to register
|
||||
* @return this builder for fluent chaining
|
||||
*/
|
||||
T listener(AgentListener listeners);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -31,9 +31,11 @@ import java.util.concurrent.ConcurrentHashMap;
|
|||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.locks.ReadWriteLock;
|
||||
import java.util.concurrent.locks.ReentrantReadWriteLock;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.Predicate;
|
||||
import java.util.stream.Collectors;
|
||||
import dev.langchain4j.service.tool.ToolExecution;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
|
|
@ -51,6 +53,7 @@ public class DefaultAgenticScope implements AgenticScope {
|
|||
|
||||
private final transient Map<String, Object> agents = new ConcurrentHashMap<>();
|
||||
private final transient Map<String, Object> executionContexts = new ConcurrentHashMap<>();
|
||||
private final transient List<CompensableExecution> compensableExecutions = Collections.synchronizedList(new ArrayList<>());
|
||||
|
||||
private static final Function<ErrorContext, ErrorRecoveryResult> DEFAULT_ERROR_RECOVERY =
|
||||
errorContext -> ErrorRecoveryResult.throwException();
|
||||
|
|
@ -231,6 +234,8 @@ public class DefaultAgenticScope implements AgenticScope {
|
|||
} else if (kind == Kind.PERSISTENT) {
|
||||
flush(registry);
|
||||
}
|
||||
|
||||
compensableExecutions.clear();
|
||||
}
|
||||
|
||||
private void flush(AgenticScopeRegistry registry) {
|
||||
|
|
@ -375,6 +380,34 @@ public class DefaultAgenticScope implements AgenticScope {
|
|||
return errorHandler.apply(new ErrorContext(agentName, this, exception));
|
||||
}
|
||||
|
||||
private record CompensableExecution(ToolExecution toolExecution, Consumer<ToolExecution> compensatingAction) {
|
||||
|
||||
public void compensate() {
|
||||
try {
|
||||
compensatingAction.accept(toolExecution());
|
||||
} catch (Exception e) {
|
||||
LOG.warn("Cross-agent compensating action failed for tool '{}': {}",
|
||||
toolExecution().request().name(), e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void registerCompensableExecution(ToolExecution toolExecution, Consumer<ToolExecution> compensatingAction) {
|
||||
compensableExecutions.add(new CompensableExecution(toolExecution, compensatingAction));
|
||||
}
|
||||
|
||||
public void compensateAll() {
|
||||
List<CompensableExecution> snapshot;
|
||||
synchronized (compensableExecutions) {
|
||||
snapshot = new ArrayList<>(compensableExecutions);
|
||||
compensableExecutions.clear();
|
||||
}
|
||||
|
||||
for (int i = snapshot.size() - 1; i >= 0; i--) {
|
||||
snapshot.get(i).compensate();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Checkpoints the current state of this scope by persisting it to the store.
|
||||
* This is a no-op for non-persistent scopes. For persistent scopes, it acquires
|
||||
|
|
|
|||
|
|
@ -45,4 +45,6 @@ public interface SupervisorAgentService<T> {
|
|||
SupervisorAgentService<T> listener(AgentListener agentListener);
|
||||
|
||||
SupervisorAgentService<T> beforeCall(Consumer<AgenticScope> beforeCall);
|
||||
|
||||
SupervisorAgentService<T> compensateOnError(boolean compensateOnError);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,497 @@
|
|||
package dev.langchain4j.agentic;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
|
||||
import dev.langchain4j.agent.tool.CompensateFor;
|
||||
import dev.langchain4j.agent.tool.P;
|
||||
import dev.langchain4j.agent.tool.Tool;
|
||||
import dev.langchain4j.agent.tool.ToolExecutionRequest;
|
||||
import dev.langchain4j.agentic.AgenticServices.AgentConfigurator;
|
||||
import dev.langchain4j.agentic.agent.AgentInvocationException;
|
||||
import dev.langchain4j.service.IllegalConfigurationException;
|
||||
import dev.langchain4j.agentic.declarative.ChatModelSupplier;
|
||||
import dev.langchain4j.agentic.declarative.SequenceAgent;
|
||||
import dev.langchain4j.data.message.AiMessage;
|
||||
import dev.langchain4j.model.chat.ChatModel;
|
||||
import dev.langchain4j.model.chat.request.ChatRequest;
|
||||
import dev.langchain4j.model.chat.response.ChatResponse;
|
||||
import dev.langchain4j.model.chat.response.ChatResponseMetadata;
|
||||
import dev.langchain4j.service.UserMessage;
|
||||
import dev.langchain4j.service.V;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Queue;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
public class CrossAgentCompensationTest {
|
||||
|
||||
static final List<String> compensationLog = Collections.synchronizedList(new ArrayList<>());
|
||||
|
||||
static ChatModel modelThatCallsTool(String toolName, String toolArgs) {
|
||||
Queue<AiMessage> responses = new ConcurrentLinkedQueue<>();
|
||||
responses.add(AiMessage.from(ToolExecutionRequest.builder()
|
||||
.id("call-1")
|
||||
.name(toolName)
|
||||
.arguments(toolArgs)
|
||||
.build()));
|
||||
responses.add(AiMessage.from("done"));
|
||||
return new ChatModel() {
|
||||
@Override
|
||||
public ChatResponse chat(ChatRequest request) {
|
||||
return ChatResponse.builder()
|
||||
.aiMessage(responses.isEmpty() ? AiMessage.from("done") : responses.poll())
|
||||
.metadata(ChatResponseMetadata.builder().build())
|
||||
.build();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
static ChatModel throwingModel() {
|
||||
return new ChatModel() {
|
||||
@Override
|
||||
public ChatResponse chat(ChatRequest request) {
|
||||
throw new RuntimeException("Agent failed");
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
// ----- Tool classes -----
|
||||
|
||||
static class CreditService {
|
||||
boolean credited = false;
|
||||
boolean compensated = false;
|
||||
|
||||
@Tool("credits an account")
|
||||
String credit(@P(name = "amount") int amount) {
|
||||
credited = true;
|
||||
compensationLog.add("credit:" + amount);
|
||||
return "credited " + amount;
|
||||
}
|
||||
|
||||
@CompensateFor("credit")
|
||||
void uncredit(int amount) {
|
||||
compensated = true;
|
||||
compensationLog.add("uncredit:" + amount);
|
||||
}
|
||||
}
|
||||
|
||||
static class DebitService {
|
||||
boolean debited = false;
|
||||
boolean compensated = false;
|
||||
|
||||
@Tool("debits an account")
|
||||
String debit(@P(name = "amount") int amount) {
|
||||
debited = true;
|
||||
compensationLog.add("debit:" + amount);
|
||||
return "debited " + amount;
|
||||
}
|
||||
|
||||
@CompensateFor("debit")
|
||||
void undebit(int amount) {
|
||||
compensated = true;
|
||||
compensationLog.add("undebit:" + amount);
|
||||
}
|
||||
}
|
||||
|
||||
static class ThrowingCompensationService {
|
||||
boolean compensated = false;
|
||||
|
||||
@Tool("does something")
|
||||
String doSomething(@P(name = "input") String input) {
|
||||
compensationLog.add("doSomething:" + input);
|
||||
return "done " + input;
|
||||
}
|
||||
|
||||
@CompensateFor("doSomething")
|
||||
void undoSomething(String input) {
|
||||
compensationLog.add("undoSomething:" + input);
|
||||
compensated = true;
|
||||
throw new RuntimeException("Compensation failed");
|
||||
}
|
||||
}
|
||||
|
||||
static class NoCompensationToolService {
|
||||
@Tool("tool without compensation")
|
||||
String noCompensation(@P(name = "input") String input) {
|
||||
compensationLog.add("noCompensation:" + input);
|
||||
return "done";
|
||||
}
|
||||
}
|
||||
|
||||
// ----- Agent interfaces -----
|
||||
|
||||
public interface CreditAgentService {
|
||||
@Agent(description = "Credits an account", outputKey = "creditResult")
|
||||
@UserMessage("Credit the account for request: {{request}}")
|
||||
String execute(@V("request") String request);
|
||||
}
|
||||
|
||||
public interface DebitAgentService {
|
||||
@Agent(description = "Debits an account", outputKey = "debitResult")
|
||||
@UserMessage("Debit the account for request: {{request}}")
|
||||
String execute(@V("request") String request);
|
||||
}
|
||||
|
||||
public interface FailingAgentService {
|
||||
@Agent(description = "An agent that fails", outputKey = "failResult")
|
||||
@UserMessage("Do something with {{request}}")
|
||||
String execute(@V("request") String request);
|
||||
}
|
||||
|
||||
public interface SimpleAgentService {
|
||||
@Agent(description = "A simple agent", outputKey = "simpleResult")
|
||||
@UserMessage("Process {{request}}")
|
||||
String execute(@V("request") String request);
|
||||
}
|
||||
|
||||
public interface NoCompAgentService {
|
||||
@Agent(description = "Agent without compensation tools", outputKey = "noCompResult")
|
||||
@UserMessage("Process {{request}}")
|
||||
String execute(@V("request") String request);
|
||||
}
|
||||
|
||||
// ----- Tests -----
|
||||
|
||||
@Test
|
||||
void should_compensate_all_tools_in_sequence_when_last_agent_fails() {
|
||||
compensationLog.clear();
|
||||
CreditService creditService = new CreditService();
|
||||
DebitService debitService = new DebitService();
|
||||
|
||||
var creditAgent = AgenticServices.agentBuilder(CreditAgentService.class)
|
||||
.chatModel(modelThatCallsTool("credit", "{\"amount\": 100}"))
|
||||
.tools(creditService)
|
||||
.name("creditAgent")
|
||||
.build();
|
||||
|
||||
var debitAgent = AgenticServices.agentBuilder(DebitAgentService.class)
|
||||
.chatModel(modelThatCallsTool("debit", "{\"amount\": 100}"))
|
||||
.tools(debitService)
|
||||
.name("debitAgent")
|
||||
.build();
|
||||
|
||||
var failAgent = AgenticServices.agentBuilder(FailingAgentService.class)
|
||||
.chatModel(throwingModel())
|
||||
.name("failAgent")
|
||||
.build();
|
||||
|
||||
interface TestAgent {
|
||||
String run(@V("request") String request);
|
||||
}
|
||||
|
||||
TestAgent sequenceAgent = AgenticServices.<TestAgent>sequenceBuilder(TestAgent.class)
|
||||
.subAgents(creditAgent, debitAgent, failAgent)
|
||||
.compensateOnError(true)
|
||||
.name("sequenceAgent")
|
||||
.build();
|
||||
|
||||
assertThrows(AgentInvocationException.class, () -> sequenceAgent.run("test"));
|
||||
|
||||
assertThat(creditService.credited).isTrue();
|
||||
assertThat(debitService.debited).isTrue();
|
||||
assertThat(creditService.compensated).isTrue();
|
||||
assertThat(debitService.compensated).isTrue();
|
||||
|
||||
int uncreditIdx = compensationLog.indexOf("uncredit:100");
|
||||
int undebitIdx = compensationLog.indexOf("undebit:100");
|
||||
assertThat(undebitIdx).isLessThan(uncreditIdx);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_not_compensate_when_flag_is_not_set() {
|
||||
compensationLog.clear();
|
||||
CreditService creditService = new CreditService();
|
||||
|
||||
var creditAgent = AgenticServices.agentBuilder(CreditAgentService.class)
|
||||
.chatModel(modelThatCallsTool("credit", "{\"amount\": 100}"))
|
||||
.tools(creditService)
|
||||
.name("creditAgent")
|
||||
.build();
|
||||
|
||||
var failAgent = AgenticServices.agentBuilder(FailingAgentService.class)
|
||||
.chatModel(throwingModel())
|
||||
.name("failAgent")
|
||||
.build();
|
||||
|
||||
interface TestAgent {
|
||||
String run(@V("request") String request);
|
||||
}
|
||||
|
||||
TestAgent sequenceAgent = AgenticServices.<TestAgent>sequenceBuilder(TestAgent.class)
|
||||
.subAgents(creditAgent, failAgent)
|
||||
.name("sequenceAgent")
|
||||
.build();
|
||||
|
||||
assertThrows(AgentInvocationException.class, () -> sequenceAgent.run("test"));
|
||||
|
||||
assertThat(creditService.credited).isTrue();
|
||||
assertThat(creditService.compensated).isFalse();
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_only_compensate_tools_with_compensateFor_annotation() {
|
||||
compensationLog.clear();
|
||||
CreditService creditService = new CreditService();
|
||||
NoCompensationToolService noCompService = new NoCompensationToolService();
|
||||
|
||||
var creditAgent = AgenticServices.agentBuilder(CreditAgentService.class)
|
||||
.chatModel(modelThatCallsTool("credit", "{\"amount\": 100}"))
|
||||
.tools(creditService)
|
||||
.name("creditAgent")
|
||||
.build();
|
||||
|
||||
var noCompAgent = AgenticServices.agentBuilder(NoCompAgentService.class)
|
||||
.chatModel(modelThatCallsTool("noCompensation", "{\"input\": \"test\"}"))
|
||||
.tools(noCompService)
|
||||
.name("noCompAgent")
|
||||
.build();
|
||||
|
||||
var failAgent = AgenticServices.agentBuilder(FailingAgentService.class)
|
||||
.chatModel(throwingModel())
|
||||
.name("failAgent")
|
||||
.build();
|
||||
|
||||
interface TestAgent {
|
||||
String run(@V("request") String request);
|
||||
}
|
||||
|
||||
TestAgent sequenceAgent = AgenticServices.<TestAgent>sequenceBuilder(TestAgent.class)
|
||||
.subAgents(creditAgent, noCompAgent, failAgent)
|
||||
.compensateOnError(true)
|
||||
.name("sequenceAgent")
|
||||
.build();
|
||||
|
||||
assertThrows(AgentInvocationException.class, () -> sequenceAgent.run("test"));
|
||||
|
||||
assertThat(creditService.compensated).isTrue();
|
||||
assertThat(compensationLog).containsExactly("credit:100", "noCompensation:test", "uncredit:100");
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_continue_compensating_when_one_compensation_fails() {
|
||||
compensationLog.clear();
|
||||
ThrowingCompensationService throwingCompService = new ThrowingCompensationService();
|
||||
CreditService creditService = new CreditService();
|
||||
|
||||
var throwingCompAgent = AgenticServices.agentBuilder(SimpleAgentService.class)
|
||||
.chatModel(modelThatCallsTool("doSomething", "{\"input\": \"test\"}"))
|
||||
.tools(throwingCompService)
|
||||
.name("throwingCompAgent")
|
||||
.build();
|
||||
|
||||
var creditAgent = AgenticServices.agentBuilder(CreditAgentService.class)
|
||||
.chatModel(modelThatCallsTool("credit", "{\"amount\": 50}"))
|
||||
.tools(creditService)
|
||||
.name("creditAgent")
|
||||
.build();
|
||||
|
||||
var failAgent = AgenticServices.agentBuilder(FailingAgentService.class)
|
||||
.chatModel(throwingModel())
|
||||
.name("failAgent")
|
||||
.build();
|
||||
|
||||
interface TestAgent {
|
||||
String run(@V("request") String request);
|
||||
}
|
||||
|
||||
TestAgent sequenceAgent = AgenticServices.<TestAgent>sequenceBuilder(TestAgent.class)
|
||||
.subAgents(throwingCompAgent, creditAgent, failAgent)
|
||||
.compensateOnError(true)
|
||||
.name("sequenceAgent")
|
||||
.build();
|
||||
|
||||
assertThrows(AgentInvocationException.class, () -> sequenceAgent.run("test"));
|
||||
|
||||
// Both compensations attempted even though the first throws
|
||||
assertThat(throwingCompService.compensated).isTrue();
|
||||
assertThat(creditService.compensated).isTrue();
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_compensate_in_reverse_chronological_order() {
|
||||
compensationLog.clear();
|
||||
CreditService creditService = new CreditService();
|
||||
DebitService debitService = new DebitService();
|
||||
|
||||
var creditAgent = AgenticServices.agentBuilder(CreditAgentService.class)
|
||||
.chatModel(modelThatCallsTool("credit", "{\"amount\": 200}"))
|
||||
.tools(creditService)
|
||||
.name("creditAgent")
|
||||
.build();
|
||||
|
||||
var debitAgent = AgenticServices.agentBuilder(DebitAgentService.class)
|
||||
.chatModel(modelThatCallsTool("debit", "{\"amount\": 300}"))
|
||||
.tools(debitService)
|
||||
.name("debitAgent")
|
||||
.build();
|
||||
|
||||
var failAgent = AgenticServices.agentBuilder(FailingAgentService.class)
|
||||
.chatModel(throwingModel())
|
||||
.name("failAgent")
|
||||
.build();
|
||||
|
||||
interface TestAgent {
|
||||
String run(@V("request") String request);
|
||||
}
|
||||
|
||||
TestAgent sequenceAgent = AgenticServices.<TestAgent>sequenceBuilder(TestAgent.class)
|
||||
.subAgents(creditAgent, debitAgent, failAgent)
|
||||
.compensateOnError(true)
|
||||
.name("sequenceAgent")
|
||||
.build();
|
||||
|
||||
assertThrows(AgentInvocationException.class, () -> sequenceAgent.run("test"));
|
||||
|
||||
assertThat(compensationLog).containsExactly(
|
||||
"credit:200", "debit:300", "undebit:300", "uncredit:200");
|
||||
}
|
||||
|
||||
static class MisconfiguredCompensationService {
|
||||
@Tool("credits an account")
|
||||
String credit(@P(name = "amount") int amount) {
|
||||
return "credited " + amount;
|
||||
}
|
||||
|
||||
@CompensateFor("credt") // typo
|
||||
void uncredit(int amount) {
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_fail_fast_on_misconfigured_compensateFor() {
|
||||
var leafAgent = AgenticServices.agentBuilder(CreditAgentService.class)
|
||||
.chatModel(modelThatCallsTool("credit", "{\"amount\": 100}"))
|
||||
.tools(new MisconfiguredCompensationService())
|
||||
.name("misconfiguredAgent")
|
||||
.build();
|
||||
|
||||
interface TestAgent {
|
||||
String run(@V("request") String request);
|
||||
}
|
||||
|
||||
assertThatThrownBy(() -> AgenticServices.<TestAgent>sequenceBuilder(TestAgent.class)
|
||||
.subAgents(leafAgent)
|
||||
.compensateOnError(true)
|
||||
.name("sequenceAgent")
|
||||
.build())
|
||||
.hasRootCauseInstanceOf(IllegalConfigurationException.class)
|
||||
.rootCause().hasMessageContaining("credt");
|
||||
}
|
||||
|
||||
static class CreditAndFailService {
|
||||
boolean credited = false;
|
||||
boolean compensated = false;
|
||||
|
||||
@Tool("credits an account")
|
||||
String credit(@P(name = "amount") int amount) {
|
||||
credited = true;
|
||||
compensationLog.add("credit:" + amount);
|
||||
return "credited " + amount;
|
||||
}
|
||||
|
||||
@CompensateFor("credit")
|
||||
void uncredit(int amount) {
|
||||
compensated = true;
|
||||
compensationLog.add("uncredit:" + amount);
|
||||
}
|
||||
|
||||
@Tool("fails always")
|
||||
String failingTool(@P(name = "input") String input) {
|
||||
throw new RuntimeException("tool failure");
|
||||
}
|
||||
}
|
||||
|
||||
static ChatModel modelThatCallsToolsSequentially(String tool1, String args1, String tool2, String args2) {
|
||||
Queue<AiMessage> responses = new ConcurrentLinkedQueue<>();
|
||||
responses.add(AiMessage.from(ToolExecutionRequest.builder()
|
||||
.id("call-1").name(tool1).arguments(args1).build()));
|
||||
responses.add(AiMessage.from(ToolExecutionRequest.builder()
|
||||
.id("call-2").name(tool2).arguments(args2).build()));
|
||||
responses.add(AiMessage.from("done"));
|
||||
return new ChatModel() {
|
||||
@Override
|
||||
public ChatResponse chat(ChatRequest request) {
|
||||
return ChatResponse.builder()
|
||||
.aiMessage(responses.isEmpty() ? AiMessage.from("done") : responses.poll())
|
||||
.metadata(ChatResponseMetadata.builder().build())
|
||||
.build();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_compensate_on_standalone_leaf_agent() {
|
||||
compensationLog.clear();
|
||||
CreditAndFailService service = new CreditAndFailService();
|
||||
|
||||
CreditAgentService agent = AgenticServices.agentBuilder(CreditAgentService.class)
|
||||
.chatModel(modelThatCallsToolsSequentially(
|
||||
"credit", "{\"amount\": 50}",
|
||||
"failingTool", "{\"input\": \"test\"}"))
|
||||
.tools(service)
|
||||
.compensateOnError(true)
|
||||
.name("standaloneAgent")
|
||||
.build();
|
||||
|
||||
agent.execute("test");
|
||||
|
||||
assertThat(service.credited).isTrue();
|
||||
assertThat(service.compensated).isTrue();
|
||||
assertThat(compensationLog).containsExactly("credit:50", "uncredit:50");
|
||||
}
|
||||
|
||||
// ----- Declarative annotation test -----
|
||||
|
||||
public interface DeclarativeCreditAgent {
|
||||
@Agent(description = "Credits an account", outputKey = "creditResult")
|
||||
@UserMessage("Credit the account for request: {{request}}")
|
||||
String execute(@V("request") String request);
|
||||
|
||||
@ChatModelSupplier
|
||||
static ChatModel chatModel() {
|
||||
return modelThatCallsTool("credit", "{\"amount\": 100}");
|
||||
}
|
||||
}
|
||||
|
||||
public interface DeclarativeFailingAgent {
|
||||
@Agent(description = "A failing agent", outputKey = "failResult")
|
||||
@UserMessage("Do something with {{request}}")
|
||||
String execute(@V("request") String request);
|
||||
|
||||
@ChatModelSupplier
|
||||
static ChatModel chatModel() {
|
||||
return throwingModel();
|
||||
}
|
||||
}
|
||||
|
||||
public interface DeclarativeCompensatingSequence {
|
||||
@SequenceAgent(
|
||||
compensateOnError = true,
|
||||
outputKey = "result",
|
||||
subAgents = {DeclarativeCreditAgent.class, DeclarativeFailingAgent.class})
|
||||
String run(@V("request") String request);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_compensate_with_declarative_annotation() {
|
||||
compensationLog.clear();
|
||||
CreditService creditService = new CreditService();
|
||||
|
||||
DeclarativeCompensatingSequence agent = AgenticServices.createAgenticSystem(
|
||||
DeclarativeCompensatingSequence.class,
|
||||
new AgentConfigurator(ctx -> {
|
||||
if (ctx.agentServiceClass() == DeclarativeCreditAgent.class) {
|
||||
ctx.agentBuilder().tools(creditService);
|
||||
}
|
||||
}, null, null));
|
||||
|
||||
assertThrows(AgentInvocationException.class, () -> agent.run("test"));
|
||||
|
||||
assertThat(creditService.credited).isTrue();
|
||||
assertThat(creditService.compensated).isTrue();
|
||||
}
|
||||
}
|
||||
|
|
@ -14,6 +14,7 @@ import static org.mockito.Mockito.verify;
|
|||
import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import dev.langchain4j.agent.tool.CompensateFor;
|
||||
import dev.langchain4j.agent.tool.P;
|
||||
import dev.langchain4j.agent.tool.Tool;
|
||||
import dev.langchain4j.agent.tool.ToolSpecification;
|
||||
|
|
@ -237,16 +238,35 @@ public class SupervisorAgentIT {
|
|||
return newBalance;
|
||||
}
|
||||
|
||||
@CompensateFor("credit")
|
||||
void reverseCredit(String user, Double amount) {
|
||||
Double balance = accounts.get(user);
|
||||
if (balance != null) {
|
||||
accounts.put(user, balance - amount);
|
||||
}
|
||||
}
|
||||
|
||||
@Tool("Withdraw the given amount with the given user and return the new balance")
|
||||
Double withdraw(@P("user name") String user, @P("amount") Double amount) {
|
||||
Double balance = accounts.get(user);
|
||||
if (balance == null) {
|
||||
throw new RuntimeException("No balance found for user " + user);
|
||||
}
|
||||
if (balance < amount) {
|
||||
throw new RuntimeException("Insufficient balance for user " + user);
|
||||
}
|
||||
Double newBalance = balance - amount;
|
||||
accounts.put(user, newBalance);
|
||||
return newBalance;
|
||||
}
|
||||
|
||||
@CompensateFor("withdraw")
|
||||
void reverseWithdraw(String user, Double amount) {
|
||||
Double balance = accounts.get(user);
|
||||
if (balance != null) {
|
||||
accounts.put(user, balance + amount);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -294,6 +314,57 @@ public class SupervisorAgentIT {
|
|||
assertThat(bankTool.getBalance("Georgios")).isEqualTo(1100.0);
|
||||
}
|
||||
|
||||
@Test
|
||||
void compensating_agentic_banker_test() {
|
||||
agentic_banker_test_with_compensation(true);
|
||||
}
|
||||
|
||||
@Test
|
||||
void non_compensating_agentic_banker_test() {
|
||||
agentic_banker_test_with_compensation(false);
|
||||
}
|
||||
|
||||
public interface BankerAgentWithMemory {
|
||||
|
||||
String execute(@MemoryId String memoryId, @V("request") String request);
|
||||
}
|
||||
|
||||
void agentic_banker_test_with_compensation(boolean compensating) {
|
||||
BankTool bankTool = new BankTool();
|
||||
bankTool.createAccount("Mario", 150.0);
|
||||
bankTool.createAccount("Georgios", 1000.0);
|
||||
|
||||
WithdrawAgent withdrawAgent = AgenticServices.agentBuilder(WithdrawAgent.class)
|
||||
.chatModel(baseModel())
|
||||
.tools(bankTool)
|
||||
.build();
|
||||
CreditAgent creditAgent = AgenticServices.agentBuilder(CreditAgent.class)
|
||||
.chatModel(baseModel())
|
||||
.tools(bankTool)
|
||||
.build();
|
||||
|
||||
BankerAgentWithMemory bankSupervisor = AgenticServices.supervisorBuilder(BankerAgentWithMemory.class)
|
||||
.chatModel(plannerModel())
|
||||
.chatMemoryProvider(memoryId -> MessageWindowChatMemory.withMaxMessages(20))
|
||||
.contextGenerationStrategy(SupervisorContextStrategy.CHAT_MEMORY) // default
|
||||
.responseStrategy(SupervisorResponseStrategy.SUMMARY)
|
||||
.subAgents(withdrawAgent, creditAgent)
|
||||
.compensateOnError(compensating)
|
||||
.build();
|
||||
|
||||
String result = bankSupervisor.execute("1", "Credit 100 USD to Georgios' account and after withdraw them from Mario's one");
|
||||
System.out.println(result);
|
||||
|
||||
assertThat(bankTool.getBalance("Mario")).isEqualTo(50.0);
|
||||
assertThat(bankTool.getBalance("Georgios")).isEqualTo(1100.0);
|
||||
|
||||
result = bankSupervisor.execute("1", "Credit 100 USD to Georgios' account and after withdraw them from Mario's one");
|
||||
System.out.println(result);
|
||||
|
||||
assertThat(bankTool.getBalance("Mario")).isEqualTo(50.0);
|
||||
assertThat(bankTool.getBalance("Georgios")).isEqualTo(compensating ? 1100.0 : 1200.0);
|
||||
}
|
||||
|
||||
public interface ExchangeAgent {
|
||||
@UserMessage(
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -91,7 +91,7 @@ public class ToolService {
|
|||
private final List<ToolSpecification> toolSpecifications = new ArrayList<>();
|
||||
private final Map<String, ToolExecutor> toolExecutors = new HashMap<>();
|
||||
private final Map<String, ReturnBehavior> returnBehaviors = new HashMap<>();
|
||||
private final Map<String, BiConsumer<ToolExecution, InvocationContext>> compensatingExecutors = new HashMap<>();
|
||||
private final Map<String, Consumer<ToolExecution>> compensatingExecutors = new HashMap<>();
|
||||
private IllegalConfigurationException compensatingToolMisconfiguration;
|
||||
private final Set<ToolProvider> toolProviders = new LinkedHashSet<>();
|
||||
private boolean compensateOnToolErrors;
|
||||
|
|
@ -105,6 +105,8 @@ public class ToolService {
|
|||
|
||||
private Consumer<BeforeToolExecution> beforeToolExecution = null;
|
||||
private Consumer<ToolExecution> afterToolExecution = null;
|
||||
private BiConsumer<ToolExecution, Consumer<ToolExecution>> onCompensableToolExecution = null;
|
||||
private Consumer<InvocationContext> onToolExecutionError = null;
|
||||
|
||||
public void hallucinatedToolNameStrategy(
|
||||
Function<ToolExecutionRequest, ToolExecutionResultMessage> toolHallucinationStrategy) {
|
||||
|
|
@ -267,8 +269,8 @@ public class ToolService {
|
|||
}
|
||||
}
|
||||
|
||||
private Map<String, BiConsumer<ToolExecution, InvocationContext>> findCompensatingActions(Object objectWithTools) {
|
||||
Map<String, BiConsumer<ToolExecution, InvocationContext>> compensatingActions = new HashMap<>();
|
||||
private Map<String, Consumer<ToolExecution>> findCompensatingActions(Object objectWithTools) {
|
||||
Map<String, Consumer<ToolExecution>> compensatingActions = new HashMap<>();
|
||||
if (compensatingToolMisconfiguration != null) {
|
||||
return compensatingActions;
|
||||
}
|
||||
|
|
@ -318,7 +320,7 @@ public class ToolService {
|
|||
if (acceptsToolExecution) {
|
||||
method.setAccessible(true);
|
||||
Method compensatingMethod = method;
|
||||
compensatingActions.put(toolName, (toolExecution, ctx) -> {
|
||||
compensatingActions.put(toolName, toolExecution -> {
|
||||
try {
|
||||
compensatingMethod.invoke(objectWithTools, toolExecution);
|
||||
} catch (Exception e) {
|
||||
|
|
@ -332,9 +334,8 @@ public class ToolService {
|
|||
.methodToInvoke(method)
|
||||
.propagateToolExecutionExceptions(true)
|
||||
.build();
|
||||
compensatingActions.put(
|
||||
toolName,
|
||||
(toolExecution, ctx) -> executor.executeWithContext(toolExecution.request(), ctx));
|
||||
compensatingActions.put(toolName, toolExecution ->
|
||||
executor.executeWithContext(toolExecution.request(), toolExecution.invocationContext()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -407,6 +408,17 @@ public class ToolService {
|
|||
return afterToolExecution;
|
||||
}
|
||||
|
||||
public void onCompensableToolExecution(BiConsumer<ToolExecution, Consumer<ToolExecution>> onCompensableToolExecution) {
|
||||
if (compensatingToolMisconfiguration != null) {
|
||||
throw compensatingToolMisconfiguration;
|
||||
}
|
||||
this.onCompensableToolExecution = onCompensableToolExecution;
|
||||
}
|
||||
|
||||
public void onToolExecutionError(Consumer<InvocationContext> onToolExecutionError) {
|
||||
this.onToolExecutionError = onToolExecutionError;
|
||||
}
|
||||
|
||||
/**
|
||||
* @since 1.4.0
|
||||
*/
|
||||
|
|
@ -595,8 +607,13 @@ public class ToolService {
|
|||
|
||||
fireToolExecutedEvent(invocationContext, request, toolExecution, context.eventListenerRegistrar);
|
||||
|
||||
if (!result.isError() && compensateOnToolErrors && compensatingExecutors.containsKey(request.name())) {
|
||||
compensableExecutions.add(new CompensableToolExecution(toolExecution, toolExecMsg));
|
||||
if (!result.isError() && compensatingExecutors.containsKey(request.name())) {
|
||||
if (compensateOnToolErrors) {
|
||||
compensableExecutions.add(new CompensableToolExecution(toolExecution, toolExecMsg));
|
||||
}
|
||||
if (onCompensableToolExecution != null) {
|
||||
onCompensableToolExecution.accept(toolExecution, compensatingExecutors.get(request.name()));
|
||||
}
|
||||
}
|
||||
|
||||
if (result.isError() && failedToolName == null) {
|
||||
|
|
@ -607,12 +624,16 @@ public class ToolService {
|
|||
}
|
||||
|
||||
if (anyToolErrored && compensableExecutions != null && !compensableExecutions.isEmpty()) {
|
||||
compensateToolsActions(compensableExecutions, invocationContext);
|
||||
compensateToolsActions(compensableExecutions);
|
||||
rewriteChatMemoryForCompensatedTools(messages, chatMemory, compensableExecutions, failedToolName);
|
||||
compensableExecutions.clear();
|
||||
rewriteCurrentResults(toolExecutionRequests, toolResults, resultMessages, failedToolName);
|
||||
}
|
||||
|
||||
if (anyToolErrored && onToolExecutionError != null) {
|
||||
onToolExecutionError.accept(invocationContext);
|
||||
}
|
||||
|
||||
for (ToolExecutionResultMessage resultMessage : resultMessages) {
|
||||
if (chatMemory != null) {
|
||||
chatMemory.add(resultMessage);
|
||||
|
|
@ -716,14 +737,13 @@ public class ToolService {
|
|||
|
||||
private record CompensableToolExecution(ToolExecution toolExecution, ToolExecutionResultMessage resultMessage) {}
|
||||
|
||||
private void compensateToolsActions(
|
||||
List<CompensableToolExecution> compensableExecutions, InvocationContext invocationContext) {
|
||||
private void compensateToolsActions(List<CompensableToolExecution> compensableExecutions) {
|
||||
for (int i = compensableExecutions.size() - 1; i >= 0; i--) {
|
||||
ToolExecution toolExecution = compensableExecutions.get(i).toolExecution();
|
||||
String toolName = toolExecution.request().name();
|
||||
BiConsumer<ToolExecution, InvocationContext> compensatingAction = compensatingExecutors.get(toolName);
|
||||
Consumer<ToolExecution> compensatingAction = compensatingExecutors.get(toolName);
|
||||
try {
|
||||
compensatingAction.accept(toolExecution, invocationContext);
|
||||
compensatingAction.accept(toolExecution);
|
||||
} catch (Exception e) {
|
||||
log.warn("Compensating action failed for tool '{}': {}", toolName, e.getMessage(), e);
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue