news 2026/6/10 14:11:40

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/4 14:55:00

从免费到付费：10款AIGC优化工具性能与价格全面分析

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

作者头像 李华
网站建设 2026/6/10 10:56:34

OpCore Simplify：系统配置优化与硬件兼容性适配的技术实践

OpCore Simplify&#xff1a;系统配置优化与硬件兼容性适配的技术实践 【免费下载链接】OpCore-Simplify A tool designed to simplify the creation of OpenCore EFI 项目地址: https://gitcode.com/GitHub_Trending/op/OpCore-Simplify 黑苹果配置的核心挑战与解决方案…

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

青岛籍影视演员韩锐携“我爱我家”组合亮相青岛春晚，歌曲《回山东过年吧》唱出家的声音

《2026年新青岛熠熠之光春节晚会》节目录制现场&#xff0c;由青岛籍影视演员、音乐创作人韩锐与家人组成的“我爱我家”家庭乐队温情登场&#xff0c;首唱了原创贺岁单曲《回山东过年吧》。这首以“回家过年”为主线的歌曲&#xff0c;凭借流畅温暖的旋律、质朴真挚的歌词&…

作者头像 李华
网站建设 2026/6/10 10:54:32

黑苹果配置3步突破：用OpCore Simplify打造高性能游戏主机

黑苹果配置3步突破&#xff1a;用OpCore Simplify打造高性能游戏主机 【免费下载链接】OpCore-Simplify A tool designed to simplify the creation of OpenCore EFI 项目地址: https://gitcode.com/GitHub_Trending/op/OpCore-Simplify 你是否曾遇到过这样的困境&#…

作者头像 李华
网站建设 2026/6/10 10:49:05

腾讯混元0.5B：超轻量AI边缘推理新标杆

腾讯混元0.5B&#xff1a;超轻量AI边缘推理新标杆 【免费下载链接】Hunyuan-0.5B-Pretrain 腾讯开源混元大模型系列中的高效轻量版本&#xff0c;专注性能与部署灵活性。0.5B参数规模兼顾边缘设备与高并发场景&#xff0c;支持256K超长上下文和混合推理模式&#xff0c;具备强大…

作者头像 李华