LangGraph 实战深入教程

一、引言

在人工智能和自然语言处理领域，随着大型语言模型（LLMs）的飞速发展，构建复杂、智能的应用程序成为了新的挑战和需求。传统的开发框架在处理复杂任务、动态流程控制和长时状态跟踪等方面逐渐显现出局限性。LangGraph 作为 LangChain 生态系统中的一个重要扩展，应运而生，为开发者提供了一种强大而灵活的方式来构建有状态、多参与者的应用程序，尤其是智能体（Agent）和多智能体工作流。

本文将深入探讨 LangGraph 的各个方面，包括其核心原理、高级功能、复杂代码示例、与其他框架的对比、性能优化方法以及未来发展趋势等，旨在帮助开发者全面掌握 LangGraph 的使用，从而构建出更加智能、高效的 AI 应用。

二、LangGraph 基础回顾

2.1 基本概念

LangGraph 是由 LangChain Inc. 开发的一个底层编排框架，它利用大型语言模型（LLMs）构建有状态、多参与者的应用程序，特别是智能体和多智能体工作流。其核心目标是为复杂的 AI 智能体任务提供可靠性、可控性和可扩展性。

与许多传统 LLM 链（通常构建为有向无环图 - DAGs）不同，LangGraph 专注于支持循环图结构。这种循环能力对于实现智能体行为至关重要，因为智能体行为通常涉及循环、重试和基于动态决策的路径选择。

2.2 核心组件

2.2.1 状态（State）

状态是 LangGraph 应用程序的基础，代表了应用程序数据在执行过程中的当前快照。状态的设计和管理对于构建复杂、可靠的智能体至关重要。

状态模式定义了图管理的信息结构，是所有节点和边的输入/输出契约。LangGraph 主要支持使用 Python 的 TypedDict 或 Pydantic BaseModel 来定义状态模式。

例如，使用 TypedDict 定义一个简单的状态：

from typing import TypedDict

class ShoppingState(TypedDict):
    # 当前状态
    current_step: str
    # 购物车商品
    cart_items: list[str]
    # 总金额
    total_amount: float
    # 用户输入
    user_input: str

2.2.2 节点（Nodes）

节点是图中的基本元素，代表了系统中的各种实体。这些实体可以是不同类型的智能体，也可以是执行特定计算或处理任务的函数。

节点通常是 Python 函数，接收当前状态作为输入，执行实际的计算、逻辑操作或副作用，并返回更新后的状态。节点可以包含 LLM 或简单的 Python 代码。

例如，定义一个简单的节点函数：

def add_to_cart(state: ShoppingState):
    item = state["user_input"]
    state["cart_items"].append(item)
    return state

2.2.3 边（Edges）

边表示节点之间的连接关系，这种关系可以是多种多样的，比如因果关系、依赖关系、通信关系等。边决定了状态在节点之间的流转路径，即下一步应该执行哪个节点。

LangGraph 支持普通边和条件边。普通边表示无条件跳转，而条件边根据状态的不同选择不同的执行路径。

例如，定义一个条件边：

def should_move_to_checkout(state: ShoppingState) -> bool:
    """判断是否应该转换到结账状态"""
    return "checkout" in state["user_input"].lower()

2.3 工作原理

LangGraph 将智能体工作流建模为图，通过节点和边的组合来定义智能体的行为。在执行过程中，系统会维护一个中央状态对象，该状态对象会根据节点的执行结果不断更新。节点根据当前状态执行相应的操作，并返回更新后的状态，边根据状态决定下一步的执行节点，直到到达结束节点。

例如，一个简单的购物流程可以用 LangGraph 表示为：

from langgraph.graph import StateGraph

class ShoppingGraph(StateGraph):
    def __init__(self):
        super().__init__()
        # 定义状态
        self.add_node("browse", self.browse_products)
        self.add_node("add_to_cart", self.add_to_cart)
        self.add_node("checkout", self.checkout)
        self.add_node("payment", self.payment)

    def browse_products(self, state):
        # 浏览商品逻辑
        return state

    def add_to_cart(self, state):
        # 添加商品到购物车逻辑
        return state

    def checkout(self, state):
        # 结账逻辑
        return state

    def payment(self, state):
        # 支付逻辑
        return state

三、LangGraph 高级功能

3.1 循环和分支

LangGraph 支持在应用程序中实现循环和条件逻辑，这对于构建复杂的代理架构至关重要。通过循环和分支，智能体可以根据不同的情况做出不同的决策，实现更加灵活和智能的行为。

