本文详解LangChain多智能体系统架构，介绍Subagents、Handoffs、Skills、Router和Custom workflow五种实现模式，并通过搜索智能体案例展示两种实际应用方案。多智能体系统通过群体智能突破单智能体能力边界，适合复杂、动态、大规模任务场景，但需注意并非所有复杂问题都需要多智能体解决。

多智能体系统（Multi-Agent System，MAS）是由多个具备感知、决策与行动能力的自主智能体，通过通信交互与协同机制，协作或竞争完成复杂任务的分布式 AI 系统，核心是用群体智能突破单智能体能力边界，适配大规模、多目标、强动态的场景。

一、LangChain中多智能体应用简介

根据LangChain 1.0官方文档介绍，多智能体主要是用来解决复杂的工作流，而且并不是每个复杂任务都需要通过多智能体来解决，很多复杂的任务其实可以正确使用提示词和工具来解决。

一般来说以下三种情况会用到多智能体技术：

1. 上下文管理：当上下文过多时，在不占用模型上下文窗口过多资源的前提下，提供专业化知识。通过构建多个智能体获取更多的上下文知识。
2. 分布式开发：支持不同团队独立开发和维护各项功能，并将其整合为一个边界清晰的大型系统。
3. 并行化处理：为子任务分配专用的执行单元，通过并行执行提升任务处理效率。

官网将多智能体的实现方式分成了以下五种模式：

1. Subagents：主智能体将子智能体作为工具进行协同调度，所有任务路由均由主智能体统一管控，由其决定调用各子智能体的时机与方式。

Subagents的关键特性有如下几点：

集中式控制：所有任务路由均由主智能体统一处理

无直接用户交互：子智能体将结果反馈至主智能体，而非直接对接用户（不过可在子智能体中设置中断机制，以支持用户交互）

工具化调用子智能体：子智能体通过工具实现调用

并行执行：主智能体可在单次交互轮次中调用多个子智能体

总体架构图：

以购买咖啡为例，它的时序图如下：

2. Handoffs：行为根据系统状态动态调整。工具调用操作会更新一个状态变量，该变量触发任务路由或配置变更，进而完成智能体切换，或调整当前智能体的关联工具与提示词。智能体之间通过工具调用实现控制权移交。每个智能体既可以将控制权交接给其他智能体，也可以直接向用户反馈结果。在这种架构中，系统行为会基于状态动态调整。其核心机制为：工具调用会更新一个可跨交互轮次持久化的状态变量（例如current_step（当前步骤）或active_agent（活跃智能体）），系统读取该变量后对自身行为进行调整 —— 要么加载不同的配置信息（系统提示词、工具集），要么将任务路由至其他智能体。这种模式既支持不同智能体之间的任务交接，也可实现单个智能体内部的动态配置变更。

Handoffs的关键特性有以下几点：

状态驱动行为：行为基于状态变量进行调整

工具触发状态切换：通过工具调用更新状态变量，以实现不同状态间的流转

直接用户交互：各状态对应的配置可直接处理用户消息

状态持久化：状态可跨对话轮次保留

总体架构图：

以购买咖啡为例，它的时序图如下：

3. Skills：这种架构下，各类专用能力被封装为可调用的 “技能组件”，以此扩展智能体的行为边界。这类技能组件本质上是由提示词驱动的专用模块，可供智能体根据需求灵活调用。

Skills关键特性有以下几点：

提示词驱动：技能组件主要由专用提示词定义

渐进式能力：技能组件根据上下文信息或用户需求按需启用

分布式开发：不同团队可独立开展技能组件的开发与维护工作

轻量化组合：技能组件的复杂度低于完整的子智能体

总体架构图：

以购买咖啡为例，它的时序图如下：

4. Router：在Router架构下，系统会通过一个路由步骤对输入内容进行分类，并将其分发至对应的专用智能体，随后系统将各智能体返回的结果整合为一份合并后的响应内容。这种架构在处理垂直领域场景时尤为实用 —— 即当存在多个相互独立的知识域，且每个知识域都需要配备专属智能体的情况。

Router的关键特性如下：

• 路由模块对查询请求进行拆解

• 并行调用零个或多个专用智能体

• 将结果整合为连贯一致的响应内容

总体架构图：

以购买咖啡为例，它的时序图如下：

5. 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_key=os.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( model=os.getenv("LLM_MODEL_ID", "gpt-4o-mini"), api_key=os.getenv("LLM_API_KEY"), base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1"), temperature=0.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(content=understand_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(content=f"我理解您的需求：{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(content=f"✅ 搜索完成！找到了相关信息，正在为您整理答案...")] } except Exception as e: error_msg = f"搜索时发生错误：{str(e)}" print(f"❌ {error_msg}") return { "search_results": error_msg, "step": "search_failed", "messages": [AIMessage(content=f"❌ 搜索遇到问题，我将基于已有知识为您回答")] } 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(content=fallback_prompt)]) return { "final_answer": response.content, "step": "completed", "messages": [AIMessage(content=response.content)] } # 基于搜索结果生成答案 answer_prompt = f"""基于以下的搜索结果为用户提供完整、准确的答案： 用户问题：{state["user_query"]} 搜索结果： {state["search_results"]} 请要求： 1. 综合搜索结果，提供准确、有用的回答 2. 如果是技术问题，提供具体的解决方案或代码 3. 引用重要信息的来源 4. 回答要结构清晰、易于理解 5. 如果搜索结果不够完整，请说明并提供补充建议 """ response = llm.invoke([SystemMessage(content=answer_prompt)]) return { "final_answer": response.content, "step": "completed", "messages": [AIMessage(content=response.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(checkpointer=memory) return app

