파이프라인 루프

파이프라인 루프 (Pipeline Loops)

Haystack 파이프라인에서 루프가 어떻게 동작하고, 어떻게 끝나며, 피드백과 자기 수정(self-correction)에 어떻게 쓰이는지 알아봐요.

출처: 공식문서

Haystack 파이프라인은 **루프(loop)**를 지원해요. 컴포넌트 그래프에서 사이클(cycle)을 만들어, 뒤쪽 컴포넌트의 출력이 앞쪽 컴포넌트로 다시 피드백되는 구조죠. 이렇게 하면 자기 수정, 검증, 반복적 개선 같은 피드백 흐름과, 더 나아가 고급 에이전틱(agentic) 동작까지 구현할 수 있어요.

실행 시점에 파이프라인은 컴포넌트의 필수 입력이 모두 다시 준비될 때마다 그 컴포넌트를 다시 실행해요. 루프가 언제 멈출지는 그래프와 라우팅 로직을 신중하게 설계하거나, 내장된 안전 한도(safety limits)를 사용해서 제어할 수 있어요.

같은 컴포넌트의 여러 번 실행

컴포넌트가 루프에 참여하면 단일 Pipeline.run() 호출 안에서 여러 번 실행될 수 있어요. 파이프라인은 각 컴포넌트에 대해 내부 방문 카운터(visit counter)를 유지해요.

컴포넌트가 실행될 때마다 방문 횟수가 1씩 증가해요.
이 방문 횟수를 breakpoint 같은 디버깅 도구에서 사용해서 루프의 특정 반복을 검사할 수 있어요.

최종 파이프라인 결과에서:

실행된 각 컴포넌트에 대해 파이프라인은 마지막에 생성된 출력만 반환해요.
중간 컴포넌트(예: validator나 router)의 출력을 최종 결과 딕셔너리에 담고 싶다면, Pipeline.run()의 include_outputs_from 인자를 사용하세요.

루프 종료와 안전 한도

루프는 결국 멈춰야 파이프라인 실행이 완료될 수 있어요. 루프가 끝나는 주요 방법은 두 가지예요.

자연 완료(Natural completion): 더 이상 실행 가능한 컴포넌트가 없음 파이프라인은 작업 큐가 비고 어떤 컴포넌트도 다시 실행할 수 없을 때(예: router가 더 이상 루프로 입력을 피드백하지 않음) 끝나요.
최대 실행 횟수 도달 모든 파이프라인에는 컴포넌트별 실행 한도가 있는데, Pipeline 생성자의 max_runs_per_component 파라미터로 제어하며 기본값은 100이에요. 어떤 컴포넌트든 이 한도를 넘기면 Haystack은 PipelineMaxComponentRuns 오류를 발생시켜요.

이 한도를 더 낮게 설정할 수도 있어요.

from haystack import Pipeline

pipe = Pipeline(max_runs_per_component=5)

한도는 각 실행 전에 확인되므로, 한도가 3인 컴포넌트는 4번째 시도에서 오류가 발생하기 전까지 3번의 실행을 성공적으로 완료해요.

이 안전장치는 새 루프나 복잡한 라우팅 로직을 실험할 때 특히 중요해요. 루프 조건이 틀렸거나 절대 충족되지 않으면, 이 오류 덕분에 파이프라인이 무한히 계속 실행되는 일을 막아 줘요.

예제: 자기 수정을 위한 피드백 루프

다음은 간단한 피드백 루프 예제예요.

ChatPromptBuilder가 이전의 틀린 답변을 포함하는 프롬프트를 만들어요.
OpenAIChatGenerator가 답변을 생성해요.
ConditionalRouter가 답변이 맞는지 확인해요.
- 맞으면 답변을 final_answer로 보내고 루프가 끝나요.
- 틀리면 답변을 ChatPromptBuilder로 되돌려 보내 다른 반복을 트리거해요.

from haystack import Pipeline
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.routers import ConditionalRouter
from haystack.dataclasses import ChatMessage

template = [
    ChatMessage.from_system(
        "Answer the following question concisely with just the answer, no punctuation.",
    ),
    ChatMessage.from_user(
        "{% if previous_replies %}"
        "Previously you replied incorrectly: {{ previous_replies[0].text }}\n"
        "{% endif %}"
        "Question: {{ query }}",
    ),
]

prompt_builder = ChatPromptBuilder(template=template, required_variables=["query"])
generator = OpenAIChatGenerator()

