news 2026/6/12 9:23:33

langchain 快速入门(六)：实现多agent协作

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
langchain 快速入门(六)：实现多agent协作

简介

多Agent协作能够将一个复杂的任务拆解成一个个子任务给专门的agent，能够解决复杂问题，实现复杂的ai工作流。

多Agent协作

不同的Agent，有不同的能力，我们可能会有各种实际需求，例如：实时识别车牌位置（Yolo）->识别车牌内容（qwen-vl）-> LLM管理记录车牌信息。通过多Agent协作的工作流，能够实现拍照答题，自动剪辑，ppt生成等一系列复杂问题。

下面用一个简单的案例，来说明。

简单的多Agent协作

示例

需求：查一下阿里、腾讯、百度的PE，并计算平均值。

/* by yours.tools - online tools website : yours.tools/zh/togif.html */ 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：

  1. agent Researcher，其有一个工具，负责搜索某些内容
  2. agent coder，其有一个工具，负责进行精确计算
  3. agent finish，其没有工具，负责总结内容
  4. agent Supervisor，其没有工具，负责管理上面3个agent，决定任务的去向
    上面案例使用的是langgraph组件，这里就不详细讲解了，请看之前文章。

代码流程：初始化工具库->初始化agent->构建图->运行

初始化工具库

/* by yours.tools - online tools website : yours.tools/zh/togif.html */ @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的输出格式，返回相应格式的字典。
  • 定义输出格式限定的类：
  1. 该类是BaseModel的子类
  2. Literal是选择，要求ai从"Researcher", "Coder", "FINISH"三选一
  3. 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)

如果❤喜欢❤本系列教程，就点个关注吧，后续不定期更新~

版权声明: 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！
网站建设 2026/6/10 21:11:05

零基础掌握智能视频增强：3大场景×5个实战技巧

零基础掌握智能视频增强&#xff1a;3大场景5个实战技巧 【免费下载链接】flowframes Flowframes Windows GUI for video interpolation using DAIN (NCNN) or RIFE (CUDA/NCNN) 项目地址: https://gitcode.com/gh_mirrors/fl/flowframes 核心功能解析 视频流畅度优化是…

作者头像 李华
网站建设 2026/6/10 16:08:33

解锁动态桌面定制新体验：Lively Wallpaper全方位创意指南

解锁动态桌面定制新体验&#xff1a;Lively Wallpaper全方位创意指南 【免费下载链接】lively Free and open-source software that allows users to set animated desktop wallpapers and screensavers powered by WinUI 3. 项目地址: https://gitcode.com/gh_mirrors/li/li…

作者头像 李华
网站建设 2026/6/10 12:47:47

智能创作升级：10大AIGC平台免费版与付费版核心差异

&#xfffd;&#xfffd; 10大降AIGC平台核心对比速览 排名 工具名称 降AIGC效率 适用场景 免费/付费 1 askpaper ⭐⭐⭐⭐⭐ 学术论文精准降AI 付费 2 秒篇 ⭐⭐⭐⭐⭐ 快速降AIGC降重 付费 3 Aibiye ⭐⭐⭐⭐ 多学科论文降AI 付费 4 Aicheck ⭐⭐⭐⭐…

作者头像 李华
网站建设 2026/6/9 17:20:52

深度测评9个AI论文软件，助继续教育学生轻松搞定毕业论文！

深度测评9个AI论文软件&#xff0c;助继续教育学生轻松搞定毕业论文&#xff01; AI 工具如何改变论文写作的未来 在当今快速发展的科技时代&#xff0c;AI 工具正逐步渗透到学术研究的各个环节&#xff0c;尤其是在论文写作中&#xff0c;AI 的应用不仅提升了效率&#xff0c;…

作者头像 李华
网站建设 2026/5/30 23:56:49

PLC控制机械手设计

机械手电气控制系统&#xff0c;除了有多工步特点之外&#xff0c;还要求有连续控制和手动控制等操作方式。工作方式的选择可以很方便地在操作面板上表示出来。当旋钮打向回原点时&#xff0c;系统自动地回到左上角位置待命。当旋钮打向自动时&#xff0c;系统自动完成各工步操…

作者头像 李华