目录

  1. 前言
  2. 深化环境依赖补充
  3. 模块一:LangGraph 可编排工作流智能体落地
  4. 模块二:基于知识库问答对的轻量化 SFT 微调
  5. 模块三:多租户数据隔离与权限管控
  6. 深化落地踩坑复盘
  7. 深化阶段面试考点自检清单写在最后

前言

前两篇把基础链路和进阶优化都跑完了,真要往企业级落地再深一步,绕不开三个高频需求:更灵活的智能体工作流、垂直领域效果瓶颈的突破、多租户权限隔离。这三个点也是面试深挖的重灾区,只说基础 RAG+Agent 很容易被问住,讲清楚深化的思路和落地细节,才能拉开差距。

这篇顺着前面的项目继续迭代,把三块内容逐个落地:用 LangGraph 把原来的黑盒 Agent 改成可编排的状态机工作流,支持重试、分支、校验;用积累的知识库问答对做轻量化 LoRA 微调,突破 Prompt 和 RAG 的效果上限;再加上多租户数据隔离和权限管控,适配企业多部门的使用场景。


2. 深化环境依赖补充

在前面依赖的基础上补装几个包,尽量保持轻量化。LangGraph 直接装官方包就行;微调部分用 LoRA 轻量化方案,不用全量训练,普通 GPU 甚至 CPU 都能跑小批量测试;多租户部分不用额外组件,靠 Milvus 原生能力 + 应用层校验实现。

pip install langgraph peft datasets transformers accelerate

需要额外准备:

  • 整理好的知识库问答对数据集(用于微调)
  • 多租户场景的部门 / 租户标识(用于权限隔离)

3. 模块一:LangGraph 可编排工作流智能体落地

原来用的 ReAct 和 Plan-and-Execute Agent,优点是开箱即用,但缺点也很明显:执行链路是黑盒,不好控制分支逻辑,碰到需要重试、答案校验、异常回退的场景特别别扭。比如检索结果为空要自动改写 query 再查、答案不符合要求要重生成、涉及敏感内容要走人工审核,这些用普通 Agent 很难优雅实现。

所以我把智能体改成了 LangGraph 状态机架构,把整个对话链路拆成独立节点,每个节点只做一件事,通过边和条件分支串起来,流程完全可控,想加什么环节直接插节点就行,扩展性强太多。

我设计的基础工作流有五个核心节点:query 改写→知识库检索→答案生成→答案校验→结果输出,校验不通过就自动回到 query 改写节点重试,最多重试 3 次,比原来盲目的工具调用稳很多。

核心代码实现

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage

# 定义工作流状态:全程共享的数据结构
class AgentState(TypedDict):
    user_query: str
    rewritten_query: str
    retrieved_docs: list
    answer: str
    retry_count: int
    max_retry: int
    is_valid: bool

# 节点1:query改写,优化检索问句
def node_query_rewrite(state: AgentState):
    """把口语化、指代不清的问题改写成标准检索问句"""
    prompt = f"""请把用户的问题改写成适合知识库检索的标准问句,保留核心语义,去除口语化表达。
用户问题:{state['user_query']}
改写后的问句:"""
    rewritten = llm.invoke(prompt).content
    return {"rewritten_query": rewritten, "retry_count": state["retry_count"] + 1}

# 节点2:知识库检索
def node_retrieve(state: AgentState):
    """调用两级检索器获取相关文档片段"""
    retriever = get_retriever(load_vector_store())
    docs = retriever.get_relevant_documents(state["rewritten_query"])
    return {"retrieved_docs": docs}

# 节点3:答案生成
def node_generate(state: AgentState):
    """基于检索上下文生成答案"""
    context = "\n".join([doc.page_content for doc in state["retrieved_docs"]])
    prompt = f"""请仅基于以下文档内容回答用户问题,禁止编造。
文档内容:{context}
用户问题:{state['user_query']}
答案:"""
    answer = llm.invoke(prompt).content
    return {"answer": answer}

