FallbackChatGenerator

여러 Chat Generator를 순차적으로 시도해서 하나가 성공할 때까지 돌리는 ChatGenerator 래퍼예요.

출처: 문서

본문

파이프라인에서의 일반적인 위치: ChatPromptBuilder 뒤에 사용해요.
필수 초기화 변수: chat_generators(순서대로 시도할 Chat Generator 컴포넌트의 비어 있지 않은 리스트)
필수 실행 변수: messages(대화를 나타내는 ChatMessage 객체 리스트 또는 일반 문자열)
출력 변수: replies(첫 번째로 성공한 Generator가 생성한 ChatMessage 인스턴스), meta(성공한 Generator 정보를 포함한 실행 메타데이터)

개요 (Overview)

FallbackChatGenerator는 여러 Chat Generator를 순차적으로 시도해서 하나가 성공할 때까지 돌리는 래퍼 컴포넌트예요. Generator가 실패하면 목록의 다음 것을 시도해요. 이렇게 하면 제공자 중단, 요금 제한, 기타 일시적 실패를 처리할 수 있어요.

이 컴포넌트는 모든 파라미터를 내부 Chat Generator에 전달하고 첫 번째 성공 결과를 반환해요. Generator가 예외를 발생시키면 다음 Generator를 시도해요. 여기에는 타임아웃 오류, 요금 제한 오류(429), 인증 오류(401), 컨텍스트 길이 오류(400), 서버 오류(500 이상) 및 기타 모든 예외가 포함돼요.

이 컴포넌트는 어떤 Generator가 성공했는지, 몇 번 시도했는지, 어떤 Generator가 실패했는지 등 실행 메타데이터를 반환해요. 모든 파라미터(messages, generation_kwargs, tools, streaming_callback)는 내부 Generator에 전달돼요. messages에 문자열을 전달하면, 전달 전에 user 역할을 가진 단일 ChatMessage를 담은 리스트로 변환돼요.

타임아웃 강제는 내부 Chat Generator에 위임돼요. 레이턴시를 제어하려면 Chat Generator를 timeout 파라미터로 구성하세요. OpenAI, Anthropic, Cohere 같은 Chat Generator는 초과 시 예외를 발생시키는 타임아웃 파라미터를 지원해요.

모니터링과 텔레메트리 (Monitoring and Telemetry)

출력의 meta 딕셔너리에는 모니터링에 유용한 정보가 들어 있어요.

from haystack.components.generators.chat import (
    FallbackChatGenerator,
    OpenAIChatGenerator,
)
from haystack.dataclasses import ChatMessage

# Set up generators
primary = OpenAIChatGenerator(model="gpt-4o")
backup = OpenAIChatGenerator(model="gpt-4o-mini")
generator = FallbackChatGenerator(chat_generators=[primary, backup])

# Run and inspect metadata
result = generator.run(messages=[ChatMessage.from_user("Hello")])

meta = result["meta"]
print(
    f"Successful generator index: {meta['successful_chat_generator_index']}",
)  # 0 for first, 1 for second, etc.
print(
    f"Successful generator class: {meta['successful_chat_generator_class']}",
)  # e.g., "OpenAIChatGenerator"
print(
    f"Total attempts made: {meta['total_attempts']}",
)  # How many Generators were tried
print(
    f"Failed generators: {meta['failed_chat_generators']}",
)  # List of failed Generator names

이 메타데이터를 사용해서 다음을 할 수 있어요.

가장 자주 사용되는 Generator 추적
각 Generator의 실패율 모니터링
폴백(fallback) 발생 시 알림 설정
성공률에 기반한 Generator 순서 조정

스트리밍 (Streaming)

FallbackChatGenerator는 streaming_callback 파라미터를 통해 스트리밍을 지원해요. 이 콜백은 내부 Generator에 직접 전달돼요.

사용법 (Usage)

단독 사용 (On its own)

주 모델에서 백업 모델로 폴백하는 기본 사용법이에요.

from haystack.components.generators.chat import (
    FallbackChatGenerator,
    OpenAIChatGenerator,
)
from haystack.dataclasses import ChatMessage

# Create primary and backup generators
primary = OpenAIChatGenerator(model="gpt-4o", timeout=30)
backup = OpenAIChatGenerator(model="gpt-4o-mini", timeout=30)

# Wrap them in a FallbackChatGenerator
generator = FallbackChatGenerator(chat_generators=[primary, backup])

