多智能体系统Multi-Agent SystemMAS是由多个具备感知、决策与行动能力的自主智能体通过通信交互与协同机制协作或竞争完成复杂任务的分布式 AI 系统核心是用群体智能突破单智能体能力边界适配大规模、多目标、强动态的场景。一、LangChain中多智能体应用简介根据LangChain 1.0官方文档介绍多智能体主要是用来解决复杂的工作流而且并不是每个复杂任务都需要通过多智能体来解决很多复杂的任务其实可以正确使用提示词和工具来解决。一般来说以下三种情况会用到多智能体技术上下文管理当上下文过多时在不占用模型上下文窗口过多资源的前提下提供专业化知识。通过构建多个智能体获取更多的上下文知识。分布式开发支持不同团队独立开发和维护各项功能并将其整合为一个边界清晰的大型系统。并行化处理为子任务分配专用的执行单元通过并行执行提升任务处理效率。官网将多智能体的实现方式分成了以下五种模式Subagents主智能体将子智能体作为工具进行协同调度所有任务路由均由主智能体统一管控由其决定调用各子智能体的时机与方式。Subagents的关键特性有如下几点集中式控制所有任务路由均由主智能体统一处理无直接用户交互子智能体将结果反馈至主智能体而非直接对接用户不过可在子智能体中设置中断机制以支持用户交互工具化调用子智能体子智能体通过工具实现调用并行执行主智能体可在单次交互轮次中调用多个子智能体总体架构图以购买咖啡为例它的时序图如下Handoffs行为根据系统状态动态调整。工具调用操作会更新一个状态变量该变量触发任务路由或配置变更进而完成智能体切换或调整当前智能体的关联工具与提示词。智能体之间通过工具调用实现控制权移交。每个智能体既可以将控制权交接给其他智能体也可以直接向用户反馈结果。在这种架构中系统行为会基于状态动态调整。其核心机制为工具调用会更新一个可跨交互轮次持久化的状态变量例如current_step当前步骤或active_agent活跃智能体系统读取该变量后对自身行为进行调整 —— 要么加载不同的配置信息系统提示词、工具集要么将任务路由至其他智能体。这种模式既支持不同智能体之间的任务交接也可实现单个智能体内部的动态配置变更。Handoffs的关键特性有以下几点状态驱动行为行为基于状态变量进行调整工具触发状态切换通过工具调用更新状态变量以实现不同状态间的流转直接用户交互各状态对应的配置可直接处理用户消息状态持久化状态可跨对话轮次保留总体架构图以购买咖啡为例它的时序图如下Skills这种架构下各类专用能力被封装为可调用的 “技能组件”以此扩展智能体的行为边界。这类技能组件本质上是由提示词驱动的专用模块可供智能体根据需求灵活调用。Skills关键特性有以下几点提示词驱动技能组件主要由专用提示词定义渐进式能力技能组件根据上下文信息或用户需求按需启用分布式开发不同团队可独立开展技能组件的开发与维护工作轻量化组合技能组件的复杂度低于完整的子智能体总体架构图以购买咖啡为例它的时序图如下Router在Router架构下系统会通过一个路由步骤对输入内容进行分类并将其分发至对应的专用智能体随后系统将各智能体返回的结果整合为一份合并后的响应内容。这种架构在处理垂直领域场景时尤为实用 —— 即当存在多个相互独立的知识域且每个知识域都需要配备专属智能体的情况。Router的关键特性如下路由模块对查询请求进行拆解并行调用零个或多个专用智能体将结果整合为连贯一致的响应内容总体架构图以购买咖啡为例它的时序图如下Custom workflow这个模式其实就是借助 LangGraph 构建定制化执行流程能够对图结构实现完全掌控包括其中的顺序执行步骤、条件分支、循环逻辑以及并行执行机制。这是完全可以自定义的模式。它的关键特性如下完全掌控图结构融合确定性逻辑与智能体自主行为支持顺序执行步骤、条件分支、循环逻辑及并行执行机制可以将其他架构模式嵌入工作流作为流程节点使用二、多智能体实际应用现在我希望构建一个搜索智能体根据用户的搜索信息确定用户的意图提取出关键词然后使用web搜索工具根据关键词搜索获取信息最后根据获取的信息组织成符合人类阅读习惯的结果返回给用户。根据上面的LangChain多智能体的构建方案我选择了两种方式来实现我的需求。方式一基于LangGraph打造工作流实现关键词获取搜索和结果反馈几个节点。方式二基于LangChain实现两个智能体一个智能体负责从用户查询中提取关键词并封装成工具web搜索也封装成工具另一个智能体调用着两个工具实现搜索信息的反馈。web搜索采用tavily来实现。tavily https://www.tavily.com/是专为 LLM 与 AI 智能体设计的企业级网络数据访问平台核心提供搜索、提取、抓取等 API能高效获取实时结构化信息并降低大模型幻觉。首先是方式一。在方式一中本质上就是基于LangGraph构建workflow流程图如下首先定义一个状态结构用于保存工作流中执行的数据和状态。同时初始化大模型和tavily客户端。# 初始化Tavily客户端 tavily_client TavilyClient(api_keyos.getenv(TAVILY_API_KEY)) # 定义状态结构 class SearchState(TypedDict): messages: Annotated[list, add_messages] user_query: str # 用户查询 search_query: str # 优化后的搜索查询 search_results: str # Tavily搜索结果 final_answer: str # 最终答案 step: str # 当前步骤 # 初始化模型和Tavily客户端 llm ChatOpenAI( modelos.getenv(LLM_MODEL_ID, gpt-4o-mini), api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL, https://api.openai.com/v1), temperature0.7 )然后基于LangGraph构建一个工作流其中提示词的设计非常重要。def understand_query_node(state: SearchState) - SearchState: 步骤1理解用户查询并生成搜索关键词 # 获取最新的用户消息 user_message for msg in reversed(state[messages]): if isinstance(msg, HumanMessage): user_message msg.content break understand_prompt f分析用户的查询{user_message} 请完成两个任务 1. 简洁总结用户想要了解什么 2. 生成最适合搜索的关键词中英文均可要精准 格式 理解[用户需求总结] 搜索词[最佳搜索关键词] response llm.invoke([SystemMessage(contentunderstand_prompt)]) # 提取搜索关键词 response_text response.content search_query user_message # 默认使用原始查询 if 搜索词 in response_text: search_query response_text.split(搜索词)[1].strip() elif 搜索关键词 in response_text: search_query response_text.split(搜索关键词)[1].strip() return { user_query: response.content, search_query: search_query, step: understood, messages: [AIMessage(contentf我理解您的需求{response.content})] } def tavily_search_node(state: SearchState) - SearchState: 步骤2使用Tavily API进行真实搜索 search_query state[search_query] try: print(f 正在搜索{search_query}) response tavily_client.search( query search_query, search_depth basic, include_answer True, include_raw_content False, max_results 5 ) # 处理搜索结果 search_results # 优先使用Tavily的综合答案 if response.get(answer): search_results f综合答案\n{response[answer]}\n\n # 添加具体的搜索结果 if response.get(results): search_results 相关信息\n for i, result in enumerate(response[results][:3], 1): title result.get(title, ) content result.get(content, ) url result.get(url, ) search_results f{i}. {title}\n{content}\n来源{url}\n\n if not search_results: search_results 抱歉没有找到相关的信息。 return { search_results: search_results, step: searched, messages: [AIMessage(contentf✅ 搜索完成找到了相关信息正在为您整理答案...)] } except Exception as e: error_msg f搜索时发生错误{str(e)} print(f❌ {error_msg}) return { search_results: error_msg, step: search_failed, messages: [AIMessage(contentf❌ 搜索遇到问题我将基于已有知识为您回答)] } def generate_answer_node(state: SearchState) - SearchState: 步骤3基于搜索结果生成最终答案 # 检查是否有搜索结果 if state[step] search_failed: # 如果搜索失败基于LLM知识问答 fallback_prompt f搜索API暂时不可用请基于您的知识回答用户的问题 用户问题{state[user_query]} 请提供一个有用的回答并说明这是基于已有知识的回答。 response llm.invoke([SystemMessage(contentfallback_prompt)]) return { final_answer: response.content, step: completed, messages: [AIMessage(contentresponse.content)] } # 基于搜索结果生成答案 answer_prompt f基于以下的搜索结果为用户提供完整、准确的答案 用户问题{state[user_query]} 搜索结果 {state[search_results]} 请要求 1. 综合搜索结果提供准确、有用的回答 2. 如果是技术问题提供具体的解决方案或代码 3. 引用重要信息的来源 4. 回答要结构清晰、易于理解 5. 如果搜索结果不够完整请说明并提供补充建议 response llm.invoke([SystemMessage(contentanswer_prompt)]) return { final_answer: response.content, step: completed, messages: [AIMessage(contentresponse.content)] } # 构建搜索工作流 def create_search_assistant(): workflow StateGraph(SearchState) # 添加三个节点 workflow.add_node(understand, understand_query_node) workflow.add_node(search, tavily_search_node) workflow.add_node(answer, generate_answer_node) # 设置线性流程 workflow.add_edge(START, understand) workflow.add_edge(understand, search) workflow.add_edge(search, answer) workflow.add_edge(answer, END) # 编译图 memory InMemorySaver() app workflow.compile(checkpointermemory) return app最后将工作流的访问包装成fast api接口供前端调用注意这里的流式输出需要通过Generator封装成StreamingResponse 通过 fast-api 封装成web接口 app FastAPI() # 配置CORS app.add_middleware( CORSMiddleware, allow_origins[*], # 允许所有来源生产环境应限制为具体域名 allow_credentialsTrue, allow_methods[*], # 允许所有方法 allow_headers[*], # 允许所有头 ) app.get(/stream_search, summary流式获取大模型回答) async def stream_search(query:str) - StreamingResponse: 接收主题和问题返回大模型的流式回答 # 定义生成器函数逐块获取 LangChain 的输出 def generate() - Generator[str, None, None]: # 初始状态 inital_state { messages: [HumanMessage(contentquery)], user_query: , search_query: , search_results: , final_answer: , step: start } config {configurable: {thread_id: fsearch-session-1}} search_app create_search_assistant() try: print(\n * 60) chunk_size 4 import time # 执行工作流 for output in search_app.stream(inital_state, configconfig): print(output) for node_name, node_output in output.items(): if messages in node_output and node_output[messages]: lastest_message node_output[messages][-1] if isinstance(lastest_message, AIMessage): if node_name understand: print(f 理解阶段{lastest_message.content}) content f 理解阶段{lastest_message.content}\n for i in range(0, len(content), chunk_size): chunk content[i:ichunk_size] print(chunk, end, flushTrue) time.sleep(0.1) yield chunk elif node_name search: print(f 搜索阶段{lastest_message.content}) content f 搜索阶段{lastest_message.content}\n for i in range(0, len(content), chunk_size): chunk content[i:ichunk_size] print(chunk, end, flushTrue) time.sleep(0.1) yield chunk elif node_name answer: print(f\n 最终回答:\n{lastest_message.content}) content f\n 最终回答:\n{lastest_message.content}\n for i in range(0, len(content), chunk_size): chunk content[i:ichunk_size] print(chunk, end, flushTrue) time.sleep(0.1) yield chunk print(\n *60 \n) except Exception as e: print(f❌ 发生错误{e}) print(fError occurred: {str(e)}) yield fdata: Error occurred: {str(e)} # 返回 StreamingResponse指定媒体类型为 text/event-stream适合 SSE return StreamingResponse( generate(), media_typetext/event-stream )配合前端实现后测试结果如下方式二这种方式的本质是把子智能体封装成工具供主智能体调用。架构图如下核心代码分析用户意图获取搜索关键词并封装成一个工具agent_intent create_agent( modelllm, system_prompt你是一个助手你需要帮助用户理解查询意图并生成最合适的搜索关键词。, ) # Wrap it as a tool tool(intent, description分析用户的查询意图并返回最合适的搜索关键词。) def call_intent_agent(query: str): intent_prompt f分析用户的查询{query} 请完成两个任务 1. 简洁总结用户想要了解什么 2. 生成最适合搜索的关键词中英文均可要精准 格式 理解[用户需求总结] 搜索词[最佳搜索关键词] result agent_intent.invoke({messages: [{role: user, content: intent_prompt}]}) print(result[messages][-1].content) print(*60) return result[messages][-1].content搜索工具tool(search, description根据关键词搜索信息) def search(keywords: str): # 调用Tavily API进行搜索 response tavily_client.search( query keywords, search_depth basic, include_answer True, include_raw_content False, max_results 5 ) # 处理搜索结果 search_results # 优先使用Tavily的综合答案 if response.get(answer): search_results f综合答案\n{response[answer]}\n\n # 添加具体的搜索结果 if response.get(results): search_results 相关信息\n for i, result in enumerate(response[results][:3], 1): title result.get(title, ) content result.get(content, ) url result.get(url, ) search_results f{i}. {title}\n{content}\n来源{url}\n\n if not search_results: search_results 抱歉没有找到相关的信息。 return search_results构建一个主智能体来回答用户问题answer_agent create_agent( modelllm, system_prompt你一个助手你需要基于搜索结果为用户提供完整、准确的答案。 当你收到搜索请求时需要先分析用户的查询意图得到查询关键词 然后进行根据关键词搜索信息最后给出答案。, tools[call_intent_agent, search], )接下来只要调用answer_agent来生成回答并流式返回给前端即可其余部分和提示词和方式一类似不再赘述。测试结果如下总之多智能体系统MAS的核心意义在于通过群体智能突破单智能体的能力边界适配复杂、动态、大规模的任务场景实现 “112” 的协同价值。不过我们也不需要遇到复杂的问题就直接考虑多智能体正如LangChain官网所说大部分复杂问题都可以通过优化提示和调用工具来解决。最后唠两句为什么AI大模型成为越来越多程序员转行就业、升职加薪的首选很简单这些岗位缺人且高薪智联招聘的最新数据给出了最直观的印证2025年2月AI领域求职人数同比增幅突破200% 远超其他行业平均水平整个人工智能行业的求职增速达到33.4%位居各行业榜首其中人工智能工程师岗位的求职热度更是飙升69.6%。AI产业的快速扩张也让人才供需矛盾愈发突出。麦肯锡报告明确预测到2030年中国AI专业人才需求将达600万人人才缺口可能高达400万人这一缺口不仅存在于核心技术领域更蔓延至产业应用的各个环节。那0基础普通人如何学习大模型 深耕科技一线十二载亲历技术浪潮变迁。我见证那些率先拥抱AI的同行如何建立起效率与薪资的代际优势。如今我将积累的大模型面试真题、独家资料、技术报告与实战路线系统整理分享于此为你扫清学习困惑共赴AI时代新程。我整理出这套 AI 大模型突围资料包【允许白嫖】✅从入门到精通的全套视频教程✅AI大模型学习路线图0基础到项目实战仅需90天✅大模型书籍与技术文档PDF✅各大厂大模型面试题目详解✅640套AI大模型报告合集✅大模型入门实战训练这份完整版的大模型 AI 学习和面试资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】①从入门到精通的全套视频教程包含提示词工程、RAG、Agent等技术点② AI大模型学习路线图0基础到项目实战仅需90天全过程AI大模型学习路线③学习电子书籍和技术文档市面上的大模型书籍确实太多了这些是我精选出来的④各大厂大模型面试题目详解⑤640套AI大模型报告合集⑥大模型入门实战训练如果说你是以下人群中的其中一类都可以来智泊AI学习人工智能找到高薪工作一次小小的“投资”换来的是终身受益应届毕业生‌无工作经验但想要系统学习AI大模型技术期待通过实战项目掌握核心技术。零基础转型‌非技术背景但关注AI应用场景计划通过低代码工具实现“AI行业”跨界‌。业务赋能 ‌突破瓶颈传统开发者Java/前端等学习Transformer架构与LangChain框架向AI全栈工程师转型‌。获取方式有需要的小伙伴可以保存图片到wx扫描二v码免费领取【保证100%免费】