第 20 期 | 群组研讨会 (Network of Agents) 模拟实战

副标题：Writer 写完交给 Editor 挑刺，再发回给 Writer 改稿的三方博弈环形流。

同学们好！我是你们的 AI 技术导师。今天，我们不玩单挑，我们要玩群殴！哦不，是群组协作！在现实世界中，任何复杂的工作都不是一个人能搞定的，智能体也一样。我们之前构建的 AI 内容机构，虽然有 Planner、Researcher、Writer、Editor 各司其职，但它们之间往往是单向流转。今天，我们要打破这种线性思维，引入一个更真实、更智能的协作模式——群组研讨会 (Network of Agents)，特别是那种带有反馈和迭代的环形流。

想象一下，你的 Writer 吭哧吭哧写了一篇稿子，信心满满地交给 Editor。Editor 一看，眉头一皱，"这不行，逻辑不顺，语气也不对，给我改！" 然后啪嗒一下，稿子又回到了 Writer 手里。Writer 带着委屈和 Editor 的批注，继续修改。这样的来回，是不是很像我们日常工作的真实写照？这就是我们今天要用 LangGraph 模拟的三方博弈环形流：Writer (写稿方)、Editor (审稿方)、以及贯穿其中的内容状态。

这种环形流，是构建任何具有迭代、优化、协商机制的复杂 AI 系统不可或缺的能力。学会它，你的 AI Agent 将不再是僵硬的流水线工人，而是能真正“思考”、“反馈”和“迭代”的智能团队成员。准备好了吗？让我们深入 LangGraph 的核心，解锁这个高级技能！

🎯 本期学习目标

完成本期学习后，你将能够：

1. 理解并构建基于条件判断的 LangGraph 循环流：掌握如何利用 add_conditional_edges 实现多智能体之间的动态反馈和迭代机制。
2. 精细化管理复杂图状态 (Graph State)：学会在多智能体协作中，如何设计和更新共享状态，以承载各智能体的输入、输出和决策信息。
3. 模拟 AI 内容机构的真实协作场景：将 Writer 和 Editor 的角色深度融合，实现从初稿到定稿的自动迭代优化过程。
4. 掌握避免循环陷阱与优化迭代流程的策略：了解在构建循环流时可能遇到的问题，并学会如何设计健壮的退出条件和优化机制。

📖 原理解析

在 LangGraph 中，构建一个多智能体协作的“群组研讨会”核心在于图状态 (Graph State) 和条件边 (Conditional Edges)。想象一下，Graph State 就是一个共享的白板，所有智能体都在上面写写画画，更新信息。条件边则像是会议的主持人，根据白板上的最新讨论结果（状态），决定下一步该由谁发言，或者是否可以散会了。

我们今天的“三方博弈”——Writer 写稿、Editor 挑刺、Writer 改稿——其本质是一个迭代优化循环。

1. Writer (写稿方)：接收一个主题或待修改的内容，输出一个初稿或修订稿。
2. Editor (审稿方)：接收 Writer 的稿件，进行评估。它会输出两样东西：
   • 反馈意见：告诉 Writer 哪里需要改进。
   • 决策：判断稿件是否已达到要求 (approved)，或者仍需修改 (needs_revision)。
3. 内容状态 (Content State)：这是我们的“白板”，上面记录了：
   • 当前稿件 (current_article)。
   • Editor 的最新反馈 (editor_feedback)。
   • 稿件的当前状态 (status: drafting, revising, approved)。
   • 甚至可以记录修订次数 (revision_count)，以防止无限循环。

这个流程的精髓在于 Editor 的决策。如果 Editor 决定 needs_revision，那么流程就会通过条件边，再次回到 Writer 节点。如果 Editor 决定 approved，那么流程就终止（或者进入下一个阶段，比如发布）。

Mermaid 图解核心架构

让我们用 Mermaid 图来直观地展示这个循环流：

graph TD
    A[开始] --> B(Writer: 撰写/修改文章);
    B --> C{Editor: 审查文章};
    C -- 稿件需要修改 (needs_revision) --> B;
    C -- 稿件通过 (approved) --> D[结束: 文章定稿];

    style A fill:#f9f,stroke:#333,stroke-width:2px;
    style B fill:#bbf,stroke:#333,stroke-width:2px;
    style C fill:#bfb,stroke:#333,stroke-width:2px;
    style D fill:#f9f,stroke:#333,stroke-width:2px;

图解说明：