# 节点4:答案校验
def node_validate(state: AgentState):
    """校验答案是否基于文档内容,有没有幻觉"""
    context = "\n".join([doc.page_content for doc in state["retrieved_docs"]])
    prompt = f"""请判断下面的答案是否完全基于提供的文档内容,有没有编造信息。
文档内容:{context}
答案:{state['answer']}
只输出Yes或No:"""
    result = llm.invoke(prompt).content.strip()
    is_valid = "yes" in result.lower()
    return {"is_valid": is_valid}

# 条件分支:校验通过就结束,不通过且没到重试次数就回去改写
def route_after_validate(state: AgentState):
    if state["is_valid"]:
        return END
    elif state["retry_count"] < state["max_retry"]:
        return "rewrite"
    else:
        return END

# 构建工作流图
def build_rag_workflow():
    graph = StateGraph(AgentState)
    
    # 添加所有节点
    graph.add_node("rewrite", node_query_rewrite)
    graph.add_node("retrieve", node_retrieve)
    graph.add_node("generate", node_generate)
    graph.add_node("validate", node_validate)
    
    # 设置入口和连接关系
    graph.set_entry_point("rewrite")
    graph.add_edge("rewrite", "retrieve")
    graph.add_edge("retrieve", "generate")
    graph.add_edge("generate", "validate")
    
    # 添加条件分支
    graph.add_conditional_edges(
        "validate",
        route_after_validate,
        {
            "rewrite": "rewrite",
            END: END
        }
    )
    
    return graph.compile()

# 运行工作流
def rag_workflow_chat(user_query: str):
    workflow = build_rag_workflow()
    initial_state = {
        "user_query": user_query,
        "rewritten_query": "",
        "retrieved_docs": [],
        "answer": "",
        "retry_count": 0,
        "max_retry": 3,
        "is_valid": False
    }
    result = workflow.invoke(initial_state)
    return result["answer"]

这套工作流跑下来,答非所问和幻觉的情况又少了很多,因为加了校验重试机制,不会一次检索不准就随便输出答案。而且后续想加新环节特别方便,比如加敏感词检测、加人工审核节点,直接插进去就行,不用改整个 Agent 的逻辑。

对应面试考点(附答题思路)

  1. LangGraph 和传统 LangChain Agent 有什么区别?优势是什么? 答:核心区别是可控性和灵活性。传统 LangChain Agent 是黑盒式的循环执行,逻辑由大模型自己控制,开发者很难干预执行流程,适合简单工具调用场景。 LangGraph 是基于状态机的显式编排,开发者可以自定义节点、分支、循环、回退,每一步做什么、怎么走分支都完全可控,适合复杂业务流程。 优势很明显:一是流程可控,能实现重试、校验、人工介入、异常兜底等复杂逻辑;二是可观测性强,每个节点的状态和结果都能清晰追踪,方便调试和排障;三是扩展性好,新增业务环节只要加节点就行,不用重构整个智能体。 简单场景用普通 Agent 就够了,复杂业务流程、对稳定性要求高的场景,用 LangGraph 更合适。

  2. 你们的智能体工作流是怎么设计的?有哪些核心节点? 答:我们基于 LangGraph 设计了 RAG 问答工作流,核心是 “改写 - 检索 - 生成 - 校验” 的闭环,支持自动重试优化。 核心节点有五个:一是 query 改写节点,把用户口语化、有指代的问题改写成标准检索问句,提升召回准确率;二是知识库检索节点,走向量召回 + Rerank 精排的两级检索;三是答案生成节点,基于检索上下文生成合规答案;四是答案校验节点,用大模型判断答案有没有幻觉、是否基于文档内容;五是条件分支,校验通过就输出结果,不通过且没到重试上限就回到改写节点重新来,最多重试 3 次。 整个流程是显式可控的,每一步的状态都可追踪,后续还可以扩展敏感词检测、人工审核、多工具调度等节点。

  3. 什么场景下适合用 LangGraph 做编排?什么场景用普通 Agent 就够? 答:普通 Agent 适合简单、线性的工具调用场景,比如单工具问答、简单多工具调度,对流程控制要求不高,追求快速上线的场景,优点是开发快、轻量。 LangGraph 适合复杂、有分支、有循环、需要严格管控流程的场景,比如:需要多轮重试、答案校验的高可靠 RAG 问答;有明确步骤的多工具协同工作流;需要人工介入、审批节点的业务流程;还有异常分支多、兜底逻辑复杂的场景。 选型不是非此即彼,可以结合用:简单任务走轻量 Agent,复杂任务走 LangGraph 工作流,按需调度。

  4. LangGraph 里的状态管理是怎么做的?怎么保证数据流转正确? 答:LangGraph 用统一的状态对象(State)在所有节点间共享数据,每个节点接收当前状态,执行逻辑后返回更新后的状态,全程状态唯一、单向流转。 状态一般定义成 TypedDict,包含用户输入、中间结果、执行计数、标记位这些全程需要共享的数据;每个节点只修改自己负责的字段,不改动其他节点的数据,避免互相干扰。 保证流转正确的手段:一是定义清晰的状态结构,字段职责明确;二是节点职责单一,每个节点只做一件事,只更新对应字段;三是用条件边做分支控制,严格按照业务逻辑流转,避免非法路径;四是加边界控制,比如重试次数上限,防止无限循环。


