想学AI应用开发?先看看我这个文旅探店助手踩过的坑(附完整代码)

我最近我动手做了一个AI项目：用自然语言问"钟楼附近有什么人均80以下的面馆"，AI自动帮你查店铺、筛优惠券、按距离排序，最后流式输出推荐结果。这篇文章不打算罗列技术术语，而是想和你聊聊——做完这个项目，我真正学到了什么，以及如果你也想入门AI应用开发，可以从中学到什么。

一、先说说这个项目是干什么的

想象这样一个场景：你到西安旅游，想吃泡馍，但又不想被坑。你打开某个APP，搜"泡馍"，刷评分，翻评论，找优惠券……至少折腾5分钟。

如果有一个助手，你直接问它：

"钟楼附近，人均80以内，有优惠券的泡馍店，推荐一下。"

它自动帮你完成：找店铺 → 按距离排序 → 筛人均 → 查优惠券 → 输出结果。整个过程就像有个西安本地朋友在帮你打听。

这就是我做的"长安文旅探店助手"，听起来很酷对吧？下面我带你看看，为了实现这个效果，我学到了哪些东西。

二、第一课：统一入口——用户一句话，该走RAG还是Agent？

这是我在这个项目中学到的最重要的一课。

问题：用户问一句话，该走哪条路？

我的系统有两种工作模式：

• RAG 模式：从知识库里检索答案（适合"兵马俑门票多少钱"这种知识类问题）
• Agent 模式：自动调用工具去查实时数据（适合"附近有什么好吃的"这种行动类问题）

所以用户发来一句话，系统要先判断：走RAG还是走Agent？

路由系统的迭代是一个典型的"试错→优化→再试错→再优化"的过程。我前后经历了四版：

V1（初版）：一层关键词规则路由。效果很差，因为用户问题往往包含多个关键词，简单规则匹配只能命中一个。

V2（试错版）：三层路由——规则→小模型→大模型。延迟太高（秒级），C端用户体验完全不可接受。

V3（实用版）：双层关键词匹配规则路由。延迟 <1ms，在361条测试集上准确率99.7%。但存在三个明显缺陷：

• 关键词覆盖不足（5-10% 盲区），口语化表达和方言无法识别
• 没有上下文感知能力（无法处理多轮追问）
• 无闲聊/拒绝模式（"你好""谢谢""哈哈哈"也会走完整管线）

V4（当前版）：四层架构 + 置信度门控 + LLM 路由缓存。这是我最终定型的设计：

各层详解

Layer 1 — 预过滤层（0ms，拦截约 8% 请求）

这是最轻量的一层。像"你好""谢谢""哈哈哈""？？？""   "这类输入，一毫秒都不需要走后面的管线。直接识别为闲聊（Chat）或拒绝（Reject），省掉下游所有开销。

CHAT_KEYWORDS = frozenset({"你好", "谢谢", "再见", "哈哈", "嗯", "哦", ...})REJECT_PATTERN = re.compile(r'^[\s\W]+$')  # 纯空白/符号/表情

Layer 2 — 关键词匹配层（<2ms，命中约 90% 请求）

这是核心执行层。相比V3的双层关键词，V4做了三个关键升级：

冲突解决的权重计算逻辑：

# Agent 强关键词（2.0 分）：帮我找、找一家、推荐、优惠券、团购、人均...# RAG 实体关键词（2.0 分）：大雁塔、兵马俑、华清池、钟楼、城墙...# 普通关键词（1.0 分）# 规则：RAG 权重大于 Agent 权重 1.5 倍时走 RAG，否则走 Agentif rag_weight > agent_weight * 1.5:return RouteType.RAG   # 例："推荐一个好玩的景点" → RAGelse:return RouteType.AGENT  # 例："推荐一家好吃的泡馍店" → Agent

Layer 2.5—置信度门控（P0/P1 优化）

这是 V4 最核心的设计模式。关键词匹配的结果不是无条件信任的——每条路由决策都附带一个置信度分数，低置信度（<0.6）时穿透到Layer 4让LLM做二次判断：

什么场景会触发低置信度？比如用户只说"多少钱？"——没有上下文，不知道问的是门票还是菜品价格，置信度只有0.45。

Layer 3—上下文感知层