• 开始 (Start)：整个流程的起点。
• Writer (撰写/修改文章)：这是一个节点，代表 Writer Agent 的操作。它会根据当前的状态（是初次撰写还是根据反馈修改）来生成或更新文章。
• Editor (审查文章)：这是另一个节点，代表 Editor Agent 的操作。它会接收 Writer 的输出，进行审查，并做出决策。
• 条件边 (Conditional Edges)：
  • 稿件需要修改 (needs_revision)：如果 Editor 判断稿件需要修改，这条条件边会将流程导回 Writer 节点，形成一个循环。
  • 稿件通过 (approved)：如果 Editor 判断稿件已达到要求，这条条件边会将流程导向结束，文章定稿。
• 结束 (End)：流程的终点，标志着文章的最终完成。

这个图清晰地展示了，通过 LangGraph 的条件边机制，我们如何将两个独立的智能体（Writer 和 Editor）连接成一个具有反馈和迭代能力的协作网络。这就是“群组研讨会”的核心魅力！

💻 实战代码演练 (Agency项目中的具体应用)

好了，理论说得再漂亮，不如代码来得实在。现在，我们把这套“三方博弈”机制，无缝集成到我们的 AI 内容创作机构中。我们将创建一个 ArticleRevisionGraph，它将包含 Writer 和 Editor 两个核心 Agent，并通过 LangGraph 的 StateGraph 和 add_conditional_edges 功能，实现文章的迭代修改。

我们将使用一个简化的 AgentState 来模拟文章内容、编辑反馈和修订状态。

import operator
from typing import TypedDict, Annotated, List, Union
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import os

# 确保你的OpenAI API Key已经设置在环境变量中
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 定义图的状态
# Define the state of the graph
class AgentState(TypedDict):
    """
    Represent the state of our content agency's article revision process.
    代表我们内容机构文章修订流程的状态。
    """
    topic: str  # 文章的主题 The topic of the article
    current_article: str  # 当前的文章内容 The current content of the article
    editor_feedback: str  # 编辑的反馈意见 Editor's feedback
    revision_count: Annotated[int, operator.add]  # 修订次数，使用operator.add累加 Revision count, accumulates using operator.add
    status: str  # 文章的当前状态 (e.g., "drafting", "revising", "approved") The current status of the article

# 模拟的Writer Agent
# Simulated Writer Agent
class WriterAgent:
    def __init__(self, llm_model: str = "gpt-4o-mini"):
        self.llm = ChatOpenAI(model=llm_model, temperature=0.7)

    def write_or_revise(self, state: AgentState) -> AgentState:
        """
        Writer Agent的节点函数：根据当前状态撰写初稿或修订文章。
        Writer Agent's node function: Writes the initial draft or revises the article based on the current state.
        """
        topic = state["topic"]
        current_article = state["current_article"]
        editor_feedback = state["editor_feedback"]
        revision_count = state["revision_count"]

        # 构建给LLM的提示
        # Construct the prompt for the LLM
        if revision_count == 0:
            prompt = HumanMessage(f"你是一位专业的文章写手。请根据以下主题撰写一篇高质量的文章：\n主题: {topic}\n\n请确保文章结构清晰，内容丰富，语言流畅。")
            print(f"\n--- Writer: 撰写初稿 (Revision {revision_count}) ---")
        else:
            prompt = HumanMessage(f"你是一位专业的文章写手。以下是之前的文章内容和编辑的反馈意见，请根据反馈进行修改和润色：\n\n旧文章内容:\n{current_article}\n\n编辑反馈:\n{editor_feedback}\n\n请提供修改后的完整文章。")
            print(f"\n--- Writer: 正在修订文章 (Revision {revision_count}) ---")
            print(f"接收到编辑反馈: {editor_feedback}")

        # 调用LLM生成内容
        # Call LLM to generate content
        response = self.llm.invoke([prompt])
        new_article = response.content

        # 更新状态
        # Update the state
        print(f"Writer 生成了新文章或修订稿:\n{new_article[:200]}...") # 打印前200字
        return {"current_article": new_article, "status": "revising", "revision_count": state["revision_count"] + 1}

