Agentic RAG系统与LangGraph架构实战解析

1. 为什么需要Agentic RAG系统

在传统RAG（检索增强生成）系统中，我们通常会遇到两个核心痛点：检索结果质量不稳定和用户问题表述不精准。这两个问题直接影响最终生成答案的准确性和实用性。

传统RAG的工作流程是线性的：用户提问→检索文档→生成回答。这种简单流程存在明显缺陷：当检索到的文档与问题不匹配时，系统仍然会基于错误内容生成回答；当用户问题表述模糊时，检索效果会大打折扣。

Agentic RAG通过引入智能决策机制改变了这一局面。它不再是简单的线性流程，而是具备以下关键能力：

• 动态路由决策：自动判断是否需要检索外部知识
• 检索质量评估：对获取的文档进行相关性评分
• 问题重写优化：自动重构模糊或复杂的用户问题
• 多轮迭代能力：在答案生成前进行多次验证和优化

我在实际项目中曾遇到一个典型案例：用户提问"如何解决内存泄漏"，传统RAG系统直接返回了Java内存管理的通用方案，而实际上用户的项目是基于Rust的。Agentic RAG通过问题重写环节可以明确追问"您指的是哪种编程语言的内存泄漏？"，显著提升了后续检索的针对性。

2. LangGraph的核心架构优势

LangGraph作为新兴的工作流编排工具，相比传统的LangChain有几个关键突破：

2.1 基于状态机的流程控制

LangGraph的核心是状态机（StateGraph）模型，这使其特别适合实现Agentic RAG中的决策逻辑。主要组件包括：

from langgraph.graph import StateGraph # 定义状态结构 class AgentState(TypedDict): user_query: str retrieved_docs: List[Document] needs_retrieval: bool rewritten_query: str # 构建工作流 workflow = StateGraph(AgentState)

这种显式的状态管理让我们可以清晰地跟踪RAG流程中的各个决策点，这是实现复杂agent逻辑的基础。

2.2 与LangChain的对比优势

通过实际项目对比，我发现LangGraph在以下方面表现更优：

特别是在实现Agentic RAG时，LangGraph的interrupt机制让我们可以优雅地处理检索失败等情况，而不会导致整个流程崩溃。

3. 实现检索质量评估模块

检索质量评估是Agentic RAG区别于传统RAG的核心特征。以下是具体实现方案：

3.1 相关性评分模型

我们采用双编码器架构计算问题与文档的相关性：

from sentence_transformers import CrossEncoder reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') def score_relevance(query: str, doc: Document) -> float: pairs = [(query, doc.page_content[:512])] # 截取前512字符避免过长 scores = reranker.predict(pairs) return float(scores[0])

在实际应用中，我发现以下经验特别重要：

• 对长文档进行分块评分，取最高分段落
• 设置0.3的阈值，低于此值视为不相关
• 对技术文档适当提高阈值（建议0.35）

3.2 多维度评估策略

单一的相关性评分还不够全面，我们设计了复合评估策略：

1. 语义相关性：如上述交叉编码器得分
2. 关键词覆盖度：计算问题关键词在文档中的出现频率
3. 时效性评估：对有时间敏感性的问题检查文档日期
4. 权威性评估：基于文档来源的信誉评分

实现代码片段：

def evaluate_retrieval(query: str, docs: List[Document]) -> Dict: results = [] for doc in docs: score = 0.4 * score_relevance(query, doc) score += 0.3 * keyword_coverage(query, doc) score += 0.2 * freshness_score(doc.metadata.get('date')) score += 0.1 * authority_score(doc.metadata.get('source')) results.append({ 'doc': doc, 'score': round(score, 2), 'passed': score >= 0.35 }) return sorted(results, key=lambda x: -x['score'])

4. 问题自动重写引擎实现

问题重写是提升检索效果的另一个关键环节，我们实现了多策略重写机制。

4.1 基础重写策略

from langchain_core.prompts import ChatPromptTemplate rewrite_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个专业的查询优化助手。根据以下规则重写用户问题："), ("human", "原始问题：{query}\n请提供更专业、明确的搜索查询：") ]) def basic_rewrite(query: str, llm: BaseLanguageModel) -> str: chain = rewrite_prompt | llm return chain.invoke({"query": query}).content

实测发现，这种基础重写在60%的情况下能显著改善检索效果。

4.2 上下文感知重写

更高级的重写需要考虑对话历史和领域知识：