4. 模块二:基于知识库问答对的轻量化 SFT 微调

RAG、Prompt 优化、分块优化都做到头之后,会碰到效果天花板:垂直领域的专业术语模型理解不到位、答案风格和企业要求不统一、回答总是太泛不够精准。这时候再抠 RAG 细节收益很小,就该上模型微调了。

我没有做全量微调,成本太高也没必要,用的是 LoRA 轻量化 SFT(有监督微调)。基于日常积累的高质量知识库问答对,只训练少量低秩参数,不用动基座模型,普通 GPU 就能跑,成本低见效快,非常适合企业垂直知识库场景。

核心思路就是:把历史上用户的问题和人工校准后的标准答案整理成数据集,用 LoRA 对基座模型做微调,让模型学会这个领域的回答风格、专业术语和知识边界,再和 RAG 结合使用,效果比纯 RAG 好一个档次。

核心代码实现

from datasets import Dataset
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from peft import LoraConfig, get_peft_model, TaskType
from transformers import TrainingArguments, Trainer

# 1. 构造微调数据集:问题 + 标准答案
def build_sft_dataset(qa_pairs: list):
    """
    构造SFT数据集,格式遵循对话模板
    qa_pairs格式:[{"question": "xxx", "answer": "xxx"}]
    """
    formatted_data = []
    for item in qa_pairs:
        prompt = f"""你是企业知识库助手,请仅基于内部知识准确回答用户问题,回答简洁严谨。
用户问题:{item['question']}
答案:"""
        formatted_data.append({
            "prompt": prompt,
            "response": item["answer"]
        })
    return Dataset.from_list(formatted_data)

# 2. LoRA 配置:轻量化微调,只训练少量参数
lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=8,          # 秩,越小参数量越少
    lora_alpha=32,
    lora_dropout=0.05,
    target_modules=["q_proj", "v_proj"],  # 只微调注意力层
    bias="none"
)

# 3. 加载基座模型和分词器
model_name = "Qwen/Qwen2-7B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    device_map="auto",
    load_in_4bit=True  # 4bit量化,进一步降显存
)

# 挂载LoRA适配器
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()  # 可训练参数一般不到1%

# 4. 训练配置
training_args = TrainingArguments(
    output_dir="./rag_sft_lora",
    per_device_train_batch_size=4,
    gradient_accumulation_steps=2,
    learning_rate=2e-4,
    num_train_epochs=3,
    logging_steps=10,
    save_strategy="epoch",
    fp16=True
)

# 5. 启动训练
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=sft_dataset,
)
trainer.train()

# 6. 推理使用:加载LoRA权重,和RAG结合
def sft_rag_answer(user_query: str, context: str):
    prompt = f"""你是企业知识库助手,请仅基于以下文档内容回答问题。
文档内容:{context}
用户问题:{user_query}
答案:"""
    inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
    outputs = model.generate(**inputs, max_new_tokens=512, temperature=0.1)
    return tokenizer.decode(outputs[0], skip_special_tokens=True)

实测下来,微调之后,垂直领域术语的准确率、答案的专业度、回答风格的统一性都有明显提升,幻觉率也降了一截。不用太多数据,几百条高质量问答对就能看到效果,性价比很高。

