[RAG在LangChain中的实现-05]利用LangGraph编写RAG应用
![[RAG在LangChain中的实现-05]利用LangGraph编写RAG应用](http://pic.xiahunao.cn/yaotu/[RAG在LangChain中的实现-05]利用LangGraph编写RAG应用)
在“让LLM在指定的上下文范围内回答问题”我们利用LangChain(create_agent工厂函数)编写了一个简单RAG应用这是一个将指定博文内容作为上下文的QA应用。现在我们使用LangGraph的编程模式重现实现它并添加如下两个功能查询和上下文相关性评估在利用查询文本从VectorStore检索出相关内容之后我们利用LLM评估它们之间是否具有相关性重新生成查询如果没有通过相关性评估我们会利用LLM重新生成具有更高质量的查询文本并重启QA流程。整个流程如下图所示入口节点retrieve_context利用查询文件检索博文内容并将其作为上下文。如果通过了上述的相关性评估直接转入generate_response节点生成回答否则转入regenerate_query节点重新生成更高质量的查询文本。regenerate_query节点完成后再次转向入口节点重启流程。接下来我们分步骤介绍整个应用的实现。![Alternative Text][1791123792692]步骤一 检索上下文我们首先通过如下的步骤创建作为上下文检索器的Retriever对象我们利用WebBaseLoader加载博文内容并转换成Document列表后者被RecursiveCharacterTextSplitter切割并将切片添加到创建的InMemoryVectorStore中最后调用InMemoryVectorStore的as_retriever方法得到所需的检索器。loaderWebBaseLoader(web_paths(https://www.cnblogs.com/artech/p/inside-asp-net-core-framework.html,),bs_kwargs{parse_only:SoupStrainer(class_(postBody))},)documentsloader.load()splitterRecursiveCharacterTextSplitter(chunk_size1000,chunk_overlap200,add_start_indexTrue,)retriever(InMemoryVectorStore.from_documents(documentssplitter.split_documents(documents),embeddingOpenAIEmbeddings(modeltext-embedding-3-small),).as_retriever())我们定义了如下的State作为整个执行流程的状态类型。为了让大家了解整个流程的执行情况我们会将每个步骤的执行信息以日志的形式写入log变量表示的列表。log[]classState(TypedDict):query:strcontext:strresult:strgenerate_times:Annotated[int,operator.add]State类型具有如下四个成员query查询文本可以是用户输入的原始查询文本也可以是在相关性评估失败后重新生成的查询context根据查询文本检索得到的作为上下文的文本result最终的回答generate_times如果评估一直失败检索-评估-查询循环将会一直持续下去generate_times对查询再生成进行计数如下所示的retrieve_context为上下文检索节点函数只需要直接调用Retriever对象将格式化后的检索内容用于更新状态的context成员。在方法返回之前我们将检索信息写入日志。defretrieve_context(query:State)-dict:retrieved_docsretriever.invoke(query[query])context\n\n.join((f来源:{doc.metadata}\n内容:{doc.page_content})fordocinretrieved_docs)log.append(f检索上下文: 检索到{len(retrieved_docs)}个相关文档\n)return{context:context,result:}步骤二相关性评估我们使用LLM评估查询文本与检索内容的相关性并利用模型的结构化输出得到一个确定的二元结果评估成功或者失败如下所示的QueryEvaluationResult为绑定的结构化输出类型两个字段成员is_relevant和reason为评估结果和理由。我们使用的模型组件为ChatOpenAI并调用with_structured_output方法完成结构化输出Schema的绑定。classQueryEvaluationResult(TypedDict):对查询与上下文相关性的评估结果is_relevant:bool查询与上下文相关性的判断结果reason:str对判断结果的简要说明llmChatOpenAI(modelgpt-5.2-chat)如下所示的evaluate_query_relevance为相关性评估函数返回的字符串表示在评估成功和失败情况下的路由节点。为了避免导致的无限循环我们将查询文本的最大生成次数设置为3。defevaluate_query_relevance(state:State)-Literal[generate_response,regenerate_query]:if(state.get(generate_times)3):returngenerate_responseevaluation_prompt(fQuery:{state[query]}\nfContext:{state[context]}\n\n请判断这个查询与上下文的相关性返回格式为{is_relevant: bool, reason: str}其中reason是对判断结果的简要说明。)responsellm.with_structured_output(QueryEvaluationResult).invoke([HumanMessage(contentevaluation_prompt)])is_relevantresponse[is_relevant]pathgenerate_responseifis_relevantelseregenerate_querylog.append(f查询相关性评估: 结果{不相关ifnotis_relevantelse相关}, 路由{path}, 原因{response[reason]}. )returnpath在使用查询文本和上下文格式化提示词后我们调用ChatOpenAI对象得到评估结果和理由并据此返回最终的路由节点名称。在方法返回之前我们将评估信息写入日志。步骤三查询文本再生成如下所示的查询文本再生成节点函数regenerate_query会根据当前状态提供的查询文本和上下文构建提示词并调用ChatOpenAI对象得到由LLM生成的高质量的查询文本。我们在利用返回的字典更新query和generate_times成员之前会将查询再生成的信息写入日志defregenerate_query(state:State)-dict:regeneration_prompt(f根据以下上下文信息重新生成一个与查询相关的新查询。\n\nfContext:{state[context]}\n\nfOriginal Query:{state[query]}\n\n请直接回复新查询的内容不要任何额外的说明和前后缀。)queryllm.invoke([HumanMessage(contentregeneration_prompt)]).content log.append(f重新生成查询 原始查询: {state[query]}, 新查询: {query} )return{query:query,generate_times:1}步骤四生成答案最终得到的答案由如下的节点函数generate_response生成它利用状态提供的查询和上下文生成提示词调用ChatOpenAI对象。在利用返回的字典将得到的结果写入状态的result成员之前我们也会将相关执行信息写入日志。defgenerate_response(state:State)-dict:response_prompt(f根据以下上下文信息使用精炼的语言回答查询尽量控制在100个字以内。\n\nfContext:{state[context]}\n\nfQuery:{state[query]}\n\n)resultllm.invoke([HumanMessage(contentresponse_prompt)]).content log.append(f生成最终回答:{result})return{result:result}步骤五图的构建和编译我们创建了一个StateGraph对象并将添加了上面定义的三个节点其中retrieve_context和generate_response作为入口和完成节点。我们在retrieve_context和generate_respons/regenerate_query节点之间添加了一个“条件边”条件分支函数为evaluate_query_relevance。regenerate_query和retrieve_context之间的静态边确保查询再生成后流程再次启动。builder(StateGraph(State).add_node(retrieve_context,retrieve_context)# type: ignore.add_node(generate_response,generate_response).add_node(regenerate_query,regenerate_query).set_entry_point(retrieve_context).set_finish_point(generate_response).add_conditional_edges(retrieve_context,evaluate_query_relevance).add_edge(regenerate_query,retrieve_context))agentbuilder.compile().with_config(recursion_limit10)payloadagent.get_graph().draw_mermaid_png()PILImage.open(io.BytesIO(payload)).show()在将StateGraph编译成Agent之后我们调用其get_graph方法得到Graph对象并将其转换成PNG图片呈现出来开篇给出的图片就是最终的呈现效果。步骤六调用Agent我们调用Agent并问了一个问题:比较一下ASP.NET Core和Express。并输出最终的状态和写入的日志。resultagent.invoke({query:比较一下ASP.NET Core和Express})# type: ignoreprint(f最终状态: query{result[query]}context{result[context][:50]}result{result[result][:50]}generate_times{result[generate_times]})print(\n\nExecution Log:)fori,entryinenumerate(log):print(f{i1}.{entry})最终的输出如下所示。从输出的日志可以清除地看出其执行流程上下文检索 - 查询相关性评估失败- 查询再生成 - 上下文检索 - 查询相关性评估失败- 查询再生成 - 上下文检索-查询相关性评估成功- 生成最终答案。最终状态: query结合RequestDelegate与FuncRequestDelegate, RequestDelegate的抽象方式深入分析ASP.NET Core中间件管道的设计动机及其对异步请求处理的支持机制 context来源: {source: https://www.cnblogs.com/artech/p/i resultASP.NET Core将请求处理抽象为RequestDelegateFuncHttpContext,Task统一同步/异步模型中间件用FuncRequestDelegate,RequestDelegate包装后续管道实现职责链组合。该设计简洁、可组合天然支持异步与管道式扩展。 generate_times2 Execution Log: 1. 检索上下文: 检索到 4 个相关文档 2. 查询相关性评估: 结果不相关, 路由regenerate_query, 原因查询要求比较 ASP.NET Core 与 Express而上下文内容仅详细介绍了 ASP.NET Core 的框架原理与实现没有涉及 Express 或两者的对比因此相关性不足。. 3. 重新生成查询 原始查询: 比较一下ASP.NET Core和Express, 新查询: 从请求处理管道和中间件设计的角度对比ASP.NET Core与Express框架的核心架构思想 4. 检索上下文: 检索到 4 个相关文档 5. 查询相关性评估: 结果不相关, 路由regenerate_query, 原因上下文内容详细介绍了 ASP.NET Core 的请求处理管道与中间件设计但未涉及 Express 框架或其架构思想无法支持对 ASP.NET Core 与 Express 的对比分析。. 6. 重新生成查询 原始查询: 从请求处理管道和中间件设计的角度对比ASP.NET Core与Express框架的核心架构思想, 新查询: 结合RequestDelegate与FuncRequestDelegate, RequestDelegate的抽象方式深入分析ASP.NET Core中间件管道的设计动机及其对异步请求处理的支持机制 7. 检索上下文: 检索到 4 个相关文档 8. 查询相关性评估: 结果相关, 路由generate_response, 原因查询聚焦于RequestDelegate与FuncRequestDelegate, RequestDelegate在ASP.NET Core中间件管道中的设计动机及异步支持机制而上下文内容正系统性地解释了RequestDelegate为何采用FuncHttpContext, Task、中间件为何抽象为FuncRequestDelegate, RequestDelegate以及这种设计如何支持异步请求处理二者高度一致。. 9. 生成最终回答: ASP.NET Core将请求处理抽象为RequestDelegateFuncHttpContext,Task统一同步/异步模型中间件用FuncRequestDelegate,RequestDelegate包装后续管道实现职责链组合。该设计简 洁、可组合天然支持异步与管道式扩展。如果运行时设置了LangSmith相关的环境变量从其提供的Trace不仅仅可以看清执行的流程还可以获取每个步骤的输入和输出。附上完整的代码。fromdotenvimportload_dotenv load_dotenv()fromtypingimportTypedDict,Literal,Annotatedfromlangchain_openaiimportChatOpenAI,OpenAIEmbeddingsfromlangchain_core.vectorstoresimportInMemoryVectorStorefromlangchain_core.messagesimportHumanMessagefrombs4.filterimportSoupStrainerfromlangchain_community.document_loadersimportWebBaseLoaderfromlangchain_text_splittersimportRecursiveCharacterTextSplitterfromlanggraph.graphimportStateGraphfromPILimportImageasPILImageimportio,operator loaderWebBaseLoader(web_paths(https://www.cnblogs.com/artech/p/inside-asp-net-core-framework.html,),bs_kwargs{parse_only:SoupStrainer(class_(postBody))},)documentsloader.load()splitterRecursiveCharacterTextSplitter(chunk_size1000,chunk_overlap200,add_start_indexTrue,)retriever(InMemoryVectorStore.from_documents(documentssplitter.split_documents(documents),embeddingOpenAIEmbeddings(modeltext-embedding-3-small),).as_retriever())log[]classState(TypedDict):query:strcontext:strresult:strgenerate_times:Annotated[int,operator.add]classQueryEvaluationResult(TypedDict):对查询与上下文相关性的评估结果is_relevant:bool查询与上下文相关性的判断结果reason:str对判断结果的简要说明defretrieve_context(query:State)-dict:retrieved_docsretriever.invoke(query[query])context\n\n.join((f来源:{doc.metadata}\n内容:{doc.page_content})fordocinretrieved_docs)log.append(f检索上下文: 检索到{len(retrieved_docs)}个相关文档\n)return{context:context,result:}llmChatOpenAI(modelgpt-5.2-chat)# llm_evaluate ChatOpenAI(modelgpt-5.2-chat)defevaluate_query_relevance(state:State)-Literal[generate_response,regenerate_query]:if(state.get(generate_times)3):returngenerate_responseevaluation_prompt(fQuery:{state[query]}\nfContext:{state[context]}\n\n请判断这个查询与上下文的相关性返回格式为{is_relevant: bool, reason: str}其中reason是对判断结果的简要说明。)responsellm.with_structured_output(QueryEvaluationResult).invoke([HumanMessage(contentevaluation_prompt)])is_relevantresponse[is_relevant]pathgenerate_responseifis_relevantelseregenerate_querylog.append(f查询相关性评估: 结果{不相关ifnotis_relevantelse相关}, 路由{path}, 原因{response[reason]}. )returnpathdefgenerate_response(state:State)-dict:response_prompt(f根据以下上下文信息使用精炼的语言回答查询尽量控制在100个字以内。\n\nfContext:{state[context]}\n\nfQuery:{state[query]}\n\n)resultllm.invoke([HumanMessage(contentresponse_prompt)]).content log.append(f生成最终回答:{result})return{result:result}defregenerate_query(state:State)-dict:regeneration_prompt(f根据以下上下文信息重新生成一个与查询相关的新查询。\n\nfContext:{state[context]}\n\nfOriginal Query:{state[query]}\n\n请直接回复新查询的内容不要任何额外的说明和前后缀。)queryllm.invoke([HumanMessage(contentregeneration_prompt)]).content log.append(f重新生成查询 原始查询: {state[query]}, 新查询: {query} )return{query:query,generate_times:1}builder(StateGraph(State).add_node(retrieve_context,retrieve_context)# type: ignore.add_node(generate_response,generate_response).add_node(regenerate_query,regenerate_query).set_entry_point(retrieve_context).set_finish_point(generate_response).add_conditional_edges(retrieve_context,evaluate_query_relevance).add_edge(regenerate_query,retrieve_context))agentbuilder.compile().with_config(recursion_limit10)payloadagent.get_graph(xrayTrue).draw_mermaid_png()PILImage.open(io.BytesIO(payload)).show()resultagent.invoke({query:比较一下ASP.NET Core和Express})# type: ignoreprint(f最终状态: query{result[query]}context{result[context][:50]}result{result[result]}generate_times{result[generate_times]})print(\n\nExecution Log:)fori,entryinenumerate(log):print(f{i1}.{entry})