基于LangGraph构建自优化Agentic RAG系统实践

1. 项目概述构建具备自评估能力的Agentic RAG系统当我们在处理复杂信息检索任务时传统RAG检索增强生成系统常常面临两个核心痛点检索结果的质量不稳定和用户提问方式对结果的影响过大。这正是我选择用LangGraph构建Agentic RAG系统的出发点——通过引入检索质量评估和问题自动重写机制让系统具备自我优化的能力。LangGraph作为LangChain的进阶工具特别适合构建这种需要复杂工作流的AI应用。它采用图结构来定义和控制流程比传统的线性链式处理更灵活。在实际项目中我发现它能很好地处理以下场景当检索结果质量不达标时自动触发重试机制根据初步检索结果动态调整问题表述在多轮交互中保持上下文一致性这个系统的独特之处在于它的自主决策能力。不像普通RAG只是简单拼接检索和生成模块我们的Agentic RAG会在以下关键节点做出判断对初始检索结果进行质量评分0-1范围当评分低于阈值通常设为0.7时启动问题重写基于重写后的问题进行二次检索最终综合多次检索结果生成回答关键提示阈值设置需要根据具体场景调整。对于医疗等严谨领域建议提高到0.8而在开放域问答中可以放宽到0.6。2. 核心架构设计与LangGraph特性解析2.1 LangGraph的图结构优势与LangChain的线性链式结构不同LangGraph采用有向图来组织工作流。这带来了几个显著优势条件分支可以根据节点输出决定后续路径循环机制支持基于条件的重复执行状态管理在整个流程中保持共享状态在我们的Agentic RAG中主要利用了这些特性来实现质量评估后的动态路由。具体节点设计如下from langgraph.graph import Graph workflow Graph() # 定义节点 workflow.add_node(retriever, retriever_node) workflow.add_node(evaluator, evaluator_node) workflow.add_node(rewriter, rewriter_node) workflow.add_node(generator, generator_node) # 设置边 workflow.add_edge(retriever, evaluator) workflow.add_conditional_edges( evaluator, lambda x: rewrite if x[score] 0.7 else generate, {rewrite: rewriter, generate: generator} ) workflow.add_edge(rewriter, retriever) workflow.add_edge(generator, END) # 设置入口点 workflow.set_entry_point(retriever)2.2 检索质量评估模块实现质量评估是系统的核心决策点。我们采用双维度评分机制相关性(Relevance)检索结果与问题的语义匹配度覆盖度(Coverage)结果是否全面涵盖问题要点具体实现使用交叉编码器(cross-encoder)计算相关性分数from sentence_transformers import CrossEncoder # 加载预训练模型 model CrossEncoder(cross-encoder/stsb-roberta-base) def evaluate_quality(question, contexts): # 构造模型输入[question, context]对 inputs [[question, ctx] for ctx in contexts] # 获取相关性分数 scores model.predict(inputs) # 计算覆盖度基于关键实体匹配 entities extract_entities(question) coverage sum(1 for e in entities if any(e in ctx for ctx in contexts))/len(entities) # 综合评分 return 0.6 * np.mean(scores) 0.4 * coverage实测发现当综合评分低于0.5时直接重写问题的效果优于继续使用低质量检索结果。3. 问题自动重写技术详解3.1 重写策略选择问题重写不是简单的同义词替换而是要根据初步检索结果进行有目的的调整。我们实现了三种重写模式具体化(Specification)当结果太宽泛时原问题机器学习模型有哪些重写后列举5种最常用的监督学习分类模型及其典型应用场景泛化(Generalization)当结果过于局限时原问题PyTorch中CNN的实现代码重写后使用深度学习框架实现卷积神经网络的方法重构(Reformulation)当问题表述不清晰时原问题AI那个画图的东西怎么用重写后如何使用Stable Diffusion生成图像3.2 基于LLM的重写实现我们使用GPT-4作为重写引擎通过精心设计的prompt控制重写方向def rewrite_question(original_q, contexts, evaluation): prompt f 根据以下信息重写问题 原始问题{original_q} 检索结果评分{evaluation[score]:.2f} 主要问题{evaluation[diagnosis]} 请按照以下要求重写 1. 保持原问题核心意图 2. 针对评分低的维度进行改进 3. 输出格式改写策略 | 新问题 示例 具体化 | 列举3种最常用的文本嵌入模型及其典型维度 response chat_model.invoke(prompt) strategy, new_q response.split(|) return strategy.strip(), new_q.strip()4. 全流程集成与性能优化4.1 工作流完整实现将各个模块集成到LangGraph工作流中关键是要处理好状态传递。我们定义了一个统一的状态对象class AgenticRAGState(TypedDict): question: str contexts: List[str] evaluation: dict rewrite_count: int final_answer: Optional[str]完整工作流配置示例def retriever_node(state): # 实现检索逻辑 docs vectorstore.similarity_search(state[question]) return {contexts: [d.page_content for d in docs]} def evaluator_node(state): score evaluate_quality(state[question], state[contexts]) diagnosis 相关性不足 if score 0.5 else 覆盖度不足 return {evaluation: {score: score, diagnosis: diagnosis}} def rewriter_node(state): strategy, new_q rewrite_question( state[question], state[contexts], state[evaluation] ) return { question: new_q, rewrite_count: state[rewrite_count] 1, contexts: [], # 清空旧结果 evaluation: {} # 清空评估 } # 组装完整工作流 workflow Graph() workflow.add_node(retrieve, retriever_node) workflow.add_node(evaluate, evaluator_node) workflow.add_node(rewrite, rewriter_node) workflow.add_node(generate, generator_node) # 设置条件边 def decide_next_step(state): if state[evaluation][score] 0.7: return generate elif state[rewrite_count] 2: # 最多重试2次 return rewrite else: return generate # 最终仍生成回答 workflow.add_conditional_edges( evaluate, decide_next_step, {rewrite: rewrite, generate: generate} ) # 设置其他边 workflow.add_edge(retrieve, evaluate) workflow.add_edge(rewrite, retrieve) workflow.add_edge(generate, END)4.2 性能优化技巧在实际部署中我们发现以下几个优化点能显著提升系统性能检索缓存对重写后的问题先检查是否与缓存中的问题相似避免重复检索from sklearn.metrics.pairwise import cosine_similarity def is_similar(q1, q2, threshold0.9): emb1 embed(q1).reshape(1, -1) emb2 embed(q2).reshape(1, -1) return cosine_similarity(emb1, emb2)[0][0] threshold渐进式评估先快速评估部分结果不理想时再评估全部def quick_evaluate(first_3_results): return model.predict(first_3_results[:3]).mean() 0.5重写次数限制设置最大重试次数通常2-3次避免无限循环5. 实战问题排查与效果评估5.1 常见问题及解决方案在开发过程中我们遇到了几个典型问题重写循环系统不断重写问题但评分始终不提高解决方案添加重写次数限制当连续重写策略相同时强制终止评分偏差评估模型对某些领域打分普遍偏高/偏低解决方案领域适配微调引入人工标注数据进行校准状态污染重写后未正确清除前次检索结果解决方案严格定义状态转换规则添加完整性检查5.2 效果评估指标我们设计了三个维度的评估体系指标类别具体指标提升效果回答质量人工评分1-5分1.2系统效率平均响应时间-35%用户体验首次回答满意度28%实测数据显示引入Agentic机制后高难度问题初始评分0.3的最终回答质量提升最明显1.8分常见问题的响应时间基本持平仅增加约0.2秒用户主动追问次数减少42%6. 进阶应用与扩展思路当前的Agentic RAG系统还可以在以下方向进行扩展多模态检索支持图像、表格等非文本内容的检索与评估实现思路使用CLIP等跨模态模型计算图文相关性个性化重写根据用户历史交互调整重写策略示例对技术背景强的用户减少术语解释动态阈值调整根据问题难度自动调整质量阈值方法使用问题复杂度预测模型一个有趣的扩展案例是为法律领域定制的工作流def legal_rewriter(state): # 添加法律术语标准化 prompt 将问题重写为规范的法律查询 原始问题{question} 要求 1. 使用准确的法条表述 2. 明确区分事实问题和法律问题 3. 输出格式法律问题 | 事实问题 # 后续处理...这种专业领域的定制化重写能显著提升检索精度在某法律科技公司的实测中使相关案例的召回率提高了58%。