RAG + LLM项目通常面临的观测盲区和解决方案

package com.ticket.util;

import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.LinkedHashMap;
import java.util.Map;

/**
 * AI聊天链路专用计时器
 * 提供关键节点的耗时统计和结构化日志输出
 * 配合ELK全链路追踪使用
 */
@Slf4j
public class AiChatStopWatch {

    private static final DateTimeFormatter FORMATTER =
            DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");

    /** 任务名称 */
    private final String taskName;

    /** 开始时间（毫秒） */
    private long startTimeMillis;

    /** 上一个检查点时间（毫秒） */
    private long lastCheckpointMillis;

    /** 各检查点耗时记录 */
    private final LinkedHashMap<String, Long> checkpoints;

    /** 是否已完成 */
    private boolean isRunning;

    /**
     * 创建AI聊天计时器
     * @param taskName 任务名称（如"ai-chat-sync"、"ai-chat-stream"）
     */
    public AiChatStopWatch(String taskName) {
        this.taskName = taskName;
        this.checkpoints = new LinkedHashMap<>();
        this.isRunning = false;
    }

    /**
     * 开始计时
     * @return 当前实例（支持链式调用）
     */
    public AiChatStopWatch start() {
        this.startTimeMillis = System.currentTimeMillis();
        this.lastCheckpointMillis = this.startTimeMillis;
        this.isRunning = true;
        return this;
    }

    /**
     * 记录检查点并计算与上一节点的时间差
     * @param checkpointName 检查点名称（如 "route_judge", "llm_call", "db_save"）
     * @return 当前实例（支持链式调用）
     */
    public AiChatStopWatch checkpoint(String checkpointName) {
        if (!isRunning) {
            throw new IllegalStateException("请先调用 start() 开始计时");
        }
        long now = System.currentTimeMillis();
        long elapsed = now - lastCheckpointMillis;
        checkpoints.put(checkpointName, elapsed);
        this.lastCheckpointMillis = now;
        return this;
    }

    /**
     * 结束计时并输出结构化日志
     * 日志格式为JSON字符串，方便ELK解析
     */
    public void stopAndLog() {
        if (!isRunning) {
            return;
        }
        isRunning = false;
        long totalElapsed = System.currentTimeMillis() - startTimeMillis;

        // 构建结构化日志（JSON格式）
        StringBuilder sb = new StringBuilder(256);
        sb.append("[AiChatTrace] {");
        appendField(sb, "taskName", taskName);
        sb.append(", ");
        appendField(sb, "traceId", TraceContext.getTraceId());
        sb.append(", ");
        appendField(sb, "totalMs", String.valueOf(totalElapsed));
        sb.append(", ");
        appendField(sb, "startTime",
                LocalDateTime.now().minusNanos(totalElapsed * 1_000_000).format(FORMATTER));
        sb.append(", \"checkpoints\": {");

        boolean first = true;
        for (Map.Entry<String, Long> entry : checkpoints.entrySet()) {
            if (!first) {
                sb.append(", ");
            }
            first = false;
            sb.append("\"").append(entry.getKey()).append("\": ")
              .append(entry.getValue()).append("ms");
        }
        sb.append("}}");

        log.info(sb.toString());
    }

    /**
     * 获取总耗时（毫秒），不停止计时
     * @return 从start()到当前的毫秒数
     */
    public long getTotalTimeMillis() {
        if (!isRunning) {
            return 0;
        }
        return System.currentTimeMillis() - startTimeMillis;
    }

    /**
     * 获取某个检查点的耗时
     * @param checkpointName 检查点名称
     * @return 该段耗时（毫秒），未找到返回-1
     */
    public long getCheckpointTime(String checkpointName) {
        Long time = checkpoints.get(checkpointName);
        return time != null ? time : -1L;
    }

    /**
     * 向StringBuilder追加JSON字段
     */
    private void appendField(StringBuilder sb, String key, String value) {
        sb.append("\"").append(key).append("\": ")
          .append(value == null ? "\"null\"" : "\"" + value + "\"");
    }
}

2.在需要ai操作的地方去触发监听器

