PipelineTool

PipelineTool

Haystack 파이프라인을 감싸서 LLM이 툴로 호출할 수 있게 해주는 도구예요. 파이프라인 전체를 하나의 툴로 노출하고 싶을 때 사용하죠.

출처: 문서

본문

항목 내용
필수 init 변수 pipeline: 감쌀 Haystack 파이프라인. name: 툴의 이름. description: 툴의 설명
API reference PipelineTool
GitHub 링크 pipeline_tool.py
패키지 이름 haystack-ai

개요

PipelineTool은 Haystack 파이프라인 전체를 감싸 LLM이 호출할 수 있는 툴로 노출시켜 줘요. 파이프라인을 먼저 SuperComponent로 감싼 다음 ComponentTool에 전달하던 예전 방식의 워크플로를 대체하죠.

PipelineTool은 파이프라인의 입력 소켓에서 툴 파라미터 스키마를 만들고, 입력 설명에는 내부 컴포넌트들의 docstring을 사용해요. input_mapping과 output_mapping으로 노출할 파이프라인 입력·출력을 선택할 수 있죠. Agent 컴포넌트와 함께 사용할 수 있는데, 직접 사용하거나 파이프라인 안에서 사용해요.

PipelineTool은 비동기 호출도 지원해요. 모든 Pipeline이 네이티브 run_async를 노출하기 때문에, 툴을 추가 설정 없이 대기(await)할 수 있어요 (예: Agent.run_async에서). 자세한 내용은 Async Tools를 참고하세요.

파라미터

  • pipeline은 필수이며 Pipeline 인스턴스여야 해요.
  • name은 필수이며 툴 이름을 지정해요.
  • description은 필수이며 툴이 무엇을 하는지 설명해요.
  • input_mapping은 선택적이에요. 툴 입력 이름을 파이프라인 입력 소켓 경로에 매핑해요. 생략하면 모든 파이프라인 입력에서 기본 매핑이 생성돼요.
  • output_mapping은 선택적이에요. 파이프라인 출력 소켓 경로를 툴 출력 이름에 매핑해요. 생략하면 모든 파이프라인 출력에서 기본 매핑이 생성돼요.
  • parameters는 선택적이며, 툴 입력에 대해 자동 생성된 JSON 스키마를 재정의해요.
  • outputs_to_string은 선택적이며, 파이프라인 출력을 LLM용 문자열로 어떻게 변환할지 제어해요. 기본적으로 전체 결과 딕셔너리가 직렬화돼요. {"source": "key"}로 단일 출력 키를 추출하거나, "handler"를 추가해 커스텀 포매터를 적용할 수 있어요.
  • inputs_from_state는 선택적이며, 에이전트 상태 키를 파이프라인 입력 파라미터에 매핑해요. 예: {"repository": "repo"}는 "repository"의 상태 값을 파이프라인의 "repo" 입력으로 전달해요.
  • outputs_to_state는 선택적이며, 파이프라인 출력 키를 에이전트 상태 키에 매핑해요. 예: {"documents": {"source": "docs"}}는 파이프라인의 "docs" 출력을 상태의 "documents"에 기록해요.

사용법

팁: Haystack에서 PipelineTool을 사용하는 권장 방식은 Agent 컴포넌트와 함께 쓰는 거예요. Agent가 툴 호출 루프를 관리해 주거든요. 아래 예제처럼 Agent를 단독 실행하거나 파이프라인에 추가할 수 있어요.

기본 사용법

기존 Haystack 파이프라인 어디서든 PipelineTool을 만들 수 있어요.

이 페이지의 예제는 sentence-transformers-haystack 패키지의 Sentence Transformers 컴포넌트(ranker와 embedder)를 사용해요. 예제를 실행하려면 설치하세요:

pip install sentence-transformers-haystack

from haystack import Document, Pipeline
from haystack.tools import PipelineTool
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack_integrations.components.rankers.sentence_transformers import (
    SentenceTransformersSimilarityRanker,
)
from haystack.document_stores.in_memory import InMemoryDocumentStore

# Create your pipeline
document_store = InMemoryDocumentStore()
# Add some example documents
document_store.write_documents(
    [
        Document(
            content="Nikola Tesla was a Serbian-American inventor and electrical engineer.",
        ),
        Document(
            content="Alternating current (AC) is an electric current which periodically reverses direction.",
        ),
        Document(
            content="Thomas Edison promoted direct current (DC) and competed with AC in the War of Currents.",
        ),
    ],
)

retrieval_pipeline = Pipeline()
retrieval_pipeline.add_component(
    "bm25_retriever",
    InMemoryBM25Retriever(document_store=document_store),
)
retrieval_pipeline.add_component(
    "ranker",
    SentenceTransformersSimilarityRanker(model="cross-encoder/ms-marco-MiniLM-L-6-v2"),
)
retrieval_pipeline.connect("bm25_retriever.documents", "ranker.documents")

