企业里知识散在文档里、工单全靠人工审、通用 AI 又不懂企业业务。这个项目用「多智能体编排 + RAG 检索 + 工具调用」把这三件事拧成一个对话入口。这篇文章讲清项目做什么、业务怎么跑、技术怎么落地。

一、项目介绍

        知库工单智联系统是面向中小企业的智能办公平台，做三件事：私有知识库管理、AI 智能问答、企业化工单协同。解决三个痛点：

1. 企业知识分散：制度、流程、手册散在各种文档里，新人想查找不到
2. 工单全靠人工审：简单咨询类工单也要人盯着处理，浪费人力
3. 通用 AI 不懂企业业务：ChatGPT 不会知道你公司的 VPN 怎么配、报销走什么流程

        核心思路是一个对话入口统一处理：用户用自然语言提问，系统自动判断意图路由到对应能力；工单提交后 AI 自动审核办结，复杂的转人工。

        技术栈用 Vue3 + FastAPI + LangGraph + DeepSeek + Ollama + Chroma + MySQL + MinIO + Redis。

二、业务流程

系统有四块核心业务，围绕一条典型用户旅程串起来：

员工上传《VPN 手册》

   → AI 审核文档，通过才发布

   → 后台异步向量化（切片 + embedding 写入 Chroma）

   → 员工在对话框问「VPN 怎么连」

   → 多智能体 supervisor 判断意图，路由到知识问答智能体

   → RAG 检索该员工可见的文档片段

   → 流式生成带引用来源的回答

2.1 私有知识库

三层隔离的文档管理：

公共库：全员可读，管理员管理（公司制度、规范）

个人库：自己上传自己管（私人笔记）

群组库：群主管理，群成员可读（部门资料）

        文档上传走 AI 审核，通过才发布，后台异步向量化。这套隔离贯穿到 RAG 检索——每个用户只能检索到自己可见范围内的知识。

2.2 AI 智能问答

        一个多智能体助手做统一入口。用户问什么，supervisor 路由给对应智能体处理：问知识走 RAG、查工单走工具、看数据走统计、闲聊直接回。回答流式逐字返回，带文档引用。

2.3 企业化工单

三态路由：

员工上行：员工提交工单给所属企业

企业下行：企业账号给员工派任务

无企业员工：工单仅个人可见，记个人日志

工单进来后跑 AI 审核工作流：简单咨询类自动办结，复杂的转人工。转人工是 HITL（Human-In-The-Loop，人在回路）的落地点——AI 跑到关键节点把决策权交回给人。员工还能对自己的工单自审/自关闭。

2.4 用户与安全

        手机号验证码注册，选员工或企业身份（企业账号填企业名生成企业码，员工选填企业码加入企业）。改密后旧 token 立即失效。

三、技术工作流

三套工作流都用 LangGraph 编排：多智能体调度、工单审核、文档向量化。挑两套重点讲。

3.1 多智能体调度（supervisor 主管模式）

        什么是主管模式：不直接让多个智能体平级对话，而是先设一个「主管」节点用 LLM 判断用户问题该交给谁，再路由过去。所有智能体听主管调度，彼此不直接通信，路由集中可控。

用户问题 → supervisor(LLM 判断意图)

            ├─ qa      → RAG 检索知识库 → 流式生成回答

            ├─ ticket  → 调工具查工单 → 生成工单状态播报

            ├─ doc     → 调工具取文档摘要 → 总结

            ├─ stats   → 调工具取统计数据 → 自然语言汇总

            └─ chat    → 无需工具，直接闲聊

supervisor 的判断逻辑是关键词优先 + LLM 兜底：

判断规则(按优先级):

涉及「查询工单」「我的工单」「工单状态」→ ticket

涉及「摘要」「总结文档」→ doc

涉及「统计」「数量」「多少」→ stats

涉及企业知识/流程/规范咨询 → qa

问候/闲聊/不确定 → chat

        为什么不用纯关键词、也不用纯 LLM？纯关键词遇到「VPN 怎么申请」（该走 qa）和「我的 VPN 工单到哪了」（该走 ticket）会判错；纯 LLM 又慢又费 token。关键词过滤掉明确意图，剩下的模糊问题交给 LLM，兼顾准确和成本。

        这里有个取舍：supervisor 输出 JSON 用了「代码块 + json.loads + 失败降级」的土办法，而不是 `with_structured_output`。原因是 DeepSeek 的 function calling 偶尔抽风，JSON 模式反而更稳。失败时降级到 `chat`，保证用户至少能收到一句问候，不卡死。

