3. LangGraph实战

1. LangGraph高级特性：总结与注意事项

LangGraph作为一个强大的图结构程序设计工具，提供了许多高级特性来支持复杂的AI应用开发。本文将深入探讨LangGraph的一些关键概念和注意事项，帮助开发者更好地利用这个工具。

1.1 数据状态与归纳函数

在LangGraph中，理解数据状态的处理方式至关重要。默认情况下，节点返回的字典数据会覆盖原始数据。这可能导致一些意料之外的结果。例如：

class MyState(TypedDict):
    messages: list

def fn1(state: MyState):
    return {"messages": [4]}

r = graph.invoke({"messages": [1, 2, 3]})
# 结果是 {"messages": [4]} 而不是 [1,2,3,4]

为了解决这个问题，LangGraph提供了两种方法来累加数据：

手动获取并更新原始状态：

1
2
3

def fn1(state: MyState):
    old = state.get("messages", [])
    return {"messages": old + [4]}

使用LangGraph的Annotated封装和归纳函数：

def concat_lists(original: list, new: list) -> list:
    return original + new

class MyState(TypedDict):
    messages: Annotated[list, concat_lists]

def fn1(state: MyState):
    return {"messages": [4]}

r = graph.invoke({"messages": [1, 2, 3]})
# 输出是 {'messages': [1, 2, 3, 4]}

使用归纳函数的优势在于它允许每个节点独立执行，无需关心其他节点的操作，同时也简化了状态结构更新时的节点修改工作。

1.2 多节点并行执行

在LangGraph中，END节点表示当前路线结束，而不是终止整个图的执行。这对于理解多节点并行执行非常重要。

同一层级的节点会并行执行，但执行顺序不确定。
可以通过调整节点连接方式来控制执行流程。

例如：

1	graph.add_edge(["left1", "right3"], "merge")

这样可以让left1和right3节点处于同一层，同时连接到merge节点。

1.3 检查点（CheckPoint）机制

检查点可以看作是一个存储介质，用于记录节点状态。主要特点包括：

可以获取最后状态和历史记录：
- graph.get_state(config): 获取最后一次存档
- graph.get_state_history(config): 获取所有存档列表
支持状态回退
允许数据修改
使用thread_id和thread_ts唯一定位存档

1.4 最佳实践建议

根据需求选择合适的数据处理方式，考虑使用归纳函数来处理累积数据。
在设计图结构时，注意节点的层级和连接方式，以实现期望的执行流程。
合理利用检查点机制，但要注意存储开销。
在处理复杂状态时，考虑使用TypedDict和Annotated来增强类型提示和数据处理逻辑。

1.5 注意事项

默认的数据覆盖行为可能导致意外结果，请谨慎处理状态更新。
多节点并行执行时，要注意执行顺序的不确定性可能带来的影响。
使用检查点机制时要考虑性能影响，特别是在处理大量数据或频繁存档时。
虽然归纳函数提供了便利，但在某些特殊操作中可能会增加复杂性，需要权衡使用。

1.6 结语

通过深入理解和合理运用这些特性，开发者可以构建出更加强大、灵活和高效的LangGraph应用。LangGraph的这些高级特性为复杂AI应用的开发提供了强大的支持，但同时也需要开发者在使用时保持谨慎，充分考虑各种情况。随着LangGraph的不断发展和完善，我们可以期待它在未来为AI应用开发带来更多可能性。

2. LangGraph的两种基础流式响应技巧

在构建复杂的AI应用时，LangGraph作为一个强大的工具，为我们提供了灵活的图结构程序设计能力。今天，我们将深入探讨LangGraph中的一个关键特性：流式响应模式。这个特性不仅能提高应用的响应速度，还能为用户提供更加流畅的交互体验。

2.1 LangGraph中的流式响应：与传统LLM有何不同？

在LangGraph中，编译后的图程序本质上是一个Runnable可运行组件。与传统的大语言模型（LLM）不同，LangGraph支持多种流式模式输出。传统LLM的流式模式通常是一个词一个词地输出，而LangGraph的流式响应每次输出的是一个节点的数据状态。这种设计为我们提供了更细粒度的控制和更丰富的数据展示方式。

2.1.1 两种基本的流式模式

LangGraph提供了两种主要的流式响应模式，每种模式都有其特定的使用场景：

values模式：
- 返回图的完整状态值（总量）
- 每个节点调用后，返回图的完整状态
- 适用于需要随时了解整个图状态的场景
updates模式：
- 返回图的状态更新（增量）
- 每个节点调用后，只返回状态的变化部分
- 适用于只关注变化部分，或需要节省带宽的场景

2.1.2 如何使用流式模式？

使用流式模式非常简单。在调用stream()函数时，只需传递stream_mode参数即可配置不同的流式响应模式。让我们以ReACT智能体为例，来看看如何使用这两种模式：

# values模式示例
inputs = {"messages": [("human", "2024年北京半程马拉松的前3名成绩是多少?")]}
for chunk in agent.stream(
    inputs,
    stream_mode="values",
):
    print(chunk["messages"][-1].pretty_print())

# updates模式示例 
for chunk in agent.stream(
    inputs,
    stream_mode="updates",
):
    print(chunk)

在values模式下，每次输出都是完整的数据状态。而在updates模式下，返回的是字典格式的增量数据，键为节点名称，值为状态的更新。

2.2 当前限制与未来展望

虽然LangGraph的流式响应机制已经为我们提供了强大的功能，但它仍然存在一些限制。目前，虽然我们可以正确获取每个节点的数据，但等待时间仍然较长，特别是涉及大语言模型的节点。这是因为节点本身也应该支持流式输出。理想情况下，大语言模型节点在图的流式输出下应该保持其固有的流式特性，而不是等待完整输出后再返回。