router = ConditionalRouter(
    routes=[
        {
            # End the loop when the answer is correct
            "condition": "{{ 'Rome' in replies[0].text }}",
            "output": "{{ replies }}",
            "output_name": "final_answer",
            "output_type": list[ChatMessage],
        },
        {
            # Loop back when the answer is incorrect
            "condition": "{{ 'Rome' not in replies[0].text }}",
            "output": "{{ replies }}",
            "output_name": "previous_replies",
            "output_type": list[ChatMessage],
        },
    ],
    unsafe=True,  # Required to handle ChatMessage objects
)

pipe = Pipeline(max_runs_per_component=3)

pipe.add_component("prompt_builder", prompt_builder)
pipe.add_component("generator", generator)
pipe.add_component("router", router)

pipe.connect("prompt_builder.prompt", "generator.messages")
pipe.connect("generator.replies", "router.replies")
pipe.connect("router.previous_replies", "prompt_builder.previous_replies")

result = pipe.run(
    {
        "prompt_builder": {
            "query": "What is the capital of Italy? If the statement 'Previously you replied incorrectly:' is missing "
            "above then answer with Milan.",
        },
    },
    include_outputs_from={"router", "prompt_builder"},
)

print(result["prompt_builder"]["prompt"][1].text)  # Shows the last prompt used
print(result["router"]["final_answer"][0].text)  # Rome

이 루프에서 일어나는 일

첫 번째 반복

prompt_builder가 query="What is the capital of Italy?"와 이전 답변 없이 실행돼요.
generator가 LLM의 답변이 담긴 ChatMessage를 반환해요.
router가 조건을 평가해서 답변에 "Rome"이 있는지 확인해요.
답변이 틀리면 previous_replies가 prompt_builder.previous_replies로 피드백돼요.

후속 반복 (필요 시)

prompt_builder가 다시 실행되는데, 이번에는 이전의 틀린 답변이 사용자 메시지에 포함돼요.
generator가 추가 컨텍스트로 새 답변을 만들어요.
router가 다시 답변에 "Rome"이 포함돼 있는지 확인해요.

종료

router가 final_answer로 라우팅하면 더 이상 루프로 입력이 피드백되지 않아요.
큐가 비워지고 파이프라인 실행이 성공적으로 끝나요.

max_runs_per_component=3을 사용했으므로, 루프를 계속하게 만드는 예상치 못한 동작이 있으면 무한히 도는 대신 PipelineMaxComponentRuns 오류가 발생해요.

루프 구축에 유용한 컴포넌트

루프를 만드는 데 특히 유용한 컴포넌트가 두 개 있어요.

ConditionalRouter: 조건에 따라 데이터를 여러 출력으로 라우팅해요. 루프에서 빠져나갈지 계속 반복할지를 결정하는 데 사용해요. 위 예제가 이 패턴을 쓰고 있어요.
BranchJoiner: 여러 소스의 입력을 하나의 출력으로 병합해요. 루프 안의 컴포넌트가 (첫 반복에서는) 초기 입력과 (후속 반복에서는) 루프백된 값, 둘 다를 받아야 할 때 사용해요. 예를 들어 BranchJoiner로 사용자 입력과 검증 오류를 같은 Generator에 주입할 수 있어요. 완전한 루프 예제는 BranchJoiner 문서를 참고하세요.

루프에서의 Greedy vs Lazy 가변 소켓 (Variadic Sockets)

일부 컴포넌트는 단일 소켓에 여러 값을 받을 수 있는 가변 입력(variadic inputs)을 지원해요. 루프에서 가변 동작은 반복 간 입력이 어떻게 소비되는지를 제어해요.

Greedy 가변 소켓: 한 번에 정확히 하나의 값을 소비하고 컴포넌트가 실행된 뒤 그 값을 제거해요. 여기에는 사용자가 제공한 입력도 포함되는데, 그래야 무한히 재트리거되지 않아요. 대부분의 가변 소켓은 기본적으로 greedy예요.
Lazy 가변 소켓: 반복 과정에서 선행 컴포넌트로부터 받은 모든 값을 누적해요. 시간이 지나며 여러 부분 결과를 모아야 할 때 유용해요(예: 진행하기 전에 여러 루프 반복의 출력을 모으는 경우).

대부분의 루프 시나리오에서는 평소처럼 컴포넌트를 연결하고 max_runs_per_component로 실수를 방어하는 것만으로 충분해요.

루프 문제 해결

파이프라인이 멈춘 것처럼 보이거나 예상보다 오래 실행된다면, 흔한 원인과 디버깅 방법을 알아둘 필요가 있어요.