多轮对话中，用户经常追问："那附近呢？""这家贵吗？""怎么样？"。如果不看上下文，这些短句无法路由。Layer 3 检测追问模式（代词指代、省略主语），从上一轮继承路由决策。

Layer 4—LLM 智能判断层（仅 <2% 请求触发）

这是兜底层。当Layer 2完全没命中任何关键词，或置信度太低时，调一次DeepSeek做轻量分类（temperature=0.1, max_tokens=150）。但这还不是全部——P2 优化给Layer 4加了缓存：

classRoutingLLMCache:"""LLM 路由结果缓存。    - TTL 1 小时，最大 1000 条    - 相同 query 命中缓存时 ~0ms（跳过 LLM 调用）    - 预计缓存命中率 60-80%    - 将 LLM 调用率从 3-5% 降至 1-2%    """

核心道理：大模型是很厉害但不是万能的。一个精心设计的关键词路由能覆盖90%的场景，剩下的10%中8%靠上下文感知解决，真正需要LLM判断的只有不到2%。这就是分层的艺术——每一层解决自己能解决的最好解决的问题，解决不了的交给下一层。

你学到了什么？

一句话总结：大模型是锤子，但不是所有问题都是钉子。我从「三层LLM路由」→「双层关键词路由」→「四层置信度门控路由」，每一步都在回答同一个问题："这个决策，真的需要大模型吗？"记住：分层设计 + 置信度门控 + 缓存兜底，是让路由系统既快又准的三板斧。

三、第二课：RAG全链路——从原始问题到可信答案的五步流水线

RAG（检索增强生成）听起来就是三步：把问题转成向量 → 去向量库搜 → 把结果喂给大模型生成答案。但魔鬼在细节里。

实际落地中，一条完整的RAG管线需要经过五道工序，每一道都在解决一个具体问题：

下面我们逐一拆解这五道工序。

3.1 你得有个"门卫"——领域守卫

直接让用户问题和你的知识库匹配，会遇到一个问题：

用户问："帮我写一段 Python 冒泡排序"

系统也会傻傻地去"文旅知识库"里检索，找不到就胡诌。当然胡诌可以通过防幻觉手段加以限制，但是放任这种和我们当前场景无关的问题，会导致大模型浪费token，增加成本。

所以在检索之前，我加了一层轻量级门卫——让大模型花极小的代价（约180个 token，¥0.0002），先判断一下：这个问题跟西安旅游有关吗？

你学到的：在设计 AI 应用时，在管线的最前端加一道轻量级过滤，可以省掉下游大量的无效计算。就像快递分拣——先看地址对不对，对的才往下送。

3.2 听懂用户的"弦外之音"——指代消解

问题：用户说"它几点关门"，AI 怎么知道"它"是谁？

这是多轮对话中最头疼的问题。看这段对话：

用户："大雁塔门票多少钱？"

AI："大雁塔门票40元，学生半价……"

用户："它几点关门？"

如果不做处理，这个"它"字直接送进向量库检索，你猜会搜到什么？一堆包含"它"字的文档——全是噪音。RAG 检索直接报废，大模型收到的上下文跟"大雁塔"半毛钱关系都没有。

解法：两级流水线——关键词快筛 + LLM 兜底

我设计了一个两级检测机制：

第一程：关键词三级分层

我把指代词按出现概率分成三级，越靠前越明确：

第三级特别容易误判——"其他"里有"他"、"尤其"里有"其"（看着像"它"的变体）。我用了一个复合词排除列表，先把"其他""其它""尤其""其实""他人""他乡"从文本中抹掉，再检查"它/他/她"是否还在——还在的才是真指代。

第二程：LLM 轻量兜底

当关键词完全没命中时，就释放"终极武器"——一次极轻量的LLM调用（max_tokens=5，温度0）。这个调用大约是 50ms，额外成本可以忽略，但能把漏判率从~15%压到~3%。

就好比：门卫先看脸（关键词），不认识的人才查身份证（LLM）。99%的人看脸就够了，只有1%需要查身份证——效率最高。

改写环节：要校验，别盲目信任

检测到指代词后，调 LLM 做改写："它几点关门？"→"大雁塔几点关门？"。但 LLM 偶尔也会翻车——返回空字符串、只输出标点符号、或者直接复制原句。所以我加了一层轻量校验：