# Wrap the pipeline as a tool
retrieval_tool = PipelineTool(
    pipeline=retrieval_pipeline,
    input_mapping={"query": ["bm25_retriever.query", "ranker.query"]},
    output_mapping={"ranker.documents": "documents"},
    name="retrieval_tool",
    description="Search short articles about Nikola Tesla, AC electricity, and related inventors",
)
print(retrieval_tool)

Agent 컴포넌트와 함께 사용하기

from haystack import Document, Pipeline
from haystack.tools import PipelineTool
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack_integrations.components.embedders.sentence_transformers import (
    SentenceTransformersTextEmbedder,
    SentenceTransformersDocumentEmbedder,
)
from haystack.components.retrievers import InMemoryEmbeddingRetriever
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.agents import Agent
from haystack.dataclasses import ChatMessage

# Initialize a document store and add some documents
document_store = InMemoryDocumentStore()
document_embedder = SentenceTransformersDocumentEmbedder(
    model="sentence-transformers/all-MiniLM-L6-v2",
)
documents = [
    Document(
        content="Nikola Tesla was a Serbian-American inventor and electrical engineer.",
    ),
    Document(
        content="He is best known for his contributions to the design of the modern alternating current (AC) electricity supply system.",
    ),
]
docs_with_embeddings = document_embedder.run(documents=documents)["documents"]
document_store.write_documents(docs_with_embeddings)

# Build a simple retrieval pipeline
retrieval_pipeline = Pipeline()
retrieval_pipeline.add_component(
    "embedder",
    SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"),
)
retrieval_pipeline.add_component(
    "retriever",
    InMemoryEmbeddingRetriever(document_store=document_store),
)
retrieval_pipeline.connect("embedder.embedding", "retriever.query_embedding")

# Wrap the pipeline as a tool
retriever_tool = PipelineTool(
    pipeline=retrieval_pipeline,
    input_mapping={"query": ["embedder.text"]},
    output_mapping={"retriever.documents": "documents"},
    name="document_retriever",
    description="For any questions about Nikola Tesla, always use this tool",
)

agent = Agent(
    system_prompt="You are an assistant that can use a retrieval tool to find information about Nikola Tesla.",
    chat_generator=OpenAIChatGenerator(model="gpt-5.4-nano"),
    tools=[retriever_tool],
)

result = agent.run([ChatMessage.from_user("Who was Nikola Tesla?")])
print("Answer:")
print(result["messages"][-1].text)

파이프라인에서 사용하기

PipelineTool을 Agent 컴포넌트에 전달해 파이프라인에서도 사용할 수 있어요.

from haystack import Document, Pipeline
from haystack.tools import PipelineTool
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack_integrations.components.embedders.sentence_transformers import (
    SentenceTransformersTextEmbedder,
    SentenceTransformersDocumentEmbedder,
)
from haystack.components.retrievers import InMemoryEmbeddingRetriever
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.agents import Agent
from haystack.dataclasses import ChatMessage

# Initialize a document store and add some documents
document_store = InMemoryDocumentStore()
document_embedder = SentenceTransformersDocumentEmbedder(
    model="sentence-transformers/all-MiniLM-L6-v2",
)
documents = [
    Document(
        content="Nikola Tesla was a Serbian-American inventor and electrical engineer.",
    ),
    Document(
        content="He is best known for his contributions to the design of the modern alternating current (AC) electricity supply system.",
    ),
]
docs_with_embeddings = document_embedder.run(documents=documents)["documents"]
document_store.write_documents(docs_with_embeddings)

# Build a simple retrieval pipeline
retrieval_pipeline = Pipeline()
retrieval_pipeline.add_component(
    "embedder",
    SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"),
)
retrieval_pipeline.add_component(
    "retriever",
    InMemoryEmbeddingRetriever(document_store=document_store),
)
retrieval_pipeline.connect("embedder.embedding", "retriever.query_embedding")

# Wrap the pipeline as a tool
retriever_tool = PipelineTool(
    pipeline=retrieval_pipeline,
    input_mapping={"query": ["embedder.text"]},
    output_mapping={"retriever.documents": "documents"},
    name="document_retriever",
    description="For any questions about Nikola Tesla, always use this tool",
)

pipeline = Pipeline()
pipeline.add_component(
    "agent",
    Agent(
        chat_generator=OpenAIChatGenerator(model="gpt-5.4-nano"),
        tools=[retriever_tool],
    ),
)
message = ChatMessage.from_user(
    "Use the document retriever tool to find information about Nikola Tesla",
)
result = pipeline.run({"agent": {"messages": [message]}})
print(result["agent"]["last_message"].text)

더 알아보기 (Learn more)