Langchain 快速入门(六)：实现多agent协作
·
简介
多Agent协作能够将一个复杂的任务拆解成一个个子任务给专门的agent，能够解决复杂问题，实现复杂的ai工作流。
多Agent协作
不同的Agent，有不同的能力，我们可能会有各种实际需求，例如：实时识别车牌位置（Yolo）->识别车牌内容（qwen-vl）-> LLM管理记录车牌信息。通过多Agent协作的工作流，能够实现拍照答题，自动剪辑，ppt生成等一系列复杂问题。
下面用一个简单的案例，来说明。
简单的多Agent协作
示例
需求：查一下阿里、腾讯、百度的PE，并计算平均值。
import os
import operator
from pydantic import BaseModel, Field
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, BaseMessage, ToolMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from typing import Annotated, List, Literal, TypedDict
from langgraph.graph import StateGraph, END
os.environ["DASHSCOPE_API_KEY"] = "apikey"
llm = ChatTongyi(model="qwen-plus")
@tool
def web_search(query: str):
"""用于查找最新的股票数据、公司财报信息。"""
results = []
if "阿里" in query: results.append("阿里巴巴(BABA) PE: 15.5")
if "腾讯" in query: results.append("腾讯控股(0700) PE: 18.2")
if "百度" in query: results.append("百度(BIDU) PE: 11.8")
if not results:
return "未找到数据"
return " ; ".join(results)
@tool
def python_calculator(code: str):
"""用于计算。输入必须是 python 代码。"""
try:
result = eval(code)
return f"计算结果: {result}"
except Exception as e:
return f"计算错误: {e}"
def create_agent(state: dict, llm, tools, system_prompt):
llm_tools = llm.bind_tools(tools)
prompt = [SystemMessage(content=system_prompt)] + state["messages"]
response = llm_tools.invoke(prompt)
results = [response]
for tool_call in response.tool_calls:
func_name = tool_call["name"]
args = tool_call["args"]
call_id = tool_call["id"]
func = next((t for t in tools if t.name == func_name), None)
if func:
tool_output = func.invoke(args)
tool_msg = ToolMessage(
content=str(tool_output),
name=func_name,
tool_call_id=call_id
)
results.append(tool_msg)
return {"messages": results}
class State(TypedDict):
messages: Annotated[List[BaseMessage], operator.add]
next: str
def researcher_node(state):
return create_agent(
state=state,
llm=llm,
tools=[web_search],
system_prompt="你是一个研究员。只负责查数据，找到数据后直接输出原话，不需要计算。"
)
def coder_node(state):
return create_agent(
state=state,
llm=llm,
tools=[python_calculator],
system_prompt="你是一个程序员。根据上面研究员查到的数据，写代码计算平均值。"
)
def finish_node(state):
return create_agent(
state=state,
llm=llm,
tools=[],
system_prompt="任务完成，简短的总结最终结果。"
)
def supervisor_node(state):
system_prompt = (
"你是项目经理。根据对话历史决定下一步交给谁。"
"查数据找 Researcher，计算找 Coder，识别图片找 Photographer。"
"如果任务完成，必须选择 FINISH。"
)
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
MessagesPlaceholder(variable_name="messages"),
("system", "根据以上情况，请做出选择。"),
])
class RouteResponse(BaseModel):
next: Literal["Researcher", "Coder", "FINISH"] = Field(
...,
description="下一步交给谁？如果任务完成请选 FINISH"
)
chain = prompt | llm.with_structured_output(RouteResponse)
response = chain.invoke(state)
return {"next": response.next}
def init_agent():
workflow = StateGraph(State)
workflow.add_node("Researcher", researcher_node)
workflow.add_node("Coder", coder_node)
workflow.add_node("Supervisor", supervisor_node)
workflow.add_node("Finish", finish_node)
workflow.add_edge("Researcher", "Supervisor")
workflow.add_edge("Coder", "Supervisor")
workflow.add_edge("Finish", END)
workflow.add_conditional_edges(
"Supervisor",
lambda state: state["next"],{
"Researcher": "Researcher",
"Coder": "Coder",
"FINISH": "Finish",
})
workflow.set_entry_point("Supervisor")
return workflow.compile()
if __name__ == "__main__":
agent = init_agent()
for result in agent.stream({
"messages": [HumanMessage(content="查一下阿里、腾讯、百度的PE，并计算平均值。")]
}):
for key, value in result.items():
if key == "Supervisor":
print("[" + key + "] 去向: " + value["next"])
else:
print("[" + key + "] 回复: " + value['messages'][-1].content)
代码解释
代码一共用到了4个agent：
- agent Researcher，其有一个工具，负责搜索某些内容
- agent coder，其有一个工具，负责进行精确计算
- agent finish，其没有工具，负责总结内容
- agent Supervisor，其没有工具，负责管理上面3个agent，决定任务的去向
上面案例使用的是langgraph组件，这里就不详细讲解了，请看之前文章。
代码流程： 初始化工具库->初始化agent->构建图->运行
初始化工具库
@tool
def web_search(query: str):
"""用于查找最新的股票数据、公司财报信息。"""
results = []
if "阿里" in query: results.append("阿里巴巴(BABA) PE: 15.5")
if "腾讯" in query: results.append("腾讯控股(0700) PE: 18.2")
if "百度" in query: results.append("百度(BIDU) PE: 11.8")
if not results:
return "未找到数据"
return " ; ".join(results)
@tool
def python_calculator(code: str):
"""用于计算。输入必须是 python 代码。"""
try:
result = eval(code)
return f"计算结果: {result}"
except Exception as e:
return f"计算错误: {e}"
这里的web_search使用的是虚假的模拟信息，上面的工具描述不够完整，但是能用，如果用实际案例，请描述完整，工具的描述参考之前文章。
初始化agent
其他3个agent
def create_agent(state: dict, llm, tools, system_prompt):
llm_tools = llm.bind_tools(tools)
prompt = [SystemMessage(content=system_prompt)] + state["messages"]
response = llm_tools.invoke(prompt)
results = [response]
for tool_call in response.tool_calls:
func_name = tool_call["name"]
args = tool_call["args"]
call_id = tool_call["id"]
func = next((t for t in tools if t.name == func_name), None)
if func:
tool_output = func.invoke(args)
tool_msg = ToolMessage(
content=str(tool_output),
name=func_name,
tool_call_id=call_id
)
results.append(tool_msg)
def researcher_node(state):
return create_agent(
state=state,
llm=llm,
tools=[web_search],
system_prompt="你是一个研究员。只负责查数据，找到数据后直接输出原话，不需要计算。"
)
def coder_node(state):
return create_agent(
state=state,
llm=llm,
tools=[python_calculator],
system_prompt="你是一个程序员。根据上面研究员查到的数据，写代码计算平均值。"
)
def finish_node(state):
return create_agent(
state=state,
llm=llm,
tools=[],
system_prompt="任务完成，简短的总结最终结果。"
)
流程相对简单，create_agent细节前面文章已经讲解，这里就不废话了。
管理agent
def supervisor_node(state):
system_prompt = (
"你是项目经理。根据对话历史决定下一步交给谁。"
"查数据找 Researcher，计算找 Coder，识别图片找 Photographer。"
"如果任务完成，必须选择 FINISH。"
)
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
MessagesPlaceholder(variable_name="messages"),
("system", "根据以上情况，请做出选择。"),
])
class RouteResponse(BaseModel):
next: Literal["Researcher", "Coder", "FINISH"] = Field(
...,
description="下一步交给谁？如果任务完成请选 FINISH"
)
chain = prompt | llm.with_structured_output(RouteResponse)
response = chain.invoke(state)
return {"next": response.next}
这一步需要简单说明：
class RouteResponse(BaseModel):
next: Literal["Researcher", "Coder", "FINISH"] = Field(
...,
description="下一步交给谁？如果任务完成请选 FINISH"
)
chain = prompt | llm.with_structured_output(RouteResponse)
- LLM中的
with_structured_output方法是langchain提供的一个组件，功能是，限定LLM的输出格式，返回相应格式的字典。 - 定义输出格式限定的类：
- 该类是
BaseModel的子类 Literal是选择，要求ai从"Researcher", "Coder", "FINISH"三选一Field描述变量，尽量详尽，描述+例子，因为是给大模型看的
class classname(BaseModel):
fieldname: fieldtype = Field(..., description="描述")
构建图（重要）
def init_agent():
workflow = StateGraph(State)
workflow.add_node("Researcher", researcher_node)
workflow.add_node("Coder", coder_node)
workflow.add_node("Supervisor", supervisor_node)
workflow.add_node("Finish", finish_node)
workflow.add_edge("Researcher", "Supervisor")
workflow.add_edge("Coder", "Supervisor")
workflow.add_edge("Finish", END)
workflow.add_conditional_edges(
"Supervisor",
lambda state: state["next"],{
"Researcher": "Researcher",
"Coder": "Coder",
"FINISH": "Finish",
})
workflow.set_entry_point("Supervisor")
return workflow.compile()
这一步相当于连接工作流，构建的流程图如下：
+---------------------------+
| 开始 |
+-------------+-------------+
|
v
+---------------------------+
| Supervisor |<----------------+
| (通过当前任务状态，返回next) | |
+-------------+-------------+ |
|(根据state中next判断去向) |
_____________|_____________ |
/ | \ |
/ | \ |
v v v |
+------------+ +------------+ +------------+ |
| Researcher | | Coder | | Finish | |
| (Agent) | | (Agent) | | (Cleanup) | |
+-----+------+ +-----+------+ +-----+------+ |
| | | |
| | v |
| | +------------+ |
+---------------+ | END | |
| +------------+ |
| |
+--------------------------------------+
(返回重新选择下一个agent)想入门 AI 大模型却找不到清晰方向？备考大厂 AI 岗还在四处搜集零散资料？
别再浪费时间啦！2025 年 AI 大模型全套学习资料已整理完毕，从学习路线到面试真题，从工具教程到行业报告，一站式覆盖你的所有需求，现在全部免费分享！
👇👇扫码免费领取全部内容👇👇
一、学习必备：100+本大模型电子书+26 份行业报告 + 600+ 套技术PPT，帮你看透 AI 趋势
想了解大模型的行业动态、商业落地案例？大模型电子书？这份资料帮你站在 “行业高度” 学 AI：
1. 100+本大模型方向电子书
2. 26 份行业研究报告：覆盖多领域实践与趋势
报告包含阿里、DeepSeek 等权威机构发布的核心内容，涵盖：
- 职业趋势：《AI + 职业趋势报告》《中国 AI 人才粮仓模型解析》；
- 商业落地：《生成式 AI 商业落地白皮书》《AI Agent 应用落地技术白皮书》；
- 领域细分：《AGI 在金融领域的应用报告》《AI GC 实践案例集》；
- 行业监测：《2024 年中国大模型季度监测报告》《2025 年中国技术市场发展趋势》。
3. 600+套技术大会 PPT：听行业大咖讲实战
PPT 整理自 2024-2025 年热门技术大会，包含百度、腾讯、字节等企业的一线实践：
- 安全方向：《端侧大模型的安全建设》《大模型驱动安全升级（腾讯代码安全实践）》；
- 产品与创新：《大模型产品如何创新与创收》《AI 时代的新范式：构建 AI 产品》；
- 多模态与 Agent：《Step-Video 开源模型（视频生成进展）》《Agentic RAG 的现在与未来》；
- 工程落地：《从原型到生产：AgentOps 加速字节 AI 应用落地》《智能代码助手 CodeFuse 的架构设计》。
二、求职必看：大厂 AI 岗面试 “弹药库”，300 + 真题 + 107 道面经直接抱走
想冲字节、腾讯、阿里、蔚来等大厂 AI 岗？这份面试资料帮你提前 “押题”，拒绝临场慌！
1. 107 道大厂面经：覆盖 Prompt、RAG、大模型应用工程师等热门岗位
面经整理自 2021-2025 年真实面试场景，包含 TPlink、字节、腾讯、蔚来、虾皮、中兴、科大讯飞、京东等企业的高频考题，每道题都附带思路解析：
2. 102 道 AI 大模型真题：直击大模型核心考点
针对大模型专属考题，从概念到实践全面覆盖，帮你理清底层逻辑：
3. 97 道 LLMs 真题：聚焦大型语言模型高频问题
专门拆解 LLMs 的核心痛点与解决方案，比如让很多人头疼的 “复读机问题”：
三、路线必明：AI 大模型学习路线图，1 张图理清核心内容
刚接触 AI 大模型，不知道该从哪学起？这份「AI大模型 学习路线图」直接帮你划重点，不用再盲目摸索！
路线图涵盖 5 大核心板块，从基础到进阶层层递进：一步步带你从入门到进阶，从理论到实战。
L1阶段：了解大模型的基础知识，以及大模型在各个行业的应用和分析，学习理解大模型的核心原理、关键技术以及大模型应用场景。
L2阶段：AI大模型RAG应用开发工程，主要学习RAG检索增强生成：包括Naive RAG、Advanced-RAG以及RAG性能评估，还有GraphRAG在内的多个RAG热门项目的分析。
L3阶段：大模型Agent应用架构进阶实现，主要学习LangChain、 LIamaIndex框架，也会学习到AutoGPT、 MetaGPT等多Agent系统，打造Agent智能体。
L4阶段：大模型的微调和私有化部署，更加深入的探讨Transformer架构，学习大模型的微调技术，利用DeepSpeed、Lamam Factory等工具快速进行模型微调，并通过Ollama、vLLM等推理部署框架，实现模型的快速部署。
四、资料领取：全套内容免费抱走，学 AI 不用再找第二份
不管你是 0 基础想入门 AI 大模型，还是有基础想冲刺大厂、了解行业趋势，这份资料都能满足你！
现在只需按照提示操作，就能免费领取：
👇👇扫码免费领取全部内容👇👇
2025 年想抓住 AI 大模型的风口？别犹豫，这份免费资料就是你的 “起跑线”！
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
5
0- 0
已为社区贡献113条内容
所有评论(0)