pokemon agent runtime 系列（三）：本地直答、缓存与 RAG 主链

系列导航：上一篇见 pokemon agent runtime 系列（二）：一次提问的完整调用链；下一篇见 pokemon agent runtime 系列（四）：LangGraph Agent 编排。本文负责回答：普通模式下，本地直答、缓存和 RAG 主链到底是怎么串起来的。

很多人一提到 AI Agent 或 RAG，就会下意识把焦点放在“多轮调度”“工具调用”“LangGraph”“supervisor”这些听起来更像智能体的部分。但如果你真的在做一个面向真实用户的问答系统，你很快会发现：很多请求根本不需要进入 Agent。真正决定系统体验的，往往不是“最复杂的那条路径”，而是普通模式下那条最常被命中的主链到底是否足够稳、足够快、足够可控。

pokemon agent 就是一个很有代表性的例子。它虽然支持 Agent 模式，但普通聊天模式本身已经不是简单的“把 query 扔给模型”。在实际执行上，这条链会先经过本地事实直答、语义缓存、Retriever 多源证据聚合，最后才进入模型生成。也就是说，它已经具备了一个成熟 RAG 系统该有的大部分工程骨架，只是没有把流程控制权完全交给 Agent Loop。

这篇文章就专门讲这一条链：一个不用 Agent 的聊天模式，如何通过短路、缓存、检索增强和 prompt 构造，做出一条既低延迟又可扩展的问答主链。

TL;DR

先厘清一个误区：不是所有问答系统都应该一上来就进 Agent

实际工程里，Agent Loop 的最大问题从来不是“能不能工作”，而是：

• 成本更高
• 延迟更高
• 路径更不稳定
• 更难约束与回放

因此，一个健壮的系统通常不会把所有请求都直接送进 Agent。更合理的做法是：先把那些高确定性、低成本、可快速结束的路径前置。只有这些路径无法解决的问题，才继续进入更昂贵的执行层。

pokemon agent 的普通模式就是这样设计的。它不是 Agent 的“简化版”，而是一条经过精心分层的执行链：

1. 能直接从本地数据回答的问题，立即返回
2. 能从缓存复用的问题，立即返回
3. 需要外部证据的问题，进入 Retriever 聚合证据
4. 只有到了最后，才交给 LLM 做语言生成

从系统设计角度看，这种结构的意义非常明确：把复杂性往后推，把确定性往前放。

一张图看懂普通模式主链

这张图最想表达的一点是：普通模式并不是“少一个 Agent”，而是“多了三个前置决策层”。如果这些前置层设计得好，系统就能在大多数日常问题上把成本、延迟和复杂度压到很低，而不需要把所有请求都提升到 Agent 编排级别。

第一层：本地事实直答为什么比 RAG 还重要

对一个垂直领域问答系统来说，最理想的情况不是“什么都用检索 + 生成来做”，而是先识别哪些问题根本不值得做检索。

pokemon agent 在 chat_router.py 里首先做的，就是对宝可梦事实类问题做本地直答短路：

# 0. Local deterministic Pokédex shortcut
if decision.intent == Intent.POKEDEX_FACTS and (decision.confidence >= 0.8 or decision.needs_clarification):
    content = format_basic_facts(rec) + "\n\n" + format_evolution(rec, data=data)
    refs = {
        "model_name": "local_dataset",
        "knowledge_base": {
            "results": [],
            "message": "Answer来自本地数据集（无 LLM 调用）",
        },
    }
    yield make_chunk(meta, content=content, status="loading")
    yield make_chunk(meta, status="finished", refs=refs)
    return

这段逻辑体现的是一个很经典、但很容易被忽视的 RAG 设计原则：

对那些可以被确定性数据源稳定回答的问题，最好的 RAG 往往是“不要走 RAG”。

为什么这一步值钱

1. 延迟最低

不需要 embedding，不需要向量检索，不需要模型生成，直接命中本地数据。

2. 成本最低

不产生额外 token，也不依赖外部服务。

3. 幻觉风险最低

在明确事实问答上，最稳的永远不是生成，而是直接查结构化或半结构化数据。

