PromptTemplate#

This page presents all APIs and classes related to prompt Templates.

class wayflowcore.templates.template.PromptTemplate(messages, output_parser=None, input_descriptors=None, pre_rendering_transforms=None, post_rendering_transforms=None, tools=None, native_tool_calling=True, response_format=None, native_structured_generation=True, generation_config=None, _partial_values=<factory>, *, id=<factory>, __metadata_info__=<factory>, name='', description=None)#

Represents a flexible and extensible template for constructing prompts to be sent to large language models (LLMs).

The PromptTemplate class enables the definition of prompt messages with variable placeholders, supports both native and custom tool calling, and allows for structured output generation. It manages input descriptors, message transforms (pre- and post chat_history rendering), and partial formatting for efficiency. The class also integrates with output parsers, tools and llm generation configurations.

Parameters:

messages (Sequence[Message | MessageAsDictT]) –
output_parser (OutputParser | List[OutputParser] | None) –
input_descriptors (List[Property] | None) –
pre_rendering_transforms (List[MessageTransform] | None) –
post_rendering_transforms (List[MessageTransform] | None) –
tools (List[Tool] | None) –
native_tool_calling (bool) –
response_format (Property | None) –
native_structured_generation (bool) –
generation_config (LlmGenerationConfig | None) –
_partial_values (Dict[str, Any]) –
id (str) –
__metadata_info__ (Dict[str, Any]) –
name (str) –
description (str | None) –

CHAT_HISTORY_PLACEHOLDER: ClassVar[Message] = Message(id='7a63eaa5-f81e-442e-a793-7ce53f102ebb', __metadata_info__={}, role='system', contents=[TextContent(content='$$__CHAT_HISTORY_PLACEHOLDER__$$')], tool_requests=None, tool_result=None, display_only=False, _prompt_cache_key=None, _reasoning_content=None, sender=None, recipients=set())#: Message placeholder in case the chat history is formatted as a chat.

CHAT_HISTORY_PLACEHOLDER_NAME: ClassVar[str] = '__CHAT_HISTORY__'#: Reserved name of the placeholder for the chat history, if rendered in one message.

RESPONSE_FORMAT_PLACEHOLDER_NAME: ClassVar[str] = '__RESPONSE_FORMAT__'#: Reserved name of the placeholder for the expected output format. Only used if non-native structured generation, to be able to specify the JSON format anywhere in the prompt.

TOOL_PLACEHOLDER_NAME: ClassVar[str] = '__TOOLS__'#: Reserved name of the placeholder for tools.

copy()#

Returns a copy of the template.

Return type:: PromptTemplate

format(inputs=None)#

Formats the prompt into a list of messages to pass to the LLM

Return type:: Prompt
Parameters:: inputs (Dict[str, Any] | None) –

async format_async(inputs=None)#

Synchronously formats the prompt into a list of messages to pass to the LLM

Return type:: Prompt
Parameters:: inputs (Dict[str, Any] | None) –

classmethod from_string(template, output_parser=None, input_descriptors=None, pre_rendering_transforms=None, post_rendering_transforms=None, tools=None, native_tool_calling=True, response_format=None, native_structured_generation=True, generation_config=None)#

Creates a prompt template from a string.

Return type:

PromptTemplate

Parameters:

template (str) –
output_parser (OutputParser | None) –
input_descriptors (List[Property] | None) –
pre_rendering_transforms (List[MessageTransform] | None) –
post_rendering_transforms (List[MessageTransform] | None) –
tools (List[Tool] | None) –
native_tool_calling (bool) –
response_format (Property | None) –
native_structured_generation (bool) –
generation_config (LlmGenerationConfig | None) –

generation_config: Optional[LlmGenerationConfig] = None#: Parameters to configure the generation.

input_descriptors: Optional[List[Property]] = None#: Input descriptors that will be picked up by PromptExecutionStep or AgentExecutionStep. Resolved by default from the variables present in the messages.

messages: Sequence[Union[Message, MessageAsDictT]]#: List of messages for the prompt.

native_structured_generation: bool = True#: Whether to use native structured generation or not. All llm providers might not support it.

native_tool_calling: bool = True#: Whether to use the native tool calling of the model or not. All llm providers might not support it.

output_parser: Union[OutputParser, List[OutputParser], None] = None#: Post-processing applied on the raw output of the LLM.

post_rendering_transforms: Optional[List[MessageTransform]] = None#: Message transform applied on the rendered list of messages. Use these to ensure the remote LLM will accept the prompt: ensuring a single system_message, ensuring alternating user/assistant messages, …

pre_rendering_transforms: Optional[List[MessageTransform]] = None#: Message transform applied before rendering the list of messages into the template. Use these to summarize messages, …