# Use it like any other Chat Generator
messages = [ChatMessage.from_user("What's Natural Language Processing? Be brief.")]
result = generator.run(messages=messages)

print(result["replies"][0].text)
print(f"Successful generator: {result['meta']['successful_chat_generator_class']}")
print(f"Total attempts: {result['meta']['total_attempts']}")

# Natural Language Processing (NLP) is a field of artificial intelligence that
# focuses on the interaction between computers and humans through natural language...
# Successful generator: OpenAIChatGenerator
# Total attempts: 1

여러 제공자와 함께 사용하는 경우:

from haystack.components.generators.chat import (
    FallbackChatGenerator,
    OpenAIChatGenerator,
    AzureOpenAIChatGenerator,
)
from haystack.dataclasses import ChatMessage
from haystack.utils import Secret

# Create generators from different providers
openai_gen = OpenAIChatGenerator(
    model="gpt-4o-mini",
    api_key=Secret.from_env_var("OPENAI_API_KEY"),
    timeout=30,
)

azure_gen = AzureOpenAIChatGenerator(
    azure_endpoint="<Your Azure endpoint>",
    api_key=Secret.from_env_var("AZURE_OPENAI_API_KEY"),
    azure_deployment="gpt-4o-mini",
    timeout=30,
)

# Fallback will try OpenAI first, then Azure
generator = FallbackChatGenerator(chat_generators=[openai_gen, azure_gen])

messages = [ChatMessage.from_user("Explain quantum computing briefly.")]
result = generator.run(messages=messages)

print(result["replies"][0].text)

스트리밍과 함께 사용하는 경우:

from haystack.components.generators.chat import (
    FallbackChatGenerator,
    OpenAIChatGenerator,
)
from haystack.dataclasses import ChatMessage

primary = OpenAIChatGenerator(model="gpt-4o")
backup = OpenAIChatGenerator(model="gpt-4o-mini")

generator = FallbackChatGenerator(chat_generators=[primary, backup])

messages = [ChatMessage.from_user("What's Natural Language Processing? Be brief.")]
result = generator.run(
    messages=messages,
    streaming_callback=lambda chunk: print(chunk.content, end="", flush=True),
)

print("\n", result["meta"])

파이프라인에서 사용 (In a Pipeline)

from haystack import Pipeline
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import (
    FallbackChatGenerator,
    OpenAIChatGenerator,
)
from haystack.dataclasses import ChatMessage

# Create primary and backup generators with timeouts
primary = OpenAIChatGenerator(model="gpt-4o", timeout=30)
backup = OpenAIChatGenerator(model="gpt-4o-mini", timeout=30)

# Wrap in fallback
fallback_generator = FallbackChatGenerator(chat_generators=[primary, backup])

# Build pipeline
prompt_builder = ChatPromptBuilder()

pipe = Pipeline()
pipe.add_component("prompt_builder", prompt_builder)
pipe.add_component("llm", fallback_generator)
pipe.connect("prompt_builder.prompt", "llm.messages")

# Run pipeline
messages = [
    ChatMessage.from_system(
        "You are a helpful assistant that provides concise answers.",
    ),
    ChatMessage.from_user("Tell me about {{location}}"),
]

result = pipe.run(
    data={
        "prompt_builder": {
            "template": messages,
            "template_variables": {"location": "Paris"},
        },
    },
)

print(result["llm"]["replies"][0].text)
print(f"Generator used: {result['llm']['meta']['successful_chat_generator_class']}")

오류 처리 (Error Handling)

모든 Generator가 실패하면 FallbackChatGenerator는 어떤 Generator가 실패했고 마지막에 어떤 오류가 발생했는지에 대한 세부 정보와 함께 RuntimeError를 발생시켜요.

from haystack.components.generators.chat import (
    FallbackChatGenerator,
    OpenAIChatGenerator,
)
from haystack.dataclasses import ChatMessage
from haystack.utils import Secret

# Create generators with invalid credentials to demonstrate error handling
primary = OpenAIChatGenerator(api_key=Secret.from_token("invalid-key-1"))
backup = OpenAIChatGenerator(api_key=Secret.from_token("invalid-key-2"))

generator = FallbackChatGenerator(chat_generators=[primary, backup])

try:
    result = generator.run(messages=[ChatMessage.from_user("Hello")])
except RuntimeError as e:
    print(f"All generators failed: {e}")
    # >> All 2 chat generators failed. Last error: ... Failed chat generators: [OpenAIChatGenerator, OpenAIChatGenerator]

FallbackChatGenerator