19204656740

Committed 09 Nov 2025 06:35AM UTC coverage: 91.577% (-0.001%) from 91.578%

Build # 19204656740

Build Type

Pull #9932

github

Committed by

web-flow

Commit Message

Merge b7d498881 into 6231296cd

Pull Request Pull Request #9932: fix: prompt-builder - jinja2 template set vars still shows required

Run Details

13732 of 14995 relevant lines covered (91.58%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.26

haystack/components/builders/chat_prompt_builder.py

# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0

import json
from copy import deepcopy
from typing import Any, Literal, Optional, Union

from jinja2 import meta
from jinja2.sandbox import SandboxedEnvironment

from haystack import component, default_from_dict, default_to_dict, logging
from haystack.dataclasses.chat_message import ChatMessage, ChatRole, TextContent
from haystack.lazy_imports import LazyImport
from haystack.utils import Jinja2TimeExtension, JinjaTemplateVariableExtractor
from haystack.utils.jinja2_chat_extension import ChatMessageExtension, templatize_part

logger = logging.getLogger(__name__)

with LazyImport("Run 'pip install \"arrow>=1.3.0\"'") as arrow_import:
    import arrow  # pylint: disable=unused-import

NO_TEXT_ERROR_MESSAGE = "ChatMessages from {role} role must contain text. Received ChatMessage with no text: {message}"

FILTER_NOT_ALLOWED_ERROR_MESSAGE = (
    "The templatize_part filter cannot be used with a template containing a list of"
    "ChatMessage objects. Use a string template or remove the templatize_part filter "
    "from the template."
)


@component
class ChatPromptBuilder:
    """
    Renders a chat prompt from a template using Jinja2 syntax.

    A template can be a list of `ChatMessage` objects, or a special string, as shown in the usage examples.

    It constructs prompts using static or dynamic templates, which you can update for each pipeline run.

    Template variables in the template are optional unless specified otherwise.
    If an optional variable isn't provided, it defaults to an empty string. Use `variable` and `required_variables`
    to define input types and required variables.

    ### Usage examples

    #### Static ChatMessage 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.")
    ```

    #### Overriding static ChatMessage 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.")

    msg = "Translate to {{ target_language }} and summarize. Context: {{ snippet }}; Summary:"
    summary_template = [ChatMessage.from_user(msg)]
    builder.run(target_language="spanish", snippet="I can't speak spanish.", template=summary_template)
    ```

    #### Dynamic ChatMessage 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-4o-mini")

    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(_role=<ChatRole.ASSISTANT: 'assistant'>, _content=[TextContent(text=
    "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!")], _name=None, _meta={'model': 'gpt-4o-mini',
    '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(_role=<ChatRole.ASSISTANT: 'assistant'>, _content=[TextContent(text=
    "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.")], _name=None, _meta={'model': 'gpt-4o-mini',
    'index': 0, 'finish_reason': 'stop', 'usage': {'prompt_tokens': 37, 'completion_tokens': 201,
    'total_tokens': 238}})]}}
    ```

    #### String prompt template
    ```python
    from haystack.components.builders import ChatPromptBuilder
    from haystack.dataclasses.image_content import ImageContent

    template = \"\"\"
    {% message role="system" %}
    You are a helpful assistant.
    {% endmessage %}

    {% message role="user" %}
    Hello! I am {{user_name}}. What's the difference between the following images?
    {% for image in images %}
    {{ image | templatize_part }}
    {% endfor %}
    {% endmessage %}
    \"\"\"

    images = [ImageContent.from_file_path("apple.jpg"), ImageContent.from_file_path("orange.jpg")]

    builder = ChatPromptBuilder(template=template)
    builder.run(user_name="John", images=images)
    ```
    """

    def __init__(
        self,
        template: Optional[Union[list[ChatMessage], str]] = None,
        required_variables: Optional[Union[list[str], Literal["*"]]] = None,
        variables: Optional[list[str]] = None,
    ):
        """
        Constructs a ChatPromptBuilder component.

        :param template:
            A list of `ChatMessage` objects or a string template. The component looks for Jinja2 template syntax and
            renders the prompt with the provided variables. Provide the template in either
            the `init` method` or the `run` method.
        :param required_variables:
            List variables that must be provided as input to ChatPromptBuilder.
            If a variable listed as required is not provided, an exception is raised.
            If set to "*", all variables found in the prompt are required. Optional.
        :param variables:
            List input variables to use in prompt templates instead of the ones inferred from the
            `template` parameter. For example, to use more variables during prompt engineering than the ones present
            in the default template, you can provide them here.
        """
        self._variables = variables
        self._required_variables = required_variables
        self.template = template

        self._env = SandboxedEnvironment(extensions=[ChatMessageExtension])
        self._env.filters["templatize_part"] = templatize_part
        if arrow_import.is_successful():
            self._env.add_extension(Jinja2TimeExtension)

        extracted_variables = []
        if template and not variables:

            def _extract_from_text(
                text: Optional[str], role: Optional[str] = None, is_filter_allowed: bool = False
            ) -> list[str]:
                if text is None:
                    raise ValueError(NO_TEXT_ERROR_MESSAGE.format(role=role or "unknown", message=text))
                if is_filter_allowed and "templatize_part" in text:
                    raise ValueError(FILTER_NOT_ALLOWED_ERROR_MESSAGE)

                ast = self._env.parse(text)
                template_variables = meta.find_undeclared_variables(ast)
                assigned_variables = JinjaTemplateVariableExtractor._extract_declared_variables(text, env=self._env)
                return list(template_variables - assigned_variables)

            if isinstance(template, list):
                for message in template:
                    if message.is_from(ChatRole.USER) or message.is_from(ChatRole.SYSTEM):
                        extracted_variables += _extract_from_text(
                            message.text, role=message.role.value, is_filter_allowed=True
                        )
            elif isinstance(template, str):
                extracted_variables = _extract_from_text(template, is_filter_allowed=False)

        extracted_variables = extracted_variables or []
        self.variables = variables or extracted_variables
        self.required_variables = required_variables or []

        if len(self.variables) > 0 and required_variables is None:
            logger.warning(
                "ChatPromptBuilder has {length} prompt variables, but `required_variables` is not set. "
                "By default, all prompt variables are treated as optional, which may lead to unintended behavior in "
                "multi-branch pipelines. To avoid unexpected execution, ensure that variables intended to be required "
                "are explicitly set in `required_variables`.",
                length=len(self.variables),
            )

        # setup inputs
        for var in self.variables:
            if self.required_variables == "*" or 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[Union[list[ChatMessage], str]] = None,
        template_variables: Optional[dict[str, Any]] = None,
        **kwargs,
    ):
        """
        Renders the prompt template with the provided variables.

        It applies the template variables to render the final prompt. You can provide variables with pipeline kwargs.
        To overwrite the default template, you can set the `template` parameter.
        To overwrite pipeline kwargs, you can set the `template_variables` parameter.

        :param template:
            An optional list of `ChatMessage` objects or string template 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 to overwrite the pipeline variables.
        :param kwargs:
            Pipeline variables used for rendering the prompt.

        :returns: A dictionary with the following keys:
            - `prompt`: The updated list of `ChatMessage` objects after rendering the 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 isinstance(template, list) and 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 = []
        if isinstance(template, list):
            for message in template:
                if message.is_from(ChatRole.USER) or message.is_from(ChatRole.SYSTEM):
                    self._validate_variables(set(template_variables_combined.keys()))
                    if message.text is None:
                        raise ValueError(NO_TEXT_ERROR_MESSAGE.format(role=message.role.value, message=message))
                    if message.text and "templatize_part" in message.text:
                        raise ValueError(FILTER_NOT_ALLOWED_ERROR_MESSAGE)
                    compiled_template = self._env.from_string(message.text)
                    rendered_text = compiled_template.render(template_variables_combined)
                    # deep copy the message to avoid modifying the original message
                    rendered_message: ChatMessage = deepcopy(message)
                    rendered_message._content = [TextContent(text=rendered_text)]
                    processed_messages.append(rendered_message)
                else:
                    processed_messages.append(message)
        elif isinstance(template, str):
            self._validate_variables(set(template_variables_combined.keys()))
            processed_messages = self._render_chat_messages_from_str_template(template, template_variables_combined)

        return {"prompt": processed_messages}

    def _render_chat_messages_from_str_template(
        self, template: str, template_variables: dict[str, Any]
    ) -> list[ChatMessage]:
        """
        Renders a chat message from a string template.

        This must be used in conjunction with the `ChatMessageExtension` Jinja2 extension
        and the `templatize_part` filter.
        """
        compiled_template = self._env.from_string(template)
        rendered = compiled_template.render(template_variables)

        messages = []
        for line in rendered.strip().split("\n"):
            line = line.strip()
            if line:
                messages.append(ChatMessage.from_dict(json.loads(line)))

        return 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.
        """
        if self.required_variables == "*":
            required_variables = sorted(self.variables)
        else:
            required_variables = self.required_variables
        missing_variables = [var for var in 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: {required_variables}. Provided variables: {provided_variables}."
            )

    def to_dict(self) -> dict[str, Any]:
        """
        Returns a dictionary representation of the component.

        :returns:
            Serialized dictionary representation of the component.
        """
        template: Optional[Union[list[dict[str, Any]], str]] = None
        if isinstance(self.template, list):
            template = [m.to_dict() for m in self.template]
        elif isinstance(self.template, str):
            template = self.template

        return default_to_dict(
            self, template=template, variables=self._variables, required_variables=self._required_variables
        )

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "ChatPromptBuilder":
        """
        Deserialize this component from a dictionary.

        :param data:
            The dictionary to deserialize and create the component.

        :returns:
            The deserialized component.
        """
        init_parameters = data["init_parameters"]
        template = init_parameters.get("template")
        if template:
            if isinstance(template, list):
                init_parameters["template"] = [ChatMessage.from_dict(d) for d in template]
            elif isinstance(template, str):
                init_parameters["template"] = template

        return default_from_dict(cls, data)

