Cách Tùy Chỉnh Cập Nhật Trạng Thái bằng Reducers

Trong LangGraph, Reducers là những hàm đặc biệt được sử dụng để tổng hợp dữ liệu hoặc kết hợp trạng thái khi có nhiều luồng song song (parallel branches) trả về kết quả khác nhau.

Nói cách khác:

Khi bạn chạy nhiều node cùng lúc (parallel), và muốn gom kết quả từ chúng lại để đưa vào bước tiếp theo – thì bạn cần một Reducer.

Bạn cần dùng Reducer khi:

• Bạn có các nhánh song song trong graph.

• Mỗi nhánh thực hiện công việc riêng và trả về một phần kết quả.

• Bạn cần gom các kết quả đó về một trạng thái chung trước khi đi tiếp.

Giả sử bạn có 3 nhánh song song:

• Node A xử lý phần tên.

• Node B xử lý địa chỉ.

• Node C xử lý số điện thoại.

Sau khi 3 node này chạy xong, bạn cần gộp kết quả lại thành:

{
  "name": "VHTSoft",
  "address": "TP. Hồ Chí Minh",
  "phone": "0123456789"
}

Reducer sẽ gom và hợp nhất những giá trị này thành một trạng thái chung.

Định nghĩa Reducer

Reducer là một hàm nhận vào nhiều trạng thái (dưới dạng danh sách) và trả về một trạng thái duy nhất.

def my_reducer(states: list[dict]) -> dict:
    result = {}
    for state in states:
        result.update(state)  # Gộp tất cả dict lại
    return result

Cách sử dụng trong LangGraph

Khi bạn xây dựng graph với StateGraph, bạn có thể chỉ định reducer cho một node nhất định như sau:

from langgraph.graph import StateGraph

builder = StateGraph(dict)
# Các node song song được thêm ở đây...

# Thêm reducer cho node tổng hợp
builder.add_reducer("merge_node", my_reducer)

Ví dụ thực tế

1. Khai báo các node

def node_1(state): return {"name": "VHTSoft"}
def node_2(state): return {"address": "TP.Hồ Chí Minh"}
def node_3(state): return {"phone": "0123456789"}

2. Tạo graph

builder = StateGraph(dict)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
builder.add_node("merge_node", lambda x: x)  # Dummy node

# Thêm reducer
def my_reducer(states):  # states là list các dict
    result = {}
    for s in states:
        result.update(s)
    return result

builder.add_reducer("merge_node", my_reducer)

# Chạy 3 node song song → gộp về merge_node
builder.add_edge("node_1", "merge_node")
builder.add_edge("node_2", "merge_node")
builder.add_edge("node_3", "merge_node")

builder.set_entry_point("node_1")  # node_1 là điểm vào chính

graph = builder.compile()