最后将工作流的访问包装成fast api接口，供前端调用，注意这里的流式输出需要通过Generator封装成StreamingResponse：

""" 通过 fast-api 封装成web接口 """ app = FastAPI() # 配置CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], # 允许所有来源，生产环境应限制为具体域名 allow_credentials=True, 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(content=query)], "user_query": "", "search_query": "", "search_results": "", "final_answer": "", "step": "start" } config = {"configurable": {"thread_id": f"search-session-1"}} search_app = create_search_assistant() try: print("\n" + "=" * 60) chunk_size = 4 import time # 执行工作流 for output in search_app.stream(inital_state, config=config): 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:i+chunk_size] print(chunk, end="", flush=True) 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:i+chunk_size] print(chunk, end="", flush=True) 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:i+chunk_size] print(chunk, end="", flush=True) time.sleep(0.1) yield chunk print("\n" + "="*60 + "\n") except Exception as e: print(f"❌ 发生错误：{e}") print(f"Error occurred: {str(e)}") yield f"data: Error occurred: {str(e)}" # 返回 StreamingResponse，指定媒体类型为 text/event-stream（适合 SSE） return StreamingResponse( generate(), media_type="text/event-stream" )

配合前端实现后，测试结果如下：

方式二：这种方式的本质是把子智能体封装成工具，供主智能体调用。架构图如下：

核心代码：

分析用户意图，获取搜索关键词，并封装成一个工具：

agent_intent = create_agent( model=llm, 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( model=llm, system_prompt="""你一个助手，你需要基于搜索结果为用户提供完整、准确的答案。 当你收到搜索请求时，需要先分析用户的查询意图，得到查询关键词， 然后进行根据关键词搜索信息，最后给出答案。""", tools=[call_intent_agent, search], )

接下来只要调用answer_agent来生成回答，并流式返回给前端即可，其余部分和提示词和方式一类似，不再赘述。测试结果如下：

总之，多智能体系统（MAS）的核心意义在于通过群体智能突破单智能体的能力边界，适配复杂、动态、大规模的任务场景，实现 “1+1>2” 的协同价值。不过我们也不需要遇到复杂的问题就直接考虑多智能体，正如LangChain官网所说，大部分复杂问题都可以通过优化提示和调用工具来解决。

AI时代，未来的就业机会在哪里？

答案就藏在大模型的浪潮里。从ChatGPT、DeepSeek等日常工具，到自然语言处理、计算机视觉、多模态等核心领域，技术普惠化、应用垂直化与生态开源化正催生Prompt工程师、自然语言处理、计算机视觉工程师、大模型算法工程师、AI应用产品经理等AI岗位。

掌握大模型技能，就是把握高薪未来。

那么，普通人如何抓住大模型风口？

AI技术的普及对个人能力提出了新的要求，在AI时代，持续学习和适应新技术变得尤为重要。无论是企业还是个人，都需要不断更新知识体系，提升与AI协作的能力，以适应不断变化的工作环境。

因此，这里给大家整理了一份《2026最新大模型全套学习资源》，包括2026最新大模型学习路线、大模型书籍、视频教程、项目实战、最新行业报告、面试题、AI产品经理入门到精通等，带你从零基础入门到精通，快速掌握大模型技术！

由于篇幅有限，有需要的小伙伴可以扫码获取！

1. 成长路线图&学习规划

要学习一门新的技术，作为新手一定要先学习成长路线图，方向不对，努力白费。这里，我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。

2. 大模型经典PDF书籍

书籍和学习文档资料是学习大模型过程中必不可少的，我们精选了一系列深入探讨大模型技术的书籍和学习文档，它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。（书籍含电子版PDF）

3. 大模型视频教程

对于很多自学或者没有基础的同学来说，书籍这些纯文字类的学习教材会觉得比较晦涩难以理解，因此，我们提供了丰富的大模型视频教程，以动态、形象的方式展示技术概念，帮助你更快、更轻松地掌握核心知识。

4. 大模型项目实战

学以致用，当你的理论知识积累到一定程度，就需要通过项目实战，在实际操作中检验和巩固你所学到的知识，同时为你找工作和职业发展打下坚实的基础。

5. 大模型行业报告

行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

6. 大模型面试题

面试不仅是技术的较量，更需要充分的准备。

在你已经掌握了大模型技术之后，就需要开始准备面试，我们将提供精心整理的大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

为什么大家都在学AI大模型？

随着AI技术的发展，企业对人才的需求从“单一技术”转向 “AI+行业”双背景。企业对人才的需求从“单一技术”转向 “AI+行业”双背景。金融+AI、制造+AI、医疗+AI等跨界岗位薪资涨幅达30%-50%。

同时很多人面临优化裁员，近期科技巨头英特尔裁员2万人，传统岗位不断缩减，因此转行AI势在必行！

这些资料有用吗？

这份资料由我们和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理，现任上海殷泊信息科技CEO，其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证，服务航天科工、国家电网等1000+企业，以第一作者在IEEE Transactions发表论文50+篇，获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的技术人员，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

大模型全套学习资料已整理打包，有需要的小伙伴可以微信扫描下方CSDN官方认证二维码，免费领取【保证100%免费】