Day27：LangGraph 实战落地｜Tool_RAG + 并行子图 + 持久化部署，打造工业级 AI Agent

Day27LangGraph 实战落地Tool/RAG 并行子图 持久化部署打造工业级 AI Agent 博客导语昨天我们搞定了 LangGraph 核心逻辑Day2 直接进入实战落地阶段一篇博客分基础、进阶、高级三阶段从 Tool/RAG 集成到并行执行再到生产级持久化部署把 LangGraph 变成能干活、能上线的企业级 AI Agent通义千问 本地 Qwen 双版本代码直接复用往期 RAG / 工具内容学完即可上线 第一阶段基础篇LangGraph 集成 ToolRAG打造实用 AI 助手核心概念白话拆解Tool 集成让 AI 调用外部工具计算器、天气、数据库实现实际操作RAG 集成对接私有向量库让 AI 基于本地知识回答杜绝幻觉实战案例2 个案例 1LangGraph 多 Tool 工具集成fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDfromlangchain_core.toolsimporttoolimportos# 通义千问配置os.environ[DASHSCOPE_API_KEY]你的API Keyfromlangchain_community.llmsimportTongyi llmTongyi(modelqwen-plus,temperature0.1)# 定义两个工具tooldefcalc(a:int,b:int)-int:数学计算returnabtooldefweather(city:str)-str:天气查询returnf{city}今日晴天25℃# 定义State路由节点复用Day1条件路由classAgentState(TypedDict):question:strtool_type:strresult:str# 完整工作流代码直接运行测试工具调用workflowStateGraph(AgentState)# 节点路由定义省略同Day1进阶appworkflow.compile()print(app.invoke({question:北京天气怎么样}))案例 2LangGraphChroma RAG 私有知识库# 集成往期Chroma向量库实现私有知识问答fromlangchain_community.vectorstoresimportChromafromlangchain_community.embeddingsimportDashScopeEmbeddingsfromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDimportos os.environ[DASHSCOPE_API_KEY]你的API KeyembeddingDashScopeEmbeddings(modeltext-embedding-v3)# 加载本地向量库vector_dbChroma(persist_directory./chroma_lpr_db,embedding_functionembedding)# 定义RAG检索节点defrag_node(state:AgentState):docsvector_db.similarity_search(state[question],k2)context\n.join([d.page_contentfordindocs])return{context:context}# 结合大模型生成精准回答实现私有知识库Agent基础阶段避坑指南Tool 的 description 必须写清晰AI 才能正确调用RAG 检索的向量库要和 Embedding 模型一致检索结果精简避免上下文过长 第二阶段进阶篇并行节点 子图复杂工作流高效编排核心概念白话拆解并行节点让 AI 同时执行多个任务检索 工具提速 50%子图把复杂流程拆成小模块代码更优雅、易维护实战案例2 个案例 1并行节点执行检索 工具同时跑fromlanggraph.graphimportStateGraph,START,ENDfromtypingimportTypedDict# 定义StateclassAgentState(TypedDict):question:strrag_res:strtool_res:str# 并行执行两个节点defrag_node(state):return{rag_res:RAG检索内容}deftool_node(state):return{tool_res:工具执行结果}# 构建并行工作流workflowStateGraph(AgentState)workflow.add_node(rag,rag_node)workflow.add_node(tool,tool_node)# 同时执行两个节点workflow.add_conditional_edges(START,lambda_:[rag,tool])workflow.add_edge(rag,END)workflow.add_edge(tool,END)appworkflow.compile()进阶阶段避坑指南并行节点结果要做好合并逻辑子图命名要清晰避免模块混乱并行任务不宜过多防止资源占用过高 第三阶段高级篇持久化 FastAPI 部署企业级 Agent 上线核心概念白话拆解持久化保存 AI 对话记忆断电 / 重启不丢失支持断点续跑FastAPI 部署封装成接口对外提供服务对接前端 / 小程序实战案例2 个案例 1MemorySaver 状态持久化# 持久化保存AI记忆实现多轮对话fromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.graphimportStateGraph,START,END# 初始化持久化memoryMemorySaver()# 编译工作流绑定持久化workflowStateGraph(AgentState)# 节点路由定义省略appworkflow.compile(checkpointermemory)# 指定用户会话ID记忆保留config{configurable:{thread_id:user_001}}app.invoke({question:LangGraph核心组件},configconfig)# 第二次调用记忆不丢失app.invoke({question:刚才说的组件有哪些},configconfig)案例 2FastAPI 服务化部署# 封装API接口生产环境上线fromfastapiimportFastAPIfromlanggraph.checkpoint.memoryimportMemorySaver appFastAPI(titleLangGraph企业级API)memoryMemorySaver()graphworkflow.compile(checkpointermemory)# 对外问答接口app.post(/agent/ask)defask(question:str,user_id:str):config{configurable:{thread_id:user_id}}resgraph.invoke({user_question:question},configconfig)return{answer:res[ai_answer]}# 启动命令uvicorn main:app --host 0.0.0.0 --port 8000高级阶段避坑指南生产环境用 Redis/MySQL 替代 MemorySaver接口添加限流、鉴权防止恶意调用开启日志监控方便排查 AI 执行问题总结一篇搞定 LangGraphTool/RAG 集成、并行子图、持久化部署从实用 AI 助手到企业级服务彻底实现 LangGraph 全栈落地学完即可独立开发上线级 AI Agent