@PostMapping("/ask")
    public ResponseUtil.Result<ChatResponse> ask(@Valid @RequestBody ChatRequest request) {
        ensureChatTraceForRequest();
        AiChatStopWatch stopWatch = new AiChatStopWatch("ai-chat-sync").start();
        String question = request.getQuestion();
        log.info("[{}] 用户提问：{}", TraceContext.getTraceId(), question);
        String answer = aiChatService.chat(question);
        stopWatch.checkpoint("controller_total");
        ChatResponse chatResponse = new ChatResponse(answer, true, System.currentTimeMillis());
        stopWatch.stopAndLog();
        return ResponseUtil.success(chatResponse);
    }

3.然后做重要的日志记录，可以异步写入 Elasticsearch 或 Jaeger（OpenTelemetry），实现可视化调用链，因为我是在每一次请求到的时候生成的一个traceId，然后再写日志的时候顺便写进去，后台监控也能查到对应的traceId。

综合来看这个其实并不好，因为traceId完全可以在请求打进来的时候通过过滤器解析出来设置到TraceContext做一个上下文管理。

但是最重要的就是还是没有做到rag层面，prompt上的监控其实我个人觉得这个方面才是重点，后续还会修改代码。

首先，RAG层面是最能体现你的系统的检索效率和检索质量的，因为这样可以确定你的Prompt提示词如何修改能够更适配，rag检索质量更好，其次Prompt检测时更能确定性的修改你的提示词，让你能够更好的组织出一个适配你当前系统和向量数据库的提示词。

在这里我做一个补充代码，后续会加上我的项目中。

1.首当其冲的就是RAG检测

我们可以通过包装ContentRetriever，来实现日志打印。

@Slf4j
public class TracedContentRetriever implements ContentRetriever {
    private final ContentRetriever delegate;
    private final String name;
    
    @Override
    public List<Content> retrieve(Query query) {
        long start = System.currentTimeMillis();
        try {
            List<Content> contents = delegate.retrieve(query);
            log.info("RAG.retrieve name={}, query={}, resultCount={}, latencyMs={}", 
                     name, query.text(), contents.size(), System.currentTimeMillis() - start);
            return contents;
        } catch (Exception e) {
            log.error("RAG.retrieve failed name={}", name, e);
            throw e;
        }
    }
}

2.然后再这个代码修改，用上面的的TracedContentRetriever

/**
     * 检索链：首先过滤掉enabled=false的 doc，然后对剩余的 doc 进行打分，并返回打分最高的 topN 个 doc。
     * 属于稀疏检索，且结果数量多，且结果质量低。
     * @param embeddingStore 向量存储
     * @param embeddingModel 向量模型
     * @param manualKbGovernance 手册治理：检索结果由外部控制
     * @param expandMultiplier 检索结果数量：检索结果数量乘数
     * @param filteredCap 过滤后的最大数量
     * @return
     */
    @Bean
    public ContentRetriever manualRagRetriever(
            EmbeddingStore<TextSegment> embeddingStore,
            EmbeddingModel embeddingModel,
            ManualDocImpl manualKbGovernance,
            @Value("${rag.retrieval-expand-multiplier:5}") int expandMultiplier,
            @Value("${rag.filtered-cap:40}") int filteredCap) {
        int saneMax = saneMaxResults(maxResults);
        double saneMin = saneMinScore(minScore);

        int expanded = Math.min(
                saneMax * Math.max(expandMultiplier, 1),
                Math.max(filteredCap, saneMax));

        log.info("RAG检索(单路): outputMax={}, expandedRetrieve={}, minScore={}", saneMax, expanded, saneMin);

        EmbeddingStoreContentRetriever inner = EmbeddingStoreContentRetriever.builder()
                .embeddingStore(embeddingStore)
                .embeddingModel(embeddingModel)
                .maxResults(expanded)
                .minScore(saneMin)
                .build();

        return new ManualContentRetriever(inner, manualKbGovernance, saneMax);
    }

因为我这个时写死的，在管理和配置方面不方便

其实还可以让ManualContentRetriever去继承TracedContentRetriever，这样就可以少些一部分，或者，直接把TracedContentRetriever 中的retriever修改进ManualContentRetriever，因为我这边后续还有一个混合索引的操作，其实更好的是把日志部分的配置，还有输出统一写道ManualContentRetriever里面的。