这也解释了一个常见误判

很多人会说：“我把知识库关了，为什么系统还能答对？”

在这个项目里，一个非常常见的答案就是：因为它根本没走知识库，它在第一层就已经命中了本地事实直答。

所以从系统可观测性角度看，单看“回答正确”远远不够，必须看 refs 或 model_name 才知道它到底走了哪条路径。

第二层：语义缓存不是为了偷懒，而是为了稳定复用低风险答案

本地直答之后，系统会尝试检查语义缓存。但这个缓存并不是“所有请求都先查一下”，而是被刻意限制在一个很保守的适用边界里：

safe_cache = (not history) and (not need_retrieve(meta))
if safe_cache:
    cached_response = cache.get(query, meta=cache_meta)
    if cached_response:
        yield make_chunk(meta, content=cached_response, status="finished")
        return

这意味着缓存只在以下条件同时满足时生效：

• 没有历史消息
• 当前请求不需要检索

为什么要限制得这么严格

因为缓存最怕两类错误：

第一类：上下文污染

同样一句话，在不同历史上下文下可能含义完全不同。如果缓存不区分上下文，复用结果就很危险。

第二类：外部状态漂移

如果一个答案依赖 KB / Web / MCP / Graph 这些外部证据源，那么缓存它意味着把一个本应动态计算的结果“冻结”下来，后续极容易过期或错误。

这个缓存还有一个工程细节很值得注意

src/knowledge/cache/cache.py 里的 SemanticCache 不是纯字符串缓存，而是 embedding 语义缓存。也就是说，它支持两类命中：

1. exact match：完全相同的 query
2. similarity match：语义相近、相似度超过阈值的 query

if entry is not None and entry.meta == norm_meta:
    logger.info("Cache HIT (exact)")
    return entry.response
 
query_embedding = np.array(model.embed_query(query))
...
if best_similarity >= self.similarity_threshold:
    logger.info(f"Cache HIT: similarity={best_similarity:.3f}")
    return best_match

这让缓存不只是“相同问题复用答案”，而是能覆盖一部分语义近似提问，从而进一步降低重复 LLM 调用。

但它为什么仍然可控

因为它还有一层 meta 命名空间约束：

• model_provider
• model_name
• system_prompt_sha

也就是说，即使 query 相同，只要模型或系统提示词环境不同，也不会错误复用缓存。这是一个非常典型的工程实践：缓存不是只按输入文本做 key，而是按执行语义空间做 key。

第三层：Retriever 不是“接一个向量库”，而是多源证据编排器

当系统发现请求需要检索时，普通模式会把问题交给 src/knowledge/core/retriever.py。这个模块在架构里的位置特别关键，因为它并不是单纯的 KB 封装器，而是负责统一管理多种证据源：

• Knowledge Base（Milvus 向量检索）
• Knowledge Graph（Neo4j 图谱问答）
• Web Search
• MCP

它的核心入口大致是：

def retrieval(self, query: str, history: list[dict[str, Any]], meta: dict[str, Any]) -> dict[str, Any]:
    refs = {"query": query, "history": history, "meta": meta}
    refs["entities"] = self.reco_entities(query, history, refs)
 
    requested = sum(
        1
        for v in (
            bool(meta.get("db_id")),
            bool(meta.get("use_graph")),
            bool(meta.get("use_web")),
            bool(meta.get("mcp_id")),
        )
        if v
    )

这段代码看上去简单，但它已经透露出 Retriever 的三个核心职责。

1. 它会统一携带上下文与引用容器

refs 不是一个临时变量，而是这条检索链的核心数据结构。它把：

• 原始 query
• history
• meta
• entities
• 各检索源结果

统一组织进一个引用对象里。后续无论是构造增强 query，还是回传给前端做引用展示，都依赖它。

2. 它会按请求能力动态选择检索源

Retriever 不会无条件查所有源，而是由 meta 决定：

• db_id 决定是否启用知识库
• use_graph 决定是否启用图谱
• use_web 决定是否启用 Web Search
• mcp_id 决定是否启用 MCP