对应面试考点(附答题思路)

  1. RAG 和模型微调有什么区别?分别适合什么场景? 答:两者是完全不同的优化方向,解决的问题不一样。 RAG 是检索增强,从外部知识库拉取上下文,不修改模型参数,优点是知识更新方便、不会产生幻觉、成本低,适合知识频繁更新、需要溯源、数据量大的场景;缺点是对复杂推理、风格统一、专业术语理解的提升有限。 微调是修改模型参数,让模型学习特定领域的知识和风格,优点是回答更专业、风格更统一、推理能力更强,适合领域术语多、回答风格要求固定、RAG 优化到瓶颈的场景;缺点是知识更新麻烦、容易产生幻觉、训练成本高,有灾难遗忘风险。 落地里一般是两者结合:RAG 负责动态知识和事实性内容,微调负责领域风格和推理能力,1+1>2。

  2. 做知识库微调的时候,训练数据怎么构造?有哪些注意点? 答:核心是用真实场景的问答对,而不是随便生成数据,不然微调出来的效果不落地。 数据来源优先选真实用户提问 + 人工校准的标准答案,这是质量最高的;数据量不够的话,可以用大模型基于知识库文档生成高质量问答对,再人工抽检。 构造注意点:一是数据分布要和真实业务一致,覆盖高频问题、边缘问题、易错问题;二是答案风格要统一,符合业务要求,不然微调出来风格乱;三是要加入 “无答案” 样本,也就是知识库没有的内容要回答不知道,防止微调后模型瞎编;四是控制数据质量,低质量数据反而会把模型训坏。 一般几百到几千条高质量数据就够了,不是越多越好,质量比数量重要。

  3. LoRA 微调的原理是什么?为什么能做到轻量化? 答:LoRA 是低秩适配,核心思想是大模型的权重更新是低秩的,也就是大部分参数不用动,只需要训练一个小的低秩矩阵就能模拟全量微调的效果。 它不修改基座模型的原始参数,而是在注意力层旁边加两个小的低秩矩阵,训练的时候只更新这两个矩阵的参数,推理的时候把低秩矩阵和原权重合并,效果和全量微调接近,但可训练参数量只有原来的千分之几到百分之几。 所以它的优势就是:训练显存需求低,普通 GPU 就能跑 7B 甚至 13B 模型;训练速度快,成本低;不修改基座模型,不会有灾难遗忘,多个任务可以用多个 LoRA 适配器切换,非常适合企业垂直场景的轻量化微调。

  4. 微调后怎么评估效果?怎么防止过拟合和灾难遗忘? 答:评估分两部分:客观指标和人工评估。客观指标可以算答案的 BLEU、ROUGE 分数,看和标准答案的重合度;还有忠实度、相关性的大模型自动打分。但最核心的还是人工评估,看准确率、幻觉率、风格匹配度、专业度这些,自动指标只能做参考。 防止过拟合:控制训练轮次,不要训太多 epoch;用小的学习率;数据要多样化,不要反复训同一批数据;加 dropout 正则化。 防止灾难遗忘:因为用 LoRA 本身就不修改基座模型,灾难遗忘风险比全量微调小很多;另外可以在训练数据里混入少量通用样本,保持模型的通用能力;还有就是 LoRA 的秩不要设太大,学习率不要太高。

  5. 为什么不直接微调模型,还要做 RAG?两者怎么结合? 答:因为单独微调解决不了所有问题,甚至会带来新问题。首先微调没法实时更新知识,企业制度一变,重新微调成本很高,RAG 改文档就行;然后微调容易产生幻觉,没法溯源,RAG 能给出引用来源,可信度更高;还有就是数据量小的时候,微调效果不如 RAG,成本还高。 两者结合是最优解:RAG 负责提供事实性知识,保证答案准确、可溯源、更新方便;微调负责优化模型的领域理解能力、回答风格和推理能力,让答案更专业更贴合业务。 具体结合方式就是:检索还是走 RAG 链路拿到上下文,然后用微调后的模型来生成答案,既保留 RAG 的知识准确性,又发挥微调的风格和推理优势。


5. 模块三:多租户数据隔离与权限管控