def_validate_rewrite(original, rewritten):ifnot rewritten ornot rewritten.strip():returnFalse# 空的，别用    stripped = rewritten.strip().rstrip("?？!！。，,、；;：:…~～ ")return len(stripped) > 0# 去掉标点还有字吗？

校验不通过就回退原句——宁可保留指代词让 RAG 打得不准，也不能把空字符串喂给后面的管线。

你学到了什么？

3.3 检索回来的东西要"去重"——MMR 算法

假设你的知识库里有 10 篇文章提了"大雁塔"，按相关性排序后，前 4 篇可能都是大雁塔的内容——虽然相关，但信息高度重复，浪费了大模型的 token 配额。

这里用了一个叫 MMR（最大边际相关性） 的算法。它的思想很简单：

打个比方：你去自助餐厅拿菜，第一盘拿最爱的红烧肉，第二盘还是红烧肉？不，你会拿点青菜——既要好吃，又要种类多。

# 伪代码解释 MMR 思想（不是完整实现）def 选菜(所有菜, 已拿的菜, 还要拿几盘):    候选 = 所有菜 - 已拿的菜for 每盘菜 in 候选:        得分 = 0.6 × 好吃程度 - 0.4 × 和已拿菜的重合度    选得分最高的那盘

你学到的：检索不只是"找相关的"，还要"找多样的"。否则大模型拿到 4 段几乎一样的内容，生成出来的答案也是重复的。

3.4 排序也要"两条腿走路"——两阶段重排

光有 MMR 还不够。向量检索（Embedding 相似度）和交叉编码器（Cross-Encoder）是两种不同粒度的打分方式：

• Embedding 相似度：把问题和文档分别编码成向量，算余弦距离——快，但粗糙（就像看封面判断一本书）
• 交叉编码器：把"问题+文档"拼在一起喂给模型打分——慢，但精确（就像翻开书读一段再判断）

我的重排管线是两阶段的：

graph LR    A[Milvus 召回8条<br/>Embedding 粗排] --> B[规则重排<br/>关键词重叠+商圈加权]    B --> C{开启API重排?}    C -->|是| D[交叉编码器精排<br/>bge-reranker-v2-m3]    C -->|否/失败| E[直接MMR去重]    D --> E    E --> F[输出Top 4]

规则重排的巧妙之处：在调 API 之前，先用 jieba 分词做一轮关键词重叠 + 商圈加权。比如用户问"回民街泡馍"，检索结果里文本中有"回民街"的就额外加0.2分——这是零成本的本地化优化。

交叉编码器的不可替代性：规则重排看的是"词"有没有对上，交叉编码器看的是"语义"对不对。比如用户搜"性价比高的面馆"，一篇讲"人均 15 的油泼面"（没出现"性价比"三个字）——规则会给低分，但交叉编码器能读懂它们在语义上是相关的。

降级保底：API 挂了怎么办？直接走规则重排，保证服务不中断。这就叫优雅降级——不因外部依赖故障影响核心链路。

# reranker.py 的精简逻辑defrerank(query, hits, top_k):if len(hits) <= top_k:return hits  # 候选少，直接返回if settings.rerank_enabled:try:return _api_rerank(query, hits, top_k)  # 交叉编码器except Exception:return _rule_based_rerank(query, hits, top_k)  # 降级return _rule_based_rerank(query, hits, top_k)

你学到的：排序不是只有一种算法。粗排保速度、精排保质量、降级保可用——三管齐下，才是一个生产级的检索系统。

3.5 防止大模型"胡说八道"——三层防线

大模型有时候会"幻觉"——明明知识库里没有，它也会编一段看起来很真的答案。我用了三层防线：

你学到的：不要指望一个 Prompt 就能解决幻觉问题。多层防御才是正道——就像你家有防盗门、监控、报警器三道防线，而不是只靠一把锁。

附加课：基础设施——LLM/Embedding 封装与 Milvus 踩坑

在讲完 RAG 和 Agent 两条主链路之后，有必要聊聊支撑它们的基础设施层。这些代码不直接面向用户，但决定了整个系统的稳定性和可维护性。

LLM 统一封装：一个入口管所有大模型调用

项目里所有调 DeepSeek 的地方（聊天、改写、摘要、质量修复……），都经过一个 llm.py 模块：

