將 Python 工作流程檢查點升級到 1.13.0

Agent Framework 1.13.0 包含對 Python 工作流程執行的輕微破壞性變更。 大多數應用程式 不需要 更改。 這些變更影響依賴精確超步數或迭代次數、設定 max_iterations 於收斂邊界、檢查初始訊息來源 ID,或對檢查點放置與排序做出假設的應用程式。

Background

在 1.13.0 之前,檢查點機制並未完全實現其承諾:完整擷取可從任何已記錄的邊界點恢復執行所需的工作流程狀態。 啟動執行器在超步驟與檢查點迴圈之前執行,因此最早的檢查點包含啟動執行器的輸出與更新狀態,但不含原始工作流程輸入。 同樣地,對請求事件的回應會在未先記錄於檢查點的情況下交付與處理。 因此,沒有任何檢查點能夠從原始輸入重新執行起始執行器,或根據已傳回的回應重現人工介入的後續流程。

行為變更

1.13.0 版本彌補了這些缺口。 啟動執行器現在在第一個超步驟執行,進入檢查點記錄該超步驟前的初始輸入,回應輸入檢查點則記錄處理前已交付的回應。 這些變更共同讓採用檢查點的工作流程可根據其輸入完整重現,包括人工介入的後續流程。

這很重要

這些變更不會影響 1.13.0 版本之前建立的檢查點。 現有的檢查點仍可支援,升級後仍可恢復。

可能需要採取行動的變更

Area 1.13.0 之前 在 1.13.0 及之後版本中 使用者影響
啟動執行人 啟動執行器在超步迴圈之前執行。 輸入會排入啟動執行器的佇列,而該執行器會在第一個超步中執行。 每次重新執行都會額外發出一個 superstep_started 事件和一個 superstep_completed 事件。
迭代次數 迭代 1 代表啟動執行器執行後的第一個超步驟。 迭代 1 執行啟動執行器。 後期的工作會依序輪流一次。 過去需要 $N 美元迭代的工作流程,現在需要 $N + 1 美元。
輸入訊息來源 最初的訊息有硬編碼的來源 ID "Workflow" 初始訊息透過啟動執行者的內部邊緣傳送,來源 ID INTERNAL_SOURCE_ID(start_executor.id)為 。 讀取或過濾初始訊息來源 ID 的程式碼必須使用該新值。

重玩性改進

Area 1.13.0 之前 在 1.13.0 及之後版本中 改進
初始檢查站 迭代-0檢查點是在啟動執行器執行後建立的。 它會擷取執行者的輸出訊息和更新狀態,但不會捕捉原始輸入。 進入檢查點會在超級步驟 1 前建立。 它會記錄為啟動執行者排隊的原始輸入。 恢復進入檢查點會重播完整執行過程,包括起始執行器。
回應檢查點 對請求事件的回應在未先記錄於檢查點的情況下交付。 回應進入檢查點會在回應傳送完成後、處理該回應的超步驟執行前建立。 還原回應項目檢查點會重新執行取用該回應的接續作業。

更新超步驟事件處理

現在重新執行一次工作流程時,會多產生一對超級步驟事件,因為起始執行器是在超級步驟 1 執行:

  • superstep_startediteration == 1
  • superstep_completediteration == 1

後續執行人的工作會多一個超階。 更新測試、遙測、進度指示器或其他假設精確事件計數或將特定執行器映射到固定迭代的程式碼。

能夠回應事件類型而不依賴事件數量或迭代次數的程式碼,則不需要改變。

檢視最大迭代次數限制

max_iterations 限制現在也包含執行啟動執行器的超步在內。 如果工作流程之前已經用盡了限制,則將設定值增加一:

from agent_framework import WorkflowBuilder

workflow = WorkflowBuilder(
    start_executor=start_executor,
    max_iterations=previous_max_iterations + 1,
).build()

如果工作流程在達到設定限制前就已經收斂,則不需要更改。

更新初始訊息來源檢查

如果啟動執行器取用了初始訊息的來源 ID,請將硬式編碼的 "Workflow" 值替換為啟動執行器內部邊的來源 ID。

1.13.0 之前:

is_workflow_input = ctx.source_executor_ids != ["Workflow"]

在 1.13.0 及之後版本中:

from agent_framework import INTERNAL_SOURCE_ID

is_workflow_input = ctx.source_executor_ids != [INTERNAL_SOURCE_ID(self.id)]

INTERNAL_SOURCE_ID(executor_id) 目前回傳 "internal:<executor_id>"。 使用輔助工具代替建構這個字串,讓你的程式碼遵循框架的原始碼 ID 格式。

更新檢查點處理方式

初始輸入檢查點

啟用檢查點後,每次全新執行現在都會在 iteration_count == 0 建立一個入口檢查點。 此檢查點包含原始輸入,作為發送給啟動執行器的飛行中訊息。 恢復後會重新執行啟動執行器,並重現完整的工作流程執行。

完成每個超步驟後,框架會持續建立一個檢查點。 對於一次有 $N$ 個超級步驟的執行,預計會有 $N+1$ 個檢查點:先是一個初始檢查點,之後每個已完成的超級步驟各有一個檢查點。

審查程式碼假設迭代-0檢查點包含啟動執行器產生的狀態。 該狀態現在會出現在超級步驟 1 後建立的檢查點中。

請求回應檢查點

當你使用 workflow.run(responses=...) 繼續工作流程時,框架現在會在將回應加入佇列後、執行處理這些回應的超級步驟前,建立一個回應項目檢查點。 恢復這個檢查點會重新傳送錄製的回應,並重播整個工作流程。

回應輸入檢查點與前一個包含待處理請求的檢查點相同 iteration_count 。 它是一個獨立的檢查點,其 previous_checkpoint_id 指向該待處理請求檢查點。

這很重要

An iteration_count 並不保證在人類參與的檢查點歷史中是唯一的。 依照 previous_checkpoint_id 鏈來判定檢查點順序。 如果你需要最新的檢查點,請使用 檢查點儲存 API,而不是選擇最大的 iteration_count

遷徙檢查清單

  • 更新依賴精確超步計數或迭代次數的斷言與事件取用端。
  • 僅對已達到先前限制的工作流程,將 max_iterations 增加 1。
  • 將初始來源 ID 檢查 "Workflow" 替換為 INTERNAL_SOURCE_ID(start_executor.id)
  • 將迭代 0 檢查點視為執行前的輸入檢查點。
  • 應該依照血統排序人為迴圈檢查點,而不是假設 iteration_count 它是唯一的。
  • 驗證重播某個進入檢查點和回應進入檢查點時,是否會產生預期的輸出和副作用。

如需實作細節,請參閱 允許工作流程檢查點具有完整可重播性