1.1 LangGraph概述与架构设计

1.1 LangGraph概述与架构设计

本节导读：通过本节学习，你将掌握LangGraph的核心定位和架构优势，理解其与传统智能体框架的本质区别，能够根据项目需求判断是否选择LangGraph。

学习目标

• 理解LangGraph的定位：低级编排框架vs高级智能体框架
• 掌握LangGraph的四大核心能力：持久化执行、流式输出、人类在回路、综合记忆
• 能够对比分析LangGraph与LangChain的适用场景
• 学会根据项目复杂度选择合适的技术栈

核心概念

LangGraph是什么？LangGraph是一个低级编排框架（Low-level Orchestration Framework）和运行时，专门用于构建、管理和部署长时间运行、有状态（long-running, stateful）的AI智能体。与传统的智能体框架相比，LangGraph提供了更底层、更灵活的控制能力，让开发者能够精确控制智能体的行为和状态。

定位对比

LangGraph与LangChain代表了两种不同的设计哲学：

LangChain（高级抽象框架）：

• 采用预定义的智能体循环模式
• 提供开箱即用的组件和模板
• 适合快速开发和简单任务
• 抽象层次较高，学习曲线平缓
• 内置丰富的工具和记忆系统

LangGraph（低级编排框架）：

• 采用自定义图结构控制流程
• 提供细粒度的状态管理和执行控制
• 适合复杂长时间运行任务
• 抽象层次较低，学习曲线陡峭
• 强调状态持久化和流式输出能力

四大核心能力详解

LangGraph的强大之处在于其专注于四个关键能力，这些能力对于构建复杂的长时间运行任务至关重要：

1. 持久化执行（Persistence）

持久化执行是LangGraph最核心的能力之一。它允许智能体在长时间运行过程中保存执行状态，即使程序崩溃或重启，也能够从上次中断的地方继续执行。

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, add_messages

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    step_count: int
    current_task: str

def process_data_node(state: AgentState):
    """数据处理节点"""
    print(f"处理数据步骤 {state['step_count']}: {state['current_task']}")
    
    # 模拟数据处理
    step = state['step_count'] + 1
    task = f"已处理: {state['current_task']}"
    
    return {
        "messages": [{"role": "assistant", "content": task}],
        "step_count": step,
        "current_task": task
    }

# 创建状态图
graph = StateGraph(AgentState)
graph.add_node("process_data", process_data_node)
graph.add_edge(START, "process_data")
graph.add_edge("process_data", END)

# 使用内存检查点实现持久化
checkpointer = MemorySaver()
compiled_graph = graph.compile(checkpointer=checkpointer)

# 第一次运行
initial_state = {
    "messages": [{"role": "user", "content": "开始数据处理"}],
    "step_count": 1,
    "current_task": "初始数据处理任务"
}
result1 = compiled_graph.invoke(initial_state)

print("第一次执行完成，状态已保存")

# 第二次运行（从检查点恢复）
result2 = compiled_graph.invoke(initial_state)
print("从检查点恢复继续执行")
​

持久化执行的关键优势：

• 故障恢复：程序崩溃后可以从最后检查点恢复
• 长时间任务：支持需要数小时甚至数天才能完成的复杂任务
• 状态管理：在多个会话间保持任务状态连续性
• 资源优化：可以在资源紧张时暂停执行，资源恢复后继续

2. 流式输出（Streaming）

流式输出能力让LangGraph能够实时输出中间结果，这对于需要实时反馈的应用场景至关重要。

# 多种流式输出模式
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI

# 定义状态
class State(MessagesState):
    pass

# 创建支持流式输出的智能体
def streaming_analysis_node(state: State):
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    print("开始流式分析...")
    
    # 模拟逐步分析过程
    steps = [
        "数据收集阶段：正在获取原始数据...",
        "数据清洗阶段：去除异常值和重复数据...", 
        "特征工程阶段：提取关键特征和模式...",
        "模型训练阶段：构建和训练预测模型...",
        "结果评估阶段：分析模型性能和准确性...",
        "报告生成阶段：生成最终的分析报告..."
    ]
    
    for i, step in enumerate(steps, 1):
        print(f"步骤 {i}/6: {step}")
        
        # 模拟处理时间
        import time
        time.sleep(0.5)
        
        # 流式输出当前进度
        response = llm.invoke([
            {"role": "system", "content": "你是数据分析专家，请报告当前分析进度"},
            {"role": "user", "content": f"进度报告: {step}"}
        ])
        
        # 将中间结果添加到消息历史
        yield {"messages": [response]}

