一开始我以为，把 AI 接进项目里就是：

前端输入问题
-> 调用大模型接口
-> 返回答案

真正做完 X-Chat AI 后，我才发现，AI 应用开发最难的地方往往不是“怎么调模型”，而是：

已有业务系统怎么安全地接入 AI？
AI 服务出错时，普通业务会不会受影响？
用户权限怎么控制？
流式回复怎么推给前端？
AI 查知识库时，怎么避免跨群泄露资料？
用户量变大后，AI 服务能不能承受？

这篇文章记录我在 X-Chat 项目里，把一个普通即时通讯系统接入 AI 服务时遇到的问题、解决方案和后续优化方向。

---

一、项目背景：不是单独做一个 AI Demo

X-Chat 原本是一个即时通讯系统，核心能力包括：

用户登录
好友聊天
群聊
消息落库
WebSocket 实时推送

后来我希望给它加上 AI 能力，比如：

1. 在群聊中问 AI
2. 根据群知识库回答问题
3. 总结聊天上下文
4. 展示 AI 工具调用过程
5. 流式返回 AI 回复

如果只是为了演示，最简单的做法是前端直接调用一个 Python AI 接口。

但这样很快会遇到问题：

用户有没有登录，谁来判断？
用户是不是群成员，谁来判断？
A 群的知识库能不能被 B 群查到？
AI 服务挂了，会不会影响普通聊天？
AI 回复是一段段生成的，前端怎么合并？

所以这个项目最后不是“前端调一下 AI 接口”，而是做成了下面这种架构：

Vue3 前端
-> Spring Boot Java 后端
   -> 鉴权
   -> 群权限校验
   -> 消息落库
   -> WebSocket 推送
-> FastAPI Python AI 服务
   -> 意图路由
   -> RAG 检索
   -> Agent 工具调用
   -> NDJSON 流式输出
-> Java WebSocket 转发
-> 前端按 messageId 追加 AI 消息

核心思想是：

Java 负责业务安全和消息链路，Python 负责 AI 能力。前端不能直接绕过 Java 访问 AI 服务。

---

二、前端直接调 Python？一开始看起来最简单

一开始我想过，既然 Python 服务负责 AI，那前端是不是可以直接请求 Python？

比如：

前端 -> Python /chat/stream -> 返回 AI 答案

这样看起来最简单，但它有一个很严重的问题：会绕过业务权限。

X-Chat 是聊天系统，用户能看到什么，不只是看接口能不能调用，还要看业务关系。

比如：

用户是不是已经登录？
用户是不是这个群的成员？
用户有没有权限访问这个群的知识库？
这个 group_id 是否属于当前用户？

如果前端直接调用 Python，并且可以自己传 group_id，那理论上就可能出现：

A 用户不是某个群成员
但手动构造请求
让 Python 去查这个群的知识库

这显然不安全。

我的处理方式：前端只调用 Java，Java 校验后再代理 Python

所以我把 AI 服务接入设计成：

前端
-> Java Controller
-> Java Service 校验用户和会话权限
-> AiFastApiClient 调 Python /chat/stream
-> Python 执行 RAG / Agent
-> Java 通过 WebSocket 推给前端

在这个设计里，Python 不直接处理 X-Chat 的登录态和群成员关系。

Python 只负责：

1. 接收 Java 传来的问题
2. 根据 Java 传来的 scope_type / scope_id 检索知识库
3. 执行 RAG 或 Agent
4. 返回流式事件

Java 负责：

1. 判断用户是谁
2. 判断用户是否有权限访问当前会话
3. 判断用户是否有权限访问当前群知识库
4. 再把合法请求转给 Python

这样就把边界分清楚了：

Java：业务权限边界
Python：AI 能力边界

这也是我后来对“业务系统接入 AI”的一个重要理解：

AI 服务不是越独立越好。它必须被业务系统安全地包起来，否则很容易绕过原有权限体系。

---

三、为什么我没有把 AI 逻辑直接写在 Java 里？

另一个问题是：既然 X-Chat 是 Java 项目，为什么不直接在 Java 里做 RAG？

