mirror of
https://github.com/deepset-ai/haystack.git
synced 2026-01-05 19:47:45 +00:00
feat: add ChatPromptBuilder, deprecate DynamicChatPromptBuilder (#7663)
This commit is contained in:
tstadel
GitHub
parent
4bc62854a9
commit
98fd270428
223
haystack/components/builders/chat_prompt_builder.py
Normal file
223
haystack/components/builders/chat_prompt_builder.py
Normal file
@ -0,0 +1,223 @@
|
||||
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
|
||||
#
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
from typing import Any, Dict, List, Optional, Set
|
||||
|
||||
from jinja2 import Template, meta
|
||||
|
||||
from haystack import component, logging
|
||||
from haystack.dataclasses.chat_message import ChatMessage, ChatRole
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@component
|
||||
class ChatPromptBuilder:
|
||||
"""
|
||||
ChatPromptBuilder is a component that renders a chat prompt from a template string using Jinja2 templates.
|
||||
|
||||
It is designed to construct prompts for the pipeline using static or dynamic templates: Users can change
|
||||
the prompt template at runtime by providing a new template for each pipeline run invocation if needed.
|
||||
|
||||
The template variables found in the init template string are used as input types for the component and are all optional,
|
||||
unless explicitly specified. If an optional template variable is not provided as an input, it will be replaced with
|
||||
an empty string in the rendered prompt. Use `variable` and `required_variables` to specify the input types and
|
||||
required variables.
|
||||
|
||||
Usage example with static prompt template:
|
||||
```python
|
||||
template = [ChatMessage.from_user("Translate to {{ target_language }}. Context: {{ snippet }}; Translation:")]
|
||||
builder = ChatPromptBuilder(template=template)
|
||||
builder.run(target_language="spanish", snippet="I can't speak spanish.")
|
||||
```
|
||||
|
||||
Usage example of overriding the static template at runtime:
|
||||
```python
|
||||
template = [ChatMessage.from_user("Translate to {{ target_language }}. Context: {{ snippet }}; Translation:")]
|
||||
builder = ChatPromptBuilder(template=template)
|
||||
builder.run(target_language="spanish", snippet="I can't speak spanish.")
|
||||
|
||||
summary_template = [ChatMessage.from_user("Translate to {{ target_language }} and summarize. Context: {{ snippet }}; Summary:")]
|
||||
builder.run(target_language="spanish", snippet="I can't speak spanish.", template=summary_template)
|
||||
```
|
||||
|
||||
Usage example with dynamic prompt template:
|
||||
```python
|
||||
from haystack.components.builders import ChatPromptBuilder
|
||||
from haystack.components.generators.chat import OpenAIChatGenerator
|
||||
from haystack.dataclasses import ChatMessage
|
||||
from haystack import Pipeline
|
||||
from haystack.utils import Secret
|
||||
|
||||
# no parameter init, we don't use any runtime template variables
|
||||
prompt_builder = ChatPromptBuilder()
|
||||
llm = OpenAIChatGenerator(api_key=Secret.from_token("<your-api-key>"), model="gpt-3.5-turbo")
|
||||
|
||||
pipe = Pipeline()
|
||||
pipe.add_component("prompt_builder", prompt_builder)
|
||||
pipe.add_component("llm", llm)
|
||||
pipe.connect("prompt_builder.prompt", "llm.messages")
|
||||
|
||||
location = "Berlin"
|
||||
language = "English"
|
||||
system_message = ChatMessage.from_system("You are an assistant giving information to tourists in {{language}}")
|
||||
messages = [system_message, ChatMessage.from_user("Tell me about {{location}}")]
|
||||
|
||||
res = pipe.run(data={"prompt_builder": {"template_variables": {"location": location, "language": language},
|
||||
"template": messages}})
|
||||
print(res)
|
||||
|
||||
>> {'llm': {'replies': [ChatMessage(content="Berlin is the capital city of Germany and one of the most vibrant
|
||||
and diverse cities in Europe. Here are some key things to know...Enjoy your time exploring the vibrant and dynamic
|
||||
capital of Germany!", role=<ChatRole.ASSISTANT: 'assistant'>, name=None, meta={'model': 'gpt-3.5-turbo-0613',
|
||||
'index': 0, 'finish_reason': 'stop', 'usage': {'prompt_tokens': 27, 'completion_tokens': 681, 'total_tokens':
|
||||
708}})]}}
|
||||
|
||||
|
||||
messages = [system_message, ChatMessage.from_user("What's the weather forecast for {{location}} in the next
|
||||
{{day_count}} days?")]
|
||||
|
||||
res = pipe.run(data={"prompt_builder": {"template_variables": {"location": location, "day_count": "5"},
|
||||
"template": messages}})
|
||||
|
||||
print(res)
|
||||
>> {'llm': {'replies': [ChatMessage(content="Here is the weather forecast for Berlin in the next 5
|
||||
days:\n\nDay 1: Mostly cloudy with a high of 22°C (72°F) and...so it's always a good idea to check for updates
|
||||
closer to your visit.", role=<ChatRole.ASSISTANT: 'assistant'>, name=None, meta={'model': 'gpt-3.5-turbo-0613',
|
||||
'index': 0, 'finish_reason': 'stop', 'usage': {'prompt_tokens': 37, 'completion_tokens': 201,
|
||||
'total_tokens': 238}})]}}
|
||||
```
|
||||
|
||||
Note how in the example above, we can dynamically change the prompt template by providing a new template to the
|
||||
run method of the pipeline.
|
||||
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
template: Optional[List[ChatMessage]] = None,
|
||||
required_variables: Optional[List[str]] = None,
|
||||
variables: Optional[List[str]] = None,
|
||||
):
|
||||
"""
|
||||
Constructs a ChatPromptBuilder component.
|
||||
|
||||
:param template:
|
||||
A list of `ChatMessage` instances. All user and system messages are treated as potentially having jinja2
|
||||
templates and are rendered with the provided template variables. If not provided, the template
|
||||
must be provided at runtime using the `template` parameter of the `run` method.
|
||||
:param required_variables: An optional list of input variables that must be provided at all times.
|
||||
If not provided, an exception will be raised.
|
||||
:param variables:
|
||||
A list of template variable names you can use in prompt construction. For example,
|
||||
if `variables` contains the string `documents`, the component will create an input called
|
||||
`documents` of type `Any`. These variable names are used to resolve variables and their values during
|
||||
pipeline execution. The values associated with variables from the pipeline runtime are then injected into
|
||||
template placeholders of a prompt text template that is provided to the `run` method.
|
||||
If not provided, variables are inferred from `template`.
|
||||
"""
|
||||
self._variables = variables
|
||||
self._required_variables = required_variables
|
||||
self.required_variables = required_variables or []
|
||||
self.template = template
|
||||
variables = variables or []
|
||||
if template and not variables:
|
||||
for message in template:
|
||||
if message.is_from(ChatRole.USER) or message.is_from(ChatRole.SYSTEM):
|
||||
# infere variables from template
|
||||
msg_template = Template(message.content)
|
||||
ast = msg_template.environment.parse(message.content)
|
||||
template_variables = meta.find_undeclared_variables(ast)
|
||||
variables += list(template_variables)
|
||||
|
||||
# setup inputs
|
||||
static_input_slots = {"template": Optional[str], "template_variables": Optional[Dict[str, Any]]}
|
||||
component.set_input_types(self, **static_input_slots)
|
||||
for var in variables:
|
||||
if var in self.required_variables:
|
||||
component.set_input_type(self, var, Any)
|
||||
else:
|
||||
component.set_input_type(self, var, Any, "")
|
||||
|
||||
@component.output_types(prompt=List[ChatMessage])
|
||||
def run(
|
||||
self,
|
||||
template: Optional[List[ChatMessage]] = None,
|
||||
template_variables: Optional[Dict[str, Any]] = None,
|
||||
**kwargs,
|
||||
):
|
||||
"""
|
||||
Executes the prompt building process.
|
||||
|
||||
It applies the template variables to render the final prompt. You can provide variables either via pipeline
|
||||
(set through `variables` or inferred from `template` at initialization) or via additional template variables
|
||||
set directly to this method. On collision, the variables provided directly to this method take precedence.
|
||||
|
||||
:param template:
|
||||
An optional list of ChatMessages to overwrite ChatPromptBuilder's default template. If None, the default template
|
||||
provided at initialization is used.
|
||||
:param template_variables:
|
||||
An optional dictionary of template variables. These are additional variables users can provide directly
|
||||
to this method in contrast to pipeline variables.
|
||||
:param kwargs:
|
||||
Pipeline variables (typically resolved from a pipeline) which are merged with the provided template variables.
|
||||
|
||||
:returns: A dictionary with the following keys:
|
||||
- `prompt`: The updated list of `ChatMessage` instances after rendering the found templates.
|
||||
:raises ValueError:
|
||||
If `chat_messages` is empty or contains elements that are not instances of `ChatMessage`.
|
||||
"""
|
||||
kwargs = kwargs or {}
|
||||
template_variables = template_variables or {}
|
||||
template_variables_combined = {**kwargs, **template_variables}
|
||||
|
||||
if template is None:
|
||||
template = self.template
|
||||
|
||||
if not template:
|
||||
raise ValueError(
|
||||
f"The {self.__class__.__name__} requires a non-empty list of ChatMessage instances. "
|
||||
f"Please provide a valid list of ChatMessage instances to render the prompt."
|
||||
)
|
||||
|
||||
if not all(isinstance(message, ChatMessage) for message in template):
|
||||
raise ValueError(
|
||||
f"The {self.__class__.__name__} expects a list containing only ChatMessage instances. "
|
||||
f"The provided list contains other types. Please ensure that all elements in the list "
|
||||
f"are ChatMessage instances."
|
||||
)
|
||||
|
||||
processed_messages = []
|
||||
for message in template:
|
||||
if message.is_from(ChatRole.USER) or message.is_from(ChatRole.SYSTEM):
|
||||
self._validate_variables(set(template_variables_combined.keys()))
|
||||
compiled_template = Template(message.content)
|
||||
rendered_content = compiled_template.render(template_variables_combined)
|
||||
rendered_message = (
|
||||
ChatMessage.from_user(rendered_content)
|
||||
if message.is_from(ChatRole.USER)
|
||||
else ChatMessage.from_system(rendered_content)
|
||||
)
|
||||
processed_messages.append(rendered_message)
|
||||
else:
|
||||
processed_messages.append(message)
|
||||
|
||||
return {"prompt": processed_messages}
|
||||
|
||||
def _validate_variables(self, provided_variables: Set[str]):
|
||||
"""
|
||||
Checks if all the required template variables are provided.
|
||||
|
||||
:param provided_variables:
|
||||
A set of provided template variables.
|
||||
:raises ValueError:
|
||||
If no template is provided or if all the required template variables are not provided.
|
||||
"""
|
||||
missing_variables = [var for var in self.required_variables if var not in provided_variables]
|
||||
if missing_variables:
|
||||
missing_vars_str = ", ".join(missing_variables)
|
||||
raise ValueError(
|
||||
f"Missing required input variables in ChatPromptBuilder: {missing_vars_str}. "
|
||||
f"Required variables: {self.required_variables}. Provided variables: {provided_variables}."
|
||||
)
|
||||
@ -2,6 +2,7 @@
|
||||
#
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
import warnings
|
||||
from typing import Any, Dict, List, Optional, Set
|
||||
|
||||
from jinja2 import Template, meta
|
||||
@ -84,6 +85,11 @@ class DynamicChatPromptBuilder:
|
||||
pipeline execution. The values associated with variables from the pipeline runtime are then injected into
|
||||
template placeholders of a ChatMessage that is provided to the `run` method.
|
||||
"""
|
||||
warnings.warn(
|
||||
"`DynamicChatPromptBuilder` is deprecated and will be removed in Haystack 2.3.0."
|
||||
"Use `ChatPromptBuilder` instead.",
|
||||
DeprecationWarning,
|
||||
)
|
||||
runtime_variables = runtime_variables or []
|
||||
|
||||
# setup inputs
|
||||
|
||||
@ -0,0 +1,7 @@
|
||||
---
|
||||
enhancements:
|
||||
- |
|
||||
`ChatPromptBuilder` now supports changing its template at runtime. This allows you to define a default template and then change it based on your needs at runtime.
|
||||
deprecations:
|
||||
- |
|
||||
`DynamicChatPromptBuilder` has been deprecated as `ChatPromptBuilder` fully covers its functionality. Use `ChatPromptBuilder` instead.
|
||||
496
test/components/builders/test_chat_prompt_builder.py
Normal file
496
test/components/builders/test_chat_prompt_builder.py
Normal file
@ -0,0 +1,496 @@
|
||||
from typing import Any, Dict, List, Optional
|
||||
from jinja2 import TemplateSyntaxError
|
||||
import pytest
|
||||
|
||||
from haystack.components.builders.chat_prompt_builder import ChatPromptBuilder
|
||||
from haystack import component
|
||||
from haystack.core.pipeline.pipeline import Pipeline
|
||||
from haystack.dataclasses.chat_message import ChatMessage
|
||||
from haystack.dataclasses.document import Document
|
||||
|
||||
|
||||
class TestChatPromptBuilder:
|
||||
def test_init(self):
|
||||
builder = ChatPromptBuilder(
|
||||
template=[
|
||||
ChatMessage.from_user(content="This is a {{ variable }}"),
|
||||
ChatMessage.from_system(content="This is a {{ variable2 }}"),
|
||||
]
|
||||
)
|
||||
assert builder.required_variables == []
|
||||
assert builder.template[0].content == "This is a {{ variable }}"
|
||||
assert builder.template[1].content == "This is a {{ variable2 }}"
|
||||
assert builder._variables is None
|
||||
assert builder._required_variables is None
|
||||
|
||||
# we have inputs that contain: template, template_variables + inferred variables
|
||||
inputs = builder.__haystack_input__._sockets_dict
|
||||
assert set(inputs.keys()) == {"template", "template_variables", "variable", "variable2"}
|
||||
assert inputs["template"].type == Optional[List[ChatMessage]]
|
||||
assert inputs["template_variables"].type == Optional[Dict[str, Any]]
|
||||
assert inputs["variable"].type == Any
|
||||
assert inputs["variable2"].type == Any
|
||||
|
||||
# response is always prompt
|
||||
outputs = builder.__haystack_output__._sockets_dict
|
||||
assert set(outputs.keys()) == {"prompt"}
|
||||
assert outputs["prompt"].type == List[ChatMessage]
|
||||
|
||||
def test_init_without_template(self):
|
||||
variables = ["var1", "var2"]
|
||||
builder = ChatPromptBuilder(variables=variables)
|
||||
assert builder.template is None
|
||||
assert builder.required_variables == []
|
||||
assert builder._variables == variables
|
||||
assert builder._required_variables is None
|
||||
|
||||
# we have inputs that contain: template, template_variables + variables
|
||||
inputs = builder.__haystack_input__._sockets_dict
|
||||
assert set(inputs.keys()) == {"template", "template_variables", "var1", "var2"}
|
||||
assert inputs["template"].type == Optional[List[ChatMessage]]
|
||||
assert inputs["template_variables"].type == Optional[Dict[str, Any]]
|
||||
assert inputs["var1"].type == Any
|
||||
assert inputs["var2"].type == Any
|
||||
|
||||
# response is always prompt
|
||||
outputs = builder.__haystack_output__._sockets_dict
|
||||
assert set(outputs.keys()) == {"prompt"}
|
||||
assert outputs["prompt"].type == List[ChatMessage]
|
||||
|
||||
def test_init_with_required_variables(self):
|
||||
builder = ChatPromptBuilder(
|
||||
template=[ChatMessage.from_user("This is a {{ variable }}")], required_variables=["variable"]
|
||||
)
|
||||
assert builder.required_variables == ["variable"]
|
||||
assert builder.template[0].content == "This is a {{ variable }}"
|
||||
assert builder._variables is None
|
||||
assert builder._required_variables == ["variable"]
|
||||
|
||||
# we have inputs that contain: template, template_variables + inferred variables
|
||||
inputs = builder.__haystack_input__._sockets_dict
|
||||
assert set(inputs.keys()) == {"template", "template_variables", "variable"}
|
||||
assert inputs["template"].type == Optional[List[ChatMessage]]
|
||||
assert inputs["template_variables"].type == Optional[Dict[str, Any]]
|
||||
assert inputs["variable"].type == Any
|
||||
|
||||
# response is always prompt
|
||||
outputs = builder.__haystack_output__._sockets_dict
|
||||
assert set(outputs.keys()) == {"prompt"}
|
||||
assert outputs["prompt"].type == List[ChatMessage]
|
||||
|
||||
def test_init_with_custom_variables(self):
|
||||
variables = ["var1", "var2", "var3"]
|
||||
template = [ChatMessage.from_user("Hello, {{ var1 }}, {{ var2 }}!")]
|
||||
builder = ChatPromptBuilder(template=template, variables=variables)
|
||||
assert builder.required_variables == []
|
||||
assert builder._variables == variables
|
||||
assert builder.template[0].content == "Hello, {{ var1 }}, {{ var2 }}!"
|
||||
assert builder._required_variables is None
|
||||
|
||||
# we have inputs that contain: template, template_variables + variables
|
||||
inputs = builder.__haystack_input__._sockets_dict
|
||||
assert set(inputs.keys()) == {"template", "template_variables", "var1", "var2", "var3"}
|
||||
assert inputs["template"].type == Optional[List[ChatMessage]]
|
||||
assert inputs["template_variables"].type == Optional[Dict[str, Any]]
|
||||
assert inputs["var1"].type == Any
|
||||
assert inputs["var2"].type == Any
|
||||
assert inputs["var3"].type == Any
|
||||
|
||||
# response is always prompt
|
||||
outputs = builder.__haystack_output__._sockets_dict
|
||||
assert set(outputs.keys()) == {"prompt"}
|
||||
assert outputs["prompt"].type == List[ChatMessage]
|
||||
|
||||
def test_run(self):
|
||||
builder = ChatPromptBuilder(template=[ChatMessage.from_user("This is a {{ variable }}")])
|
||||
res = builder.run(variable="test")
|
||||
assert res == {"prompt": [ChatMessage.from_user("This is a test")]}
|
||||
|
||||
def test_run_template_variable(self):
|
||||
builder = ChatPromptBuilder(template=[ChatMessage.from_user("This is a {{ variable }}")])
|
||||
res = builder.run(template_variables={"variable": "test"})
|
||||
assert res == {"prompt": [ChatMessage.from_user("This is a test")]}
|
||||
|
||||
def test_run_template_variable_overrides_variable(self):
|
||||
builder = ChatPromptBuilder(template=[ChatMessage.from_user("This is a {{ variable }}")])
|
||||
res = builder.run(template_variables={"variable": "test_from_template_var"}, variable="test")
|
||||
assert res == {"prompt": [ChatMessage.from_user("This is a test_from_template_var")]}
|
||||
|
||||
def test_run_without_input(self):
|
||||
builder = ChatPromptBuilder(template=[ChatMessage.from_user("This is a template without input")])
|
||||
res = builder.run()
|
||||
assert res == {"prompt": [ChatMessage.from_user("This is a template without input")]}
|
||||
|
||||
def test_run_with_missing_input(self):
|
||||
builder = ChatPromptBuilder(template=[ChatMessage.from_user("This is a {{ variable }}")])
|
||||
res = builder.run()
|
||||
assert res == {"prompt": [ChatMessage.from_user("This is a ")]}
|
||||
|
||||
def test_run_with_missing_required_input(self):
|
||||
builder = ChatPromptBuilder(
|
||||
template=[ChatMessage.from_user("This is a {{ foo }}, not a {{ bar }}")], required_variables=["foo", "bar"]
|
||||
)
|
||||
with pytest.raises(ValueError, match="foo"):
|
||||
builder.run(bar="bar")
|
||||
with pytest.raises(ValueError, match="bar"):
|
||||
builder.run(foo="foo")
|
||||
with pytest.raises(ValueError, match="foo, bar"):
|
||||
builder.run()
|
||||
|
||||
def test_run_with_variables(self):
|
||||
variables = ["var1", "var2", "var3"]
|
||||
template = [ChatMessage.from_user("Hello, {{ name }}! {{ var1 }}")]
|
||||
|
||||
builder = ChatPromptBuilder(template=template, variables=variables)
|
||||
|
||||
template_variables = {"name": "John"}
|
||||
expected_result = {"prompt": [ChatMessage.from_user("Hello, John! How are you?")]}
|
||||
|
||||
assert builder.run(template_variables=template_variables, var1="How are you?") == expected_result
|
||||
|
||||
def test_run_with_variables_and_runtime_template(self):
|
||||
variables = ["var1", "var2", "var3"]
|
||||
|
||||
builder = ChatPromptBuilder(variables=variables)
|
||||
|
||||
template = [ChatMessage.from_user("Hello, {{ name }}! {{ var1 }}")]
|
||||
template_variables = {"name": "John"}
|
||||
expected_result = {"prompt": [ChatMessage.from_user("Hello, John! How are you?")]}
|
||||
|
||||
assert (
|
||||
builder.run(template=template, template_variables=template_variables, var1="How are you?")
|
||||
== expected_result
|
||||
)
|
||||
|
||||
def test_run_overwriting_default_template(self):
|
||||
default_template = [ChatMessage.from_user("Hello, {{ name }}!")]
|
||||
|
||||
builder = ChatPromptBuilder(template=default_template)
|
||||
|
||||
template = [ChatMessage.from_user("Hello, {{ var1 }}{{ name }}!")]
|
||||
expected_result = {"prompt": [ChatMessage.from_user("Hello, John!")]}
|
||||
|
||||
assert builder.run(template, name="John") == expected_result
|
||||
|
||||
def test_run_overwriting_default_template_with_template_variables(self):
|
||||
default_template = [ChatMessage.from_user("Hello, {{ name }}!")]
|
||||
|
||||
builder = ChatPromptBuilder(template=default_template)
|
||||
|
||||
template = [ChatMessage.from_user("Hello, {{ var1 }} {{ name }}!")]
|
||||
template_variables = {"var1": "Big"}
|
||||
expected_result = {"prompt": [ChatMessage.from_user("Hello, Big John!")]}
|
||||
|
||||
assert builder.run(template, template_variables, name="John") == expected_result
|
||||
|
||||
def test_run_overwriting_default_template_with_variables(self):
|
||||
variables = ["var1", "var2", "name"]
|
||||
default_template = [ChatMessage.from_user("Hello, {{ name }}!")]
|
||||
|
||||
builder = ChatPromptBuilder(template=default_template, variables=variables)
|
||||
|
||||
template = [ChatMessage.from_user("Hello, {{ var1 }} {{ name }}!")]
|
||||
expected_result = {"prompt": [ChatMessage.from_user("Hello, Big John!")]}
|
||||
|
||||
assert builder.run(template, name="John", var1="Big") == expected_result
|
||||
|
||||
def test_run_with_invalid_template(self):
|
||||
builder = ChatPromptBuilder()
|
||||
|
||||
template = [ChatMessage.from_user("Hello, {{ name }!")]
|
||||
template_variables = {"name": "John"}
|
||||
with pytest.raises(TemplateSyntaxError):
|
||||
builder.run(template, template_variables)
|
||||
|
||||
def test_init_with_invalid_template(self):
|
||||
template = [ChatMessage.from_user("Hello, {{ name }!")]
|
||||
with pytest.raises(TemplateSyntaxError):
|
||||
ChatPromptBuilder(template)
|
||||
|
||||
def test_run_without_template(self):
|
||||
prompt_builder = ChatPromptBuilder()
|
||||
with pytest.raises(
|
||||
ValueError, match="The ChatPromptBuilder requires a non-empty list of ChatMessage instances"
|
||||
):
|
||||
prompt_builder.run()
|
||||
|
||||
def test_run_with_empty_chat_message_list(self):
|
||||
prompt_builder = ChatPromptBuilder(template=[], variables=["documents"])
|
||||
with pytest.raises(
|
||||
ValueError, match="The ChatPromptBuilder requires a non-empty list of ChatMessage instances"
|
||||
):
|
||||
prompt_builder.run()
|
||||
|
||||
def test_chat_message_list_with_mixed_object_list(self):
|
||||
prompt_builder = ChatPromptBuilder(
|
||||
template=[ChatMessage.from_user("Hello"), "there world"], variables=["documents"]
|
||||
)
|
||||
with pytest.raises(
|
||||
ValueError, match="The ChatPromptBuilder expects a list containing only ChatMessage instances"
|
||||
):
|
||||
prompt_builder.run()
|
||||
|
||||
def test_provided_template_variables(self):
|
||||
prompt_builder = ChatPromptBuilder(variables=["documents"], required_variables=["city"])
|
||||
|
||||
# both variables are provided
|
||||
prompt_builder._validate_variables({"name", "city"})
|
||||
|
||||
# provided variables are a superset of the required variables
|
||||
prompt_builder._validate_variables({"name", "city", "age"})
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
prompt_builder._validate_variables({"name"})
|
||||
|
||||
def test_example_in_pipeline(self):
|
||||
default_template = [
|
||||
ChatMessage.from_user("Here is the document: {{documents[0].content}} \\n Answer: {{query}}")
|
||||
]
|
||||
prompt_builder = ChatPromptBuilder(template=default_template, variables=["documents"])
|
||||
|
||||
@component
|
||||
class DocumentProducer:
|
||||
@component.output_types(documents=List[Document])
|
||||
def run(self, doc_input: str):
|
||||
return {"documents": [Document(content=doc_input)]}
|
||||
|
||||
pipe = Pipeline()
|
||||
pipe.add_component("doc_producer", DocumentProducer())
|
||||
pipe.add_component("prompt_builder", prompt_builder)
|
||||
pipe.connect("doc_producer.documents", "prompt_builder.documents")
|
||||
|
||||
template = [ChatMessage.from_user("Here is the document: {{documents[0].content}} \n Query: {{query}}")]
|
||||
result = pipe.run(
|
||||
data={
|
||||
"doc_producer": {"doc_input": "Hello world, I live in Berlin"},
|
||||
"prompt_builder": {
|
||||
"template": template,
|
||||
"template_variables": {"query": "Where does the speaker live?"},
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
assert result == {
|
||||
"prompt_builder": {
|
||||
"prompt": [
|
||||
ChatMessage.from_user(
|
||||
"Here is the document: Hello world, I live in Berlin \n Query: Where does the speaker live?"
|
||||
)
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
def test_example_in_pipeline_simple(self):
|
||||
default_template = [ChatMessage.from_user("This is the default prompt:\n Query: {{query}}")]
|
||||
prompt_builder = ChatPromptBuilder(template=default_template)
|
||||
|
||||
pipe = Pipeline()
|
||||
pipe.add_component("prompt_builder", prompt_builder)
|
||||
|
||||
# using the default prompt
|
||||
result = pipe.run(data={"query": "Where does the speaker live?"})
|
||||
expected_default = {
|
||||
"prompt_builder": {
|
||||
"prompt": [ChatMessage.from_user("This is the default prompt:\n Query: Where does the speaker live?")]
|
||||
}
|
||||
}
|
||||
assert result == expected_default
|
||||
|
||||
# using the dynamic prompt
|
||||
result = pipe.run(
|
||||
data={
|
||||
"query": "Where does the speaker live?",
|
||||
"template": [ChatMessage.from_user("This is the dynamic prompt:\n Query: {{query}}")],
|
||||
}
|
||||
)
|
||||
expected_dynamic = {
|
||||
"prompt_builder": {
|
||||
"prompt": [ChatMessage.from_user("This is the dynamic prompt:\n Query: Where does the speaker live?")]
|
||||
}
|
||||
}
|
||||
assert result == expected_dynamic
|
||||
|
||||
|
||||
class TestChatPromptBuilderDynamic:
|
||||
def test_multiple_templated_chat_messages(self):
|
||||
prompt_builder = ChatPromptBuilder()
|
||||
language = "French"
|
||||
location = "Berlin"
|
||||
messages = [
|
||||
ChatMessage.from_system("Write your response in this language:{{language}}"),
|
||||
ChatMessage.from_user("Tell me about {{location}}"),
|
||||
]
|
||||
|
||||
result = prompt_builder.run(template_variables={"language": language, "location": location}, template=messages)
|
||||
assert result["prompt"] == [
|
||||
ChatMessage.from_system("Write your response in this language:French"),
|
||||
ChatMessage.from_user("Tell me about Berlin"),
|
||||
], "The templated messages should match the expected output."
|
||||
|
||||
def test_multiple_templated_chat_messages_in_place(self):
|
||||
prompt_builder = ChatPromptBuilder()
|
||||
language = "French"
|
||||
location = "Berlin"
|
||||
messages = [
|
||||
ChatMessage.from_system("Write your response ins this language:{{language}}"),
|
||||
ChatMessage.from_user("Tell me about {{location}}"),
|
||||
]
|
||||
|
||||
res = prompt_builder.run(template_variables={"language": language, "location": location}, template=messages)
|
||||
assert res == {
|
||||
"prompt": [
|
||||
ChatMessage.from_system("Write your response ins this language:French"),
|
||||
ChatMessage.from_user("Tell me about Berlin"),
|
||||
]
|
||||
}, "The templated messages should match the expected output."
|
||||
|
||||
def test_some_templated_chat_messages(self):
|
||||
prompt_builder = ChatPromptBuilder()
|
||||
language = "English"
|
||||
location = "Paris"
|
||||
messages = [
|
||||
ChatMessage.from_system("Please, respond in the following language: {{language}}."),
|
||||
ChatMessage.from_user("I would like to learn more about {{location}}."),
|
||||
ChatMessage.from_assistant("Yes, I can help you with that {{subject}}"),
|
||||
ChatMessage.from_user("Ok so do so please, be elaborate."),
|
||||
]
|
||||
|
||||
result = prompt_builder.run(template_variables={"language": language, "location": location}, template=messages)
|
||||
|
||||
expected_messages = [
|
||||
ChatMessage.from_system("Please, respond in the following language: English."),
|
||||
ChatMessage.from_user("I would like to learn more about Paris."),
|
||||
ChatMessage.from_assistant(
|
||||
"Yes, I can help you with that {{subject}}"
|
||||
), # assistant message should not be templated
|
||||
ChatMessage.from_user("Ok so do so please, be elaborate."),
|
||||
]
|
||||
|
||||
assert result["prompt"] == expected_messages, "The templated messages should match the expected output."
|
||||
|
||||
def test_example_in_pipeline(self):
|
||||
prompt_builder = ChatPromptBuilder()
|
||||
|
||||
pipe = Pipeline()
|
||||
pipe.add_component("prompt_builder", prompt_builder)
|
||||
|
||||
location = "Berlin"
|
||||
system_message = ChatMessage.from_system(
|
||||
"You are a helpful assistant giving out valuable information to tourists."
|
||||
)
|
||||
messages = [system_message, ChatMessage.from_user("Tell me about {{location}}")]
|
||||
|
||||
res = pipe.run(data={"prompt_builder": {"template_variables": {"location": location}, "template": messages}})
|
||||
assert res == {
|
||||
"prompt_builder": {
|
||||
"prompt": [
|
||||
ChatMessage.from_system("You are a helpful assistant giving out valuable information to tourists."),
|
||||
ChatMessage.from_user("Tell me about Berlin"),
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
messages = [
|
||||
system_message,
|
||||
ChatMessage.from_user("What's the weather forecast for {{location}} in the next {{day_count}} days?"),
|
||||
]
|
||||
|
||||
res = pipe.run(
|
||||
data={
|
||||
"prompt_builder": {"template_variables": {"location": location, "day_count": "5"}, "template": messages}
|
||||
}
|
||||
)
|
||||
assert res == {
|
||||
"prompt_builder": {
|
||||
"prompt": [
|
||||
ChatMessage.from_system("You are a helpful assistant giving out valuable information to tourists."),
|
||||
ChatMessage.from_user("What's the weather forecast for Berlin in the next 5 days?"),
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
def test_example_in_pipeline_with_multiple_templated_messages(self):
|
||||
# no parameter init, we don't use any runtime template variables
|
||||
prompt_builder = ChatPromptBuilder()
|
||||
|
||||
pipe = Pipeline()
|
||||
pipe.add_component("prompt_builder", prompt_builder)
|
||||
|
||||
location = "Berlin"
|
||||
system_message = ChatMessage.from_system(
|
||||
"You are a helpful assistant giving out valuable information to tourists in {{language}}."
|
||||
)
|
||||
messages = [system_message, ChatMessage.from_user("Tell me about {{location}}")]
|
||||
|
||||
res = pipe.run(
|
||||
data={
|
||||
"prompt_builder": {
|
||||
"template_variables": {"location": location, "language": "German"},
|
||||
"template": messages,
|
||||
}
|
||||
}
|
||||
)
|
||||
assert res == {
|
||||
"prompt_builder": {
|
||||
"prompt": [
|
||||
ChatMessage.from_system(
|
||||
"You are a helpful assistant giving out valuable information to tourists in German."
|
||||
),
|
||||
ChatMessage.from_user("Tell me about Berlin"),
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
messages = [
|
||||
system_message,
|
||||
ChatMessage.from_user("What's the weather forecast for {{location}} in the next {{day_count}} days?"),
|
||||
]
|
||||
|
||||
res = pipe.run(
|
||||
data={
|
||||
"prompt_builder": {
|
||||
"template_variables": {"location": location, "day_count": "5", "language": "English"},
|
||||
"template": messages,
|
||||
}
|
||||
}
|
||||
)
|
||||
assert res == {
|
||||
"prompt_builder": {
|
||||
"prompt": [
|
||||
ChatMessage.from_system(
|
||||
"You are a helpful assistant giving out valuable information to tourists in English."
|
||||
),
|
||||
ChatMessage.from_user("What's the weather forecast for Berlin in the next 5 days?"),
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
def test_pipeline_complex(self):
|
||||
@component
|
||||
class ValueProducer:
|
||||
def __init__(self, value_to_produce: str):
|
||||
self.value_to_produce = value_to_produce
|
||||
|
||||
@component.output_types(value_output=str)
|
||||
def run(self):
|
||||
return {"value_output": self.value_to_produce}
|
||||
|
||||
pipe = Pipeline()
|
||||
pipe.add_component("prompt_builder", ChatPromptBuilder(variables=["value_output"]))
|
||||
pipe.add_component("value_producer", ValueProducer(value_to_produce="Berlin"))
|
||||
pipe.connect("value_producer.value_output", "prompt_builder")
|
||||
|
||||
messages = [
|
||||
ChatMessage.from_system("You give valuable information to tourists."),
|
||||
ChatMessage.from_user("Tell me about {{value_output}}"),
|
||||
]
|
||||
|
||||
res = pipe.run(data={"template": messages})
|
||||
assert res == {
|
||||
"prompt_builder": {
|
||||
"prompt": [
|
||||
ChatMessage.from_system("You give valuable information to tourists."),
|
||||
ChatMessage.from_user("Tell me about Berlin"),
|
||||
]
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user