mirror of
https://github.com/deepset-ai/haystack.git
synced 2025-07-19 15:01:40 +00:00
63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
|
|
#
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
from pathlib import Path
|
|
from pprint import pprint
|
|
|
|
from haystack.core.pipeline import Pipeline
|
|
from haystack.testing.sample_components import StringJoiner, StringListJoiner, Hello, TextSplitter
|
|
|
|
import logging
|
|
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
|
|
|
|
def test_joiner(tmp_path):
|
|
pipeline = Pipeline()
|
|
pipeline.add_component("hello_one", Hello())
|
|
pipeline.add_component("hello_two", Hello())
|
|
pipeline.add_component("hello_three", Hello())
|
|
pipeline.add_component("joiner", StringJoiner())
|
|
|
|
pipeline.connect("hello_one", "hello_two")
|
|
pipeline.connect("hello_two", "joiner")
|
|
pipeline.connect("hello_three", "joiner")
|
|
|
|
pipeline.draw(tmp_path / "joiner_pipeline.png")
|
|
|
|
results = pipeline.run({"hello_one": {"word": "world"}, "hello_three": {"word": "my friend"}})
|
|
assert results == {"joiner": {"output": "Hello, my friend! Hello, Hello, world!!"}}
|
|
|
|
|
|
def test_joiner_with_lists(tmp_path):
|
|
pipeline = Pipeline()
|
|
pipeline.add_component("first", TextSplitter())
|
|
pipeline.add_component("second", TextSplitter())
|
|
pipeline.add_component("joiner", StringListJoiner())
|
|
|
|
pipeline.connect("first", "joiner")
|
|
pipeline.connect("second", "joiner")
|
|
|
|
pipeline.draw(tmp_path / "joiner_list_pipeline.png")
|
|
|
|
results = pipeline.run({"first": {"sentence": "Hello world!"}, "second": {"sentence": "How are you?"}})
|
|
assert results == {"joiner": {"output": ["Hello", "world!", "How", "are", "you?"]}}
|
|
|
|
|
|
def test_joiner_with_pipeline_run(tmp_path):
|
|
pipeline = Pipeline()
|
|
pipeline.add_component("hello", Hello())
|
|
pipeline.add_component("joiner", StringJoiner())
|
|
pipeline.connect("hello", "joiner")
|
|
|
|
pipeline.draw(tmp_path / "joiner_with_pipeline_run.png")
|
|
|
|
results = pipeline.run({"hello": {"word": "world"}, "joiner": {"input_str": "another string!"}})
|
|
assert results == {"joiner": {"output": "another string! Hello, world!"}}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_joiner(Path(__file__).parent)
|
|
test_joiner_with_lists(Path(__file__).parent)
|
|
test_joiner_with_pipeline_run(Path(__file__).parent)
|