[FEATURE] Hopefully the model build will support custom httpClient

[nn200433]

Is your feature request related to a problem? Please describe.

I hope the community models can customize httpClientBuilder like the official ollama so that they can better customize http requests to support interrupted streaming responses.

希望社区的模型都能跟官方的 ollama 一样自定义 httpClientBuilder ，这样可以更好的自定义 http 请求，以支持中断流式响应。

Describe the solution you'd like

Currently, when sending a question to a large language model (like ChatGPT), I create an SseSession Manager for each request. I store the session ID and request ID in the thread context. Then, in the onPartialResponse callback, I check the interruption status based on these IDs to decide whether to continue sending the streaming response to the frontend.

While this method works for controlling interruption, it has several drawbacks:

1. Memory contamination issue: Langchain4j automatically manages conversation memory, and we don’t have full control over when it saves context. If the stream is interrupted halfway, some incomplete or irrelevant content might still be recorded.
2. Not an elegant approach: Even if we stop streaming to the frontend, Langchain4j still receives the full response from the model provider. This isn’t a clean or proper interruption.

Later, I discovered that Langchain4j allows customization of the HTTP client. Internally, it uses the method:

public abstract <T> CompletableFuture<HttpResponse<T>> sendAsync(HttpRequest request, BodyHandler<T> responseBodyHandler)

This gave me a cleaner solution:

• Before making the request, I bind the request ID with the corresponding CompletableFuture task;
• If the user chooses to interrupt the request, I simply cancel the async task. This stops the entire response stream cleanly — avoiding memory pollution and achieving true interruption behavior.

目前我在调用大模型（如 ChatGPT 等）时，每次提问都会创建一个 SseSession 管理器，并将会话 ID 和请求 ID 存储在线程上下文中。随后，在收到流式回复（onPartialResponse 方法）时，我会通过这些 ID 去判断是否中断输出，以决定是否继续将内容推送给前端。

这个做法虽然可以实现中断控制，但存在几个问题：

会话记忆污染：Langchain4j 的记忆机制是自动管理的，我们无法完全控制何时保存信息。如果在中断时仍接收了部分回复内容，可能会将无效或不完整的数据记录进记忆中。

实现不优雅：虽然我们在前端停止了输出，但底层 Langchain4j 实际上还是在接收来自模型提供商的完整响应流。这不是真正的“中断”。

后来我发现 Langchain4j 支持自定义 HTTP 客户端，而底层是通过 sendAsync 方法异步处理流式响应的。于是我找到了一个更优雅的方式：

在发起请求前，将请求 ID 与 CompletableFuture 任务进行绑定；

如果用户需要中断请求，直接取消这个异步任务即可，整个响应就会彻底终止，既不会污染记忆，也更符合流式处理的预期行为。

---

@Slf4j
public class InterruptibleJdkHttpClient implements HttpClient {

    private final java.net.http.HttpClient delegate;
    private final Duration readTimeout;
    
    /** 存储活跃的HTTP请求，key是requestId，value是CompletableFuture */
    private static final Map<String, CompletableFuture<?>> ACTIVE_REQUESTS = new ConcurrentHashMap<>();

    public InterruptibleJdkHttpClient(InterruptibleJdkHttpClientBuilder builder) {
        java.net.http.HttpClient.Builder httpClientBuilder =
                getOrDefault(builder.httpClientBuilder(), java.net.http.HttpClient::newBuilder);
        if (builder.connectTimeout() != null) {
            httpClientBuilder.connectTimeout(builder.connectTimeout());
        }
        this.delegate = httpClientBuilder.build();
        this.readTimeout = builder.readTimeout();
    }

    public static InterruptibleJdkHttpClientBuilder builder() {
        return new InterruptibleJdkHttpClientBuilder();
    }

