Tutorial 1: LangGraph 入门

什么是 LangGraph?

LangGraph 是一个用于构建有状态、多步骤 AI 应用的框架。它的核心理念是:

将 Agent 工作流建模为图(Graph)

  • 节点(Node): 执行具体操作的函数

  • 边(Edge): 连接节点,定义执行顺序

  • 状态(State): 在节点之间传递的数据

LangGraph vs LangChain AgentExecutor

特性

AgentExecutor

LangGraph

流程控制

隐式,由 LLM 决定

显式,由图结构定义

状态管理

有限

完整的状态机

循环控制

有限

完全支持

人工干预

困难

原生支持

多 Agent

需要额外封装

原生支持

调试

困难

可视化图结构

持久化

需要额外实现

内置支持

安装配置

pip install langgraph langchain-openai

核心概念

1. StateGraph(状态图)

StateGraph 是 LangGraph 的核心类,用于定义工作流:

from langgraph.graph import StateGraph

# 定义状态类型
from typing import TypedDict

class MyState(TypedDict):
    messages: list
    current_step: str

# 创建图
graph = StateGraph(MyState)

2. Node(节点)

节点是执行具体操作的函数:

def my_node(state: MyState) -> dict:
    # 处理状态
    return {"current_step": "completed"}

# 添加节点
graph.add_node("my_node", my_node)

3. Edge(边)

边定义节点之间的连接:

# 普通边:A -> B
graph.add_edge("node_a", "node_b")

# 条件边:根据条件选择下一个节点
graph.add_conditional_edges(
    "node_a",
    routing_function,
    {"option1": "node_b", "option2": "node_c"}
)

第一个 LangGraph 程序

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI

# 1. 定义状态
class State(TypedDict):
    topic: str
    title: str
    outline: str

# 2. 定义节点函数
llm = ChatOpenAI(model="gpt-4o-mini")

def generate_title(state: State) -> dict:
    """生成标题"""
    topic = state["topic"]
    response = llm.invoke(f"为主题「{topic}」生成一个吸引人的标题,只输出标题")
    return {"title": response.content}

def generate_outline(state: State) -> dict:
    """生成大纲"""
    title = state["title"]
    response = llm.invoke(f"为标题「{title}」生成文章大纲,3-5个章节")
    return {"outline": response.content}

# 3. 构建图
graph = StateGraph(State)

# 添加节点
graph.add_node("generate_title", generate_title)
graph.add_node("generate_outline", generate_outline)

# 添加边
graph.add_edge(START, "generate_title")
graph.add_edge("generate_title", "generate_outline")
graph.add_edge("generate_outline", END)

# 4. 编译图
app = graph.compile()

# 5. 运行
result = app.invoke({"topic": "AI编程入门"})

print(f"主题: {result['topic']}")
print(f"标题: {result['title']}")
print(f"大纲:\n{result['outline']}")

图的可视化

# 生成 Mermaid 图
print(app.get_graph().draw_mermaid())

输出:

graph TD
    __start__ --> generate_title
    generate_title --> generate_outline
    generate_outline --> __end__

LangGraph 执行流程

┌─────────────────────────────────────────────────────────────┐
│                    LangGraph 执行流程                        │
│                                                              │
│  初始状态 ──► [START] ──► [Node 1] ──► [Node 2] ──► [END]   │
│     │            │           │            │           │      │
│     │            │           │            │           │      │
│     └────────────┴───────────┴────────────┴───────────┘      │
│                         状态传递                              │
│                                                              │
│  每个节点:                                                   │
│  1. 接收当前状态                                             │
│  2. 执行操作                                                 │
│  3. 返回状态更新                                             │
│  4. 状态合并后传递给下一个节点                               │
│                                                              │
└─────────────────────────────────────────────────────────────┘

自媒体项目预览

在接下来的教程中,我们将逐步构建一个完整的自媒体内容工作流:

# 最终我们将实现这样的工作流
from selfmedia_workflow import create_content_workflow

workflow = create_content_workflow()

# 执行完整流程
result = workflow.invoke({
    "topic": "AI Agent 开发",
    "target_platforms": ["微信公众号", "知乎"],
    "style": "专业但易懂"
})

# 结果包含
# - 话题分析
# - 内容大纲
# - 完整文章
# - 多平台适配版本
# - 发布结果

下一步

在下一个教程中,我们将深入学习 State 和 Graph 的基础知识。

Tutorial 2: State 与 Graph 基础