例如，一个智能客服系统可以根据用户的问题类型进行不同的处理：

from langgraph.graph import StateGraph, START, END
from typing import Dict, Any

def conditional_branch(state: Dict[str, Any]) -> str:
    """根据输入的状态决定下一个节点"""
    if "technical" in state.get('question_type', ''):
        return 'technical_support'
    else:
        return 'general_support'

builder = StateGraph()
builder.add_node("start", lambda state: {"processed": True})
builder.add_node("branch_point", conditional_branch)
builder.add_node("technical_support", lambda state: {"result": "Technical support response"})
builder.add_node("general_support", lambda state: {"result": "General support response"})

# 定义起始到分叉点以及后续两条可能路径的关系
builder.add_edge(START, "start")
builder.add_edge("start", "branch_point")
builder.add_edge("branch_point", "technical_support", condition=lambda s: "technical" in s['question_type'])
builder.add_edge("branch_point", "general_support", condition=lambda s: "technical" not in s['question_type'])

for end_node in ["technical_support", "general_support"]:
    builder.add_edge(end_node, END)

compiled_graph = builder.compile()

# 测试不同的初始状态如何影响最终的结果
print(compiled_graph.invoke({'question_type': 'technical'}))  # 输出 {'result': 'Technical support response'}
print(compiled_graph.invoke({'question_type': 'general'}))  # 输出 {'result': 'General support response'}

3.2 状态持久化

LangGraph 内置了状态持久化机制，允许在图的每一步之后自动保存状态。这使得系统可以在任何点暂停和恢复图的执行，支持错误恢复、人工干预工作流、时间旅行等功能。

例如，使用 MemorySaver 保存对话历史：

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END

# 定义状态类
class ChatState:
    def __init__(self):
        self.messages = []

# 定义节点函数
def chatbot_node(state: ChatState):
    user_input = input("User: ")
    state.messages.append({"role": "user", "content": user_input})
    response = f"Echo: {user_input}"
    state.messages.append({"role": "assistant", "content": response})
    print(f"Assistant: {response}")
    return state

# 构建图
graph_builder = StateGraph(ChatState)
graph_builder.add_node("chatbot", chatbot_node)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

# 运行图
state = ChatState()
graph.invoke(state)

3.3 人工干预

LangGraph 能中断图的执行，以便人工批准或编辑代理计划的下一个行动。这在一些需要人工决策的场景中非常有用，例如医疗诊断、金融风险评估等。

例如，在一个智能客服系统中，当遇到复杂问题时，可以将问题转交给人工客服处理：

from langgraph.graph import StateGraph, START, END
from typing import Dict, Any

def is_complex_question(state: Dict[str, Any]) -> bool:
    """判断问题是否复杂"""
    return "complex" in state.get('question_type', '')

def transfer_to_human(state: Dict[str, Any]):
    """转人工处理"""
    print("Transferring to human support...")
    return state

builder = StateGraph()
builder.add_node("start", lambda state: {"processed": True})
builder.add_node("check_complexity", is_complex_question)
builder.add_node("transfer_to_human", transfer_to_human)
builder.add_node("auto_response", lambda state: {"result": "Auto response"})

# 定义起始到检查复杂度节点以及后续两条可能路径的关系
builder.add_edge(START, "start")
builder.add_edge("start", "check_complexity")
builder.add_edge("check_complexity", "transfer_to_human", condition=lambda s: s)
builder.add_edge("check_complexity", "auto_response", condition=lambda s: not s)

for end_node in ["transfer_to_human", "auto_response"]:
    builder.add_edge(end_node, END)

compiled_graph = builder.compile()

# 测试不同的初始状态如何影响最终的结果
print(compiled_graph.invoke({'question_type': 'complex'}))  # 输出 Transferring to human support...
print(compiled_graph.invoke({'question_type': 'simple'}))  # 输出 {'result': 'Auto response'}

3.4 流式支持

LangGraph 支持按每个节点生成的顺序流式传输输出，包括令牌流。这使得系统可以实时反馈执行状态，提升用户体验。

例如，在一个文本生成任务中，可以实时输出生成的文本：

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

# 定义状态
class State(TypedDict):
    messages: Annotated[List[HumanMessage | AIMessage], "Conversation messages"]

model = ChatOpenAI(temperature=0, streaming=True)

def generate_text(state: State):
    messages = state["messages"]
    for chunk in model.stream(messages):
        print(chunk.content, end="", flush=True)
    return state

