파이프라인 중단점
파이프라인 중단점 (Pipeline Breakpoints)
복잡한 실행 흐름을 디버깅할 때 파이프라인을 일시 중지하고 재개하는 방법을 알아봐요. 중단점(Breakpoint)으로 실행을 멈춰 상태를 검사하고, 저장된 스냅샷에서 워크플로를 이어갈 수 있어요.
출처: 공식문서
소개
Haystack 파이프라인은 복잡한 실행 흐름을 디버깅하기 위한 중단점을 지원해요. Breakpoint는 특정 컴포넌트에서 실행을 일시 중지하고, 파이프라인 상태를 검사하며, 저장된 스냅샷에서 실행을 재개할 수 있게 해 줘요. 이 기능은 일반 컴포넌트뿐 아니라 Agent 컴포넌트에서도 동작해요.
파이프라인의 어떤 컴포넌트에든 특정 방문 횟수(visit count)로 Breakpoint를 설정할 수 있어요. 트리거되면 시스템이 Pipeline의 실행을 멈추고 현재 파이프라인 상태의 스냅샷을 캡처해요. 스냅샷 파일 저장이 활성화돼 있으면 상태를 JSON 파일로 저장할 수 있는데, 자세한 내용은 아래 '스냅샷 파일 저장'에서 다뤄요.
일반 컴포넌트에 Breakpoint 설정하기
컴포넌트 이름과 트리거할 방문 횟수를 지정해서 Breakpoint를 만들어요. 루프가 있는 파이프라인에서 특히 유용해요. 기본 visit_count 값은 0이에요.
from haystack.dataclasses.breakpoints import Breakpoint
from haystack.core.errors import BreakpointException
# Create a breakpoint that triggers on the first visit to the "llm" component
break_point = Breakpoint(
component_name="llm",
visit_count=0, # 0 = first visit, 1 = second visit, etc.
snapshot_file_path="/path/to/snapshots", # Optional: save snapshot to file
)
# Run pipeline with breakpoint
try:
result = pipeline.run(data=input_data, break_point=break_point)
except BreakpointException as e:
print(f"Breakpoint triggered at component: {e.component}")
print(f"Component inputs: {e.inputs}")
print(f"Pipeline results so far: {e.results}")
BreakpointException이 발생하는데, 여기에는 실행이 중단된 시점까지의 컴포넌트 입력과 파이프라인 출력이 담겨 있어요. 예를 들어 위 예제의 llm처럼 중단점과 연결된 컴포넌트가 실행되기 직전의 상태죠.
Breakpoint에 snapshot_file_path가 지정되고 스냅샷 파일 저장이 활성화되어 있으면, 시스템은 BreakpointException과 같은 정보를 담은 JSON 스냅샷을 저장해요. disk로의 스냅샷 파일 저장은 기본적으로 비활성화되어 있어요. 자세한 내용은 아래 '스냅샷 파일 저장'을 참고하세요.
커스텀 스냅샷 콜백 사용하기
Pipeline.run()에 snapshot_callback을 전달하면 파일로 저장하는 대신 스냅샷을 직접 처리할 수 있어요. 중단점이 트리거되거나 오류 시 스냅샷이 생성되면, PipelineSnapshot 객체와 함께 콜백이 호출돼요.
from haystack.core.errors import BreakpointException
from haystack.dataclasses.breakpoints import Breakpoint, PipelineSnapshot
def my_snapshot_callback(snapshot: PipelineSnapshot) -> None:
# Custom handling: e.g. save to DB, send to API, or log
print(f"Snapshot at component: {snapshot.component_name}")
snapshot_callback이 제공되면 파일 저장은 건너뛰고, 콜백이 스냅샷 처리를 책임져요.
스냅샷 파일 저장
disk로의 스냅샷 파일 저장은 기본적으로 비활성화되어 있어요. 중단점이 트리거되거나 파이프라인 실패 시 스냅샷을 JSON 파일로 저장하려면 환경 변수 HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED를 "true" 또는 "1"(대소문자 무관)로 설정하세요.
활성화되면 스냅샷은 중단점의 snapshot_file_path에 지정된 경로로 쓰여지고, 실행이 실패하면 '오류 복구 (Error Recovery with Snapshots)'의 기본 디렉터리에 쓰여져요.
import os
# Enable saving snapshot files to disk
os.environ["HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED"] = "true"
break_point = Breakpoint(
component_name="llm",
visit_count=0,
snapshot_file_path="/path/to/snapshots",
)
# When the breakpoint triggers, a JSON file will be written to /path/to/snapshots
중단점에서 파이프라인 실행 재개하기
중단점에서 파이프라인 실행을 재개하려면 파이프라인 실행 시 생성된 JSON 파일 경로를 pipeline_snapshot으로 전달해요. 먼저 load_pipeline_snapshot()으로 JSON을 불러온 뒤 파이프라인에 넘기면 돼요.
from haystack.core.pipeline.breakpoint import load_pipeline_snapshot
# Load the snapshot
snapshot = load_pipeline_snapshot("llm_2025_05_03_11_23_23.json")
# Resume execution from the snapshot
result = pipeline.run(data={}, pipeline_snapshot=snapshot)
print(result)
오류 복구와 스냅샷 (Error Recovery with Snapshots)
파이프라인은 실행이 실패하면 자동으로 마지막 유효 상태의 스냅샷을 만들어요. 스냅샷에는 실패 시점까지의 입력, 방문 횟수, 중간 출력이 포함돼요. 이를 검사하고 문제를 고친 뒤, 전체 실행을 다시 시작하는 대신 그 체크포인트에서 실행을 재개할 수 있어요.
실패 시 스냅샷 접근하기
스냅샷 파일 저장이 활성화되어 있으면(위 '스냅샷 파일 저장' 참고), Haystack은 같은 스냅샷을 disk에도 JSON 파일로 저장해요. 디렉터리는 다음 순서로 자동 선택돼요.
~/.haystack/pipeline_snapshot/tmp/haystack/pipeline_snapshot./.haystack/pipeline_snapshot
파일 이름은 {component_name}_{visit_nr}_{YYYY_MM_DD_HH_MM_SS}.json 패턴을 가져요.
스냅샷에서 재개하기
인메모리 스냅샷에서 직접 재개하거나 disk에서 불러와 재개할 수 있어요.
메모리에서 재개:
result = pipeline.run(data={}, pipeline_snapshot=snapshot)
disk에서 재개:
from haystack.core.pipeline.breakpoint import load_pipeline_snapshot
snapshot = load_pipeline_snapshot(
"/path/to/.haystack/pipeline_snapshot/reader_0_2025_09_20_12_33_10.json",
)
result = pipeline.run(data={}, pipeline_snapshot=snapshot)