Token 与成本失控

对于一个商业项目，每次对话消耗多少 Token 看不到，预算超了都不知道，这个是最危险的，因为这决定了一个用户可以通过这种方式去攻击我们的LLM服务，导致整体业务宕机或者token耗尽。

解决方案

1.我是实现了一个服务的chat接口，代码如下

package com.ticket.aichat.service;

import dev.langchain4j.service.Result;
import dev.langchain4j.service.SystemMessage;
import dev.langchain4j.service.UserMessage;

/**
 * 客服 AI 服务接口（LangChain4j 生成实现）
 */
public interface KnowledgeAssistant {

    @SystemMessage("""
            你是一个...""")
    @UserMessage("{{it}}")
    Result<String> chat(String question);
}

然后LangChain4j会返回一个Result<String>，这是LangChain4j为我们提供的类，源码是这样的

//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//

package dev.langchain4j.service;

import dev.langchain4j.internal.Utils;
import dev.langchain4j.internal.ValidationUtils;
import dev.langchain4j.model.output.FinishReason;
import dev.langchain4j.model.output.TokenUsage;
import dev.langchain4j.rag.content.Content;
import dev.langchain4j.service.tool.ToolExecution;
import java.util.List;

public class Result<T> {
    private final T content;
    private final TokenUsage tokenUsage;
    private final List<Content> sources;
    private final FinishReason finishReason;
    private final List<ToolExecution> toolExecutions;

    public Result(T content, TokenUsage tokenUsage, List<Content> sources, FinishReason finishReason, List<ToolExecution> toolExecutions) {
        this.content = ValidationUtils.ensureNotNull(content, "content");
        this.tokenUsage = tokenUsage;
        this.sources = Utils.copyIfNotNull(sources);
        this.finishReason = finishReason;
        this.toolExecutions = Utils.copyIfNotNull(toolExecutions);
    }

    public static <T> ResultBuilder<T> builder() {
        return new ResultBuilder();
    }

    public T content() {
        return this.content;
    }

    public TokenUsage tokenUsage() {
        return this.tokenUsage;
    }

    public List<Content> sources() {
        return this.sources;
    }

    public FinishReason finishReason() {
        return this.finishReason;
    }

    public List<ToolExecution> toolExecutions() {
        return this.toolExecutions;
    }

    public static class ResultBuilder<T> {
        private T content;
        private TokenUsage tokenUsage;
        private List<Content> sources;
        private FinishReason finishReason;
        private List<ToolExecution> toolExecutions;

        ResultBuilder() {
        }

        public ResultBuilder<T> content(T content) {
            this.content = content;
            return this;
        }

        public ResultBuilder<T> tokenUsage(TokenUsage tokenUsage) {
            this.tokenUsage = tokenUsage;
            return this;
        }

        public ResultBuilder<T> sources(List<Content> sources) {
            this.sources = sources;
            return this;
        }

        public ResultBuilder<T> finishReason(FinishReason finishReason) {
            this.finishReason = finishReason;
            return this;
        }

        public ResultBuilder<T> toolExecutions(List<ToolExecution> toolExecutions) {
            this.toolExecutions = toolExecutions;
            return this;
        }

        public Result<T> build() {
            return new Result(this.content, this.tokenUsage, this.sources, this.finishReason, this.toolExecutions);
        }

        public String toString() {
            ....
        }
    }
}

在这里面我们看到，就有tokenusage等一些相关方法可以使用。

然后我们使用tokenUsage()，就可以获取对应的token用量了，接下来是我的代码。

