ReAct、规划与反思ReAct、规划与反思

import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;

public final class BoundedAgentOrchestrator {
    enum RunStatus {
        READY, RUNNING, WAITING_APPROVAL, SUCCEEDED, FAILED, CANCELLED
    }

    record Action(String effectId, String tool, String resourceId, boolean sideEffect) {
        Action {
            Objects.requireNonNull(effectId);
            Objects.requireNonNull(tool);
            Objects.requireNonNull(resourceId);
        }
    }

    sealed interface Decision permits Invoke, Finish, Abort {}
    record Invoke(Action action) implements Decision {}
    record Finish(String answer) implements Decision {}
    record Abort(String reason) implements Decision {}
    record PendingApproval(Action action, String operationId) {}
    record ToolExecution(Action action, Observation observation) {}

    record Observation(
            boolean success,
            boolean retryable,
            boolean outcomeKnown,
            String payload) {
        Observation {
            Objects.requireNonNull(payload);
        }
    }

    interface Planner {
        Decision next(String goal, List<Observation> observations, int remainingSteps);
    }

    interface ToolGateway {
        Observation invoke(Action action, String operationId);
        Optional<Observation> findResult(String operationId);
    }

    interface ApprovalPolicy {
        boolean approved(String runId, PendingApproval pending);
    }

    static final class Run {
        private final String runId;
        private final String goal;
        private final Instant deadline;
        private final int maxSteps;
        private final List<Observation> observations = new ArrayList<>();
        private final Map<String, Observation> completedOperations = new HashMap<>();
        private RunStatus status = RunStatus.READY;
        private PendingApproval pendingApproval;
        private int step;
        private String answer;

        Run(String runId, String goal, Instant deadline, int maxSteps) {
            this.runId = Objects.requireNonNull(runId);
            this.goal = Objects.requireNonNull(goal);
            this.deadline = Objects.requireNonNull(deadline);
            if (maxSteps <= 0) throw new IllegalArgumentException("maxSteps must be positive");
            this.maxSteps = maxSteps;
        }
    }

    private final Planner planner;
    private final ToolGateway tools;
    private final ApprovalPolicy approvals;
    private final Clock clock;

    public BoundedAgentOrchestrator(
            Planner planner, ToolGateway tools, ApprovalPolicy approvals, Clock clock) {
        this.planner = Objects.requireNonNull(planner);
        this.tools = Objects.requireNonNull(tools);
        this.approvals = Objects.requireNonNull(approvals);
        this.clock = Objects.requireNonNull(clock);
    }

    public Run execute(Run run) {
        requireState(run, RunStatus.READY, RunStatus.RUNNING);
        run.status = RunStatus.RUNNING;

        while (run.status == RunStatus.RUNNING) {
            if (!clock.instant().isBefore(run.deadline)) {
                fail(run, "deadline exceeded");
                break;
            }
            if (run.step >= run.maxSteps) {
                fail(run, "step budget exceeded");
                break;
            }

            int remaining = run.maxSteps - run.step;
            Decision decision = planner.next(run.goal, List.copyOf(run.observations), remaining);
            run.step++;

            switch (decision) {
                case Finish finish -> {
                    run.answer = finish.answer();
                    run.status = RunStatus.SUCCEEDED;
                }
                case Abort abort -> fail(run, abort.reason());
                case Invoke invoke -> executeAction(run, invoke.action());
            }
        }
        return run;
    }

    public Run resumeAfterApproval(Run run) {
        requireState(run, RunStatus.WAITING_APPROVAL);
        PendingApproval pending = Objects.requireNonNull(run.pendingApproval);
        if (!approvals.approved(run.runId, pending)) return run;
        // 审批等待期间 run 可能已过绝对截止时间；必须在副作用前复核。
        if (!clock.instant().isBefore(run.deadline)) {
            fail(run, "deadline exceeded while waiting for approval");
            return run;
        }
        if (run.step > run.maxSteps) {
            fail(run, "step budget exceeded while waiting for approval");
            return run;
        }

        run.status = RunStatus.RUNNING;
        run.pendingApproval = null;
        executeAction(run, pending.action());
        return run.status == RunStatus.RUNNING ? execute(run) : run;
    }

    private void executeAction(Run run, Action action) {
        requireState(run, RunStatus.RUNNING);
        if (!clock.instant().isBefore(run.deadline)) {
            fail(run, "deadline exceeded before tool invocation");
            return;
        }
        // 同一逻辑副作用的全部重试都复用稳定 effectId；新业务动作才分配新 effectId。
        String operationId = run.runId + ":effect:" + action.effectId();
        PendingApproval pending = new PendingApproval(action, operationId);
        if (action.sideEffect() && !approvals.approved(run.runId, pending)) {
            run.pendingApproval = pending;
            run.status = RunStatus.WAITING_APPROVAL;
            return;
        }

        Observation cached = run.completedOperations.get(operationId);
        if (cached != null) {
            run.observations.add(cached);
            return;
        }

        Observation result = tools.invoke(action, operationId);
        if (!result.outcomeKnown()) {
            result = tools.findResult(operationId).orElse(result);
            if (!result.outcomeKnown()) {
                fail(run, "side-effect outcome is unknown; manual reconciliation required");
                return;
            }
        }

        if (result.success()) {
            run.completedOperations.put(operationId, result);
            run.observations.add(result);
        } else if (result.retryable()) {
            // 持久化 nextAttemptAt 后异步退避；重试仍使用 action.effectId() 对应的 operationId。
            run.observations.add(result);
        } else {
            fail(run, result.payload());
        }
    }

    private static void fail(Run run, String reason) {
        run.answer = reason;
        run.status = RunStatus.FAILED;
    }

    private static void requireState(Run run, RunStatus... allowed) {
        for (RunStatus status : allowed) {
            if (run.status == status) return;
        }
        throw new IllegalStateException("cannot execute run in state " + run.status);
    }

    public static void main(String[] args) {
        Planner planner = (goal, observations, remaining) ->
                observations.isEmpty()
                        ? new Invoke(new Action("query-order", "order.query", "order-42", false))
                        : new Finish("已基于订单系统返回结果完成处理");

        ToolGateway gateway = new ToolGateway() {
            private final Map<String, ToolExecution> results = new HashMap<>();

            @Override
            public Observation invoke(Action action, String operationId) {
                ToolExecution existing = results.get(operationId);
                if (existing != null) {
                    if (!existing.action().equals(action)) {
                        throw new IllegalArgumentException(
                                "operationId reused with different action");
                    }
                    return existing.observation();
                }
                Observation result = new Observation(true, false, true, "orderStatus=PAID");
                results.put(operationId, new ToolExecution(action, result));
                return result;
            }

            @Override
            public Optional<Observation> findResult(String operationId) {
                return Optional.ofNullable(results.get(operationId))
                        .map(ToolExecution::observation);
            }
        };

        var orchestrator = new BoundedAgentOrchestrator(
                planner, gateway,
                (runId, pending) -> !pending.action().sideEffect(), Clock.systemUTC());
        var run = new Run("run-1001", "查询订单状态", Instant.now().plus(Duration.ofSeconds(5)), 4);
        Run completed = orchestrator.execute(run);
        System.out.println(completed.status + ": " + completed.answer);
    }
}