# llm.py 核心设计defchat_sync(messages, temperature=0.7, max_tokens=None, timeout=15.0):"""非流式：一次性返回完整文本（改写/摘要等短任务用）"""defchat_stream(messages, temperature=0.7, timeout=60.0):"""流式：逐段 yield 文本增量（SSE 打字机效果）"""

为什么分成两个函数？

短任务（改写一句话）理应秒回，超时 15s 就说明 API 异常；长回答可能需要几十秒，给 60s 宽限。

全局单例 + 连接池：OpenAI SDK 内部自带 httpx 连接池，整个进程只创建一个 Client 实例——所有请求复用同一条连接。

Embedding 封装：抽象层 + 工厂模式

向量化也做了类似封装，但多了一层抽象：

classEmbeddingClient(ABC):# 抽象基类    @abstractmethoddefembed_texts(self, texts: list[str]) -> list[list[float]]:        ...classSiliconflowEmbeddingClient(EmbeddingClient):# 硅基流动实现defembed_texts(self, texts):        resp = self._client.embeddings.create(model=self._model, input=texts)return [item.embedding for item in resp.data]@lru_cache(maxsize=1)  # 单例缓存defget_embedding_client() -> EmbeddingClient:return SiliconflowEmbeddingClient(...)

为什么加抽象层？因为将来换模型（比如从 bge-m3 换成 OpenAI text-embedding-3），只需新建一个实现类，改一行工厂函数——上层代码零改动。

Milvus 向量库的三个真实踩坑

选了 Milvus 做向量存储，踩了三个文档里没写的坑：

坑 1：VARCHAR 的 max_length 是按字节算的

我一开始设 text 字段 max_length=600，结果存中文就报错——因为 UTF-8 中文每个字占 3 字节，600 字节只能存 200 个汉字。

后来改成 max_length=16384（约 5000 中文字），够用了。

坑 2：delete 后 row_count 不减少

Milvus 的 delete 是软删除（写 delta log），不是真删。所以 get_collection_stats() 返回的 row_count 包含已删除的行。

正确做法是用 count(*) 聚合查询获取真实的存活行数。

坑 3：默认一致性是 Bounded

Milvus 3.0 默认 consistency_level="Bounded"——意思是"最终一致"，delete 后立即读可能读到旧数据。

我的解法：**查询侧显式设置 consistency_level="Strong"**。千级数据量下，强一致性的开销可以忽略不计。

# vector_store.py 关键配置hits = self._client.query(    collection_name=COLLECTION_NAME,    data=query_vector,    limit=top_k,    consistency_level="Strong",  # 强一致性，避免读到旧数据    ...)

你学到了什么？

四、第三课：Agent 全链路——缓存加速 → 并行执行 → 记忆管理

Agent 链路的核心挑战和 RAG 不同：**RAG 关心"答案准不准"，Agent 关心"执行快不快、记不记得住"**。

因为 Agent 需要反复调外部工具（查店铺、查优惠券、查笔记），每一次都是网络 I/O。如果让它们一个接一个跑，用户等得黄花菜都凉了。

我的 Agent 链路分五步：

4.1 工具结果缓存——同样的查询别再调第二次

Agent 调用工具是有成本的——Java HTTP 往返、向量库查询、LLM API 调用。如果用户接着问"那这家店的优惠券呢？"，Agent 可能再次调 get_shop_detail 查同一家店——完全没必要的开销。

我给工具加了分级缓存，按数据时效性分三档：

缓存键 = 工具名 + 参数哈希（SHA1），相同参数一定命中。用 LRU 淘汰策略 + 容量限制——优惠券缓存最多 500 条、店铺缓存最多 1000 条，防止内存泄漏。

还有一个细节：错误结果不缓存。如果 Java 后端挂了，返回的是 {"error": "查询失败"}，这种不能写缓存——否则 5 秒内的所有请求都吃到过期错误。

4.2 让工具"多线程工作"——并行执行

什么是 Agent？

简单说，Agent 就是能自己调用工具的大模型。你问"钟楼附近有什么泡馍店"，它会：

1. 思考：我需要查店铺
2. 调用工具：search_shops_by_name("泡馍")
3. 拿到结果：5 家店
4. 再思考：用户说"钟楼附近"，我得再查距离
5. 调用工具：list_shops_by_type("美食", x=108.94, y=34.26)
6. 综合结果，生成回答

问题：工具是一个一个调用，还是一起调用？

