文章探讨了单一智能体在处理复杂旅行规划任务时的局限性提出了使用LangGraph搭建多智能体系统的解决方案。该系统通过主智能体和子智能体的分工合作实现了可中断的权限控制和可委派的专业分工有效解决了多步骤、多工具和有风险操作的组合任务问题。文章详细介绍了系统的架构设计包括状态管理、工具分类机制、智能体协作机制以及关键实现等并提供了一份可直接运行的完整代码骨架帮助读者理解和实践多智能体系统的构建。开会前十分钟你同事突然丢来一句“我下周去北京帮我把行程改一下——航班改签、旧酒店取消、换一家离新会议地点近的、再加个租车。”你脑子里已经开始自动列 checklist 了先查余票再看退改规则再核对酒店取消政策再对比新酒店位置和预算还要确认租车取还车时间……而很多“传统对话式 AI”的尴尬就在这它会聊天但一遇到这种“多步骤 多工具 有风险操作”的组合任务就容易把事情越办越乱。这篇文章我想从“为什么单一智能体扛不住”讲到“怎么用 LangGraph 搭一个可中断、可委派的多智能体旅行系统”最后给你一份可以直接跑的完整代码骨架包含权限中断流程。一、当单一AI无法应对复杂旅行场景1痛点复杂旅行规划不是“回答问题”而是“执行流程”想象这个场景用户帮我改签航班然后取消旧酒店再订新酒店最后租车。这里面至少包含四类能力搜索查航班、查酒店、查租车多为只读策略什么时候先改签、什么时候先取消避免损失规则退改签/取消政策、时间冲突、预算约束执行真正“改”“退”“订”这类动作往往不可逆如果你用单一智能体硬扛通常会变成这样它必须同时掌握所有工具、所有规则、所有策略一会儿在算价格一会儿又在做取消操作用户中途插一句“等等先别取消”模型上下文更容易混乱最终结果要么不敢动只给建议要么动错用户信任直接崩掉。2解决方案LangGraph 多智能体的核心设计我更喜欢用一个类比传统单体像一个“全能客服”什么都能聊但高压场景容易手忙脚乱LangGraph 多智能体像“专业团队 智能调度中心”主智能体负责分派、兜底、把控全局子智能体各司其职航班、酒店、租车、签证……一句话概括这套系统的价值可中断的权限控制 可委派的专业分工让系统既能自动化推进又不会越权乱改。二、架构核心理解系统的“骨架”与“血液”2.1 状态管理State全局记忆体LangGraph 的关键不是“写几个 prompt”而是你要认真设计 State整个系统的记忆与控制面板。一个典型的旅行系统状态可以这样定义后面代码会给完整实现messages对话历史user_info用户偏好、证件信息、常用城市等agent_stack对话状态栈重点agent_stack为什么它是“能委派、能返回”的关键agent_stack就像浏览器的历史栈进入子智能体push记录“我从主助手进入航班助手”子智能体完成pop返回上一级上下文任何时候都能“回到主助手继续聊别的”可视化一下栈变化示意初始[main]主助手委派航班[main, flights]航班助手又需要酒店信息[main, flights, hotels]酒店完成返回[main, flights]航班完成返回[main]这就是“多智能体协作但不丢上下文”的底层支撑。2.2 工具分类机制安全与敏感的“操作开关”旅行系统里工具大体分两类安全工具只读可自动执行例如search_flights、search_hotels敏感工具修改/下单/取消必须授权例如cancel_hotel、update_ticket设计意义很直白在“自动化效率”与“用户控制权”之间找到平衡。你会发现只要把敏感操作统一做成“可中断”用户的信任感会明显提升系统不再是“擅自帮你取消”而是“我准备这么做你点头我再执行”。三、智能体协作机制系统如何“思考”与“流转”3.1 主智能体调度中心主智能体的任务不是“专业处理一切”而是识别意图航班酒店租车签证判断是否需要委派简单问题自己直接答比如“改签一般有哪些费用”遇到跨域任务时拆解顺序例如先改签确认再取消酒店在代码层面主智能体通常会有四类“委派工具”本质是路由信号ToFlightAssistantToHotelAssistantToCarAssistantToVisaAssistant扩展时加3.2 子智能体的专业化设计以航班为例航班助手只需要关心航班相关工具安全search_flights查余票、价格敏感update_ticket、cancel_ticket触发授权专业化带来的好处很实际prompt 更短、更精准工具集更小模型更不容易“拿错工具”token 消耗更可控尤其是多轮对话场景3.3 核心流转机制剖析入口节点Entry Node智能体切换时的“上下文生成器”你可以把create_entry_node理解为“当我把任务丢给子助手时自动生成一段系统消息交代背景和目标”。它解决的问题是主助手和子助手之间不需要靠“人肉复制粘贴上下文”而是通过结构化状态自动串起来。CompleteOrEscalate子智能体的“完成/向上移交”协议子智能体处理时会遇到两种情况完成需求满足返回主助手移交遇到策略冲突或超出权限例如用户要求违反政策需要回主助手统一协调这相当于给多智能体协作定义了“收口协议”系统不会在子助手里无限兜圈。四、关键实现如何构建“可中断”的权限流程4.1 敏感工具的中断-授权流程流程拆开看其实很清晰用户提出修改请求“帮我取消酒店”子智能体准备调用cancel_hotel敏感LangGraph 触发interrupt暂停执行系统把“待执行操作详情”展示给用户请确认用户确认继续执行用户拒绝终止/返回核心是工具不是不能自动调用而是调用前必须过“授权闸门”。4.2 错误处理与回退现实世界里工具会失败酒店已过取消时间订单号不存在网络超时好的体验应该是给用户可理解的失败原因agent_stack不要乱失败也要能回退到正确上下文允许用户选择下一步换方案/联系客服/只保留建议五、扩展性设计如何添加新智能体以签证为例扩展到“签证办理智能体”一般只需要增量定义新助手与工具如check_visa_requirements、prepare_visa_documents主助手新增一条委派路由ToVisaAssistant状态结构如需新增字段再加多数情况下不需要这也是多智能体架构真正“能长期演进”的地方你不会因为加一个新模块把整个系统 prompt 写崩。六、完整代码示例可运行骨架包含中断授权下面这份代码是一个“可跑的最小系统骨架”具备State(messages/user_info/agent_stack)具备主助手 航班/酒店两个子助手具备敏感工具的interrupt授权流程用“伪工具实现”模拟真实 API你接入自己的供应商接口即可依赖langgraph、langchain-core、langchain-openai环境变量OPENAI_API_KEY把下面保存为multi_agent_travel.pyfrom __future__ import annotations import osfrom dataclasses import dataclassfrom typing import Any, Dict, List, Literal, Optional, TypedDictfrom langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessagefrom langchain_core.tools import toolfrom langchain_openai import ChatOpenAIfrom langgraph.graph import StateGraph, ENDfrom langgraph.types import interrupt# ----------------------------# 1) State: 全局记忆体# ----------------------------class UserInfo(TypedDict, totalFalse): name: str loyalty_id: str preferences: Dict[str, Any]class TravelState(TypedDict): messages: List[Any] user_info: UserInfo agent_stack: List[Literal[main, flights, hotels, cars, visa]] pending_action: Optional[Dict[str, Any]] # 存放待授权的敏感操作# ----------------------------# 2) 工具安全 vs 敏感# ----------------------------tooldef search_flights(origin: str, destination: str, date: str) - Dict[str, Any]: 安全工具查询航班只读 # 真实场景接供应商 API这里用模拟数据 return { origin: origin, destination: destination, date: date, options: [ {flight_no: MU123, depart: 09:10, arrive: 13:40, price: 2100}, {flight_no: NH008, depart: 12:00, arrive: 16:30, price: 2850}, ], }tooldef search_hotels(city: str, checkin: str, checkout: str) - Dict[str, Any]: 安全工具查询酒店只读 return { city: city, checkin: checkin, checkout: checkout, options: [ {hotel: Shinjuku Stay, price_per_night: 980, refundable: True}, {hotel: Ginza Business, price_per_night: 1180, refundable: False}, ], }tooldef cancel_hotel(order_id: str) - Dict[str, Any]: 敏感工具取消酒店会修改订单 # 真实场景调用取消接口这里模拟成功 return {order_id: order_id, status: cancelled, refund: 1200}tooldef update_ticket(ticket_id: str, new_flight_no: str) - Dict[str, Any]: 敏感工具改签机票会修改订单 return {ticket_id: ticket_id, new_flight_no: new_flight_no, status: rebooked}SAFE_TOOLS [search_flights, search_hotels]SENSITIVE_TOOLS [cancel_hotel, update_ticket]# ----------------------------# 3) 委派信号路由工具# ----------------------------tooldef ToFlightAssistant() - str: 委派到航班助手 return route:flightstooldef ToHotelAssistant() - str: 委派到酒店助手 return route:hotelsROUTING_TOOLS [ToFlightAssistant, ToHotelAssistant]# ----------------------------# 4) LLM# ----------------------------def make_llm() - ChatOpenAI: return ChatOpenAI( modelgpt-4o-mini, temperature0.2, )# ----------------------------# 5) Entry Node: 切换上下文# ----------------------------def create_entry_node(target: Literal[flights, hotels]) - Any: def _entry(state: TravelState) - TravelState: stack state[agent_stack] stack.append(target) background ( 你是旅行系统里的专业助手。/n f你现在负责{target}。/n 你会收到主助手已有的对话上下文请直接推进任务。/n 如果任务完成返回 CompleteOrEscalate 工具并说明原因。/n 如果需要执行敏感操作取消/改签先触发授权流程不要擅自执行。 ) return { state, messages: state[messages] [SystemMessage(contentbackground)], agent_stack: stack, } return _entrytooldef CompleteOrEscalate(reason: str) - str: 子助手完成或需要上移交时调用 return fcomplete:{reason}# ----------------------------# 6) 权限中断敏感工具网关# ----------------------------def sensitive_tool_gateway(state: TravelState) - TravelState: 如果 pending_action 存在则触发 interrupt 请求用户授权。 用户同意后继续执行用户拒绝则清空并返回上级。 pending state.get(pending_action) if not pending: return state # 触发中断把将要执行的动作展示给用户 decision interrupt( { type: approval_required, title: 需要确认敏感操作, action: pending, prompt: 请回复确认 / 取消, } ) # decision 由外部 runner 注入见 main 中的示例 if isinstance(decision, dict) and decision.get(approved) is True: tool_name pending[tool] args pending[args] # 执行真实敏感工具 if tool_name cancel_hotel: result cancel_hotel.invoke(args) elif tool_name update_ticket: result update_ticket.invoke(args) else: result {error: funknown sensitive tool: {tool_name}} return { state, pending_action: None, messages: state[messages] [ToolMessage(contentstr(result), tool_call_idpending[tool_call_id])], } # 用户拒绝清空 pending_action并回到主助手 stack state[agent_stack] if len(stack) 1: stack.pop() return { state, pending_action: None, agent_stack: stack, messages: state[messages] [AIMessage(content好的我不会执行该操作。我们回到主流程继续。)], }# ----------------------------# 7) 各智能体主/航班/酒店# ----------------------------def main_assistant(state: TravelState) - TravelState: llm make_llm().bind_tools(ROUTING_TOOLS) sys SystemMessage( content( 你是旅行规划系统的主助手调度中心。/n 职责识别用户意图必要时委派到子助手航班/酒店简单问题直接回答。/n 当用户提出需要取消/改签等修改动作时引导到对应子助手处理并强调需要用户确认。/n ) ) msgs [sys] state[messages] out llm.invoke(msgs) return {state, messages: state[messages] [out]}def flight_assistant(state: TravelState) - TravelState: llm make_llm().bind_tools([*SAFE_TOOLS, *SENSITIVE_TOOLS, CompleteOrEscalate]) sys SystemMessage( content( 你是航班助手。/n 你可以用 search_flights 查询如果要改签(update_ticket)或退票等敏感操作必须先走授权。/n 当你准备执行敏感工具时不要直接调用工具请在 state.pending_action 写入待执行信息并让系统走授权网关。/n ) ) out llm.invoke([sys] state[messages]) # 如果模型直接发起敏感 tool call拦截并转 pending_action if hasattr(out, tool_calls) and out.tool_calls: for tc in out.tool_calls: name tc[name] if name in {update_ticket}: return { state, messages: state[messages] [out], pending_action: { tool: name, args: tc[args], tool_call_id: tc[id], }, } return {state, messages: state[messages] [out]}def hotel_assistant(state: TravelState) - TravelState: llm make_llm().bind_tools([*SAFE_TOOLS, *SENSITIVE_TOOLS, CompleteOrEscalate]) sys SystemMessage( content( 你是酒店助手。/n 你可以用 search_hotels 查询如果要取消订单(cancel_hotel)等敏感操作必须先走授权。/n 当你准备执行敏感工具时不要直接调用工具请在 state.pending_action 写入待执行信息并让系统走授权网关。/n ) ) out llm.invoke([sys] state[messages]) if hasattr(out, tool_calls) and out.tool_calls: for tc in out.tool_calls: name tc[name] if name in {cancel_hotel}: return { state, messages: state[messages] [out], pending_action: { tool: name, args: tc[args], tool_call_id: tc[id], }, } return {state, messages: state[messages] [out]}# ----------------------------# 8) Graph路由与返回# ----------------------------def route_from_main(state: TravelState) - str: last state[messages][-1] if hasattr(last, tool_calls) and last.tool_calls: name last.tool_calls[0][name] if name ToFlightAssistant: return to_flights if name ToHotelAssistant: return to_hotels return end_or_waitdef route_after_child(state: TravelState) - str: # 如果有待授权敏感操作先走网关 if state.get(pending_action): return sensitive_gateway # 如果子助手显式 CompleteOrEscalate则返回主助手 last state[messages][-1] if hasattr(last, tool_calls) and last.tool_calls: for tc in last.tool_calls: if tc[name] CompleteOrEscalate: # pop 子助手 stack state[agent_stack] if len(stack) 1: stack.pop() state[agent_stack] stack return back_to_main # 否则继续留在当前子助手 return staydef build_graph(): g StateGraph(TravelState) g.add_node(main, main_assistant) g.add_node(entry_flights, create_entry_node(flights)) g.add_node(entry_hotels, create_entry_node(hotels)) g.add_node(flights, flight_assistant) g.add_node(hotels, hotel_assistant) g.add_node(sensitive_gateway, sensitive_tool_gateway) g.set_entry_point(main) g.add_conditional_edges( main, route_from_main, { to_flights: entry_flights, to_hotels: entry_hotels, end_or_wait: END, }, ) g.add_edge(entry_flights, flights) g.add_edge(entry_hotels, hotels) g.add_conditional_edges( flights, route_after_child, {sensitive_gateway: sensitive_gateway, back_to_main: main, stay: flights}, ) g.add_conditional_edges( hotels, route_after_child, {sensitive_gateway: sensitive_gateway, back_to_main: main, stay: hotels}, ) # 授权网关执行完回到当前栈顶对应助手 def after_gateway_route(state: TravelState) - str: top state[agent_stack][-1] return {main: main, flights: flights, hotels: hotels}.get(top, main) g.add_conditional_edges( sensitive_gateway, after_gateway_route, {main: main, flights: flights, hotels: hotels}, ) return g.compile()# ----------------------------# 9) 本地演示 Runner# ----------------------------def run_demo(): graph build_graph() state: TravelState { messages: [], user_info: {name: Alex, preferences: {hotel_budget: 1200}}, agent_stack: [main], pending_action: None, } print(输入 exit 退出。) while True: text input(/n你).strip() if text.lower() exit: break state[messages].append(HumanMessage(contenttext)) # invoke 可能触发 interrupt这里用 try/except 演示一个“简化版人工审批” try: state graph.invoke(state) except Exception as e: # LangGraph 的 interrupt 在不同版本里表现略有差异 # 这里用一个通用思路检测到中断后人工收集确认并继续。 # 如果你的版本提供更直接的中断捕获方式可按官方文档替换。 print(f/n[系统中断] {e}) # 简化读取 pending_action 来展示 pending state.get(pending_action) if pending: print(f待执行{pending[tool]} args{pending[args]}) decision input(确认执行(确认/取消)).strip() approved decision 确认 # 将 decision 以可被 interrupt 消费的形式写回示例做法 # 实战中建议用 checkpointer / stream API 处理更优雅 state[messages].append( HumanMessage(contentf审批结果{同意 if approved else 拒绝}) ) # 这里直接调用网关节点 state sensitive_tool_gateway({state, pending_action: pending}) state graph.invoke(state) # 打印最后一条 AI 输出 for m in state[messages][-3:]: if isinstance(m, AIMessage): print(f/n助手{m.content})if __name__ __main__: if not os.getenv(OPENAI_API_KEY): raise RuntimeError(请先设置环境变量 OPENAI_API_KEY) run_demo()运行方式pip install langgraph langchain-core langchain-openaiexport OPENAI_API_KEY你的keypython multi_agent_travel.py你可以试试输入“帮我查一下长沙到北京2026-01-10 的航班”“把我的机票改签到 NH008”“取消酒店订单 H10086”会触发授权中断七、总结多智能体系统听起来挺复杂但本质就是分工合作这个古老智慧的数字化实现。LangGraph提供了很好的框架让我们能够把这种协作逻辑优雅地表达出来。这套“可中断、可委派”的多智能体旅行系统核心优势很明确专业化每个子智能体只做一类事提示词更准、工具更少、成本更可控权限控制敏感操作统一中断授权用户放心系统也更安全模块化扩展加一个“签证助手”就是加一组工具 一条路由不用重写全局如何学习大模型 AI 由于新岗位的生产效率要优于被取代岗位的生产效率所以实际上整个社会的生产效率是提升的。但是具体到个人只能说是“最先掌握AI的人将会比较晚掌握AI的人有竞争优势”。这句话放在计算机、互联网、移动互联网的开局时期都是一样的道理。我在一线科技企业深耕十二载见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事早已在效率与薪资上形成代际优势我意识到有很多经验和知识值得分享给大家也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。我们整理出这套AI 大模型突围资料包✅ 从零到一的 AI 学习路径图✅ 大模型调优实战手册附医疗/金融等大厂真实案例✅ 百度/阿里专家闭门录播课✅ 大模型当下最新行业报告✅ 真实大厂面试真题✅ 2026 最新岗位需求图谱所有资料 ⚡️ 朋友们如果有需要《AI大模型入门进阶学习资源包》下方扫码获取~① 全套AI大模型应用开发视频教程包含提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点② 大模型系统化学习路线作为学习AI大模型技术的新手方向至关重要。 正确的学习路线可以为你节省时间少走弯路方向不对努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划带你从零基础入门到精通③ 大模型学习书籍文档学习AI大模型离不开书籍文档我精选了一系列大模型技术的书籍和学习文档电子版它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。④ AI大模型最新行业报告2025最新行业报告针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。⑤ 大模型项目实战配套源码学以致用在项目实战中检验和巩固你所学到的知识同时为你找工作就业和职业发展打下坚实的基础。⑥ 大模型大厂面试真题面试不仅是技术的较量更需要充分的准备。在你已经掌握了大模型技术之后就需要开始准备面试我精心整理了一份大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。以上资料如何领取为什么大家都在学大模型最近科技巨头英特尔宣布裁员2万人传统岗位不断缩减但AI相关技术岗疯狂扩招有3-5年经验大厂薪资就能给到50K*20薪不出1年“有AI项目经验”将成为投递简历的门槛。风口之下与其像“温水煮青蛙”一样坐等被行业淘汰不如先人一步掌握AI大模型原理应用技术项目实操经验“顺风”翻盘这些资料真的有用吗这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理现任上海殷泊信息科技CEO其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证服务航天科工、国家电网等1000企业以第一作者在IEEE Transactions发表论文50篇获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的技术人员这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。以上全套大模型资料如何领取 小白程序员也能学会：LangGraph搭建可中断、可委派的多智能体旅行系统
文章探讨了单一智能体在处理复杂旅行规划任务时的局限性提出了使用LangGraph搭建多智能体系统的解决方案。该系统通过主智能体和子智能体的分工合作实现了可中断的权限控制和可委派的专业分工有效解决了多步骤、多工具和有风险操作的组合任务问题。文章详细介绍了系统的架构设计包括状态管理、工具分类机制、智能体协作机制以及关键实现等并提供了一份可直接运行的完整代码骨架帮助读者理解和实践多智能体系统的构建。开会前十分钟你同事突然丢来一句“我下周去北京帮我把行程改一下——航班改签、旧酒店取消、换一家离新会议地点近的、再加个租车。”你脑子里已经开始自动列 checklist 了先查余票再看退改规则再核对酒店取消政策再对比新酒店位置和预算还要确认租车取还车时间……而很多“传统对话式 AI”的尴尬就在这它会聊天但一遇到这种“多步骤 多工具 有风险操作”的组合任务就容易把事情越办越乱。这篇文章我想从“为什么单一智能体扛不住”讲到“怎么用 LangGraph 搭一个可中断、可委派的多智能体旅行系统”最后给你一份可以直接跑的完整代码骨架包含权限中断流程。一、当单一AI无法应对复杂旅行场景1痛点复杂旅行规划不是“回答问题”而是“执行流程”想象这个场景用户帮我改签航班然后取消旧酒店再订新酒店最后租车。这里面至少包含四类能力搜索查航班、查酒店、查租车多为只读策略什么时候先改签、什么时候先取消避免损失规则退改签/取消政策、时间冲突、预算约束执行真正“改”“退”“订”这类动作往往不可逆如果你用单一智能体硬扛通常会变成这样它必须同时掌握所有工具、所有规则、所有策略一会儿在算价格一会儿又在做取消操作用户中途插一句“等等先别取消”模型上下文更容易混乱最终结果要么不敢动只给建议要么动错用户信任直接崩掉。2解决方案LangGraph 多智能体的核心设计我更喜欢用一个类比传统单体像一个“全能客服”什么都能聊但高压场景容易手忙脚乱LangGraph 多智能体像“专业团队 智能调度中心”主智能体负责分派、兜底、把控全局子智能体各司其职航班、酒店、租车、签证……一句话概括这套系统的价值可中断的权限控制 可委派的专业分工让系统既能自动化推进又不会越权乱改。二、架构核心理解系统的“骨架”与“血液”2.1 状态管理State全局记忆体LangGraph 的关键不是“写几个 prompt”而是你要认真设计 State整个系统的记忆与控制面板。一个典型的旅行系统状态可以这样定义后面代码会给完整实现messages对话历史user_info用户偏好、证件信息、常用城市等agent_stack对话状态栈重点agent_stack为什么它是“能委派、能返回”的关键agent_stack就像浏览器的历史栈进入子智能体push记录“我从主助手进入航班助手”子智能体完成pop返回上一级上下文任何时候都能“回到主助手继续聊别的”可视化一下栈变化示意初始[main]主助手委派航班[main, flights]航班助手又需要酒店信息[main, flights, hotels]酒店完成返回[main, flights]航班完成返回[main]这就是“多智能体协作但不丢上下文”的底层支撑。2.2 工具分类机制安全与敏感的“操作开关”旅行系统里工具大体分两类安全工具只读可自动执行例如search_flights、search_hotels敏感工具修改/下单/取消必须授权例如cancel_hotel、update_ticket设计意义很直白在“自动化效率”与“用户控制权”之间找到平衡。你会发现只要把敏感操作统一做成“可中断”用户的信任感会明显提升系统不再是“擅自帮你取消”而是“我准备这么做你点头我再执行”。三、智能体协作机制系统如何“思考”与“流转”3.1 主智能体调度中心主智能体的任务不是“专业处理一切”而是识别意图航班酒店租车签证判断是否需要委派简单问题自己直接答比如“改签一般有哪些费用”遇到跨域任务时拆解顺序例如先改签确认再取消酒店在代码层面主智能体通常会有四类“委派工具”本质是路由信号ToFlightAssistantToHotelAssistantToCarAssistantToVisaAssistant扩展时加3.2 子智能体的专业化设计以航班为例航班助手只需要关心航班相关工具安全search_flights查余票、价格敏感update_ticket、cancel_ticket触发授权专业化带来的好处很实际prompt 更短、更精准工具集更小模型更不容易“拿错工具”token 消耗更可控尤其是多轮对话场景3.3 核心流转机制剖析入口节点Entry Node智能体切换时的“上下文生成器”你可以把create_entry_node理解为“当我把任务丢给子助手时自动生成一段系统消息交代背景和目标”。它解决的问题是主助手和子助手之间不需要靠“人肉复制粘贴上下文”而是通过结构化状态自动串起来。CompleteOrEscalate子智能体的“完成/向上移交”协议子智能体处理时会遇到两种情况完成需求满足返回主助手移交遇到策略冲突或超出权限例如用户要求违反政策需要回主助手统一协调这相当于给多智能体协作定义了“收口协议”系统不会在子助手里无限兜圈。四、关键实现如何构建“可中断”的权限流程4.1 敏感工具的中断-授权流程流程拆开看其实很清晰用户提出修改请求“帮我取消酒店”子智能体准备调用cancel_hotel敏感LangGraph 触发interrupt暂停执行系统把“待执行操作详情”展示给用户请确认用户确认继续执行用户拒绝终止/返回核心是工具不是不能自动调用而是调用前必须过“授权闸门”。4.2 错误处理与回退现实世界里工具会失败酒店已过取消时间订单号不存在网络超时好的体验应该是给用户可理解的失败原因agent_stack不要乱失败也要能回退到正确上下文允许用户选择下一步换方案/联系客服/只保留建议五、扩展性设计如何添加新智能体以签证为例扩展到“签证办理智能体”一般只需要增量定义新助手与工具如check_visa_requirements、prepare_visa_documents主助手新增一条委派路由ToVisaAssistant状态结构如需新增字段再加多数情况下不需要这也是多智能体架构真正“能长期演进”的地方你不会因为加一个新模块把整个系统 prompt 写崩。六、完整代码示例可运行骨架包含中断授权下面这份代码是一个“可跑的最小系统骨架”具备State(messages/user_info/agent_stack)具备主助手 航班/酒店两个子助手具备敏感工具的interrupt授权流程用“伪工具实现”模拟真实 API你接入自己的供应商接口即可依赖langgraph、langchain-core、langchain-openai环境变量OPENAI_API_KEY把下面保存为multi_agent_travel.pyfrom __future__ import annotations import osfrom dataclasses import dataclassfrom typing import Any, Dict, List, Literal, Optional, TypedDictfrom langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessagefrom langchain_core.tools import toolfrom langchain_openai import ChatOpenAIfrom langgraph.graph import StateGraph, ENDfrom langgraph.types import interrupt# ----------------------------# 1) State: 全局记忆体# ----------------------------class UserInfo(TypedDict, totalFalse): name: str loyalty_id: str preferences: Dict[str, Any]class TravelState(TypedDict): messages: List[Any] user_info: UserInfo agent_stack: List[Literal[main, flights, hotels, cars, visa]] pending_action: Optional[Dict[str, Any]] # 存放待授权的敏感操作# ----------------------------# 2) 工具安全 vs 敏感# ----------------------------tooldef search_flights(origin: str, destination: str, date: str) - Dict[str, Any]: 安全工具查询航班只读 # 真实场景接供应商 API这里用模拟数据 return { origin: origin, destination: destination, date: date, options: [ {flight_no: MU123, depart: 09:10, arrive: 13:40, price: 2100}, {flight_no: NH008, depart: 12:00, arrive: 16:30, price: 2850}, ], }tooldef search_hotels(city: str, checkin: str, checkout: str) - Dict[str, Any]: 安全工具查询酒店只读 return { city: city, checkin: checkin, checkout: checkout, options: [ {hotel: Shinjuku Stay, price_per_night: 980, refundable: True}, {hotel: Ginza Business, price_per_night: 1180, refundable: False}, ], }tooldef cancel_hotel(order_id: str) - Dict[str, Any]: 敏感工具取消酒店会修改订单 # 真实场景调用取消接口这里模拟成功 return {order_id: order_id, status: cancelled, refund: 1200}tooldef update_ticket(ticket_id: str, new_flight_no: str) - Dict[str, Any]: 敏感工具改签机票会修改订单 return {ticket_id: ticket_id, new_flight_no: new_flight_no, status: rebooked}SAFE_TOOLS [search_flights, search_hotels]SENSITIVE_TOOLS [cancel_hotel, update_ticket]# ----------------------------# 3) 委派信号路由工具# ----------------------------tooldef ToFlightAssistant() - str: 委派到航班助手 return route:flightstooldef ToHotelAssistant() - str: 委派到酒店助手 return route:hotelsROUTING_TOOLS [ToFlightAssistant, ToHotelAssistant]# ----------------------------# 4) LLM# ----------------------------def make_llm() - ChatOpenAI: return ChatOpenAI( modelgpt-4o-mini, temperature0.2, )# ----------------------------# 5) Entry Node: 切换上下文# ----------------------------def create_entry_node(target: Literal[flights, hotels]) - Any: def _entry(state: TravelState) - TravelState: stack state[agent_stack] stack.append(target) background ( 你是旅行系统里的专业助手。/n f你现在负责{target}。/n 你会收到主助手已有的对话上下文请直接推进任务。/n 如果任务完成返回 CompleteOrEscalate 工具并说明原因。/n 如果需要执行敏感操作取消/改签先触发授权流程不要擅自执行。 ) return { state, messages: state[messages] [SystemMessage(contentbackground)], agent_stack: stack, } return _entrytooldef CompleteOrEscalate(reason: str) - str: 子助手完成或需要上移交时调用 return fcomplete:{reason}# ----------------------------# 6) 权限中断敏感工具网关# ----------------------------def sensitive_tool_gateway(state: TravelState) - TravelState: 如果 pending_action 存在则触发 interrupt 请求用户授权。 用户同意后继续执行用户拒绝则清空并返回上级。 pending state.get(pending_action) if not pending: return state # 触发中断把将要执行的动作展示给用户 decision interrupt( { type: approval_required, title: 需要确认敏感操作, action: pending, prompt: 请回复确认 / 取消, } ) # decision 由外部 runner 注入见 main 中的示例 if isinstance(decision, dict) and decision.get(approved) is True: tool_name pending[tool] args pending[args] # 执行真实敏感工具 if tool_name cancel_hotel: result cancel_hotel.invoke(args) elif tool_name update_ticket: result update_ticket.invoke(args) else: result {error: funknown sensitive tool: {tool_name}} return { state, pending_action: None, messages: state[messages] [ToolMessage(contentstr(result), tool_call_idpending[tool_call_id])], } # 用户拒绝清空 pending_action并回到主助手 stack state[agent_stack] if len(stack) 1: stack.pop() return { state, pending_action: None, agent_stack: stack, messages: state[messages] [AIMessage(content好的我不会执行该操作。我们回到主流程继续。)], }# ----------------------------# 7) 各智能体主/航班/酒店# ----------------------------def main_assistant(state: TravelState) - TravelState: llm make_llm().bind_tools(ROUTING_TOOLS) sys SystemMessage( content( 你是旅行规划系统的主助手调度中心。/n 职责识别用户意图必要时委派到子助手航班/酒店简单问题直接回答。/n 当用户提出需要取消/改签等修改动作时引导到对应子助手处理并强调需要用户确认。/n ) ) msgs [sys] state[messages] out llm.invoke(msgs) return {state, messages: state[messages] [out]}def flight_assistant(state: TravelState) - TravelState: llm make_llm().bind_tools([*SAFE_TOOLS, *SENSITIVE_TOOLS, CompleteOrEscalate]) sys SystemMessage( content( 你是航班助手。/n 你可以用 search_flights 查询如果要改签(update_ticket)或退票等敏感操作必须先走授权。/n 当你准备执行敏感工具时不要直接调用工具请在 state.pending_action 写入待执行信息并让系统走授权网关。/n ) ) out llm.invoke([sys] state[messages]) # 如果模型直接发起敏感 tool call拦截并转 pending_action if hasattr(out, tool_calls) and out.tool_calls: for tc in out.tool_calls: name tc[name] if name in {update_ticket}: return { state, messages: state[messages] [out], pending_action: { tool: name, args: tc[args], tool_call_id: tc[id], }, } return {state, messages: state[messages] [out]}def hotel_assistant(state: TravelState) - TravelState: llm make_llm().bind_tools([*SAFE_TOOLS, *SENSITIVE_TOOLS, CompleteOrEscalate]) sys SystemMessage( content( 你是酒店助手。/n 你可以用 search_hotels 查询如果要取消订单(cancel_hotel)等敏感操作必须先走授权。/n 当你准备执行敏感工具时不要直接调用工具请在 state.pending_action 写入待执行信息并让系统走授权网关。/n ) ) out llm.invoke([sys] state[messages]) if hasattr(out, tool_calls) and out.tool_calls: for tc in out.tool_calls: name tc[name] if name in {cancel_hotel}: return { state, messages: state[messages] [out], pending_action: { tool: name, args: tc[args], tool_call_id: tc[id], }, } return {state, messages: state[messages] [out]}# ----------------------------# 8) Graph路由与返回# ----------------------------def route_from_main(state: TravelState) - str: last state[messages][-1] if hasattr(last, tool_calls) and last.tool_calls: name last.tool_calls[0][name] if name ToFlightAssistant: return to_flights if name ToHotelAssistant: return to_hotels return end_or_waitdef route_after_child(state: TravelState) - str: # 如果有待授权敏感操作先走网关 if state.get(pending_action): return sensitive_gateway # 如果子助手显式 CompleteOrEscalate则返回主助手 last state[messages][-1] if hasattr(last, tool_calls) and last.tool_calls: for tc in last.tool_calls: if tc[name] CompleteOrEscalate: # pop 子助手 stack state[agent_stack] if len(stack) 1: stack.pop() state[agent_stack] stack return back_to_main # 否则继续留在当前子助手 return staydef build_graph(): g StateGraph(TravelState) g.add_node(main, main_assistant) g.add_node(entry_flights, create_entry_node(flights)) g.add_node(entry_hotels, create_entry_node(hotels)) g.add_node(flights, flight_assistant) g.add_node(hotels, hotel_assistant) g.add_node(sensitive_gateway, sensitive_tool_gateway) g.set_entry_point(main) g.add_conditional_edges( main, route_from_main, { to_flights: entry_flights, to_hotels: entry_hotels, end_or_wait: END, }, ) g.add_edge(entry_flights, flights) g.add_edge(entry_hotels, hotels) g.add_conditional_edges( flights, route_after_child, {sensitive_gateway: sensitive_gateway, back_to_main: main, stay: flights}, ) g.add_conditional_edges( hotels, route_after_child, {sensitive_gateway: sensitive_gateway, back_to_main: main, stay: hotels}, ) # 授权网关执行完回到当前栈顶对应助手 def after_gateway_route(state: TravelState) - str: top state[agent_stack][-1] return {main: main, flights: flights, hotels: hotels}.get(top, main) g.add_conditional_edges( sensitive_gateway, after_gateway_route, {main: main, flights: flights, hotels: hotels}, ) return g.compile()# ----------------------------# 9) 本地演示 Runner# ----------------------------def run_demo(): graph build_graph() state: TravelState { messages: [], user_info: {name: Alex, preferences: {hotel_budget: 1200}}, agent_stack: [main], pending_action: None, } print(输入 exit 退出。) while True: text input(/n你).strip() if text.lower() exit: break state[messages].append(HumanMessage(contenttext)) # invoke 可能触发 interrupt这里用 try/except 演示一个“简化版人工审批” try: state graph.invoke(state) except Exception as e: # LangGraph 的 interrupt 在不同版本里表现略有差异 # 这里用一个通用思路检测到中断后人工收集确认并继续。 # 如果你的版本提供更直接的中断捕获方式可按官方文档替换。 print(f/n[系统中断] {e}) # 简化读取 pending_action 来展示 pending state.get(pending_action) if pending: print(f待执行{pending[tool]} args{pending[args]}) decision input(确认执行(确认/取消)).strip() approved decision 确认 # 将 decision 以可被 interrupt 消费的形式写回示例做法 # 实战中建议用 checkpointer / stream API 处理更优雅 state[messages].append( HumanMessage(contentf审批结果{同意 if approved else 拒绝}) ) # 这里直接调用网关节点 state sensitive_tool_gateway({state, pending_action: pending}) state graph.invoke(state) # 打印最后一条 AI 输出 for m in state[messages][-3:]: if isinstance(m, AIMessage): print(f/n助手{m.content})if __name__ __main__: if not os.getenv(OPENAI_API_KEY): raise RuntimeError(请先设置环境变量 OPENAI_API_KEY) run_demo()运行方式pip install langgraph langchain-core langchain-openaiexport OPENAI_API_KEY你的keypython multi_agent_travel.py你可以试试输入“帮我查一下长沙到北京2026-01-10 的航班”“把我的机票改签到 NH008”“取消酒店订单 H10086”会触发授权中断七、总结多智能体系统听起来挺复杂但本质就是分工合作这个古老智慧的数字化实现。LangGraph提供了很好的框架让我们能够把这种协作逻辑优雅地表达出来。这套“可中断、可委派”的多智能体旅行系统核心优势很明确专业化每个子智能体只做一类事提示词更准、工具更少、成本更可控权限控制敏感操作统一中断授权用户放心系统也更安全模块化扩展加一个“签证助手”就是加一组工具 一条路由不用重写全局如何学习大模型 AI 由于新岗位的生产效率要优于被取代岗位的生产效率所以实际上整个社会的生产效率是提升的。但是具体到个人只能说是“最先掌握AI的人将会比较晚掌握AI的人有竞争优势”。这句话放在计算机、互联网、移动互联网的开局时期都是一样的道理。我在一线科技企业深耕十二载见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事早已在效率与薪资上形成代际优势我意识到有很多经验和知识值得分享给大家也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。我们整理出这套AI 大模型突围资料包✅ 从零到一的 AI 学习路径图✅ 大模型调优实战手册附医疗/金融等大厂真实案例✅ 百度/阿里专家闭门录播课✅ 大模型当下最新行业报告✅ 真实大厂面试真题✅ 2026 最新岗位需求图谱所有资料 ⚡️ 朋友们如果有需要《AI大模型入门进阶学习资源包》下方扫码获取~① 全套AI大模型应用开发视频教程包含提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点② 大模型系统化学习路线作为学习AI大模型技术的新手方向至关重要。 正确的学习路线可以为你节省时间少走弯路方向不对努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划带你从零基础入门到精通③ 大模型学习书籍文档学习AI大模型离不开书籍文档我精选了一系列大模型技术的书籍和学习文档电子版它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。④ AI大模型最新行业报告2025最新行业报告针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。⑤ 大模型项目实战配套源码学以致用在项目实战中检验和巩固你所学到的知识同时为你找工作就业和职业发展打下坚实的基础。⑥ 大模型大厂面试真题面试不仅是技术的较量更需要充分的准备。在你已经掌握了大模型技术之后就需要开始准备面试我精心整理了一份大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。以上资料如何领取为什么大家都在学大模型最近科技巨头英特尔宣布裁员2万人传统岗位不断缩减但AI相关技术岗疯狂扩招有3-5年经验大厂薪资就能给到50K*20薪不出1年“有AI项目经验”将成为投递简历的门槛。风口之下与其像“温水煮青蛙”一样坐等被行业淘汰不如先人一步掌握AI大模型原理应用技术项目实操经验“顺风”翻盘这些资料真的有用吗这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理现任上海殷泊信息科技CEO其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证服务航天科工、国家电网等1000企业以第一作者在IEEE Transactions发表论文50篇获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的技术人员这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。以上全套大模型资料如何领取