graph_builder = StateGraph(State)
graph_builder.add_node("generate_text", generate_text)
graph_builder.add_edge(START, "generate_text")
graph_builder.add_edge("generate_text", END)

graph = graph_builder.compile()

# 运行图
state = State(messages=[HumanMessage(content="Generate a short story.")])
graph.invoke(state)

四、复杂代码示例

4.1 多智能体协作系统

在许多复杂的应用场景中，需要多个智能体协作完成任务。LangGraph 提供了一种灵活的方式来构建多智能体协作系统。

以下是一个简单的多智能体协作系统示例，包括一个信息检索智能体和一个答案生成智能体：

from langgraph.graph import StateGraph, START, END
from langchain_core.tools import tool
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated, List
from langchain_core.messages import HumanMessage, AIMessage, FunctionMessage

# 定义工具
@tool
def search(query: str):
    """Call to surf the web."""
    search_tool = TavilySearchResults(max_results=1)
    return search_tool.run(query)

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[List[HumanMessage | AIMessage | FunctionMessage], "Conversation messages"]
    tool_calls: list = []  # 记录工具调用历史

# 初始化模型
model = ChatOpenAI(temperature=0).bind_tools([search])

# 定义节点函数
def agent_node(state: AgentState):
    messages = state["messages"]
    response = model.invoke(messages)
    return {"messages": [response]}

def tool_node(state: AgentState):
    last_message = state["messages"][-1]
    if "function_call" in last_message.additional_kwargs:
        function_name = last_message.additional_kwargs["function_call"]["name"]
        parameters = last_message.additional_kwargs["function_call"]["arguments"]
        if function_name == "search":
            query = parameters["query"]
            result = search(query)
            return {"messages": [FunctionMessage(content=str(result), name=function_name)]}
    return state

# 定义条件函数
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if "function_call" in last_message.additional_kwargs:
        return "tool_node"
    return END

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tool", tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue, {"tool_node": "tool", END: END})
workflow.add_edge("tool", "agent")

app = workflow.compile()

# 运行测试
final_state = app.invoke(
    {"messages": [HumanMessage(content="What is the capital of France?")]}
)
print(final_state["messages"][-1].content)

4.2 任务规划与执行系统

LangGraph 可以用于构建任务规划与执行系统，通过定义不同的节点和条件边，实现复杂的任务调度和执行逻辑。

以下是一个简单的任务规划与执行系统示例，包括任务分配、任务执行和结果检查：

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, List

# 定义状态
class TaskState(TypedDict):
    tasks: List[str]
    current_task: str
    task_result: str

# 定义节点函数
def assign_task(state: TaskState):
    if state["tasks"]:
        state["current_task"] = state["tasks"].pop(0)
    return state

def execute_task(state: TaskState):
    # 模拟任务执行
    state["task_result"] = f"Task {state['current_task']} executed successfully."
    return state

def check_result(state: TaskState):
    if state["task_result"]:
        return "assign_task" if state["tasks"] else END
    return "execute_task"

# 构建图
graph_builder = StateGraph(TaskState)
graph_builder.add_node("assign_task", assign_task)
graph_builder.add_node("execute_task", execute_task)
graph_builder.add_node("check_result", check_result)
graph_builder.add_edge(START, "assign_task")
graph_builder.add_edge("assign_task", "execute_task")
graph_builder.add_edge("execute_task", "check_result")
graph_builder.add_conditional_edges("check_result", check_result, {"assign_task": "assign_task", END: END})

graph = graph_builder.compile()

# 运行图
state = TaskState(tasks=["Task 1", "Task 2", "Task 3"], current_task="", task_result="")
final_state = graph.invoke(state)
print(final_state["task_result"])

五、LangGraph 与其他框架的对比

5.1 与 LangChain 的对比

LangGraph 是 LangChain 生态系统的一个扩展或模块，通常与 LangChain 的组件一起使用，但也可以独立运行。LangChain 是构建由大型语言模型 (LLMs) 驱动的应用程序最受欢迎的框架之一，它提供了一种灵活的代码优先方法，允许开发者将文档检索、摘要和问答等任务串联成统一的工作流程。

而 LangGraph 专注于复杂、有状态流程的编排，它解决了 LangChain 在处理循环、状态持久化和条件逻辑等方面的局限性。LangGraph 支持循环图结构，使得智能体可以进行多轮交互和动态决策，同时提供了强大的状态管理机制，方便开发者管理和维护上下文信息。

5.2 与 AutoGen 的对比