最初我的 Agent 是顺序调用工具的：

但实际上这三个工具互不依赖——查店铺不需要等优惠券结果，它们是独立的！

所以我改成了并行执行：

从700ms降到320ms，快了54%。

你学到了什么？

这就好比做饭：

• 顺序执行：先洗菜（5 分钟）→ 再切菜（3 分钟）→ 再烧水（5 分钟）= 13 分钟
• 并行执行：洗菜和烧水同时做，然后切菜 = 8 分钟

一句话总结：分析你的工具（函数）之间的依赖关系，互不依赖的就让它们同时跑，这是最简单的性能优化。

4.3 Agent 的"武器库"——6 个工具的设计哲学

并行执行解决了"快"的问题，但工具本身怎么设计？这直接决定了 Agent 能力边界。

工具分层：时效性决定数据来源

我的 Agent 有 6 个工具，按数据时效性分成两层：

为什么这样分？

• 优惠券库存每秒都在变——用户刚看到"剩 5 张"，下单时可能已经没了，所以必须实时查 Java
• 文旅知识（大雁塔历史）一年才更新一次——没必要每次都调 Java，本地向量库快照就够了

工具失败兜底：让 LLM 自己换策略

每个工具内部都有 try/except，但我不让它抛异常——而是返回结构化错误文本：

@tooldefsearch_shops_by_name(name: str) -> str:try:with JavaClient() as jc:            shops = jc.list_shops_by_name(name)return _json({"count": len(shops), "shops": shops})except JavaClientError as e:return _json({"error": f"查询失败: {e}"})

为什么这样做？因为 LangGraph 的 ToolNode 会把工具返回包成 ToolMessage 喂回 LLM。如果 LLM 看到的是 "查询失败: Java 服务超时"，它会自动换一个工具或者换个参数重试——就像人一样，这条路不通就换一条。

如果把异常直接抛出，整个 Agent 链路就断了，用户只能看到一个 500 错误页面。

返回 JSON 而非自然语言

所有工具都返回 JSON 字符串。为什么？因为 LLM 从结构化数据里提取字段，比从自然语言里"猜"字段准确得多：

// ✅ JSON 格式：LLM 能精确提取{"count": 5, "shops": [{"id": 1, "name": "老孙家泡馍", "avgPrice": 65}]}// ❌ 自然语言格式：LLM 可能漏掉字段"找到了 5 家店，其中老孙家泡馍人均 65 元..."

4.4 Python 怎么调 Java？——客户端封装的三个踩坑点

Agent 的 4 个实时工具都要调 Java 后端。这个"桥梁"我封装成了 JavaClient 类，踩了三个坑：

坑 1：连接池不复用 = 慢死

最开始我每次调用都 httpx.get(...) ——每次都新建 TCP 连接、TLS 握手、HTTP/2 协商……查一次店铺 50ms，其中 30ms 花在网络握手上了。

后来改成连接池复用：

classJavaClient:def__init__(self):        self._client = httpx.Client(            base_url=settings.java_base_url,            timeout=settings.java_timeout_seconds,        )# 整个生命周期复用同一个连接池

分页循环查20次店铺，从20次握手变成1次握手 + 19 次复用——延迟降了60%。

坑 2：Java 偶发抖动怎么办？

Java 服务偶尔会抖动（GC 停顿、网络瞬断），直接报错太粗暴。我加了指数退避重试：

• 第 1 次失败 → 等 0.5s → 重试
• 第 2 次失败 → 等 1s → 再重试（间隔翻倍）
• 第 3 次还失败 → 放弃，抛异常

重要细节：只有网络异常才重试（HTTPError），业务失败（success=false）不重试——数据错了重试 100 次也是错。

坑 3：统一信封格式解包

Java 后端所有接口都返回统一信封：{success, errorMsg, data}。我在 _get_list() 里统一解包，上层代码永远看不到信封，只看到干净的数据：

def_get_list(self, path, params):    resp = self._client.get(path, params=params)    body = resp.json()ifnot body.get("success"):raise JavaClientError(f"{path}: {body['errorMsg']}")return body["data"]  # 上层只拿到 data 列表

你学到了什么？

4.5 会话记忆——Redis 持久化 + 智能摘要

最开始我用 Python 内存字典存对话历史——服务一重启，记忆全丢。用户刚聊了两轮，刷新页面就"失忆"了。

