文章介绍了多智能体系统(MAS)的概念和在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大模型 ？

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。【保证100%免费】🆓

CSDN粉丝独家福利

这份完整版的 AI 大模型学习资料已经上传CSDN，朋友们如果需要可以扫描下方二维码&点击下方CSDN官方认证链接免费领取 【保证100%免费】

读者福利： 👉👉CSDN大礼包：《最新AI大模型学习资源包》免费分享 👈👈

（👆👆👆安全链接，放心点击）

对于0基础小白入门：

如果你是零基础小白，想快速入门大模型是可以考虑的。

一方面是学习时间相对较短，学习内容更全面更集中。
二方面是可以根据这些资料规划好学习计划和方向。

👉1.大模型入门学习思维导图👈

要学习一门新的技术，作为新手一定要先学习成长路线图，方向不对，努力白费。

对于从来没有接触过AI大模型的同学，我们帮你准备了详细的学习成长路线图&学习规划。可以说是最科学最系统的学习路线，大家跟着这个大的方向学习准没问题。（全套教程文末领取哈）

👉2.AGI大模型配套视频👈

很多朋友都不喜欢晦涩的文字，我也为大家准备了视频教程，每个章节都是当前板块的精华浓缩。

👉3.大模型实际应用报告合集👈

这套包含640份报告的合集，涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师，还是对AI大模型感兴趣的爱好者，这套报告合集都将为您提供宝贵的信息和启示。（全套教程文末领取哈）

👉4.大模型实战项目&项目源码👈

光学理论是没用的，要学会跟着一起做，要动手实操，才能将自己的所学运用到实际当中去，这时候可以搞点实战项目来学习。（全套教程文末领取哈）

👉5.大模型经典学习电子书👈

随着人工智能技术的飞速发展，AI大模型已经成为了当今科技领域的一大热点。这些大型预训练模型，如GPT-3、BERT、XLNet等，以其强大的语言理解和生成能力，正在改变我们对人工智能的认识。 那以下这些PDF籍就是非常不错的学习资源。（全套教程文末领取哈）

👉6.大模型面试题&答案👈

截至目前大模型已经超过200个，在大模型纵横的时代，不仅大模型技术越来越卷，就连大模型相关的岗位和面试也开始越来越卷了。为了让大家更容易上车大模型算法赛道，我总结了大模型常考的面试题。（全套教程文末领取哈）

为什么分享这些资料?

只要你是真心想学AI大模型，我这份资料就可以无偿分享给你学习，我国在这方面的相关人才比较紧缺，大模型行业确实也需要更多的有志之士加入进来，我也真心希望帮助大家学好这门技术，如果日后有什么学习上的问题，欢迎找我交流，有技术上面的问题，我是很愿意去帮助大家的！

这些资料真的有用吗?

这份资料由我和鲁为民博士共同整理，鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位，在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利，同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。

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

CSDN粉丝独家福利

这份完整版的 AI 大模型学习资料已经上传CSDN，朋友们如果需要可以扫描下方二维码&点击下方CSDN官方认证链接免费领取 【保证100%免费】

读者福利： 👉👉CSDN大礼包：《最新AI大模型学习资源包》免费分享 👈👈

（👆👆👆安全链接，放心点击）