3.2 工单 AI 审核（LangGraph 状态机）

工单提交后自动跑审核工作流，决定「自动办结」还是「转人工」：

parse_ticket → classify_intent → retrieve_knowledge → decide

                                                       ├─(auto_close)─→ auto_resolve → END

                                                       └─(escalate)──→ escalate ────→ END

• parse_ticket：读工单信息，把产出字段初始化成默认值。不调 LLM、不查 RAG
• classify_intent：LLM 判断意图（故障报修/VPN申请/账号申请等）+ 复杂度（simple/complex）。容错有个坑：LLM 可能裹一层 ```json 标记，要去掉开头标记（不能逐字符删，会把正文开头的 j、s、o、n 误删）；解析失败降级成关键词规则
• retrieve_knowledge：复用 qa_agent.retrieve 检索知识库，带上提交者可见文档集合做用户隔离，只检索有权限的文档。报错降级成空，不卡流程
• decide：纯规则，不调 LLM。三条转人工规则——优先级紧急或高、复杂度复杂、知识库没命中（没依据就不自动办结）；否则自动办结
• auto_resolve：LLM 基于 RAG 内容生成回复，工单状态置为「已完结」。失败降级转人工
• escalate：套模板生成提示，工单状态置为「处理中」等管理员复核

整个链路有 WorkflowLog + WorkflowStep 记录每一步耗时和结果，方便排查「为什么这个工单被转人工了」。

四、RAG 检索增强：从单跳到全链路

这是整个项目技术含量最高的部分。最初的版本只有「单次 top-k 向量检索 + L2 阈值过滤」，问题很明显：口语化问题召回差（「门禁卡咋办」匹配不到「门禁权限申请流程」）、单次召回容易漏、检索为空直接放弃。

后来重构成一套「细检索 + 深检索」的完整链路：

Step1 查询改写    DeepSeek 把口语问题改写成检索友好的关键词 + 1-2 个子问题

Step2 多路召回    main_query + 每个子问题各检索 top-5，并发 + 按 content 去重

Step3 Rerank      Cross-Encoder（bge-reranker-base）对候选切片打分重排，取 top-4

Step4 阈值过滤    L2 距离 ≤ 0.8 且 rerank 分数 ≥ 0.0 双阈值

Step5 失败重试    过滤后为空 → 用改写后的关键词重召回，阈值放宽到 0.85

Step6 邻居扩展    命中切片的前后两片拼进来，给 LLM 更完整的上下文

4.1 查询改写：解决口语化问题

用户不会按关键词提问。一句「门禁卡咋办」，直接 embedding 出来跟「门禁权限申请流程」相似度很低。我用 DeepSeek 先把问题改写：

输入: 门禁卡咋办

输出: {"main_query": "门禁卡 申请流程", "sub_queries": ["门禁权限申请审批", "门禁卡领取签字"]}

        `main_query` 去掉口语词、保留核心实体，`sub_queries` 拆出不同角度。子查询最多 2 条，避免召回爆炸。改写失败降级为原问题（退化为单路召回）。

4.2 多路召回 + 去重

改写后得到 1 个 main_query + N 个 sub_queries，每路各检索 top-5，用 `asyncio.gather` 并发跑，按 content 去重时保留 L2 距离更小的那个（更相似的优先）。原本单次检索漏掉的相关切片，子查询能补回来。

4.3 Cross-Encoder Rerank：真正提升精度的关键

向量检索用的是双塔模型（query 和 doc 各自 embedding 再算距离），快但粗。Rerank 用的是 Cross-Encoder（query 和 doc 拼一起过一遍 transformer），慢但准。我用 `BAAI/bge-reranker-base`：

节选自 app/ai/tools/reranker.py

model = CrossEncoder(model_name_or_path="BAAI/bge-reranker-base",

                     max_length=512, cache_folder=settings.RERANK_CACHE_DIR,

                     device="cpu")

pairs = [(query, c) for c, _, _ in candidates]  # (query, doc) 对

scores = await asyncio.to_thread(model.predict, pairs)  # 同步 CPU 密集，用 to_thread 包

        Rerank 模型加载是同步阻塞的，用 `asyncio.to_thread` 包成异步；加载失败时降级为「按 L2 距离排序取 top-k」，不阻塞问答。

4.4 邻居切片扩展：零数据迁移的上下文增强

切片是 500 字一段，一个完整流程可能跨 3-4 个切片。如果只把命中的那一片喂给 LLM，上下文是断的。解决办法是取命中切片的前后邻居拼起来。

怎么取邻居？最直觉的想法是按 vector_id 命名约定（`f"doc{doc_id}_chunk_{i}"`）构造 id 去查，但这强依赖命名约定，改一处全炸。我用的是 **Chroma 的 metadata where 过滤：

节选自 app/ai/tools/vectorstore.py

where = {

    "$and": [

        {"doc_id": doc_id},

        {"chunk_index": {"$in": [chunk_index - 1, chunk_index + 1]}},

    ]

}

result = get_vectorstore().get(where=where, include=["metadatas", "documents"])

        这里踩了个坑：Chroma 的 where 不支持顶层多 key 隐式 AND，必须用 `$and` 显式包裹，否则报 `Expected where to have exactly one operator`。边界情况（首尾切片）天然降级返回空，不影响主流程。

        邻居拼好后写到 metadata 的 `extended_content` 字段，`content` 原值不动——保证 `extract_sources` 按 doc_title 去重的逻辑不变，SSE 契约不破坏。

4.5 失败重试：深检索层

        阈值过滤后 hits 为空时，不是直接返回「没找到」，而是用改写后的 main_query 重召回一次（fetch_k 扩大到 k×4），阈值放宽（L2 到 0.85，rerank 到 -2.0）。仍空才降级。这样对「相关但相似度边缘」的查询能多捞一次。

        整个 `retrieve` 最外层包了 try/except，任何环节异常都回退到 `_legacy_retrieve`（原单跳逻辑完整保留）。问答永远不会因为 RAG 增强失败而阻塞，这是最重要的工程保证。

五、MCP 工具层：标准化工具协议

        智能体要调工具（查工单、查文档、查统计），我没有让 LLM 直接写 SQL，而是封装了一套**标准化工具协议**。每个工具有元数据描述（name/description/parameters schema），LLM 只需决定「调哪个工具、传什么参数」，具体执行由后端代码完成。

节选自 app/ai/tools/mcp_tools.py

TOOL_DEFINITIONS = [

    {"name": "query_tickets",

     "description": "查询企业工单列表,可按状态过滤、关键词搜索",

     "parameters": {"status": "string(可选)", "keyword": "string(可选)", "limit": "int(默认5)"}},

    # ... list_docs / get_doc_summary / get_dashboard_stats / generate_report

]

为什么要封装这层？三个好处：

1. 一是用户隔离：工具内部按 user_id 过滤，员工只能查自己的工单、看自己有权限的文档
2. 二是审计溯源：每次工具调用写 ToolLog（谁、什么时候、调了什么、结果如何），供管理员追溯
3. 三是可扩展：新加工具只要写实现函数 + 注册到 TOOL_REGISTRY，智能体就能用

六、数据存储：按特性分库

没有用「一个数据库打天下」，而是按数据特性分到四个存储：

存储	存什么	为什么
MySQL	用户/工单/对话/文档元数据/日志	结构化、要事务、要复杂查询
MinIO	工单附件（PDF/DOC/ZIP）	非结构化大文件，对象存储合适
Redis	验证码/频率限制	短时高频，内存低延迟
Chroma	文档向量	相似性检索

Redis 还做了降级机制：连不上时降级为进程内 dict，验证码功能不挂掉。

七、几个值得讲的工程取舍

1. SSE 事件契约不能破坏：前端对接了 `tool_start → tool_end(sources) → token* → done` 的事件流。RAG 增强时 hits 结构加了 `rerank_score`/`extended_content` 字段，但 `extract_sources` 按 doc_title 去重的逻辑完全不动，sources 还是文档标题列表，前端无感知。

2. 后台任务的竞态：文档向量化是后台异步任务，如果在向量化过程中文档被删除，会导致外键约束失败和 Chroma 孤儿向量。修复办法是向量化前校验文档存在性，不存在则清理已写入向量。

3. 全链路日志埋点：每个环节打 `[RAG]` 前缀的结构化日志（改写结果、多路召回数量、rerank 分数、邻居扩展字数）。排查问题能 grep 出完整链路，面试讲解时也能直接展示「这个查询走了哪几步」。

八、总结

这个项目最难的不是某个单点技术，而是**把多个技术拼成一个能跑的链路，还要保证每一层失败都不阻塞主流程。RAG 增强那套（查询改写→多路召回→rerank→邻居扩展→失败重试）每一层都有降级路径，最外层回退原逻辑，这是工程上最值得讲的部分。