第 05 期 | 初见成效：把「调查-写作」闭环跑起来

🎯 本期学习目标

嘿，各位未来的 AI 架构师们，欢迎回到我们的 LangGraph 专家课！上期我们聊了 LangGraph 的哲学，是时候把那些理论变成实实在在的代码了。别急着构建复杂的决策树，我们先从最基础、但也是最关键的一步开始：让我们的 AI 内容机构 MVP 版本跑起来！

本期，我们将：

• 掌握 LangGraph 顺序执行流的核心机制： 理解 add_node 和 add_edge 如何编织出固定的工作流。
• 将「研究员 (Researcher)」和「写作者 (Writer)」Agent 成功融入 LangGraph 工作流： 它们不再是独立的脚本，而是我们机构体系中的一环。
• 学会定义和管理 Graph 状态 (State)： 确保信息在不同 Agent 之间流畅、准确地传递。
• 成功运行「AI 内容机构」的「调查-写作」MVP 闭环： 亲手见证第一个内容创作流程的诞生！

📖 原理解析

为什么从固定顺序开始？—— MVP 思维的胜利

在软件工程领域，我们总强调一个概念：MVP (Minimum Viable Product)，最小可行产品。对于我们的 AI 内容机构，MVP 是什么？就是能够完成一次「从主题到内容」的最小闭环。最直接的方式，就是让一个研究员先去调查，然后把调查结果交给一个写作者去撰写。这中间，没有复杂的决策，没有重试，没有分支，就是一条直线。

为什么要这样？

1. 降低复杂性： 初学者最容易犯的错误就是一开始就想把所有逻辑都塞进去。LangGraph 固然强大，但如果你连最简单的顺序流都玩不转，何谈复杂的条件分支？先跑起来，验证核心功能，这是王道。
2. 快速验证： 通过固定顺序，我们可以迅速验证各个 Agent 的功能是否正常，以及它们之间的数据传递是否顺畅。一旦发现问题，排查起来也更简单。
3. 奠定基础： 顺序流是所有复杂图的基础。你理解了 add_node 和 add_edge 的精髓，未来引入 add_conditional_edges 也会水到渠成。这就像盖房子，地基打不牢，上层建筑再华丽也是空中楼阁。

LangGraph 的基本构建块：Node, Edge, State

还记得我们上期提到的 LangGraph 核心吗？它就是由 节点 (Node) 和 边 (Edge) 构成的有向图。

• 节点 (Node)： 在我们的「AI 内容机构」中，每个 Agent（比如 Researcher、Writer）就是一个节点。节点接收输入，执行任务，然后输出结果。这个「执行任务」可以是调用一个 LLM，也可以是执行一个工具，甚至只是一个简单的 Python 函数。
• 边 (Edge)： 边定义了节点之间的流转方向。从节点 A 到节点 B，意味着节点 A 完成任务后，其输出将作为节点 B 的输入（或者说，状态会更新，然后 B 读取状态）。
• 状态 (State)： 这是 LangGraph 的灵魂！Graph 的所有节点共享一个可变状态。当一个节点完成任务后，它会更新这个共享状态。下一个节点会从这个更新后的状态中读取它所需的信息。这种状态传递机制，是多 Agent 协作的基础。

本期，我们主要利用 add_node 来注册我们的 Agent，并用 add_edge 来定义它们之间的固定顺序流转。

Mermaid 图解：初见成效的「调查-写作」工作流

这是我们本期要构建的核心工作流，非常简洁，直观明了：

graph TD
    A[开始] --> B(研究员 Researcher);
    B --> C(写作者 Writer);
    C --> D[结束];

这个图清晰地展示了：

1. 开始 (Start)：工作流的起点，我们在这里传入初始的主题。
2. 研究员 (Researcher)：接收主题，进行调查，并将调查结果更新到共享状态。
3. 写作者 (Writer)：从共享状态中读取调查结果，然后撰写内容，并将草稿更新到共享状态。
4. 结束 (End)：工作流的终点，我们可以从最终状态中获取撰写好的内容。

是不是很简单？但正是这种简单，构成了复杂系统的基石。

💻 实战代码演练

好了，理论说得再好，不如撸起袖子干一场！我们将用 Python 来实现这个「调查-写作」闭环。

import operator
from typing import Annotated, List, Tuple, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_core.runnables import Runnable
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END

