AG Day 12 狀態設計:TypedDict、reducer 與訊息累積
執行需求:CPU 可跑。本篇文章聚焦於 LangGraph 狀態架構的進階聚合語意,所有範例皆在本地 CPU 環境下運算,不需要外部模型 API 金鑰,即可完整演練 TypedDict 定義、自訂縮減器(Reducer)、訊息增量堆疊與識別碼更新機制。
引言
在上一篇 AG Day 11(原文連結)中,我們正式引進了 LangGraph 1.x,透過 StateGraph、START 與 END 打造了包含規劃、佐證收集與報告產出的三階段狀態圖。在昨天的基礎範例中,各節點之間的狀態更新採用的是標準的「覆蓋模式(Overwrite Mode)」:當節點回傳一個字典時,圖引擎便直接以新值取代狀態中對應鍵值的舊資料。
然而,當我們開始建構真實對話或長流程研究助理時,「直接覆蓋」的策略很快就會遭遇瓶頸。在一個多輪推理迴圈中,系統不可能在第二輪對話時直接抹去第一輪的使用者提問與助理回應;同樣地,當多個搜尋節點並行執行時,我們期望將新擷取到的文獻網址「追加」進現有的來源清單中,而不是讓後執行的節點將先執行節點的心血覆蓋蒸發。換言之,我們需要讓狀態具備「累積(Accumulation)」與「智慧縮減(Reduction)」的能力。
LangGraph 透過 Python 3.9+ 的 typing.Annotated 語法,優雅地解決了這項工程難題。藉由在型別標註中綁定縮減器函式(Reducer Functions),開發者能夠為狀態中的每一個欄位獨立定義合併語意。本文將深入剖析 LangGraph 的狀態聚合核心,從自訂清單合併器出發,深入探索官方內建的 add_messages 縮減器,並為貫穿專案 research-agent 建立具備訊息記憶與知識累積能力的強大狀態架構。
TypedDict 狀態架構與預設覆蓋行為
在深入縮減器之前,我們必須徹底理解 LangGraph 的狀態基礎。在 Python 生態中,定義結構化資料主要有 TypedDict、dataclass 與 Pydantic BaseModel 三種選擇。LangGraph 官方強烈推薦以 TypedDict 作為圖狀態的首選規格。
原因在於執行效能與架構純粹度。TypedDict 本質上是標準 Python 字典(dict)的型別提示外衣,它在執行階段毫無額外的封裝開銷,序列化與反序列化速度極快,記憶體佔用極低。更重要的是,它天生具備字典的扁平語意,極度契合分散式狀態機在節點之間傳遞「增量修訂(Partial Updates)」的運作模式。
在預設情況下,如果在 TypedDict 中只宣告型別而未附加任何元資料,LangGraph 會對該鍵值採取標準的字典合併邏輯(相當於 current_state[key] = new_value)。例如,如果狀態中原本有 {"counter": 1},節點回傳 {"counter": 2},狀態便更新為 2。這種預設行為適用於旗標開關、階段標記或單一報告文本;但一旦遇到對話訊息或資料集合,覆蓋將導致歷史遺失。
Annotated 與 Reducer 的數學本質
為了解決集合累積的痛點,LangGraph 引入了函式語言程式設計中的核心概念——縮減器(Reducer)。縮減器的數學本質是一個二元運算子:Reducer(current_value, update_value) -> next_value。它接收「目前的狀態值」與「節點傳來的更新值」,經過特定的邏輯運算後,回傳「合併後的全新狀態值」。
在 Python 中,我們使用標準庫 typing.Annotated 來將這個運算子附加至型別標註之上:
# Annotated 語法結構:Annotated[型別, 縮減器函式]
from typing import Annotated, List
def append_list(current: List[str], update: List[str]) -> List[str]:
"""最基礎的清單累加縮減器。"""
return current + update
class CustomState(TypedDict):
# 宣告 keywords 欄位在合併時使用 append_list 縮減器進行追加
keywords: Annotated[List[str], append_list]
當節點執行完畢並回傳 {"keywords": ["LangGraph"]} 時,圖引擎不會粗暴地用 ["LangGraph"] 覆蓋原本的清單,而是會自動呼叫 append_list(current_keywords, ["LangGraph"]),並將結合後的新串列寫入系統狀態中。這賦予了每個欄位極為細緻的自主進化能力。
add_messages 深入解析:訊息更新、覆蓋與刪除
在所有縮減器中,最核心、也最常被使用的是 LangGraph 專門為對話歷史打造的 add_messages。初學者常誤以為 add_messages 只是一個單純的 list.extend(),但事實上它是一個功能極其完備的「訊息資料庫管理員」。
add_messages 具備三大核心超能力:
- 型別自動提升(Type Coercion):它不僅接受標準的 LangChain 訊息實體(如
HumanMessage、AIMessage、SystemMessage、ToolMessage),也支援以元組(Tuple,如("human", "你好"))或原生字典形式傳入,並在底層自動轉換為標準訊息實體。 - 基於識別碼的就地更新(ID-based Upsert):每則訊息都可以具備一個全域唯一識別碼(
id)。當傳入的新訊息所帶有的id與現有狀態中的某一則訊息完全相同時,add_messages不會重複追加,而是會直接替換該則既有訊息。這項機制在串流輸出、以及人工修正先前發言時扮演了決定性的角色。 - 語意化訊息刪除(Message Deletion):當需要從對話歷史中修剪或刪除特定訊息以節省權杖配額時,我們不需要手動走訪串列,只需傳入
RemoveMessage(id="target_id"),add_messages就會自動在狀態中剔除對應的歷史記錄。
完整實作:為 research-agent 建立多維狀態與聚合縮減器
現在,我們將進階狀態設計整合進貫穿專案 research-agent 中。我們將在 src/research_agent/graph.py 中實作一個具備多輪對話訊息累積、去重複技術標籤聚合、以及步階計數器累加的完整狀態機架構。
第一步,定義自訂縮減器函式與核心 AdvancedResearchState 架構:
"""src/research_agent/graph.py:定義具備縮減器的進階狀態。"""
from typing import TypedDict, Annotated, List, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, RemoveMessage
from langgraph.graph.message import add_messages
from langgraph.graph import StateGraph, START, END
def merge_unique_tags(current: List[str], update: List[str]) -> List[str]:
"""自訂縮減器:將新標籤追加至清單中,並自動去除重複標籤與前後空白。"""
combined = list(current or [])
for tag in update:
cleaned = tag.strip()
if cleaned and cleaned not in combined:
combined.append(cleaned)
return combined
def step_accumulator(current: int, increment: int) -> int:
"""自訂縮減器:步數累加器。"""
return (current or 0) + increment
class AdvancedResearchState(TypedDict):
"""研究助理在圖中流轉的進階狀態架構。"""
# 使用官方 add_messages 管理訊息生命週期(支援追加、ID 覆蓋與刪除)
messages: Annotated[Sequence[BaseMessage], add_messages]
# 使用自訂縮減器維護不重複的技術標籤
tags: Annotated[List[str], merge_unique_tags]
# 使用累加器記錄圖執行的總步數
step_count: Annotated[int, step_accumulator]
# 標準無縮減器欄位:節點回傳時直接覆蓋
current_summary: str
第二步,實作展示訊息累積與標籤追加的節點函式。請注意,由於綁定了縮減器,節點在回傳時絕對不可回傳整份累積清單,而只需回傳本次運算的「增量切片」:
"""定義產生狀態增量的純函式節點。"""
def user_intake_node(state: AdvancedResearchState) -> dict:
"""第一階段:解析使用者最新輸入,並初步提取主題標籤。"""
print("[節點: user_intake] 正在解析使用者需求...")
# 增量回傳:新增技術標籤與單步計數
return {
"tags": ["Agent架構", "LangGraph"],
"step_count": 1,
"current_summary": "已完成使用者需求剖析。",
}
def research_worker_node(state: AdvancedResearchState) -> dict:
"""第二階段:執行研究檢索,追加助理訊息與新標籤。"""
print("[節點: research_worker] 正在進行深度檢索並生成助理回覆...")
ai_reply = AIMessage(
content="依據檢索結果,LangGraph 透過 Annotated 與 Reducer 提供了強大的狀態合併能力。",
id="ai_msg_001", # 指定訊息識別碼
)
return {
"messages": [ai_reply],
"tags": ["LangGraph", "狀態管理", "Reducer"], # 包含重複標籤以測試去重複
"step_count": 1,
"current_summary": "已檢索完成並產出第一版分析。",
}
def report_curator_node(state: AdvancedResearchState) -> dict:
"""第三階段:審查報告,並展示透過同 ID 訊息更新既有內容的機制。"""
print("[節點: report_curator] 正在進行報告修訂,覆蓋先前的助理訊息...")
# 建立相同 id 的訊息以測試 upsert 機制
updated_reply = AIMessage(
content="【經專家審查修訂】LangGraph 的 Reducer 機制徹底消除了多節點狀態競爭衝突。",
id="ai_msg_001", # 相同 ID 將直接覆蓋先前的 ai_msg_001
)
return {
"messages": [updated_reply],
"tags": ["專家審查"],
"step_count": 1,
"current_summary": "研究總結報告已定稿。",
}
第三步,建立圖組裝函式 build_advanced_graph() 並完成編譯:
"""組裝具備進階縮減器的狀態圖。"""
def build_advanced_graph():
"""組裝具備進階狀態縮減器的 LangGraph 實體。"""
builder = StateGraph(AdvancedResearchState)
builder.add_node("intake", user_intake_node)
builder.add_node("worker", research_worker_node)
builder.add_node("curator", report_curator_node)
builder.add_edge(START, "intake")
builder.add_edge("intake", "worker")
builder.add_edge("worker", "curator")
builder.add_edge("curator", END)
return builder.compile()
第四步,撰寫端到端執行示範,觀察訊息的堆疊、識別碼覆蓋以及標籤的去重複合併過程:
"""執行進階狀態圖並詳細印出每一步的累積狀態。"""
from src.research_agent.graph import build_advanced_graph
from langchain_core.messages import HumanMessage
app = build_advanced_graph()
# 初始輸入狀態包含一則人類提問
initial_state = {
"messages": [HumanMessage(content="請分析 LangGraph 的狀態聚合機制。", id="user_001")],
"tags": ["初始研究"],
"step_count": 0,
"current_summary": "等待系統啟動",
}
print("=== 啟動圖執行流程 ===")
final_output = app.invoke(initial_state)
print("\n" + "=" * 20 + " 最終聚合狀態審查 " + "=" * 20)
print(f"累計執行總步數: {final_output['step_count']}")
print(f"最新摘要內容: {final_output['current_summary']}")
print(f"去重複後的技術標籤: {final_output['tags']}")
print(f"最終訊息庫筆數: {len(final_output['messages'])} 則")
for i, msg in enumerate(final_output["messages"], 1):
role = msg.__class__.__name__
print(f" [{i}] ({role}, ID: {msg.id}): {msg.content}")
# 輸出(範例輸出):
# === 啟動圖執行流程 ===
# [節點: user_intake] 正在解析使用者需求...
# [節點: research_worker] 正在進行深度檢索並生成助理回覆...
# [節點: report_curator] 正在進行報告修訂,覆蓋先前的助理訊息...
#
# ==================== 最終聚合狀態審查 ====================
# 累計執行總步數: 3
# 最新摘要內容: 研究總結報告已定稿。
# 去重複後的技術標籤: ['初始研究', 'Agent架構', 'LangGraph', '狀態管理', 'Reducer', '專家審查']
# 最終訊息庫筆數: 2 則
# [1] (HumanMessage, ID: user_001): 請分析 LangGraph 的狀態聚合機制。
# [2] (AIMessage, ID: ai_msg_001): 【經專家審查修訂】LangGraph 的 Reducer 機制徹底消除了多節點狀態競爭衝突。
第五步,展示透過 RemoveMessage 進行歷史修剪的高階操作:
"""示範使用 RemoveMessage 縮減對話歷史權杖。"""
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage, RemoveMessage
# 模擬現存的歷史訊息清單
existing_messages = [
HumanMessage(content="第一輪問題", id="m1"),
AIMessage(content="第一輪回答", id="m2"),
HumanMessage(content="第二輪問題", id="m3"),
]
# 傳送刪除指令,指定移除 ID 為 m1 的訊息
updates = [RemoveMessage(id="m1")]
pruned_messages = add_messages(existing_messages, updates)
print("修剪前訊息 ID:", [m.id for m in existing_messages])
print("修剪後訊息 ID:", [m.id for m in pruned_messages])
# 輸出(範例輸出):
# 修剪前訊息 ID: ['m1', 'm2', 'm3']
# 修剪後訊息 ID: ['m2', 'm3']
第六步,我們為自訂縮減器與狀態合併邏輯編寫自動化單元測試 tests/test_state_reducers.py,利用 pytest 全面驗證去重複、步階加總與訊息 ID 覆蓋的正確性:
"""tests/test_state_reducers.py:驗證狀態縮減器之合併行為。"""
import pytest
from langchain_core.messages import HumanMessage, AIMessage
from src.research_agent.graph import merge_unique_tags, step_accumulator, build_advanced_graph
def test_merge_unique_tags():
"""測試技術標籤去重複與空白清理。"""
base = ["Python", "LangGraph"]
new_tags = ["LangGraph", " Pydantic ", "Python", "Docker"]
res = merge_unique_tags(base, new_tags)
assert res == ["Python", "LangGraph", "Pydantic", "Docker"]
def test_step_accumulator():
"""測試步階數值累積。"""
assert step_accumulator(0, 1) == 1
assert step_accumulator(5, 3) == 8
def test_advanced_graph_message_upsert():
"""測試圖執行時相同訊息 ID 的就地覆蓋機制。"""
app = build_advanced_graph()
res = app.invoke({
"messages": [HumanMessage(content="初始測試", id="test_01")],
"tags": [],
"step_count": 0,
"current_summary": "",
})
# 驗證訊息總筆數只有 2 筆(一則 Human,一則被 curator 覆蓋的 AI)
assert len(res["messages"]) == 2
assert res["messages"][1].id == "ai_msg_001"
assert "【經專家審查修訂】" in res["messages"][1].content
assert res["step_count"] == 3
讀者可以在終端機執行以下指令確認所有單元測試皆順利通過:
# 執行狀態縮減器單元測試
uv run pytest tests/test_state_reducers.py -v
常見錯誤與踩雷
在設計與使用 LangGraph 縮減器時,有四個極具迷惑性的常見陷阱:
- 在節點中回傳整份累積清單(指數級重複災難):這是初學者最常犯的重大錯誤。當一個欄位宣告了
Annotated[List[T], add]或add_messages時,縮減器會自動幫你執行current + update。如果節點函式內部自己先寫了all_msgs = state["messages"] + [new_msg],並回傳{"messages": all_msgs},縮減器將會把all_msgs再次附加到state["messages"]之後!這會導致狀態中的資料以指數級速度瘋狂重複膨脹,迅速撐爆記憶體與上下文視窗。記住核心鐵律:有縮減器的欄位,節點永遠只回傳本次的新增量。 - 忘記使用
Annotated宣告縮減器:如果在TypedDict中寫下messages: List[BaseMessage],而漏掉了Annotated[..., add_messages],LangGraph 將會退回預設的覆蓋行為。每當某個節點回傳{"messages": [new_reply]}時,先前的所有歷史對話記錄將會瞬間被覆蓋清空,導致代理失去對話記憶。 - 在自訂縮減器中直接就地修改傳入物件:在撰寫自訂縮減器時,絕對不要在
current物件上直接呼叫.append()或.extend()。必須像我們的merge_unique_tags一樣,先複製或建立新物件後再回傳。就地修改破壞了資料的不可變性(Immutability),會直接摧毀 LangGraph 檢查點機制的時光旅行回溯能力。 - 訊息缺少 ID 導致無法精確更新:在使用
add_messages時,如果建立訊息時未指定id,系統會自動生成隨機 UUID。這意味著你傳送的每一則訊息都會被視為全新訊息追加在末尾。若想要實作串流更新同一則訊息、或者覆蓋先前的思考草稿,務必為主動修訂的訊息指派固定且一致的id。
效能與實務提醒
狀態是 Agent 系統運作的心臟,合理的狀態設計能帶來顯著的效能收益:
第一點是對話視窗長度控制。雖然 add_messages 提供了無限累積訊息的能力,但大型語言模型的上下文視窗(Context Window)始終有物理極限,且過長的歷史訊息會導致每一次 API 呼叫的計費飆升。在生產環境中,必須搭配訊息修剪策略(例如使用 langchain_core.messages.trim_messages),在狀態送入模型前自動保留最近的 K 則訊息或限制總權杖數上限,避免無止境累積拖垮效能。
第二點是大型二進位資料與向量內容的存放分離。不要把大容量的 PDF 原始文字、高解析度圖片或龐大的向量陣列直接塞進狀態的 TypedDict 中。狀態中的每一個欄位在每一次節點躍遷時都會被序列化並記入檢查點快照。過於龐大的狀態物件會大幅拖慢磁碟 I/O。正確做法是將大檔案儲存在外部物件儲存或 SQLite(如 data/knowledge.db)中,狀態字典內僅保留指向該資源的 URI 或資料表主鍵識別碼。
第三點是型別安全性維護。在大型協同專案中,隨意向狀態中塞入未定義型別的鍵值會破壞系統的可維護性。強烈建議在 CI/CD 管線中開啟 mypy --strict 或 pyright,讓靜態型別檢查器在編譯前捕捉所有針對狀態欄位的拼寫錯誤與型別不符問題。
小結
今天我們深入解構了 LangGraph 的狀態聚合心臟。我們釐清了 TypedDict 的預設覆蓋行為與工程價值,掌握了透過 typing.Annotated 綁定縮減器(Reducer)的數學本質與實作語法。我們在 research-agent 專案中實作了自訂的標籤去重複縮減器與步數計數器,並全面剖析了官方 add_messages 縮減器在追加、識別碼覆蓋以及 RemoveMessage 修剪上的強大機制。
透過縮減器的宣告,我們的狀態圖擺脫了被動的整包覆蓋,獲得了優雅、可控的集合累積與時間維度管理能力,為建構具備深厚記憶的多輪代理打下了最關鍵的架構基石。
結語
至此,我們已經掌握了固定邊的線性躍遷以及狀態字典的精準累積。然而,現實中的智慧代理之所以被稱為「智慧」,關鍵在於它能夠「依據當前情境做出自主抉擇」:若檢索資料充足,則直接撰寫結論;若資訊存在矛盾,則轉向延伸搜尋;若發生錯誤,則轉向自我修復節點。這種動態抉擇在 LangGraph 中正是透過「條件邊(Conditional Edges)」來實現。
明天,我們會進入「AG Day 13 條件邊:讓流程學會分岔」,學習如何使用條件邊、路由函式與路徑對照字典(Path Mapping),讓我們的研究助理學會自主思考、評估條件並在不同節點之間靈活分岔,正式解鎖複雜決策圖的強大威力。
延伸資源
- LangGraph 狀態管理官方文件:
https://langchain-ai.github.io/langgraph/concepts/low_level/#state。深入了解狀態與縮減器的運作底層。 - LangGraph add_messages API 規格說明:
https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.message.add_messages。查閱訊息聚合與修剪的詳細行為。 - Python typing.Annotated 官方規格說明(PEP 593):
https://peps.python.org/pep-0593/。掌握現代型別元資料附加語法。 - LangChain Core 訊息架構指南:
https://python.langchain.com/docs/concepts/messages/。了解 BaseMessage 家族的完整定義。
留言
張貼留言