    @Override
    public SuccessfulHttpResponse execute(HttpRequest request) throws HttpException {
        try {
            java.net.http.HttpRequest jdkRequest = toJdkRequest(request);

            java.net.http.HttpResponse<String> jdkResponse = delegate.send(jdkRequest, BodyHandlers.ofString());

            if (!isSuccessful(jdkResponse)) {
                throw new HttpException(jdkResponse.statusCode(), jdkResponse.body());
            }

            return fromJdkResponse(jdkResponse, jdkResponse.body());
        } catch (HttpTimeoutException e) {
            throw new TimeoutException(e);
        } catch (IOException | InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void execute(HttpRequest request, ServerSentEventParser parser, ServerSentEventListener listener) {
        // 从ThreadLocal获取请求信息
        String requestId = ThreadLocalUtil.get(RagConstants.LOCAL_THREAD_REQUEST_ID);
        String conversationId = ThreadLocalUtil.get(RagConstants.LOCAL_THREAD_CONVERSATION_ID);
        
        log.debug("开始流式HTTP请求: requestId={}, conversationId={}", requestId, conversationId);
        
        java.net.http.HttpRequest jdkRequest = toJdkRequest(request);

        CompletableFuture<Void> future = delegate.sendAsync(jdkRequest, BodyHandlers.ofInputStream())
                .thenAccept(jdkResponse -> {
                    // 在处理响应前再次检查是否被中断
                    if (isRequestInterrupted(requestId, conversationId)) {
                        log.info("HTTP响应处理前检测到中断: requestId={}", requestId);
                        return;
                    }

                    if (!isSuccessful(jdkResponse)) {
                        HttpException exception = new HttpException(jdkResponse.statusCode(), readBody(jdkResponse));
                        ignoringExceptions(() -> listener.onError(exception));
                        return;
                    }

                    SuccessfulHttpResponse response = fromJdkResponse(jdkResponse, null);
                    ignoringExceptions(() -> listener.onOpen(response));

                    try (InputStream inputStream = jdkResponse.body()) {
                        // 使用包装的listener进行流解析，支持中断检查
                        parseWithInterruptCheck(inputStream, parser, listener, requestId, conversationId);
                        ignoringExceptions(listener::onClose);
                    } catch (IOException e) {
                        if (!isRequestInterrupted(requestId, conversationId)) {
                            throw new RuntimeException(e);
                        }
                        log.info("HTTP流解析被中断: requestId={}", requestId);
                    }
                })
                .exceptionally(throwable -> {
                    if (throwable instanceof CancellationException) {
                        log.info("HTTP请求被取消: requestId={}", requestId);
                    } else if (throwable.getCause() instanceof HttpTimeoutException) {
                        ignoringExceptions(() -> listener.onError(new TimeoutException(throwable)));
                    } else {
                        ignoringExceptions(() -> listener.onError(throwable));
                    }
                    return null;
                })
                .whenComplete((result, throwable) -> {
                    // 请求完成后清理
                    ACTIVE_REQUESTS.remove(requestId);
                    log.debug("HTTP请求完成，清理缓存: requestId={}", requestId);
                });
        
        // 如果有requestId，则缓存这个future用于后续中断
        if (StrUtil.isNotBlank(requestId)) {
            ACTIVE_REQUESTS.put(requestId, future);
            log.debug("缓存HTTP请求用于中断控制: requestId={}", requestId);
        }
    }

    private java.net.http.HttpRequest toJdkRequest(HttpRequest request) {
        java.net.http.HttpRequest.Builder builder = java.net.http.HttpRequest.newBuilder()
                .uri(URI.create(request.url()));

        request.headers().forEach((name, values) -> {
            if (values != null) {
                values.forEach(value -> builder.header(name, value));
            }
        });

        BodyPublisher bodyPublisher;
        if (request.body() != null) {
            bodyPublisher = BodyPublishers.ofString(request.body());
        } else {
            bodyPublisher = BodyPublishers.noBody();
        }
        builder.method(request.method().name(), bodyPublisher);

        if (readTimeout != null) {
            builder.timeout(readTimeout);
        }

        return builder.build();
    }

    private static SuccessfulHttpResponse fromJdkResponse(java.net.http.HttpResponse<?> response, String body) {
        return SuccessfulHttpResponse.builder()
                .statusCode(response.statusCode())
                .headers(response.headers().map())
                .body(body)
                .build();
    }

    private static boolean isSuccessful(java.net.http.HttpResponse<?> response) {
        int statusCode = response.statusCode();
        return statusCode >= 200 && statusCode < 300;
    }

    private static String readBody(java.net.http.HttpResponse<InputStream> response) {
        try (InputStream inputStream = response.body();
             BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))) {
            return reader.lines().collect(joining(System.lineSeparator()));
        } catch (IOException e) {
            return "Cannot read error response body: " + e.getMessage();
        }
    }
    
