diff --git a/src/main/java/com/knowledge/base/application/filter/LogFilter.java b/src/main/java/com/knowledge/base/application/filter/LogFilter.java index b331c4d..bb027fc 100644 --- a/src/main/java/com/knowledge/base/application/filter/LogFilter.java +++ b/src/main/java/com/knowledge/base/application/filter/LogFilter.java @@ -1,5 +1,6 @@ package com.knowledge.base.application.filter; +import cn.hutool.core.util.IdUtil; import cn.hutool.core.util.StrUtil; import cn.hutool.json.JSONUtil; import com.knowledge.base.application.enums.MDCKeyEnum; @@ -20,79 +21,112 @@ import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; import java.io.IOException; import java.net.InetAddress; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; import java.util.Optional; -import java.util.UUID; /** - * Description: 交易日志记录 + * 通用交易日志过滤器 */ @Component public class LogFilter extends OncePerRequestFilter { - private final static Logger LOGGER = LoggerFactory.getLogger(LogFilter.class); + + private static final Logger LOGGER = LoggerFactory.getLogger(LogFilter.class); + private static final int MAX_LOG_BODY_LENGTH = 3000; @Autowired private DynamicConfig dynamicConfig; @Override - protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain) throws ServletException, IOException { - ContentCachingRequestWrapper contentCachingRequestWrapper = new ContentCachingRequestWrapper(request); - ContentCachingResponseWrapper contentCachingResponseWrapper = new ContentCachingResponseWrapper(response); - if (!"Y".equals(dynamicConfig.getRecordMsgBody())) { - // 不记录请求和响应信息 - filterChain.doFilter(contentCachingRequestWrapper, contentCachingResponseWrapper); - contentCachingResponseWrapper.copyBodyToResponse(); + protected void doFilterInternal(HttpServletRequest request, + HttpServletResponse response, + FilterChain filterChain) throws ServletException, IOException { + + ContentCachingRequestWrapper requestWrapper = new ContentCachingRequestWrapper(request); + ContentCachingResponseWrapper responseWrapper = new ContentCachingResponseWrapper(response); + + if (!"Y".equalsIgnoreCase(dynamicConfig.getRecordMsgBody())) { + filterChain.doFilter(requestWrapper, responseWrapper); + responseWrapper.copyBodyToResponse(); return; } StopWatch stopWatch = new StopWatch(); + String traceId = null; try { stopWatch.start(); - //获取http header 中的traceid,没有默认生成一个,放入到mdc中,进行日志打印 - String traceId = Optional.ofNullable(request.getHeader("traceId")) - .orElse(UUID.randomUUID().toString().replace("-", "")); + + // traceId 生成或提取 + traceId = Optional.ofNullable(request.getHeader("traceId")) + .orElse(IdUtil.fastSimpleUUID()); + InetAddress addr = InetAddress.getLocalHost(); MDC.put(MDCKeyEnum.HOST_NAME.name, addr.getHostName()); MDC.put(MDCKeyEnum.TRACE_ID.name, traceId); - filterChain.doFilter(contentCachingRequestWrapper, contentCachingResponseWrapper); + + filterChain.doFilter(requestWrapper, responseWrapper); } catch (Exception e) { - LOGGER.warn("交易跟踪号[{}]记录交易报文信息异常:{}", MDC.get(MDCKeyEnum.TRACE_ID.name), e.getMessage()); - + LOGGER.warn("记录交易信息异常(traceId={}): {}", MDC.get(MDCKeyEnum.TRACE_ID.name), e.getMessage()); } finally { - byte[] requestBody = contentCachingRequestWrapper.getContentAsByteArray(); - byte[] responseBody = contentCachingResponseWrapper.getContentAsByteArray(); - try { - //获取请求路径 - String url = contentCachingRequestWrapper.getRequestURL().toString(); - //获取报文头交易码 - String tranCode = Optional.ofNullable(contentCachingRequestWrapper.getHeader("TRANCODE")).orElse("UNKOWN"); - //获取IP地址 - String remoteIp = getRemortIP(contentCachingRequestWrapper); - //获取IP端口 - int remotePort = contentCachingRequestWrapper.getRemotePort(); - stopWatch.stop(); - - LOGGER.info("交易跟踪号[{}]\n请求路径[{}]\n交易码[{}]\n远程地址[{}]\n远程端口[{}]\n请求内容:[{}]\n响应内容:[{}]\n响应时间:[{}]毫秒", - MDC.get(MDCKeyEnum.TRACE_ID.name), - url, - tranCode, - remoteIp, - remotePort, - JSONUtil.toJsonStr(new String(requestBody)), - JSONUtil.toJsonStr(new String(responseBody)), - stopWatch.getTotalTimeMillis()); - } catch (Exception e) { - LOGGER.warn("登记交易信息异常:" + e.getMessage()); - } - MDC.remove(MDCKeyEnum.HOST_NAME.name); - MDC.remove(MDCKeyEnum.TRACE_ID.name); - contentCachingResponseWrapper.copyBodyToResponse(); + responseWrapper.setHeader(MDCKeyEnum.TRACE_ID.name, traceId); + stopWatch.stop(); + logTraceMessage(requestWrapper, responseWrapper, stopWatch.getTotalTimeMillis()); + MDC.clear(); + responseWrapper.copyBodyToResponse(); } } - private String getRemortIP(HttpServletRequest request) { - return StrUtil.isBlank(request.getHeader("x-forwarded-for")) ? - request.getRemoteAddr() : - request.getHeader("x-forwarded-for"); + private void logTraceMessage(ContentCachingRequestWrapper request, + ContentCachingResponseWrapper response, + long timeCostMs) { + + String url = request.getRequestURL().toString(); + String tranCode = Optional.ofNullable(request.getHeader("TRANCODE")).orElse("UNKNOWN"); + String remoteIp = getRemoteIP(request); + int remotePort = request.getRemotePort(); + + String reqBody = extractRequestBody(request); + String respBody = extractResponseBody(response); + + Map logMap = new HashMap<>(); + logMap.put("traceId", MDC.get(MDCKeyEnum.TRACE_ID.name)); + logMap.put("url", url); + logMap.put("tranCode", tranCode); + logMap.put("ip", remoteIp); + logMap.put("port", remotePort); + logMap.put("request", reqBody); + logMap.put("response", respBody); + logMap.put("timeMs", timeCostMs); + + LOGGER.info("log_trace_record: {}", JSONUtil.toJsonStr(logMap)); + } + + private String extractRequestBody(ContentCachingRequestWrapper request) { + String contentType = request.getContentType(); + byte[] content = request.getContentAsByteArray(); + + if (StrUtil.isNotBlank(contentType) && contentType.toLowerCase().contains("multipart/form-data")) { + // 表单上传只记录参数，不打印文件体 + Map paramMap = request.getParameterMap(); + return "[multipart/form-data] " + JSONUtil.toJsonStr(paramMap); + } + + if (content.length == 0) return ""; + String body = new String(content, StandardCharsets.UTF_8); + return body.length() > MAX_LOG_BODY_LENGTH ? body.substring(0, MAX_LOG_BODY_LENGTH) + "...[truncated]" : body; + } + + private String extractResponseBody(ContentCachingResponseWrapper response) { + byte[] content = response.getContentAsByteArray(); + if (content.length == 0) return ""; + String body = new String(content, StandardCharsets.UTF_8); + return body.length() > MAX_LOG_BODY_LENGTH ? body.substring(0, MAX_LOG_BODY_LENGTH) + "...[truncated]" : body; + } + + private String getRemoteIP(HttpServletRequest request) { + String xff = request.getHeader("x-forwarded-for"); + return StrUtil.isBlank(xff) ? request.getRemoteAddr() : xff; } } diff --git a/src/main/java/com/knowledge/base/application/service/LLMAppService.java b/src/main/java/com/knowledge/base/application/service/LLMAppService.java new file mode 100644 index 0000000..6a66eb2 --- /dev/null +++ b/src/main/java/com/knowledge/base/application/service/LLMAppService.java @@ -0,0 +1,13 @@ +package com.knowledge.base.application.service; + +import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; + +import java.util.Map; + +public interface LLMAppService { + String getToken(String password) throws Exception; + ResponseBodyEmitter ask(String userId, 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 new file mode 100644 index 0000000..b2bf621 --- /dev/null +++ b/src/main/java/com/knowledge/base/application/service/LLMAppServiceImpl.java @@ -0,0 +1,59 @@ +package com.knowledge.base.application.service; + +import com.knowledge.base.infrastructure.south.llm.LLMService; +import com.knowledge.base.infrastructure.south.llm.LLMServiceFactory; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; + +import java.io.PipedInputStream; +import java.io.PipedOutputStream; +import java.util.Map; + +@Service +@RequiredArgsConstructor +public class LLMAppServiceImpl implements LLMAppService { + + private final LLMServiceFactory llmServiceFactory; + + @Override + public String getToken(String password) throws Exception { + return llmServiceFactory.current().fetchToken(password); + } + + @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); + + 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)); + } + emitter.complete(); + } catch (Exception e) { + emitter.completeWithError(e); + } + }).start(); + + return emitter; + } +} + 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 e03d90f..801ce89 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 @@ -1,93 +1,36 @@ package com.knowledge.base.infrastructure.north.controller; -import com.knowledge.base.infrastructure.util.HttpHelper; +import com.knowledge.base.application.service.LLMAppService; import lombok.RequiredArgsConstructor; -import org.springframework.web.bind.annotation.RequestMapping; -import org.springframework.web.bind.annotation.RestController; - -import com.fasterxml.jackson.databind.ObjectMapper; -import com.knowledge.base.infrastructure.config.LLmConfig; -import lombok.extern.slf4j.Slf4j; -import okhttp3.Response; import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.*; import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; -import java.io.BufferedReader; -import java.io.InputStreamReader; -import java.nio.charset.StandardCharsets; import java.util.Map; - -/** - * @author Luke.ye - * @date 2025/6/22 08:08 - */ - -@Slf4j @RestController @RequestMapping("/api/v1/llm") @RequiredArgsConstructor public class LLMController { - private final HttpHelper okHttpHelper; - private final LLmConfig llmConfig; - private final ObjectMapper objectMapper; + private final LLMAppService llmAppService; - /** - * POST /token - * 请求：{ password: xxx } - * 返回：{ token: xxx } - */ @PostMapping("/token") - public Map getToken(@RequestBody Map body) throws Exception { - String url = llmConfig.getBaseUrl() + "/api/token"; - return okHttpHelper.postJson(url, body, null); + public Map getToken(@RequestHeader("Authorization") String token, + @RequestBody Map body) throws Exception { + String password = String.valueOf(body.get("password")); + String resultToken = llmAppService.getToken(password); + return Map.of("token", resultToken); } - /** - * GET /workspaces - * Header: Authorization: Bearer xxx - * 返回：{ workspaces: [...] } - */ - @GetMapping("/workspaces") - public Map getWorkspaces(@RequestHeader("Authorization") String token) throws Exception { - String url = llmConfig.getBaseUrl() + "/api/workspaces"; - return okHttpHelper.get(url, token); - } - - /** - * POST /workspace/{slug}/stream-chat - * Header: Authorization - * Body: { message: "...", attachments: [...] } - * 返回：text/event-stream - */ - @PostMapping(value = "/workspace/{slug}/stream-chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE) - public ResponseBodyEmitter chatStream(@PathVariable String slug, - @RequestHeader("Authorization") String token, - @RequestBody Map body) { - String url = llmConfig.getBaseUrl() + "/api/workspace/" + slug + "/stream-chat"; - ResponseBodyEmitter emitter = new ResponseBodyEmitter(); - - new Thread(() -> { - try (Response resp = okHttpHelper.postStream(url, body, token); - BufferedReader reader = new BufferedReader(new InputStreamReader(resp.body().byteStream(), StandardCharsets.UTF_8))) { - - String line; - while ((line = reader.readLine()) != null) { - emitter.send(line + "\n"); - } - emitter.complete(); - } catch (Exception e) { - log.error("LLM 流式回答异常", e); - try { - emitter.send("data: {\"textResponse\":\"LLM服务异常\"}\n\n"); - } catch (Exception ignore) {} - emitter.completeWithError(e); - } - }).start(); - - return emitter; + @PostMapping(value = "/ask", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public ResponseBodyEmitter 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); } } + + 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 new file mode 100644 index 0000000..f76624a --- /dev/null +++ b/src/main/java/com/knowledge/base/infrastructure/south/llm/AnythingLLMServiceImpl.java @@ -0,0 +1,75 @@ +package com.knowledge.base.infrastructure.south.llm; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.knowledge.base.infrastructure.util.HttpHelper; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import okhttp3.Response; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; + +import java.io.BufferedReader; +import java.io.InputStreamReader; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; + +@Slf4j +@Service +@RequiredArgsConstructor +public class AnythingLLMServiceImpl implements LLMService { + + @Value("${llm.remote.base-url}") + private String baseUrl; + + private final HttpHelper httpHelper; + private final ObjectMapper objectMapper; + + private String cachedToken; + private long cachedAt; + private final long cacheExpireMs = 60 * 60 * 1000; + + @Override + public boolean supports(String type) { + return "anything".equalsIgnoreCase(type); + } + + @Override + public String fetchToken(String password) throws Exception { + String url = baseUrl + "/api/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"); + } + + @Override + public void streamAnswer(String userId, String question, Map params, OutputStream output) throws Exception { + String slug = String.valueOf(params.getOrDefault("slug", "kb")); + String url = baseUrl + "/api/workspace/" + slug + "/stream-chat"; + + Map body = Map.of( + "message", question, + "attachments", params.getOrDefault("attachments", List.of()) + ); + + try (Response response = httpHelper.postStream(url, body, "Bearer " + getValidToken()); + 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(); + } + } + } +} 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 new file mode 100644 index 0000000..fb487be --- /dev/null +++ b/src/main/java/com/knowledge/base/infrastructure/south/llm/LLMService.java @@ -0,0 +1,13 @@ +package com.knowledge.base.infrastructure.south.llm; + +import java.io.OutputStream; +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; + + boolean supports(String type); +} + diff --git a/src/main/java/com/knowledge/base/infrastructure/south/llm/LLMServiceFactory.java b/src/main/java/com/knowledge/base/infrastructure/south/llm/LLMServiceFactory.java new file mode 100644 index 0000000..b59b3f7 --- /dev/null +++ b/src/main/java/com/knowledge/base/infrastructure/south/llm/LLMServiceFactory.java @@ -0,0 +1,25 @@ +package com.knowledge.base.infrastructure.south.llm; + +import lombok.RequiredArgsConstructor; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +import java.util.List; + +@Component +@RequiredArgsConstructor +public class LLMServiceFactory { + + @Value("${llm.provider:anything}") + private String provider; + + private final List services; + + public LLMService current() { + return services.stream() + .filter(s -> s.supports(provider)) + .findFirst() + .orElseThrow(() -> new IllegalStateException("不支持的LLM服务类型: " + provider)); + } +} + diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 96cfe2d..e9dc451 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -34,4 +34,5 @@ object.storage.public-buckets[0]=kbase object.storage.public-buckets[1]=public-media # 大模型相关 +llm.provider=anything llm.remote.base-url=http://llm.wisdompulse.cn \ No newline at end of file diff --git a/src/test/resources/application.properties b/src/test/resources/application.properties index 4fcff6c..813ec33 100644 --- a/src/test/resources/application.properties +++ b/src/test/resources/application.properties @@ -23,6 +23,7 @@ import.schedule.enabled=true import.schedule.cron=0 0 * * * * # 大模型相关 +llm.provider=anything llm.remote.base-url=http://llm.wisdompulse.cn