预算受限下的智能体搜索：从混合召回到LLM路由的工程实践

最近在做知识库问答和垂直搜索相关的项目时，有一个感受越来越明显：传统的搜索结果页已经很难满足用户对“答案”的预期，但完全依赖大模型把搜索词直接“翻译”成答案，成本又高得让人不敢放开用。随着“智能体搜索”这个概念被反复提起，越来越多团队开始尝试用 Agent 的思路重构搜索链路，却发现卡点往往不在算法，而在“预算”——有限的 token、有限的 API 调用额度、有限的计算资源。这篇文章就从工程落地的角度，系统拆解一下预算受限场景下智能体搜索的设计思路，并结合一套可运行的轻量检索系统，把查询理解、混合召回、粗排精排、LLM 预算路由这些关键环节讲清楚。无论你是正在做 RAG 应用，还是想优化搜索体验，这篇文章都能给你一份可以照做的方案参考。

1. 智能体搜索是什么：从关键词搜索到意图认知

智能体搜索并不是一个全新的搜索引擎，而是把传统搜索系统与 LLM 的推理、规划、工具调用能力结合起来，形成一个“能理解问题、能拆解任务、能调用检索工具、能组织答案”的完整链路。它和传统搜索最本质的区别在于：传统搜索返回的是“链接列表”，智能体搜索返回的是“经过推理和整合后的答案”。

1.1 传统搜索的核心局限

传统搜索以关键词匹配为核心，用户输入“什么显卡跑深度学习性价比高”，搜索引擎返回的是包含“显卡”“深度学习”“性价比”这些词的网页列表。这里有几个很明显的问题：

• 语义鸿沟：用户表达的是“意图”，系统匹配的是“字面词”，同义词、口语化表达、隐含条件都容易被漏掉。
• 信息碎片化：用户需要点开多个链接，自己拼接答案，判断哪个信息更权威、更符合自己的预算和使用场景。
• 无法处理复杂任务：比如“对比 4090 和 7900 XTX 在 24GB 显存下的训练效果差异，再推荐两套整机配置”，这种多条件、多步骤的问题，传统搜索几乎无能为力。

1.2 智能体搜索带来的变化

智能体搜索把搜索从一个“查询到结果”的单跳过程，变成了一个“理解到生成”的多跳过程。整个链路通常是这样：

1. 用户输入自然语言问题。
2. Agent 对问题进行意图识别和查询改写，拆出关键实体、约束条件、对比维度。
3. 根据任务需要，调用多个检索工具：全文检索、向量检索、数据库查询、外部 API。
4. 对召回结果进行重排，筛选出真正有价值的信息片段。
5. 把筛选后的信息交给 LLM 组织成结构化的答案，并提供引用。

这个过程解决了两类关键问题：一类是“用户表达不精确”的问题，通过改写和意图识别来对齐；另一类是“信息过载”的问题，通过检索、重排、过滤，让 LLM 只看到高价值信息，而不是把整个互联网都塞进上下文。

1.3 预算受限带来的新挑战

但智能体搜索并不是免费午餐。每一个环节都在消耗资源：向量化需要计算资源，召回需要数据库查询，重排需要模型推理，最后还有一次甚至多次 LLM 调用。如果对每个 query 都做完整的 Agent 链路，成本会迅速失控。

这就是“预算受限下的智能体搜索”要回答的核心问题：如何在效果接近完全版智能体搜索的前提下，把单位查询成本压到可以接受的范围。换句话说，不是每个问题都需要完整的 Agent 规划，也不是每个问题都需要调用最强的大模型。聪明的搜索系统应该学会“按需分配”。

2. 预算从哪儿来：智能体搜索的成本构成

在做任何优化之前，先要把成本结构拆清楚。否则优化就很容易变成“拆东墙补西墙”，表面上省了钱，实际效果却大打折扣。

2.1 成本构成四大块

一个智能体搜索请求的成本，通常由四部分组成：

这里最容易被忽略的是第一项和第四项。Embedding 虽然单次便宜，但它是所有流量都要经过的环节；LLM 生成看起来只是最后一步，实际上因为每次都会携带几千甚至上万 token 的上下文，成本远高于前几步。

2.2 一个简单的成本估算示例

假设每天有 10 万次搜索请求，每次请求带着 2000 token 的上下文调用一次 LLM，输出 300 token。按比较克制的价格估算（不同模型差异很大，这里只是为了说明计算思路）：

