4.1 本节挑战与目标

入门篇的Agent和LCEL链，其本质是有向无环图（DAG）。数据从一端流入，经过一系列固定的处理步骤，从另一端流出。这个模型非常适合线性的、一步到位的任务。但它无法解决以下关键挑战：

1. 循环与修正 (Cycles & Revisions): 真实的工作流充满了“返工”。如果初稿不合格，需要退回修改；如果信息不充分，需要重新搜集。线性链无法实现这种“回头”的能力。
2. 动态路由 (Dynamic Routing): 任务的下一步往往取决于上一步的结果。例如，搜集完信息后，是直接开始写作，还是发现信息不足需要补充搜索？这种条件判断和动态分支，简单的链难以实现。
3. 角色专业化 (Role Specialization): 复杂任务需要不同角色的专家。一个擅长搜索，一个擅长分析，一个擅长批判性评估。试图将所有能力塞进一个Agent的Prompt里，会导致性能下降和维护困难。
4. 状态管理 (State Management): 团队协作需要一个共享的工作空间，记录着任务的最新进展、搜集到的资料、草稿等。传统链的“记忆”通常只关注对话历史，无法管理复杂的任务状态。

本节目标：
我们将彻底拥抱LangGraph，从“链式思维”跃迁到“图式思维”。我们的目标是构建一个多智能体协作的AI研究团队，它将包含不同角色的Agent，并能在一个共享的状态下，通过循环和动态路由，协同完成一个复杂的研究任务。具体来说，你将掌握：

1. 使用StatefulGraph和TypedDict来定义和管理团队的共享状态。
2. 将不同的Agent或功能封装为图中的节点 (Nodes)。
3. 设计一个“主管”Agent，通过**条件边 (Conditional Edges)**来动态调度团队成员。
4. 构建一个包含循环修正机制的完整工作流。

4.2 核心概念深潜

想象一下，你正在领导一个项目团队完成一份市场分析报告。

• 入门级Agent：就像你把任务的所有要求写在一张长长的纸条上，交给一个非常能干但只会按部就班的实习生。他会从头到尾做一遍，但如果中间某个环节做得不好，或者需要根据新发现调整计划，他就无能为力了。

• LangGraph多智能体系统：则完全是另一幅景象。它更像一个真正的项目团队：

  • 共享项目板 (The State): 有一块白板（比如Trello或Jira），上面记录着项目的核心信息：原始需求、搜集到的数据、报告初稿、评审意见、下一步负责人等。这就是LangGraph中的State，通常用一个TypedDict来定义。团队所有成员都能看到并更新这个白板。
  • 团队成员 (The Nodes): 你有几个各司其职的专家：
    • 小张 (Researcher Agent): 擅长用各种工具搜集信息。他的任务是看一眼白板上的原始需求，然后把找到的数据写回白板。
    • 小李 (Writer Agent): 擅长分析和写作。他会查看白板上的数据，然后撰写一份报告初稿并更新到白板上。
    • 王总 (Critic Agent): 经验丰富，负责评审。他会阅读报告初稿，给出评审意见，然后更新白板。
  • 你，项目经理 (The Supervisor/Router): 你是团队的大脑。你不会亲自做具体工作，但你会不断地查看项目白板，并决定下一步该谁来做。
    • “嗯，刚开始，让小张去搜集数据。” (Entry Point -> Researcher)
    • “小张把数据放上来了，现在让小李来写初稿。” (Researcher -> Writer)
    • “小李写完了，让王总评审一下。” (Writer -> Critic)
    • “王总觉得初稿信息不全，需要补充XX数据。OK，把任务退回给小张，并把王总的意见告诉他！” (循环发生！ Critic -> Researcher)
    • “王总觉得报告OK了！太棒了，项目结束。” (Critic -> END)

