第 30 期 | 终局项目路演：完备的 AI 创作发行机构 (Capstone)

从接受指令、全网调研、主编写作、编辑审查、到图文排版，三十期功夫将在这 500 行的 Graph 中完美收官。

各位同学，欢迎来到《LangGraph 多智能体专家课》的最终回！我是你们的老朋友，陪伴了大家整整三十期的导师。

回首这三十个日夜，我们从最基础的单节点 LLM 调用，一路打怪升级，搞定了 Tool Calling、Memory 机制、Human-in-the-loop（人类介入）、以及各种复杂的 Routing 逻辑。大家还记得我们在第 1 期立下的 Flag 吗？我们要亲手打造一个**「AI 万能内容创作机构 (AI Content Agency)」**。

今天，就是检验真理的时刻。

在真实的企业级 AI 架构中，单打独斗的 Agent 早就活不下去了。你需要的是一个高度协同的“虚拟公司”。今天，我们将把之前 29 期写过的零散模块——运筹帷幄的主管 Planner、不知疲倦的研究员 Researcher、文笔犀利的主笔 Writer、吹毛求疵的编辑 Editor，以及精通排版的发行 Publisher——全部装进一个巨大的 StateGraph 中。

深呼吸，打开你的 IDE，让我们完成这最后一块拼图。

🎯 本期学习目标

在本次终局路演中，你将获得以下高阶收益：

1. 掌握全局状态管理 (Global State Management)：定义一个能承载整个内容生产生命周期的“数据总线”，让 5 个独立 Agent 共享且安全地修改上下文。
2. 构建复杂的条件流转与重试熔断 (Conditional Routing & Circuit Breaker)：实现 Editor 与 Writer 之间的“相爱相杀”（打回重写），并设置最大修改次数防止“死循环”。
3. 完成全链路代码组装 (End-to-End Orchestration)：将三十期的理论化作一个可直接运行的、具备工业级雏形的 Capstone 项目代码。

📖 原理解析

在写代码之前，老规矩，先看架构图。不要一上来就撸起袖子敲键盘，高级架构师的时间有 70% 是在画图和设计 State。

我们的 AI Content Agency 工作流如下：

1. 用户 (User) 丢出一个宽泛的选题（比如：“写一篇关于苹果 Vision Pro 销量惨淡的深度分析”）。
2. Planner 接收指令，拆解出详细的写作大纲和需要调研的核心问题。
3. Researcher 根据大纲，调用搜索引擎工具（模拟）获取全网最新数据。
4. Writer 拿着大纲和调研数据，挥洒汗水写出初稿。
5. Editor 登场，用极其严苛的标准审阅初稿。如果不行，给出修改意见（Review Comments），打回给 Writer 重新写。
6. Publisher：一旦 Editor 审核通过（或者达到了最大重试次数，被迫妥协），Publisher 将接手进行最终的 Markdown 排版和配图（模拟），并输出终稿。

让我们用 Mermaid 将这个宏大的工作流具象化：

graph TD
    %% 定义样式
    classDef human fill:#f9f,stroke:#333,stroke-width:2px;
    classDef agent fill:#bbf,stroke:#333,stroke-width:2px;
    classDef router fill:#fbf,stroke:#333,stroke-width:2px;
    classDef endnode fill:#bfb,stroke:#333,stroke-width:2px;

    START((开始)) --> UserInput[用户输入选题]:::human
    UserInput --> Planner[Planner: 拆解大纲与调研方向]:::agent
    Planner --> Researcher[Researcher: 执行全网调研]:::agent
    Researcher --> Writer[Writer: 撰写文章初稿/修改稿]:::agent
    Writer --> Editor[Editor: 质量审查]:::agent
    
    Editor --> EditorRouter{审核通过了吗?}:::router
    
    EditorRouter -- 否 (打回修改) --> Writer
    EditorRouter -- 是 (或达到最大修改次数) --> Publisher[Publisher: 图文排版与定稿]:::agent
    
    Publisher --> END((结束)):::endnode

    %% 状态说明框
    subgraph State [全局状态 State (数据总线)]
        direction LR
        topic[选题]
        outline[大纲]
        research[调研数据]
        draft[当前草稿]
        comments[修改意见]
        revision_count[修改次数]
    end

