From 45537ab33962ef68ecd5eac70cf3dc6027de99db Mon Sep 17 00:00:00 2001 From: luke Date: Mon, 23 Jun 2025 16:12:24 +0800 Subject: [PATCH] opt --- .../base/application/filter/AuthFilter.java | 2 +- .../application/service/LLMAppService.java | 4 +- .../service/LLMAppServiceImpl.java | 39 ++++------- .../cache/iface/UserCacheService.java | 5 ++ .../cache/impl/RedisUserCacheServiceImpl.java | 11 ++++ .../infrastructure/config/ConstantConfig.java | 8 +++ .../north/controller/LLMController.java | 10 +-- .../south/llm/AnythingLLMServiceImpl.java | 65 ++++++++++++------- .../infrastructure/south/llm/LLMService.java | 5 +- .../base/infrastructure/util/HttpHelper.java | 18 +++-- .../util/http/SseOutputAdapter.java | 22 +++++++ .../util/http/WriterAdapter.java | 8 +++ .../application-dev-windows.properties | 3 +- 13 files changed, 136 insertions(+), 64 deletions(-) create mode 100644 src/main/java/com/knowledge/base/infrastructure/util/http/SseOutputAdapter.java create mode 100644 src/main/java/com/knowledge/base/infrastructure/util/http/WriterAdapter.java diff --git a/src/main/java/com/knowledge/base/application/filter/AuthFilter.java b/src/main/java/com/knowledge/base/application/filter/AuthFilter.java index d7aeefb..505a293 100644 --- a/src/main/java/com/knowledge/base/application/filter/AuthFilter.java +++ b/src/main/java/com/knowledge/base/application/filter/AuthFilter.java @@ -33,7 +33,7 @@ public class AuthFilter implements Filter { String path = request.getRequestURI(); // 需要鉴权的路径前缀 - if (path.startsWith("/api/v1/doc")) { + if (path.startsWith("/api/v1/doc") || path.startsWith("/api/v1/llm")) { String token = request.getHeader("Authorization"); if (StrUtil.isBlank(token)) { diff --git a/src/main/java/com/knowledge/base/application/service/LLMAppService.java b/src/main/java/com/knowledge/base/application/service/LLMAppService.java index 6a66eb2..89e9130 100644 --- a/src/main/java/com/knowledge/base/application/service/LLMAppService.java +++ b/src/main/java/com/knowledge/base/application/service/LLMAppService.java @@ -1,12 +1,12 @@ package com.knowledge.base.application.service; -import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import java.util.Map; public interface LLMAppService { String getToken(String password) throws Exception; - ResponseBodyEmitter ask(String userId, String question, Map params) throws Exception; + SseEmitter ask(String llmToken, String question, Map params) throws Exception; } diff --git a/src/main/java/com/knowledge/base/application/service/LLMAppServiceImpl.java b/src/main/java/com/knowledge/base/application/service/LLMAppServiceImpl.java index b2bf621..fd950cf 100644 --- a/src/main/java/com/knowledge/base/application/service/LLMAppServiceImpl.java +++ b/src/main/java/com/knowledge/base/application/service/LLMAppServiceImpl.java @@ -1,17 +1,18 @@ package com.knowledge.base.application.service; -import com.knowledge.base.infrastructure.south.llm.LLMService; import com.knowledge.base.infrastructure.south.llm.LLMServiceFactory; +import com.knowledge.base.infrastructure.util.http.SseOutputAdapter; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; -import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; -import java.io.PipedInputStream; -import java.io.PipedOutputStream; +import java.io.*; import java.util.Map; @Service @RequiredArgsConstructor +@Slf4j public class LLMAppServiceImpl implements LLMAppService { private final LLMServiceFactory llmServiceFactory; @@ -22,33 +23,19 @@ public class LLMAppServiceImpl implements LLMAppService { } @Override - public ResponseBodyEmitter ask(String userId, String question, Map params) throws Exception { - LLMService service = llmServiceFactory.current(); - - ResponseBodyEmitter emitter = new ResponseBodyEmitter(); - PipedOutputStream out = new PipedOutputStream(); - PipedInputStream in = new PipedInputStream(out); + public SseEmitter ask(String llmToken, String question, Map params) throws Exception { + SseEmitter emitter = new SseEmitter(0L); // 不超时 new Thread(() -> { try { - service.streamAnswer(userId, question, params, out); - } catch (Exception e) { - try { - emitter.send("data: {\"error\": \"LLM内部异常\"}\n\n"); - } catch (Exception ignore) {} - emitter.completeWithError(e); - } - }).start(); - - new Thread(() -> { - try (in) { - byte[] buf = new byte[1024]; - int len; - while ((len = in.read(buf)) != -1) { - emitter.send(new String(buf, 0, len)); - } + llmServiceFactory.current().streamAnswer(llmToken, question, params, new SseOutputAdapter(emitter)); emitter.complete(); } catch (Exception e) { + try { + emitter.send(SseEmitter.event().data("{\"error\": \"LLM异常\"}")); + } catch (IOException ignored) { + log.error("error", e); + } emitter.completeWithError(e); } }).start(); diff --git a/src/main/java/com/knowledge/base/infrastructure/cache/iface/UserCacheService.java b/src/main/java/com/knowledge/base/infrastructure/cache/iface/UserCacheService.java index d314d47..7bb6169 100644 --- a/src/main/java/com/knowledge/base/infrastructure/cache/iface/UserCacheService.java +++ b/src/main/java/com/knowledge/base/infrastructure/cache/iface/UserCacheService.java @@ -22,4 +22,9 @@ public interface UserCacheService { * @param userJsonStr */ void cacheUserJsonByToken(String token, String userJsonStr); + + + default void cacheAnythingLLMSlugId(String wsName, String slugId) {} + + default Optional getAnythingLLMSlugIdByWorkspaceName(String wsName) {return Optional.empty();} } diff --git a/src/main/java/com/knowledge/base/infrastructure/cache/impl/RedisUserCacheServiceImpl.java b/src/main/java/com/knowledge/base/infrastructure/cache/impl/RedisUserCacheServiceImpl.java index e0def7f..203fc14 100644 --- a/src/main/java/com/knowledge/base/infrastructure/cache/impl/RedisUserCacheServiceImpl.java +++ b/src/main/java/com/knowledge/base/infrastructure/cache/impl/RedisUserCacheServiceImpl.java @@ -8,6 +8,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Service; +import java.time.Duration; import java.util.Optional; import java.util.concurrent.TimeUnit; @@ -39,4 +40,14 @@ public class RedisUserCacheServiceImpl implements UserCacheService { public void cacheUserJsonByToken(String token, String userJsonStr) { redisTemplate.opsForValue().set(key("token", token), userJsonStr, ConstantConfig.USER_CACHE_EXPIRED_MINUTES, TimeUnit.MINUTES); } + + @Override + public void cacheAnythingLLMSlugId(String wsName, String slugId) { + redisTemplate.opsForValue().set(key("ws-name", wsName), slugId, Duration.ofDays(10).toMinutes(), TimeUnit.MINUTES); + } + + @Override + public Optional getAnythingLLMSlugIdByWorkspaceName(String wsName) { + return Optional.ofNullable(redisTemplate.opsForValue().get(key("ws-name", wsName))); + } } diff --git a/src/main/java/com/knowledge/base/infrastructure/config/ConstantConfig.java b/src/main/java/com/knowledge/base/infrastructure/config/ConstantConfig.java index 5857221..0c44082 100644 --- a/src/main/java/com/knowledge/base/infrastructure/config/ConstantConfig.java +++ b/src/main/java/com/knowledge/base/infrastructure/config/ConstantConfig.java @@ -28,4 +28,12 @@ public class ConstantConfig { public static final String DEFAULT_UPLOADER = "ADMIN"; public static final String UNKNOWN_LOCAL_RELA_PATH = "UNKNOWN"; + + + /** + * 以下是AnythingLLM相关 + */ + public static final String DEFAULT_SLUG_ID = "87e14982-a821-48d8-9c6b-3557d0bb2f96"; + + } diff --git a/src/main/java/com/knowledge/base/infrastructure/north/controller/LLMController.java b/src/main/java/com/knowledge/base/infrastructure/north/controller/LLMController.java index 801ce89..bfc1889 100644 --- a/src/main/java/com/knowledge/base/infrastructure/north/controller/LLMController.java +++ b/src/main/java/com/knowledge/base/infrastructure/north/controller/LLMController.java @@ -4,7 +4,7 @@ import com.knowledge.base.application.service.LLMAppService; import lombok.RequiredArgsConstructor; import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.*; -import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import java.util.Map; @@ -24,11 +24,11 @@ public class LLMController { } @PostMapping(value = "/ask", produces = MediaType.TEXT_EVENT_STREAM_VALUE) - public ResponseBodyEmitter ask(@RequestHeader("Authorization") String token, - @RequestBody Map body) throws Exception { + public SseEmitter ask(@RequestHeader("Authorization") String token, + @RequestBody Map body) throws Exception { String question = String.valueOf(body.get("question")); - String userId = String.valueOf(body.getOrDefault("userId", "anonymous")); - return llmAppService.ask(userId, question, body); + String llmToken = String.valueOf(body.getOrDefault("llmToken", "anonymous")); + return llmAppService.ask(llmToken, question, body); } } diff --git a/src/main/java/com/knowledge/base/infrastructure/south/llm/AnythingLLMServiceImpl.java b/src/main/java/com/knowledge/base/infrastructure/south/llm/AnythingLLMServiceImpl.java index f76624a..cb56d09 100644 --- a/src/main/java/com/knowledge/base/infrastructure/south/llm/AnythingLLMServiceImpl.java +++ b/src/main/java/com/knowledge/base/infrastructure/south/llm/AnythingLLMServiceImpl.java @@ -1,8 +1,13 @@ package com.knowledge.base.infrastructure.south.llm; -import com.fasterxml.jackson.databind.ObjectMapper; +import cn.hutool.core.collection.CollectionUtil; +import cn.hutool.core.util.StrUtil; +import com.knowledge.base.infrastructure.cache.iface.UserCacheService; +import com.knowledge.base.infrastructure.config.ConstantConfig; import com.knowledge.base.infrastructure.util.HttpHelper; +import com.knowledge.base.infrastructure.util.http.WriterAdapter; import lombok.RequiredArgsConstructor; +import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; import okhttp3.Response; import org.springframework.beans.factory.annotation.Value; @@ -12,8 +17,10 @@ import java.io.BufferedReader; import java.io.InputStreamReader; import java.io.OutputStream; import java.nio.charset.StandardCharsets; +import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; @Slf4j @Service @@ -24,11 +31,7 @@ public class AnythingLLMServiceImpl implements LLMService { private String baseUrl; private final HttpHelper httpHelper; - private final ObjectMapper objectMapper; - - private String cachedToken; - private long cachedAt; - private final long cacheExpireMs = 60 * 60 * 1000; + private final UserCacheService userCacheService; @Override public boolean supports(String type) { @@ -37,24 +40,18 @@ public class AnythingLLMServiceImpl implements LLMService { @Override public String fetchToken(String password) throws Exception { - String url = baseUrl + "/api/token"; + String url = baseUrl + "/api/request-token"; Map result = httpHelper.postJson(url, Map.of("password", password), null); - String token = String.valueOf(result.get("token")); - cachedToken = token; - cachedAt = System.currentTimeMillis(); - return token; - } - - private String getValidToken() throws Exception { - if (cachedToken != null && (System.currentTimeMillis() - cachedAt) < cacheExpireMs) { - return cachedToken; - } - return fetchToken("wisdom2025"); + return String.valueOf(result.get("token")); } @Override - public void streamAnswer(String userId, String question, Map params, OutputStream output) throws Exception { - String slug = String.valueOf(params.getOrDefault("slug", "kb")); + public void streamAnswer(String llmToken, String question, Map params, WriterAdapter writer) throws Exception { + String slug = (String) params.get("slug"); + if (StrUtil.isBlank(slug)) { + slug = fetchSlugByWsName(llmToken, (String) params.getOrDefault("wsName", "部门知识库")); + } + String url = baseUrl + "/api/workspace/" + slug + "/stream-chat"; Map body = Map.of( @@ -62,14 +59,36 @@ public class AnythingLLMServiceImpl implements LLMService { "attachments", params.getOrDefault("attachments", List.of()) ); - try (Response response = httpHelper.postStream(url, body, "Bearer " + getValidToken()); + try (Response response = httpHelper.postStream(url, body, "Bearer " + llmToken); BufferedReader reader = new BufferedReader(new InputStreamReader(response.body().byteStream(), StandardCharsets.UTF_8))) { String line; while ((line = reader.readLine()) != null) { - output.write((line + "\n").getBytes(StandardCharsets.UTF_8)); - output.flush(); + writer.writeLine(line); // 已是 SSE 格式 } } } + + @SneakyThrows + private String fetchSlugByWsName(String llmToken, String wsName) { + Optional slugId = userCacheService.getAnythingLLMSlugIdByWorkspaceName(wsName); + if(slugId.isPresent()) { + return slugId.get(); + } + String url = baseUrl + "/api/workspaces"; + Map resMap = httpHelper.get(url, llmToken); + List workspaces = (List)resMap.get("workspaces"); + if(CollectionUtil.isEmpty(workspaces)) { + return ConstantConfig.DEFAULT_SLUG_ID; + } + Optional targetWorkspace = workspaces.stream().filter(workspace -> workspace.get("name").equals(wsName)).findFirst(); + if(targetWorkspace.isEmpty()) { + return ConstantConfig.DEFAULT_SLUG_ID; + } + + String slug = (String) targetWorkspace.get().get("slug"); + // slug数据进缓存 + userCacheService.cacheAnythingLLMSlugId(wsName, slug); + return StrUtil.isBlank(slug) ? ConstantConfig.DEFAULT_SLUG_ID : slug; + } } diff --git a/src/main/java/com/knowledge/base/infrastructure/south/llm/LLMService.java b/src/main/java/com/knowledge/base/infrastructure/south/llm/LLMService.java index fb487be..49e19cf 100644 --- a/src/main/java/com/knowledge/base/infrastructure/south/llm/LLMService.java +++ b/src/main/java/com/knowledge/base/infrastructure/south/llm/LLMService.java @@ -1,12 +1,13 @@ package com.knowledge.base.infrastructure.south.llm; -import java.io.OutputStream; +import com.knowledge.base.infrastructure.util.http.WriterAdapter; + import java.util.Map; public interface LLMService { String fetchToken(String password) throws Exception; - void streamAnswer(String userId, String question, Map params, OutputStream output) throws Exception; + void streamAnswer(String llmToken, String question, Map params, WriterAdapter writer) throws Exception; boolean supports(String type); } diff --git a/src/main/java/com/knowledge/base/infrastructure/util/HttpHelper.java b/src/main/java/com/knowledge/base/infrastructure/util/HttpHelper.java index 76be105..e7d6e70 100644 --- a/src/main/java/com/knowledge/base/infrastructure/util/HttpHelper.java +++ b/src/main/java/com/knowledge/base/infrastructure/util/HttpHelper.java @@ -22,8 +22,8 @@ public class HttpHelper { public HttpHelper() { this.client = new OkHttpClient.Builder() - .connectTimeout(20, TimeUnit.SECONDS) - .readTimeout(30, TimeUnit.SECONDS) + .connectTimeout(60, TimeUnit.SECONDS) + .readTimeout(300, TimeUnit.SECONDS) .build(); this.objectMapper = new ObjectMapper(); } @@ -48,8 +48,18 @@ public class HttpHelper { * 流式 POST 请求，返回 Response 实例，调用方负责关闭 response.body().close() */ public Response postStream(String url, Map body, String token) throws IOException { - Request request = buildJsonRequest(url, body, token, "POST"); - return client.newCall(request).execute(); // 返回后调用方处理 body() + String json = objectMapper.writeValueAsString(body); + RequestBody requestBody = RequestBody.create(json, MediaType.get("application/json; charset=utf-8")); + + Request request = new Request.Builder() + .url(url) + .addHeader("Authorization", token) + .addHeader("Accept", "text/event-stream") + .addHeader("Connection", "keep-alive") + .post(requestBody) + .build(); + + return client.newCall(request).execute(); } /** diff --git a/src/main/java/com/knowledge/base/infrastructure/util/http/SseOutputAdapter.java b/src/main/java/com/knowledge/base/infrastructure/util/http/SseOutputAdapter.java new file mode 100644 index 0000000..7f91c9d --- /dev/null +++ b/src/main/java/com/knowledge/base/infrastructure/util/http/SseOutputAdapter.java @@ -0,0 +1,22 @@ +package com.knowledge.base.infrastructure.util.http; + +import cn.hutool.core.util.StrUtil; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.io.IOException; + +public class SseOutputAdapter implements WriterAdapter { + + private final SseEmitter emitter; + + public SseOutputAdapter(SseEmitter emitter) { + this.emitter = emitter; + } + + @Override + public void writeLine(String line) throws IOException { + if (StrUtil.isNotBlank(line)) { + emitter.send(SseEmitter.event().data(line)); + } + } +} diff --git a/src/main/java/com/knowledge/base/infrastructure/util/http/WriterAdapter.java b/src/main/java/com/knowledge/base/infrastructure/util/http/WriterAdapter.java new file mode 100644 index 0000000..92a73d0 --- /dev/null +++ b/src/main/java/com/knowledge/base/infrastructure/util/http/WriterAdapter.java @@ -0,0 +1,8 @@ +package com.knowledge.base.infrastructure.util.http; + +import java.io.IOException; + +public interface WriterAdapter { + void writeLine(String line) throws IOException; +} + diff --git a/src/main/resources/application-dev-windows.properties b/src/main/resources/application-dev-windows.properties index 468702a..b4c4c25 100644 --- a/src/main/resources/application-dev-windows.properties +++ b/src/main/resources/application-dev-windows.properties @@ -58,4 +58,5 @@ knowledge.base.redis.password= # 可留空 knowledge.base.redis.default-expire-minutes=1440 # 大模型相关 -llm.remote.base-url=http://localhost:3001 \ No newline at end of file +llm.remote.base-url=http://localhost:3001 +#llm.remote.base-url=http://llm.wisdompulse.cn \ No newline at end of file