LangGraph的本质，就是提供了一个框架，让我们能够精确地定义这个“项目团队”的成员（Nodes）、**共享白板（State）**以及最重要的——你作为项目经理的决策逻辑（Edges）。它将AI应用的构建从“编写线性脚本”提升到了“设计协作流程”的高度。

4.3 LangChain高级组件解析

1. langgraph.graph.StatefulGraph:

   • 是什么: 这是LangGraph的核心，一个用于构建有状态图的类。它与普通图不同，它的每个节点都能修改一个在整个图的执行过程中持续存在的、共享的状态对象。
   • 如何工作: 你首先需要定义状态的结构（通常是一个TypedDict）。然后，你将函数或Runnable对象添加为图的节点。当图运行时，它会将当前的状态对象传递给你定义的节点，节点执行完毕后返回一个字典，该字典会被用于更新状态对象。
   • 与LCEL链的区别: LCEL链是无状态的，每个组件的输出只是下一个组件的输入。StatefulGraph则拥有一个贯穿始终的、可变的共享状态，这是实现循环和复杂逻辑的基础。

2. typing.TypedDict as State:

   • 是什么: Python的TypedDict允许我们创建一个类似字典的类型，但带有明确的键和对应的值类型。在LangGraph中，它被用作定义状态结构的“蓝图”。
   • 为什么重要: 它为我们的“共享白板”提供了结构和规范。这使得代码更易于理解、维护和调试。我们可以清晰地知道状态中应该包含哪些字段（如question, documents, draft, revisions），以及它们应该是什么类型。LangGraph会利用这些类型信息来智能地更新状态（例如，使用+操作符合并列表）。

3. Nodes (节点):

   • 是什么: 图中的一个工作单元。在LangGraph中，任何接收一个状态对象并返回一个更新字典的函数或Runnable都可以作为一个节点。
   • 角色: 代表我们AI团队中的一个“专家”或一个“任务步骤”，如researcher、writer。

4. Edges (边):

   • add_edge(start_node, end_node): 定义了一个固定的、无条件的连接。一旦start_node完成，工作流总是会流向end_node。
   • add_conditional_edges(source_node, path_function, path_map): 这是实现动态路由的“魔法”。
     • source_node: 决策发生的节点，通常是我们的“主管”Agent。
     • path_function: 一个函数，它接收当前的状态，并返回一个字符串。这个字符串代表了下一步应该走哪条路。
     • path_map: 一个字典，将path_function返回的字符串映射到实际的目标节点名。例如，{"rewrite": "writer", "research": "researcher", "end": END}。

4.4 实战代码演练

我们将构建一个AI研究团队，包含三个角色：研究员(Researcher)、分析师/作家(Analyst) 和 主管(Supervisor)。

环境与工具准备

# 安装必要的库
# !pip install langchain langchain_core langchain_community langgraph dashscope tavily-python langchain-tavily -U

import os
from dotenv import load_dotenv
from typing import TypedDict, List, Annotated
import operator

# 加载环境变量
load_dotenv()

# 确保API密钥已设置
os.environ["TAVILY_API_KEY"] = os.getenv("TAVILY_API_KEY")
if os.getenv("DASHSCOPE_API_KEY") is None or os.getenv("TAVILY_API_KEY") is None:
    print("请设置环境变量 DASHSCOPE_API_KEY 和 TAVILY_API_KEY")
    exit()

from langchain_community.chat_models import ChatTongyi
# 修正：从 langchain_tavily 导入更新后的工具类 TavilySearch
from langchain_tavily import TavilySearch
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import BaseMessage, HumanMessage

# --- 1. 定义工具和模型 ---
# 为了让研究员能上网搜索，我们使用Tavily搜索工具
# 修正：使用新的 TavilySearch 类
tool = TavilySearch(max_results=3)
tools = [tool]
llm = ChatTongyi(model_name="qwen-plus", temperature=0)
print("工具和模型初始化完毕。")