导师犀利点评： 看到这个图，很多同学可能会问：“老师，为啥不让 Researcher 在 Writer 修改时再次去搜索？” 这是一个极好的问题！在工业界，这取决于你的成本和延迟预算。为了保证今天这 500 行代码的清晰度，我们设定 Researcher 只在初期进行一次深度调研，后续 Writer 的修改仅基于 Editor 的反馈。懂得在架构上做减法，才是真正的高手。

💻 实战代码演练

废话不多说，上代码。这段代码是我们三十期心血的结晶。为了让大家能直接跑通，我使用了 langchain_openai 并对部分耗时的 Tool 进行了 Mock（模拟）。

请仔细阅读代码中的双语注释，里面藏着大量的实战细节。

import operator
from typing import TypedDict, Annotated, List
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

# ==========================================
# 1. 定义全局状态 (Define the Global State)
# ==========================================
# 导师注：State 就是我们 Agency 的中央数据库。
# 注意 revision_count 的 Annotated 用法，它会在每次打回时自动 +1。
class AgencyState(TypedDict):
    topic: str                      # 用户输入的初始选题
    outline: str                    # Planner 产出的大纲
    research_data: str              # Researcher 产出的调研素材
    draft: str                      # Writer 产出的文章草稿
    review_comments: str            # Editor 给出的修改意见
    revision_count: Annotated[int, operator.add] # 记录被 Editor 打回的次数
    final_article: str              # Publisher 产出的终稿

# 初始化 LLM (建议使用 GPT-4o 或 Claude-3.5-Sonnet 以获得最佳的多智能体效果)
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# ==========================================
# 2. 定义各个 Agent 节点 (Define Agent Nodes)
# ==========================================

def planner_node(state: AgencyState):
    """Planner: 负责将宽泛的选题拆解为结构化的大纲"""
    print("👨‍💼 [Planner] 正在拆解选题并制定大纲...")
    sys_msg = SystemMessage(content="你是一个资深媒体主编。请根据用户的选题，输出一个包含 3-4 个核心段落的详细写作大纲。")
    user_msg = HumanMessage(content=f"选题：{state['topic']}")
    
    response = llm.invoke([sys_msg, user_msg])
    # 状态更新：将大纲写入 State，同时初始化修改次数为 0
    return {"outline": response.content, "revision_count": 0}

def researcher_node(state: AgencyState):
    """Researcher: 根据大纲进行素材搜集 (这里用模拟数据代替真实的 Search Tool)"""
    print("🕵️‍♂️ [Researcher] 正在全网搜集深度素材...")
    # 在真实项目中，这里会 bind_tools(SearchTool) 并执行循环。
    # 为保证 Capstone 顺畅运行，我们让 LLM 直接基于大纲生成伪调研数据。
    sys_msg = SystemMessage(content="你是一个王牌研究员。请根据主编的大纲，提供丰富的数据、案例和事实作为写作素材。")
    user_msg = HumanMessage(content=f"大纲如下：\n{state['outline']}")
    
    response = llm.invoke([sys_msg, user_msg])
    return {"research_data": response.content}

def writer_node(state: AgencyState):
    """Writer: 结合大纲、素材以及可能的修改意见进行撰写"""
    print(f"✍️ [Writer] 正在奋笔疾书 (当前修改次数: {state.get('revision_count', 0)})...")
    
    sys_prompt = "你是一个金牌主笔。请根据大纲和调研数据写出引人入胜的文章。"
    # 如果有 Editor 的修改意见，Writer 必须听从
    if state.get("review_comments"):
        sys_prompt += f"\n\n注意！这是主编的修改意见，请务必遵从：\n{state['review_comments']}"
        
    sys_msg = SystemMessage(content=sys_prompt)
    user_msg = HumanMessage(content=f"大纲：\n{state['outline']}\n\n素材：\n{state['research_data']}")
    
    response = llm.invoke([sys_msg, user_msg])
    return {"draft": response.content}