# =============================================================================
# 1. 定义我们的 AI 内容机构的共享状态 (Agency State)
#    这是整个 Graph 运行过程中，所有节点共享和更新的数据。
# =============================================================================
class AgencyState(TypedDict):
    """
    代表 AI 内容机构在内容创作过程中的共享状态。
    This represents the shared state of our AI Content Agency during content creation.
    """
    topic: str  # 创作的主题 The topic for content creation.
    research_data: str  # 研究员收集到的数据 Data collected by the researcher.
    draft_content: str  # 写作者撰写的初稿 Draft content written by the writer.
    # messages: Annotated[List[BaseMessage], operator.add] # 如果需要存储消息历史，可以这样定义
    # (Optional) If you need to store message history, you can define it like this.

# =============================================================================
# 2. 模拟我们的 Agent (节点函数)
#    为了聚焦 LangGraph 本身，我们这里先用简单的 Python 函数来模拟 Agent 的行为。
#    实际项目中，这里会是更复杂的 LangChain Agent 或自定义的 Runnable。
# =============================================================================

# 模拟研究员 Agent
def research_agent(state: AgencyState) -> AgencyState:
    """
    模拟研究员 Agent。
    它接收当前状态，从 'topic' 中获取主题，模拟进行调查，并更新 'research_data'。
    Simulates the Researcher Agent.
    It takes the current state, gets the topic from 'topic', simulates research,
    and updates 'research_data'.
    """
    print(f"\n--- 研究员正在调查主题: {state['topic']} ---") # Researcher is researching topic.
    # 模拟一个耗时操作和调查结果
    # Simulate a time-consuming operation and research results.
    simulated_research_result = f"关于 '{state['topic']}' 的初步调查结果：\n" \
                                "1. 市场趋势显示，此类内容近期关注度高。\n" \
                                "2. 主要竞品内容侧重于技术原理，缺乏应用场景。\n" \
                                "3. 用户普遍反馈希望有更多实战案例和避坑指南。"
    
    print("--- 研究员调查完成，更新状态 ---") # Researcher finished research, updating state.
    # 更新状态，返回新状态字典
    # Update state and return the new state dictionary.
    return {"research_data": simulated_research_result}

# 模拟写作者 Agent
def writer_agent(state: AgencyState) -> AgencyState:
    """
    模拟写作者 Agent。
    它接收当前状态，从 'research_data' 中获取调查结果，模拟撰写内容，并更新 'draft_content'。
    Simulates the Writer Agent.
    It takes the current state, gets research data from 'research_data', simulates writing,
    and updates 'draft_content'.
    """
    print(f"\n--- 写作者正在基于调查结果撰写内容 ---") # Writer is writing content based on research results.
    print(f"调查结果摘要:\n{state['research_data'][:100]}...") # Research results summary.

    # 模拟一个耗时操作和写作过程
    # Simulate a time-consuming operation and writing process.
    simulated_draft = f"标题：深入浅出：{state['topic']} 实战指南与避坑\n\n" \
                      f"前言：在当今快速发展的AI时代，{state['topic']} 已成为不可忽视的关键技术。本指南将结合最新市场趋势和用户反馈，为您提供一份详尽的实战策略。\n\n" \
                      f"第一部分：市场洞察与竞品分析\n{state['research_data']}\n\n" \
                      f"第二部分：核心技术原理与应用场景\n（此处省略具体技术细节，聚焦实战）\n\n" \
                      f"第三部分：常见问题与避坑指南\n（此处基于用户反馈提供建议）\n\n" \
                      f"结语：掌握 {state['topic']}，赋能未来AI应用！"
    
    print("--- 写作者撰写完成，更新状态 ---") # Writer finished writing, updating state.
    # 更新状态，返回新状态字典
    # Update state and return the new state dictionary.
    return {"draft_content": simulated_draft}

# =============================================================================
# 3. 构建 LangGraph 工作流
#    使用 StateGraph 来定义我们的 Agent 节点和它们之间的顺序流转。
# =============================================================================

# 实例化 StateGraph，并指定我们定义的共享状态类型
# Instantiate StateGraph, specifying our defined shared state type.
workflow = StateGraph(AgencyState)

# 注册节点 (Add Nodes)
# 每个节点都关联一个我们上面定义的 Agent 函数。
# Each node is associated with an Agent function we defined above.
workflow.add_node("researcher", research_agent) # 注册研究员节点 Register the researcher node.
workflow.add_node("writer", writer_agent)       # 注册写作者节点 Register the writer node.

# 设置入口点 (Set Entry Point)
# 定义 Graph 从哪里开始执行。
# Define where the Graph starts execution.
workflow.set_entry_point("researcher") # 从研究员开始 Start from the researcher.

# 添加边 (Add Edges)
# 定义节点之间的顺序流转。
# Define the sequential flow between nodes.
workflow.add_edge("researcher", "writer") # 研究员完成任务后，流转到写作者。
                                          # After the researcher completes its task, flow to the writer.