后来我发现，拆成两个服务更合理。

Java 后端更适合处理：

用户鉴权
好友关系
群成员权限
消息落库
会话更新
WebSocket 推送

Python 更适合处理：

LangChain
Embedding
Chroma 向量库
RAG
Agent
大模型流式调用
文档解析

所以最终服务边界是：

X-Chat
-> Java / Spring Boot
-> 负责用户、群聊、消息、权限、WebSocket

X-RAG Agent
-> Python / FastAPI
-> 负责 RAG、Agent、LLM 调用、流式输出

这样做有几个好处。

第一，职责清楚。Java 不需要硬接 Python AI 生态，Python 也不需要直接操作 X-Chat 的业务库。

第二，稳定性更好。AI 服务挂了，普通聊天仍然可以使用。

第三，AI 能力可以独立迭代。后面要换模型、加 rerank、改 RAG 检索逻辑，都可以主要在 Python 服务里完成。

面试或者项目介绍时，我会这样总结：

X-Chat 的 Java 后端负责业务和权限，Python FastAPI 负责 AI 能力。两者通过内部接口通信。这样既能利用 Python 的 AI 生态，又不会破坏原有 Java 业务系统的稳定性。

---

四、用户消息还没保存成功，AI 就开始回复会发生什么？

接入 AI 后，我遇到一个很容易忽略的问题：

用户发消息后，什么时候触发 AI 回复？

最简单的做法是：

用户发送消息
-> Java 保存消息
-> 立刻调用 Python
-> 返回 AI 回复

但这里有一个事务问题。

用户消息发送并不是只插入一条消息，通常还会更新会话、最后一条消息等数据。

如果这些数据库操作还没提交，就提前触发 AI，可能出现：

AI 已经根据这条消息开始回复
但 Java 消息事务后来失败回滚

这样前端可能看到 AI 回复了一条实际上不存在的用户消息。

我的处理方式：事务提交后异步触发 AI

所以我的处理方式是：

用户消息先落库
-> 会话数据更新
-> WebSocket 推送普通消息
-> 事务提交成功后
-> 异步触发 AI 回复

核心思路是：只有用户消息真正保存成功后，AI 才开始回复。

同时，AI 调用是异步的，不会阻塞普通发消息接口。

这个点让我理解到：

AI 是增强能力，不应该破坏原有业务链路。普通消息发送必须稳定，AI 回复可以稍后生成。

---

五、AI 服务失败了，普通聊天还能不能用？

业务系统接入 AI 时，不能只考虑成功情况，还要考虑失败时怎么收场。

AI 调用链路比普通接口长得多：

Java 后端
-> 调用 Python AI 服务
-> Python 做意图判断
-> 检索知识库
-> 调用大模型
-> 流式返回结果
-> Java WebSocket 推给前端

这条链路里任何一步都可能出问题：

1. Python AI 服务没有启动
2. Java 调 Python 超时
3. 大模型接口调用失败
4. RAG 检索异常
5. 流式返回到一半中断
6. 前端 WebSocket 断开
7. 工具调用失败

如果没有兜底，就可能出现很差的体验：

用户消息已经发出去了
但 AI 回复一直卡着
前端不知道是生成中还是失败了
数据库里可能留下半条 AI 消息
用户也不知道该不该重新提问

我的处理方式：AI 失败只影响 AI 回复，不影响普通消息

我在项目里的思路是：

普通聊天主链路优先保证成功
AI 回复失败只影响这一次 AI 回复
不能影响用户自己的消息发送

也就是说，用户发消息时：

用户消息先落库
-> 普通 WebSocket 消息正常推送
-> 事务提交后再异步触发 AI

如果 AI 服务失败，最多是：

AI 回复失败

而不是：

用户消息发送失败
聊天页面崩溃
整个会话不可用

如果 Java 调用 Python /chat/stream 失败，比较合理的处理是：

1. 捕获异常
2. 更新 AI 占位消息为失败状态
3. 通过 WebSocket 推送失败事件
4. 前端展示“AI 回复失败，请稍后重试”

用户看到的应该是一个明确提示，而不是一直转圈。

正常情况：