# 构建支持流式输出的图
graph = StateGraph(State)
graph.add_node("analysis", streaming_analysis_node)
graph.add_edge(START, "analysis")
graph.add_edge("analysis", END)

compiled_graph = graph.compile()

# 使用不同流式模式执行
print("=== 使用values流式模式 ===")
for chunk in compiled_graph.stream(
    {"messages": [{"role": "user", "content": "分析销售趋势"}]}, 
    stream_mode="values"
):
    print("当前状态:", chunk.get("messages", []))

print("\n=== 使用messages流式模式 ===")
for message in compiled_graph.stream(
    {"messages": [{"role": "user", "content": "分析客户行为"}]}, 
    stream_mode="messages"
):
    if hasattr(message, 'content'):
        print(message.content, end="", flush=True)
​

流式输出的应用场景：

• 实时聊天应用：提供即时回复和交互体验
• 数据分析任务：实时显示分析进度和中间结果
• 多步骤工作流：逐步显示任务执行状态
• 教育培训系统：提供实时反馈和指导

3. 人类在回路（Human-in-the-Loop）

人类在回路能力允许智能体在执行过程中暂停并等待人工干预，这对于需要人类决策和审核的场景非常有用。

# 定义人工审核节点
def human_review_node(state: State):
    print("\n=== 需要人工审核 ===")
    print("待审核内容:")
    for i, message in enumerate(state["messages"]):
        print(f"{i+1}. {message.content}")
    
    print("\n请选择审核结果:")
    print("1. 批准通过")
    print("2. 需要修改")
    print("3. 拒绝执行")
    
    # 模拟用户输入
    user_input = input("请输入选择 (1-3): ")
    
    if user_input == "1":
        decision = "批准通过，继续执行下一步"
        action = "continue"
    elif user_input == "2":
        decision = "需要修改，请提供修改意见"
        action = "modify"
    else:
        decision = "拒绝执行，终止任务"
        action = "stop"
    
    return {
        "messages": [{
            "role": "assistant", 
            "content": f"人工审核结果: {decision}",
            "action": action
        }]
    }

# 定义修改节点
def modification_node(state: State):
    print("\n=== 修改节点 ===")
    print("请提供修改意见:")
    
    # 模拟用户输入修改意见
    feedback = input("修改意见: ")
    
    return {
        "messages": [{
            "role": "assistant", 
            "content": f"已根据意见修改: {feedback}"
        }]
    }

# 构建包含人工审核的图
graph = StateGraph(State)
graph.add_node("analysis", streaming_analysis_node)
graph.add_node("review", human_review_node)
graph.add_node("modify", modification_node)

# 设置边和中断点
graph.add_edge(START, "analysis")
graph.add_edge("analysis", "review")
graph.add_edge("review", "modify", lambda x: x.get("messages", [{}])[-1].get("action") == "modify")
graph.add_edge("review", END, lambda x: x.get("messages", [{}])[-1].get("action") == "stop")
graph.add_edge("modify", "analysis")  # 修改后重新进行分析

# 编译时配置中断策略
checkpointer = MemorySaver()
compiled_graph = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["review", "modify"]
)

print("=== 人工协作智能体演示 ===")
print("这个智能体会执行分析并在关键节点暂停等待人工审核。")

# 执行到第一个中断点
try:
    result = compiled_graph.invoke({
        "messages": [{"role": "user", "content": "分析产品改进建议"}]
    })
    print("执行到中断点，等待人工干预...")
except Exception as e:
    print(f"预期中的暂停: {e}")

# 人工干预后继续执行
print("\n模拟人工审批通过...")
final_result = compiled_graph.invoke(result)
print("最终执行完成")
​

人类协作的优势：

• 质量控制：在关键节点进行人工审核确保输出质量
• 道德约束：对敏感决策进行人工把关
• 用户定制：根据用户反馈调整执行策略
• 复杂决策：处理需要人类判断的复杂场景

4. 综合记忆（Comprehensive Memory）

LangGraph提供了丰富的记忆管理能力，包括短期工作记忆和长期持久化记忆。