# --- 2. 定义团队成员 (Agent) ---
# 我们将为每个Agent创建一个绑定了工具的LLM实例
# 这样每个Agent就知道自己能使用哪些工具

# a. 研究员 Agent: 负责使用搜索工具进行网络搜索
researcher_llm = llm.bind_tools(tools)

# b. 分析师/作家 Agent: 负责整理信息并撰写报告，它不需要工具
analyst_llm = llm


# --- 3. 定义共享状态 (State) ---
# 这是我们团队的“共享项目板”
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]  # 对话历史，使用operator.add来累加消息
    sender: str  # 记录当前消息的发送者


# --- 4. 定义节点函数 ---
# 节点是图中的工作单元，代表团队中的一个成员或一个步骤

def researcher_node(state: AgentState):
    """
    研究员节点：接收指令，使用工具进行搜索，并返回结果。
    """
    print("--- 当前节点: 研究员 ---")
    # 传递整个消息列表给LLM，为LLM提供了完整的上下文。
    response = researcher_llm.invoke(state["messages"])
    # 将LLM的响应（可能包含工具调用）也添加到消息历史中
    return {"messages": [response], "sender": "Researcher"}


def analyst_node(state: AgentState):
    """
    分析师/作家节点：接收所有信息，进行分析和撰写。
    """
    print("--- 当前节点: 分析师 ---")
    # 分析师需要看到完整的历史信息来进行总结
    response = analyst_llm.invoke(state["messages"])
    return {"messages": [response], "sender": "Analyst"}


# 创建一个 ToolNode。这个节点会自动执行 AIMessage 中请求的工具，
# 并返回一个包含结果的 ToolMessage。
tool_node = ToolNode(tools)


# --- 5. 定义主管/路由器 (Conditional Edge Logic) ---
# 主管决定下一个由谁来工作

def supervisor_router(state: AgentState):
    """
    主管路由器：检查最后一条消息，决定下一个节点。
    """
    print("--- 当前节点: 主管 (决策中) ---")
    last_message = state["messages"][-1]

    # 如果最后一条消息是工具调用，说明研究员需要执行工具。
    # 我们将任务路由到 'tool_node'。
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        print("决策：路由到工具执行节点 (tool_node)")
        return "tool_node"

    # 如果最后一条消息来自研究员，并且没有工具调用，说明研究已经完成，轮到分析师
    if state["sender"] == "Researcher":
        print("决策：路由到分析师 (研究员工作完成)")
        return "Analyst"

    # 如果最后一条消息来自分析师，说明报告已完成，任务结束
    if state["sender"] == "Analyst":
        print("决策：任务结束")
        return "END"

    # 默认情况或初始情况
    print("决策：默认路由到研究员")
    return "Researcher"


# --- 6. 构建图 ---
# a. 初始化StatefulGraph
workflow = StateGraph(AgentState)

# b. 添加节点
workflow.add_node("Researcher", researcher_node)
workflow.add_node("Analyst", analyst_node)
workflow.add_node("tool_node", tool_node)

# c. 设置入口点
workflow.set_entry_point("Researcher")

# d. 添加条件边 (这是核心)
# 重构图的边，以包含正确的工具使用流程
workflow.add_conditional_edges(
    # 决策的起点是研究员节点
    "Researcher",
    # 决策逻辑由主管执行
    supervisor_router,
    # 根据主管的决策，路由到下一个节点
    {
        "tool_node": "tool_node",
        "Analyst": "Analyst",
        "END": END
    }
)

# 添加从工具节点到研究员节点的边。
# 这意味着在工具执行完毕后，流程会返回给研究员，
# 让研究员看到工具的输出结果并进行下一步思考。
workflow.add_edge("tool_node", "Researcher")

# 分析师完成工作后，流程应该直接结束。
workflow.add_edge("Analyst", END)

# e. 编译图
# .compile()会返回一个可执行的Runnable对象
app = workflow.compile()
print("LangGraph应用编译完成！")