AI 正在输入...
-> AI 返回 delta
-> AI 回复完成

异常情况：

AI 正在输入...
-> Python 调用失败
-> 显示：AI 回复失败，请稍后重试

这样用户至少知道这次 AI 回复失败了，不会误以为系统还在生成。

流式回复中途失败怎么办？

流式回复还有一个特殊问题：它可能已经返回了一半。

比如前端已经看到：

根据知识库内容，Netty 是一个异步事件驱动的...

这时候 Python 或大模型接口突然断开。

如果不处理，前端会一直以为 AI 还在输入。

所以流式异常也要兜底：

1. 已经收到的 delta 可以保留
2. 但要给这条消息标记生成失败
3. 前端停止 loading
4. 可以追加一段提示：本次回复未完整生成

比如展示成：

根据知识库内容，Netty 是一个异步事件驱动的...

[本次 AI 回复中途失败，内容可能不完整，请稍后重试]

这比直接丢掉半条回复更友好，也比一直卡住更清楚。

这一部分可以总结成：

异常兜底的本质，是把 AI 服务从“强依赖”变成“可降级能力”。普通聊天必须稳定，AI 回复可以失败；但失败时要有明确状态、明确提示和可恢复方式。

---

六、AI 一边生成一边返回，前端怎么把它显示成一条消息？

普通接口一般是：

请求
-> 等待
-> 一次性返回完整结果

但 AI 回复通常比较长，如果等全部生成完再返回，用户体验会比较差。

所以我在 Python 服务里使用流式输出。

Python 返回的不是一个完整 JSON，而是一行一行的 NDJSON 事件：

{"type":"delta","content":"你好"}
{"type":"delta","content":"，这个问题可以这样理解"}
{"type":"tool_call","tool_name":"knowledge_search"}
{"type":"tool_result","content":"命中 3 个知识库片段"}
{"type":"done"}

Java 后端按行读取这些事件，然后通过 WebSocket 推给前端。

整体链路是：

Python /chat/stream
-> 返回 NDJSON
-> Java AiFastApiClient 按行解析
-> Java WebSocket 推送
-> 前端 chatStore 按 messageId 追加内容

这里还有一个关键点：messageId。

AI 回复不是一次性生成的，而是一段一段推送的。如果前端每收到一个 delta 都新增一条消息，聊天窗口就会出现很多碎片气泡。

所以我的做法是：

1. Java 先创建一条 AI 占位消息，拿到 messageId
2. 前端先显示一条空的 AI 消息
3. 每次收到 delta，都根据 messageId 找到这条消息
4. 把 delta 追加到 messageContent 后面

这样用户看到的是一条 AI 消息逐渐变长，而不是一堆零散消息。

---

七、工具调用信息应该放进回答正文里吗？

在 RAG / Agent 场景中，AI 不只是生成文本，还可能调用工具。

比如：

AI 开始查知识库
AI 命中了哪些文档
AI 是否进入回答阶段

一开始我也纠结：这些工具调用信息要不要混进最终回答？

后来我发现，最好分开。

最终回答是给用户看的自然语言内容。

工具调用过程是调试和可解释性信息。

所以 Python 返回两类事件：

delta
-> AI 正文片段

tool_call / tool_result
-> 工具调用过程

Java 再转成 WebSocket 事件：

ai_stream_delta
ai_tool_call
ai_tool_result

前端把工具事件挂在同一条 AI 消息下面，而不是混进正文。

这样做的好处是：

用户能看到 AI 正在查知识库
开发者能看到工具是否命中
最终回答仍然保持干净

这让我意识到：

AI 系统不能只是“给答案”，还要能解释答案是怎么来的。

---

八、群知识库为什么不能所有群共用一份？

X-Chat 是群聊系统，不同群可能上传不同项目资料。

比如：

A 群上传需求文档
B 群上传接口文档
C 群上传会议纪要

如果 RAG 检索时不做隔离，就可能出现：

A 群用户提问
-> AI 命中了 B 群知识库
-> 私有资料泄露

所以知识库必须带作用域。

在项目里，我用类似下面的字段表示知识库范围：