这意味着 Retriever 是一个runtime router，而不只是一个“数据库访问类”。

3. 它支持并行证据召回

当同一个请求开启多个能力源时，它会并行执行：

if requested >= 2:
    with ThreadPoolExecutor(max_workers=4) as ex:
        fut_kb = ex.submit(self.query_knowledgebase, query, history, refs)
        fut_graph = ex.submit(self.query_graph, query, history, refs)
        fut_web = ex.submit(self.query_web, query, history, refs)
        fut_mcp = ex.submit(self.query_mysql_mcp, query, history, refs)

这点非常重要，因为这意味着系统对多源 RAG 的理解不是“多查几个地方”，而是“尽量把延迟压低到接近最慢那条分支”。

知识库路径：向量检索不是默认入口，而是带 feature flag 的可选能力

知识库检索的具体逻辑在 query_knowledgebase(...)，它有两个前置条件：

1. 当前请求带了 db_id
2. enable_knowledge_base 为真

def query_knowledgebase(self, query, history, refs):
    db_id = meta.get("db_id")
    if not db_id or not feature_enabled("enable_knowledge_base"):
        response["message"] = "知识库未启用、或未指定知识库、或知识库不存在"
        return response

这意味着 KB 不会因为“系统里存在 Milvus”就自动生效，它同时依赖：

• 前端是否真的选中了数据库
• 后端 feature flag 是否开启

检索前还会先做 query rewrite

真正进入 KB 搜索前，Retriever 会调用 rewrite_query(...)：

rw_query = self.rewrite_query(query, history, refs)
kb_res = kb.search(
    query=rw_query,
    db_id=db_id,
    distance_threshold=meta.get("distanceThreshold", self.default_distance_threshold),
    rerank=True,
    top_k=meta.get("topK", self.top_k),
)

这也是这条链和一个“最简向量检索 Demo”最大的区别之一：它不是直接拿用户原始问题去查，而是给查询改写留了一个明确的钩子位。

在实现上，这一步支持至少三种模式：

• off：不改写
• on：调用 prompt 做 query rewrite
• hyde：走 HyDEOperator

if rewrite_query_span == "off":
    rewritten_query = query
else:
    rewritten_query_prompt = rewritten_query_prompt_template.format(history=history_query, query=query)
    rewritten_query = model.invoke(rewritten_query_prompt).content
 
if rewrite_query_span == "hyde":
    res = HyDEOperator.call(model_callable=model.invoke, query=query, context_str=history_query)
    rewritten_query = res.content

这意味着这个项目的 RAG 不是“只做向量召回”，而是已经把查询变换纳入了可调度链路里。

一个很常见的误解：rerank 开了，到底改变了什么

很多人第一次看 RAG 链路时，会把 rerank 理解成“开了才会把检索结果塞进 prompt，不开就不塞”，或者反过来理解成“不开就把所有结果全塞进去”。这两种理解都不准确。

更准确的说法是：

• 不开 rerank：系统直接按向量检索结果取前 top_k
• 开 rerank：系统会先多召回一批候选，再用 reranker 二次打分、过滤、排序，最后仍然只保留前 top_k

也就是说，rerank 改变的是候选结果如何被二次筛选和重排，而不是“是否把所有结果都塞进 prompt”。真正进入 prompt 的，始终只是被筛过的一小部分上下文。

这点非常重要，因为它解释了为什么打开 rerank 往往会提升回答质量：它并不是让系统“知道更多”，而是减少那些“向量相似但回答并不真正相关”的片段进入最终上下文。

如果用一句非常工程化的话来概括，那就是：

rerank 不改变“要不要进 prompt”，而是改变“哪些片段更值得进 prompt”。

图谱、Web、MCP：这不是单一 RAG，而是多源增强

如果只盯着 KB，很容易把这个系统误解成“一个向量库问答系统”。但实际上它更像一个多源增强框架：

这张表很关键，因为它说明：在普通模式下，系统已经具备一种“轻量 orchestrator”特征。虽然它没有进入 LangGraph worker 调度，但它已经会根据请求上下文动态打开不同证据源，并把它们聚合到同一个生成入口。