무한 루프의 흔한 원인

조건이 충족되지 않음: 종료 조건(예: 답변에 "Rome"이 있는지)이 LLM 동작이나 데이터 문제로 절대 참이 되지 않을 수 있어요. 안전망으로 항상 합리적인 max_runs_per_component을 설정하세요.
선택적 출력에 의존: 컴포넌트에 출력 소켓이 여러 개인데 그중 일부만 반환하면, 반환되지 않은 출력은 다운스트림 연결을 트리거하지 않아요. 루프에서 혼란을 일으킬 수 있죠.

예를 들어 다음 패턴은 문제가 될 수 있어요.

@component
class Validator:
    @component.output_types(valid=str, invalid=Optional[str])
    def run(self, text: str):
        if is_valid(text):
            return {"valid": text}  # "invalid" is never returned
        else:
            return {"invalid": text}

invalid를 재시도용 업스트림 컴포넌트에 연결했는데, 루프를 유지하는 다른 연결도 있다면 예상치 못한 동작이 생길 수 있어요.

대신 명시적이고 상호 배타적인 조건을 가진 ConditionalRouter를 쓰세요.

router = ConditionalRouter(
    routes=[
        {
            "condition": "{{ is_valid }}",
            "output": "{{ text }}",
            "output_name": "valid",
            "output_type": str,
        },
        {
            "condition": "{{ not is_valid }}",
            "output": "{{ text }}",
            "output_name": "invalid",
            "output_type": str,
        },
    ]
)

사용자 입력이 루프를 재트리거: 루프 안의 소켓에 사용자 제공 입력이 연결되어 있으면 루프가 예상치 못하게 다시 시작될 수 있어요.

# Problematic: user input goes directly to a component inside the loop
result = pipe.run(
    {
        "generator": {
            "prompt": query
        },  # This input persists and may retrigger the loop
    }
)

# Better: use an entry-point component outside the loop
result = pipe.run(
    {
        "prompt_builder": {"query": query},  # Entry point feeds into the loop once
    }
)

입력이 어떻게 소비되는지에 대한 자세한 내용은 "Greedy vs. Lazy Variadic Sockets"를 참고하세요.

같은 컴포넌트로 가는 여러 경로: 루프 안의 컴포넌트가 여러 소스에서 입력을 받으면, 어떤 경로든 입력을 제공할 때마다 실행돼요.

# Component receives from two sources – runs when either provides input
pipe.connect("source_a.output", "processor.input")
pipe.connect("source_b.output", "processor.input")  # Variadic input

각 경로가 언제 출력을 만들지 이해하거나, BranchJoiner로 병합 지점을 명시적으로 제어하세요.

디버깅 팁

낮은 한도로 시작: 루프를 개발할 때 max_runs_per_component=3 등으로 설정해요. 이렇게 하면 타임아웃을 기다리는 대신 명확한 오류로 문제를 조기에 잡을 수 있어요.
include_outputs_from 사용: (router 같은) 중간 컴포넌트를 추가해서 각 단계에서 무슨 일이 일어나는지 보세요.

result = pipe.run(data, include_outputs_from={"router", "validator"})

트레이싱 활성화: 트레이싱으로 입력·출력을 포함한 모든 컴포넌트 실행을 볼 수 있어요. 루프의 각 반복을 쉽게 따라갈 수 있죠. 빠른 디버깅에는 LoggingTracer(설정 안내 참고), 더 깊은 분석에는 Langfuse 같은 트레이싱 백엔드를 통합하세요.
파이프라인 시각화: pipe.draw()나 pipe.show()로 그래프 구조를 보고 연결이 올바른지 확인하세요. 자세한 내용은 Pipeline Visualization 문서를 참고.
브레이크포인트 사용: 특정 컴포넌트와 방문 횟수에 Breakpoint를 설정해서 그 반복 시점의 상태를 검사하세요. Pipeline Breakpoints 문서 참고.
막힌 파이프라인 확인: PipelineComponentsBlockedError가 보이면 어떤 컴포넌트도 실행할 수 없다는 뜻이에요. 보통 연결 누락이나 순환 의존성 때문이죠. 모든 필수 입력이 제공되는지 확인하세요.

신중한 그래프 설계, 컴포넌트별 실행 한도, 그리고 이런 디버깅 도구를 조합하면 Haystack 파이프라인에 견고한 피드백 루프를 만들 수 있어요.

더 알아보기 (Learn more)

파이프라인 루프 (Pipeline Loops) - 공식문서