在前四篇文章中,我们构建了能够渲染组件、流式解析、多工具并发和实时搜索的 AI 助手。然而,当面对复杂的多步骤任务时,传统的 while 循环开始显露局限性…
你是否遇到过这样的场景:
- 用户说:“我要去一个北京现在气温在 15 度以上的公园”
- AI 需要先搜索气温 → 如果不满足再搜索其他城市 → 找到合适城市后规划路线 → 预订门票
- 传统的 Agent 代码变成了一团乱麻,嵌套的 if-else 让人窒息
今天我们来解决这个问题,从简单 Agent升级到图化决策!
😱 回顾问题:传统 Agent 的"线性困境"
在第四篇文章中,我们使用 while 循环实现了简单的工具调用:
while(!isFinalPass){conststream=awaitmodel.stream(currentMessages);letgathered:AIMessageChunk|null=null;forawait(constchunkofstream){gathered=gathered?gathered.concat(chunk):chunk;}consttoolCalls=gathered?.tool_calls||[];if(toolCalls.length>0){// 执行工具...currentMessages.push(...toolMessages);}else{isFinalPass=true;}}❌ 传统 Agent 的问题
逻辑分散
业务逻辑散落在工具定义、System Prompt、执行代码中，想要修改决策流程,需要改动多个地方
示例场景:
// 工具定义constrawTools=[searchWeather,planRoute,bookTicket];// System PromptconstSYSTEM_PROMPT="先查天气,再规划,再订票";// 执行代码中还有逻辑if(call.name==='searchWeather'){if(result.temp<15){// 特殊处理...}}状态管理混乱
所有状态都在currentMessages中,难以区分"用户输入"、“AI 决策”、“工具结果”，想要追踪某个步骤的中间结果,需要遍历整个消息数组
示例:
// 想知道当前搜索的是哪个城市的天气?// 需要在 messages 数组中往前翻找...constlastToolCall=messages.findLast(m=>m.tool_calls?.[0]?.name==='searchWeather');constcity=lastToolCall.tool_calls[0].args.city;流程无法可视化
while循环是一个"黑盒",看不到决策路径，流程无法可视化
🎨 LangGraph 核心概念:图化决策
LangGraph是一个专门为LLM Agent设计的图编排框架,它把复杂的决策过程可视化为状态图。
🏗️ 三大核心概念
1️⃣ 节点 (Nodes)
节点是图中的"执行单元",可以是一个 AI 调用、一个工具执行,或一个业务逻辑。
newStateGraph(AgentState).addNode("check_weather",checkWeatherNode).addNode("search_alternative",searchAlternativeNode).addNode("plan_route",planRouteNode).addNode("book_tickets",bookTicketsNode)2️⃣ 边 (Edges)
边定义了节点之间的连接关系,分为两种类型:
条件边:
addConditionalEdges("check_weather",(state)=>(state.isWarmEnough?"satisfy":"fail"),{satisfy:"plan_route",fail:"search_alternative"})普通边:
addEdge(START,"check_weather")addEdge("search_alternative","plan_route")addEdge("book_tickets",END);3️⃣ 状态 (State)
状态是图中流动的数据,每个节点都可以读取和修改状态。
// TravelGraph.tsexportconstAgentState=Annotation.Root({// 对话消息历史（自动追加）messages:Annotation<BaseMessage[]>({reducer:(x,y)=>x.concat(y),default:()=>[],}),// 目标城市targetCity:Annotation<string>({reducer:(x,y)=>y??x,default:()=>"南京"}),// 天气是否达标（驱动条件路由）isWarmEnough:Annotation<boolean>({reducer:(x,y)=>y??x,default:()=>false}),});💻 完整案例实现:智能出行助手
让我们用 LangGraph 实现一个复杂场景:
用户说:“我要去现在气温在 15 度以上的公园”
AI 需要:
- 搜索北京当前气温
- 判断是否 ≥ 15°C
- 如果不满足,搜索其他城市
- 找到合适城市后,规划路线
- 预订公园门票
完整代码实现
整体架构一览
┌─────────────────────────────────────────┐ │ 前端 (React) │ │ graph-demo.tsx │ │ - 用户输入 & 消息渲染 │ │ - SSE 流式接收 & 插件渲染 │ └──────────────────┬──────────────────────┘ │ POST /api/chat (SSE) ┌──────────────────▼──────────────────────┐ │ 后端 (Express) │ │ server.ts │ │ - 接收请求，启动 LangGraph 流 │ │ - 将每个节点输出实时推送给前端 │ └──────────────────┬──────────────────────┘ │ langGraphApp.stream() ┌──────────────────▼──────────────────────┐ │ 工作流引擎 (LangGraph) │ │ TravelGraph.ts │ │ - 状态机定义 & 四个节点编排 │ │ - DeepSeek LLM + Tavily 搜索 │ └─────────────────────────────────────────┘完整执行流程图
逐节点深度解析
check_weather
这是工作流的第一个节点，负责查询目标城市的实时天气。
constcheckWeatherNode=async(state:typeofAgentState.State)=>{// 1. 用 Tavily 实时搜索天气constquery=`${state.targetCity}今日实时气温、天气状况`;constsearchRes=awaitsearchTool.invoke({query});// 2. 用结构化输出提取关键数据conststructuredModel=model.withStructuredOutput(WeatherSchema);constweatherData=awaitstructuredModel.invoke([{role:"system",content:`你是气象数据提取助手，严格提取【${state.targetCity}】的数据`},{role:"user",content:`搜索结果如下：\n${searchRes}`}]);// 3. 更新状态 & 生成带 tool_call 的 AI 消息return{isWarmEnough:weatherData.temperature>=15,// ← 驱动后续路由targetCity:weatherData.city,messages:[newAIMessage({content:`查询到${weatherData.city}当前气温为${weatherData.temperature}°C。`,tool_calls:[{name:"render_weather",args:weatherData,id:"weather_"+Date.now()}]})]};};工作流编排
// TravelGraph.ts — 工作流编排constworkflow=newStateGraph(AgentState).addNode("check_weather",checkWeatherNode).addNode("search_alternative",searchAlternativeNode).addNode("plan_route",planRouteNode).addNode("book_tickets",bookTicketsNode).addEdge(START,"check_weather")// ↓ 条件边：根据 isWarmEnough 决定下一个节点.addConditionalEdges("check_weather",(state)=>(state.isWarmEnough?"satisfy":"fail"),{satisfy:"plan_route",fail:"search_alternative"}).addEdge("search_alternative","plan_route").addEdge("plan_route","book_tickets").addEdge("book_tickets",END);search_alternative
当目标城市天气不达标时，此节点启动，实时搜索全国当前气温达标的城市作为替代方案：
constsearchAlternativeNode=async()=>{constsearchResults=awaitsearchTool.invoke({query:`请搜索中国哪些城市当前气温在15度以上，并且适合旅游？`});conststructuredLlm=model.withStructuredOutput(AlternativeCitySchema);constrecommendation=awaitstructuredLlm.invoke([{role:"system",content:`从搜索结果中挑选一个气温在15度以上的城市`},{role:"user",content:`搜索结果如下：\n${searchResults}`}]);return{targetCity:recommendation.city,// ← 更新目标城市，后续节点使用新城市isWarmEnough:true,messages:[newAIMessage({content:`...`,tool_calls:[...]})]};};plan_route
拿到确认达标的城市后，搜索景点并规划带坐标的地图路线：
constplanRouteNode=async(state:typeofAgentState.State)=>{// 使用最终确认的 targetCity（可能已被 search_alternative 更新）constquery=`${state.targetCity}适合15度以上天气游玩的公园和景点坐标`;constsearchResults=awaitsearchTool.invoke({query});conststructuredLlm=model.withStructuredOutput(MapItinerarySchema);constitinerary=awaitstructuredLlm.invoke([{role:"system",content:`为${state.targetCity}规划2-3个景点，提供准确经纬度`},{role:"user",content:`搜索结果：\n${searchResults}`}]);return{messages:[newAIMessage({content:`我已为您规划了${state.targetCity}的游玩路线，包含${itinerary.points.length}个推荐点。`,tool_calls:[{name:"render_map_itinerary",args:itinerary,id:"route_"+Date.now()}]})]};};book_tickets
这是最后一个节点，它从消息历史中提取上一步规划的景点，然后为第一个景点模拟预订：
constbookTicketsNode=async(state:typeofAgentState.State)=>{// 从消息历史反向查找 render_map_itinerary tool_callconstlastMessageWithRoute=[...state.messages].reverse().find(m=>(masAIMessage).tool_calls?.some(tc=>tc.name==="render_map_itinerary"));constrouteArgs=(lastMessageWithRouteasAIMessage)?.tool_calls?.find(tc=>tc.name==="render_map_itinerary")?.args;// 取第一个景点预订constspotToBook=routeArgs?.points?.[0]?.name||`${state.targetCity}热门景点`;// 调用 LLM 生成真实感票据conststructuredLlm=model.withStructuredOutput(TicketSchema);constticketData=awaitstructuredLlm.invoke([{role:"system",content:"你是票务系统助手，为景点生成预订成功的票据信息"},{role:"user",content:`请为景点"${spotToBook}"办理预订。`}]);return{messages:[newAIMessage({content:`祝贺！我已经为您成功预订了${ticketData.spotName}的门票。`,tool_calls:[{name:"book_tickets",args:{...ticketData,status:"success",confirmationCode:"TIC-"+Math.random().toString(36).substr(2,9).toUpperCase()}}]})]};};后端如何实时"推流"？
server.ts是连接 LangGraph 与前端的桥梁，核心是使用SSE（Server-Sent Events）将每个节点的输出实时推给浏览器：
server.post('/api/chat',async(req,res)=>{const{query}=req.body;// 设置 SSE 响应头res.setHeader('Content-Type','text/event-stream');res.setHeader('Cache-Control','no-cache');res.setHeader('Connection','keep-alive');// 启动 LangGraph 流，streamMode: "updates" 表示每个节点完成后推一次conststream=awaitlangGraphApp.stream({messages:[newHumanMessage(query)]},{streamMode:"updates"});// 每个节点完成 → 立即推送给前端forawait(constchunkofstream){res.write(`data:${JSON.stringify(chunk)}\n\n`);}res.end();});关键点：streamMode: "updates"让每个节点一旦执行完毕就立即推送，而不是等全部完成。用户能实时看到"天气查好了 → 路线规划中 → 票已预订"的进度，体验远优于等待。
为什么用 SSE 而不是普通 HTTP？
节点级实时推送：每个节点完成后立刻推一次，不等全流程结束
streamMode: “updates”：只推送本节点的增量输出，数据量小
用户体验：天气卡片 → 地图 → 票据依次出现，有明显的"AI 在工作"感知
格式：标准 SSE 格式data: <JSON>\n\n，浏览器原生支持
前端如何接收并渲染？
graph-demo.tsx使用ReadableStream API消费 SSE 流，并通过插件注册表将tool_calls渲染成实际 UI 组件：
// 1. 读取 SSE 流constreader=response.body.getReader();while(true){const{done,value}=awaitreader.read();if(done)break;constchunkText=decoder.decode(value);for(constlineofchunkText.split('\n')){if(!line.startsWith('data: '))continue;constchunk=JSON.parse(line.replace('data: ',''));constnodeName=Object.keys(chunk)[0];// e.g. "check_weather"constoutput=chunk[nodeName];// 2. 提取 tool_calls，构建 pluginMapconsttoolCalls=output.messages[last].kwargs.tool_calls;toolCalls.forEach(tc=>{newPluginMap[tc.id]={pluginName:tc.name,// e.g. "render_weather"argsParsed:tc.args,// 传给插件组件的 props};});// 3. 实时更新消息（插件合并，不覆盖）updateLastMessage({content:"...",pluginMap:newPluginMap});}}
插件化渲染的精髓：后端不返回纯文字，而是通过tool_calls下发"渲染指令"（pluginName + args）。前端的AIPluginRegistry充当一个组件工厂——根据pluginName找到React组件，把args直接当props传入。这实现了后端工作流直接驱动前端 UI 的效果。
端到端完整数据流
🎊 核心技术点总结
1. 状态机驱动，逻辑清晰
传统 Agent 用if/else堆砌，流程难以维护。LangGraph 用图结构 + 条件边表达业务逻辑，天气达标/不达标的路由一目了然：
.addConditionalEdges("check_weather",(state)=>state.isWarmEnough?"satisfy":"fail",{satisfy:"plan_route",fail:"search_alternative"})2. 结构化输出，数据可靠
每个节点都用model.withStructuredOutput(Schema)约束 LLM 的输出格式，坐标、温度、票价都是强类型数据，不会出现"LLM 返回了一段话但解析不出来"的问题。
3. SSE 流式推送，体验流畅
streamMode: "updates"+ SSE 的组合让每个节点完成后立即呈现结果，用户不需要等待整个流程结束，天气卡片 → 地图 → 票据依次出现，有明显的"AI 在工作"的感知。
4. 插件化 UI 渲染
后端通过tool_calls传递渲染指令（而非纯文字），前端的AIPluginRegistry根据pluginName找到对应 React 组件并传入参数。这样后端工作流的输出直接驱动前端 UI，实现了真正的"AI 生成界面"。
5. 消息历史作为通信总线
bookTicketsNode没有专门的"景点名称"参数，而是从state.messages消息历史中反向查找render_map_itinerary的 tool_call 参数。这是一种优雅的节点间数据传递方式——共享状态即通信。
🎉 总结
这个项目最核心的价值不在于"会查天气、会订票"，而在于：
用 LangGraph 把一个复杂的多步骤 AI 任务，拆解成可读、可测、可扩展的图结构；用 SSE 流式输出把每个步骤的进展实时传递给用户；用插件化 UI 把 AI 的结构化输出变成真实可交互的界面组件。
这三者的结合，才是构建"真正好用的 AI 应用"的正确姿势。