输入：2000 token × 10万次 = 2亿 token 输出：300 token × 10万次 = 3000万 token

如果输入侧的单价是每百万 token 若干元，这意味着一项日成本轻松达到上百甚至上千元。这还只是 LLM 生成部分，没有算 embedding、检索和重排。

所以预算受限下的第一原则是：让 LLM 只处理那些真正需要它处理的问题。能靠普通检索解决的，就不要让 Agent 上；能靠小模型解决的，就不要上大模型；能靠缓存解决的，就一次都不要多计算。

2.3 预算约束如何影响架构设计

预算约束不只是“省钱”，它会直接改变系统的架构形态：

• 从“一次查询一条完整链路”变成“按问题难度动态选择链路”。
• 从“全部向量检索”变成“向量 + 关键词混合检索”，减少无效召回。
• 从“每次都调 LLM”变成“缓存优先、规则兜底、LLM 最后兜底”。
• 从“单一强模型”变成“大小模型分级路由”。

预算受限本质上是在倒逼系统做“合理性判断”：这个问题值不值得花这么多钱去回答？这个答案能不能先用低成本方式拿到？整个架构设计的核心，就是把有限的预算花在最能提升用户体验的那个环节上。

3. 预算受限下的智能体搜索整体架构

在预算受限的前提下，我比较推荐一种“分层漏斗式”架构。这个架构的核心思路是：让大多数简单请求在低成本的早期阶段就被满足，只有少数复杂请求才进入高成本的 LLM 生成阶段。

3.1 分层漏斗架构

整个流程分成四个层级，每一层都有能力直接返回结果：

用户查询 │ ▼ 第 1 层：缓存层（命中直接返回） │ 未命中 ▼ 第 2 层：轻量检索层（关键词 + 向量混合召回，规则重排后直接返回 Top 结果） │ 置信度不足 ▼ 第 3 层：LLM 精排 + 生成层（小模型先精排，强模型最后兜底） │ ▼ 第 4 层：完整 Agent 规划层（多步搜索、工具调用、对比总结）

在这个架构里，不是每个请求都要走到底。大多数简单问题在第二层就可以结束，只有那些包含对比、推理、多条件约束的问题才需要进入第三层甚至第四层。这样既保证了效果，又把预算控制在了合理范围。

3.2 关键决策点：什么时候“升级”

分层架构中最重要的问题是：如何判断一个查询需要在当前层返回，还是继续向上“升级”？

这里需要建立一套信号体系：

• 检索结果的分数分布：如果 Top 1 和 Top 5 的分数差距很大，说明答案比较明确，可以直接返回；如果差距很小，说明候选结果本身就很接近，需要 LLM 进一步判断。
• 查询复杂度：包含“对比”“区别”“推荐”“为什么”“如果”等词的查询，大概率需要多跳推理。
• 用户反馈：如果用户对返回结果不满意，可以通过“换种方式问”或“继续追问”的交互，触发生成层。
• 业务规则：某些高价值场景（例如付费咨询、精准医疗、法律问答）可以直接指定进入 LLM 层，保证回答质量。

这套信号体系需要结合具体业务调优，但总体原则是清晰的：能用低成本解决的问题，绝不让高成本模型出手。

3.3 预算控制的三个开关

在架构上，我建议为每个请求都设计三个预算开关：

• 模型选择开关：不同复杂度的查询走不同的模型，简单分类用规则或小模型，复杂生成用强模型。
• 上下文长度开关：控制送入 LLM 的上下文长度。2400 token 能回答的问题，不必塞 8000 token。
• 工具调用开关：控制 Agent 可调用的工具数量。普通查询只调检索工具，复杂问题才允许调外部 API。

三个开关的本质是把“每次查询成本”从固定值变成变量，让预算能够根据查询的实际需要弹性分配。

4. 从零搭建一个轻量智能体检索系统

接下来我们用一个完整的 Python 示例，搭建一个“预算受限的智能体搜索”最小可运行系统。这个示例会把前面讲的架构理念落地，包含缓存、混合检索、RRF 融合、规则重排、LLM 路由这几部分。

演示环境说明：示例使用 Python 3.x，向量化部分使用 sentence-transformers 或任意 OpenAI 兼容的 Embedding API，倒排索引用 sklearn 的 TfidfVectorizer 演示，生产环境可以用 Elasticsearch 替换。LLM 调用部分以 OpenAI 兼容接口为例，实际使用时请替换为自己的模型服务地址和密钥。

