19358970555

Committed 14 Nov 2025 08:36AM UTC coverage: 91.441% (+0.001%) from 91.44%

Build # 19358970555

Build Type

Pull #9932

github

Committed by

web-flow

Commit Message

Merge 24b0c631f into 41f02010a

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

Run Details

13739 of 15025 relevant lines covered (91.44%)

0.91 hits per line

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

98.28

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 Jinja2TemplateVariableExtractor, Jinja2TimeExtension
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:
                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)
                jinja_var_extractor = Jinja2TemplateVariableExtractor(env=self._env)
                assigned_variables = jinja_var_extractor._extract_from_text(template_str=text)
                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 Jinja2TemplateVariableExtractor, Jinja2TimeExtension	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:
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	jinja_var_extractor = Jinja2TemplateVariableExtractor(env=self._env)	1✔
186	assigned_variables = jinja_var_extractor._extract_from_text(template_str=text)	1✔
187	return list(template_variables - assigned_variables)	1✔
188
189	if isinstance(template, list):	1✔
190	for message in template:	1✔
191	if message.is_from(ChatRole.USER) or message.is_from(ChatRole.SYSTEM):	1✔
192	extracted_variables += _extract_from_text(	1✔
193	message.text, role=message.role.value, is_filter_allowed=True
194	)
195	elif isinstance(template, str):	1✔
196	extracted_variables = _extract_from_text(template, is_filter_allowed=False)	1✔
197
198	extracted_variables = extracted_variables or []	1✔
199	self.variables = variables or extracted_variables	1✔
200	self.required_variables = required_variables or []	1✔
201
202	if len(self.variables) > 0 and required_variables is None:	1✔
203	logger.warning(	1✔
204	"ChatPromptBuilder has {length} prompt variables, but `required_variables` is not set. "
205	"By default, all prompt variables are treated as optional, which may lead to unintended behavior in "
206	"multi-branch pipelines. To avoid unexpected execution, ensure that variables intended to be required "
207	"are explicitly set in `required_variables`.",
208	length=len(self.variables),
209	)
210
211	# setup inputs
212	for var in self.variables:	1✔
213	if self.required_variables == "*" or var in self.required_variables:	1✔
214	component.set_input_type(self, var, Any)	1✔
215	else:
216	component.set_input_type(self, var, Any, "")	1✔
217
218	@component.output_types(prompt=list[ChatMessage])	1✔
219	def run(	1✔
220	self,
221	template: Optional[Union[list[ChatMessage], str]] = None,
222	template_variables: Optional[dict[str, Any]] = None,
223	**kwargs,
224	):
225	"""
226	Renders the prompt template with the provided variables.
227
228	It applies the template variables to render the final prompt. You can provide variables with pipeline kwargs.
229	To overwrite the default template, you can set the `template` parameter.
230	To overwrite pipeline kwargs, you can set the `template_variables` parameter.
231
232	:param template:
233	An optional list of `ChatMessage` objects or string template to overwrite ChatPromptBuilder's default
234	template.
235	If `None`, the default template provided at initialization is used.
236	:param template_variables:
237	An optional dictionary of template variables to overwrite the pipeline variables.
238	:param kwargs:
239	Pipeline variables used for rendering the prompt.
240
241	:returns: A dictionary with the following keys:
242	- `prompt`: The updated list of `ChatMessage` objects after rendering the templates.
243	:raises ValueError:
244	If `chat_messages` is empty or contains elements that are not instances of `ChatMessage`.
245	"""
246	kwargs = kwargs or {}	1✔
247	template_variables = template_variables or {}	1✔
248	template_variables_combined = {kwargs, template_variables}	1✔
249
250	if template is None:	1✔
251	template = self.template	1✔
252
253	if not template:	1✔
254	raise ValueError(	1✔
255	f"The {self.__class__.__name__} requires a non-empty list of ChatMessage instances. "
256	f"Please provide a valid list of ChatMessage instances to render the prompt."
257	)
258
259	if isinstance(template, list) and not all(isinstance(message, ChatMessage) for message in template):	1✔
260	raise ValueError(	1✔
261	f"The {self.__class__.__name__} expects a list containing only ChatMessage instances. "
262	f"The provided list contains other types. Please ensure that all elements in the list "
263	f"are ChatMessage instances."
264	)
265
266	processed_messages = []	1✔
267	if isinstance(template, list):	1✔
268	for message in template:	1✔
269	if message.is_from(ChatRole.USER) or message.is_from(ChatRole.SYSTEM):	1✔
270	self._validate_variables(set(template_variables_combined.keys()))	1✔
271	if message.text is None:	1✔
272	raise ValueError(NO_TEXT_ERROR_MESSAGE.format(role=message.role.value, message=message))	×
273	if message.text and "templatize_part" in message.text:	1✔
274	raise ValueError(FILTER_NOT_ALLOWED_ERROR_MESSAGE)	1✔
275	compiled_template = self._env.from_string(message.text)	1✔
276	rendered_text = compiled_template.render(template_variables_combined)	1✔
277	# deep copy the message to avoid modifying the original message
278	rendered_message: ChatMessage = deepcopy(message)	1✔
279	rendered_message._content = [TextContent(text=rendered_text)]	1✔
280	processed_messages.append(rendered_message)	1✔
281	else:
282	processed_messages.append(message)	1✔
283	elif isinstance(template, str):	1✔
284	self._validate_variables(set(template_variables_combined.keys()))	1✔
285	processed_messages = self._render_chat_messages_from_str_template(template, template_variables_combined)	1✔
286
287	return {"prompt": processed_messages}	1✔
288
289	def _render_chat_messages_from_str_template(	1✔
290	self, template: str, template_variables: dict[str, Any]
291	) -> list[ChatMessage]:
292	"""
293	Renders a chat message from a string template.
294
295	This must be used in conjunction with the `ChatMessageExtension` Jinja2 extension
296	and the `templatize_part` filter.
297	"""
298	compiled_template = self._env.from_string(template)	1✔
299	rendered = compiled_template.render(template_variables)	1✔
300
301	messages = []	1✔
302	for line in rendered.strip().split("\n"):	1✔
303	line = line.strip()	1✔
304	if line:	1✔
305	messages.append(ChatMessage.from_dict(json.loads(line)))	1✔
306
307	return messages	1✔
308
309	def _validate_variables(self, provided_variables: set[str]):	1✔
310	"""
311	Checks if all the required template variables are provided.
312
313	:param provided_variables:
314	A set of provided template variables.
315	:raises ValueError:
316	If no template is provided or if all the required template variables are not provided.
317	"""
318	if self.required_variables == "*":	1✔
319	required_variables = sorted(self.variables)	1✔
320	else:
321	required_variables = self.required_variables	1✔
322	missing_variables = [var for var in required_variables if var not in provided_variables]	1✔
323	if missing_variables:	1✔
324	missing_vars_str = ", ".join(missing_variables)	1✔
325	raise ValueError(	1✔
326	f"Missing required input variables in ChatPromptBuilder: {missing_vars_str}. "
327	f"Required variables: {required_variables}. Provided variables: {provided_variables}."
328	)
329
330	def to_dict(self) -> dict[str, Any]:	1✔
331	"""
332	Returns a dictionary representation of the component.
333
334	:returns:
335	Serialized dictionary representation of the component.
336	"""
337	template: Optional[Union[list[dict[str, Any]], str]] = None	1✔
338	if isinstance(self.template, list):	1✔
339	template = [m.to_dict() for m in self.template]	1✔
340	elif isinstance(self.template, str):	1✔
341	template = self.template	1✔
342
343	return default_to_dict(	1✔
344	self, template=template, variables=self._variables, required_variables=self._required_variables
345	)
346
347	@classmethod	1✔
348	def from_dict(cls, data: dict[str, Any]) -> "ChatPromptBuilder":	1✔
349	"""
350	Deserialize this component from a dictionary.
351
352	:param data:
353	The dictionary to deserialize and create the component.
354
355	:returns:
356	The deserialized component.
357	"""
358	init_parameters = data["init_parameters"]	1✔
359	template = init_parameters.get("template")	1✔
360	if template:	1✔
361	if isinstance(template, list):	1✔
362	init_parameters["template"] = [ChatMessage.from_dict(d) for d in template]	1✔
363	elif isinstance(template, str):	1✔
364	init_parameters["template"] = template	1✔
365
366	return default_from_dict(cls, data)	1✔

deepset-ai / haystack / 19358970555

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