# --- 7. 运行多智能体协作 ---
# 使用.stream()来观察每一步的执行过程
query = "请调研最新的AI模型 'Llama 3' 和 'GPT-4o'，对比它们的特点和发布时间。"
inputs = {"messages": [HumanMessage(content=query)], "sender": "User"}
events = app.stream(
    inputs,
    config={"recursion_limit": 10}  # 设置递归限制，防止意外的无限循环
)

print("\n--- 开始执行研究任务 ---")
final_response = None
for event in events:
    for key, value in event.items():
        # 打印出每个节点的输出，观察状态的变化
        print(f"节点: {key}")
        print("---")
        # 打印完整的状态，以便调试
        print(value)
        # 捕获最终的分析师输出
        if key == "Analyst":
            final_response = value['messages'][-1].content
        print("\n====================\n")

print("\n--- 任务执行完毕 ---")
print("最终报告：")
print(final_response)

4.5 代码架构详解

让我们用一张图来梳理我们刚刚构建的AI团队的协作流程：

            +-----------------+
            |   User Query    |
            +-----------------+
                    |
                    v
(Entry Point)-->+------------+
                | Researcher | --(has tool_calls?)--> (self-loop)
                +------------+
                    | (no tool_calls)
                    v
            +------------+
            |  Supervisor| --(sender is Researcher?)--> +---------+
            |  (Router)  |                              | Analyst |
            +------------+                              +---------+
                    ^                                       |
                    |                                       v
                    +------------------(sender is Analyst?)-+
                    |
                    +--(is END?)--> +-----+
                                    | END |
                                    +-----+

1. 状态驱动 (State-Driven): 整个架构的核心是AgentState这个共享状态。它像一个“接力棒”，在不同的节点间传递。每个节点的工作都是读取当前状态，然后更新状态，再把“接力棒”传给下一个节点。operator.add的使用确保了消息历史是不断累加的，而不是被覆盖。

2. 去中心化的控制流 (Decentralized Control Flow): 我们的“主管”逻辑并不是一个独立的节点，而是被嵌入到了条件边中。这意味着控制流的决策是在节点转换的“瞬间”发生的。这种设计非常灵活，我们可以为不同的节点转换设置不同的决策逻辑。在这个例子中，我们简化了设计，让所有节点工作后都调用同一个supervisor_router，形成了一个集中的决策点。

3. 角色专业化与解耦 (Specialization & Decoupling):

   • researcher_node只关心一件事：使用工具进行搜索。它不需要知道如何写作。
   • analyst_node只关心一件事：基于现有信息进行分析和总结。它甚至不知道这些信息是哪里来的，也不需要知道如何使用工具。
   • 这种高内聚、低耦合的设计使得每个Agent都可以被独立地开发、测试和优化。例如，我们可以给研究员换用更强大的搜索工具，或者给分析师更换一个更擅长写作的LLM，而无需改动其他部分。

4. 循环的实现 (Implementing Cycles): supervisor_router中的逻辑 if hasattr(last_message, 'tool_calls') ... return "Researcher"，以及add_conditional_edges中将"Researcher"映射回"Researcher"节点的配置，共同实现了一个自循环。这使得研究员可以连续多次调用工具，直到完成所有的搜索任务，然后才将控制权交给分析师。这是一个简单但非常强大的循环模式。

4.6 总结与展望

在本章中，我们完成了从“链式思维”到“图式思维”的巨大飞跃。我们不再是构建线性的“流水线”，而是在设计一个动态的“协作团队”。我们掌握了：

• 使用StatefulGraph和TypedDict来构建多智能体协作的“共享工作区”。
• 将专业化的Agent封装为图中的节点。
• 通过条件边实现了一个“主管”，它能根据任务状态动态地调度团队成员。
• 成功构建了一个能够自主进行研究、分析和写作的AI团队，并亲眼见证了其循环和协作的过程。

