파이프라인 직렬화

파이프라인 직렬화 (Serializing Pipelines)

파이프라인을 커스텀 포맷으로 저장하고, 직렬화 옵션들을 함께 살펴봐요.

**직렬화(serialization)**는 파이프라인을 디스크에 저장했다가 나중에 다시 불러올 수 있는 포맷으로 변환하는 것을 뜻해요. Haystack은 파이프라인 직렬화를 위해 YAML 포맷을 지원합니다.

출처: 공식문서

파이프라인을 YAML로 변환하기 (Converting a Pipeline to YAML)

dumps() 메서드를 사용하면 Pipeline 객체를 YAML로 변환할 수 있어요.

from haystack import Pipeline

pipe = Pipeline()
print(pipe.dumps())
# >> components: {}
# >> connections: []
# >> max_runs_per_component: 100
# >> metadata: {}

또한 dump() 메서드로 파이프라인의 YAML 표현을 파일에 저장할 수도 있습니다.

with open("/content/test.yml", "w") as file:
    pipe.dump(file)

파이프라인을 다시 Python으로 변환하기 (Converting a Pipeline Back to Python)

YAML 파이프라인을 다시 Python으로 변환할 수 있어요. loads() 메서드는 파이프라인의 문자열 표현(str, bytes 또는 bytearray)을, load() 메서드는 파일류 객체(file-like object)로 표현된 파이프라인을 대응하는 Python 객체로 변환합니다.

두 로딩 메서드 모두 역직렬화 과정 중에 컴포넌트를 수정할 수 있는 콜백을 지원해요. 역직렬화는 **신뢰된 모듈 허용 목록(allowlist)**으로 제한되므로, 허용 목록 밖의 클래스를 참조하는 파이프라인은 허용 목록을 확장할 때까지 로드에 실패합니다. 자세한 내용은 아래 역직렬화 보안에서 볼게요.

다음은 그 예제 스크립트입니다.

from haystack import Pipeline
from haystack.core.serialization import DeserializationCallbacks
from typing import Type, Dict, Any

# This is the YAML you want to convert to Python:
pipeline_yaml = """
components:
  cleaner:
    init_parameters:
      remove_empty_lines: true
      remove_extra_whitespaces: true
      remove_regex: null
      remove_repeated_substrings: false
      remove_substrings: null
    type: haystack.components.preprocessors.document_cleaner.DocumentCleaner
  converter:
    init_parameters:
      encoding: utf-8
    type: haystack.components.converters.txt.TextFileToDocument

connections:
- receiver: cleaner.documents
  sender: converter.documents
max_runs_per_component: 100
metadata: {}
"""

def component_pre_init_callback(
    component_name: str,
    component_cls: Type,
    init_params: Dict[str, Any],
):
    # This function gets called every time a component is deserialized.
    if component_name == "cleaner":
        assert "DocumentCleaner" in component_cls.__name__
        # Modify the init parameters. The modified parameters are passed to
        # the init method of the component during deserialization.
        init_params["remove_empty_lines"] = False
        print("Modified 'remove_empty_lines' to False in 'cleaner' component")
    else:
        print(f"Not modifying component {component_name} of class {component_cls}")

pipe = Pipeline.loads(
    pipeline_yaml,
    callbacks=DeserializationCallbacks(component_pre_init_callback),
)

역직렬화 보안 (Deserialization Security)

파이프라인을 로드하면 직렬화된 데이터에 참조된 클래스들이 인스턴스화됩니다. 조작된 YAML 파일이 임의의 클래스를 import 하고 인스턴스화하는 것을 막기 위해, Pipeline.load, Pipeline.loads, Pipeline.from_dict는 신뢰된 모듈 허용 목록 밖의 모듈에서 클래스를 import 하는 것을 거부하고 대신 DeserializationError를 발생시킵니다.

기본적으로 허용 목록에는 haystack, haystack_integrations, haystack_experimental, builtins, typing, collections가 들어 있어요. 위험한 내장 eval, exec, compile, __import__, open, getattr은 builtins가 허용 목록에 있더라도 차단됩니다.

커스텀 모듈 허용하기 (Allowing Custom Modules)