现在换成了 Redis：

但这还不够。当你聊了 10 轮之后，每轮都要把全部历史喂给 LLM，token 费用越来越高，响应也越来越慢。

所以我又加了一层智能摘要记忆：超过 3 轮对话后，自动调 LLM 把历史压缩成结构化摘要：

{"user_preferences": ["偏好泡馍", "人均预算80以内"],"constraints": ["钟楼附近", "要有优惠券"],"decisions": ["选择了老孙家泡馍"],"key_entities": ["回民街", "老孙家泡馍", "人均65"]}

压缩后，10 轮对话变成 3 条消息（摘要 + 最近 1 轮），token 消耗降低 70%+。而且摘要失败自动降级——LLM 调用超时或报错，就回退到原始滑动窗口，不阻塞对话。

你学到了什么？

五、第四课：知识库的质量比数量重要——PDF 解析的血泪史

问题：PDF 转文本，一塌糊涂

我收集了 37 篇西安旅游文档。Markdown 格式的文本很干净，但 PDF 就头疼了——解析出来的内容经常是：

• 🚫 乱码："大雁塔建于唐⿏永徽三年"——"⿏"是什么鬼？
• 🚫 缺段：原文 10 段，解析出来只剩 6 段
• 🚫 表格碎裂：门票价格表变成了乱七八糟的数字堆

垃圾进，垃圾出——如果一堆乱码进了向量库，用户检索到的就是乱码。

解法：给每篇文档体检

我设计了一个"文档体检"流程：

你学到了什么？

文档怎么切？——分块策略的三种模式

质量体检解决了"文档脏不脏"的问题，但还有一个更基础的问题：一篇 5000 字的文档，怎么切成小块喂给向量库？

这看起来简单（按固定字数切不就行了？），但实际上是个语义完整性问题。

模式一：按标题切——语料 Markdown 的首选

我的文旅知识库大部分是 Markdown 格式，天然有 ## 二级标题 结构：

## 大雁塔历史大雁塔建于唐永徽三年（652年）...## 开放时间旺季（3月-11月）：8:00-18:00...

如果按固定 500 字切，可能把"历史"的最后一句和"开放时间"的第一句拼到一起——语义就乱了。

所以我选择按 ## 标题切分，每个标题下的内容自成一个 chunk。这样检索时命中的块，语义是完整的。

模式二：按段落切——PDF/Word 的 fallback

PDF 和 Word 没有标题结构，退而求其次按空行（段落）切分。段落是作者写好的自然语义边界，比固定字数靠谱。

模式三：表格特殊处理——带表头重复拆分

这是最 tricky 的地方。普通文本用 overlap（重叠）就能保证上下文连续，但表格不行：

我的做法是：检测到管道符表格后，第一行当表头，后续数据行贪心装箱，每块都带上表头。这样无论命中哪一块，都能看到列名。

超长内容怎么办？

不管哪种模式，单个 chunk 超过 600 字就按句子边界续切（。！？\n），相邻块重叠 120 字（约 20%，业界推荐区间 15-25%）。为什么按句子？因为不能把一句话拦腰截断——"大雁塔建于唐永徽三年"切成"大雁塔建于唐永" + "徽三年"就没法看了。

结构化数据：模板化而非字段裸拼

店铺、笔记、优惠券这些结构化数据，我不硬切——一条记录就是一个 chunk。但直接把 JSON 字段拼进去（name=老孙家 avgPrice=65）效果很差，向量模型在自然语言上表现最好。

所以我把字段模板化成自然中文：

defshop_to_text(shop):returnf"""店铺名称：{shop['name']}人均消费：{shop['avgPrice']}元地址：{shop['address']}营业时间：{shop['openHours']}评分：{shop['score']}分简介：{shop['summary']}"""

入库流水线：从文件到向量的完整旅程

体检和分块都准备好了，怎么把它们串起来跑？我设计了一个格式无关的入库流水线：

核心设计：

1. Reader 注册机制：新增格式只需创建一个 XxxReader 类 + @register_reader 装饰器，ingest_service 自动发现——零侵入
2. 幂等设计：每个 source 先 delete_by_source 清空再重灌——数据改了重跑不会残留旧 chunk
3. 部分失败容错：Java 源失败不影响 corpus 源入库，最终报告里带 errors 字段，谁挂了一目了然

