iT邦幫忙

2026 iThome 鐵人賽

DAY 21
0
AI Engineering

AI Agent 系統開發 30 天系列 第 21 篇

Agent 記憶與狀態持久化

  • 分享至 

  • xImage
  •  

呼叫大型語言模型(LLM)的 API 是無狀態(Stateless)的。在先前的 讓 LangGraph 接手對話狀態管理 中,我們已使用 Checkpointer 依據 thread_id 實現了單一對話 Session 的短期狀態管理。

除了單次對話的短期狀態,真實應用中的 Agent 同樣需要能跨越 Session、持久保存使用者偏好的長期記憶。兩者的生命週期與資料範圍不同,識別碼也不同:短期記憶以 thread_id 識別,換了新的 thread_id 就代表開啟新對話,對話紀錄重置為空白,避免舊話題干擾新任務;長期記憶以 user_id 識別,同一名使用者不論開啟多少次新對話,都能存取相同的設定。

以旅遊客服為例,旅客在對話中提到的「偏好靠窗座位」或「習慣家庭旅遊」,屬於長久有效的使用者屬性。長期記憶會將這些結論提取出來寫入掛在 user_id 底下的 Profile(可以是在使用者說「請記住」時透過 Tool 即時寫入,或在對話 Session 結束時由模型抽取存入)。當同一個旅客換了新的 thread_id 詢問新行程時,上一趟旅行的景點對話已被清空,但 Agent 依然能從長期記憶載入偏好,自動安排靠窗座位與適合家庭的行程。

長期記憶:用 Store 隔離與保存跨對話的使用者 Profile

在規劃 Agent 的長期記憶時,我們不需要將整段聊天歷史原封不動存進長期資料庫,否則既浪費空間,又會讓每次讀取的上下文膨脹。

因為這是專門處理旅遊諮詢的 Agent,業務上記錄的資訊範圍非常明確:旅客的座位偏好、旅遊類型(如家庭旅遊、獨旅)與回覆語言。對於這種特定領域的 Agent,我們會事先定義好強型別的 Schema,確保只有符合預期的屬性才能進入長期記憶庫。

範例中定義的 TravelerProfile 只允許三項明確的業務欄位進入長期記憶庫:

  • 座位偏好(seat_preference):允許 window(靠窗)、aisle(靠走道)、no_preference。
  • 旅遊類型偏好(travel_style):允許 family(家庭旅遊)、solo(獨自旅遊)、couple(雙人旅遊)、business(商務)、no_preference。
  • 回覆語言偏好(response_language):允許 zh-TW(繁中)、en(英文)、ja(日文)。

使用 Pydantic 配合 Literal 枚舉型別與 extra="forbid",能防止 LLM 生成像「想坐舒服一點」這種非標準、無法對接 API 的自由文字,確保記憶庫欄位品質:

class TravelerProfile(BaseModel):
    model_config = ConfigDict(extra="forbid")

    seat_preference: Literal["window", "aisle", "no_preference"] | None = None
    travel_style: (
        Literal["family", "solo", "couple", "business", "no_preference"]
        | None
    ) = None
    response_language: Literal["zh-TW", "en", "ja"] | None = None

為避免不同使用者的偏好互相混淆,Store 採用命名空間(Namespace)建立隔離邊界:

def profile_namespace(user_id: str) -> tuple[str, ...]:
    return ("travelers", user_id, "profile")

每輪對話執行時,chatbot 節點會先透過 user_id 從 Store 讀取該名使用者的 Profile,並將其轉化為 System Prompt 的一部分注入模型:

def chatbot(state: MessagesState, runtime: Runtime[Context]):
    profile = read_profile(runtime.store, runtime.context.user_id)
    profile_prompt = render_profile(profile)
    system_prompt = BASE_SYSTEM_PROMPT
    if profile_prompt:
        system_prompt = f"{system_prompt}\n\n{profile_prompt}"

    response = model_with_tools.invoke(
        [SystemMessage(content=system_prompt), *state["messages"]]
    )
    return {"messages": [response]}

透過這種方式,無論開啟多少個新的 thread_id,只要 user_id 保持一致,模型都能掌握使用者的長遠偏好。

長期記憶的寫入時機:即時 Tool 與 Session 結束抽取

將偏好存入 Store 有兩條路徑:

1. 即時 Tool 寫入

當使用者說出「請記住」或「以後都要」等明確指令時,模型的最佳選擇是呼叫 Tool。定義寫入偏好的 Tool:

@tool
def save_traveler_preference(field: PreferenceField, value: str) -> str:
    """保存旅客明確要求跨旅行沿用的偏好。"""
    return f"已保存旅客偏好:{field}={value}"

當模型判定需要記錄偏好時,會觸發 Tool 呼叫,Graph 路由至 save_preference 節點:驗證欄位是否合規、與既有 Profile 合併,並寫回 Store。當輪對話便能立即看到更新後的記憶生效。

2. Session 結束抽取(Extractor)

