LangGraph 入门实战：用流水线思维打造 AI 文章生成器

from typing import TypedDict

class ArticleState(TypedDict):
    """文章生成的状态定义"""
    title: str              # 文章标题
    outline: str            # 文章大纲
    content: str            # 正文内容
    polished_content: str   # 润色后的内容
    current_step: str       # 当前执行到哪一步

def input_title(state: ArticleState) -> ArticleState:
    """接收并确认文章标题"""
    print("\n=== 第1步：标题确认 ===")
    title = state.get("title", "")

    if not title:
        raise ValueError("标题不能为空")

    print(f"标题: {title}")

    return {
        **state,  # 保留原有状态
        "current_step": "title_confirmed"  # 更新当前步骤
    }

def generate_outline(state: ArticleState) -> ArticleState:
    """基于标题生成文章大纲"""
    print("\n=== 第2步：大纲生成 ===")
    title = state["title"]  # 从状态中获取标题

    # 构建提示词
    prompt = f"""请为以下标题生成一个文章大纲：

标题：{title}

要求：
1. 包含3-5个主要章节
2. 每个章节有简短的描述
3. 逻辑清晰，结构合理"""

    # 调用 LLM 生成大纲
    response = llm.invoke(prompt)
    outline = response.content

    print(outline)

    return {
        **state,
        "outline": outline,  # 保存大纲
        "current_step": "outline_generated"
    }

def write_content(state: ArticleState) -> ArticleState:
    """根据大纲撰写完整文章"""
    # 读取大纲，调用 LLM 生成正文
    # ...
    return {**state, "content": content}

def polish_content(state: ArticleState) -> ArticleState:
    """优化和润色文章内容"""
    # 读取正文，调用 LLM 润色
    # ...
    return {**state, "polished_content": polished_content}

from langgraph.graph import StateGraph, END

def create_article_workflow():
    """创建文章生成工作流"""

    # 1. 创建状态图
    workflow = StateGraph(ArticleState)

    # 2. 添加节点
    workflow.add_node("input_title", input_title)
    workflow.add_node("generate_outline", generate_outline)
    workflow.add_node("write_content", write_content)
    workflow.add_node("polish_content", polish_content)

    # 3. 设置入口点（从哪个节点开始）
    workflow.set_entry_point("input_title")

    # 4. 添加边，定义执行顺序
    workflow.add_edge("input_title", "generate_outline")      # 标题 → 大纲
    workflow.add_edge("generate_outline", "write_content")    # 大纲 → 正文
    workflow.add_edge("write_content", "polish_content")      # 正文 → 润色
    workflow.add_edge("polish_content", END)                  # 润色 → 结束

    # 5. 编译图
    return workflow.compile()

def main():
    # 创建工作流
    app = create_article_workflow()

    # 准备初始状态
    initial_state = {
        "title": "人工智能在医疗领域的应用",
        "outline": "",
        "content": "",
        "polished_content": "",
        "current_step": "initial"
    }

    # 执行工作流
    final_state = app.invoke(initial_state)

    # 输出最终结果
    print(final_state["polished_content"])

workflow.add_node("quality_check", quality_check_func)
workflow.add_edge("polish_content", "quality_check")
workflow.add_edge("quality_check", END)

# 1. 定义状态
class MyState(TypedDict):
    field1: str
    field2: str

# 2. 定义节点函数
def node1(state: MyState) -> MyState:
    # 处理逻辑
    return {**state, "field1": "new_value"}

# 3. 构建图
workflow = StateGraph(MyState)
workflow.add_node("node1", node1)
workflow.add_node("node2", node2)
workflow.set_entry_point("node1")
workflow.add_edge("node1", "node2")
workflow.add_edge("node2", END)

# 4. 运行
app = workflow.compile()
result = app.invoke(initial_state)