4.1 项目结构

我们创建一个预算受限的智能体搜索演示项目，目录结构如下：

budget_agent_search/ ├── config.py # 全局配置与预算参数 ├── index_builder.py # 构建向量索引和关键词索引 ├── search.py # 混合检索与 RRF 融合 ├── rerank.py # 轻量重排，计算置信度 ├── llm_router.py # LLM 路由与调用 ├── main.py # 主流程 └── data/ └── faq_docs.jsonl # 示例文档

4.2 准备示例数据

先在data/faq_docs.jsonl中放一些演示文档。这里用云服务器相关的知识库内容做例子，方便大家理解检索效果。

{"id": 1, "title": "云服务器如何选择 CPU 核数", "content": "选择 CPU 核数时，需要根据业务类型判断。Web 应用和小型数据库通常 2 核 4G 即可；视频处理、科学计算建议 8 核以上。核数越高，并发处理能力越强，但成本也线性增加。"} {"id": 2, "title": "GPU 云服务器适合哪些场景", "content": "GPU 云服务器适合深度学习训练、图形渲染、视频转码和高性能计算等场景。建议根据显存需求选择 T4、A10、A100 等不同规格。"} {"id": 3, "title": "云服务器带宽怎么选", "content": "带宽选择取决于业务流量。个人网站选择 3-5Mbps 即可，视频直播或文件下载场景建议按峰值流量估算，必要时搭配 CDN 降低源站带宽压力。"} {"id": 4, "title": "云硬盘与本地盘的区别", "content": "云硬盘数据可靠性高，支持快照和随时扩容，适合数据库等需要持久化的场景。本地盘延迟更低，但数据可靠性依赖物理机，建议用于临时缓存和日志存储。"} {"id": 5, "title": "如何降低云服务器成本", "content": "降低云服务器成本可以从实例规格、付费方式、架构三方面入手：按量付费改为包年包月、使用竞价实例处理弹性负载、用 Serverless 承载低频任务，都可以显著降低账单。"} {"id": 6, "title": "DDoS 防护如何配置", "content": "DDoS 防护需要结合高防 IP、流量清洗和 CDN 加速能力。攻击流量较小时可以通过安全组限制来源 IP；大流量攻击场景建议接入专业高防服务。"}

4.3 配置层：预算参数集中管理

创建config.py，把所有与预算相关的参数集中管理。这样后续调优时只需要改一个文件，不需要在业务代码里到处找魔法数字。

# 文件路径：budget_agent_search/config.py class BudgetConfig: # 检索参数 TOP_K_KEYWORD = 10 # 关键词召回数量 TOP_K_VECTOR = 10 # 向量召回数量 TOP_K_FUSION = 5 # RRF 融合后保留数量 # 置信度阈值 # 当最高分与次高分差距大于 CONFIDENT_GAP 时，直接返回结果，不调用 LLM CONFIDENT_GAP = 0.15 MIN_CONFIDENT_SCORE = 0.40 # 最高分低于该值时，认为检索结果不可信 # LLM 路由参数 ENABLE_LLM = True # 是否启用 LLM 生成层 MAX_CONTEXT_TOKENS = 2048 # 发送给 LLM 的最大上下文 token MODEL_STRONG = "deepseek-chat" # 强模型，用于复杂生成 MODEL_LIGHT = "deepseek-lite" # 轻量模型，用于简单总结 # 缓存参数 CACHE_SIZE = 1000 # LRU 缓存容量 # Embedding 模型设置 EMBEDDING_MODEL = "text-embedding-v2" # 如果希望使用本地模型，可以改为： # EMBEDDING_MODEL = "local:BAAI/bge-small-zh-v1.5" # 伪文档集合，仅用于演示 DOC_PATH = "data/faq_docs.jsonl"

配置集中管理的价值在于：在系统上线后，你可以只修改阈值、模型名称、缓存大小来完成大部分预算策略调整，而不需要改代码逻辑。

4.4 构建索引：向量召回 + 关键词召回

创建index_builder.py，负责加载文档并构建两类索引。

这里的关键思路是“双向召回”：向量索引负责语义匹配，关键词索引负责精确匹配。两者互补，可以避免单纯向量检索导致的“明明有标准答案却召不回来”的尴尬。

