feat(harness): add chat application use case
This commit is contained in:
zhuyongxin
@@ -0,0 +1,259 @@
|
||||
package com.superbiz.agent.harness.application;
|
||||
|
||||
import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.superbiz.agent.harness.application.persistence.ChatRunStore;
|
||||
import com.superbiz.agent.harness.application.persistence.RoutingHistory;
|
||||
import com.superbiz.agent.harness.application.routing.IntentRouterInput;
|
||||
import com.superbiz.agent.harness.application.routing.IntentRoutingException;
|
||||
import com.superbiz.agent.harness.contract.IntentType;
|
||||
import com.superbiz.agent.harness.contract.PreviousTurn;
|
||||
import com.superbiz.agent.harness.contract.PublishedResult;
|
||||
import com.superbiz.agent.harness.contract.ReleaseOutcome;
|
||||
import com.superbiz.agent.harness.core.DiagnosisHarnessCore;
|
||||
import com.superbiz.agent.harness.core.RunCancellationReason;
|
||||
import com.superbiz.agent.harness.core.RunContext;
|
||||
import com.superbiz.agent.harness.core.RunState;
|
||||
import com.superbiz.agent.harness.retry.RetryExecutionException;
|
||||
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
import java.util.function.Supplier;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
public final class ChatApplicationUseCase {
|
||||
|
||||
private static final Pattern SAFE_ID = Pattern.compile("[A-Za-z0-9][A-Za-z0-9._-]{0,63}");
|
||||
|
||||
private final DiagnosisHarnessCore core;
|
||||
private final Supplier<String> sessionIdSupplier;
|
||||
private final ChatRunStore runStore;
|
||||
private final IntentRouting router;
|
||||
private final SystemChatOperation systemChat;
|
||||
private final KnowledgeQueryOperation knowledgeQuery;
|
||||
private final DiagnosisOperation diagnosis;
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
public ChatApplicationUseCase(DiagnosisHarnessCore core,
|
||||
Supplier<String> sessionIdSupplier,
|
||||
ChatRunStore runStore,
|
||||
IntentRouting router,
|
||||
SystemChatOperation systemChat,
|
||||
KnowledgeQueryOperation knowledgeQuery,
|
||||
DiagnosisOperation diagnosis,
|
||||
ObjectMapper objectMapper) {
|
||||
this.core = Objects.requireNonNull(core, "core must not be null");
|
||||
this.sessionIdSupplier = Objects.requireNonNull(
|
||||
sessionIdSupplier, "sessionIdSupplier must not be null");
|
||||
this.runStore = Objects.requireNonNull(runStore, "runStore must not be null");
|
||||
this.router = Objects.requireNonNull(router, "router must not be null");
|
||||
this.systemChat = Objects.requireNonNull(systemChat, "systemChat must not be null");
|
||||
this.knowledgeQuery = Objects.requireNonNull(knowledgeQuery, "knowledgeQuery must not be null");
|
||||
this.diagnosis = Objects.requireNonNull(diagnosis, "diagnosis must not be null");
|
||||
this.objectMapper = Objects.requireNonNull(objectMapper, "objectMapper must not be null");
|
||||
}
|
||||
|
||||
public ChatApplicationResult execute(ChatApplicationRequest request) {
|
||||
return execute(request, ChatApplicationObserver.noop());
|
||||
}
|
||||
|
||||
public ChatApplicationResult execute(ChatApplicationRequest request,
|
||||
ChatApplicationObserver observer) {
|
||||
Objects.requireNonNull(request, "request must not be null");
|
||||
Objects.requireNonNull(observer, "observer must not be null");
|
||||
String sessionId = resolveSessionId(request.sessionId());
|
||||
|
||||
Optional<RoutingHistory> history;
|
||||
Optional<PreviousTurn> previousTurn;
|
||||
try {
|
||||
history = runStore.findLatestRoutingHistory(sessionId);
|
||||
previousTurn = runStore.findPreviousTurn(sessionId);
|
||||
} catch (RuntimeException exception) {
|
||||
throw new ChatApplicationException(
|
||||
ChatFailureCode.RUN_PERSISTENCE_FAILED,
|
||||
"无法读取会话上下文，请稍后重试", exception);
|
||||
}
|
||||
RunContext context = core.startRun(sessionId);
|
||||
long startedNanos = System.nanoTime();
|
||||
IntentType intent = null;
|
||||
try {
|
||||
persistStart(context, request.query());
|
||||
observer.onStarted(new CoreRunControl(core, context));
|
||||
observer.onStatus(ChatApplicationStatus.ROUTING);
|
||||
intent = router.route(context, new IntentRouterInput(
|
||||
request.query(),
|
||||
history.map(RoutingHistory::intent).orElse(null),
|
||||
history.map(RoutingHistory::userQuery).orElse(null)));
|
||||
persistIntent(context.runId(), intent);
|
||||
|
||||
PathResult path = executePath(
|
||||
intent, context, request.query(), previousTurn.orElse(null), observer);
|
||||
core.checkActive(context);
|
||||
core.completeSuccess(context);
|
||||
String safeJson = write(path.content());
|
||||
persistFinish(context, intent, path.outcome(), safeJson,
|
||||
path.publishedResult(), durationMillis(startedNanos));
|
||||
return new ChatApplicationResult(
|
||||
context.sessionId(), context.runId(), intent, path.outcome(),
|
||||
path.content().contentType(), path.content());
|
||||
} catch (RuntimeException exception) {
|
||||
ReleaseOutcome terminal = terminalOutcome(context);
|
||||
try {
|
||||
runStore.finish(context, intent, terminal, null, null,
|
||||
durationMillis(startedNanos));
|
||||
} catch (RuntimeException persistenceFailure) {
|
||||
exception.addSuppressed(persistenceFailure);
|
||||
}
|
||||
throw safeFailure(exception, terminal);
|
||||
}
|
||||
}
|
||||
|
||||
private PathResult executePath(IntentType intent,
|
||||
RunContext context,
|
||||
String query,
|
||||
PreviousTurn previousTurn,
|
||||
ChatApplicationObserver observer) {
|
||||
return switch (intent) {
|
||||
case SYSTEM_CHAT -> {
|
||||
observer.onStatus(ChatApplicationStatus.SYSTEM_RESPONDING);
|
||||
yield new PathResult(
|
||||
ReleaseOutcome.SUCCESS, systemChat.execute(context, query), null);
|
||||
}
|
||||
case KNOWLEDGE_QUERY -> {
|
||||
observer.onStatus(ChatApplicationStatus.KNOWLEDGE_SEARCHING);
|
||||
observer.onStatus(ChatApplicationStatus.KNOWLEDGE_ANSWERING);
|
||||
yield new PathResult(
|
||||
ReleaseOutcome.SUCCESS, knowledgeQuery.execute(context, query), null);
|
||||
}
|
||||
case DIAGNOSIS -> {
|
||||
DiagnosisExecutionResult result = diagnosis.execute(
|
||||
context, query, previousTurn, observer::onStatus);
|
||||
yield new PathResult(result.outcome(), result.content(), result.publishedResult());
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private void persistStart(RunContext context, String query) {
|
||||
try {
|
||||
runStore.start(context, query);
|
||||
} catch (RuntimeException exception) {
|
||||
throw new ChatApplicationException(
|
||||
ChatFailureCode.RUN_PERSISTENCE_FAILED,
|
||||
"无法创建诊断运行记录，请稍后重试", exception);
|
||||
}
|
||||
}
|
||||
|
||||
private void persistIntent(String runId, IntentType intent) {
|
||||
try {
|
||||
runStore.markIntent(runId, intent);
|
||||
} catch (RuntimeException exception) {
|
||||
throw new ChatApplicationException(
|
||||
ChatFailureCode.RUN_PERSISTENCE_FAILED,
|
||||
"无法记录请求类型，请稍后重试", exception);
|
||||
}
|
||||
}
|
||||
|
||||
private void persistFinish(RunContext context,
|
||||
IntentType intent,
|
||||
ReleaseOutcome outcome,
|
||||
String safeJson,
|
||||
PublishedResult publishedResult,
|
||||
int durationMs) {
|
||||
try {
|
||||
runStore.finish(context, intent, outcome, safeJson, publishedResult, durationMs);
|
||||
} catch (RuntimeException exception) {
|
||||
throw new ChatApplicationException(
|
||||
ChatFailureCode.RUN_PERSISTENCE_FAILED,
|
||||
"无法记录运行终态，请稍后重试", exception);
|
||||
}
|
||||
}
|
||||
|
||||
private ReleaseOutcome terminalOutcome(RunContext context) {
|
||||
RunState state = context.lifecycle().state();
|
||||
if (state == RunState.CANCELLED) {
|
||||
return ReleaseOutcome.CANCELLED;
|
||||
}
|
||||
if (!state.isTerminal()) {
|
||||
core.completeFailure(context, "APPLICATION_EXECUTION_FAILED");
|
||||
}
|
||||
return context.lifecycle().state() == RunState.CANCELLED
|
||||
? ReleaseOutcome.CANCELLED : ReleaseOutcome.FAILED;
|
||||
}
|
||||
|
||||
private ChatApplicationException safeFailure(RuntimeException exception,
|
||||
ReleaseOutcome terminal) {
|
||||
if (terminal == ReleaseOutcome.CANCELLED) {
|
||||
return new ChatApplicationException(
|
||||
ChatFailureCode.RUN_CANCELLED, "Chat Run was cancelled", exception);
|
||||
}
|
||||
if (exception instanceof IntentRoutingException) {
|
||||
return new ChatApplicationException(
|
||||
ChatFailureCode.ROUTING_UNAVAILABLE,
|
||||
"当前暂时无法识别请求类型，请稍后重试", exception);
|
||||
}
|
||||
if (exception instanceof ChatApplicationException applicationFailure) {
|
||||
return applicationFailure;
|
||||
}
|
||||
if (exception instanceof RetryExecutionException retry
|
||||
&& retry.failure() == com.superbiz.agent.harness.retry.RetryFailure.CANCELLED) {
|
||||
return new ChatApplicationException(
|
||||
ChatFailureCode.RUN_CANCELLED, "Chat Run was cancelled", exception);
|
||||
}
|
||||
return new ChatApplicationException(
|
||||
ChatFailureCode.INTERNAL_FAILURE, "当前暂时无法处理该请求，请稍后重试", exception);
|
||||
}
|
||||
|
||||
private String resolveSessionId(String requested) {
|
||||
String value = requested == null ? sessionIdSupplier.get() : requested;
|
||||
if (value == null || !SAFE_ID.matcher(value).matches()) {
|
||||
throw new IllegalArgumentException("sessionId is invalid");
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
private String write(ChatApplicationContent content) {
|
||||
try {
|
||||
return objectMapper.writeValueAsString(content);
|
||||
} catch (JsonProcessingException exception) {
|
||||
throw new ChatApplicationException(
|
||||
ChatFailureCode.INTERNAL_FAILURE, "Public content is not serializable", exception);
|
||||
}
|
||||
}
|
||||
|
||||
private static int durationMillis(long startedNanos) {
|
||||
long millis = Math.max(0L, (System.nanoTime() - startedNanos) / 1_000_000L);
|
||||
return millis >= Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) millis;
|
||||
}
|
||||
|
||||
private record PathResult(
|
||||
ReleaseOutcome outcome,
|
||||
ChatApplicationContent content,
|
||||
PublishedResult publishedResult) {
|
||||
}
|
||||
|
||||
private static final class CoreRunControl implements ChatRunControl {
|
||||
|
||||
private final DiagnosisHarnessCore core;
|
||||
private final RunContext context;
|
||||
|
||||
private CoreRunControl(DiagnosisHarnessCore core, RunContext context) {
|
||||
this.core = core;
|
||||
this.context = context;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String sessionId() {
|
||||
return context.sessionId();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String runId() {
|
||||
return context.runId();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean cancelClientDisconnect() {
|
||||
return core.cancel(context, RunCancellationReason.CLIENT_DISCONNECTED);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user