【AI神器】多Agent协作大揭秘！LangChain+Python构建智能工作流，小白也能秒变AI高手！

简介

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

多Agent协作

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

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

简单的多Agent协作

示例

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

import osimport operatorfrom pydantic import BaseModel, Fieldfrom langchain_community.chat_models.tongyi import ChatTongyifrom langchain_core.tools import toolfrom langchain_core.messages import HumanMessage, BaseMessage, ToolMessage, SystemMessagefrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom typing import Annotated, List, Literal, TypedDictfrom langgraph.graph import StateGraph, ENDos.environ["DASHSCOPE_API_KEY"] = "apikey"llm = ChatTongyi(model="qwen-plus")@tooldef 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)@tooldef 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: strdef 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->构建图->运行

初始化工具库

@tooldef 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)@tooldef python_calculator(code: str): """用于计算。输入必须是 python 代码。""" try: result = eval(code) return f"计算结果: {result}" except Exception as e: return f"计算错误: {e}"

这里的web_search使用的是虚假的模拟信息，上面的工具描述不够完整，但是能用，如果用实际案例，请描述完整，工具的描述参考之前文章。

初始化agent

Researcher，coder，finish3个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细节前面文章已经讲解，这里就不废话了。

Supervisor 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)

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

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