다른 패키지에 있는 커스텀 컴포넌트나 콜러블을 참조하는 파이프라인은, 그 모듈을 허용 목록에 추가할 때까지 로드에 실패해요. 허용 목록을 확장하는 방법은 세 가지입니다.

from haystack import Pipeline

# 1. Per call: pass additional module patterns for this deserialization only
pipe = Pipeline.load(open("pipeline.yaml"), allowed_modules=["mypkg.*"])

# 2. Process-wide: extend the allowlist programmatically
from haystack.core.serialization import allow_deserialization_module
allow_deserialization_module("mypkg")

# 3. Environment variable with comma-separated patterns, read on every deserialization call
export HAYSTACK_DESERIALIZATION_ALLOWLIST="mypkg.*,otherpkg.*"

패턴들은 기본적으로 접두사로 일치합니다("mypkg"는 mypkg와 그 모든 하위 모듈에 일치). 끝의 .*가 아닌 곳에 *, ?, [가 들어 있으면 fnmatch 글로브로 일치해요. 끝의 .*는 접두사 일치로 취급되므로 "mypkg"와 "mypkg.*"는 동일하게 동작합니다.

직렬화된 데이터의 출처를 완전히 신뢰할 수 있다면, unsafe=True로 허용 목록을 아예 우회할 수 있어요.

pipe = Pipeline.load(open("pipeline.yaml"), unsafe=True)

unsafe=True는 직렬화된 파이프라인이 어디서 왔는지 완전히 신뢰할 때만 사용하세요. 위험한 내장에 대한 차단도 함께 풀리기 때문입니다.

중첩 init 파라미터 검증 (Nested Init Parameter Validation)

추가 안전장치로, 역직렬화는 중첩된 {"type": "...", "init_parameters": {...}} 딕셔너리로 재귀하기 전에 init_parameters의 키를 클래스의 __init__ 시그니처에 대해 검증합니다. 허용되지 않은 파라미터 이름이 키인 중첩 딕셔너리는 중첩 타입이 import 되기 전에 DeserializationError로 거부되어, 사용되지 않는 파라미터 슬롯에 신뢰할 수 없는 클래스를 밀어 넣는 시도를 차단해요. 생성자가 **kwargs를 받는 클래스는 허용되는 파라미터 집합을 정적으로 결정할 수 없으므로 면제됩니다.

이 검증은 YAML 파일에 있던 기존 버그를 드러낼 수 있어요. 예를 들어 오타, 이름이 바뀌거나 제거된 파라미터의 잔재, 더 오래된 Haystack 버전에서 온 낡은 스냅샷 같은 것들이죠. 해결책은 YAML을 갱신해서 각 중첩 컴포넌트의 키가 부모 클래스의 실제 __init__ 파라미터와 일치하도록 하는 것입니다.

기본 직렬화 동작 (Default Serialization Behavior)

직렬화 시스템은 default_to_dict와 default_from_dict를 사용해 많은 객체 타입을 자동으로 처리해요. 보통 다음 항목들에 대해 커스텀 to_dict/from_dict를 구현할 필요가 없습니다.

  • Secret: 민감한 값이 평문으로 저장되지 않도록 자동으로 직렬화·역직렬화됩니다.
  • ComponentDevice: 디바이스 설정이 자동으로 감지되고 복원됩니다.
  • 자체 to_dict/from_dict가 있는 객체: to_dict()를 정의한 타입의 init 파라미터는 그것을 호출해 직렬화하며, from_dict()가 있는 클래스를 가리키는 type 키가 있는 init_parameters 안의 딕셔너리는 자동으로 역직렬화됩니다.

단일 컴포넌트를 직렬화하거나 역직렬화하려면 haystack.core.serialization의 component_to_dict와 component_from_dict를 사용할 수 있어요. 이들은 컴포넌트가 커스텀 to_dict/from_dict를 정의하지 않았을 때 위의 기본 동작을 폴백으로 사용합니다.

from haystack import component
from haystack.core.serialization import component_from_dict, component_to_dict

@component
class Greeter:
    def __init__(self, message: str = "Hello"):
        self.message = message

    @component.output_types(greeting=str)
    def run(self, name: str):
        return {"greeting": f"{self.message}, {name}!"}

# Serialize a component instance to a dictionary
greeter = Greeter(message="Hi")
data = component_to_dict(greeter, "my_greeter")

# Deserialize back to a component instance
restored = component_from_dict(Greeter, data, "my_greeter")
assert restored.message == greeter.message

:::note init 파라미터는 인스턴스 속성으로 저장되어야 합니다. 기본 직렬화는 init 파라미터 이름과 인스턴스 속성 사이에 1:1 매핑이 있을 때만 동작해요. __init__의 모든 인자에 대해, 컴포넌트는 같은 이름의 속성에 그것을 할당해야 합니다. 예를 들어 def __init__(self, prompt: str)가 있다면 클래스 안에 self.prompt = prompt가 있어야 합니다. 그렇지 않으면 직렬화 로직이 직렬화할 값을 찾지 못해 오류를 내거나, 파라미터에 기본값이 있으면 기본값을 사용합니다. :::

커스텀 직렬화 수행하기 (Performing Custom Serialization)

Haystack의 파이프라인과 컴포넌트는 커스텀 컴포넌트를 포함한 간단한 컴포넌트를 기본으로 직렬화할 수 있어요. 이런 코드는 그냥 동작합니다.

from haystack import component

@component
class RepeatWordComponent:
    def __init__(self, times: int):
        self.times = times

    @component.output_types(result=str)
    def run(self, word: str):
        return word * self.times

반면 아래 코드는 최종 포맷이 JSON이면 동작하지 않아요. set 타입은 JSON으로 직렬화할 수 없기 때문이죠.

from haystack import component

@component
class SetIntersector:
    def __init__(self, intersect_with: set):
        self.intersect_with = intersect_with

    @component.output_types(result=set)
    def run(self, data: set):
        return data.intersection(self.intersect_with)

이런 경우 컴포넌트에 직접 from_dict와 to_dict 구현을 제공할 수 있어요.

from haystack import component, default_from_dict, default_to_dict

class SetIntersector:
    def __init__(self, intersect_with: set):
        self.intersect_with = intersect_with

    @component.output_types(result=set)
    def run(self, data: set):
        return data.intersect(self.intersect_with)

    def to_dict(self):
        return default_to_dict(self, intersect_with=list(self.intersect_with))

    @classmethod
    def from_dict(cls, data):
        # convert the set into a list for the dict representation,
        # so it can be converted to JSON
        data["intersect_with"] = set(data["intersect_with"])
        return default_from_dict(cls, data)

파이프라인을 커스텀 포맷으로 저장하기 (Saving a Pipeline to a Custom Format)

파이프라인을 dictionary 포맷으로 만들었다면, 직렬화의 마지막 단계는 그 dictionary를 저장하거나 네트워크로 보낼 수 있는 포맷으로 변환하는 거예요. Haystack은 YAML을 기본으로 지원하지만, 다른 포맷이 필요하면 커스텀 Marshaller를 작성할 수 있습니다.

Marshaller는 특정 포맷에 따라 텍스트를 dictionary로, dictionary를 텍스트로 변환하는 책임을 가진 Python 클래스예요. Marshaller는 Marshaller 프로토콜을 지켜야 하며, marshal과 unmarshal 메서드를 제공해야 합니다.

rtoml 라이브러리에 의존하는 커스텀 TOML marshaller 코드는 이렇습니다.

# This code requires a `pip install rtoml`
from typing import Dict, Any, Union
import rtoml

class TomlMarshaller:
    def marshal(self, dict_: Dict[str, Any]) -> str:
        return rtoml.dumps(dict_)

    def unmarshal(self, data_: Union[str, bytes]) -> Dict[str, Any]:
        return dict(rtoml.loads(data_))

그런 다음 Marshaller 인스턴스를 dump, dumps, load, loads 메서드에 전달할 수 있어요.

from haystack import Pipeline
from my_custom_marshallers import TomlMarshaller

pipe = Pipeline()
pipe.dumps(TomlMarshaller())
# >> 'max_runs_per_component = 100\nconnections = []\n\n[metadata]\n\n[components]\n'

더 알아보기 (Learn more)