scope_type = global
scope_id = ""
表示全局知识库

scope_type = group
scope_id = 当前群 ID
表示某个群的知识库

群聊问答时，只能检索：

全局知识库 + 当前群知识库

不能检索：

其它群知识库

这里 Java 和 Python 又有一次职责分工：

Java：判断当前用户是不是群成员
Python：根据 scope_type / scope_id 做知识库过滤

这个设计让我理解到，RAG 的权限问题不是“查数据库时多加一个条件”这么简单，而是 AI 应用里的安全边界。

---

九、用户量变大后，Python AI 服务能接住这么多请求吗？

当前项目里，用户消息保存成功后，Java 后端会异步触发 AI 回复。

这样已经避免了 AI 调用阻塞普通消息发送。

但如果用户量变大，会出现新的问题。

比如很多用户同时在群里问 AI：

用户 A 问 AI
用户 B 问 AI
用户 C 问 AI
用户 D 问 AI
...

如果 Java 后端每次都立刻创建一个异步任务去调用 Python，就可能出现几个风险：

1. 同一时间 AI 请求太多，Java 线程池被打满
2. Python AI 服务压力过大，响应变慢甚至超时
3. 大模型接口本身有并发限制或限流
4. 某些 AI 请求失败后，不方便统一重试
5. 系统很难知道还有多少 AI 任务正在排队

Python 的 FastAPI 服务理论上可以并发处理多个请求，特别是调用大模型这种 I/O 型任务。

但 AI 请求和普通接口不一样。

普通接口可能几十毫秒就结束，而 AI 请求可能包括：

RAG 检索
-> 构造 prompt
-> 调用大模型
-> 流式生成回答
-> Java 转发 WebSocket

一次请求可能持续几秒甚至更久。

如果大量请求同时打到 Python 服务，会带来明显压力：

1. 同时连接数变多
2. 每个流式请求占用时间较长
3. Chroma / SQLite / Embedding 查询压力增加
4. 大模型接口可能有 QPS 或并发限制
5. Python 服务内存和 CPU 消耗上升

所以不能简单认为：

用户越多 -> Python 直接全部并发处理

更稳的做法是控制并发。

优化方向：把 AI 调用任务队列化

可以把 AI 调用从“立即执行”改成“先放进任务队列，再由后台 worker 慢慢消费”。

当前做法：

用户消息保存成功
-> Java 立即异步调用 Python
-> Python 生成 AI 回复
-> Java 推送给前端

优化后的做法：

用户消息保存成功
-> Java 创建一条 AI 任务
-> 把任务放进队列
-> 后台 worker 从队列里取任务
-> 调用 Python AI 服务
-> 得到结果后更新 AI 消息
-> 通过 WebSocket 推送给前端

这里的任务可以包含这些信息：

task_id：任务 ID
user_id：是谁发起的
session_id：属于哪个会话
message_id：用户原始消息 ID
ai_message_id：AI 占位消息 ID
question：用户问题
scope_type / scope_id：知识库作用域
status：pending / running / success / failed
retry_count：重试次数
created_at：创建时间

这样设计以后，AI 调用就不再是一个不可控的异步请求，而是变成了一个可以管理的后台任务。

队列的作用不是让 Python 无限并发，而是保护 Python 服务

比如 Java 侧收到 100 个 AI 请求，不需要让它们同时打到 Python。

可以这样处理：

100 个 AI 请求
-> 全部进入队列
-> 5 个 worker 并发处理
-> 剩余任务等待

也就是说：

Python AI 服务同时最多承受 5 个左右的 AI 调用
其余任务在队列里排队

没有队列时：

100 个请求同时打到 Python
-> Python 服务变慢
-> 部分请求超时
-> 大模型接口被限流
-> 用户体验整体下降

有队列后：

100 个请求先进入队列
-> 只放行固定数量请求给 Python
-> Python 按稳定并发处理
-> 失败任务可以重试
-> 系统整体更可控

所以队列不是为了让 AI 服务无限并发，而是为了把突发流量变成平稳流量。

这一节可以总结成：