# 模拟的Editor Agent
# Simulated Editor Agent
class EditorAgent:
    def __init__(self, llm_model: str = "gpt-4o-mini", max_revisions: int = 3):
        self.llm = ChatOpenAI(model=llm_model, temperature=0.7)
        self.max_revisions = max_revisions # 设置最大修订次数

    def review_article(self, state: AgentState) -> AgentState:
        """
        Editor Agent的节点函数：审查文章，提供反馈，并决定是否通过或需要修改。
        Editor Agent's node function: Reviews the article, provides feedback, and decides whether to approve or require revision.
        """
        current_article = state["current_article"]
        revision_count = state["revision_count"]

        # 构建给LLM的提示
        # Construct the prompt for the LLM
        prompt = HumanMessage(f"你是一位资深的内容编辑。请审查以下文章，并给出详细的修改意见。如果文章已经达到发布标准，请在反馈中明确说明 '文章已通过'。否则，请详细指出需要改进的地方。\n\n文章内容:\n{current_article}")
        print(f"\n--- Editor: 正在审查文章 (Revision {revision_count-1}的成果) ---")

        # 调用LLM生成反馈
        # Call LLM to generate feedback
        response = self.llm.invoke([prompt])
        feedback = response.content

        # 判断文章是否通过
        # Determine if the article is approved
        status = "needs_revision"
        if "文章已通过" in feedback or "已达到发布标准" in feedback:
            status = "approved"
            print("Editor 判定: 文章已通过！")
        elif revision_count >= self.max_revisions:
            status = "approved" # 达到最大修订次数，强制通过（实际项目中可能需要人工介入）
            feedback += "\n[系统提示]：已达到最大修订次数，文章强制通过，可能需要人工复审。"
            print(f"Editor 判定: 达到最大修订次数 {self.max_revisions}，强制通过。")
        else:
            print("Editor 判定: 文章需要修改。")

        # 更新状态
        # Update the state
        print(f"Editor 提供了反馈:\n{feedback[:200]}...") # 打印前200字
        return {"editor_feedback": feedback, "status": status}

# 定义一个路由函数，用于决定下一步的流向
# Define a routing function to decide the next flow
def route_article(state: AgentState) -> str:
    """
    根据文章的'status'决定下一步的节点。
    Routes the article based on its 'status'.
    """
    if state["status"] == "approved":
        print("\n--- 路由: 文章已通过，流程结束 ---")
        return "end"
    elif state["status"] == "revising":
        print("\n--- 路由: 文章需要修改，返回给 Writer ---")
        return "writer"
    else:
        # 默认情况，例如初始状态，或者未知状态，通常会回到Writer
        # Default case, e.g., initial state, or unknown state, usually goes back to Writer
        print("\n--- 路由: 初始状态或未知状态，返回给 Writer ---")
        return "writer"

# 构建LangGraph
# Build the LangGraph
def build_article_revision_graph(llm_model: str = "gpt-4o-mini", max_revisions: int = 3):
    writer_agent = WriterAgent(llm_model=llm_model)
    editor_agent = EditorAgent(llm_model=llm_model, max_revisions=max_revisions)

    # 初始化StateGraph
    # Initialize StateGraph
    workflow = StateGraph(AgentState)

    # 添加节点
    # Add nodes
    workflow.add_node("writer", writer_agent.write_or_revise)
    workflow.add_node("editor", editor_agent.review_article)

    # 设置入口点
    # Set the entry point
    workflow.set_entry_point("writer")

    # 添加边
    # Add edges
    # Writer 节点完成后，总是交给 Editor 审查
    # After the Writer node completes, it always goes to the Editor for review
    workflow.add_edge("writer", "editor")

    # Editor 节点完成后，根据路由函数决定是返回 Writer 还是结束
    # After the Editor node completes, a routing function decides whether to return to Writer or end
    workflow.add_conditional_edges(
        "editor",      # 来源节点 From node
        route_article, # 路由函数 Routing function
        {              # 映射表 Mapping table
            "writer": "writer", # 如果路由函数返回"writer"，则流向"writer"节点 If routing function returns "writer", flow to "writer" node
            "end": END          # 如果路由函数返回"end"，则流程结束 If routing function returns "end", the process ends
        }
    )

    # 编译图
    # Compile the graph
    app = workflow.compile()
    return app

# 运行模拟
# Run the simulation
if __name__ == "__main__":
    # 确保OpenAI API Key已设置
    if not os.getenv("OPENAI_API_KEY"):
        print("请设置环境变量 OPENAI_API_KEY。")
        exit()

    # 构建并运行图
    # Build and run the graph
    app = build_article_revision_graph(llm_model="gpt-4o-mini", max_revisions=3) # 可以尝试 gpt-3.5-turbo 或 gpt-4o-mini

    initial_state = {
        "topic": "探索人工智能在教育领域的应用与挑战",
        "current_article": "",
        "editor_feedback": "",
        "revision_count": 0,
        "status": "drafting"
    }

    print("--- 开始文章创作与修订流程 ---")
    final_state = None
    for s in app.stream(initial_state):
        print(s) # 打印每一步的状态变化
        final_state = s

    print("\n--- 流程结束 ---")
    print("\n最终文章状态:")
    print(f"主题: {final_state['editor']['topic']}")
    print(f"最终修订次数: {final_state['editor']['revision_count'] - 1}") # 减1是因为第一次是0，然后每次+1
    print(f"最终状态: {final_state['editor']['status']}")
    print("\n--- 最终文章内容 ---")
    print(final_state['editor']['current_article'])
    print("\n--- 最终编辑反馈 ---")
    print(final_state['editor']['editor_feedback'])

