目录
- 前言
- 高阶环境依赖补充
- 模块一:LLM 网关统一管理大模型调用
- 模块二:精细化用户行为分析与知识库迭代
- 模块三:知识图谱增强 RAG 落地
- 高阶落地踩坑复盘
- 高阶阶段面试考点自检清单
- 写在最后
前言
业务方要求同时接入通义和 DeepSeek 做容灾,每家接口格式、参数命名都不一样,改一次适配要动好几处代码;运营天天来问用户都在搜什么、哪些问题始终答不上来,光靠好评点踩两个指标根本说不清楚优化方向;还有员工问流程审批、组织关系这类问题,纯向量检索总是差口气,明明文档里有答案,就是散在不同片段里串不起来。
索性把这三块一起落地了:搭一个统一的 LLM 网关管所有大模型调用,不用再挨个适配接口;做细用户行为数据分析,反过来指导知识库迭代;用轻量知识图谱补向量检索的短板,解决关系类、流程类问题答不准的问题。
2. 高阶环境依赖补充
不用引入太重的中间件,小团队也能直接落地。LLM 网关自己封装轻量版本,不用上开源的重型网关;行为分析用 SQLite 存日志,Python 脚本做统计;知识图谱用 NetworkX 做轻量存储,不用单独部署图数据库。
pip install networkx jieba需要额外准备:
- 至少两家大模型的 API 密钥(用于网关容灾测试)
- 结构化的制度 / 流程类文档(用于抽取知识图谱三元组)
3. 模块一:LLM 网关统一管理大模型调用
之前的代码里大模型是直接初始化死的,接的模型多了之后特别乱:每家 SDK 不一样、错误码不一样、计费方式不一样,出了问题不好排查,也没法做容灾降级。所以我封装了一层轻量 LLM 网关,把所有大模型调用统一收口。
核心做五件事:统一接口适配,不管底层接哪家模型,对外调用方式都一样;失败自动降级重试,主模型挂了自动切备用模型;多模型智能路由,简单问题用便宜的模型,复杂问题用好模型;统一 Token 计费,所有模型的消耗统一统计;全链路日志,每次调用的模型、耗时、Token 量、错误信息都记下来。
核心代码实现
import time from typing import Optional from langchain_community.chat_models import ChatTongyi from langchain_community.chat_models import ChatOpenAI from langchain.schema import BaseMessage class LLMGateway: def __init__(self): # 注册所有可用模型,配置优先级和角色 self.models = { "qwen-plus": { "model": ChatTongyi(model="qwen-plus", api_key=DASHSCOPE_API_KEY, temperature=0.1), "priority": 1, # 主模型,优先级最高 "cost_per_1k_input": 0.008, "cost_per_1k_output": 0.02 }, "qwen-turbo": { "model": ChatTongyi(model="qwen-turbo", api_key=DASHSCOPE_API_KEY, temperature=0.1), "priority": 2, # 轻量模型,简单问题用 "cost_per_1k_input": 0.002, "cost_per_1k_output": 0.006 }, "deepseek-chat": { "model": ChatOpenAI( model="deepseek-chat", api_key=DEEPSEEK_API_KEY, base_url="https://api.deepseek.com/v1", temperature=0.1 ), "priority": 3, # 备用容灾模型 "cost_per_1k_input": 0.001, "cost_per_1k_output": 0.002 } } self.failed_models = set() # 故障模型集合,熔断用 self.call_logs = [] # 调用日志,可异步落库 def _get_available_model(self, prefer_level: str = "standard") -> str: """根据场景选择可用模型,简单场景优先便宜的,故障时自动降级""" # 简单场景优先用轻量模型 if prefer_level == "fast": order = ["qwen-turbo", "qwen-plus", "deepseek-chat"] # 标准场景优先用主力模型 else: order = ["qwen-plus", "deepseek-chat", "qwen-turbo"] for model_name in order: if model_name not in self.failed_models: return model_name raise Exception("所有模型均不可用") def invoke(self, messages: list[BaseMessage], prefer_level: str = "standard") -> dict: """ 统一调用入口,自动重试、降级、计费、日志 返回:{content: 答案, model: 实际用的模型, token_usage: token消耗, cost: 费用, latency: 耗时} """ start_time = time.time() model_name = self._get_available_model(prefer_level) max_retry = 2 last_error = None for i in range(max_retry): try: llm = self.models[model_name]["model"] response = llm.invoke(messages) # 统计token和费用 input_tokens = response.usage_metadata["input_tokens"] output_tokens = response.usage_metadata["output_tokens"] cost = (input_tokens / 1000 * self.models[model_name]["cost_per_1k_input"] + output_tokens / 1000 * self.models[model_name]["cost_per_1k_output"]) latency = time.time() - start_time result = { "content": response.content, "model": model_name, "input_tokens": input_tokens, "output_tokens": output_tokens, "cost": round(cost, 6), "latency": round(latency, 3) } # 异步记录日志 self.call_logs.append(result) # 调用成功,把模型从故障集里移除 self.failed_models.discard(model_name) return result except Exception as e: last_error = e # 连续失败就熔断这个模型,切下一个 if i == max_retry - 1: self.failed_models.add(model_name) # 还有可用模型就降级重试 remaining = [m for m in self.models if m not in self.failed_models] if remaining: model_name = remaining[0] continue raise last_error # 全局网关实例 llm_gateway = LLMGateway()用了网关之后,上层业务代码不用关心底层接的是哪家模型,统一调用网关就行;主模型出故障的时候自动切备用,用户几乎感知不到。后面再加新模型,只要在网关注册一下就行,不用改业务代码。
对应面试考点(附答题思路)
为什么要做 LLM 网关?它解决了什么问题?答:核心是解决多模型场景下的管理混乱、容灾能力弱、成本不可控的问题,把大模型调用从零散的直连模式,变成统一管控的网关模式。 具体解决四类问题:第一是接口统一,屏蔽不同厂商模型的接口差异,业务层不用挨个适配,加新模型不用改业务代码;第二是容灾降级,单家模型故障时自动切换备用模型,保障服务可用;第三是成本管控,统一统计所有模型的 Token 消耗和费用,做智能路由降本,统一限流防超额;第四是可观测性,所有调用统一打日志、埋点,方便排查问题和统计分析。 用户量小、只用一家模型的时候感受不明显,一旦接入多模型、对可用性有要求,网关是必须的。
LLM 网关的核心功能有哪些?答:核心能力可以分成五层,从基础到高级: 最基础的是统一接入层,适配不同厂商的 API,对外提供统一的调用接口,屏蔽参数、格式、鉴权的差异。 第二层是流量管理层,包括限流、熔断、降级、负载均衡,控制调用量,防止打崩下游,故障时自动切备用。 第三层是智能路由层,比如按问题复杂度选不同模型、按成本选最便宜的可用模型、按延迟选最快的模型,平衡效果和成本。 第四层是成本与审计层,统一统计 Token 消耗、核算费用、按租户 / 部门分摊成本,所有调用留痕可审计。 第五层是增强服务层,比如语义缓存、Prompt 统一管理、内容安全审核,在网关层统一做,不用每个业务都做一遍。 落地不用一开始就做全,先从统一接入和容灾开始,再逐步加路由、成本管控。
多模型容灾降级怎么设计?怎么防止降级把备用模型打挂?答:核心是分级降级 + 熔断保护,不能一故障就全量切过去,很容易把备用模型也打挂,造成雪崩。 首先是故障检测:连续多次调用失败、超时率超过阈值,就判定模型故障,触发熔断,暂时不往这个模型发流量。 然后是降级策略:分梯度降级,先降级非核心场景的流量,比如简单问答先切轻量模型,核心复杂问答再切备用大模型,不要一下全量切过去。 还要有限流保护:备用模型有承载上限,降级的时候控制流量速率,不要超过备用模型的 QPS 限制,防止打挂。 最后是恢复机制:熔断后每隔一段时间探活一次,发少量请求试试,模型恢复了就逐步切回流量,不是一下子全量恢复。 我们之前就踩过全量降级的坑,加了熔断和限流之后就稳了。
智能路由怎么判断问题该用大模型还是小模型?答:常用的有三种方式,从简单到精准,按需选就行。 第一种是规则路由,最简单也最常用,比如按问题长度、关键词、场景来判断,比如简单查询、短问题用小模型,复杂推理、长上下文用大模型,实现简单,性能好,大部分场景够用。 第二种是分类模型路由,训练一个小的分类模型,专门判断问题复杂度,决定用什么模型,比规则准一点,成本也不高。 第三种是大小模型级联,先让小模型回答,回答质量不够再切大模型重答,效果最好,但耗时和成本会高一点。 我们目前用的是规则路由,覆盖了 80% 的场景,足够用;后续可以升级分类模型,进一步提升准确率。
4. 模块二:精细化用户行为分析与知识库迭代
之前只有好评点踩两个反馈指标,太粗了,运营想优化知识库都不知道从哪下手。比如不知道用户最常问什么、哪些问题从来答不上来、用户问完一次就走还是会追问。
所以我补了一套用户行为分析体系,从会话、问题、用户三个维度做统计,定期输出分析报告,反过来指导知识库优化:高频问题重点优化答案,高频无答案问题补充文档,用户总追问的问题优化分块和检索。
核心统计三个维度:
- 问题维度:热门问题 TOP 榜、未命中问题榜、问题分类分布
- 会话维度:平均对话轮次、解决率、用户追问率
- 用户维度:活跃用户、部门使用分布、人均提问量
核心代码实现
import sqlite3 from datetime import datetime, timedelta from collections import Counter class UserBehaviorAnalyzer: def __init__(self, db_path: str = "./behavior.db"): self.db_path = db_path self._init_db() def _init_db(self): conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 会话表 cursor.execute(''' CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, user_id TEXT, tenant_id TEXT, start_time TEXT, end_time TEXT, turn_count INTEGER, is_solved INTEGER, -- 1解决 0未解决 feedback INTEGER ) ''') # 提问明细表 cursor.execute(''' CREATE TABLE IF NOT EXISTS queries ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, user_query TEXT, answer TEXT, retrieval_hit INTEGER, -- 检索是否命中 latency REAL, create_time TEXT ) ''') conn.commit() conn.close() def record_query(self, session_id: str, user_id: str, tenant_id: str, user_query: str, answer: str, retrieval_hit: bool, latency: float): """记录单条提问""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute( "INSERT INTO queries VALUES (NULL, ?, ?, ?, ?, ?, ?)", (session_id, user_query, answer, 1 if retrieval_hit else 0, latency, datetime.now().strftime("%Y-%m-%d %H:%M:%S")) ) conn.commit() conn.close() def get_daily_report(self, date: str = None) -> dict: """生成日报:热门问题、未命中问题、会话统计""" if not date: date = datetime.now().strftime("%Y-%m-%d") conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 1. 热门问题TOP10 cursor.execute(''' SELECT user_query, COUNT(*) as cnt FROM queries WHERE date(create_time) = ? GROUP BY user_query ORDER BY cnt DESC LIMIT 10 ''', (date,)) top_queries = cursor.fetchall() # 2. 检索未命中问题TOP10 cursor.execute(''' SELECT user_query, COUNT(*) as cnt FROM queries WHERE date(create_time) = ? AND retrieval_hit = 0 GROUP BY user_query ORDER BY cnt DESC LIMIT 10 ''', (date,)) miss_queries = cursor.fetchall() # 3. 会话整体数据 cursor.execute(''' SELECT COUNT(DISTINCT session_id), AVG(turn_count), AVG(is_solved) FROM sessions WHERE date(start_time) = ? ''', (date,)) session_stats = cursor.fetchone() conn.close() return { "date": date, "total_sessions": session_stats[0], "avg_turns": round(session_stats[1], 2), "solve_rate": round(session_stats[2], 4), "top_queries": top_queries, "miss_queries": miss_queries }有了数据之后,优化就有方向了:高频未命中的问题,优先补充对应文档;用户总追问的问题,检查是不是分块太碎、上下文不全;某个部门用得少,就去调研是不是内容不符合需求。比之前瞎调参数高效太多。
对应面试考点(附答题思路)
为什么要做用户行为分析?对 RAG 系统有什么价值?答:核心价值是把 “凭感觉优化” 变成 “数据驱动优化”,让系统迭代有方向。 具体有三个层面的价值:第一是发现知识库盲区,通过未命中问题、低好评问题,知道哪些内容是用户需要但知识库没有的,指导内容运营补全文档;第二是优化系统效果,通过分析用户追问、多轮对话,知道哪些场景检索不准、生成不好,针对性优化分块、检索、Prompt;第三是衡量业务价值,看使用量、解决率、活跃用户,知道系统到底有没有帮到用户,有没有减少工单、提升效率。 只做功能不做数据分析,系统就会停滞不前,永远不知道哪里该优化。
用户行为分析核心看哪些维度?答:我一般分成四个维度,从宏观到微观逐层看。 第一个是大盘指标,比如日活用户、总提问量、人均提问数、整体解决率、好评率,看系统整体的使用情况和健康度。 第二个是问题维度,热门问题 TOP 榜、未命中问题榜、问题分类分布,知道用户最关心什么、什么内容知识库没有,是内容优化的核心依据。 第三个是会话维度,平均对话轮次、追问率、单轮解决率,看用户是不是一次就能得到答案,还是要反复问,反映回答的完整性和准确性。 第四个是用户维度,部门 / 租户使用分布、用户活跃度分层,知道哪些部门用得多、哪些用得少,方便针对性运营。 不用一开始就做很全,先把热门问题和未命中问题跑起来,就能解决 80% 的优化方向问题。
怎么用行为数据来优化 RAG 效果?答:不同的数据对应不同的优化手段,精准施策,不是一股脑全调。 如果是高频未命中问题:优先补充对应的知识库文档,填补内容盲区;如果文档里有但搜不到,就优化分块和检索参数,或者把这类问题加入 query 改写的示例。 如果是用户频繁追问:大概率是答案不完整、太笼统,就优化分块大小,增加上下文长度,或者调整 Prompt 要求回答更详细;如果是用户指代不清,就优化 query 改写和上下文补全。 如果是好评率低的问题:先看是检索不准还是生成幻觉,检索不准就调召回和 Rerank,生成幻觉就加强 Prompt 约束、调低 temperature。 如果是某类问题集中反馈不好:可以针对性补充微调数据,或者做专项的 Prompt 优化。 核心就是数据定位问题,再针对性解决,比盲目调参效率高太多。
怎么区分 “知识库没有答案” 和 “有答案但搜不到”?答:这两种情况的优化方向完全不一样,必须区分开,处理方式也不同。 最简单的方法是人工抽检,从未命中问题里抽一部分,人工去文档里找有没有答案,就能算出比例,大概知道是内容缺失多还是检索效果差。 技术上可以辅助判断:看检索返回的最高相似度分数,如果分数特别低,大概率是真没有;如果分数不低但内容不相关,就是检索效果有问题。还可以用大模型做判断,把检索结果和问题一起给大模型,判断能不能回答问题。 如果是 “没有答案”,就补文档、加知识库内容,属于运营侧的优化;如果是 “有答案搜不到”,就优化分块、Embedding、检索、Rerank,属于技术侧的优化。
5. 模块三:知识图谱增强 RAG 落地
纯向量 RAG 用久了会发现一个明显的短板:处理关系类、流程类、层级类问题特别吃力。比如 “员工入职的审批流程有哪几个节点”“销售部的直属上级部门是什么”,答案散在好几个文档片段里,向量检索只能捞到零散的片段,串不起来完整的逻辑,大模型很容易答漏或者答错。
所以我做了一个轻量的知识图谱来补这个短板:从结构化的制度、流程、组织文档里,抽取「实体 - 关系 - 实体」三元组,构建知识图谱;用户提问时,先从问题里抽取实体和关系,去图谱里查结构化的结论,再和向量检索的文档片段拼在一起,一起喂给大模型。结构化信息保证逻辑准确,非结构化文档补充细节,两者互补。
没有上重型的图数据库,用 NetworkX 做内存图谱,中小规模的知识完全够用,不用额外部署服务。
核心代码实现
import networkx as nx import jieba from typing import List, Tuple class KnowledgeGraphRAG: def __init__(self): self.graph = nx.DiGraph() # 有向图,存储实体和关系 def add_triples(self, triples: List[Tuple[str, str, str]]): """批量添加三元组:(实体1, 关系, 实体2)""" for head, relation, tail in triples: self.graph.add_edge(head, tail, relation=relation) print(f"知识图谱更新完成,当前共{self.graph.number_of_nodes()}个实体,{self.graph.number_of_edges()}条关系") def extract_entities_from_query(self, query: str) -> List[str]: """从用户问题中提取实体,简单用关键词匹配,生产可用NER模型""" # 这里简化实现,实际可以用领域NER模型抽取 all_entities = list(self.graph.nodes) matched = [e for e in all_entities if e in query] return matched def query_graph(self, query: str) -> str: """查询知识图谱,返回结构化的关系结论""" entities = self.extract_entities_from_query(query) if not entities: return "" results = [] for entity in entities: if entity not in self.graph: continue # 查找该实体的所有出边和入边 for neighbor in self.graph.neighbors(entity): rel = self.graph[entity][neighbor]["relation"] results.append(f"{entity} {rel} {neighbor}") for predecessor in self.graph.predecessors(entity): rel = self.graph[predecessor][entity]["relation"] results.append(f"{predecessor} {rel} {entity}") if not results: return "" return "知识图谱结构化信息:\n" + "\n".join(results) def enhanced_retrieve(self, base_retriever, query: str) -> str: """融合检索:知识图谱结果 + 向量检索结果""" # 1. 查知识图谱,拿结构化信息 graph_result = self.query_graph(query) # 2. 查向量库,拿非结构化文档 docs = base_retriever.get_relevant_documents(query) doc_result = "文档片段:\n" + "\n\n".join([doc.page_content for doc in docs]) # 3. 合并返回,一起给大模型 return graph_result + "\n\n" + doc_result # 示例:初始化制度流程知识图谱 kg_rag = KnowledgeGraphRAG() # 从文档中抽取的三元组,实际可通过大模型批量抽取 process_triples = [ ("员工入职", "审批节点", "部门负责人"), ("员工入职", "审批节点", "HR"), ("员工入职", "所需材料", "身份证"), ("销售部", "直属上级", "营销中心"), ("年假", "申请条件", "入职满一年") ] kg_rag.add_triples(process_triples)实测下来,流程、组织关系类的问题,加了知识图谱之后,回答的准确率和完整度提升特别明显,不会再漏节点、说错关系。而且图谱是结构化的,可信度比纯文本高,也能减少大模型的幻觉。
对应面试考点(附答题思路)
知识图谱增强 RAG 解决了什么痛点?答:主要解决纯向量 RAG 的三个短板:关系推理弱、结构化信息不准、碎片化严重。 第一是关系类问题答不准,比如层级、流程、隶属关系,纯向量检索只能捞到零散片段,串不起来完整的关系链,知识图谱能直接给出准确的实体关系,保证逻辑正确。 第二是结构化信息容易遗漏,比如流程的多个节点、公式的多个参数,散在不同文档块里,向量检索可能只捞到一部分,知识图谱能完整返回所有关联信息。 第三是减少幻觉,知识图谱里的三元组是结构化的事实,可信度高,大模型基于图谱信息回答,比纯靠自由文本生成更不容易瞎编。 它不是替代向量检索,是补充,两者结合效果最好:非结构化的长文本、描述性内容用向量检索,结构化的关系、流程、实体用知识图谱。
知识图谱 RAG 和纯向量 RAG 有什么区别?各自适用场景?答:核心区别是存储和检索的单元不一样,一个是结构化的实体关系,一个是非结构化的文本片段,适用场景互补。 纯向量 RAG:存的是文本块的向量,靠语义相似度检索,适合描述性、开放性、长文本的问题,比如制度解释、操作说明、政策解读,覆盖场景广,实现简单,不用提前做结构化处理;缺点是关系推理弱,碎片化严重。 知识图谱 RAG:存的是实体和关系三元组,按实体和关系匹配,适合事实类、关系类、流程类的问题,比如组织架构、审批流程、参数定义,答案准确结构化,推理能力强;缺点是构建成本高,需要抽取三元组,覆盖的场景有限。 落地里一般是两者结合,叫混合增强 RAG:通用问题走向量检索,关系流程类问题叠加图谱结果,兼顾覆盖度和准确率。
知识图谱的三元组怎么构建?有哪些方式?答:从成本和精度从低到高,有三种常用方式。 第一种是人工构建,针对核心的、高频的领域知识,人工整理三元组,优点是准确率最高,缺点是成本高、速度慢,适合体量小、很重要的知识,比如核心审批流程、组织架构。 第二种是大模型抽取,把文档喂给大模型,让大模型自动抽取三元组,再人工抽检校准,速度快、成本中等,是目前最常用的方式,适合中等规模的知识图谱构建。 第三种是NER + 关系抽取模型,训练专门的信息抽取模型,批量自动抽取,适合大规模、持续增长的知识,优点是自动化程度高,缺点是需要标注数据、模型训练成本高。 我们目前是大模型抽取加人工抽检,性价比最高,核心流程人工校验,保证准确率。
知识图谱和向量检索的结果怎么融合?答:常用的融合方式有三种,在不同阶段融合,效果和复杂度不一样。 第一种是检索层融合,也就是召回阶段就把两者的结果都拿出来,一起做重排,然后再传给大模型,实现最简单,就是把图谱的结构化结论当成一个特殊的文档片段,和向量结果拼在一起,我们目前用的就是这种。 第二种是生成层融合,大模型先生成一版答案,再用知识图谱的事实去校验和修正,准确率更高,能减少幻觉,但实现复杂一点,要多一次大模型调用。 第三种是深度融合,把图谱的实体向量和文本向量结合起来,一起做检索,效果最好,但技术复杂度最高,需要改向量表示。 一般场景用检索层融合就够了,实现简单,收益也很明显;对准确率要求特别高的场景,可以做生成层的校验融合。
知识图谱增强 RAG 有什么局限性?答:不是所有场景都适合上知识图谱,它有明显的短板。 首先是构建成本高,需要做信息抽取、数据治理,不是扔一堆文档进去就能用,前期投入比纯向量 RAG 大很多。 然后是覆盖范围有限,只能处理实体关系类的问题,描述性、开放性的问题帮不上忙,不能替代向量检索。 还有就是维护成本高,知识更新的时候,要同步更新图谱,不像纯向量 RAG 更新文档就行,运维更复杂。 所以要不要上知识图谱,要看业务场景:如果大量涉及流程、组织、参数这类结构化知识,纯向量效果差,就值得做;如果都是纯文档描述类问题,纯向量就够用,不用硬上。
6. 高阶落地踩坑复盘
这几个模块看起来技术含量高,实际落地的时候踩的坑也更隐蔽,很多都是想当然设计,上线后才发现问题。
坑 1:全量降级打挂备用模型,引发雪崩
现象:一次主模型接口故障,网关自动把所有流量切到备用模型,结果备用模型 QPS 超限直接被限流,整个服务都不可用了,比单模型故障还严重。排查思路:看网关日志,发现故障触发后没有限流,瞬间全量流量打过去,备用模型承载不住,直接触发了厂商的限流策略,等于两个模型都用不了。解决方案:加熔断和分级降级机制。首先给每个模型设置 QPS 上限,降级的时候按速率切流量,不是一下全量过去;然后分场景降级,先降非核心、简单场景的流量,核心复杂场景保留主模型重试,不要一刀切全切;再加故障恢复探活,逐步切回流量,避免反复横跳。
坑 2:行为数据没去重,分析结论完全失真
现象:日报里显示某个问题一天被问了几百次,以为是超级高频问题,补了文档之后发现没什么人问,数据完全不准。排查思路:翻原始日志,发现是同一个用户同一个问题反复追问、刷新页面,都被算成了独立提问,统计出来的次数严重虚高。解决方案:多层去重。首先按会话去重,同一个会话里相同的问题不重复统计;然后按用户去重,统计问题热度的时候看有多少独立用户问过,不只是次数;再加语义去重,意思相近的不同问法归为同一个问题。改完之后的数据才真正有参考价值,不会误导优化方向。
坑 3:三元组抽取错误,图谱反而帮倒忙
现象:加了知识图谱之后,部分问题的回答反而错得更离谱,比如把部门关系说反了,还不如纯向量检索。排查思路:查对应的图谱数据,发现大模型批量抽取三元组的时候,把部分关系的方向搞反了,实体和关系对应错了,图谱里的事实本身就是错的,大模型基于错误的信息回答,自然会错。解决方案:首先控制抽取质量,核心领域的三元组必须人工抽检校准,不能全靠大模型自动抽;然后给图谱结果加置信度,置信度低的只做参考,不放在最前面;最后在 Prompt 里说明,图谱信息和文档冲突时,以文档内容为准。知识图谱是加分项,不能让它成为错误来源。
对应高阶故障排查类面试题(附答题思路)
LLM 网关切换备用模型后,服务反而完全不可用了,可能是什么原因?答:大概率是降级策略不合理引发了雪崩,常见三个原因: 第一是没有限流,全量流量瞬间打过去,超过了备用模型的承载上限,触发厂商限流或者直接打挂服务,导致两个模型都不可用。 第二是没有熔断机制,主模型反复故障恢复,流量来回切,两个模型都被打不稳定。 第三是备用模型本身能力不匹配,比如把复杂长文本的请求切给了小模型,导致大量超时、报错,等于服务不可用。 解决方式就是分级降级、限流保护、熔断机制三件套:降级时分场景逐步切流量,控制 QPS 不超过备用上限;故障后熔断,不要反复重试;不同复杂度的请求切对应能力的模型,保证切过去也能正常服务。
加了知识图谱之后,部分问题回答准确率反而下降了,怎么回事?答:一般是图谱本身的质量问题,或者融合方式不对,常见三个原因: 第一是三元组抽取错误,比如关系方向反了、实体对应错了,图谱里的事实本身就是错的,大模型基于错误信息回答,自然更差,这是最常见的原因。 第二是融合权重不对,把图谱结果放在最前面,优先级给得太高,大模型优先信图谱,哪怕图谱错了也跟着错,冲突的时候不会以文档为准。 第三是实体匹配错误,从问题里抽取实体的时候抽错了,查到了不相关的图谱信息,引入了噪声,干扰大模型判断。 排查的时候先查对应问题的图谱返回结果,看信息对不对、相关不相关,就能定位到是抽取问题、匹配问题还是融合问题。
用户行为统计数据和实际感受不符,热门问题没人认可,怎么排查?答:大概率是统计口径有问题,数据没做清洗去重,常见四类原因: 第一是没有去重,同一个用户反复刷新、追问同一个问题,都被统计成多次,导致次数虚高。 第二是没有语义归并,同一个意思的不同问法,被当成了不同的问题,分散了计数,看起来每个问题次数都不多。 第三是包含了大量测试、机器人调用的流量,没有过滤掉,污染了真实用户数据。 第四是统计维度不对,比如只看提问次数,不看独立用户数,某个用户反复问同一个问题,就会被当成热门问题。 解决方式就是先做数据清洗:过滤测试流量、按用户去重、语义归并相似问题,统计独立用户数和提问次数两个维度,数据就准了。
7. 高阶阶段面试考点自检清单
整理成纯问题清单,配合前面四篇的内容,从基础落地到高阶优化,全链路的面试题基本全覆盖。
LLM 网关类
- LLM 网关的核心价值与解决的问题
- LLM 网关的核心功能模块
- 多模型容灾降级与雪崩防护方案
- 智能路由的实现方式与选型
用户行为分析类
- 用户行为分析对 RAG 系统的价值
- 行为分析的核心维度与指标
- 如何用行为数据驱动 RAG 优化
- 区分 “内容缺失” 和 “检索失效” 的方法
知识图谱增强类
- 知识图谱增强 RAG 解决的核心痛点
- 知识图谱与纯向量 RAG 的区别与适用场景
- 三元组的构建方式与选型
- 图谱与向量检索的融合方案
- 知识图谱增强的局限性
- 搭建轻量 LLM 网关,统一多模型接入,实现容灾降级、智能路由与统一计费,服务可用性与成本管控能力显著提升
- 落地精细化用户行为分析体系,数据驱动知识库迭代,明确优化方向,内容命中率与问题解决率持续提升
- 实现知识图谱增强 RAG 方案,融合结构化实体关系与非结构化文本,流程与关系类问题准确率大幅提升
- 覆盖从基础功能、效果优化、运营落地到高阶增强的全链路能力,具备完整的企业级 RAG 系统规划与落地经验
写在最后
从最开始搭一个能跑的 Demo,到现在一步步加了重排、工作流、评测、网关、知识图谱。
经常有人问我,做 RAG 要不要上知识图谱、要不要做网关、要不要搞微调。我一直觉得没有标准答案,完全看你有没有对应的痛点。用户量小、就几个人用,直接调用 API 就够了,搞网关纯属过度设计;都是纯文档类问题,纯向量检索就够用,不用硬上知识图谱。技术从来不是越复杂越好,能解决问题的方案才是好方案。企业招人是来解决问题的,不是来背知识点的。