简介

多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%免费】