代码解析：

1. AgentState 定义：
   • 我们使用 TypedDict 定义了 AgentState，它包含了 topic、current_article、editor_feedback、revision_count 和 status。
   • revision_count: Annotated[int, operator.add] 是一个 LangGraph 特有的高级用法，它告诉 LangGraph 在合并状态时，对于 revision_count 字段，要使用 operator.add 进行累加，而不是简单的覆盖。这意味着每次节点返回新的 revision_count，都会在原有基础上累加。这非常适合计数器。
2. WriterAgent：
   • write_or_revise 方法是 Writer 节点的核心逻辑。
   • 它根据 revision_count 判断是初次撰写还是进行修改。
   • 如果是修改，它会把 current_article 和 editor_feedback 作为上下文传递给 LLM，让 LLM 进行有针对性的修改。
   • 返回一个新的 current_article 和更新后的 status 及 revision_count。
3. EditorAgent：
   • review_article 方法是 Editor 节点的核心逻辑。
   • 它接收 current_article，并要求 LLM 给出反馈。
   • 关键在于，它会检查 LLM 的反馈中是否包含“文章已通过”或“已达到发布标准”等关键词，以此来判断文章是否可以定稿。
   • 为了防止无限循环，我们引入了 max_revisions 参数。如果达到最大修订次数，即使 Editor 还没说通过，也会强制通过，这在实际项目中是防止死循环的重要机制（当然，通常会伴随人工介入的提醒）。
   • 返回 editor_feedback 和更新后的 status。
4. route_article 路由函数：
   • 这是 add_conditional_edges 的核心，它接收当前的 AgentState，并返回一个字符串，这个字符串将作为键去匹配 add_conditional_edges 的映射表，从而决定流程的走向。
   • 如果 status 是 approved，返回 "end"，流程终止。
   • 如果 status 是 revising，返回 "writer"，流程回到 Writer 节点。
5. 图的构建 (build_article_revision_graph)：
   • workflow = StateGraph(AgentState)：创建了一个基于 AgentState 的状态图。
   • workflow.add_node("writer", writer_agent.write_or_revise) 和 workflow.add_node("editor", editor_agent.review_article)：将 Writer 和 Editor 的方法注册为图中的节点。
   • workflow.set_entry_point("writer")：指定流程从 Writer 开始。
   • workflow.add_edge("writer", "editor")：Writer 完成后，无条件地将结果传递给 Editor。
   • workflow.add_conditional_edges("editor", route_article, {"writer": "writer", "end": END})：这是本期的核心！Editor 完成后，根据 route_article 函数的判断结果，决定是回到 writer 节点，还是走向 END（结束）。
6. 运行模拟 (if __name__ == "__main__":)：
   • 设置初始状态 initial_state，包括文章主题、空内容、空反馈和初始修订次数。
   • app.stream(initial_state) 迭代执行图，每次迭代都会打印当前的状态变化，让你清晰地看到流程是如何一步步推进的。

通过这段代码，你亲手构建了一个能够自我迭代、自我优化的内容创作流程。这不仅仅是两个 Agent 的简单连接，它是一个真正意义上的智能体网络 (Network of Agents)，能够模拟人类团队的协作和反馈循环。

坑与避坑指南

这种带有循环和反馈的复杂工作流，虽然强大，但也容易踩坑。作为高级讲师，我必须给你提前打好预防针。

1. 无限循环陷阱 (Infinite Loop Trap)：
   • 坑点：Editor 总是觉得不满意，Writer 总是改不到位，导致流程永远在 Writer-Editor 之间循环，永不停止。你的 API 费用会像流水一样哗哗地流走。
   • 避坑指南：
     • 设置最大修订次数 (max_revisions)：这是最直接有效的方法。在 Editor Agent 中加入一个计数器，当达到预设的最大修订次数时，强制结束循环（例如，标记为“需要人工复审”或直接“通过”）。我们在代码中已经实现了这一点。
     • 清晰的通过标准：在给 LLM (Editor) 的提示中，明确指出“通过”的标准，例如“文章满足以下所有条件：结构完整、论点清晰、语言流畅、无语法错误”。让 LLM 有明确的判断依据。
     • 逐步收敛的反馈：设计 Editor 的反馈，使其每次都尝试让 Writer 更接近目标，而不是每次都提出新的、不相关的问题。