# 文件路径：budget_agent_search/index_builder.py import json import numpy as np from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity class IndexBuilder: def __init__(self, doc_path, embedding_func=None): self.doc_path = doc_path self.embedding_func = embedding_func self.docs = [] self.tfidf_vectorizer = TfidfVectorizer(analyzer="char_wb", ngram_range=(1, 2)) self.tfidf_matrix = None self.doc_embeddings = None def load_docs(self): """从 JSONL 文件加载文档""" with open(self.doc_path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line: item = json.loads(line) # 把标题和正文拼接成一个可检索的文本 text = f"{item['title']}。{item['content']}" self.docs.append({ "id": item["id"], "title": item["title"], "content": item["content"], "text": text }) print(f"加载文档数：{len(self.docs)}") def build_keyword_index(self): """基于 TF-IDF 构建关键词索引（演示用，生产环境建议用 ES）""" corpus = [d["text"] for d in self.docs] self.tfidf_vectorizer.fit(corpus) self.tfidf_matrix = self.tfidf_vectorizer.transform(corpus) def build_vector_index(self): """调用 Embedding 接口构建向量索引""" if self.embedding_func is None: raise ValueError("请提供 embedding_func，例如调用 Embedding API 的函数") texts = [d["text"] for d in self.docs] self.doc_embeddings = self.embedding_func(texts) # 如果 embedding_func 返回的是 list，统一转成 numpy 数组方便计算 if isinstance(self.doc_embeddings, list): self.doc_embeddings = np.array(self.doc_embeddings) def keyword_search(self, query, top_k=10): """关键词检索，返回 (doc_index, score) 列表""" query_vec = self.tfidf_vectorizer.transform([query]) scores = cosine_similarity(query_vec, self.tfidf_matrix)[0] top_indices = scores.argsort()[::-1][:top_k] return [(int(i), float(scores[i])) for i in top_indices if scores[i] > 0] def vector_search(self, query_embedding, top_k=10): """向量检索，返回 (doc_index, score) 列表""" if self.doc_embeddings is None: return [] query_vec = np.array(query_embedding).reshape(1, -1) scores = cosine_similarity(query_vec, self.doc_embeddings)[0] top_indices = scores.argsort()[::-1][:top_k] return [(int(i), float(scores[i])) for i in top_indices]

需要提醒的是，TF-IDF 在中文场景里建议用char_wb的 bi-gram 甚至 tri-gram 方式处理，单纯按空格分词对中文不友好。生产环境如果使用 Elasticsearch，可以考虑搭配 IK 分词器，效果会更好。

4.5 混合检索与 RRF 融合

创建search.py，实现混合检索。这里使用 RRF（Reciprocal Rank Fusion，倒数排名融合）而不是简单的分数加权。

RRF 是一种不依赖分数量纲的融合方式。因为关键词检索的分数分布和向量检索的分数分布完全不同，直接相加会导致某一类结果总是占据优势。而 RRF 只看排名位置，公式为：

score(doc) = Σ 1 / (k + rank(doc))

k 通常取 60 左右，作用是避免排名第一的文档拿到过高的权重。

# 文件路径：budget_agent_search/search.py class HybridSearcher: def __init__(self, index_builder): self.index_builder = index_builder @staticmethod def rrf_fusion(keyword_results, vector_results, k=60, top_k=5): """RRF 融合两组检索结果""" fusion_scores = {} for idx, score in keyword_results: fusion_scores[idx] = fusion_scores.get(idx, 0) + 1.0 / (k + 1) for idx, score in vector_results: fusion_scores[idx] = fusion_scores.get(idx, 0) + 1.0 / (k + 1) sorted_items = sorted(fusion_scores.items(), key=lambda x: x[1], reverse=True) return sorted_items[:top_k] def search(self, query, query_embedding=None): """混合检索入口""" kw_results = self.index_builder.keyword_search(query, top_k=10) vec_results = [] if query_embedding is not None: vec_results = self.index_builder.vector_search(query_embedding, top_k=10) fused = self.rrf_fusion(kw_results, vec_results, top_k=5) # 保留原始 doc 信息 docs = self.index_builder.docs results = [] for idx, fusion_score in fused: results.append({ "doc_id": docs[idx]["id"], "title": docs[idx]["title"], "content": docs[idx]["content"], "fusion_score": fusion_score, "keyword_score": dict(keyword_results)[idx] if False else None, }) return results

4.6 轻量重排与置信度判断