    /**
     * 带中断检查的流解析
     */
    private void parseWithInterruptCheck(InputStream inputStream, ServerSentEventParser parser,
                                         ServerSentEventListener listener, String requestId, String conversationId){
        // 包装listener，在每次事件处理前检查中断状态
        ServerSentEventListener wrappedListener = new ServerSentEventListener() {
            @Override
            public void onOpen(SuccessfulHttpResponse response) {
                if (!isRequestInterrupted(requestId, conversationId)) {
                    listener.onOpen(response);
                }
            }

            @Override
            public void onEvent(ServerSentEvent event) {
                if (!isRequestInterrupted(requestId, conversationId)) {
                    listener.onEvent(event);
                } else {
                    log.debug("检测到中断，直接取消HTTP请求: requestId={}", requestId);
                    // 直接取消HTTP请求，更加优雅和高效
                    cancelCurrentRequest(requestId);
                }
            }

            @Override
            public void onError(Throwable throwable) {
                if (!isRequestInterrupted(requestId, conversationId)) {
                    listener.onError(throwable);
                }
            }

            @Override
            public void onClose() {
                if (!isRequestInterrupted(requestId, conversationId)) {
                    listener.onClose();
                }
            }
        };
        
        parser.parse(inputStream, wrappedListener);
    }
    
    /**
     * 检查请求是否被中断
     */
    private boolean isRequestInterrupted(String requestId, String conversationId) {
        if (StrUtil.isBlank(requestId) || StrUtil.isBlank(conversationId)) {
            return false;
        }
        
        // 检查会话状态管理器
        boolean shouldContinue = ConversationStateManager.shouldContinueProcessing(conversationId, requestId);
        if (!shouldContinue) {
            log.debug("会话状态管理器指示应该中断: requestId={}, conversationId={}", requestId, conversationId);
            return true;
        }
        
        // 检查ThreadLocal中断标记
        Boolean isInterrupted = ThreadLocalUtil.get(RagConstants.LOCAL_THREAD_INTERRUPT_FLAG);
        if (Boolean.TRUE.equals(isInterrupted)) {
            log.debug("ThreadLocal中断标记为true: requestId={}", requestId);
            return true;
        }
        
        return false;
    }
    
    /**
     * 取消当前HTTP请求
     */
    private void cancelCurrentRequest(String requestId) {
        if (StrUtil.isBlank(requestId)) {
            return;
        }
        
        CompletableFuture<?> future = ACTIVE_REQUESTS.get(requestId);
        if (future != null && !future.isDone()) {
            boolean cancelled = future.cancel(true);
            if (cancelled) {
                ACTIVE_REQUESTS.remove(requestId);
                log.info("成功取消HTTP请求: requestId={}", requestId);
            } else {
                log.debug("请求已完成，无法取消: requestId={}", requestId);
            }
        } else {
            log.debug("未找到活跃的HTTP请求: requestId={}", requestId);
        }
    }

}

Additional context

Functions in development

System Environment

JDK: 17
Spring Boot: 3.4.3
LangChain4j: 1.2.0
OS: Win 11

[Martin7-1]

@nn200433 Hi! Thanks for you feature request. I'm wondering which model provider do you use now? Because not all model providers are accessed through HTTP (for example, langchain4j-community-dashscope is accessed through SDK)

[nn200433]

@nn200433 Hi! Thanks for you feature request. I'm wondering which model provider do you use now? Because not all model providers are accessed through HTTP (for example, langchain4j-community-dashscope is accessed through SDK)

Thank you for your reply. Currently, I mainly use locally deployed Ollama and Xinference.

[codexvn]

I have encountered the same issue. When the user triggers the stop generation action, I hope there can be a mechanism to proactively close the connection to the model service, thereby notifying it to release resources. Given the significant resource consumption of large models, merely ignoring the model’s output at the application layer without actually closing the connection is a waste of resources.

[nn200433]

I have encountered the same issue. When the user triggers the stop generation action, I hope there can be a mechanism to proactively close the connection to the model service, thereby notifying it to release resources. Given the significant resource consumption of large models, merely ignoring the model’s output at the application layer without actually closing the connection is a waste of resources.

It's like this, xinference actually provides an official interface to turn off the output, but for the sake of operational generality, it's more reliable to break the link.

是这样子，xinference官方其实提供了关闭输出的接口，但是为了操作的通用性，还是中断链接比较靠谱。