触发方式：支持两种触发：

• 命令行：python scripts/ingest.py（离线批量）
• HTTP API：POST /api/ai/admin/ingest（仅限本机 127.0.0.1）

你学到了什么？（分块与入库篇）

六、第五课：怎么知道你的 AI 做得好不好？

这个问题我在平时开发中就意识到了，如何知道自己的决策对不对，做出了的东西好不好。很多人会凭感觉，但感觉是不能量化的。

我的四层评估体系

路由评估从最初的 300 条手工用例，升级为场景矩阵自动生成——按「景点×模板」「美食×模板」「地点×店型×模板」系统构造测试集，覆盖 RAG:Agent:Boundary ≈ 60:30:10 的真实分布比例，避免手工标注的偏差。

一个关键技巧：不要让 AI 自己评自己

评估 AI 生成质量时，如果你的评估模型和生成模型是同一个，它很容易"偏心"——自己的回答自己评，当然高分。

我的做法是跨家族评估：用阿里的 Qwen3.7 来评估 DeepSeek 生成的回答，公平性大幅提升。

这就好比：让语文老师批改数学作业，虽然不如数学老师精准，但至少不会包庇自己的学生。

你学到了什么？

一句话总结：没有评估的 AI 系统就像没有仪表盘的汽车——你根本不知道它跑得怎么样。

七、第六课：给 AI 应用装上"体检仪"——Prometheus + Grafana 监控栈

问题：代码跑起来了，但你真的知道它"健康"吗？

评估体系（第五课）告诉你"做得对不对"，但那是离线评估——用测试集跑一轮，看分数。它不能告诉你线上正在发生什么：现在 QPS 是多少？P99 延迟有没有飙升？错误率突然从 0.1% 跳到 5% 了？

生产环境需要的是实时可观测性。

我的监控方案：两层视角

运维视角（Prometheus）：用 prometheus-fastapi-instrumentator 一行代码自动采集所有 HTTP 指标，暴露 /metrics 端点。Docker 一键启动 Prometheus + Grafana：

# docker-compose.monitoring.ymlservices:prometheus:image:prom/prometheus:latestports:["9090:9090"]grafana:image:grafana/grafana:latestports:["3000:3000"]

业务视角（RoutingMonitor）：Prometheus 只管 HTTP 层面的事，但路由决策的质量需要业务诊断。我专门写了一个 RoutingMonitor，记录每次路由决策的维度信息：

• 这次走了 Agent 还是 RAG？
• 有没有命中关键词？命中了哪些？
• 有没有发生意图冲突？冲突怎么解决的？
• 使用了哪一层路由？（Layer 1 预过滤 / Layer 2 关键词 / Layer 3 上下文 / Layer 4 LLM）
• 置信度是多少？有没有触发置信度门控？
• 有没有"无特征命中→兜底 RAG"的案例？（说明关键词需要补充）

最有价值的功能是可疑案例自动标记——自动识别两类可疑路由：

1. 意图冲突案例：Agent 和 RAG 关键词同时命中，冲突解决可能出错
2. 无特征命中案例：完全没命中任何关键词，走到了 Layer 4 兜底 → 说明关键词有盲区

这些案例自动汇集，定期人工审核——发现"哦，原来用户问'这个地方怎么样'里的'怎么样'应该归类为 Agent"，然后补充关键词。形成数据驱动的路由优化闭环。

你学到了什么？

一句话总结：没有监控的 AI 应用就像黑盒——你不知道它在干什么，也不知道什么时候会出问题。

八、第七课：不只会聊天——AI 辅助写作功能

不止是问答，还能帮你写探店笔记

你可能注意到，这个项目不光能做问答，还能帮用户生成笔记标题、润色正文、分析情感。这些功能是给谁用的？给发探店笔记的用户——在平台上写了笔记后，点一下"AI 生成标题"，5 个候选标题秒出。

代码架构：Java 做"门卫"，Python 做"厨师"

注意这里的职责分工：