response_format: Optional[Property] = None#: Specific format the llm answer should follow.

tools: Optional[List[Tool]] = None#: Tools to use in the prompt.

with_additional_post_rendering_transform(transform, append=True)#

Returns a copy of the prompt template with an additional post rendering transform

Return type:

PromptTemplate

Parameters:

transform (MessageTransform) –
append (bool) –

with_additional_pre_rendering_transform(transform, append=True)#

Returns a copy of the prompt template with an additional pre rendering transform

Return type:

PromptTemplate

Parameters:

transform (MessageTransform) –
append (bool) –

with_generation_config(generation_config, override=True)#

Override: Whether the template config should be overridden or should overridden this config.

Return type:

PromptTemplate

Parameters:

generation_config (LlmGenerationConfig | None) –
override (bool) –

with_output_parser(output_parser)#

Replaces the output parser of this template.

Return type:: PromptTemplate
Parameters:: output_parser (OutputParser | List[OutputParser]) –

with_partial(inputs)#

Partially formats the prompt with the given inputs (to avoid formatting everything at each call, if some inputs do not change). These inputs are not rendered directly, but stored for a later call to format().

Return type:: PromptTemplate
Parameters:: inputs (Dict[str, Any]) –

with_response_format(response_format)#

Returns a copy of the template equipped with a given response format.

Return type:: PromptTemplate
Parameters:: response_format (Property | None) –

with_tools(tools)#

Returns a copy of the template equipped with the given tools.

Return type:: PromptTemplate
Parameters:: tools (List[Tool] | None) –

OutputParser#

class wayflowcore.outputparser.OutputParser(__metadata_info__=None, id=None)#

Abstract base class for output parsers that process LLM outputs.

Parameters:

__metadata_info__ (Dict[str, Any] | None) –
id (str | None) –

abstract parse_output(content)#

Parses the LLM raw output

Return type:: Message
Parameters:: content (Message) –

abstract async parse_output_streaming(content)#

Can parse the result returned by streaming By default does nothing until the message has been completely generated, but can implement specific stream methods if we want to stream something specific

Return type:: Any
Parameters:: content (Any) –

class wayflowcore.outputparser.RegexOutputParser(regex_pattern, strict=True, *, id=<factory>, __metadata_info__=<factory>)#

Parses some text with Regex, potentially several regex to fill a dict

Examples

>>> import re
>>> from wayflowcore.messagelist import Message
>>> from wayflowcore.outputparser import RegexOutputParser, RegexPattern
>>> RegexOutputParser(
...     regex_pattern=RegexPattern(pattern=r"Solution is: (.*)", flags=re.DOTALL)
... ).parse_output(Message(content="Solution is: Bern is the capital of Switzerland")).content
'Bern is the capital of Switzerland'

>>> RegexOutputParser(
...     regex_pattern={
...         'thought': "THOUGHT: (.*) ACTION:",
...         'action': "ACTION: (.*)",
...     }
... ).parse_output(Message("THOUGHT: blahblah ACTION: doing")).content
'{"thought": "blahblah", "action": "doing"}'

Parameters:

regex_pattern (str | RegexPattern | Dict[str, str | RegexPattern]) –
strict (bool) –
id (str) –
__metadata_info__ (Dict[str, Any]) –

parse_output(message)#

Parses the LLM raw output

Return type:: Message
Parameters:: message (Message) –

async parse_output_streaming(content)#

Can parse the result returned by streaming By default does nothing until the message has been completely generated, but can implement specific stream methods if we want to stream something specific

Return type:: Any
Parameters:: content (Any) –

regex_pattern: Union[str, RegexPattern, Dict[str, Union[str, RegexPattern]]]#: Regex pattern to use

strict: bool = True#: Whether to return empty string if no match is found or return the raw text

class wayflowcore.outputparser.RegexPattern(pattern, match='first', flags=None)#

Represents a regex pattern and matching options for output parsing.

Parameters:

pattern (str) –
match (Literal['first', 'last']) –
flags (int | RegexFlag | None) –

flags: Union[int, RegexFlag, None] = None#: Potential regex flags to use (re.DOTALL for multiline matching for example)

static from_str(pattern, flags=RegexFlag.DOTALL)#

Return type:

RegexPattern

Parameters:

pattern (str | RegexPattern) –
flags (int | RegexFlag | None) –

match: Literal['first', 'last'] = 'first'#: Whether to take the first match or the last match

pattern: str#: Regex pattern to match

class wayflowcore.outputparser.JsonOutputParser(properties=None, *, id=<factory>, __metadata_info__=<factory>)#