[codexvn]

Especially when implementing Human-In-The-Loop, since users may provide new change requests at any time, the ability to stop at any moment is crucial.

[Loahrs]

Being able to easily interrupt response generation would be great. Thank you for providing your workaround implementation of an InteruptibleJdkHttpClient.

If I may ask:
What did you do to make the cancelCurrentRequest-Method callable from the UI? OllamaStreamingChatModelBuilder only takes an HttpClientBuilder, so we can't just create the HttpClient outside of it and keep a reference. Have you also implemented a InteruptibleOllamaStreamingChatModel or similar?

[nn200433]

Being able to easily interrupt response generation would be great. Thank you for providing your workaround implementation of an InteruptibleJdkHttpClient.

If I may ask: What did you do to make the cancelCurrentRequest-Method callable from the UI? OllamaStreamingChatModelBuilder only takes an HttpClientBuilder, so we can't just create the HttpClient outside of it and keep a reference. Have you also implemented a InteruptibleOllamaStreamingChatModel or similar?

English

You can simply implement a custom HttpClient , where the interrupt flag is retrieved from thread-local variables. For example, in my InterruptibleJdkHttpClient implementation:

// 从 ThreadLocal 获取请求信息
String requestId = ThreadLocalUtil.get(RagConstants.LOCAL_THREAD_REQUEST_ID);
String conversationId = ThreadLocalUtil.get(RagConstants.LOCAL_THREAD_CONVERSATION_ID);

In my approach, whenever the user sends a question, a unique request ID (requestId) and conversation ID (conversationId) are generated. These two IDs are then used to create a unique ConversationStateManager, initialize the conversation’s interrupt state as isInterrupted = false, and store the IDs in ThreadLocal for retrieval by InterruptibleJdkHttpClient. This way, as long as the frontend can obtain these two IDs and send them with an interrupt request to the backend, the backend can update the interrupt state and stop the execute method of InterruptibleJdkHttpClient.

chinese

实现一个自定义的 HttpClient 即可。中断标识可以通过线程变量获取，例如我在 InterruptibleJdkHttpClient 中的实现：

// 从 ThreadLocal 获取请求信息
String requestId = ThreadLocalUtil.get(RagConstants.LOCAL_THREAD_REQUEST_ID);
String conversationId = ThreadLocalUtil.get(RagConstants.LOCAL_THREAD_CONVERSATION_ID);

我的做法是：每次用户发送问题时，会生成一个唯一的请求 ID (requestId) 和会话 ID (conversationId)，并基于这两个 ID 创建一个唯一的会话管理器 ConversationStateManager，将会话的中断状态初始化为 isInterrupted = false ，同时将这两个 ID 存入 ThreadLocal，供 InterruptibleJdkHttpClient 获取。这样，只要前端能够获取到这两个 ID，并在发起中断请求时将它们传给后端，后端就可以修改中断状态，从而终止 InterruptibleJdkHttpClient 的 execute 方法。

Code Example

The following code is assisted by chatgpt.

requestId can be returned within the SSE stream, allowing the frontend to retrieve and store it for use when terminating the stream.

SseSessionManager.send(userId, requestId, SseEmitter.event()
                    .comment("AI Think")
                    .data(ChatResp.builder()
                                  .conversation_id(conversationId)
                                  .msg_id(assistantMsgId)
                                  // this
                                  .req_id(requestId)
                                  .answer(response)
                                  .is_completion(Boolean.FALSE)
                                  .dataType(AnswerDataTypeEnum.THINK)
                                  .is_biz(Boolean.FALSE)
                                  .hit(Boolean.FALSE)
                                  .build()));

ConversationStateManager

/**
 * 会话状态管理器 用于管理每个请求的独立状态，防止中断请求污染后续请求
 *
 * @date 2025-08-01
 */
@Slf4j
public class ConversationStateManager {

    /** 会话状态映射：conversationId_requestId -> ConversationState */
    private static final ConcurrentHashMap<String, ConversationState> stateMap           = new ConcurrentHashMap<>();
    /** 清理超时时间（10分钟） */
    private static final long                                         CLEANUP_TIMEOUT_MS = 10 * 60 * 1000;

    /**
     * 生成会话状态键
     */
    private static String generateStateKey(String conversationId, String requestId) {
        return conversationId + StrPool.UNDERLINE + requestId;
    }