重排环节我们不做复杂模型，而是用一个轻量策略：在前 5 个候选文档中，计算检索分数差距，并利用一个简单的规则模型判断“当前结果是否足够可信”。

置信度判断是预算控制的关键，它决定了查询是否需要“升级”到 LLM 层。我们这里采用两个信号：最高融合分数、第一名与第二名之间的分数差。如果第一名明显领先，说明答案很明确；如果前两名分数接近，说明存在多种可能，需要 LLM 介入。必要时，你也可以接入一个 cross-encoder 模型用 query 与 doc 的相关性分数替换这里的规则分数。

# 文件路径：budget_agent_search/rerank.py class LightReranker: """轻量重排：基于规则和阈值判断是否需要 LLM 兜底""" def __init__(self, min_score=0.4, gap=0.15): self.min_score = min_score self.gap = gap def is_confident(self, fused_results): """ 根据融合分数判断是否可以直接返回结果。 返回 True 表示检索结果足够可信，不需要调用 LLM。 """ if not fused_results: return False top_score = fused_results[0]["fusion_score"] if top_score < self.min_score: return False if len(fused_results) >= 2: second_score = fused_results[1]["fusion_score"] if top_score - second_score < self.gap: return False return True def get_top_result(self, fused_results): """直接返回最相关的文档片段""" if not fused_results: return None top = fused_results[0] return { "title": top["title"], "snippet": top["content"][:120], "doc_id": top["doc_id"], }

4.7 LLM 路由与预算控制

LLM 路由是整个架构中最体现“预算意识”的模块。它做的事情是：对需要 LLM 介入的查询，进一步判断用强模型还是轻模型，并控制送入模型的上下文长度。

这里要区分两种场景：一种是简单总结，直接从 Top 文档中截取片段生成答案；另一种是复杂推理，需要把多个文档片段拼接后让模型对比分析。场景不同，模型选择和上下文长度都应该不同。

# 文件路径：budget_agent_search/llm_router.py import json import hashlib from functools import lru_cache class LLMRouter: def __init__(self, config, llm_call_func): self.config = config self.llm_call_func = llm_call_func # 一个函数，接受 (system_prompt, user_prompt, model) 参数 @lru_cache(maxsize=1000) def cached_answer(self, query_hash): """基于查询哈希的缓存，后续相同或相似问题直接命中""" pass def is_complex_query(self, query): """判断查询是否属于复杂类型，需要强模型处理""" complex_markers = ["对比", "区别", "推荐", "为什么", "如果", "方案", "分析"] for marker in complex_markers: if marker in query: return True return False def route(self, query, candidates): """ 对候选文档进行 LLM 生成。 如果查询复杂，使用强模型；否则使用轻量模型。 """ if not candidates: return "未找到相关答案，请尝试换一种描述方式。" # 截取上下文，控制 token 开销 context_text = "\n\n".join( [f"[{idx+1}] {c['title']}：{c['content'][:200]}" for idx, c in enumerate(candidates)] ) # 这里只是演示截断，实际场景可以使用 tiktoken 精确计算 context_text = context_text[:self.config.MAX_CONTEXT_TOKENS] if self.is_complex_query(query): model = self.config.MODEL_STRONG system_prompt = "你是一个专业的搜索助手。请根据提供的资料，结合用户问题，给出准确、结构化的回答。如果资料不足，请明确说明。" else: model = self.config.MODEL_LIGHT system_prompt = "你是一个简洁的搜索助手。请从候选资料中找出最相关的信息，用一两句话回答用户问题。" user_prompt = f"用户问题：{query}\n\n候选资料：\n{context_text}\n\n请回答：" # 调用实际的 LLM 接口 # 注意：这里的 llm_call_func 需要你在 main.py 中实现 # 推荐使用 OpenAI 兼容接口，例如： # response = client.chat.completions.create( # model=model, # messages=[{"role": "system", "content": system_prompt}, # {"role": "user", "content": user_prompt}], # temperature=0.3 # ) answer = self.llm_call_func(system_prompt, user_prompt, model) return answer

这里把复杂查询判断做成规则版，是为了让大家先跑通流程。生产环境中，更推荐用一个小的分类模型，或者直接让 Agent 的规划模块来自动判断。规则版的好处是零成本、可解释、方便快速上线。

4.8 主流程串联

最后创建main.py，把上面的模块串起来，模拟一个带预算控制的查询入口。