Parses output as JSON, repairing and serializing as needed.

Parameters:

properties (Dict[str, str] | None) –
id (str) –
__metadata_info__ (Dict[str, Any]) –

parse_output(content)#

Parses the LLM raw output

Return type:: Message
Parameters:: content (Message) –

async parse_output_streaming(content)#

Can parse the result returned by streaming By default does nothing until the message has been completely generated, but can implement specific stream methods if we want to stream something specific

Return type:: Any
Parameters:: content (Any) –

properties: Optional[Dict[str, str]] = None#: Dictionary of property names and jq queries to manipulate the loaded JSON

class wayflowcore.outputparser.ToolOutputParser(tools=None, *, id=<factory>, __metadata_info__=<factory>)#

Base parser for tool requests

Parameters:

tools (List[Tool] | None) –
id (str) –
__metadata_info__ (Dict[str, Any]) –

parse_output(message)#

Separates the raw output into thoughts and calls, and then parses the calls into ToolRequests

Return type:: Message
Parameters:: message (Message) –

async parse_output_streaming(content)#

Can parse the result returned by streaming By default does nothing until the message has been completely generated, but can implement specific stream methods if we want to stream something specific

Return type:: Any
Parameters:: content (Any) –

parse_thoughts_and_calls(raw_txt)#

Default function to separate thoughts and tool calls

Return type:: Tuple[str, str]
Parameters:: raw_txt (str) –

abstract parse_tool_request_from_str(raw_txt)#

Return type:: List[ToolRequest]
Parameters:: raw_txt (str) –

tools: Optional[List[Tool]] = None#

with_tools(tools)#

Enhances the tool parser with some validation of the parsed tool calls according to specific tools

Return type:: ToolOutputParser
Parameters:: tools (List[Tool] | None) –

Message transforms#

class wayflowcore.transforms.MessageTransform#

Abstract base class for message transforms.

Subclasses should implement the __call__ method to transform a list of Message objects and return a new list of Message objects, typically for preprocessing or postprocessing message flows in the system.

async call_async(messages)#

Implement this method for asynchronous work (IO-bounded, with LLM calls, DB loading …)

Return type:: List[Message]
Parameters:: messages (List[Message]) –

class wayflowcore.transforms.CoalesceSystemMessagesTransform#

Transform that merges consecutive system messages at the start of a message list into a single system message. This is useful for reducing redundancy and ensuring that only one system message appears at the beginning of the conversation.

class wayflowcore.transforms.RemoveEmptyNonUserMessageTransform#

Transform that removes messages which are empty and not from the user.

Any message with empty content and no tool requests, except for user messages, will be filtered out from the message list.

This is useful in case the template contains optional messages, which will be discarded if their content is empty (with a string template such as “{% if __PLAN__ %}{{ __PLAN__ }}{% endif %}”).

class wayflowcore.transforms.AppendTrailingSystemMessageToUserMessageTransform#

Transform that appends the content of a trailing system message to the previous user message.

If the last message in the list is a system message and the one before it is a user message, this transform merges the system message content into the user message, reducing message clutter.

This is useful if the underlying LLM does not support system messages at the end.

class wayflowcore.transforms.SplitPromptOnMarkerMessageTransform(marker=None)#

Split prompts on a marker into multiple messages with the same role. Only apply to the messages without tool_requests and tool_result.

This transform is useful for script-based execution flows, where a single prompt script can be converted into multiple conversation turns for step-by-step reasoning.

Parameters:: marker (str | None) –

class wayflowcore.transforms.MessageSummarizationTransform(llm, max_message_size=20000, summarization_instructions='Please make a summary of this message. Include relevant information and keep it short. Your response will replace the message, so just output the summary directly, no introduction needed.', summarized_message_template='Summarized message: {{summary}}', datastore=None, cache_collection_name='summarized_messages_cache', max_cache_size=10000, max_cache_lifetime=14400)#

Summarizes oversized messages using an LLM and optionally caches summaries.

This is useful for long conversations where the context can become too large for the LLM to handle.

Parameters:

llm (LlmModel) – LLM to use for the summarization. If the agent’s llm supports images, then this llm should also support images.
max_message_size (int) – The maximum size in number of characters for the content of a message. This is converted to an estimated token count using heuristics (approximately max_message_size / 4). Images in the message are also converted to estimated token counts (assuming 16x16 patches and defaulting to 2048x2048 if the image type is not PNG, JPEG, or JPEG2000). Summarization is triggered when the total estimated token count (text + images) of a message exceeds this threshold.
summarization_instructions (str) – Instruction for the LLM on how to summarize the messages.
summarized_message_template (str) – Jinja2 template on how to present the summary (with variable summary) to the agent using the transform.
datastore (Optional[Datastore]) –
Datastore on which to store the cache. If None, an in-memory Datastore will be created automatically.