2. 状态管理混乱 (Chaotic State Management)：
   • 坑点：AgentState 设计不合理，关键信息没有在节点间正确传递或更新，导致某个 Agent 拿不到它需要的数据，或者拿到的是旧数据。
   • 避坑指南：
     • TypedDict 明确结构：始终使用 TypedDict 定义 AgentState，它提供了类型检查，能有效避免拼写错误和数据结构不一致。
     • Annotated 用于合并策略：对于需要累加或特殊合并的字段（如 revision_count），使用 Annotated 配合 operator 模块，确保状态正确更新。
     • 节点职责单一：每个节点只负责更新自己相关的状态字段，不要试图在一个节点里修改所有状态，避免副作用。
     • 日志打印：在每个节点执行前后，打印关键状态信息，可以帮助你追踪数据流向和状态变化。
3. LLM 输出不确定性 (LLM Output Volatility)：
   • 坑点：LLM 有时候会“胡说八道”，Editor LLM 可能不会按照你预期的格式给出“通过”或“需要修改”的明确信号，导致路由函数判断失误。
   • 避坑指南：
     • 鲁棒的路由逻辑：route_article 函数应该对 LLM 的输出有足够的容错性。例如，不要只检查一个精确的字符串，而是检查多个同义词或短语 ("文章已通过" in feedback or "已达到发布标准" in feedback)。
     • 系统级提示工程：在 Agent 的 LLM 提示中，明确要求 LLM 以特定格式输出关键信息，例如“请在反馈的最后一行，明确写出 'STATUS: APPROVED' 或 'STATUS: REVISE'”。这样路由函数就能更准确地解析。
     • 温度参数调整：在关键决策点，可以适当降低 LLM 的 temperature 参数，使其输出更稳定和确定。
4. 调试复杂图困难 (Debugging Complex Graphs)：
   • 坑点：当图的节点和边增多时，追踪问题变得非常困难。
   • 避坑指南：
     • 使用 app.stream()：如代码所示，stream() 方法可以让你逐步观察每个节点执行后的状态变化，这是调试复杂图的利器。
     • 细致的 print 日志：在每个 Agent 内部加入详细的 print 语句，输出当前接收到的输入、生成的输出和做出的决策。
     • 可视化工具：LangGraph 提供了 get_graph().draw_mermaid_png() 等方法来生成图的可视化，这对于理解图的结构和潜在问题非常有帮助。

记住，构建复杂的 AI 系统，就像搭乐高积木，每一块都要放对位置，并且要预想好它可能带来的连锁反应。多思考，多实践，你就能成为 LangGraph 的高手！

📝 本期小结

恭喜你！在本期《LangGraph 多智能体专家课》中，我们攻克了一个高级且至关重要的模式——群组研讨会 (Network of Agents)。我们不再满足于线性的任务流，而是深入 LangGraph 的核心，利用 StateGraph、add_conditional_edges 和巧妙的 AgentState 管理，构建了一个能够自我迭代、自我反馈的文章创作与修订循环。

我们模拟了 AI 内容机构中 Writer 和 Editor 之间真实的“三方博弈”：Writer 提交初稿，Editor 严格审查并给出反馈，如果稿件未达标，则发回 Writer 进行修改，直到 Editor 满意为止。这个环形流不仅提升了内容产出的质量，更让你的 AI Agents 具备了真正的**“学习”和“优化”**能力。

你学会了如何：

• 巧妙设计 AgentState，特别是利用 Annotated 进行状态的累加合并。
• 构建 Writer 和 Editor 智能体，使其能够根据上下文进行撰写、审查和决策。
• 运用 add_conditional_edges 和路由函数 (route_article)，根据智能体的决策动态调整工作流。
• 以及，最重要的，掌握了防止无限循环和有效调试复杂图的高级技巧。

这种带有反馈循环的智能体网络，是构建任何复杂、适应性强 AI 系统的基石。无论是内容创作、代码审查、产品设计，还是医疗诊断，只要有迭代和优化的需求，这个模式都将大放异彩。

下一期，我们将在现有基础上，继续探索更多高级的协作模式和优化技巧。保持热情，下期再见！