13.将手写 Agent 主流程迁移为 LangGraph 最小闭环，并接回 FastAPI + session 外壳

目 录前 言开始动手项目结构重构数据State化函数Node化串起Node形成Graph收尾前 言咱们前面的代码是通过手写Agent工作流程实现了一个论文RAG问答系统但是在实际生产环境中不会用这种纯手写工作逻辑项目更多的是使用现有框架比如LangGraph和LangChainLangChain是一种高级封装后的框架更适合需要借助智能体完成固定任务的非专业用户而LangGraph是一种更细粒度的智能体开发框架允许完全自定义图结构适合定制更复杂的智能体以满足个性化的任务需求。打个比较形象的比方相信很多人都喜欢摄影专业摄影师会有各种长枪短炮并且在摄影时会调整相机参数比如曝光度、白平衡等等然后看到有灵感的风景会考虑景深和前后景的关系最终拍出来一张美图。基于LangGraph的开发与这类似整个过程细节可调的地方非常多所以做出来的东西比较专业定制化也比较高。相反基于LangChain的智能体开发有点像是一个人拿着那种开了美颜的傻瓜机在拍照各种参数和接口都已经被提前设定好了也可以拍出能看的照片但是这种东西有上限。开始动手咱们手写Agent是写了三个函数choose_tool、execute_tool和generate_final_answer通过数据在三个函数间的流动实现了一个最小的智能体循环用户问题 → 思考 → 调工具 → 看结果 → 再思考 → 再决定下一步 → 最终回答当进入到LangGraph后咱们对数据的概念需要发生一点变化在这个框架中数据称为State对数据进行操作函数称为Node不同节点间通过Edge连接构成流这个框架命中带Graph可能是因为这个工作流结构和图论中的图一样。为了更好的理解LangGraph咱们还是在原有的代码上做迁移但是最好还是新建一个项目Python版本选择3.11随后安装咱们前面用到的依赖并且还要把LangGraph等依赖安装上pip install-U langgraph langchain langchain-openai pip install-U fastapi uvicorn openai python-dotenv pypdf numpy pydantic pip install-U faiss-cpu项目结构重构咱们的项目需要重构可以按照下面的目录重新组织:LangChain-for-A-Paper-Rag-Agent/ ├─ app/ │ ├─ __init__.py │ ├─ config.py │ ├─ logger_config.py │ ├─ llm_utils.py │ ├─ data_loader.py │ ├─ rag_system.py │ ├─ tools.py │ ├─ session_manager.py │ ├─ main.py │ └─ graph/ │ ├─ __init__.py │ ├─ state.py │ ├─ nodes.py │ └─ builder.py ├─ data/ ├─ .env └─ requirements.txt数据State化智能体对数据的操作原理上是Graph中的每个节点只负责读state,改state再把结果交给下一个节点。所以我们首先在app/graph/目录下创建state.py用于定义在图中流的statefromtypingimportAnyfromtyping_extensionsimportTypedDictclassAgentState(TypedDict,totalFalse):# 当前会话信息session_id:str# 当前用户问题query:str# 历史对话chat_history:list[dict[str,str]]# 路由决策结果# 例如: {tool: rag, input: what is ...}decision:dict[str,Any]# 工具执行结果# 例如:# {# tool_name: rag,# tool_input: ...,# tool_output: ...# }tool_result:dict[str,Any]# 最终返回给用户的答案final_answer:str# 预留错误字段后面做异常兜底会用到error:str函数Node化在LangGraph中函数称为Node我们需要咱们最早在agent.py中定义的三个函数进行改造让他们变成可以接受数据state的node首先需要在app/graph/目录下新建node.py然后定义两个工厂函数和一个节点关于为什么用工厂函数后面会详细解释。importjsonfromtypingimportAnyfromapp.graph.stateimportAgentStatefromapp.llm_utilsimportclientfromapp.configimportCHAT_MODELfromapp.logger_configimportsetup_logger loggersetup_logger()defbuild_choose_tool_node(tools:list[dict[str,Any]]):defchoose_tool_node(state:AgentState)-AgentState:querystate[query]tool_desc\n.join([f{t[name]}:{t[description]}fortintools])promptf You are an AI agent. Available tools:{tool_desc}User question:{query}Return JSON: {{tool: ..., input: ...}} contenttry:responseclient.chat.completions.create(modelCHAT_MODEL,messages[{role:user,content:prompt}])contentresponse.choices[0].message.content decisionjson.loads(content)exceptException:logger.warning(fTool decision parse failed:{content})decision{tool:llm,input:query}return{decision:decision}returnchoose_tool_nodedefbuild_execute_tool_node(tools:list[dict[str,Any]],ragNone):defexecute_tool_node(state:AgentState)-AgentState:decisionstate[decision]chat_historystate.get(chat_history,[])tool_namedecision[tool]tool_inputdecision[input]fortintools:ift[name]tool_name:iftool_namerag:resultt[func](tool_input,rag,chat_historychat_history)eliftool_namellm:resultt[func](tool_input,chat_historychat_history)else:resultt[func](tool_input)return{tool_result:{tool_name:tool_name,tool_input:tool_input,tool_output:result}}return{tool_result:{tool_name:none,tool_input:tool_input,tool_output:No valid tool found.}}returnexecute_tool_nodedefgenerate_answer_node(state:AgentState)-AgentState:querystate[query]tool_resultstate[tool_result]promptf You are an AI assistant. The user asked:{query}A tool was used: Tool name:{tool_result[tool_name]}Tool input:{tool_result[tool_input]}Tool output:{tool_result[tool_output]}Now provide a final helpful answer to the user. responseclient.chat.completions.create(modelCHAT_MODEL,messages[{role:user,content:prompt}])final_answerresponse.choices[0].message.contentreturn{final_answer:final_answer}串起Node形成Graph到这一步咱们就到了LangGraph的核心操作将各个node串起形成工作流前面咱们说过咱们在创建节点的时候用到了工厂函数在这一节我也想解释原因不够首先咱们还是先在app/graph/目录下新建一个builder.py文件fromlanggraph.graphimportStateGraph,START,ENDfromapp.graph.stateimportAgentStatefromapp.graph.nodesimport(build_choose_tool_node,build_execute_tool_node,generate_answer_node,)defbuild_agent_graph(tools,ragNone):graph_builderStateGraph(AgentState)# 1. 注册节点graph_builder.add_node(choose_tool,build_choose_tool_node(tools))graph_builder.add_node(execute_tool,build_execute_tool_node(tools,ragrag))graph_builder.add_node(generate_answer,generate_answer_node)# 2. 连接流程graph_builder.add_edge(START,choose_tool)graph_builder.add_edge(choose_tool,execute_tool)graph_builder.add_edge(execute_tool,generate_answer)graph_builder.add_edge(generate_answer,END)# 3. 编译 graphreturngraph_builder.compile()相信看到这里你已经看明白了LangGraph的逻辑这个框架将数据定义state将函数定义为node通过添加edge实现业务逻辑间的工作流。可能打一个不太恰当的比方通过LangGraph创建的智能体他身上像是绑定了一堆state数据你给它定义了node和edge它就会按照你定义的工作顺序去拿node修改自身绑定的state最后返回给你操作结果。关于工厂函数这里我解释一下我也是踩了坑了选择工具node和执行node不同于咱们的生成回答node生成回答node只需要给她传入state就可以得到响应的结果但是前两个node你发现他们在创建的时候是需要一些外部依赖的所以咱们需要定义工厂函数将外部依赖传入再生成出响应的node所以这里用到了工厂函数。一句话概括一些node的生成需要外部依赖所以咱们需要用工厂函数加工再生成node。收尾最后就是启动咱们的项目了在app目标下新建main.py文件然后在原有的代码基础上增加一步创建图的操作graph build_agent_graph(TOOLS, ragrag)就可以启动新的系统了这样便实现了手写Agent到LangGraph的迁移fromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModelfromapp.configimportDATA_DIRfromapp.data_loaderimportload_pdfs,process_documentsfromapp.rag_systemimportRAGSystemfromapp.toolsimportTOOLSfromapp.session_managerimportSessionManagerfromapp.graph.builderimportbuild_agent_graphfromapp.logger_configimportsetup_logger loggersetup_logger()appFastAPI()session_managerSessionManager(max_turns3)ragNonegraphNoneclassQueryRequest(BaseModel):session_id:strquestion:strapp.on_event(startup)defstartup_event():globalrag,graph logger.info(Loading RAG system...)docsload_pdfs(DATA_DIR)logger.info(fdocs数量:{len(docs)})chunksprocess_documents(docs)logger.info(fchunks数量:{len(chunks)})ragRAGSystem(chunks)rag.build_index()graphbuild_agent_graph(TOOLS,ragrag)logger.info(RAG LangGraph ready!)app.post(/ask)defask_question(req:QueryRequest):try:historysession_manager.get_history(req.session_id)state{session_id:req.session_id,query:req.question,chat_history:history,}resultgraph.invoke(state)answerresult[final_answer]session_manager.append_turn(req.session_id,req.question,answer)return{session_id:req.session_id,question:req.question,answer:answer}exceptExceptionase:logger.exception(Error occurred in /ask)raiseHTTPException(status_code500,detailstr(e))app.post(/clear/{session_id})defclear_session(session_id:str):session_manager.clear_session(session_id)return{session_id:session_id,message:session cleared}如果这篇文章对你有帮助可以点个赞完整代码地址https://github.com/1186141415/LangChain-for-A-Paper-Rag-Agent