1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package org.codelibs.fess.api.chat;
17
18 import java.io.IOException;
19 import java.io.PrintWriter;
20 import java.util.ArrayList;
21 import java.util.HashMap;
22 import java.util.List;
23 import java.util.Locale;
24 import java.util.Map;
25 import java.util.Set;
26
27 import org.apache.logging.log4j.LogManager;
28 import org.apache.logging.log4j.Logger;
29 import org.codelibs.core.lang.StringUtil;
30 import org.codelibs.fess.Constants;
31 import org.codelibs.fess.api.BaseApiManager;
32 import org.codelibs.fess.chat.ChatClient.ChatResult;
33 import org.codelibs.fess.chat.ChatPhaseCallback;
34 import org.codelibs.fess.entity.ChatMessage.ChatSource;
35 import org.codelibs.fess.entity.FacetQueryView;
36 import org.codelibs.fess.entity.SearchRequestParams;
37 import org.codelibs.fess.helper.SystemHelper;
38 import org.codelibs.fess.llm.LlmException;
39 import org.codelibs.fess.mylasta.direction.FessConfig;
40 import org.codelibs.fess.util.ComponentUtil;
41
42 import com.fasterxml.jackson.core.JsonProcessingException;
43 import com.fasterxml.jackson.databind.ObjectMapper;
44
45 import jakarta.annotation.PostConstruct;
46 import jakarta.servlet.FilterChain;
47 import jakarta.servlet.ServletException;
48 import jakarta.servlet.http.HttpServletRequest;
49 import jakarta.servlet.http.HttpServletResponse;
50
51
52
53
54
55
56 public class ChatApiManager extends BaseApiManager {
57
58 private static final Logger logger = LogManager.getLogger(ChatApiManager.class);
59
60 private static final String CHAT_API_PATH = "/api/v1/chat";
61 private static final String STREAM_API_PATH = "/api/v1/chat/stream";
62
63 private static final ObjectMapper objectMapper = new ObjectMapper();
64
65
66 protected String pathPrefix = CHAT_API_PATH;
67
68
69
70
71 public ChatApiManager() {
72
73 }
74
75
76
77
78 @PostConstruct
79 public void register() {
80 if (logger.isInfoEnabled()) {
81 logger.info("Registering ChatApiManager");
82 }
83 ComponentUtil.getWebApiManagerFactory().add(this);
84 }
85
86 @Override
87 public boolean matches(final HttpServletRequest request) {
88 final FessConfig fessConfig = ComponentUtil.getFessConfig();
89 final boolean ragChatEnabled = fessConfig.isRagChatEnabled();
90 if (!ragChatEnabled) {
91 if (logger.isTraceEnabled()) {
92 logger.trace("ChatApiManager.matches() returning false. ragChatEnabled={}", ragChatEnabled);
93 }
94 return false;
95 }
96
97 final String servletPath = request.getServletPath();
98 final boolean matches = servletPath.startsWith(CHAT_API_PATH);
99 if (logger.isTraceEnabled()) {
100 logger.trace("ChatApiManager.matches() checking path. servletPath={}, expectedPrefix={}, matches={}", servletPath,
101 CHAT_API_PATH, matches);
102 }
103 return matches;
104 }
105
106 @Override
107 public void process(final HttpServletRequest request, final HttpServletResponse response, final FilterChain chain)
108 throws IOException, ServletException {
109 final String servletPath = request.getServletPath();
110
111 if (logger.isDebugEnabled()) {
112 logger.debug("Processing chat API request. path={}, method={}", servletPath, request.getMethod());
113 }
114
115 if (servletPath.equals(STREAM_API_PATH) || servletPath.equals(STREAM_API_PATH + "/")) {
116 processStreamRequest(request, response);
117 } else if (servletPath.equals(CHAT_API_PATH) || servletPath.equals(CHAT_API_PATH + "/")) {
118 processChatRequest(request, response);
119 } else {
120 if (logger.isDebugEnabled()) {
121 logger.debug("Unknown chat API path. path={}", servletPath);
122 }
123 writeJsonResponse(response, HttpServletResponse.SC_NOT_FOUND, createErrorResponse("Not found"));
124 }
125 }
126
127
128
129
130
131
132
133
134 protected void processChatRequest(final HttpServletRequest request, final HttpServletResponse response) throws IOException {
135 if (!"POST".equalsIgnoreCase(request.getMethod())) {
136 if (logger.isDebugEnabled()) {
137 logger.debug("Invalid method for chat request. method={}", request.getMethod());
138 }
139 writeJsonResponse(response, HttpServletResponse.SC_METHOD_NOT_ALLOWED, createErrorResponse("Method not allowed"));
140 return;
141 }
142
143 try {
144 final String message = request.getParameter("message");
145 final String sessionId = request.getParameter("sessionId");
146 final String clearParam = request.getParameter("clear");
147
148 if (logger.isDebugEnabled()) {
149 logger.debug("Processing chat request. sessionId={}, messageLength={}, clear={}", sessionId,
150 message != null ? message.length() : 0, clearParam);
151 }
152
153 if (StringUtil.isBlank(message)) {
154 if ("true".equals(clearParam) && StringUtil.isNotBlank(sessionId)) {
155 final String clearUserId = getUserId(request);
156 final boolean cleared = ComponentUtil.getChatSessionManager().clearSession(sessionId, clearUserId);
157 if (cleared) {
158 if (logger.isDebugEnabled()) {
159 logger.debug("Session cleared. sessionId={}, userId={}", sessionId, clearUserId);
160 }
161 writeJsonResponse(response, HttpServletResponse.SC_OK, createSuccessResponse(sessionId, "Session cleared", null));
162 } else {
163 if (logger.isDebugEnabled()) {
164 logger.debug("Session not found or not owned. sessionId={}, userId={}", sessionId, clearUserId);
165 }
166 writeJsonResponse(response, HttpServletResponse.SC_NOT_FOUND, createErrorResponse("Session not found"));
167 }
168 return;
169 }
170 if (logger.isDebugEnabled()) {
171 logger.debug("Message is required but was empty");
172 }
173 writeJsonResponse(response, HttpServletResponse.SC_BAD_REQUEST, createErrorResponse("Message is required"));
174 return;
175 }
176
177 final FessConfig fessConfig = ComponentUtil.getFessConfig();
178 final int maxMessageLength = getMaxMessageLength(fessConfig);
179 if (message.length() > maxMessageLength) {
180 logger.warn("Chat message exceeds max length. length={}, max={}", message.length(), maxMessageLength);
181 writeJsonResponse(response, HttpServletResponse.SC_BAD_REQUEST,
182 createErrorResponse("Message is too long (max " + maxMessageLength + " characters)"));
183 return;
184 }
185
186 final String userId = getUserId(request);
187
188
189 request.setAttribute(Constants.SEARCH_LOG_ACCESS_TYPE,
190 ComponentUtil.getFessConfig().getSystemProperty("rag.llm.name", "ollama"));
191
192 final Map<String, String[]> fields = parseFieldFilters(request);
193 final String[] extraQueries = parseExtraQueries(request);
194 final ChatResult result;
195 if (fields.isEmpty() && extraQueries.length == 0) {
196 result = ComponentUtil.getChatClient().chat(sessionId, message, userId);
197 } else {
198 result = ComponentUtil.getChatClient().chat(sessionId, message, userId, fields, extraQueries);
199 }
200
201 if (logger.isDebugEnabled()) {
202 logger.debug("Chat request completed. sessionId={}, responseLength={}", result.getSessionId(),
203 result.getMessage().getContent() != null ? result.getMessage().getContent().length() : 0);
204 }
205
206 writeJsonResponse(response, HttpServletResponse.SC_OK,
207 createSuccessResponse(result.getSessionId(), result.getMessage().getContent(), result.getMessage().getSources()));
208
209 } catch (final Exception e) {
210 logger.warn("[RAG] Failed to process chat request. message={}", e.getMessage(), e);
211 writeJsonResponse(response, HttpServletResponse.SC_INTERNAL_SERVER_ERROR, createErrorResponse("Internal server error"));
212 }
213 }
214
215
216
217
218
219
220
221
222
223 protected void processStreamRequest(final HttpServletRequest request, final HttpServletResponse response) throws IOException {
224 if (!"GET".equalsIgnoreCase(request.getMethod()) && !"POST".equalsIgnoreCase(request.getMethod())) {
225 if (logger.isDebugEnabled()) {
226 logger.debug("Invalid method for stream request. method={}, expected GET or POST", request.getMethod());
227 }
228 writeJsonResponse(response, HttpServletResponse.SC_METHOD_NOT_ALLOWED, createErrorResponse("Method not allowed"));
229 return;
230 }
231
232 final String message = request.getParameter("message");
233 final String sessionId = request.getParameter("sessionId");
234
235 if (logger.isDebugEnabled()) {
236 logger.debug("Processing stream request. sessionId={}, messageLength={}", sessionId, message != null ? message.length() : 0);
237 }
238
239 if (StringUtil.isBlank(message)) {
240 if (logger.isDebugEnabled()) {
241 logger.debug("Message is required but was empty for stream request");
242 }
243 writeJsonResponse(response, HttpServletResponse.SC_BAD_REQUEST, createErrorResponse("Message is required"));
244 return;
245 }
246
247 final FessConfig fessConfig = ComponentUtil.getFessConfig();
248 final int maxMessageLength = getMaxMessageLength(fessConfig);
249 if (message.length() > maxMessageLength) {
250 logger.warn("Stream message exceeds max length. length={}, max={}", message.length(), maxMessageLength);
251 writeJsonResponse(response, HttpServletResponse.SC_BAD_REQUEST,
252 createErrorResponse("Message is too long (max " + maxMessageLength + " characters)"));
253 return;
254 }
255
256
257 response.setContentType("text/event-stream");
258 response.setCharacterEncoding("UTF-8");
259 response.setHeader("Cache-Control", "no-cache");
260 response.setHeader("Connection", "keep-alive");
261 response.setHeader("X-Accel-Buffering", "no");
262
263 try (final PrintWriter writer = response.getWriter()) {
264 final String userId = getUserId(request);
265
266
267 request.setAttribute(Constants.SEARCH_LOG_ACCESS_TYPE,
268 ComponentUtil.getFessConfig().getSystemProperty("rag.llm.name", "ollama"));
269
270
271 final ChatPhaseCallback phaseCallback = new ChatPhaseCallback() {
272 @Override
273 public void onPhaseStart(final String phase, final String phaseMessage) {
274 onPhaseStart(phase, phaseMessage, null);
275 }
276
277 @Override
278 public void onPhaseStart(final String phase, final String phaseMessage, final String keywords) {
279 try {
280 final Map<String, Object> data = new HashMap<>();
281 data.put("phase", phase);
282 data.put("status", "start");
283 data.put("message", phaseMessage);
284 if (keywords != null) {
285 data.put("keywords", keywords);
286 }
287 sendSseEvent(writer, "phase", data);
288 if (logger.isDebugEnabled()) {
289 logger.debug("SSE phase start event sent. phase={}, message={}, keywords={}", phase, phaseMessage, keywords);
290 }
291 } catch (final Exception e) {
292 if (logger.isDebugEnabled()) {
293 logger.debug("Failed to send phase start event. phase={}, error={}", phase, e.getMessage());
294 }
295 }
296 }
297
298 @Override
299 public void onPhaseComplete(final String phase) {
300 try {
301 sendSseEvent(writer, "phase", Map.of("phase", phase, "status", "complete"));
302 if (logger.isDebugEnabled()) {
303 logger.debug("SSE phase complete event sent. phase={}", phase);
304 }
305 } catch (final Exception e) {
306 if (logger.isDebugEnabled()) {
307 logger.debug("Failed to send phase complete event. phase={}, error={}", phase, e.getMessage());
308 }
309 }
310 }
311
312 @Override
313 public void onChunk(final String content, final boolean done) {
314 try {
315 if (content != null && !content.isEmpty()) {
316 sendSseEvent(writer, "chunk", Map.of("content", content));
317 }
318 } catch (final Exception e) {
319 if (logger.isDebugEnabled()) {
320 logger.debug("Failed to send SSE chunk. error={}", e.getMessage());
321 }
322 }
323 }
324
325 @Override
326 public void onError(final String phase, final String errorCode) {
327 try {
328 sendSseEvent(writer, "error", Map.of("phase", phase, "message", errorCode, "errorCode", errorCode));
329 if (logger.isDebugEnabled()) {
330 logger.debug("SSE error event sent. phase={}, error={}", phase, errorCode);
331 }
332 } catch (final Exception e) {
333 if (logger.isDebugEnabled()) {
334 logger.debug("Failed to send error event. phase={}, error={}", phase, e.getMessage());
335 }
336 }
337 }
338 };
339
340
341 final Map<String, String[]> fields = parseFieldFilters(request);
342 final String[] extraQueries = parseExtraQueries(request);
343
344
345 final ChatResult result;
346 if (fields.isEmpty() && extraQueries.length == 0) {
347 result = ComponentUtil.getChatClient().streamChatEnhanced(sessionId, message, userId, phaseCallback);
348 } else {
349 result = ComponentUtil.getChatClient().streamChatEnhanced(sessionId, message, userId, fields, extraQueries, phaseCallback);
350 }
351
352
353 final List<ChatSource> sources = result.getMessage().getSources();
354 if (sources != null && !sources.isEmpty()) {
355 sendSseEvent(writer, "sources", Map.of("sources", sources));
356 if (logger.isDebugEnabled()) {
357 logger.debug("SSE sources event sent. sourcesCount={}", sources.size());
358 }
359 }
360
361
362 final Map<String, Object> doneData = new HashMap<>();
363 doneData.put("sessionId", result.getSessionId());
364 final String htmlContent = result.getMessage().getHtmlContent();
365 if (htmlContent != null) {
366 doneData.put("htmlContent", htmlContent);
367 }
368 sendSseEvent(writer, "done", doneData);
369 if (logger.isDebugEnabled()) {
370 logger.debug("SSE stream completed. sessionId={}, hasHtmlContent={}", result.getSessionId(), htmlContent != null);
371 }
372
373 } catch (final LlmException e) {
374
375 logger.warn("LLM error during stream request. sessionId={}, errorCode={}, message={}", sessionId, e.getErrorCode(),
376 e.getMessage(), e);
377 } catch (final Exception e) {
378 logger.warn("[RAG] Failed to process stream request. sessionId={}, message={}", sessionId, e.getMessage(), e);
379 if (!response.isCommitted()) {
380 try (final PrintWriter writer = response.getWriter()) {
381 sendSseEvent(writer, "error", Map.of("message", "Internal server error", "errorCode", LlmException.ERROR_UNKNOWN));
382 } catch (final IOException ioe) {
383 logger.warn("Failed to send error response. error={}", ioe.getMessage());
384 }
385 }
386 }
387 }
388
389
390
391
392
393
394
395
396 protected void sendSseEvent(final PrintWriter writer, final String event, final Map<String, Object> data) {
397 try {
398 writer.write("event: " + event + "\n");
399 writer.write("data: " + objectMapper.writeValueAsString(data) + "\n\n");
400 writer.flush();
401 } catch (final JsonProcessingException e) {
402 logger.warn("[RAG] Failed to serialize SSE data. event={}", event, e);
403 }
404 }
405
406
407
408
409
410
411
412
413
414 protected void writeJsonResponse(final HttpServletResponse response, final int status, final Map<String, Object> data)
415 throws IOException {
416 response.setStatus(status);
417 response.setContentType("application/json");
418 response.setCharacterEncoding("UTF-8");
419 response.getWriter().write(objectMapper.writeValueAsString(data));
420 }
421
422
423
424
425
426
427
428
429
430 protected Map<String, Object> createSuccessResponse(final String sessionId, final String content, final List<ChatSource> sources) {
431 final Map<String, Object> result = new HashMap<>();
432 result.put("status", "ok");
433 result.put("sessionId", sessionId);
434 result.put("content", content);
435 if (sources != null) {
436 result.put("sources", sources);
437 }
438 return result;
439 }
440
441
442
443
444
445
446
447 protected Map<String, Object> createErrorResponse(final String message) {
448 final Map<String, Object> result = new HashMap<>();
449 result.put("status", "error");
450 result.put("message", message);
451 return result;
452 }
453
454
455
456
457
458
459
460 protected String getUserId(final HttpServletRequest request) {
461 final SystemHelper systemHelper = ComponentUtil.getSystemHelper();
462 final String username = systemHelper.getUsername();
463 if (!org.codelibs.fess.Constants.GUEST_USER.equals(username)) {
464 return username;
465 }
466
467 return ComponentUtil.getUserInfoHelper().getUserCode();
468 }
469
470
471
472
473
474
475
476 protected int getMaxMessageLength(final FessConfig fessConfig) {
477 try {
478 return Integer.parseInt(fessConfig.getOrDefault("rag.chat.message.max.length", "4000"));
479 } catch (final NumberFormatException e) {
480 logger.warn("Invalid rag.chat.message.max.length config, using default 4000");
481 return 4000;
482 }
483 }
484
485
486
487
488
489
490
491
492 protected Map<String, String[]> parseFieldFilters(final HttpServletRequest request) {
493 final Map<String, String[]> fields = new HashMap<>();
494 final String[] labels = request.getParameterValues("fields.label");
495 if (labels != null && labels.length > 0) {
496
497 final Locale requestLocale = request.getLocale() != null ? request.getLocale() : Locale.ROOT;
498 final Set<String> allowedLabels = new java.util.HashSet<>();
499 ComponentUtil.getLabelTypeHelper()
500 .getLabelTypeItemList(SearchRequestParams.SearchRequestType.SEARCH, requestLocale)
501 .stream()
502 .map(m -> m.get("value"))
503 .forEach(allowedLabels::add);
504 if (!Locale.ROOT.equals(requestLocale)) {
505 ComponentUtil.getLabelTypeHelper()
506 .getLabelTypeItemList(SearchRequestParams.SearchRequestType.SEARCH, Locale.ROOT)
507 .stream()
508 .map(m -> m.get("value"))
509 .forEach(allowedLabels::add);
510 }
511 final List<String> validLabels = new ArrayList<>();
512 for (final String label : labels) {
513 if (label != null && allowedLabels.contains(label)) {
514 validLabels.add(label);
515 } else if (logger.isDebugEnabled()) {
516 logger.debug("Rejected unknown label filter value: {}", label);
517 }
518 }
519 if (!validLabels.isEmpty()) {
520 fields.put("label", validLabels.toArray(new String[0]));
521 }
522 }
523 return fields;
524 }
525
526
527
528
529
530
531
532
533 protected String[] parseExtraQueries(final HttpServletRequest request) {
534 final String[] extraQueries = request.getParameterValues("ex_q");
535 if (extraQueries == null || extraQueries.length == 0) {
536 return new String[0];
537 }
538
539 final List<FacetQueryView> facetQueryViewList = ComponentUtil.getViewHelper().getFacetQueryViewList();
540 final Set<String> allowedQueries = new java.util.HashSet<>();
541 for (final FacetQueryView view : facetQueryViewList) {
542 allowedQueries.addAll(view.getQueryMap().values());
543 }
544 final List<String> validQueries = new ArrayList<>();
545 for (final String eq : extraQueries) {
546 if (eq != null && allowedQueries.contains(eq)) {
547 validQueries.add(eq);
548 } else if (logger.isDebugEnabled()) {
549 logger.debug("Rejected unknown extra query filter value: {}", eq);
550 }
551 }
552 if (validQueries.isEmpty()) {
553 return new String[0];
554 }
555
556 final Set<String> used = new java.util.HashSet<>();
557 final List<String> groupedQueries = new ArrayList<>();
558 for (final FacetQueryView view : facetQueryViewList) {
559 final Set<String> viewValues = new java.util.HashSet<>(view.getQueryMap().values());
560 final List<String> matched = new ArrayList<>();
561 for (final String vq : validQueries) {
562 if (viewValues.contains(vq)) {
563 matched.add(vq);
564 used.add(vq);
565 }
566 }
567 if (matched.size() == 1) {
568 groupedQueries.add(matched.get(0));
569 } else if (matched.size() > 1) {
570 groupedQueries.add(String.join(" OR ", matched));
571 }
572 }
573 for (final String vq : validQueries) {
574 if (!used.contains(vq)) {
575 groupedQueries.add(vq);
576 }
577 }
578 return groupedQueries.toArray(new String[0]);
579 }
580
581 @Override
582 protected void writeHeaders(final HttpServletResponse response) {
583 ComponentUtil.getFessConfig().getApiJsonResponseHeaderList().forEach(e -> response.setHeader(e.getFirst(), e.getSecond()));
584 }
585 }