# 文件路径：budget_agent_search/main.py import hashlib from functools import lru_cache from config import BudgetConfig from index_builder import IndexBuilder from search import HybridSearcher from rerank import LightReranker from llm_router import LLMRouter # 演示用的 embedding 函数 # 生产环境请替换为真实的 Embedding API 或本地模型 def dummy_embedding(texts): """使用简单的哈希特征模拟 embedding 向量（仅演示用）""" import numpy as np vectors = [] for text in texts: # 这里用一个固定维度的伪向量，实际项目要用真实 embedding 模型 vec = np.zeros(128) for i, ch in enumerate(text): vec[hash(ch) % 128] += 1 # 归一化 norm = np.linalg.norm(vec) vec = vec / norm if norm > 0 else vec vectors.append(vec) return np.array(vectors) # 演示用的 LLM 调用函数 # 生产环境请替换为真实的模型调用代码 def dummy_llm_call(system_prompt, user_prompt, model): """模拟 LLM 返回结果，实际使用请替换为模型 API 调用""" # 生产环境示例： # from openai import OpenAI # client = OpenAI(api_key="your-key", base_url="https://your-endpoint") # resp = client.chat.completions.create( # model=model, # messages=[{"role": "system", "content": system_prompt}, # {"role": "user", "content": user_prompt}] # ) # return resp.choices[0].message.content # 演示时从 user_prompt 中提取候选标题，返回模拟答案 titles = [] for line in user_prompt.split("\n"): line = line.strip() if line.startswith("[") and "：" in line: titles.append(line.split("：", 1)[1][:20]) return f"根据候选资料，推荐你重点参考：{ '、'.join(titles[:2]) }。建议结合自身业务场景进一步验证。" class BudgetAgentSearch: def __init__(self, config): self.config = config self.index_builder = IndexBuilder(config.DOC_PATH, embedding_func=dummy_embedding) self.index_builder.load_docs() self.index_builder.build_keyword_index() self.index_builder.build_vector_index() self.searcher = HybridSearcher(self.index_builder) self.reranker = LightReranker(min_score=config.MIN_CONFIDENT_SCORE, gap=config.CONFIDENT_GAP) self.llm_router = LLMRouter(config, llm_call_func=dummy_llm_call) @lru_cache(maxsize=1000) def search(self, query_embedding_tuple, query): """带缓存的查询入口""" query_embedding = list(query_embedding_tuple) # 1. 混合检索 candidates = self.searcher.search(query, query_embedding=query_embedding) # 2. 置信度判断 if self.reranker.is_confident(candidates): top = self.reranker.get_top_result(candidates) return { "source": "retrieval", "answer": top["snippet"], "title": top["title"], "cost_level": "low", } # 3. 如果不自信或需要 LLM，判断是否启用 LLM 层 if self.config.ENABLE_LLM: answer = self.llm_router.route(query, candidates) return { "source": "llm", "answer": answer, "cost_level": "high" if self.llm_router.is_complex_query(query) else "medium", } # 4. 未启用 LLM 时，直接返回 top1 top = self.reranker.get_top_result(candidates) if top: return {"source": "retrieval", "answer": top["snippet"], "cost_level": "low"} return {"source": "empty", "answer": "未找到相关答案。", "cost_level": "zero"} def handle_query(self, query): """对外查询接口""" # 生成查询的 embedding 向量（演示函数） query_embedding = dummy_embedding([query])[0] # lru_cache 要求参数可哈希，所以这里传 tuple return self.search(tuple(query_embedding), query) if __name__ == "__main__": config = BudgetConfig() agent = BudgetAgentSearch(config) test_queries = [ "云服务器怎么选 CPU 核数", "GPU 云服务器适合什么场景", "对比一下云硬盘和本地盘", "如何降低服务器成本", "今天天气怎么样", # 预期：找不到相关内容 ] for q in test_queries: print("=" * 50) print(f"问题：{q}") result = agent.handle_query(q) print(f"来源：{result['source']}, 成本级别：{result['cost_level']}") print(f"回答：{result['answer']}")

4.9 运行与预期结果

在项目目录下执行：

cd budget_agent_search python main.py

预期输出如下：