def editor_node(state: AgencyState):
    """Editor: 严苛的质量把控者。决定文章是进入下一环节还是打回重写"""
    print("🧐 [Editor] 正在用放大镜审阅草稿...")
    sys_msg = SystemMessage(content="""
    你是一个极其严苛的文字总监。请审阅草稿。
    如果你觉得完美，请只回复 "APPROVED"。
    如果觉得有待改进，请列出具体的修改意见（不要自己改，指出问题即可）。
    """)
    user_msg = HumanMessage(content=f"当前草稿：\n{state['draft']}")
    
    response = llm.invoke([sys_msg, user_msg])
    comments = response.content
    
    if "APPROVED" in comments.upper():
        print("   ✅ [Editor] 审核通过！")
        return {"review_comments": "APPROVED"}
    else:
        print("   ❌ [Editor] 审核不通过，打回重写！")
        # 重点：打回时，除了记录 comments，还要让 revision_count + 1
        return {"review_comments": comments, "revision_count": 1}

def publisher_node(state: AgencyState):
    """Publisher: 最终的排版与美化"""
    print("🖨️ [Publisher] 正在进行精美的 Markdown 排版并生成终稿...")
    sys_msg = SystemMessage(content="你是一个排版大师。请为文章添加合适的 Markdown 标题、加粗重点，并在合适的地方插入 [图片占位符]。")
    user_msg = HumanMessage(content=f"定稿内容：\n{state['draft']}")
    
    response = llm.invoke([sys_msg, user_msg])
    return {"final_article": response.content}

# ==========================================
# 3. 核心路由逻辑 (Conditional Routing)
# ==========================================
def editor_router(state: AgencyState) -> str:
    """
    决定 Editor 之后的走向。
    熔断机制：如果修改次数超过 2 次，强制通过，防止死循环。
    """
    if state["review_comments"] == "APPROVED":
        return "to_publisher"
    elif state["revision_count"] >= 2:
        print("   ⚠️ [System] 达到最大修改次数，触发熔断，强制进入排版环节！")
        return "to_publisher"
    else:
        return "back_to_writer"

# ==========================================
# 4. 组装终极 Graph (Build the Capstone Graph)
# ==========================================
workflow = StateGraph(AgencyState)

# 添加所有节点
workflow.add_node("Planner", planner_node)
workflow.add_node("Researcher", researcher_node)
workflow.add_node("Writer", writer_node)
workflow.add_node("Editor", editor_node)
workflow.add_node("Publisher", publisher_node)

# 定义边 (工作流顺序)
workflow.add_edge(START, "Planner")
workflow.add_edge("Planner", "Researcher")
workflow.add_edge("Researcher", "Writer")
workflow.add_edge("Writer", "Editor")

# 添加条件边 (Editor 的抉择)
workflow.add_conditional_edges(
    "Editor",
    editor_router,
    {
        "to_publisher": "Publisher",
        "back_to_writer": "Writer"
    }
)

workflow.add_edge("Publisher", END)

# 编译 Graph
agency_app = workflow.compile()

# ==========================================
# 5. 见证奇迹的时刻：运行演示 (Run the Demo)
# ==========================================
if __name__ == "__main__":
    print("\n🚀 欢迎来到 AI Content Agency！系统启动中...\n" + "="*50)
    
    initial_state = {
        "topic": "解析2024年AI大模型在医疗领域的真实落地案例与挑战"
    }
    
    # 使用 stream 模式运行，以便我们能看到每一步的执行过程
    for output in agency_app.stream(initial_state):
        # 打印当前执行完的节点名称
        for key, value in output.items():
            print(f"--- 节点 [{key}] 执行完毕 ---")
            
    print("\n" + "="*50 + "\n🎉 终稿已生成！\n")
    # 获取最终状态并打印终稿
    final_state = agency_app.get_state(config={"configurable": {"thread_id": "1"}}).values 
    # 注：如果不使用 checkpointer，直接从 stream 的最后一次输出获取即可。
    # 为了简化，我们可以直接这样拿：
    print(value.get("final_article", "生成失败"))