# 设置结束点 (Set Finish Point)
# 定义 Graph 在哪里结束执行。
# Define where the Graph finishes execution.
workflow.set_finish_point("writer") # 写作者完成任务后，Graph 结束。
                                    # After the writer completes its task, the Graph finishes.

# 编译 Graph
# 将定义好的工作流编译成一个可执行的 Runnable。
# Compile the defined workflow into an executable Runnable.
app = workflow.compile()

# =============================================================================
# 4. 运行我们的 AI 内容机构 MVP
#    传入初始状态，并观察 Graph 的执行过程和最终结果。
# =============================================================================
if __name__ == "__main__":
    print("--- AI 内容机构 MVP 启动 ---") # AI Content Agency MVP started.

    # 定义初始状态：我们想要创作的主题
    # Define initial state: the topic we want to create content about.
    initial_state = {"topic": "LangGraph 多智能体协同在内容创作中的应用", "research_data": "", "draft_content": ""}

    # 运行 Graph
    # 调用 app.stream() 可以逐步看到每个节点的状态更新。
    # Calling app.stream() allows you to see state updates at each node step by step.
    final_state = None
    for s in app.stream(initial_state):
        # s 是一个字典，键是当前正在执行的节点名，值是该节点执行后的状态更新。
        # s is a dictionary where the key is the name of the currently executing node,
        # and the value is the state update after that node's execution.
        print(s)
        print("----") # Separator for readability.
        final_state = s # Keep track of the last state for final output.

    print("\n--- AI 内容机构 MVP 运行结束 ---") # AI Content Agency MVP finished running.

    # 打印最终生成的内容
    # Print the final generated content.
    if final_state and "writer" in final_state: # 确保最终状态存在且包含 writer 节点的结果
        # Ensure final state exists and contains the result from the writer node.
        print("\n🎉 最终生成的内容草稿 (Final Generated Content Draft):")
        print(final_state["writer"]["draft_content"])
    elif final_state and END in final_state: # 如果最终状态直接是 END，则从 END 节点获取
        print("\n🎉 最终生成的内容草稿 (Final Generated Content Draft):")
        print(final_state[END]["draft_content"])
    else:
        print("\n⚠️ 未能获取最终内容草稿。") # Failed to retrieve final content draft.

代码解析：步步为营

1. AgencyState 定义 (TypedDict)

   • 我们用 TypedDict 定义了 AgencyState。这是整个 Graph 共享的唯一数据源。
   • topic：工作开始时，我们给机构一个主题。
   • research_data：研究员会把调查结果写到这里。
   • draft_content：写作者会把初稿写到这里。
   • 划重点： 这种单一的、可变的状态是 LangGraph 的核心。每个节点都从这里读取它需要的信息，并把它的输出写回到这里。

2. 模拟 Agent (research_agent, writer_agent)

   • 为了让大家专注于 LangGraph 的结构，我这里用简单的 Python 函数来模拟 Agent。
   • 每个函数都接收一个 AgencyState 对象（或其子集），执行一些模拟的“工作”，然后返回一个字典，这个字典会用来更新当前的 AgencyState。
   • 注意： 返回的字典会与当前状态进行合并（默认是浅合并，对于字典值会覆盖，对于列表值 Annotated[List[..., operator.add]] 会追加）。这里我们直接覆盖 research_data 和 draft_content。

3. 构建 LangGraph 工作流 (workflow = StateGraph(AgencyState))

   • StateGraph(AgencyState)：初始化一个状态图，并明确告诉它我们使用的共享状态类型是 AgencyState。
   • workflow.add_node("researcher", research_agent)：将我们的 research_agent 函数注册为名为 "researcher" 的节点。当 Graph 流转到这个节点时，就会执行 research_agent 函数。
   • workflow.set_entry_point("researcher")：设定 Graph 的入口点。这意味着当 Graph 启动时，它会从 "researcher" 节点开始执行。
   • workflow.add_edge("researcher", "writer")：添加一条从 "researcher" 到 "writer" 的边。这表示 "researcher" 节点执行完毕后，控制流将无条件地转移到 "writer" 节点。
   • workflow.set_finish_point("writer")：设定 Graph 的结束点。当 "writer" 节点执行完毕后，Graph 就会停止，并返回最终的状态。
   • app = workflow.compile()：将我们定义好的工作流编译成一个可执行的 LangChain Runnable 对象。这个 app 对象就是我们整个 AI 内容机构的「大脑」。