================================================== 问题：云服务器怎么选 CPU 核数 来源：retrieval, 成本级别：low 回答：选择 CPU 核数时，需要根据业务类型判断。Web 应用和小型数据库通常 2 核 4G 即可；视频处理、科学计算建议 8 核以上。核数越高，并发处理能力越强，但成本也线性增加。 ================================================== 问题：GPU 云服务器适合什么场景 来源：retrieval, 成本级别：low 回答：GPU 云服务器适合深度学习训练、图形渲染、视频转码和高性能计算等场景。建议根据显存需求选择 T4、A10、A100 等不同规格。 ================================================== 问题：对比一下云硬盘和本地盘 来源：llm, 成本级别：high 回答：根据候选资料，推荐你重点参考：云硬盘与本地盘的区别、云服务器带宽怎么选。建议结合自身业务场景进一步验证。 ================================================== 问题：如何降低服务器成本 来源：retrieval, 成本级别：low 回答：降低云服务器成本可以从实例规格、付费方式、架构三方面入手：按量付费改为包年包月、使用竞价实例处理弹性负载、用 Serverless 承载低频任务，都可以显著降低账单。 ================================================== 问题：今天天气怎么样 来源：empty, 成本级别：zero 回答：未找到相关答案。 ==================================================

从输出可以看到：

• 简单问题时，系统直接走检索层，返回文档片段，成本为 low。
• 对比类问题时，系统识别为复杂查询，走 LLM 层，并标记为 high 成本。
• 完全无关的问题时，系统直接返回空结果，没有浪费任何预算。

这就是预算控制的效果：每一分钱都花在“需要花”的地方。

5. 效果评估与成本核算

系统上线前，一定要建立一套效果和成本的双重评估体系。只看效果不看成本，智能体搜索很难持续跑下去；只看成本不看效果，业务方也不会满意。

5.1 离线评估指标

建议准备 200-500 条带标准答案的测试集，至少包含以下指标：

其中“直接命中率”和“LLM 调用率”是预算控制的核心指标。如果你发现 LLM 调用率超过 60%，说明置信度阈值设置得太严格，预算会快速耗尽；如果低于 10%，则要检查是不是检索层返回了大量低质量答案。

5.2 在线成本测算

上线后，建议每天做一次成本盘点：

当日总成本 = 检索层成本 + Embedding 成本 + LLM 生成成本 检索层成本 = 单次检索成本 × 总查询数 Embedding 成本 = 单次 Embedding 成本 × 总查询数 LLM 生成成本 = 单次平均成本 × 触发 LLM 的查询数

通过拆解可以快速定位成本增长点。如果 LLM 调用量没涨但成本涨了，就要检查上下文 token 是不是超了；如果调用量本身涨了，就要检查置信度阈值和模型路由规则。

5.3 预算控制回路的调优顺序

当预算超支时，我建议按以下顺序调整，而不是直接降低模型质量：

1. 优先扩大缓存：很多用户问题都是重复的，缓存命中率每提升 10%，总成本能降不少。
2. 调整置信度阈值：把CONFIDENT_GAP从 0.15 提升到 0.2，MIN_CONFIDENT_SCORE从 0.4 提升到 0.5，能明显降低 LLM 调用率。
3. 优化召回策略：如果检索结果不够精准导致 LLM 频繁介入，优先优化召回，解决“源头”问题。
4. 控制上下文长度：在效果允许范围内，把送入 LLM 的上下文从 2048 降到 1500，多轮累积起来能省不少。
5. 最后才考虑换模型：如果上述手段都用尽，才考虑用更便宜的模型，但一定要评估效果回退幅度。

这个顺序的核心逻辑是：优先用工程手段省没有感知的钱，而不是直接牺牲回答质量。

6. 常见问题与排查思路

在实际落地过程中，下面这些问题出现的频率最高。这里整理成一张排查表，你可以直接对照处理。

这里单独说一下上下文 token 的问题。很多人以为控制字符串长度就可以控制 token，但中文的一个字大约占 1-2 个 token，英文一个单词约 1-2 个 token，直接按字符截断很容易超限或浪费。生产环境建议使用模型对应的 tokenizer 来精确截断。

另外，缓存设计也要注意不要过度缓存。知识库内容会更新，如果文档做了修改，旧的缓存答案会过时。建议在缓存 key 中加入知识库版本号，知识库更新时自动失效。

7. 工程落地最佳实践

前面把系统搭建和调优讲了一遍，下面补充一些工程层面的最佳实践。这些经验来自实际项目踩坑，价值不一定比代码低。

7.1 接口层设计更加重要