企业内部用起来之后,很快就会碰到多租户的需求:不同部门、不同业务线都有自己的知识库,数据不能互通,还要按角色控制访问权限。比如人力的知识库销售不能看,研发的文档行政不能查,这就需要做多租户隔离和权限管控。

我做了两层方案适配:中小租户、数据量小的,用单 collection 加租户 ID 标量过滤,成本低管理方便;大租户、数据量大、对隔离性要求高的,单独建 collection 物理隔离。应用层加统一权限校验,每次检索都校验租户权限,防止越权访问。

核心代码实现

from langchain_community.vectorstores import Milvus

# 多租户向量库管理
class MultiTenantVectorStore:
    def __init__(self, tenant_id: str, isolation_mode: str = "filter"):
        """
        isolation_mode: filter 标量过滤模式(中小租户)
                       collection 物理隔离模式(大租户)
        """
        self.tenant_id = tenant_id
        self.isolation_mode = isolation_mode
        self.embedding = embedding

    def _get_collection_name(self):
        if self.isolation_mode == "collection":
            return f"kb_{self.tenant_id}"  # 每个租户独立集合
        return "enterprise_shared_kb"      # 共享集合,标量过滤

    def add_documents(self, chunks: list[Document]):
        """写入文档,自动注入租户ID元数据"""
        for chunk in chunks:
            chunk.metadata["tenant_id"] = self.tenant_id
        
        vector_store = Milvus.from_documents(
            documents=chunks,
            embedding=self.embedding,
            collection_name=self._get_collection_name(),
            connection_args={"host": MILVUS_HOST, "port": MILVUS_PORT}
        )
        return vector_store

    def as_retriever(self, top_k: int = 10):
        """获取带租户权限过滤的检索器"""
        search_kwargs = {"k": top_k}
        if self.isolation_mode == "filter":
            # 标量过滤:只查当前租户的数据
            search_kwargs["filter"] = {"tenant_id": self.tenant_id}
        
        vector_store = Milvus(
            embedding_function=self.embedding,
            collection_name=self._get_collection_name(),
            connection_args={"host": MILVUS_HOST, "port": MILVUS_PORT}
        )
        return vector_store.as_retriever(
            search_type="similarity",
            search_kwargs=search_kwargs
        )

# 应用层权限校验中间件
def permission_check(tenant_id: str, user_role: str, allowed_roles: list):
    """简单权限校验,实际可对接企业权限系统"""
    if user_role not in allowed_roles:
        raise PermissionError("无权限访问该租户知识库")
    return True

# 带权限校验的问答入口
def tenant_chat(user_query: str, tenant_id: str, user_role: str):
    # 第一步:权限校验
    permission_check(tenant_id, user_role, ["admin", "member"])
    # 第二步:获取对应租户的检索器
    store = MultiTenantVectorStore(tenant_id, isolation_mode="filter")
    retriever = store.as_retriever()
    # 第三步:走RAG问答链路
    docs = retriever.get_relevant_documents(user_query)
    context = "\n".join([doc.page_content for doc in docs])
    prompt = f"""仅基于文档回答问题。文档:{context} 问题:{user_query}"""
    return llm.invoke(prompt).content

这样设计的好处是灵活,小租户共享资源降成本,大租户独立集合保性能和隔离性,后续租户多了也能平滑扩容,不用重构架构。