4. 运行我们的 MVP (app.stream(initial_state))

   • 我们创建一个 initial_state 字典，只包含 topic。research_data 和 draft_content 初始化为空字符串。
   • app.stream(initial_state)：启动 Graph。它会返回一个迭代器，每次迭代都会输出一个字典，表示当前节点执行后的状态变化。这对于调试和理解流程非常有帮助。
   • 最终，我们从 final_state 中提取 draft_content，这就是我们机构完成的第一个作品！

坑与避坑指南

作为一名架构师，我不仅要教你“怎么做”，更要告诉你“别掉进哪些坑”。

1. 状态设计陷阱：信息丢失或污染

   • 坑： AgencyState 设计不当，例如，忘记将某个 Agent 的输出字段添加到状态中，导致下一个 Agent 无法获取所需信息。或者，不同 Agent 意外地覆盖了对方的关键数据。
   • 避坑：
     • 提前规划状态： 在写代码之前，画个图，列出每个 Agent 需要什么输入，会产生什么输出。这些输入输出就是你的 AgencyState 的字段。
     • 单一职责原则 (SRP) 应用于状态字段： 尽量让每个状态字段的更新职责归属于一个特定的 Agent。例如，research_data 应该主要由 researcher 更新，draft_content 主要由 writer 更新。
     • 利用 Annotated 和 operator.add 处理列表： 如果你的状态中需要累积列表（比如消息历史），使用 Annotated[List[BaseMessage], operator.add] 可以确保新消息被追加而不是覆盖。我们本期没有用到，但未来会用到。

2. 节点职责不清晰：大杂烩 Agent

   • 坑： 把太多逻辑塞到一个 Agent 节点里，比如一个 Agent 既做研究又做写作。这样会导致节点变得臃肿，难以维护、调试和复用。
   • 避坑：
     • 保持节点粒度适中： 每个节点应该只负责一个明确的、逻辑上独立的任务。本期的 research_agent 只负责研究，writer_agent 只负责写作，就是很好的范例。
     • 便于调试： 职责单一的节点更容易测试和定位问题。如果一个节点出错了，你立刻知道是研究逻辑有问题，还是写作逻辑有问题。

3. 调试困难：黑盒运行

   • 坑： 直接调用 app.invoke() 而不打印中间状态，导致 Graph 运行结果不符合预期时，你不知道是哪个环节出了问题。
   • 避坑：
     • 利用 app.stream()： 就像我们代码中示范的，stream() 方法是你的最佳调试伙伴。它能让你看到每个节点执行后的状态更新，帮助你追踪数据流和逻辑。
     • 在 Agent 函数内部打印日志： 在 research_agent 和 writer_agent 内部，我加入了 print 语句。这在开发阶段是极其有用的，能让你直观地看到每个 Agent 正在做什么，以及它接收到了什么输入。

4. 何时引入条件逻辑？过早优化

   • 坑： 在 MVP 还没跑起来的时候，就开始琢磨着如何加条件判断、循环重试、人工审批等复杂逻辑。
   • 避坑：
     • 先跑起来，再优化： 再次强调 MVP 思维。确保最简单的「调查-写作」闭环能稳定运行，输出符合预期的内容。这是你迈向复杂工作流的坚实基础。
     • 逐步迭代： 等你对顺序流掌握炉火纯青后，再考虑引入 add_conditional_edges 来实现复杂的决策逻辑。那将是下一阶段的挑战，但有了本期的基础，你会更有信心。

📝 本期小结

恭喜你！我们刚刚亲手搭建并运行了「AI 内容机构」的第一个 MVP 版本——一个能够从主题到内容草稿的「调查-写作」闭环。

本期我们：

• 明确了 MVP 思维 在复杂系统开发中的重要性，以及为什么从固定顺序流开始是明智之举。
• 深入理解了 LangGraph 的 Node、Edge 和 State 概念，它们是构建一切复杂工作流的基石。
• 通过实战代码，将模拟的 Researcher 和 Writer Agent 成功编排进了 LangGraph，并利用 add_node、set_entry_point、add_edge 和 set_finish_point 构建了顺序流。
• 掌握了状态管理的关键，确保信息在 Agent 之间无缝传递。
• 学习了在开发过程中常见的陷阱和避坑策略，这都是作为一名高级架构师的宝贵经验。

现在，你已经有了一个能跑起来的「骨架」。下期，我们将开始思考：如果研究员发现主题太模糊，或者写作者觉得调查数据不够，我们如何让它们进行反馈和重试？没错，我们将开始探索 LangGraph 的条件逻辑和循环机制，让我们的机构变得更加智能和健壮！

准备好了吗？下一期，精彩继续！