使用者在聊天過程中未必每次都會說「請記住」。當輸入 /new 或 /exit 結束對話 Session 時,系統會取得該 Session 的完整對話紀錄,傳給設定了結構化輸出的 Extractor 模型:

snapshot = graph.get_state(config)
messages = snapshot.values["messages"]

result = extractor.invoke(
    [
        SystemMessage(content=EXTRACTION_SYSTEM_PROMPT),
        HumanMessage(content=conversation_text(messages)),
    ]
)
update = TravelerProfileUpdate.model_validate(result)

Extractor Prompt 會過濾掉目的地、日期與預算等一次性需求,只擷取符合 TravelerProfile 的通用偏好(如家庭旅遊風格),並自動更新至 Store。這讓 Agent 不用在每輪對話頻繁呼叫工具,也能在背景補齊使用者畫像。

組裝與呼叫:在 Graph 中掛載 Store 與 Checkpointer

範例專案的完整實作位於 langgraph-traveler-memory。

定義好負責對話的 chatbot 節點、負責寫入偏好的 save_preference 節點與 Store 之後,最後一步是將它們組裝進 StateGraph 並完成編譯。

在 LangGraph 中,節點無法憑空存取外部儲存;我們必須在 compile() 時將 store 注入 Graph,節點的 runtime.store 才能正常運作。同時,為了維持多輪對話上下文,我們也會一併掛上負責短期記憶的 checkpointer。

1. 在 build_graph() 中 compile

在 main.py 的 build_graph() 函式中,我們宣告圖的拓撲結構,並在 compile() 時同時注入兩組儲存後端:

# 位於 main.py 的 build_graph()
def build_graph(model: Any, *, checkpointer: Any, store: BaseStore) -> Any:
    ...
    builder = StateGraph(MessagesState, context_schema=Context)
    builder.add_node("chatbot", chatbot)
    builder.add_node("save_preference", save_preference)
    builder.add_edge(START, "chatbot")
    builder.add_conditional_edges("chatbot", route_after_model)
    builder.add_edge("save_preference", "chatbot")

    # 同時掛載 Checkpointer(短期)與 Store(長期)
    return builder.compile(
        checkpointer=checkpointer,
        store=store,
    )

2. 在 main() 中呼叫 graph.invoke()

在 main.py 的進入點函式 main() 中,實體化儲存物件並執行對話迴圈。呼叫 graph.invoke() 時,兩組識別鍵值必須透過不同通道傳入:

# 位於 main.py 的 main() 進入點
store = InMemoryStore()
graph = build_graph(
    model,
    checkpointer=InMemorySaver(),
    store=store,
)

# 呼叫時雙通道分流:
result = graph.invoke(
    {"messages": [HumanMessage(content=question)]},
    {"configurable": {"thread_id": session_id}},  # 短期對話通道(給 Checkpointer)
    context=Context(user_id=user_id),  # 長期儲存通道(給 Store)
)
  • configurable={"thread_id": session_id}:短期對話通道。傳給 Checkpointer 用於定位該次 Session 的狀態快照。
  • context=Context(user_id=user_id):長期儲存通道。傳入執行環境,供節點從 Store 存取該名使用者的 Profile。

這套組合正是實現「清空舊對話、保留舊偏好」的關鍵:

  1. 短期記憶接管:Checkpointer 依據 thread_id 自動載入上一輪 Checkpoint 狀態,追加本輪傳入的 HumanMessage,執行節點並更新 State,執行完畢後自動將最新 State 寫回 Checkpoint。
  2. 長期記憶注入:chatbot 節點執行時,依據傳入的 user_id 從 Store 讀取長久 Profile,將其動態渲染注入 System Prompt;若對話中觸發 Tool,則由 save_preference 節點更新至 Store。
  3. 換 Session 隔離驗證:換成新的 thread_id 時,Checkpointer 找不到過去紀錄,自然會以全新的空白 State 開始執行,上一趟對話細節完全不外流;但因為 user_id 不變,Agent 依然能精準載入長期偏好!

在開發與測試階段,使用 InMemorySaver 與 InMemoryStore 可以省去資料庫架設與設定檔維護。然而,記憶體後端僅存在於 Python 程序執行期間,關閉 CLI 或重啟伺服器後資料即會消失。

正式上線時,不需要修改 Graph 的架構邏輯或節點程式碼,只需將編譯時的後端替換為持久化資料庫套件(例如 PostgreSQL Saver 與 Store):

# 生產環境替換範例
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore

# checkpointer 與 store 改為連線至實體資料庫
graph = builder.compile(
    checkpointer=postgres_checkpointer,
    store=postgres_store,
)

不論底層是記憶體陣列還是 PostgreSQL 資料庫,thread_id 控制對話生命週期、user_id 隔離使用者記憶的核心原則皆維持不變。


上一篇
用預先定義的欄位約束 Agent 的資料查詢
系列文
AI Agent 系統開發 30 天 共 21 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言