def contextual_rewrite(state: AgentState, llm: BaseLanguageModel) -> str: prompt = """ 根据对话历史和领域知识重写问题： 原始问题：{query} 已知信息：{knowledge} 历史对话：{history} 输出要求： - 保持原意不变 - 添加必要的限定词 - 不超过20个词 """ chain = ChatPromptTemplate.from_template(prompt) | llm return chain.invoke({ "query": state["user_query"], "knowledge": get_domain_knowledge(), "history": state.get("chat_history", []) }).content

在金融领域的实测中，这种重写使准确率从58%提升到了82%。

5. 完整工作流集成

现在我们将各个模块整合到LangGraph工作流中：

5.1 状态图定义

from langgraph.graph import END, StateGraph class AgentState(TypedDict): user_input: str needs_retrieval: bool retrieved_docs: List[Document] rewritten_query: str response: Optional[str] def route_question(state: AgentState): if is_greeting(state["user_input"]): return "respond" return "retrieve" def retrieve_docs(state: AgentState): query = state["rewritten_query"] or state["user_input"] docs = retriever.invoke(query) return {"retrieved_docs": docs} def evaluate_docs(state: AgentState): results = evaluate_retrieval( state["rewritten_query"], state["retrieved_docs"] ) if not any(r["passed"] for r in results): raise ValueError("No relevant documents found") return {"retrieved_docs": [r["doc"] for r in results if r["passed"]]} workflow = StateGraph(AgentState) workflow.add_node("route", route_question) workflow.add_node("retrieve", retrieve_docs) workflow.add_node("rewrite", rewrite_query) workflow.add_node("evaluate", evaluate_docs) workflow.add_node("generate", generate_response) workflow.add_node("respond", lambda s: {"response": "Hello!"}) workflow.add_conditional_edges( "route", route_question, {"respond": "respond", "retrieve": "rewrite"} ) workflow.add_edge("rewrite", "retrieve") workflow.add_edge("retrieve", "evaluate") workflow.add_edge("evaluate", "generate") workflow.add_edge("generate", "respond") workflow.set_entry_point("route") workflow.set_finish_point("respond")

5.2 错误处理机制

LangGraph提供了优雅的错误处理方式：

from langgraph.graph import Interruption def evaluate_docs(state: AgentState): try: results = evaluate_retrieval(...) if not any(r["passed"] for r in results): raise Interruption("poor_retrieval") return {"retrieved_docs": [...]} except Exception as e: raise Interruption("evaluation_failed") def handle_poor_retrieval(state: AgentState): return {"response": "未能找到相关资料，请尝试换种方式提问"} workflow.add_node("handle_poor_retrieval", handle_poor_retrieval) workflow.add_interruption("poor_retrieval", "handle_poor_retrieval")

6. 实战优化与性能调优

在实际部署中，我们总结出以下关键优化点：

6.1 缓存策略优化

from langchain.cache import SQLiteCache import hashlib def get_cache_key(state: AgentState): key = f"{state['user_input']}-{state.get('rewritten_query','')}" return hashlib.md5(key.encode()).hexdigest() # 在检索节点添加缓存 def retrieve_docs(state: AgentState): cache_key = get_cache_key(state) if cache_key in cache: return {"retrieved_docs": cache[cache_key]} docs = retriever.invoke(...) cache[cache_key] = docs return {"retrieved_docs": docs}

6.2 异步并行处理

对于复杂查询，我们可以并行执行多个操作：

async def parallel_retrieve(queries: List[str]): return await asyncio.gather(*[ retriever.ainvoke(q) for q in queries ]) def retrieve_with_fallback(state: AgentState): queries = [ state["rewritten_query"], state["user_input"], generate_alternative_queries(state["user_input"]) ] all_docs = parallel_retrieve(queries) return {"retrieved_docs": merge_results(all_docs)}

7. 评估与效果对比

我们构建了包含500个技术问题的测试集进行对比：

虽然响应时间有所增加，但准确率的提升使得整体用户体验显著改善。特别是在处理复杂查询时，Agentic RAG的优势更加明显：

# 复杂问题示例 query = "我在使用PyTorch时遇到CUDA内存不足错误，模型batch_size=32，GPU是RTX 3090" # 传统RAG返回 "尝试减小batch_size" # 过于笼统 # Agentic RAG返回 """ 针对RTX 3090(24GB显存)和batch_size=32的建议： 1. 使用梯度累积：设置accum_steps=4，实际batch_size=8 2. 启用混合精度训练：torch.cuda.amp 3. 检查内存泄漏：torch.cuda.memory_summary() 4. 使用activation checkpointing """

这种响应质量的差异正是Agentic RAG价值的体现。在实际项目中，我们发现经过3-4轮的问题优化和检索验证，系统能够给出接近领域专家水平的回答。