package com.ticket.aichat.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.ticket.dto.mq.ChatRecordEvent;
import com.ticket.entity.ChatRecord;
import com.ticket.entity.ChatSession;
import com.ticket.entity.KnowledgeBase;
import com.ticket.aichat.mapper.ChatRecordMapper;
import com.ticket.aichat.service.AIChatService;
import com.ticket.aichat.service.KnowledgeAssistant;
import com.ticket.service.RocketMQProducerService;
import com.ticket.aichat.service.ChatSessionService;
import com.ticket.aichat.service.HumanTransferPublisher;
import com.ticket.aichat.service.QuickFaqMatchService;
import com.ticket.util.MQIdempotentUtil;
import dev.langchain4j.memory.ChatMemory;
import dev.langchain4j.service.Result;
import com.ticket.util.UserContext;
import com.ticket.util.TraceContext;
import com.ticket.util.AiChatStopWatch;
import com.ticket.util.TokenCountUtil;
import com.ticket.enums.BusinessStatus;
import dev.langchain4j.service.UserMessage;
import reactor.core.publisher.Flux;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;

/** * 负责处理AI聊天业务逻辑，调用底层AI服务
 */
@Service
public class AIChatServiceImpl implements AIChatService {

    private static final Logger log = LoggerFactory.getLogger(AIChatServiceImpl.class);
    @Resource
    private KnowledgeAssistant knowledgeAssistant;

    @Resource
    private ChatRecordMapper chatRecordMapper;

    @Resource
    private ChatMemory chatMemory;

    @Resource
    private ChatSessionService chatSessionService;

    @Resource
    private RocketMQProducerService rocketMQProducerService;

    @Resource
    private MQIdempotentUtil idempotentUtil;

    @Resource
    private HumanTransferPublisher humanTransferPublisher;

    @Resource
    private QuickFaqMatchService quickFaqMatchService;
    @Override
    @Transactional
    public String chat(@UserMessage String question) {
       ..........

        // 调用AI（Result<String>获取TokenUsage）
        String answer;
        int inputTokens = 0;
        int outputTokens = 0;
        try {
            Result<String> result = knowledgeAssistant.chat(question);
            answer = result.content();
            if (answer == null || answer.isBlank()) {
                answer = EMPTY_MODEL_FALLBACK;
            }
            if (result.tokenUsage() != null) {
                inputTokens = result.tokenUsage().inputTokenCount();
                outputTokens = result.tokenUsage().outputTokenCount();
                log.info("[{}] Token消耗 - input:{}, output:{}",
                        TraceContext.getTraceId(), inputTokens, outputTokens);
            } else {
                // 回退：使用估算
                outputTokens = TokenCountUtil.estimate(answer);
                log.warn("[{}] 无法获取精确TokenUsage，使用估算值 output:{}",
                        TraceContext.getTraceId(), outputTokens);
            }
        } catch (IllegalStateException e) {
            log.error("[{}] RAG服务不可用，返回降级消息: {}");
            outputTokens = TokenCountUtil.estimate(answer);
        } catch (Exception e) {
            log.error("[{}] AI 调用异常: {}", TraceContext.getTraceId(), e.getMessage(), e);
            answer = EMPTY_MODEL_FALLBACK;
            outputTokens = TokenCountUtil.estimate(answer);
        }
        stopWatch.checkpoint("llm_call");
        saveChatRecord(answer, BusinessStatus.MSG_TYPE_ROBOT, BigDecimal.ONE, sessionId, null, 0, inputTokens, outputTokens);
        stopWatch.checkpoint("db_save_ai");
        stopWatch.stopAndLog();
        return answer;
    }


    /**
     * 保存聊天记录（异步：通过RocketMQ发送，由消费者负责DB写入和WS广播）
     */
    private void saveChatRecord(String message, String msgType, BigDecimal confidence,
                                String sessionId, Long employeeId, Integer isRead,
                                int inputTokens, int outputTokens) {
        try {
            ChatRecordEvent event = new ChatRecordEvent();
            event.setMessageId(idempotentUtil.generateMessageId());
            event.setUserId(UserContext.getCurrentUserId());
            event.setSessionId(sessionId);
            event.setMessage(message);
            event.setMsgType(msgType);
            event.setEmployeeId(employeeId);
            event.setConfidence(confidence);
            event.setInputTokens(inputTokens);
            event.setOutputTokens(outputTokens);
            event.setCreatedAt(LocalDateTime.now());
            event.setTraceId(TraceContext.getTraceId());
            rocketMQProducerService.sendChatRecordEvent(event);
            log.debug("已发送聊天记录到RocketMQ异步处理: userId={}, msgType={}",
                    UserContext.getCurrentUserId(), msgType);
        } catch (Exception e) {
            log.error("发送聊天记录事件失败", e);
        }
    }

  
}

这里我们还可以进一步做操作，比如说，添加配置类，去限制用户级别和服务级别的Token额度，避免超额。

其次我们还要添加一个降级服务，如果已经限额了，或者服务异常了，我们可以通过降级维护服务，确保其正确运行。

后面其实还可以根据用户token的使用，去做一些个性化的操作，或者去做日志分析等等，都是一个可靠的依据

对话质量无法评估

这个问题主要是因为LLM的回答不一定是按照我们的要求去回答的。我们其实可以去做一些询问，比如：答案是否忠实于检索内容？是否幻觉？是否回答了用户问题？

解决方案

第一个方案就是给予一个用户的反馈按钮，我们可以在每次回答后面添加一个评价，让用户去评价是否解决问题了，然后将答案收集起来，然后后台定期做分析，然后投诉的多，我们就知道，大模型回答有问题，可以选择优化提示词，rag检索方向，模型选择等等方面去做优化，但是这一点不好的就是用户如果经常体验感不好的话，就不会使用该服务了，对于维护客户来说是不好的。

第二个方案就是使用一个更强大的 LLM（如 GPT-5.5等等强大的模型）对随机抽样的对话进行 幻觉检测 和 相关性评分，为每个回答存储评估得分，以便后续分析哪些检索策略、模型或 prompt 效果更好。

第三个方案就是人工检查，这种方案的成本高，精力消耗大，不适合做短时间的评估，但是对于大批量的评估综合，或者质量检查来说，人工检查更加准确。

缺乏调试入口

这个的主要问题就是：如果发现某个问题回答不好，无法快速复现场景、调整 Prompt 并验证。

解决方案

1.可以添加一个调试接口，基于之前实现的TraceId去获取整个的回答流程，麻烦一点的可以搭一个内部的监控平台，利用平台去管理整个的服务，但是成本太高了。

2.可以在日志中输出 Prompt，利用之前的ELK去检查哪一个prompt有问题，然后通过调试接口去测试是否复现，然后再去修改Prompt

总结

在RAG+LLM的项目中，AI的可观测性，可控性是最重要的，如果一个商业项目，对AI服务有很好的监控性，是能够将可用变成可靠，降本增效的一个很好的服务。

2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

企业AI科普：企业AI是什么，企业通常用它解决哪些工作问题

企业AI，是面向组织业务场景构建的人工智能应用集合，区别于面向个人用户的消费级AI工具。它依托大语言模型、多模态能力、智能体编排、知识库检索等底层技术，嵌入企业现有办公、业务流程，承接团队协作、信息处理、内容生产、数据分析等工作任务。企业AI不是单一聊天窗口，而是一套可以对接组织内部知识、权限体系与第三方工具的应用体系。从技术形态上划分，企业AI包含基础对话助手、知识库问答、自动化工作流，以及当下

2048 AI社区

工程架构演化总结

分离之后，不太占用资源的前端，只需要冗余2个，便可以稳定运行，不需要为它扩展太多资源。现在AI开始流行，引入了AI智能体，AI智能体是一种AI工程的架构，它包括四个部分，分别为工具、记忆、规划、执行。微服务架构，是为了节省资源，把资源更多用在瓶颈服务上，因此把服务拆得更细，这其中拆服务也是借鉴了领域驱动开发的思想。云原生架构，并没有颠覆前两种架构，它是部署架构的升级，通过云原生，它的目标是对资源进

2048 AI社区

Gateway 的设计核心是 “多入口 + 单内核” 架构

每台主机上只有一个 Gateway，它垄断了与这些外部平台的会话，例如 WhatsApp 的登录会话只能由它开启。，将其注册为系统服务（macOS 的 LaunchAgent 或 Linux 的 systemd 服务），实现 7×24 小时常驻。理解 Gateway 的最佳方式，是看它在一次完整请求中的位置。，将来自不同渠道的请求，分发给正确的 Agent 去处理，再将结果返回给对应的渠道。将消息