第11讲：多Agent协作系统设计——角色分工、任务调度与群聊抽象逻辑

weixin_47429110

581人浏览 · 2026-06-18 10:36:39

weixin_47429110 · 2026-06-18 10:36:39 发布

欢迎来到《Python + AI Agent 实战开发完全指南》专栏！

在前面的课程中，我们一步步为 Agent 赋予了记忆、检索、推理规划甚至代码执行的能力。但如果你仔细观察现实世界中的企业级项目（如复杂的软件开发或深度的市场调研），你会发现一个残酷的现实：单靠一个全能 Agent 往往难以胜任。当上下文越来越长、工具越来越多时，单一模型极易出现"注意力涣散"、指令遗忘和逻辑幻觉。

在这一讲中，我们将迎来本专栏最激动人心的篇章——多智能体协作系统（Multi-Agent System）。我们将借鉴人类公司的组织架构，让多个拥有不同专长（Role）的 Agent 组成团队，通过任务交接与群聊机制，共同完成超大规模的复杂项目！

一、架构升级：从"孤胆英雄"到"专家团队"

在传统的单 Agent 架构中，所有的 Prompt 都试图把大模型塑造成一个全知全能的超级天才。但在多智能体架构下，我们遵循 “高内聚、低耦合” 的软件工程原则：

核心设计理念

角色专精化：每个 Agent 只负责自己最擅长的领域（例如：研究员只负责查资料，程序员只负责写代码）。这大幅降低了单个 Agent 的认知负担。
状态隔离与共享：Agent 之间不需要共享冗长的历史对话，而是通过结构化的消息传递（Handoff）进行高效的信息同步。
动态调度：引入"路由/管理者"节点，根据任务的进展自动决定下一步该由谁来接手。

这种模式不仅提高了任务的成功率，还让整个工作流变得极其透明且易于调试。

二、核心实战：构建"调研-写作"双专家协作流

为了演示多智能体的威力，我们设定一个真实的业务场景：撰写一份关于"AI行业趋势"的简报。我们将构建两个专职 Agent：Researcher（数据收集员）和 Writer（资深主编），以及一个 Router（前台接待）来统筹全局。

2.1 环境准备

import os
os.environ["PYTHONUTF8"] = "1"

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.messages import HumanMessage, SystemMessage

# --- 1. 初始化 LLM ---
API_KEY = "sk-9a64b8b09c41489f923cd59417b7dff8"
BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
MODEL = "qwen-turbo"

llm = ChatOpenAI(api_key=API_KEY, base_url=BASE_URL, model=MODEL)

2.2 定义专职 Agent 角色

# 研究员 Agent：只负责收集和整理事实信息
researcher_prompt = SystemMessage(content="""你是一个严谨的行业研究员。你的唯一任务是搜集并总结客观事实和数据。
不要发表主观评价，不要写文章。请将收集到的信息以要点列表的形式返回。""")

# 作家 Agent：只负责基于现有素材进行内容创作
writer_prompt = SystemMessage(content="""你是一个资深的科技媒体主编。你的任务是根据研究员提供的要点，撰写一篇引人入胜、逻辑连贯的文章。
不要再去查证事实，直接基于提供的信息进行润色和扩写。""")

# 路由 Agent：负责判断当前对话应该交给谁处理
router_prompt = SystemMessage(content="""你是项目总监。请分析用户的请求和当前的对话记录：
- 如果用户提出了新的研究课题，请指示交给 Researcher。
- 如果已经有了充足的研究资料，需要成文，请指示交给 Writer。
- 如果文章已经写好，请直接结束流程。
你只需要输出下一个接手的 Agent 名称（Researcher / Writer / END）。""")

2.3 节点函数与条件路由逻辑

def router_node(state: MessagesState):
    response = llm.invoke([router_prompt] + state["messages"])
    return {"messages": response}

def researcher_node(state: MessagesState):
    response = llm.invoke([researcher_prompt] + state["messages"])
    return {"messages": response}

def writer_node(state: MessagesState):
    response = llm.invoke([writer_prompt] + state["messages"])
    return {"messages": response}

# 【核心】条件路由：解析路由 Agent 的输出，决定流转方向
def route_to_agent(state: MessagesState):
    last_message = state["messages"][-1].content.lower()
    if "researcher" in last_message:
        return "researcher"
    elif "writer" in last_message:
        return "writer"
    else:
        return END

# --- 编译多智能体状态图 ---
builder = StateGraph(MessagesState)
builder.add_node("router", router_node)
builder.add_node("researcher", researcher_node)
builder.add_node("writer", writer_node)

builder.add_edge(START, "router")
builder.add_conditional_edges("router", route_to_agent)
# 无论是研究员还是作家完成工作，都必须回到路由器重新评估下一步
builder.add_edge("researcher", "router") 
builder.add_edge("writer", "router") 

checkpointer = InMemorySaver()
app = builder.compile(checkpointer=checkpointer)

三、测试团队协作流水线

现在，让我们下达一个复杂的指令，看看这个"微型公司"是如何运转的：

if __name__ == "__main__":
    config = {"configurable": {"thread_id": "session_multi_agent_001"}}
    
    print("--- 测试：多智能体协作生成报告 ---")
    user_question = "帮我写一份关于2024年AI Agent发展趋势的简报。"
    res = app.invoke({"messages": [HumanMessage(content=user_question)]}, config=config)
    print(f"\n最终输出:\n{res['messages'][-1].content}")