LangGraph为我们打开了一扇通往更高级、更自主的AI应用的大门。我们今天构建的这个“研究团队”只是一个开始。

展望未来：

• 更复杂的团队结构: 我们可以构建更复杂的团队，比如加入一个“评审员(Critic)”节点，如果它对分析师的报告不满意，可以将任务打回给研究员或分析师进行修改，形成更复杂的修正循环。
• 层级化团队 (Hierarchical Teams): 我们可以将一个LangGraph本身作为一个节点，嵌入到另一个更大的LangGraph中，从而构建出“部门-小组”式的层级化AI组织。
• 人机协作 (Human-in-the-Loop): 我们可以在图的某个节点暂停，等待人类的输入、审批或修正，然后再继续执行，实现真正的人机协同。

我们已经拥有了构建复杂AI系统的“蓝图”。但一个系统设计得再好，我们如何知道它运行得好不好？在下一章，我们将学习如何为我们的AI应用装上“仪表盘”和“黑匣子”——我们将深入LangSmith，学习如何评估、追踪和调试我们构建的这些复杂系统。

4.7 架构权衡与工程实践

构建多智能体系统是一项强大的技术，但也带来了新的复杂性。作为架构师，我们需要清醒地认识其优缺点和工程挑战。

1. 优点:

   • 模块化与可维护性: 每个Agent职责单一，易于开发、测试和迭代。
   • 可扩展性: 向团队中增加新角色（新节点）相对容易，而不会破坏现有逻辑。
   • 性能优化: 可以为不同任务选择最合适的模型（例如，为路由选择一个快速廉价的模型，为写作选择一个强大的模型）。
   • 解决复杂问题: 能够通过循环、分支和状态管理，解决单一Agent无法处理的、非线性的复杂任务。

2. 缺点/成本:

   • 编排开销 (Orchestration Overhead): 设计和维护图的结构、状态和路由逻辑本身就有一定的工作量，比写一个简单的链要复杂得多。
   • 延迟与成本增加: 任务的每一步都可能是一次LLM调用。一个多步的图会累积多次LLM调用的延迟和费用。我们的例子中，研究、分析、路由都可能调用LLM。
   • 状态管理的复杂性: 随着团队成员和任务复杂度的增加，共享状态AgentState可能会变得非常庞大和难以管理。决定哪些信息应该进入状态，哪些应该作为临时变量，是一个关键的设计决策。
   • 调试难度: 虽然LangSmith极大地简化了调试，但理解一个复杂图的执行流程和状态变化，仍然比调试线性链更具挑战性。

3. 工程实践与“坑”:

   • 先画图，再编码: 在写任何代码之前，先在白板或绘图工具上清晰地画出你的状态图、节点和边的逻辑。这会帮你理清思路，避免在代码中迷失。
   • 避免无限循环: 这是图结构中最常见的“坑”。务必在状态中加入一个计数器（如revision_count），并在路由逻辑中检查它。当循环次数超过某个阈值时，强制结束或转入错误处理流程。
   • 原子化的节点: 尽量让每个节点的功能保持“原子性”，即完成一个独立的、定义明确的子任务。这有助于复用和测试。
   • LangSmith是必需品，不是可选项: 对于任何非平凡的LangGraph应用，没有LangSmith的可视化追踪，调试工作将是一场噩梦。从第一天起就集成它。
   • 考虑“工具Agent”: 对于一些固定的、非LLM的逻辑（如格式化数据、调用API），可以将其封装成一个不调用LLM的普通Python函数节点。这能显著降低成本和延迟。
   • 从简单开始，逐步迭代: 不要一开始就尝试构建一个包含十个节点的复杂图。从两三个核心节点的协作开始，验证其可行性，然后再逐步增加新的角色和逻辑。

LangGraph赋予了我们前所未有的能力，但能力越大，责任越大。作为架构师，我们的职责就是驾驭这种复杂性，设计出既强大又健壮、既灵活又可控的AI协作系统。