1	# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2	#
3	# SPDX-License-Identifier: Apache-2.0
4
5	import json	1✔
6	from copy import deepcopy	1✔
7	from typing import Any, Literal, Optional, Union	1✔
8
9	from jinja2 import meta	1✔
10	from jinja2.sandbox import SandboxedEnvironment	1✔
11
12	from haystack import component, default_from_dict, default_to_dict, logging	1✔
13	from haystack.dataclasses.chat_message import ChatMessage, ChatRole, TextContent	1✔
14	from haystack.lazy_imports import LazyImport	1✔
15	from haystack.utils import Jinja2TimeExtension, JinjaTemplateVariableExtractor	1✔
16	from haystack.utils.jinja2_chat_extension import ChatMessageExtension, templatize_part	1✔
17
18	logger = logging.getLogger(__name__)	1✔
19
20	with LazyImport("Run 'pip install \"arrow>=1.3.0\"'") as arrow_import:	1✔
21	import arrow # pylint: disable=unused-import	1✔
22
23	NO_TEXT_ERROR_MESSAGE = "ChatMessages from {role} role must contain text. Received ChatMessage with no text: {message}"	1✔
24
25	FILTER_NOT_ALLOWED_ERROR_MESSAGE = (	1✔
26	"The templatize_part filter cannot be used with a template containing a list of"
27	"ChatMessage objects. Use a string template or remove the templatize_part filter "
28	"from the template."
29	)
30
31
32	@component	1✔
33	class ChatPromptBuilder:	1✔
34	"""
35	Renders a chat prompt from a template using Jinja2 syntax.
36
37	A template can be a list of `ChatMessage` objects, or a special string, as shown in the usage examples.
38
39	It constructs prompts using static or dynamic templates, which you can update for each pipeline run.
40
41	Template variables in the template are optional unless specified otherwise.
42	If an optional variable isn't provided, it defaults to an empty string. Use `variable` and `required_variables`
43	to define input types and required variables.
44
45	### Usage examples
46
47	#### Static ChatMessage prompt template
48
49	```python
50	template = [ChatMessage.from_user("Translate to {{ target_language }}. Context: {{ snippet }}; Translation:")]
51	builder = ChatPromptBuilder(template=template)
52	builder.run(target_language="spanish", snippet="I can't speak spanish.")
53	```
54
55	#### Overriding static ChatMessage template at runtime
56
57	```python
58	template = [ChatMessage.from_user("Translate to {{ target_language }}. Context: {{ snippet }}; Translation:")]
59	builder = ChatPromptBuilder(template=template)
60	builder.run(target_language="spanish", snippet="I can't speak spanish.")
61
62	msg = "Translate to {{ target_language }} and summarize. Context: {{ snippet }}; Summary:"
63	summary_template = [ChatMessage.from_user(msg)]
64	builder.run(target_language="spanish", snippet="I can't speak spanish.", template=summary_template)
65	```
66
67	#### Dynamic ChatMessage prompt template
68
69	```python
70	from haystack.components.builders import ChatPromptBuilder
71	from haystack.components.generators.chat import OpenAIChatGenerator
72	from haystack.dataclasses import ChatMessage
73	from haystack import Pipeline
74	from haystack.utils import Secret
75
76	# no parameter init, we don't use any runtime template variables
77	prompt_builder = ChatPromptBuilder()
78	llm = OpenAIChatGenerator(api_key=Secret.from_token("<your-api-key>"), model="gpt-4o-mini")
79
80	pipe = Pipeline()
81	pipe.add_component("prompt_builder", prompt_builder)
82	pipe.add_component("llm", llm)
83	pipe.connect("prompt_builder.prompt", "llm.messages")
84
85	location = "Berlin"
86	language = "English"
87	system_message = ChatMessage.from_system("You are an assistant giving information to tourists in {{language}}")
88	messages = [system_message, ChatMessage.from_user("Tell me about {{location}}")]
89
90	res = pipe.run(data={"prompt_builder": {"template_variables": {"location": location, "language": language},
91	"template": messages}})
92	print(res)
93
94	>> {'llm': {'replies': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>, _content=[TextContent(text=
95	"Berlin is the capital city of Germany and one of the most vibrant
96	and diverse cities in Europe. Here are some key things to know...Enjoy your time exploring the vibrant and dynamic
97	capital of Germany!")], _name=None, _meta={'model': 'gpt-4o-mini',
98	'index': 0, 'finish_reason': 'stop', 'usage': {'prompt_tokens': 27, 'completion_tokens': 681, 'total_tokens':
99	708}})]}}
100
101	messages = [system_message, ChatMessage.from_user("What's the weather forecast for {{location}} in the next
102	{{day_count}} days?")]
103
104	res = pipe.run(data={"prompt_builder": {"template_variables": {"location": location, "day_count": "5"},
105	"template": messages}})
106
107	print(res)
108	>> {'llm': {'replies': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>, _content=[TextContent(text=
109	"Here is the weather forecast for Berlin in the next 5
110	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
111	closer to your visit.")], _name=None, _meta={'model': 'gpt-4o-mini',
112	'index': 0, 'finish_reason': 'stop', 'usage': {'prompt_tokens': 37, 'completion_tokens': 201,
113	'total_tokens': 238}})]}}
114	```
115
116	#### String prompt template
117	```python
118	from haystack.components.builders import ChatPromptBuilder
119	from haystack.dataclasses.image_content import ImageContent
120
121	template = \"\"\"
122	{% message role="system" %}
123	You are a helpful assistant.
124	{% endmessage %}
125
126	{% message role="user" %}
127	Hello! I am {{user_name}}. What's the difference between the following images?
128	{% for image in images %}
129	{{ image \| templatize_part }}
130	{% endfor %}
131	{% endmessage %}
132	\"\"\"
133
134	images = [ImageContent.from_file_path("apple.jpg"), ImageContent.from_file_path("orange.jpg")]
135
136	builder = ChatPromptBuilder(template=template)
137	builder.run(user_name="John", images=images)
138	```
139	"""
140
141	def __init__(	1✔
142	self,
143	template: Optional[Union[list[ChatMessage], str]] = None,
144	required_variables: Optional[Union[list[str], Literal["*"]]] = None,
145	variables: Optional[list[str]] = None,
146	):
147	"""
148	Constructs a ChatPromptBuilder component.
149
150	:param template:
151	A list of `ChatMessage` objects or a string template. The component looks for Jinja2 template syntax and
152	renders the prompt with the provided variables. Provide the template in either
153	the `init` method` or the `run` method.
154	:param required_variables:
155	List variables that must be provided as input to ChatPromptBuilder.
156	If a variable listed as required is not provided, an exception is raised.
157	If set to "*", all variables found in the prompt are required. Optional.
158	:param variables:
159	List input variables to use in prompt templates instead of the ones inferred from the
160	`template` parameter. For example, to use more variables during prompt engineering than the ones present
161	in the default template, you can provide them here.
162	"""
163	self._variables = variables	1✔
164	self._required_variables = required_variables	1✔
165	self.template = template	1✔
166
167	self._env = SandboxedEnvironment(extensions=[ChatMessageExtension])	1✔
168	self._env.filters["templatize_part"] = templatize_part	1✔
169	if arrow_import.is_successful():	1✔
170	self._env.add_extension(Jinja2TimeExtension)	1✔
171
172	extracted_variables = []	1✔
173	if template and not variables:	1✔
174
175	def _extract_from_text(	1✔
176	text: Optional[str], role: Optional[str] = None, is_filter_allowed: bool = False
177	) -> list[str]:
178	if text is None:	1✔
179	raise ValueError(NO_TEXT_ERROR_MESSAGE.format(role=role or "unknown", message=text))	×
180	if is_filter_allowed and "templatize_part" in text:	1✔
181	raise ValueError(FILTER_NOT_ALLOWED_ERROR_MESSAGE)	1✔
182
183	ast = self._env.parse(text)	1✔
184	template_variables = meta.find_undeclared_variables(ast)	1✔
185	assigned_variables = JinjaTemplateVariableExtractor._extract_declared_variables(text, env=self._env)	1✔
186	return list(template_variables - assigned_variables)	1✔
187
188	if isinstance(template, list):	1✔
189	for message in template:	1✔
190	if message.is_from(ChatRole.USER) or message.is_from(ChatRole.SYSTEM):	1✔
191	extracted_variables += _extract_from_text(	1✔
192	message.text, role=message.role.value, is_filter_allowed=True
193	)
194	elif isinstance(template, str):	1✔
195	extracted_variables = _extract_from_text(template, is_filter_allowed=False)	1✔
196
197	extracted_variables = extracted_variables or []	1✔
198	self.variables = variables or extracted_variables	1✔
199	self.required_variables = required_variables or []	1✔
200
201	if len(self.variables) > 0 and required_variables is None:	1✔
202	logger.warning(	1✔
203	"ChatPromptBuilder has {length} prompt variables, but `required_variables` is not set. "
204	"By default, all prompt variables are treated as optional, which may lead to unintended behavior in "
205	"multi-branch pipelines. To avoid unexpected execution, ensure that variables intended to be required "
206	"are explicitly set in `required_variables`.",
207	length=len(self.variables),
208	)
209
210	# setup inputs
211	for var in self.variables:	1✔
212	if self.required_variables == "*" or var in self.required_variables:	1✔
213	component.set_input_type(self, var, Any)	1✔
214	else:
215	component.set_input_type(self, var, Any, "")	1✔
216
217	@component.output_types(prompt=list[ChatMessage])	1✔
218	def run(	1✔
219	self,
220	template: Optional[Union[list[ChatMessage], str]] = None,
221	template_variables: Optional[dict[str, Any]] = None,
222	**kwargs,
223	):
224	"""
225	Renders the prompt template with the provided variables.
226
227	It applies the template variables to render the final prompt. You can provide variables with pipeline kwargs.
228	To overwrite the default template, you can set the `template` parameter.
229	To overwrite pipeline kwargs, you can set the `template_variables` parameter.
230
231	:param template:
232	An optional list of `ChatMessage` objects or string template to overwrite ChatPromptBuilder's default
233	template.
234	If `None`, the default template provided at initialization is used.
235	:param template_variables:
236	An optional dictionary of template variables to overwrite the pipeline variables.
237	:param kwargs:
238	Pipeline variables used for rendering the prompt.
239
240	:returns: A dictionary with the following keys:
241	- `prompt`: The updated list of `ChatMessage` objects after rendering the templates.
242	:raises ValueError:
243	If `chat_messages` is empty or contains elements that are not instances of `ChatMessage`.
244	"""
245	kwargs = kwargs or {}	1✔
246	template_variables = template_variables or {}	1✔
247	template_variables_combined = {kwargs, template_variables}	1✔
248
249	if template is None:	1✔
250	template = self.template	1✔
251
252	if not template:	1✔
253	raise ValueError(	1✔
254	f"The {self.__class__.__name__} requires a non-empty list of ChatMessage instances. "
255	f"Please provide a valid list of ChatMessage instances to render the prompt."
256	)
257
258	if isinstance(template, list) and not all(isinstance(message, ChatMessage) for message in template):	1✔
259	raise ValueError(	1✔
260	f"The {self.__class__.__name__} expects a list containing only ChatMessage instances. "
261	f"The provided list contains other types. Please ensure that all elements in the list "
262	f"are ChatMessage instances."
263	)
264
265	processed_messages = []	1✔
266	if isinstance(template, list):	1✔
267	for message in template:	1✔
268	if message.is_from(ChatRole.USER) or message.is_from(ChatRole.SYSTEM):	1✔
269	self._validate_variables(set(template_variables_combined.keys()))	1✔
270	if message.text is None:	1✔
271	raise ValueError(NO_TEXT_ERROR_MESSAGE.format(role=message.role.value, message=message))	×
272	if message.text and "templatize_part" in message.text:	1✔
273	raise ValueError(FILTER_NOT_ALLOWED_ERROR_MESSAGE)	1✔
274	compiled_template = self._env.from_string(message.text)	1✔
275	rendered_text = compiled_template.render(template_variables_combined)	1✔
276	# deep copy the message to avoid modifying the original message
277	rendered_message: ChatMessage = deepcopy(message)	1✔
278	rendered_message._content = [TextContent(text=rendered_text)]	1✔
279	processed_messages.append(rendered_message)	1✔
280	else:
281	processed_messages.append(message)	1✔
282	elif isinstance(template, str):	1✔
283	self._validate_variables(set(template_variables_combined.keys()))	1✔
284	processed_messages = self._render_chat_messages_from_str_template(template, template_variables_combined)	1✔
285
286	return {"prompt": processed_messages}	1✔
287
288	def _render_chat_messages_from_str_template(	1✔
289	self, template: str, template_variables: dict[str, Any]
290	) -> list[ChatMessage]:
291	"""
292	Renders a chat message from a string template.
293
294	This must be used in conjunction with the `ChatMessageExtension` Jinja2 extension
295	and the `templatize_part` filter.
296	"""
297	compiled_template = self._env.from_string(template)	1✔
298	rendered = compiled_template.render(template_variables)	1✔
299
300	messages = []	1✔
301	for line in rendered.strip().split("\n"):	1✔
302	line = line.strip()	1✔
303	if line:	1✔
304	messages.append(ChatMessage.from_dict(json.loads(line)))	1✔
305
306	return messages	1✔
307
308	def _validate_variables(self, provided_variables: set[str]):	1✔
309	"""
310	Checks if all the required template variables are provided.
311
312	:param provided_variables:
313	A set of provided template variables.
314	:raises ValueError:
315	If no template is provided or if all the required template variables are not provided.
316	"""
317	if self.required_variables == "*":	1✔
318	required_variables = sorted(self.variables)	1✔
319	else:
320	required_variables = self.required_variables	1✔
321	missing_variables = [var for var in required_variables if var not in provided_variables]	1✔
322	if missing_variables:	1✔
323	missing_vars_str = ", ".join(missing_variables)	1✔
324	raise ValueError(	1✔
325	f"Missing required input variables in ChatPromptBuilder: {missing_vars_str}. "
326	f"Required variables: {required_variables}. Provided variables: {provided_variables}."
327	)
328
329	def to_dict(self) -> dict[str, Any]:	1✔
330	"""
331	Returns a dictionary representation of the component.
332
333	:returns:
334	Serialized dictionary representation of the component.
335	"""
336	template: Optional[Union[list[dict[str, Any]], str]] = None	1✔
337	if isinstance(self.template, list):	1✔
338	template = [m.to_dict() for m in self.template]	1✔
339	elif isinstance(self.template, str):	1✔
340	template = self.template	1✔
341
342	return default_to_dict(	1✔
343	self, template=template, variables=self._variables, required_variables=self._required_variables
344	)
345
346	@classmethod	1✔
347	def from_dict(cls, data: dict[str, Any]) -> "ChatPromptBuilder":	1✔
348	"""
349	Deserialize this component from a dictionary.
350
351	:param data:
352	The dictionary to deserialize and create the component.
353
354	:returns:
355	The deserialized component.
356	"""
357	init_parameters = data["init_parameters"]	1✔
358	template = init_parameters.get("template")	1✔
359	if template:	1✔
360	if isinstance(template, list):	1✔
361	init_parameters["template"] = [ChatMessage.from_dict(d) for d in template]	1✔
362	elif isinstance(template, str):	1✔
363	init_parameters["template"] = template	1✔
364
365	return default_from_dict(cls, data)	1✔

deepset-ai / haystack / 19204656740

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous