AI 客服笔记(DeepSeek Function Calling + Redis 会话 + SSE 流式)

AI 客服笔记(DeepSeek Function Calling + Redis 会话 + SSE 流式)

〇、功能与架构总览

AI 客服(挂载在 mall-user)能让用户用自然语言找商品、看商品详情、要推荐、查自己的订单。它不是简单调一次大模型,而是:

  1. Function Calling(函数调用):把”搜索商品””查订单”等能力以工具(Tool)的形式声明给 DeepSeek,模型自己判断何时调用、带什么参数,服务端执行后把结果喂回去,模型再组织成最终回答——实时数据全部来自数据库,模型只负责理解和表达,禁止编造。
  2. Redis 会话记忆:每轮对话历史按 chat:session:{userId}:{sessionId} 存入 Redis(TTL 24h),多轮对话有上下文(”再便宜点的”这类指代能被理解)。
  3. SSE 流式响应:AI 回复逐段推送到前端打字机式渲染,而不是等全部生成完再返回。
用户 → ui-app AI客服页 ──fetch POST /api/user/chat/stream──► 网关(验JWT,透传用户头)
                                                              ▼
                                                      mall-user ChatController
                                                              ▼
                                              ChatService(会话记忆 + 工具调度编排)
                                                    │            ▲
                                                    ▼            │ tool结果
                                          DeepSeekService      executeTool()
                                          (WebClient→DeepSeek)      │
                                                    ▲               ▼
                                                    └──── Function Calling 循环(≤3轮)
                                                                    │
                          ┌──────────────────┬──────────────────────┴──────────┐
                          ▼                  ▼                                   ▼
                   search_products     get_product_detail / get_recommendations   query_order
                   ProductToolService ──Feign──► product-service    OrderToolService ──Feign──► order-service

一、前提认知:DeepSeek Chat Completions 协议

DeepSeek 兼容 OpenAI 的 Chat Completions 协议。一次请求的核心是消息列表(messages),每条消息有 role:

role 含义 本项目中的使用
system 系统提示词,设定 AI 人设与规则 “你是思途商城的AI客服……”,每次请求固定放第一条
user 用户输入 当前消息 + 历史里的用户消息
assistant 模型回复 可能带 content(正文),也可能带 tool_calls(请求调工具)
tool 工具执行结果 必须携带 tool_call_id 与请求调用的 assistant 消息对应

Function Calling 一轮交互的完整序列:

[system, ...历史, user"帮我找100元以内的手机"]
    ↓ 模型判断需要先搜商品,返回:
assistant: {tool_calls:[{id:"call_1", function:{name:"search_products", arguments:"{\"max_price\":100}"}}]}
    ↓ 服务端执行工具,把结果作为tool消息追加:
tool: {tool_call_id:"call_1", content:"{\"products\":[{...}]}"}      ← 结果是JSON字符串
    ↓ 带着"工具已执行"的上下文再次调用模型,返回:
assistant: {content:"为您找到以下100元以内的商品:1. xxx(¥xx)..."}   ← 最终回答

流式模式(stream:true)下,响应变成一串 SSE chunk:正文以 delta.content 增量下发;工具调用也是增量的(id/name 先到,arguments 分多段拼接),需要服务端累积。

二、DTO 协议层(mall-user/.../dto/)

DeepSeekMessage——统一消息模型

@Data
@NoArgsConstructor
@AllArgsConstructor
@JsonInclude(JsonInclude.Include.NON_NULL)   // null字段不序列化:user消息不带tool_call_id
public class DeepSeekMessage {

    /** 角色:system / user / assistant / tool */
    private String role;
    private String content;

    @JsonProperty("tool_calls")              // 蛇形命名对齐API协议
    private List<DeepSeekToolCall> toolCalls;

    @JsonProperty("tool_call_id")
    private String toolCallId;

    /** 纯文本消息构造(system / user / assistant) */
    public DeepSeekMessage(String role, String content) {
        this.role = role;
        this.content = content;
    }
}

DeepSeekTool——工具声明(发给API的JSON Schema)

@Data
@AllArgsConstructor
public class DeepSeekTool {
    private String type;                     // 固定 "function"
    private DeepSeekFunctionDef function;

    @Data
    @AllArgsConstructor
    public static class DeepSeekFunctionDef {
        private String name;                 // 函数名
        private String description;          // 描述——模型靠它决定何时调用
        private Map<String, Object> parameters;  // JSON Schema: 参数名/类型/描述/必填
    }
}

description 是 Function Calling 的灵魂:模型完全靠参数和字段的描述文字来决定”要不要调、传什么值”。描述写得含糊,模型就会乱传参数或不该调时调。

其余 DTO

  • DeepSeekToolCall{id, type, function:DeepSeekFunction{name, arguments}}——模型返回的调用请求,arguments 是 JSON 字符串(不是对象)
  • DeepSeekResponse{choices[{message, finish_reason}], usage}——阻塞式响应
  • DeepSeekChunk{choices[{delta{content, tool_calls}, finish_reason}]}——流式增量(@JsonIgnoreProperties(ignoreUnknown=true) 防新字段报错)
  • ChatRequest{sessionId, message} / ChatResponse{sessionId, reply}——前后端会话契约

三、DeepSeekService:大模型 HTTP 客户端

@Slf4j
@Component
public class DeepSeekService {

    @Value("${deepseek.api-key}")
    private String apiKey;
    @Value("${deepseek.url}")         // https://api.deepseek.com/v1/chat/completions
    private String apiUrl;
    @Value("${deepseek.model}")       // deepseek-chat
    private String model;
    @Value("${deepseek.max-tokens}")
    private int maxTokens;
    @Value("${deepseek.temperature}")
    private double temperature;

    private final WebClient webClient;

    public DeepSeekService() {
        this.webClient = WebClient.builder()
                .codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(10 * 1024 * 1024))
                .build();
    }

    /** 阻塞式调用(支持 Function Calling),一次性返回完整assistant消息 */
    public DeepSeekMessage call(List<DeepSeekMessage> messages, List<DeepSeekTool> tools) {
        Map<String, Object> body = Map.of(
                "model", model,
                "messages", messages,
                "tools", tools != null ? tools : List.of(),
                "max_tokens", maxTokens,
                "temperature", temperature
        );

        DeepSeekResponse response = webClient.post()
                .uri(apiUrl)
                .header("Authorization", "Bearer " + apiKey)
                .contentType(MediaType.APPLICATION_JSON)
                .bodyValue(body)
                .retrieve()
                .onStatus(
                        status -> status.isError(),
                        clientResponse -> clientResponse.bodyToMono(String.class)
                                .flatMap(errBody -> {
                                    log.error("DeepSeek API调用失败: status={}, body={}",
                                            clientResponse.statusCode(), errBody);
                                    return Mono.error(new RuntimeException("AI服务暂不可用"));
                                })
                )
                .bodyToMono(DeepSeekResponse.class)
                .block();            //Servlet栈里同步等待结果

        if (response == null || response.getChoices() == null || response.getChoices().isEmpty()) {
            log.error("DeepSeek返回空响应");
            throw new RuntimeException("AI返回为空");
        }
        return response.getChoices().get(0).getMessage();
    }

    /** 流式调用(stream=true),返回SSE原始data载荷的Flux,最后一个是字面量"[DONE]" */
    public Flux<String> streamCall(List<DeepSeekMessage> messages, List<DeepSeekTool> tools) {
        Map<String, Object> body = Map.of(
                "model", model,
                "messages", messages,
                "tools", tools != null ? tools : List.of(),
                "max_tokens", maxTokens,
                "temperature", temperature,
                "stream", true
        );

        return webClient.post()
                .uri(apiUrl)
                .header("Authorization", "Bearer " + apiKey)
                .contentType(MediaType.APPLICATION_JSON)
                .accept(MediaType.TEXT_EVENT_STREAM)
                .bodyValue(body)
                .retrieve()
                .onStatus(status -> status.isError(), /* 同上 */)
                .bodyToFlux(String.class);   //每个元素=一个SSE data载荷
    }
}