    /**
     * 创建或获取会话状态
     */
    public static ConversationState createConversationState(String conversationId, String requestId) {
        String key = generateStateKey(conversationId, requestId);
        return stateMap.computeIfAbsent(key, k -> {
            log.debug("创建会话状态: conversationId={}, requestId={}", conversationId, requestId);
            return new ConversationState(conversationId, requestId);
        });
    }

    /**
     * 获取会话状态
     */
    public static ConversationState getConversationState(String conversationId, String requestId) {
        String key = generateStateKey(conversationId, requestId);
        return stateMap.get(key);
    }

    /**
     * 标记请求完成
     */
    public static void markCompleted(String conversationId, String requestId) {
        ConversationState state = getConversationState(conversationId, requestId);
        if (state != null) {
            state.markCompleted();
            log.debug("标记请求完成: conversationId={}, requestId={}", conversationId, requestId);
        }
    }

    /**
     * 标记请求中断
     */
    public static void markInterrupted(String conversationId, String requestId) {
        ConversationState state = getConversationState(conversationId, requestId);
        if (state != null) {
            state.markInterrupted();
            log.debug("标记请求中断: conversationId={}, requestId={}", conversationId, requestId);
        }
    }

    /**
     * 检查请求是否应该继续处理
     */
    public static boolean shouldContinueProcessing(String conversationId, String requestId) {
        ConversationState state = getConversationState(conversationId, requestId);
        return state != null && state.isValid();
    }

    /**
     * 检查是否有未完成的请求（用于判断是否需要清理会话记忆）
     */
    public static boolean hasIncompleteRequests(String conversationId) {
        return stateMap.values().stream()
                .filter(state -> conversationId.equals(state.getConversationId()))
                .anyMatch(state -> !state.isCompleted());
    }

    /**
     * 获取会话下最新的有效请求ID
     */
    public static String getLatestValidRequestId(String conversationId) {
        return stateMap.values().stream()
                .filter(state -> conversationId.equals(state.getConversationId()))
                .filter(ConversationState::isValid)
                .max((s1, s2) -> Long.compare(s1.getCreateTime(), s2.getCreateTime()))
                .map(ConversationState::getRequestId)
                .orElse(null);
    }

    /**
     * 移除特定请求的状态
     *
     * @param conversationId 会话 ID
     * @param requestId      请求 ID
     */
    public static void removeRequestState(String conversationId, String requestId) {
        String            key     = generateStateKey(conversationId, requestId);
        ConversationState removed = stateMap.remove(key);
        if (removed != null) {
            log.debug("移除请求状态: conversationId={}, requestId={}", conversationId, requestId);
        }
    }

    /**
     * 清理指定会话的所有状态
     */
    public static void cleanupConversation(String conversationId) {
        stateMap.entrySet().removeIf(entry -> {
            ConversationState state = entry.getValue();
            if (conversationId.equals(state.getConversationId())) {
                log.debug("清理会话状态: conversationId={}, requestId={}",
                          state.getConversationId(), state.getRequestId());
                return true;
            }
            return false;
        });
    }

    /**
     * 清理超时的状态
     */
    public static void cleanupExpiredStates() {
        long currentTime = System.currentTimeMillis();
        stateMap.entrySet().removeIf(entry -> {
            ConversationState state     = entry.getValue();
            boolean           isExpired = (currentTime - state.getLastActiveTime()) > CLEANUP_TIMEOUT_MS;
            if (isExpired) {
                log.debug("清理超时状态: conversationId={}, requestId={}, 超时时间={}ms", state.getConversationId(), state.getRequestId(), currentTime - state.getLastActiveTime());
            }
            return isExpired;
        });
    }

    /**
     * 获取状态统计信息
     */
    public static String getStateStats() {
        int  total       = stateMap.size();
        long completed   = stateMap.values().stream().mapToLong(s -> s.isCompleted() ? 1 : 0).sum();
        long interrupted = stateMap.values().stream().mapToLong(s -> s.isInterrupted() ? 1 : 0).sum();
        long active      = total - completed - interrupted;
        return StrUtil.format("ConversationState统计: 总计={}, 活跃={}, 完成={}, 中断={}", total, active, completed, interrupted);
    }

}

ConversationState

/**
 * 会话状态信息
 */
