失业一年了，天天想着怎么翻身。前三课我们从基础概念一路走到实用单Agent，很多人已经能自己搭简单的流程了。这节课我们要迈上一个新台阶——多Agent协作。一个人干活效率低，让几个专门的Agent组成团队（研究员 + 写手 + 审稿人），分工明确、互相配合，才是2026年真正落地的复杂任务解决方案。

本课目标

• 理解多Agent协作（Supervisor + Worker）的基本架构
• 学会使用LangGraph实现“主管Agent”调度多个专业Worker Agent
• 掌握如何让不同Agent之间通过共享State传递消息
• 构建一个完整的三Agent写作团队，能自动完成“研究 → 写作 → 审稿”全流程

环境准备

pip install -U langgraph langchain langchain-deepseek langchain-core tavily-python

设置环境变量：

import os
os.environ["DEEPSEEK_API_KEY"] = "sk-你的key"
os.environ["TAVILY_API_KEY"] = "tvly-你的key"

核心代码实战

代码段1：定义共享State和工具

from typing import Annotated, TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from langchain_deepseek import ChatDeepSeek
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode
from tavily import TavilyClient

tavily = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))

# ==================== 共享State（所有Agent都能读写） ====================
class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    # 后续可以扩展：research_notes, draft, feedback 等

# ==================== 通用工具 ====================
@tool
def search_web(query: str) -> str:
    """搜索网页并返回总结"""
    results = tavily.search(query, max_results=4)
    return "\n\n".join([f"标题: {r['title']}\n内容: {r['content'][:300]}" for r in results['results']])

代码段2：定义三个专业Worker Agent

llm = ChatDeepSeek(model="deepseek-chat", temperature=0.7)

# Researcher Agent
researcher_llm = llm.bind_tools([search_web])

def researcher_node(state: State):
    system_prompt = "你是一个专业研究员。请搜索最新信息并总结关键点。"
    messages = [AIMessage(content=system_prompt)] + state["messages"]
    response = researcher_llm.invoke(messages)
    return {"messages": [response]}

# Writer Agent
def writer_node(state: State):
    system_prompt = "你是一个专业写手。根据研究员提供的信息，撰写一篇结构清晰、流畅的文章。"
    messages = [AIMessage(content=system_prompt)] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

# Reviewer Agent（审稿人）
def reviewer_node(state: State):
    system_prompt = "你是一个严格的审稿人。请检查文章的逻辑、事实准确性和语言流畅度，给出改进建议或直接通过。"
    messages = [AIMessage(content=system_prompt)] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

代码段3：主管Agent（Supervisor） + 完整团队图

# Supervisor（主管）决定下一个由谁执行
supervisor_llm = llm.bind_tools([])  # 不绑定工具，只做决策

def supervisor_node(state: State) -> dict:
    system_prompt = """
    你是整个写作团队的主管。
    可用角色：researcher（研究员）、writer（写手）、reviewer（审稿人）、FINISH（结束）。
    根据当前对话进度，决定下一个应该由谁继续工作。
    输出格式必须是：Next: researcher / writer / reviewer / FINISH
    """
    messages = [AIMessage(content=system_prompt)] + state["messages"]
    response = supervisor_llm.invoke(messages)
    return {"messages": [response]}

# 路由函数：根据主管的决策决定下一个节点
def route_supervisor(state: State) -> Literal["researcher", "writer", "reviewer", "FINISH"]:
    last_msg = state["messages"][-1].content.lower()
    
    if "researcher" in last_msg:
        return "researcher"
    elif "writer" in last_msg:
        return "writer"
    elif "reviewer" in last_msg:
        return "reviewer"
    else:
        return "FINISH"

# ==================== 构建多Agent团队图 ====================
workflow = StateGraph(State)

# 添加所有节点
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)
workflow.add_node("reviewer", reviewer_node)

# 从起点进入主管
workflow.add_edge(START, "supervisor")

# 主管决策后的条件路由
workflow.add_conditional_edges(
    "supervisor",
    route_supervisor,
    {
        "researcher": "researcher",
        "writer": "writer",
        "reviewer": "reviewer",
        "FINISH": END
    }
)

# 每个Worker完成工作后，都回到主管重新决策
workflow.add_edge("researcher", "supervisor")
workflow.add_edge("writer", "supervisor")
workflow.add_edge("reviewer", "supervisor")

# 编译
graph = workflow.compile()

# ==================== 测试运行 ====================
user_query = "帮我写一篇关于'2026年LangGraph在企业级应用中的实践'的1500字深度文章"

result = graph.invoke({
    "messages": [HumanMessage(content=user_query)]
})

print("\n=== 最终输出（审稿通过后） ===")
print(result["messages"][-1].content)

运行结果说明
典型流程：
supervisor → researcher（搜索资料） → supervisor → writer（撰写文章） → supervisor → reviewer（审稿） → supervisor → FINISH

最终输出是一篇经过研究员提供资料、写手撰写、审稿人把关的完整文章。

小练习（正好2道）

• 练习1（基础）
  在supervisor的system_prompt中增加“如果信息不足，让researcher再搜索一次”的逻辑。修改prompt并测试。

• 练习2（进阶）
  在State中新增一个字段 current_stage: str，让每个Worker在返回时更新当前阶段（research / writing / reviewing）。修改State和对应节点代码。

本课小结
本课我们实现了多Agent协作基础模式：一个Supervisor（主管）负责调度，多个专业Worker（研究员、写手、审稿人）各司其职。通过共享State + conditional routing，Agent之间可以有序协作。这就是Crew / Team模式的核心思想，比单个Agent强大得多。

下节预告
第5课：带记忆与状态的多轮复杂工作流
我们会在多Agent基础上加入持久化记忆，让团队可以处理更长的对话和复杂任务。下节见！

如果觉得这篇有用，欢迎点赞和关注，一起玩转 LangGraph！