• Java 负责鉴权：/ai/assist/** 路径不在 MvcConfig 白名单中，LoginInterceptor 天然拦截——零配置获取鉴权能力。为什么要登录才能用？因为这是辅助你写笔记的功能，没登录哪来的笔记？
• Python 负责 AI：标题生成、正文润色、情感分析，全是调 DeepSeek 完成。
• Java→Python 用 HTTP 转发：复用已有的 Hutool（项目 pom 里本来就有），不引入新依赖。

三个 AI 辅助功能

这里有个工程常识：生成任务高温、分析任务低温。标题要多样性，温度拉到 0.9；情感分析要稳定，温度压到 0.1。同样一个模型，温度调对了就是不同的"人格"。

容错设计：解析 AI 输出是个技术活

LLM 输出不完全可靠。比如让它生成 5 个标题，它可能：

• 每行加个编号"1. 2. 3."
• 包在 markdown 代码围栏里 ...
• 前面来一段废话"好的，以下是 5 个标题："

解析代码要做多层容错：先按行拆分 → 去掉编号前缀 → 过滤空行和超长行 → 不够 5 个就截断原文前 30 字兜底。

情感分析的 JSON 解析更麻烦——LLM 可能在 JSON 外包一堆文字。所以我用正则先捞出第一个 {...}：

match = re.search(r"\{.*\}", resp, re.S)data = json.loads(match.group())  # 这样就不会被外围文字干扰

解析失败？回退默认值 {"sentiment": "中性", "score": 50}——辅助功能不能挡住用户发笔记。

你学到了什么？

九、第八课：架构设计的"分家哲学"

为什么用 Java + Python 两种语言？

很多人问我：一个项目用两种语言不麻烦吗？我的回答是：不同的问题，用最合适的工具。

• Java 管业务：用户系统、店铺管理、订单、优惠券——这是 Spring Boot 的舒适区
• Python 管 AI：调用大模型、向量检索、Agent 编排——这是 Python AI 生态的天下
• Nginx 做分流：根据 URL 前缀，自动把请求送到对应的服务

你学到了什么？

核心理念：不要让一个语言干所有事。就像你不会用菜刀拧螺丝、用螺丝刀切菜——每个工具都有自己最擅长的场景。

十、第九课：那些让你怀疑人生的 Bug

做项目不可能一帆风顺。分享两个让我印象深刻的坑：

Bug 1：向量检索偶尔卡死 30 秒

现象：Milvus 向量检索大部分时间正常，但偶尔会卡死 30 秒以上。

原因：Milvus 的 segment 是按需加载到内存的。首次查询某个 segment 时，会触发 index loading，这个过程会阻塞其他查询。

教训：给所有外部调用加超时（我的设置是 5 秒），超时了就优雅降级——返回空结果总比用户干等好。

Bug 2：前端打字机效果变成"一句话突然全出来"

现象：后端是流式输出，但前端看到的效果是卡几秒后突然全部显示。

原因：Nginx 默认会缓冲响应数据，即使后端用的是 SSE（Server-Sent Events）流式协议，Nginx 也会等数据积累到一定量再发给前端。

教训：用了反向代理就要检查它的缓冲设置。加一行 proxy_buffering off; 就解决了。

十一、总结：如果只记住三件事

做完这个项目，如果让我只总结三句话：

1️⃣ 能用简单方法的，别用大模型

四层路由中，99% 的请求在 Layer 1-2 就被处理（关键词匹配 + 预过滤），<2ms 延迟、零成本。剩下 1-2% 模糊 case 才会调用 LLM，而且有缓存兜底——真正需要大模型的场景，只占总流量的不到 2%。

2️⃣ 多层防御 > 一层完美

防幻觉有三层（分数闸门 + Prompt 约束 + 引用溯源），质量有四级评估。不要幻想一个环节做到 100% 就能解决所有问题。

3️⃣ 分而治之，各司其职

Java 管业务，Python 管 AI，Nginx 管分流，Milvus 管检索，Prometheus 管监控。每个组件只做自己最擅长的事。

最后的话：这个项目前后花了我几个月的时间，从"能跑就行"到"有评估、有监控、有降级"，每一轮重构都在加深对 AI 应用开发的理解。如果你也在学习，不要怕做得丑——先让它跑起来，再慢慢优化。一个能跑的项目，胜过十个完美的计划。

项目开源在 GitHub：changAn_AI^[1]，欢迎Star⭐和交流！ 任何项目总有漏洞，有问题的话欢迎提出，我们一起优化。

本文所有数据来自由项目 scripts/eval_*.py 系列评测脚本，可完整复现。所有技术细节可在项目源码中验证。