public class ConversationState {

    /** 请求是否完成 */
    private final AtomicBoolean         isCompleted    = new AtomicBoolean(false);
    /** 请求是否被中断 */
    private final AtomicBoolean         isInterrupted  = new AtomicBoolean(false);
    /** 会话ID */
    private final String                conversationId;
    /** 请求ID */
    private final String                requestId;
    /** 创建时间 */
    private final long                  createTime;
    /** 最后活跃时间 */
    private final AtomicReference<Long> lastActiveTime = new AtomicReference<>();

    public ConversationState(String conversationId, String requestId) {
        this.conversationId = conversationId;
        this.requestId      = requestId;
        this.createTime     = System.currentTimeMillis();
        this.lastActiveTime.set(createTime);
    }

    public boolean isCompleted() {
        return isCompleted.get();
    }

    public void markCompleted() {
        isCompleted.set(true);
        updateLastActiveTime();
    }

    /**
     * 是否中断
     *
     * @return boolean
     */
    public boolean isInterrupted() {
        return isInterrupted.get();
    }

    /**
     * 标记中断
     */
    public void markInterrupted() {
        isInterrupted.set(true);
        updateLastActiveTime();
    }

    /**
     * 更新上次活动时间
     */
    public void updateLastActiveTime() {
        lastActiveTime.set(System.currentTimeMillis());
    }

    /**
     * 获取每次聊天的唯一会话主键
     *
     * @return {@link String }
     */
    public String getConversationId() {
        return conversationId;
    }

    /**
     * 获取请求主键
     *
     * @return {@link String }
     */
    public String getRequestId() {
        return requestId;
    }

    /**
     * 获取创建时间
     *
     * @return long
     */
    public long getCreateTime() {
        return createTime;
    }

    /**
     * 获取最后活动时间
     *
     * @return long
     */
    public long getLastActiveTime() {
        return lastActiveTime.get();
    }

    /**
     * 检查状态是否有效（未完成且未中断）
     */
    public boolean isValid() {
        return !isCompleted() && !isInterrupted();
    }

}

AgentController

/**
 * 与大模型聊天
 *
 * @param req 请求
 * @return {@link Object }
 */
@PostMapping("/chat/completions")
public Object chat(@RequestBody ChatReq req) {
    final Boolean isStream = Convert.toBool(req.getIsStream(), Boolean.FALSE);
    // 设置请求id
    req.setRequestId(IdUtil.getSnowflakeNextIdStr());
    if (isStream) {
        final String userId      = req.getUserId();
        final String requestId = req.getRequestId();
        try {
            // 1. 创建会话状态 及 sse会话管理器 
            ConversationStateManager.createConversationState(req.getConversationId(), requestId);
            SseSessionManager.createSession(userId, requestId, req.getConversationId());
            // 2. 请求 LLM 部署平台
            return SseSessionManager.streaming(userId, requestId, () -> chatService.chat(req));
        } catch (Exception e) {
            log.warn("SSE流式请求异常: {}", e.getMessage(), e);
            SseSession sseSession = SseSessionManager.getSession(userId, requestId);
            if (sseSession != null) {
                SseSessionManager.error(userId, requestId, e.getMessage());
                return sseSession.getEmitter();
            } else {
                // 可以返回一个立即关闭的SseEmitter，也可以直接返回null前端自行处理
                return SseSessionManager.buildSingleNewError(req.getConversationId(), requestId, e.getMessage());
            }
        }
    } else {
        return chatService.text(req);
    }
}

/**
 * 终止sse流
 *
 * @param userId 用户主键
 * @return {@link Result }<{@link String }>
 */
@PostMapping("/sse/stop/{userId}")
public Result<String> stopStream(@PathVariable String userId, @RequestParam(value = "requestId") String requestId) {

    // You can obtain requestId and conversationId from SseSessionManager
    // 1. Immediately set the session output interrupt flag
    // 2. Clear the sending queue to prevent outdated data from being sent
    // 3. Set the completion flag so the sending thread can finish as soon as possible
    // 4. Immediately and synchronously clear the session memory to prevent LangChain4j from writing dirty data
    // 5. Mark the request as interrupted, i.e., call the `ConversationStateManager#markCompleted` method
    // ...
    SseSessionManager.setContinueFlag(userId, requestId, false);
    return Result.ofSuccess("终止成功");

}