AutoGen 是微软推出的一个多智能体框架，它涉及两个核心代理：用户代理 (User Agent) 和助手代理 (Assistant Agent)。用户代理与助手代理交互，助手代理负责生成并执行代码。AutoGen 专为代码任务的多代理编排设计，同时也能处理其他任务，支持人类在交互过程中的指导。

相比之下，LangGraph 更加通用和灵活，它不仅可以用于多智能体编排，还可以用于各种复杂的工作流编排。LangGraph 提供了更细粒度的流程控制和状态管理，允许开发者根据具体需求自定义智能体的行为和交互逻辑。

5.3 与 CrewAI 的对比

CrewAI 是一个构建快速演示的首选框架，它非常直观且易于设置，主要依赖提示工程。CrewAI 可以轻松创建新代理并加入生态系统，非技术用户也能快速上手。

而 LangGraph 更适合构建复杂的、需要高度定制的智能体系统。LangGraph 提供了丰富的功能和灵活的架构，允许开发者根据具体需求设计智能体的行为和交互逻辑，同时支持状态持久化、人工干预等高级功能。

六、性能优化方法

6.1 序列化库优化

LangGraph 0.3.20 版本将默认的序列化库换成了 ormsgpack（一个高性能的 MessagePack 实现），根据内部测试，序列化/反序列化的速度提升了 30% 以上。这意味着任务调度更快、网络传输开销更低、整体响应更流畅。

例如，将 LangGraph 升级到 0.3.20 版本后，序列化/反序列化的性能对比：

序列化库	平均耗时（万次操作）
旧库	1200ms
ormsgpack	800ms

6.2 任务 ID 优化

LangGraph 0.3.20 版本将任务 ID 的哈希算法升级到了 128 位，这使得任务 ID 的唯一性更强，几乎不可能出现哈希冲突的情况。对于大规模分布式任务调度来说，这大大提高了系统的稳定性和可靠性。

6.3 子图容错机制

在 0.3.20 版本中，LangGraph 引入了日志记录 + 自动容错机制。即使子图的 Schema 出现问题，系统也会记录错误并继续执行，而不是直接崩溃。这对于生产环境来说非常重要，可以避免因为一个小问题导致整个流程中断。

6.4 代码优化

在编写 LangGraph 代码时，可以采用一些常见的代码优化技巧，如减少不必要的计算、避免重复的操作、合理使用缓存等。同时，优化节点函数的性能，确保节点函数的执行效率。

例如，在节点函数中避免使用复杂的循环和递归，尽量使用高效的算法和数据结构。

七、未来发展趋势

7.1 功能扩展

未来，LangGraph 可能会进一步扩展其功能，例如支持更多的编程语言、提供更多的预构建组件和模板、增强与其他框架和工具的集成等。这将使得开发者能够更加方便地使用 LangGraph 构建各种复杂的应用程序。

7.2 性能提升

随着技术的不断发展，LangGraph 可能会不断优化其性能，提高系统的响应速度和吞吐量。例如，采用更高效的算法和数据结构、优化状态管理和节点调度机制等。

7.3 应用场景拓展

LangGraph 在金融服务、医疗健康、智能制造、教育科技等领域都具有广阔的应用前景。未来，随着这些领域对智能应用的需求不断增加，LangGraph 可能会在更多的行业得到应用和推广。

7.4 社区生态建设

LangGraph 的社区生态建设将越来越重要。未来，可能会有更多的开发者参与到 LangGraph 的开发和贡献中来，提供更多的插件、工具和示例代码，形成一个更加繁荣的社区生态。

八、总结

LangGraph 作为 LangChain 生态系统中的一个重要扩展，为开发者提供了一种强大而灵活的方式来构建有状态、多参与者的应用程序。通过其独特的图结构和状态管理机制，LangGraph 支持循环、条件分支、状态持久化等高级功能，适用于各种复杂的智能体场景。

在实际应用中，开发者可以根据具体需求，利用 LangGraph 的各种功能和特性，构建出高效、可靠的智能应用。同时，随着 LangGraph 的不断发展和完善，相信它将在人工智能领域发挥越来越重要的作用。

九、参考资料

• LangGraph 官方文档

• LangChain 博客

• 相关 CSDN 博客文章

• 相关掘金文章

希望通过本文的深入教程，你能够全面掌握 LangGraph 的使用方法，并在自己的项目中应用。如果你在使用过程中遇到任何问题，欢迎查阅参考资料或在社区中寻求帮助。