BranchJoiner

BranchJoiner

이 컴포넌트로 파이프라인의 서로 다른 가지(branch)를 하나의 출력으로 합칠 수 있어요.

파이프라인에서 가장 흔한 위치: 유연함 — 파이프라인 시작이나 루프의 시작에 올 수 있음 필수 init 변수: type_ — 앞선 컴포넌트들이 기대하는 데이터 타입 필수 run 변수: **kwargs — 초기화 때 정의한 어떤 입력 데이터 타입이든. 이 입력은 variadic이라 가변 개수의 컴포넌트를 연결할 수 있어요. 출력 변수: value — 연결된 컴포넌트들에서 받은 첫 번째 값 API reference: Joiners GitHub link: https://github.com/deepset-ai/haystack/blob/main/haystack/components/joiners/branch.py Package name: haystack-ai

출처: 문서

본문

Overview

BranchJoiner는 파이프라인의 여러 가지를 합쳐, 그 출력을 단일 가지로 통합하게 해줘요. 다음에 오는 단일 컴포넌트로 넘어가기 전에 여러 가지를 통합해야 하는 파이프라인에서 특히 유용합니다.

BranchJoiner는 다른 컴포넌트들로부터 같은 타입의 여러 데이터 연결을 받아, 받은 첫 번째 값을 단일 출력으로 전달해요. 파이프라인에서 루프를 닫거나 결정 컴포넌트의 여러 가지를 조정하는 데 필수적이죠.

BranchJoiner는 __init__ 함수에서 선언된 하나의 데이터 타입의 입력 하나만 처리할 수 있어요. 파이프라인 가지들에 걸쳐 데이터 타입이 일관되게 유지되도록 보장합니다. run이 호출될 때 입력에 대해 값이 두 개 이상 받아지면 컴포넌트가 오류를 발생시켜요:

from haystack.components.joiners import BranchJoiner

bj = BranchJoiner(int)
bj.run(value=[3, 4, 5])
# ValueError: BranchJoiner expects only one input, but 3 were received.

Usage

On its own

매 실행마다 입력 값은 하나만 허용되지만, variadic 특성 때문에 BranchJoiner는 여전히 목록을 기대합니다. 예시:

from haystack.components.joiners import BranchJoiner

# an example where input and output are strings
bj = BranchJoiner(str)
bj.run(value=["hello"])
# {"value" : "hello"}

# an example where input and output are integers
bj = BranchJoiner(int)
bj.run(value=[3])
# {"value": 3}

In a pipeline

Enabling loops

아래는 BranchJoiner로 루프를 닫는 예시예요. 이 예시에서 BranchJoiner는 JsonSchemaValidator로부터 돌아온 ChatMessage 객체 목록을 받아, 재생성을 위해 OpenAIChatGenerator로 내려보냅니다.

import json
from haystack import Pipeline
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.joiners import BranchJoiner
from haystack.components.validators import JsonSchemaValidator
from haystack.dataclasses import ChatMessage

person_schema = {
    "type": "object",
    "properties": {
        "first_name": {"type": "string", "pattern": "^[A-Z][a-z]+$"},
        "last_name": {"type": "string", "pattern": "^[A-Z][a-z]+$"},
        "nationality": {
            "type": "string",
            "enum": ["Italian", "Portuguese", "American"],
        },
    },
    "required": ["first_name", "last_name", "nationality"],
}

# Initialize a pipeline
pipe = Pipeline()
# Add components to the pipeline
pipe.add_component("joiner", BranchJoiner(list[ChatMessage]))
pipe.add_component("fc_llm", OpenAIChatGenerator(model="gpt-4.1-mini"))
pipe.add_component("validator", JsonSchemaValidator(json_schema=person_schema))

# Connect components
pipe.connect("joiner", "fc_llm")
pipe.connect("fc_llm.replies", "validator.messages")
pipe.connect("validator.validation_error", "joiner")

result = pipe.run(
    data={
        "fc_llm": {"generation_kwargs": {"response_format": {"type": "json_object"}}},
        "joiner": {
            "value": [ChatMessage.from_user("Create json object from Peter Parker")],
        },
    },
)
print(json.loads(result["validator"]["validated"][0].text))
# >> {'first_name': 'Peter', 'last_name': 'Parker', 'nationality': 'American', 'name': 'Spider-Man', 'occupation':
# >> 'Superhero', 'age': 23, 'location': 'New York City'}

Reconciling branches

이 예시에서 TextLanguageRouter 컴포넌트는 쿼리를 세 개의 언어별 Retriever 중 하나로 보내요. 다음 컴포넌트는 PromptBuilder이겠지만, 여러 Retriever를 단일 PromptBuilder에 직접 연결할 수는 없죠. 대신 모든 Retriever를 BranchJoiner 컴포넌트에 연결합니다. BranchJoiner는 실제로 호출된 Retriever에서 출력을 받아 단일 문서 목록으로 PromptBuilder에 전달해요. BranchJoiner는 Retriever들의 서로 다른 출력을 추가 처리를 위한 통합 연결로 통합해, 파이프라인이 여러 언어를 매끄럽게 처리할 수 있게 보장합니다.

이 페이지의 예시는 langdetect-haystack 패키지의 언어 분류 컴포넌트를 사용해요. 예시를 실행하려면 설치하세요:

pip install langdetect-haystack

from haystack import Document, Pipeline
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.joiners import BranchJoiner
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack_integrations.components.routers.langdetect import TextLanguageRouter
from haystack.dataclasses import ChatMessage

prompt_template = [
    ChatMessage.from_user(
        """Answer the question based on the given reviews.
Reviews:  
  {% for doc in documents %}
    {{ doc.content }}
  {% endfor %}
Question: {{ query}}
Answer:""",
    ),
]
documents = [
    Document(
        content="Super appartement. Juste au dessus de plusieurs bars qui ferment très tard. A savoir à l'avance. (Bouchons d'oreilles fournis !)",
    ),
    Document(
        content="El apartamento estaba genial y muy céntrico, todo a mano. Al lado de la librería Lello y De la Torre de los clérigos. Está situado en una zona de marcha, así que si vais en fin de semana , habrá ruido, aunque a nosotros no nos molestaba para dormir",
    ),
    Document(
        content="The keypad with a code is convenient and the location is convenient. Basically everything else, very noisy, wi-fi didn't work, check-in person didn't explain anything about facilities, shower head was broken, there's no cleaning and everything else one may need is charged.",
    ),
    Document(
        content="It is very central and appartement has a nice appearance (even though a lot IKEA stuff), *W A R N I N G** the appartement presents itself as a elegant and as a place to relax, very wrong place to relax - you cannot sleep in this appartement, even the beds are vibrating from the bass of the clubs in the same building - you get ear plugs from the hotel.",
    ),
    Document(
        content="Céntrico. Muy cómodo para moverse y ver Oporto. Edificio con terraza propia en la última planta. Todo refor..." ,
    ),
]
# ... (파이프라인 구성 계속, 원문 참조)

더 알아보기 (Learn more)

  • Joiners — Joiner 컴포넌트
  • Routing — 파이프라인 라우팅