对应面试考点(附答题思路)

  1. 多租户场景下,向量数据有哪几种隔离方案?各有什么优劣? 答:主流有三种方案,从隔离性从低到高分别是: 第一种是共享集合 + 标量字段过滤,所有租户数据存在同一个 collection 里,加 tenant_id 字段,查询的时候过滤。优点是资源利用率高、管理简单、成本低;缺点是隔离性弱,有数据越权风险,数据量大了之后单集合性能会下降。适合中小租户、数据量不大的场景。 第二种是每个租户独立 collection,物理隔离,每个租户有自己的向量集合。优点是隔离性强、数据安全、性能互不影响;缺点是管理成本高,租户多了之后集合数量爆炸,资源利用率低。适合大租户、数据量大、对安全隔离要求高的场景。 第三种是独立部署实例,每个租户单独一套向量库服务。优点是隔离性最强、完全物理隔离;缺点是成本极高、运维复杂。只适合对数据安全要求极高的大客户。 落地一般是混合方案:中小租户用标量过滤,大租户用独立集合,兼顾成本和安全性。

  2. 你们的多租户权限管控是怎么做的?完整链路是什么? 答:我们做了两层管控:应用层权限校验 + 数据层租户隔离。 完整链路:用户发起请求后,首先在应用层做权限校验,校验用户身份、所属租户、角色权限,没有权限直接拒绝;校验通过后,根据租户 ID 和租户等级选择对应的隔离模式,构造带租户过滤条件的检索器;检索的时候强制带上租户过滤条件,只能查询当前租户的数据;最后返回结果,全程不会接触到其他租户的数据。 另外还有审计日志,所有访问都记录租户、用户、查询内容,方便追溯。核心原则是:权限校验前置,数据层强制过滤,不相信前端传参,全程防止越权。

  3. 怎么防止越权访问其他租户的知识库数据? 答:核心是多层防护,不能只靠前端传参,也不能只在数据层过滤。 第一层是应用层权限校验,用户登录后从 token 里解析租户和角色,不接受前端传的租户 ID,防止篡改;每次请求都校验用户是否有权限访问目标租户。 第二层是数据层强制过滤,检索逻辑里强制注入租户过滤条件,不是由前端传参控制,避免绕过过滤;独立 collection 模式下,直接限制只能访问对应租户的集合。 第三层是审计和监控,记录所有查询日志,定期巡检有没有越权访问的异常请求;还有数据加密,敏感数据存储加密,进一步保障安全。 最关键的一点是:租户过滤条件必须在后端强制拼接,不能由前端或者入参控制,防止恶意篡改。

  4. 租户数量很多的时候,怎么管理向量库资源?怎么平衡性能和成本? 答:核心是分级管理,差异化对待,不搞一刀切。 首先是租户分级:按数据量和重要性分成大中小三级,大租户独立 collection 甚至独立节点,保障性能和隔离性;中小租户共享集合,用标量过滤,降低成本。 然后是资源调度:对共享集合做容量规划,单集合数据量到阈值就拆分,防止数据太多导致性能下降;定期清理过期、无效数据,控制数据量。 还有就是冷热分离:高频访问的数据做缓存、建优化索引;低频冷数据归档存储,降成本。 这样既不会为了小租户浪费资源,也不会让大租户受共享资源的性能影响,平衡成本和体验。


6. 深化落地踩坑复盘

坑 1:LangGraph 状态无限膨胀,token 频繁超限

现象:工作流重试几次之后,输入 token 暴涨,频繁报上下文超限,而且越跑越慢。 排查思路:打印了每一步的状态数据,发现每次重试都会把上一轮的检索结果、生成答案全留在状态里,越堆越多,几轮下来上下文就炸了。 解决方案:精简状态字段,状态里只存必要的核心数据,不存全量中间结果;检索结果和生成答案只保留最新一份,历史结果做摘要或者直接丢弃;同时严格限制最大重试次数,最多 3 次,防止无限循环。改完之后状态大小基本稳定,不会越跑越膨胀。

坑 2:微调后通用能力下降,出现灾难遗忘

现象:垂直领域问题答得更好了,但通用问题、常识问题回答质量明显下降,甚至出现低级错误。 排查思路:对比了微调前后的效果,确认是微调导致的灾难遗忘;因为训练数据全是垂直领域问答,而且学习率设高了,LoRA 秩也设大了,影响了基座模型的通用能力。 解决方案:首先降低 LoRA 秩,从 16 降到 8,减少可训练参数;调低学习率,从 5e-4 降到 2e-4;然后在训练数据里混入 10% 左右的通用问答样本,保持模型通用能力;同时减少训练轮次,从 5 轮降到 3 轮,防止过拟合。调整之后通用能力基本恢复,垂直领域效果也没掉。

坑 3:多租户标量过滤性能劣化