Important

The datastore needs to have a collection called cache_collection_name. This collection’s entries should be defined using MessageSummarizationTransform.get_entity_definition or as follows: ` Entity({ "cache_key": StringProperty(), "cache_content": StringProperty(), "created_at": FloatProperty(), "last_used_at": FloatProperty(), }) `
cache_collection_name (str) – Name of the collection/table in the cache for storing summarized messages.
max_cache_size (Optional[int]) – The number of cache entries (messages) kept in the cache. If None, there is no limit on cache size and no eviction occurs.
max_cache_lifetime (Optional[int]) – max lifetime of a message in the cache in seconds. If None, cached data persists indefinitely.

Examples

>>> from wayflowcore.transforms import MessageSummarizationTransform
>>> summarization_transform = MessageSummarizationTransform(
...     llm=llm,
...     max_message_size=30_000
... )

DEFAULT_CACHE_COLLECTION_NAME = 'summarized_messages_cache'#

async call_async(messages)#

Implement this method for asynchronous work (IO-bounded, with LLM calls, DB loading …)

Return type:: List[Message]
Parameters:: messages (List[Message]) –

static get_entity_definition()#

Return type:: Entity

async summarize_if_needed(contents, conv_id, msg_idx, _type='content')#

Return type:

List[MessageContent]

Parameters:

contents (List[MessageContent]) –
conv_id (str) –
msg_idx (int) –
_type (Literal['toolres', 'content']) –

class wayflowcore.transforms.ConversationSummarizationTransform(llm, max_num_messages=50, min_num_messages=10, summarization_instructions='Please make a summary of this conversation. Include relevant information and keep it short. Your response will replace the messages, so just output the summary directly, no introduction needed.', summarized_conversation_template='Summarized conversation: {{summary}}', datastore=None, max_cache_size=10000, max_cache_lifetime=14400, cache_collection_name='summarized_conversations_cache')#

Parameters:

llm (LlmModel) – LLM to use for the summarization.
max_num_messages (int) – Number of message after which we trigger summarization. Tune this parameter depending on the context length of your model and the price you are willing to pay (higher means longer conversation prompts and more tokens).
min_num_messages (int) – Number of recent messages to keep from summarizing. Tune this parameter to prevent from summarizing very recent messages and keep a very responsive and relevant agent.
summarization_instructions (str) – Instruction for the LLM on how to summarize the conversation.
summarized_conversation_template (str) – Jinja2 template on how to present the summary (with variable summary) to the agent using the transform.
datastore (Optional[Datastore]) – Datastore on which to store the cache. If not specified, an in-memory Datastore will be created automatically. If None, caching is disabled (not recommended)
max_cache_size (Optional[int]) – The maximum number of entries kept in the cache
max_cache_lifetime (Optional[int]) – max lifetime of an element in the cache in seconds
cache_collection_name (str) – the collection in the cache datastore where summarized conversations will be stored

Examples

>>> from wayflowcore.transforms import ConversationSummarizationTransform
>>> summarization_transform = ConversationSummarizationTransform(
...     llm=llm,
...     max_num_messages=30,
...     min_num_messages=10
... )

DEFAULT_CACHE_COLLECTION_NAME = 'summarized_conversations_cache'#

async call_async(messages)#

Implement this method for asynchronous work (IO-bounded, with LLM calls, DB loading …)

Return type:: List[Message]
Parameters:: messages (List[Message]) –

static get_entity_definition()#

Return type:: Entity

Helpers#

wayflowcore.templates.structuredgeneration.adapt_prompt_template_for_json_structured_generation(prompt_template)#

Adapts a prompt template for native structured generation to one that leverages a special system prompt and a JSON Output Parser.

Parameters:: prompt_template (PromptTemplate) – The prompt template to adapt
Returns:: The new prompt template, with the special system prompt and output parsers configured
Return type:: PromptTemplate
Raises:: ValueError – If the prompt template is already configured to use non-native structured generation, or the prompt template has no response format.

class wayflowcore._utils._templating_helpers.ToolRequestAsDictT#

args: Dict[str, Any]#

name: str#

tool_request_id: str#

class wayflowcore._utils._templating_helpers.ToolResultAsDictT#

content: Any#

tool_request_id: str#

class wayflowcore._utils._templating_helpers.MessageAsDictT#

content: str#

role: Literal['user', 'assistant', 'system']#

tool_requests: Optional[List[ToolRequestAsDictT]]#

tool_result: Optional[ToolResultAsDictT]#