construct_query()：RAG 的真正关键往往不是检索，而是如何把证据送进模型

检索完成后，Retriever 不会直接返回答案，而是会调用 construct_query(...) 把结果转成最终增强输入。

external_parts = []
 
graph_answer = refs.get("graph_base", {}).get("answer")
if isinstance(graph_answer, str) and graph_answer.strip():
    external_parts.append("图谱回答:\n" + graph_answer.strip())
 
kb_res = refs.get("knowledge_base", {}).get("results", [])
if kb_res:
    kb_text = "\n".join(self.clean_kb_text(kb_res))
    external_parts.append("知识库信息:\n" + kb_text)
 
web_res = refs.get("web_search", {}).get("results", [])
...
 
if external_parts:
    external = "\n\n".join(external_parts)
    query = knowbase_qa_template.format(external=external, query=query)

这段逻辑揭示了一个 RAG 工程里非常常见、但也最容易被忽视的事实：

检索并不会自动改善生成，真正起作用的是你如何把检索结果组织成模型更容易利用的上下文。

在这个项目里，construct_query(...) 做了三件事：

1. 统一格式化不同证据源
2. 对 KB 结果进行去重与截断
3. 将多源证据拼成一个统一外部上下文，再嵌入模板

这说明它对“生成前上下文组织”是有明确设计意识的，而不是简单把一堆检索结果直接 append 在 query 后面。

一张图看懂 Retriever 的内部角色

从这个视角看，Retriever 更像一个evidence orchestrator，它负责把多个来源的证据组织成一个模型更容易消费的输入，而不是简单把每个检索器暴露给上层各自调用。

知识库本身：为什么它被设计成懒加载能力，而不是启动时强绑定

src/knowledge/store/knowledgebase.py 也很值得一看，因为它揭示了这个系统对知识库的设计思路：知识库能力并不是服务启动时就强行初始化，而是懒加载的。

# 初始化（轻量）
self._check_migration()
# 注意：Milvus/Embedding 初始化会阻塞/失败（服务未启动、未配置 key 等），
# 因此这里改为懒加载，直到真正调用 KB 相关能力时再连接。

并且在真正使用前会统一经过 _ensure_ready()：

if not self._is_enabled():
    raise RuntimeError("KnowledgeBase is disabled (enable_knowledge_base=false).")
 
if self.embed_model is None:
    self._load_embedding_model(self._embedding_config)
 
if self.client is None:
    self._connect_milvus(self._milvus_uri)

这种设计的工程价值很高，因为它解决了一个真实问题：

• 后端本身应该能启动
• 但向量库、embedding 服务、reranker 服务不一定每次都齐全
• 即使 KB 依赖缺失，也不应该拖垮整个聊天主服务

所以知识库在这里不是“应用启动的前提条件”，而是“按需激活的可选能力”。这和很多现代 AI 系统的架构取向是一致的：主服务可运行，增强能力按需挂载。

用一张表总结普通模式的各层职责

这张表很适合作为普通模式的核心理解框架：每一层都不是为了“多做一步”，而是为了把最昂贵、最不确定的生成尽可能往后推。

为什么这条链很像 RAG，但又不只是传统 RAG

如果只用教科书定义，RAG 是“先检索，再生成”。但 pokemon agent 的普通模式显然比这复杂得多。它更像一条带前置短路和多源增强的执行链：

• 前面有本地事实直答
• 中间有 embedding 语义缓存
• 检索阶段不是单一向量检索，而是 KB / Graph / Web / MCP 的联合增强
• 查询在进入 KB 之前还可能被 rewrite 或 HyDE
• 最终不是直接把检索结果交给用户，而是先被拼装成增强 query 再进入模型

因此，更准确地说，这不是一个“向量库聊天系统”，而是一个带多层前置优化的增强生成主链。

这类系统比一个极简 RAG Demo 更接近真实生产实践，因为它真正处理了工程里最棘手的问题：

• 什么问题根本不值得检索
• 什么问题可以直接缓存
• 多个证据源如何统一组织
• 最终上下文怎么构造成模型更容易利用的形式