很多团队在做智能体搜索时，把大量精力放在模型和算法上，忽略了接口层的设计。实际上，接口层是预算控制的第一道防线。建议在设计 API 时增加以下参数：

• max_cost_level：调用方可以指定本次查询允许的最高成本级别。例如普通问答场景只允许 low，重要用户允许 high。
• force_llm：某些业务场景强制走 LLM 生成，保证回答质量。
• callback_url：复杂搜索采用异步回调，避免长耗时阻塞同步请求。

这样做的好处是，预算控制从“系统内部策略”变成了“产品可配置能力”，产品和运营也能参与调优。

7.2 日志与观测是预算控制的眼睛

搜索系统有一个特点：问题种类多、失败模式隐蔽。如果不记录详细的处理链路日志，很难定位预算消耗在哪个环节。

建议每条查询至少记录以下字段：

query_id, query, 缓存是否命中, 检索层耗时, 检索分数top5, 是否触发LLM, 模型名称, 输入token, 输出token, 最终来源, 耗时

有了这些日志，你可以随时分析：哪些查询在持续触发高成本模型？哪些文档被反复引用但用户不满意？哪些查询值得做定向优化？

成本监控可以设置告警阈值，比如当日 LLM 调用率超过 50% 或单日成本超过预估值的 130% 时触发告警，及时介入。

7.3 安全与越权问题

搜索系统很容易忽略安全问题，尤其是接入了 LLM 后，风险面会变大。以下几点在工程化时一定要考虑：

• Embedding 和 LLM 调用的密钥必须存放在服务端环境变量或密钥管理系统中，绝不能硬编码在前端代码里。
• 对用户输入的 query 做长度限制和敏感词过滤，避免恶意构造超长输入打爆 token 预算。
• 如果检索的数据涉及权限，需要在召回阶段就做数据权限过滤，不能把所有文档都送进 LLM 上下文再从答案里“挑出”越权内容。
• 对 LLM 的输出可以加一轮合规检测，防止生成不安全、不合适的回答。

7.4 渐进式灰度发布

预算受限的智能体搜索系统，不建议一次全量切流量。推荐按以下阶段灰度：

• 第一阶段：切 10% 流量，同时保留旧的搜索逻辑，对比两边的点击率、满意度、成本。
• 第二阶段：根据成本数据调整置信度阈值，扩大到 50% 流量。
• 第三阶段：确认效果稳定后，全量切换。

灰度期间重点观察两个指标：成本增长率是否在预期内，用户对“直接答案”的满意度是否高于旧的链接列表。这里可以加一个简单的“满意/不满意”按钮，用真实用户反馈辅助决策。

7.5 数据更新与索引维护

知识库数据不会一成不变。工程技术上要建立一套索引更新机制。最简单的方式是每天凌晨全量重建索引；数据量大时可以采用增量更新。增量更新的关键在于文档版本管理，要保证向量库、倒排索引、缓存三者的数据是一致的。否则可能会出现“文档删了，向量还在；缓存还在返回旧答案”的问题。

一个实用的做法是给每篇文档维护updated_at字段，增量任务只处理最近变更的文档，同时将涉及变更文档的缓存 key 失效。

8. 总结与下一步学习路线

这篇内容围绕“预算受限下的智能体搜索”展开，核心思路可以提炼成一句话：不要把每个查询都当成复杂任务处理，而是让系统学会识别问题的难度，把有限的预算精准地分配到最值得花的环节。围绕这个思路，我们从成本拆解、分层架构、混合召回、RRF 融合、置信度判断、LLM 路由、成本调优到工程落地方案，完整走了一遍。

如果你准备在自己的项目中落地这套方案，建议下一步按这个顺序深入：

• 先看一下当前搜索链路最贵的是哪个环节，用日志把成本拆出来。
• 实现一个最简单的“检索 + 置信度判断 + LLM 兜底”的三层结构，先把直接命中率跑出来。
• 然后逐步加入缓存、查询改写、向量检索、rerank、分级模型。
• 最后再做灰度发布和效果评测。

模型和技术选型只是其中一环，真正决定智能体搜索能不能长期跑下去的，是你对预算的感知能力和对检索链路的精细控制能力。从一个小规模的垂直知识库开始，把链路跑通，把数据积累起来，再逐步扩展，可能是预算受限场景下最稳妥的路径。如果这篇文章对你有一点帮助，可以先收藏备用，动手搭一套最小系统，跑出自己的第一份成本数据。