当前版本是“事务提交后异步触发 AI”，适合项目初期和小规模使用。后续如果用户量变大，可以把 AI 调用改成任务队列模式：Java 只负责创建 AI 任务和占位消息，后台 worker 按并发限制消费任务、调用 Python、更新结果并推送前端。这样可以支持削峰填谷、失败重试、任务状态追踪和服务保护。

---

十、这个项目和普通 AI Demo 的区别

如果只是普通 AI Demo，可能只有：

输入问题
-> 调用模型
-> 返回答案

而 X-Chat AI 需要考虑更多工程问题：

1. 前端不能直接绕过业务权限访问 AI 服务
2. Java 和 Python 要分清职责边界
3. 用户消息要先落库，再触发 AI
4. AI 调用要异步，不能阻塞普通聊天
5. Python 要用 NDJSON 流式返回
6. Java 要把流式事件转成 WebSocket
7. 前端要按 messageId 追加消息
8. 群知识库要按作用域隔离
9. 工具调用过程要和最终回答分开展示
10. AI 服务异常时，普通聊天不能受影响
11. 用户量变大后，要通过队列保护 AI 服务

做完这个项目后，我最大的感受是：AI 应用开发不是把模型接口接上去就结束了，真正难的是让它在已有业务系统里安全地运行、稳定地失败、可控地扩展。

---

十一、后续优化方向

目前这个版本已经跑通了完整链路，但如果继续优化，我会从下面几个方向做。

1. 服务间鉴权

现在前端不能直连 Python，但 Java 调 Python 的内部接口也应该继续增强安全性。

后续可以增加：

内部服务 token
请求签名
时间戳防重放
IP 白名单

这样可以避免 Python AI 服务被非 Java 后端直接调用。

2. AI 调用任务队列化

当前 AI 调用是异步触发，适合项目初期和小规模并发。

后续如果用户量变大，可以引入任务队列：

用户消息提交
-> 写入 AI 任务表 / 消息队列
-> 后台 worker 调 Python
-> 生成结果后回写消息

这样可以支持：

失败重试
削峰填谷
任务状态查询
超时控制
并发保护

3. 流式异常处理更细

流式回复相比普通接口更复杂。

后续可以继续优化：

前端断开连接怎么处理？
Python 已生成一半内容是否要保留？
重复推送 delta 如何幂等？
done 事件丢失怎么办？
用户刷新页面后如何恢复 AI 消息状态？

流式链路越长，异常情况越多，这部分需要更细的状态设计。

4. RAG 检索继续增强

当前已经有关键词 + 向量混合检索、证据门控和群知识库隔离。

后续可以继续做：

rerank 重排序
更大的评测集
检索耗时打点
按问题类型调整 top_k
更完整的 RAG 调试日志

这样可以让 AI 回答不只是“能答”，而是更稳定、更可解释。

5. 可观察性和审计日志

业务系统接入 AI 后，排查问题很重要。

后续可以记录：

本次 AI 请求是谁发起的
属于哪个会话或群
调用了哪个模型
检索了哪些知识库
命中了哪些 chunk
是否触发拒答
总耗时是多少
是否发生异常

这样当用户反馈“AI 答错了”时，后端可以知道问题出在权限、检索、证据判断还是模型生成。

---

十二、小结

做完 X-Chat AI 后，我对 AI 应用开发有了一个更具体的理解。

AI 应用不是简单地把模型接口接到前端，而是要把 AI 能力放进已有业务系统里，并处理好这些问题：

谁可以问？
可以查哪些资料？
什么时候触发 AI？
AI 出错怎么办？
流式结果怎么展示？
工具调用过程怎么解释？
AI 服务和业务服务怎么解耦？
高并发时如何保护 AI 服务？

在这个项目里，Java 后端负责业务安全和实时消息链路，Python FastAPI 负责 RAG、Agent 和大模型调用。前端只和 Java 交互，Java 校验权限后再代理 Python。Python 返回 NDJSON 流式事件，Java 再通过 WebSocket 推给前端。

这个过程让我明白：

真正的 AI 应用开发，不只是让模型回答问题，而是设计一条安全、稳定、可解释、可维护的业务链路。