坑与避坑指南 (高阶视角的排错经验)

各位同学，代码跑通了不代表万事大吉。在过去带团队落地这种多智能体架构时，我踩过无数的坑。今天作为毕业礼物，把我压箱底的避坑指南送给你们：

💣 坑一：Editor 与 Writer 的“无限死循环” (The Infinite Death Loop)

现象：Editor 觉得 Writer 写得像一坨屎，永远不给 "APPROVED"；而 Writer 偏执地坚持自己的写法，或者 LLM 的能力上限就到这了，改不出 Editor 想要的效果。结果 Graph 就在这两个节点之间疯狂循环，直到把你的 OpenAI API 余额烧光。 避坑：必须实现熔断机制（Circuit Breaker）。在上面的代码中，我们引入了 revision_count 并在 editor_router 中强制判断 >= 2 时跳出循环。在企业级项目中，你甚至可以引入一个 Human_Intervention 节点，当修改超过 3 次时，直接发飞书/钉钉消息让真人主编介入。

💣 坑二：State 状态无限膨胀导致 Context Window 撑爆

现象：如果你在 State 中把每一次的 draft 都用 Annotated[list, operator.add] 存成一个数组，经历几次修改后，Prompt 的长度会呈指数级增长，直接触发 LLM 的 Token 上限报错。 避坑：区分“需要追加的状态”和“需要覆盖的状态”。在我们的 AgencyState 中，draft 和 review_comments 都是 str 类型，这意味着每一次执行都会覆盖旧的值。Writer 只需要知道“当前最新的草稿”和“最新的修改意见”即可，不需要知道上古时期的废稿。

💣 坑三：Prompt 漂移 (Prompt Drift)

现象：Writer 节点在接收到 Editor 极其严厉的批评后，不仅修改了文章，还会在文章开头加上一句：“好的主编，我非常抱歉，我已经根据您的要求修改了文章，以下是正文...”。这导致最终输出给 Publisher 的文本包含了这种废话。 避坑：这是 LLM 的“讨好型人格”导致的。在 Writer 的 System Prompt 中，必须加上极其严厉的约束指令："你只允许输出文章正文，绝不允许输出任何解释性语句、道歉或与正文无关的对话！" 甚至可以使用 LangChain 的 Structured Output (Pydantic) 来强制约束输出格式。

📝 本期小结

各位同学，伴随着控制台里打印出的那句 🎉 终稿已生成！，我们这 30 期的《LangGraph 多智能体专家课》正式画上了句号。

从第 1 期懵懵懂懂地认识 Graph 的概念，到今天我们用区区几百行代码，就构建起了一个包含指令分发、全网检索、内容创作、质量审核、循环修改、图文排版的完备 AI 机构。你们不仅掌握了 LangGraph 的底层逻辑，更建立起了一种高阶的“多智能体架构思维”。

请记住：LangGraph 只是一个框架，它赋予了 Agent 骨架；而你们设计的 State 流转和精心调优的 Prompt，才是赋予这个系统灵魂的关键。

毕业不是结束，而是新的开始。现在的你，已经具备了去重构企业级复杂工作流的能力。无论是做金融研报 Agent、代码审查 Agent，还是我们今天做的 Content Agency，底层的“道”都是相通的。

带上这 30 期的兵器谱，去 AI 的江湖里大杀四方吧！别忘了，遇到 Bug 搞不定的时候，回来看一眼导师的教程。我们，江湖再见！🚀