现象:租户数量和数据量上来之后,共享集合的检索耗时越来越长,比单租户慢好几倍。 排查思路:排查了 Milvus 的查询日志,发现租户 ID 字段没有建标量索引,每次过滤都是全量扫描,数据量越大越慢;而且单集合数据量破百万之后,检索性能本身也会下降。 解决方案:首先给 tenant_id 字段建立标量索引,加速过滤;然后做租户分级,数据量大的大租户拆分到独立 collection,不占共享资源;最后给共享集合设置数据量上限,到阈值就扩容拆分。优化之后,检索耗时基本恢复到单租户水平。

对应深化故障排查类面试题(附答题思路)

  1. LangGraph 工作流运行久了 token 越来越长,经常超限,怎么优化? 答:核心原因是状态冗余和循环累积,从三个方向优化: 第一是精简状态结构,状态里只存必要的字段和最新结果,不保存所有历史中间产物,比如检索结果、生成答案只留最新的,历史的直接丢弃或者做摘要压缩。 第二是限制循环次数,给重试、循环节点加上限,比如最多重试 3 次,防止无限循环导致上下文越堆越多。 第三是上下文压缩,需要保留的中间结果用摘要或者关键信息提取,不用全量留存;长文本先压缩再放进状态。 另外还可以在关键节点加 token 计数,超限自动触发截断和压缩,兜底防止报错。

  2. 模型微调后垂直领域效果好了,但通用问题回答变差了,怎么回事? 答:这是典型的灾难遗忘 / 领域偏移问题,常见原因有几个: 一是训练数据太单一,全是垂直领域数据,没有通用样本,模型把通用知识 “忘了”;二是训练强度太大,学习率太高、轮次太多、LoRA 秩太大,改动的参数太多,影响了基座能力;三是微调方式不对,全量微调比 LoRA 更容易出现这个问题。 解决方式:首先优先用 LoRA 轻量化微调,不修改基座模型,本身就能大幅降低灾难遗忘风险;然后在训练数据里混入一定比例的通用样本,保持模型通用能力;接着调小学习率和 LoRA 秩,减少训练强度,控制训练轮次,不要过拟合;最后如果是多个领域,可以用多个独立 LoRA 适配器,按需切换,不用一个模型塞所有能力。

  3. 多租户数据量上来之后,检索耗时越来越长,怎么排查优化? 答:先定位是共性变慢还是个别租户变慢,再逐层排查: 先看数据层:首先检查过滤字段有没有建标量索引,没索引的话全量扫描肯定慢;然后看单集合数据量是不是太大,超过性能阈值;还有索引是不是失效了,有没有定期优化。 再看架构层:如果是共享集合模式,数据量大会互相影响,可以把大租户拆分到独立 collection,物理隔离;小租户继续共享,控制单集合的数据量。 最后做性能优化:给高频查询加缓存,减少重复检索;优化索引类型,根据数据量选合适的索引算法;还有冷热数据分离,冷数据归档,热数据优化。 一般先加标量索引、再拆分大租户,就能解决大部分性能问题。


7. 深化阶段面试考点自检清单

工作流编排类

  1. LangGraph 与传统 Agent 的核心区别与优势
  2. RAG 工作流的核心节点设计与流转逻辑
  3. LangGraph 的适用场景与选型判断
  4. LangGraph 状态管理的实现方式与注意事项

模型微调类

  1. RAG 与模型微调的区别与适用场景
  2. SFT 微调数据集的构造方法与注意事项
  3. LoRA 微调的核心原理与轻量化优势
  4. 微调效果评估方法与过拟合、灾难遗忘的解决方案
  5. RAG 与微调的结合方式与各自分工

多租户权限类

  1. 向量数据多租户隔离方案对比与选型
  2. 多租户权限管控完整链路
  3. 防止越权访问的多层防护手段
  4. 大规模租户下的资源管理与成本平衡

  • 基于 LangGraph 重构 RAG 智能体为可编排状态机工作流,新增答案校验与自动重试机制,幻觉率进一步降低
  • 落地 LoRA 轻量化 SFT 微调方案,基于知识库问答对优化垂直领域回答效果,专业度与风格统一性显著提升
  • 设计多租户混合隔离架构,支持标量过滤与物理集合双模式,兼顾数据安全、性能与成本
  • 完成从基础链路到进阶优化、再到深化落地的全栈迭代,具备完整的 RAG 系统工程化落地能力
Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