要点:

  • 为什么用 WebClient(WebFlux)而不是 RestTemplate:流式响应需要响应式流(Flux)逐段消费,Flux<String> 天然把 SSE 拆成逐元素的数据流;阻塞式接口复用同一个客户端。WebFlux 依赖与现有 Servlet 栈共存——Spring Boot 检测到 spring-webmvc 仍按 Servlet 应用启动,WebClient 只是作为 HTTP 客户端使用。
  • api-key 走 Authorization: Bearer 头,密钥从环境变量 DEEPSEEK_API_KEY 注入,不明文写死(yml 里的默认值是开发便利,生产应去掉)。

配置(mall-user/application.yml):

deepseek:
  api-key: ${DEEPSEEK_API_KEY:sk-...}
  url: ${DEEPSEEK_URL:https://api.deepseek.com/v1/chat/completions}
  model: ${DEEPSEEK_MODEL:deepseek-chat}
  max-tokens: 2048
  temperature: 0.7

四、ChatService:会话 + 工具调度编排(核心)

1. 阻塞式主流程 processMessage

public ChatResponse processMessage(String sessionId, String message) {
    Long userId = (Long) LoginContext.getLoginInfo().get("id");   //当前登录人

    if (sessionId == null || sessionId.isBlank()) {
        sessionId = UUID.randomUUID().toString();                  //首次自动生成
    }

    List<DeepSeekMessage> history = loadConversation(userId, sessionId);  //Redis加载历史
    List<DeepSeekTool> tools = buildTools();

    //组装完整消息列表:system prompt + 历史 + 当前用户消息
    List<DeepSeekMessage> messages = new ArrayList<>();
    messages.add(buildSystemPrompt());
    messages.addAll(history);
    messages.add(new DeepSeekMessage("user", message));

    // Function Calling 循环(最多 3 轮,防止模型反复调工具不收口)
    for (int i = 0; i < MAX_TOOL_CALL_ITERATIONS; i++) {
        DeepSeekMessage assistantMsg = deepSeekService.call(messages, tools);

        if (assistantMsg.getToolCalls() != null && !assistantMsg.getToolCalls().isEmpty()) {
            messages.add(assistantMsg);                            //先记下"我要调工具"这条
            for (DeepSeekToolCall toolCall : assistantMsg.getToolCalls()) {
                String toolResult = executeTool(toolCall);         //真正执行工具
                messages.add(new DeepSeekMessage("tool", toolResult, null, toolCall.getId()));
            }
            //继续循环:让模型基于工具结果继续回答
        } else {
            messages.add(assistantMsg);                            //最终回复
            saveConversation(userId, sessionId, messages.subList(1, messages.size()));
            //subList(1,...)跳过system prompt——它每轮都重新构造,不需要存
            return new ChatResponse(sessionId, assistantMsg.getContent() != null
                    ? assistantMsg.getContent() : "");
        }
    }
    return new ChatResponse(sessionId, "抱歉,暂时无法处理您的问题,请稍后再试。");
}

为什么要循环:模型一轮只能”请求调一次工具”,拿到结果后它可能还要再调(比如先查商品再查订单),所以循环直到模型给出不带 tool_calls 的最终回复;上限 3 轮防失控。

2. System Prompt

private DeepSeekMessage buildSystemPrompt() {
    String prompt = """
            你是"思途商城"的AI客服助手,服务本商城的注册用户。
            你可以:回答购物常见问题、搜索商品、查询商品详情、推荐商品、查询用户自己的订单。

            可用工具:
            1. search_products —— 用户想找商品时使用,支持关键字、价格区间、分类过滤
            2. get_product_detail —— 需要某个商品的详细信息时使用
            3. get_recommendations —— 用户没有明确目标、想逛逛时,推荐最新商品
            4. query_order —— 用户查询订单时使用,订单号只能从用户提供的信息里取

            规则:
            - 回复热情、专业、简洁,推荐商品时列出名称和价格
            - 涉及实时数据(商品、价格、库存、订单状态)必须调用工具获取,禁止编造
            - 工具返回error时,如实告知用户并建议稍后再试
            - 与购物无关或需要人工处理的问题(如退款纠纷),引导用户联系人工客服
            """;
    return new DeepSeekMessage("system", prompt.trim());
}

3. 工具定义(JSON Schema)

private List<DeepSeekTool> buildTools() {
    return List.of(
            new DeepSeekTool("function", new DeepSeekTool.DeepSeekFunctionDef(
                    "search_products",
                    "根据关键字、价格区间、分类搜索商品",
                    Map.of(
                            "type", "object",
                            "properties", Map.of(
                                    "keyword", Map.of("type", "string", "description", "搜索关键字(商品名称)"),
                                    "min_price", Map.of("type", "number", "description", "最低价格"),
                                    "max_price", Map.of("type", "number", "description", "最高价格"),
                                    "category_id", Map.of("type", "integer", "description", "分类ID"),
                                    "limit", Map.of("type", "integer", "description", "返回条数,默认5")
                            ),
                            "required", List.of()
                    )
            )),
            new DeepSeekTool("function", new DeepSeekTool.DeepSeekFunctionDef(
                    "get_product_detail", "根据商品ID获取商品详细信息",
                    Map.of("type", "object",
                           "properties", Map.of("product_id", Map.of("type", "integer", "description", "商品ID")),
                           "required", List.of("product_id")))),
            new DeepSeekTool("function", new DeepSeekTool.DeepSeekFunctionDef(
                    "get_recommendations", "获取商城推荐/最新商品",
                    Map.of("type", "object", "properties", Map.of(), "required", List.of()))),
            new DeepSeekTool("function", new DeepSeekTool.DeepSeekFunctionDef(
                    "query_order", "根据订单号查询用户自己的订单信息",
                    Map.of("type", "object",
                           "properties", Map.of("order_id", Map.of("type", "string", "description", "订单号")),
                           "required", List.of("order_id"))))
    );
}

4. 工具执行器

private String executeTool(DeepSeekToolCall toolCall) {
    String funcName = toolCall.getFunction().getName();
    String arguments = toolCall.getFunction().getArguments();
    try {
        switch (funcName) {
            case "search_products": {
                Map<String, Object> args = objectMapper.readValue(arguments,
                        new TypeReference<Map<String, Object>>() {});
                String keyword = (String) args.get("keyword");
                Number minPrice = (Number) args.get("min_price");
                Number maxPrice = (Number) args.get("max_price");
                Number categoryId = (Number) args.get("category_id");
                Number limit = (Number) args.getOrDefault("limit", 5);
                return productToolService.searchProducts(
                        keyword,
                        minPrice != null ? minPrice.doubleValue() : null,
                        maxPrice != null ? maxPrice.doubleValue() : null,
                        categoryId != null ? categoryId.longValue() : null,
                        limit.intValue());
            }
            case "get_product_detail": { /* 同理解析product_id后调productToolService */ }
            case "get_recommendations": { return productToolService.getRecommendations(5); }
            case "query_order": { /* 解析order_id后调orderToolService */ }
            default:
                return "{\"error\": \"未知工具: " + funcName + "\"}";
        }
    } catch (Exception e) {
        log.error("工具调用失败: funcName={}, args={}", funcName, arguments, e);
        return "{\"error\": \"工具执行失败,请稍后再试\"}";
    }
}

要点:工具执行永远不向上抛异常——失败也返回结构化的 {"error": ...} JSON 喂给模型,让它向用户解释,而不是让整次会话 500。

五、工具的数据来源:跨服务 Feign(与参考项目的关键差异)

参考项目(user_service)的工具直接用 JdbcTemplate 查本库的商品/订单表。本项目是微服务架构,商品在 product-service、订单在 order-service,工具改为通过 Feign 跨服务取数:

// ProductClient(mall-api):AI搜索工具的数据源
@GetMapping("/product/search")
Result<List<Product>> search(@RequestParam("keyword") String keyword,
                             @RequestParam("minPrice") BigDecimal minPrice,
                             @RequestParam("maxPrice") BigDecimal maxPrice,
                             @RequestParam("categoryId") Long categoryId,
                             @RequestParam("limit") Integer limit);

// OrderClient(mall-api,新建)
@FeignClient(value = "order-service")
public interface OrderClient {
    @GetMapping("/order/detail/{orderNo}")
    Result<OrderVO> detail(@PathVariable("orderNo") Long orderNo);
}

配套的两个安全/架构要点:

  1. 用户只能查自己的订单:/order/detail 在 order-service 内校验 order.userId == 当前登录人(从网关透传的用户头取得)。Feign 调用时 DefaultFeignConfig 的拦截器自动把这些头带上——即使模型被提示注入诱导去查别人的订单号,服务端校验也会拒绝。
  2. mall-api 里的镜像 DTO:mall-order 的 OrderVO 不能被 mall-api 反向引用(会循环依赖),所以在 mall-api 定义字段名一致的 OrderVO/OrderItemVO 镜像类,JSON 按字段名映射。

工具返回给模型的结果做了字段裁剪(只留 id/name/price/stock/subtitle),控制 token 消耗。

六、SSE 流式链路

1. ChatService:流式递归编排

/**
 * SSE事件协议:
 * event=meta  data={"sessionId":"..."} —— 建立会话(首个事件)
 * event=delta data="文本增量"(JSON编码) —— AI回复逐段推送
 * event=done  —— 回复结束(会话已持久化)
 * event=error data=提示文本 —— 出错兜底
 */
public Flux<ServerSentEvent<String>> processMessageStream(String sessionId, String message) {
    return Flux.defer(() -> {
        Long userId = (Long) LoginContext.getLoginInfo().get("id");

        //lambda内引用的局部变量必须是final,所以重新赋给一个final变量
        final String sid = (sessionId == null || sessionId.isBlank())
                ? UUID.randomUUID().toString() : sessionId;

        List<DeepSeekMessage> history = loadConversation(userId, sid);
        List<DeepSeekMessage> messages = new ArrayList<>();
        messages.add(buildSystemPrompt());
        messages.addAll(history);
        messages.add(new DeepSeekMessage("user", message));

        ServerSentEvent<String> meta = ServerSentEvent
                .<String>builder(JSONUtil.toJsonStr(Map.of("sessionId", sid)))
                .event("meta").build();

        return Flux.concat(Flux.just(meta), streamLoop(userId, sid, messages, 0))
                .onErrorResume(e -> {
                    log.error("AI流式会话异常", e);
                    return Flux.just(ServerSentEvent
                            .<String>builder("\"服务繁忙,请稍后再试\"")
                            .event("error").build());
                });
    });
}

private Flux<ServerSentEvent<String>> streamLoop(Long userId, String sessionId,
                                                 List<DeepSeekMessage> messages, int depth) {
    return Flux.defer(() -> {
        //超过最大工具轮次:直接给兜底文本,防止无限递归
        if (depth >= MAX_TOOL_CALL_ITERATIONS) {
            return Flux.just(
                    ServerSentEvent.<String>builder(JSONUtil.toJsonStr("抱歉,暂时无法处理您的问题,请稍后再试。"))
                            .event("delta").build(),
                    ServerSentEvent.<String>builder("").event("done").build());
        }

        StringBuilder contentBuf = new StringBuilder();                  //累积回复内容
        Map<Integer, DeepSeekToolCall> toolCallBuf = new LinkedHashMap<>(); //按index累积tool_calls
        AtomicInteger toolCallCount = new AtomicInteger();               //>0表示本轮是工具调用轮

        Flux<ServerSentEvent<String>> deltas = deepSeekService
                .streamCall(messages, buildTools())
                .takeWhile(payload -> !"[DONE]".equals(payload))         //流结束标记
                .map(this::parseChunk)
                .filter(Objects::nonNull)
                .map(chunk -> {
                    DeepSeekChunk.ChunkChoice choice = (chunk.getChoices() == null
                            || chunk.getChoices().isEmpty()) ? null : chunk.getChoices().get(0);
                    if (choice == null) return null;
                    DeepSeekChunk.Delta delta = choice.getDelta();

                    //正文增量:透传给前端
                    if (delta != null && delta.getContent() != null && !delta.getContent().isEmpty()) {
                        contentBuf.append(delta.getContent());
                        return ServerSentEvent.<String>builder(JSONUtil.toJsonStr(delta.getContent()))
                                .event("delta").build();
                    }
                    //工具调用增量:累积,不透传
                    if (delta != null && delta.getToolCalls() != null) {
                        for (DeepSeekChunk.DeltaToolCall dtc : delta.getToolCalls()) {
                            int idx = dtc.getIndex() == null ? 0 : dtc.getIndex();
                            DeepSeekToolCall acc = toolCallBuf.computeIfAbsent(idx, k -> {
                                toolCallCount.incrementAndGet();
                                DeepSeekToolCall t = new DeepSeekToolCall();
                                t.setType("function");
                                t.setFunction(new DeepSeekFunction());
                                return t;
                            });
                            if (dtc.getId() != null) acc.setId(dtc.getId());
                            if (dtc.getFunction() != null) {
                                if (dtc.getFunction().getName() != null)
                                    acc.getFunction().setName(dtc.getFunction().getName());
                                String args = acc.getFunction().getArguments();
                                String piece = dtc.getFunction().getArguments();
                                acc.getFunction().setArguments(args == null ? piece : args + piece); //分片拼接
                            }
                        }
                        return null;
                    }
                    return null;
                })
                .filter(Objects::nonNull);

        //流结束后:要么执行工具递归下一轮,要么持久化会话并收尾
        Mono<Flux<ServerSentEvent<String>>> completion = Mono.fromCallable(() -> {
            if (toolCallCount.get() > 0) {
                DeepSeekMessage assistantMsg = new DeepSeekMessage(
                        "assistant", null, new ArrayList<>(toolCallBuf.values()), null);
                messages.add(assistantMsg);
                for (DeepSeekToolCall toolCall : assistantMsg.getToolCalls()) {
                    messages.add(new DeepSeekMessage("tool", executeTool(toolCall), null, toolCall.getId()));
                }
                return streamLoop(userId, sessionId, messages, depth + 1);   //递归下一轮
            }
            messages.add(new DeepSeekMessage("assistant", contentBuf.toString()));
            saveConversation(userId, sessionId, messages.subList(1, messages.size()));
            return Flux.just(ServerSentEvent.<String>builder("").event("done").build());
        });

        return Flux.concat(deltas, completion.flatMapMany(f -> f));
    });
}

设计要点:

  • content 增量即时透传,tool_calls 增量静默累积:流式下模型若决定调工具,arguments 分很多小片到达,必须在服务端拼完整才能执行;这段时间用户端只看到”正在思考…”。
  • contentBuf/toolCallBuf 用普通局部变量:Reactor 默认串行执行(无并发订阅),副作用安全;比 AtomicReference 包裹可读性好。
  • 递归通过 Mono.fromCallable + flatMapMany 拼接:concat 保证只有上一轮流结束才订阅下一轮。
  • delta 用 JSON 编码(JSONUtil.toJsonStr(text)):文本里的换行若直接进 SSE data 会破坏事件边界,JSON 编码后 \n 变成 \\n,前端 JSON.parse 还原。

2. ChatController

//阻塞式:一次性返回完整回复(保留作为兼容)
@PostMapping("/send")
public Result<ChatResponse> send(@RequestBody ChatRequest request) {
    return Result.ok(chatService.processMessage(request.getSessionId(), request.getMessage()));
}

//流式:SSE
@PostMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> stream(@RequestBody ChatRequest request) {
    return chatService.processMessageStream(request.getSessionId(), request.getMessage());
}

控制器挂在 /user/chat——复用网关已有的 /user/** 路由与 JWT 鉴权,AI 客服仅登录用户可用。

七、前端:fetch 手动解析 SSE(Assistant.vue)

不能用 EventSource——它只支持 GET 且不能带自定义 body/请求头;这里是 POST + 需要 Authorization 头,所以用 fetch + ReadableStream 手动解析:

const send = async () => {
  const text = input.value.trim()
  if (!text || sending.value) return
  sending.value = true
  input.value = ''
  messages.value.push({role: 'user', content: text})

  const aiMsg = {role: 'assistant', content: ''}   //AI回复占位,流式往里追加
  messages.value.push(aiMsg)
  scrollToBottom()

  try {
    const resp = await fetch('/api/user/chat/stream', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': tokenStore.token || ''    //fetch不会走axios拦截器,token手动带
      },
      body: JSON.stringify({sessionId: sessionId.value, message: text})
    })
    if (!resp.ok) {
      aiMsg.content = resp.status === 401 ? '请先登录' : '服务异常,请稍后再试'
      return
    }
    //逐块读取响应流,按空行切分SSE事件
    const reader = resp.body.getReader()
    const decoder = new TextDecoder('utf-8')
    let buffer = ''
    while (true) {
      const {done, value} = await reader.read()
      if (done) break
      buffer += decoder.decode(value, {stream: true})   //注意stream:true,多字节中文不乱码
      let idx
      while ((idx = buffer.indexOf('\n\n')) >= 0) {     //SSE事件以空行分隔
        handleEvent(buffer.slice(0, idx), aiMsg)
        buffer = buffer.slice(idx + 2)
      }
    }
  } catch (e) {
    if (!aiMsg.content) aiMsg.content = '网络异常,请稍后再试'
  } finally {
    sending.value = false
    if (!aiMsg.content) aiMsg.content = '抱歉,暂时无法处理您的问题,请稍后再试。'
    scrollToBottom()
  }
}

//解析单个SSE事件块
const handleEvent = (raw, aiMsg) => {
  const lines = raw.split('\n')
  let event = 'message'
  const dataLines = []
  for (const line of lines) {
    if (line.startsWith('event:')) event = line.slice(6).trim()
    else if (line.startsWith('data:')) dataLines.push(line.slice(5).replace(/^ /, ''))
  }
  const data = dataLines.join('\n')          //SSE规范:多行data用\n连接
  if (event === 'meta') {
    sessionId.value = JSON.parse(data).sessionId
  } else if (event === 'delta') {
    aiMsg.content += JSON.parse(data)        //后端JSON编码的增量,parse还原
    scrollToBottom()
  } else if (event === 'error') {
    aiMsg.content += JSON.parse(data)
  }
}

UI 细节:AI 气泡流式渲染时末尾带一个闪烁光标 ▌;首个增量到达前显示”正在思考…”(工具调用轮不产生 delta,这段时间全靠它)。

八、安全设计

风险 防线
未登录白嫖 token /user/chat 不在网关白名单,JWT 鉴权
AI 查别人的订单 order-service 端校验 order.userId == 登录人,与提示注入无关的硬校验
前端伪造 userId 工具链路不接收前端传的 userId,一律从网关透传头解析
会话串号 Redis key 按 userId:sessionId 双重隔离
token 消耗膨胀 历史截断最近 20 条;工具结果字段裁剪

九、踩坑记录

  1. HttpStatus::isError 方法引用编译不过:Spring 6.1 的 onStatus 接收 Predicate<HttpStatusCode>,HttpStatus::isError 推导出的是 Predicate<HttpStatus>,类型不兼容 → 改 lambda status -> status.isError()。
  2. Mono.flatMap 不能返回 Flux:completion 是 Mono<Flux<...>>,拼接要用 flatMapMany(f -> f),flatMap 会报”无法推断 R”。
  3. lambda 引用的局部变量必须 effectively final:sessionId 在 defer 里被重新赋值导致编译错误,改赋给 final String sid。
  4. 微服务反向依赖:OrderStatus 枚举在 mall-order,mall-user 不能引用——工具服务本地维护一份状态码→中文的映射,注释说明两处需同步。
  5. api-key 默认值为空:不设环境变量时每次调用 401;开发期 yml 里给了默认密钥,生产必须走环境变量。
  6. Map.of 不允许 null 值:组装商品 JSON 时 subtitle/detail 等可空字段先做 == null ? "" : 处理,否则 NPE。

十、可改进的点

  1. WebClient 未设超时:DeepSeek 挂起会占住线程,应配 connect/read timeout。
  2. 无限流:/user/chat/stream 每次调用消耗真实 token,应按用户限流(如 Redis 计数每分钟 N 次)。
  3. 工具轮用户无感知:模型调工具的 1~3 秒里前端只有”正在思考…”,可下发 event: tool 告知”正在查询商品…”提升体验。
  4. 历史截断可能切断语义:按条数截断20条,更优做法按 token 预算截断或做滚动摘要。
  5. mall-common 的 AgentConfig(”康养小智”)是遗留代码,与本功能无关却随公共模块加载,应清理。
上一篇
下一篇