chore: import upstream snapshot with attribution
CodeQL / Analyze (csharp) (push) Waiting to run
CodeQL / Analyze (python) (push) Waiting to run

This commit is contained in:
wehub-resource-sync
2026-07-13 13:21:23 +08:00
commit b957a53def
5423 changed files with 863745 additions and 0 deletions
@@ -0,0 +1,15 @@
# Copyright (c) Microsoft. All rights reserved.
from semantic_kernel.prompt_template.handlebars_prompt_template import HandlebarsPromptTemplate
from semantic_kernel.prompt_template.input_variable import InputVariable
from semantic_kernel.prompt_template.jinja2_prompt_template import Jinja2PromptTemplate
from semantic_kernel.prompt_template.kernel_prompt_template import KernelPromptTemplate
from semantic_kernel.prompt_template.prompt_template_config import PromptTemplateConfig
__all__ = [
"HandlebarsPromptTemplate",
"InputVariable",
"Jinja2PromptTemplate",
"KernelPromptTemplate",
"PromptTemplateConfig",
]
@@ -0,0 +1,9 @@
# Copyright (c) Microsoft. All rights reserved.
from typing import Literal
KERNEL_TEMPLATE_FORMAT_NAME: Literal["semantic-kernel"] = "semantic-kernel"
HANDLEBARS_TEMPLATE_FORMAT_NAME: Literal["handlebars"] = "handlebars"
JINJA2_TEMPLATE_FORMAT_NAME: Literal["jinja2"] = "jinja2"
TEMPLATE_FORMAT_TYPES = Literal["semantic-kernel", "handlebars", "jinja2"]
@@ -0,0 +1,113 @@
# Copyright (c) Microsoft. All rights reserved.
import logging
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
from pybars import Compiler, PybarsError
from pydantic import PrivateAttr, field_validator
from semantic_kernel.exceptions import HandlebarsTemplateRenderException, HandlebarsTemplateSyntaxError
from semantic_kernel.functions.kernel_arguments import KernelArguments
from semantic_kernel.prompt_template.const import HANDLEBARS_TEMPLATE_FORMAT_NAME
from semantic_kernel.prompt_template.prompt_template_base import PromptTemplateBase
from semantic_kernel.prompt_template.utils import HANDLEBAR_SYSTEM_HELPERS, create_template_helper_from_function
if TYPE_CHECKING:
from semantic_kernel.kernel import Kernel
from semantic_kernel.prompt_template.prompt_template_config import PromptTemplateConfig
logger: logging.Logger = logging.getLogger(__name__)
class HandlebarsPromptTemplate(PromptTemplateBase):
"""Create a Handlebars prompt template.
Handlebars are parsed as a whole and therefore do not have variables that can be extracted,
also with handlebars there is no distinction in syntax between a variable and a value,
a value that is encountered is tried to resolve with the arguments and the functions,
if not found, the literal value is returned.
Args:
prompt_template_config (PromptTemplateConfig): The prompt template configuration
This is checked if the template format is 'handlebars'
allow_dangerously_set_content (bool = False): Allow content without encoding throughout, this overrides
the same settings in the prompt template config and input variables.
This reverts the behavior to unencoded input.
Raises:
ValueError: If the template format is not 'handlebars'
HandlebarsTemplateSyntaxError: If the handlebars template has a syntax error
"""
_template_compiler: Any = PrivateAttr()
@field_validator("prompt_template_config")
@classmethod
def validate_template_format(cls, v: "PromptTemplateConfig") -> "PromptTemplateConfig":
"""Validate the template format."""
if v.template_format != HANDLEBARS_TEMPLATE_FORMAT_NAME:
raise ValueError(f"Invalid prompt template format: {v.template_format}. Expected: handlebars")
return v
def model_post_init(self, __context: Any) -> None:
"""Post init model."""
if not self.prompt_template_config.template:
self._template_compiler = None
return
try:
self._template_compiler = Compiler().compile(self.prompt_template_config.template)
except PybarsError as e:
logger.error(f"Invalid handlebars template: {self.prompt_template_config.template}")
raise HandlebarsTemplateSyntaxError(
f"Invalid handlebars template: {self.prompt_template_config.template}"
) from e
async def render(self, kernel: "Kernel", arguments: "KernelArguments | None" = None) -> str:
"""Render the prompt template.
Using the prompt template, replace the variables with their values
and execute the functions replacing their reference with the
function result.
Args:
kernel: The kernel instance
arguments: The kernel arguments
Returns:
The prompt template ready to be used for an AI request
"""
if not self._template_compiler:
return ""
if arguments is None:
arguments = KernelArguments()
arguments = self._get_trusted_arguments(arguments)
allow_unsafe_function_output = self._get_allow_dangerously_set_function_output()
helpers: dict[str, Callable[..., Any]] = {}
for plugin in kernel.plugins.values():
helpers.update({
function.fully_qualified_name: create_template_helper_from_function(
function,
kernel,
arguments,
self.prompt_template_config.template_format,
allow_unsafe_function_output,
)
for function in plugin
})
helpers.update(HANDLEBAR_SYSTEM_HELPERS)
try:
return self._template_compiler(
arguments,
helpers=helpers,
)
except PybarsError as exc:
logger.error(
f"Error rendering prompt template: {self.prompt_template_config.template} with arguments: {arguments}"
)
raise HandlebarsTemplateRenderException(
f"Error rendering prompt template: {self.prompt_template_config.template} "
f"with arguments: {arguments}: error: {exc}"
) from exc
@@ -0,0 +1,26 @@
# Copyright (c) Microsoft. All rights reserved.
from typing import Any
from semantic_kernel.kernel_pydantic import KernelBaseModel
class InputVariable(KernelBaseModel):
"""Input variable for a prompt template.
Args:
name: The name of the input variable.
description: The description of the input variable.
default: The default value of the input variable.
is_required: Whether the input variable is required.
json_schema: The JSON schema for the input variable.
allow_dangerously_set_content: Allow content without encoding, this controls
if this variable is encoded before use, default is False.
"""
name: str
description: str | None = ""
default: Any | None = ""
is_required: bool | None = True
json_schema: str | None = ""
allow_dangerously_set_content: bool = False
@@ -0,0 +1,115 @@
# Copyright (c) Microsoft. All rights reserved.
import logging
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
from jinja2 import BaseLoader, TemplateError
from jinja2.sandbox import ImmutableSandboxedEnvironment
from pydantic import PrivateAttr, field_validator
from semantic_kernel.exceptions import Jinja2TemplateRenderException
from semantic_kernel.functions.kernel_arguments import KernelArguments
from semantic_kernel.prompt_template.const import JINJA2_TEMPLATE_FORMAT_NAME
from semantic_kernel.prompt_template.prompt_template_base import PromptTemplateBase
from semantic_kernel.prompt_template.prompt_template_config import PromptTemplateConfig
from semantic_kernel.prompt_template.utils import JINJA2_SYSTEM_HELPERS
from semantic_kernel.prompt_template.utils.template_function_helpers import create_template_helper_from_function
if TYPE_CHECKING:
from semantic_kernel.kernel import Kernel
logger: logging.Logger = logging.getLogger(__name__)
class Jinja2PromptTemplate(PromptTemplateBase):
"""Creates and renders Jinja2 prompt templates to text.
Jinja2 templates support advanced features such as variable substitution, control structures,
and inheritance, making it possible to dynamically generate text based on input arguments
and predefined functions. This class leverages Jinja2's flexibility to render prompts that
can include conditional logic, loops, and functions, based on the provided template configuration
and arguments.
Note that the fully qualified function name (in the form of "plugin-function") is not allowed
in Jinja2 because of the hyphen. Therefore, the function name is replaced with an underscore,
which are allowed in Python function names.
Args:
prompt_template_config (PromptTemplateConfig): The configuration object for the prompt template.
This should specify the template format as 'jinja2' and include any necessary
configuration details required for rendering the template.
allow_dangerously_set_content (bool = False): Allow content without encoding throughout, this overrides
the same settings in the prompt template config and input variables.
This reverts the behavior to unencoded input.
Raises:
ValueError: If the template format specified in the configuration is not 'jinja2'.
Jinja2TemplateSyntaxError: If there is a syntax error in the Jinja2 template.
"""
_env: ImmutableSandboxedEnvironment | None = PrivateAttr()
@field_validator("prompt_template_config")
@classmethod
def validate_template_format(cls, v: "PromptTemplateConfig") -> "PromptTemplateConfig":
"""Validate the template format."""
if v.template_format != JINJA2_TEMPLATE_FORMAT_NAME:
raise ValueError(f"Invalid prompt template format: {v.template_format}. Expected: jinja2")
return v
def model_post_init(self, _: Any) -> None:
"""Post init model."""
if not self.prompt_template_config.template:
self._env = None
return
self._env = ImmutableSandboxedEnvironment(loader=BaseLoader(), enable_async=True)
async def render(self, kernel: "Kernel", arguments: "KernelArguments | None" = None) -> str:
"""Render the prompt template.
Using the prompt template, replace the variables with their values
and execute the functions replacing their reference with the
function result.
Args:
kernel: The kernel instance
arguments: The kernel arguments
Returns:
The prompt template ready to be used for an AI request
"""
if not self._env:
return ""
if arguments is None:
arguments = KernelArguments()
arguments = self._get_trusted_arguments(arguments)
allow_unsafe_function_output = self._get_allow_dangerously_set_function_output()
helpers: dict[str, Callable[..., Any]] = {}
helpers.update(JINJA2_SYSTEM_HELPERS)
for plugin in kernel.plugins.values():
helpers.update({
function.fully_qualified_name.replace("-", "_"): create_template_helper_from_function(
function,
kernel,
arguments,
self.prompt_template_config.template_format,
allow_unsafe_function_output,
enable_async=True,
)
for function in plugin
})
if self.prompt_template_config.template is None:
raise Jinja2TemplateRenderException("Error rendering template, template is None")
try:
template = self._env.from_string(self.prompt_template_config.template, globals=helpers)
return await template.render_async(**arguments)
except TemplateError as exc:
logger.error(
f"Error rendering prompt template: {self.prompt_template_config.template} with arguments: {arguments}"
)
raise Jinja2TemplateRenderException(
f"Error rendering prompt template: {self.prompt_template_config.template} with "
f"arguments: {arguments}: error: {exc}"
) from exc
@@ -0,0 +1,151 @@
# Copyright (c) Microsoft. All rights reserved.
import logging
from html import escape
from typing import TYPE_CHECKING, Any
from pydantic import PrivateAttr, field_validator
from semantic_kernel.exceptions import TemplateRenderException
from semantic_kernel.functions.kernel_arguments import KernelArguments
from semantic_kernel.prompt_template.const import KERNEL_TEMPLATE_FORMAT_NAME
from semantic_kernel.prompt_template.input_variable import InputVariable
from semantic_kernel.prompt_template.prompt_template_base import PromptTemplateBase
from semantic_kernel.template_engine.blocks.block import Block
from semantic_kernel.template_engine.blocks.code_block import CodeBlock
from semantic_kernel.template_engine.blocks.named_arg_block import NamedArgBlock
from semantic_kernel.template_engine.blocks.var_block import VarBlock
from semantic_kernel.template_engine.template_tokenizer import TemplateTokenizer
if TYPE_CHECKING:
from semantic_kernel.kernel import Kernel
from semantic_kernel.prompt_template.prompt_template_config import PromptTemplateConfig
logger: logging.Logger = logging.getLogger(__name__)
class KernelPromptTemplate(PromptTemplateBase):
"""Create a Kernel prompt template."""
_blocks: list[Block] = PrivateAttr(default_factory=list)
@field_validator("prompt_template_config")
@classmethod
def validate_template_format(cls, v: "PromptTemplateConfig") -> "PromptTemplateConfig":
"""Validate the template format."""
if v.template_format != KERNEL_TEMPLATE_FORMAT_NAME:
raise ValueError(f"Invalid prompt template format: {v.template_format}. Expected: semantic-kernel")
return v
def model_post_init(self, _: Any) -> None:
"""Post init model."""
self._blocks = self.extract_blocks()
# Add all of the existing input variables to our known set. We'll avoid adding any
# dynamically discovered input variables with the same name.
seen = {iv.name.lower() for iv in self.prompt_template_config.input_variables}
# Enumerate every block in the template, adding any variables that are referenced.
for block in self._blocks:
if isinstance(block, VarBlock):
# Add all variables from variable blocks, e.g. "{{$a}}".
self._add_if_missing(block.name, seen)
continue
if isinstance(block, CodeBlock):
for sub_block in block.tokens:
if isinstance(sub_block, VarBlock):
# Add all variables from code blocks, e.g. "{{p.bar $b}}".
self._add_if_missing(sub_block.name, seen)
continue
if isinstance(sub_block, NamedArgBlock) and sub_block.variable:
# Add all variables from named arguments, e.g. "{{p.bar b = $b}}".
# represents a named argument for a function call.
# For example, in the template {{ MyPlugin.MyFunction var1=$boo }}, var1=$boo
# is a named arg block.
self._add_if_missing(sub_block.variable.name, seen)
def extract_blocks(self) -> list[Block]:
"""Given the prompt template, extract all the blocks (text, variables, function calls)."""
logger.debug(f"Extracting blocks from template: {self.prompt_template_config.template}")
if not self.prompt_template_config.template:
return []
return TemplateTokenizer.tokenize(self.prompt_template_config.template)
def _add_if_missing(self, variable_name: str, seen: set):
# Convert variable_name to lower case to handle case-insensitivity
if variable_name and variable_name.lower() not in seen:
seen.add(variable_name.lower())
self.prompt_template_config.input_variables.append(InputVariable(name=variable_name))
async def render(self, kernel: "Kernel", arguments: "KernelArguments | None" = None) -> str:
"""Render the prompt template.
Using the prompt template, replace the variables with their values
and execute the functions replacing their reference with the
function result.
Args:
kernel ("Kernel"): The kernel to use for functions.
arguments ("KernelArguments | None"): The arguments to use for rendering. (Default value = None)
Returns:
str: The prompt template ready to be used for an AI request
"""
return await self.render_blocks(self._blocks, kernel, arguments)
async def render_blocks(
self, blocks: list[Block], kernel: "Kernel", arguments: "KernelArguments | None" = None
) -> str:
"""Given a list of blocks render each block and compose the final result.
Args:
blocks (list[Block]): Template blocks generated by ExtractBlocks
kernel ("Kernel"): The kernel to use for functions
arguments ("KernelArguments | None"): The arguments to use for rendering (Default value = None)
Returns:
str: The prompt template ready to be used for an AI request
"""
from semantic_kernel.template_engine.protocols.code_renderer import CodeRenderer
from semantic_kernel.template_engine.protocols.text_renderer import TextRenderer
logger.debug(f"Rendering list of {len(blocks)} blocks")
rendered_blocks: list[str] = []
arguments = self._get_trusted_arguments(arguments or KernelArguments())
allow_unsafe_function_output = self._get_allow_dangerously_set_function_output()
for block in blocks:
if isinstance(block, TextRenderer):
rendered_blocks.append(block.render(kernel, arguments))
continue
if isinstance(block, CodeRenderer):
try:
rendered = await block.render_code(kernel, arguments)
except Exception as exc:
logger.error(f"Error rendering code block: {exc}")
raise TemplateRenderException(f"Error rendering code block: {exc}") from exc
rendered_blocks.append(rendered if allow_unsafe_function_output else escape(rendered))
prompt = "".join(rendered_blocks)
logger.debug(f"Rendered prompt: {prompt}")
return prompt
@staticmethod
def quick_render(template: str, arguments: dict[str, Any]) -> str:
"""Quick render a Kernel prompt template, only supports text and variable blocks.
Args:
template: The template to render
arguments: The arguments to use for rendering
Returns:
str: The prompt template ready to be used for an AI request
"""
from semantic_kernel import Kernel
from semantic_kernel.template_engine.protocols.code_renderer import CodeRenderer
blocks = TemplateTokenizer.tokenize(template)
if any(isinstance(block, CodeRenderer) for block in blocks):
raise ValueError("Quick render does not support code blocks.")
kernel = Kernel()
return "".join([block.render(kernel, arguments) for block in blocks]) # type: ignore
@@ -0,0 +1,126 @@
# Copyright (c) Microsoft. All rights reserved.
from abc import ABC, abstractmethod
from html import escape
from typing import TYPE_CHECKING, Any
from semantic_kernel.kernel_pydantic import KernelBaseModel
from semantic_kernel.prompt_template.prompt_template_config import PromptTemplateConfig
if TYPE_CHECKING:
from semantic_kernel.functions.kernel_arguments import KernelArguments
from semantic_kernel.kernel import Kernel
class PromptTemplateBase(KernelBaseModel, ABC):
"""Base class for prompt templates."""
prompt_template_config: PromptTemplateConfig
allow_dangerously_set_content: bool = False
@abstractmethod
async def render(self, kernel: "Kernel", arguments: "KernelArguments | None" = None) -> str:
"""Render the prompt template."""
pass
def _get_trusted_arguments(
self,
arguments: "KernelArguments",
) -> "KernelArguments":
"""Get the trusted arguments.
If the prompt template allows unsafe content, then we do not encode the arguments.
Otherwise, each argument is checked against the input variables to see if it allowed to be unencoded.
For string arguments, applies HTML encoding. For complex types, throws an exception unless
allow_dangerously_set_content is set to true.
Args:
arguments: The kernel arguments
"""
if self.allow_dangerously_set_content:
return arguments
from semantic_kernel.functions.kernel_arguments import KernelArguments
new_args = KernelArguments(settings=arguments.execution_settings)
for name, value in arguments.items():
new_args[name] = self._get_encoded_value_or_default(name, value)
return new_args
def _get_allow_dangerously_set_function_output(self) -> bool:
"""Get the allow_dangerously_set_content flag.
If the prompt template allows unsafe content, then we do not encode the function output,
unless explicitly allowed by the prompt template config
"""
allow_dangerously_set_content = self.allow_dangerously_set_content
if self.prompt_template_config.allow_dangerously_set_content:
allow_dangerously_set_content = True
return allow_dangerously_set_content
def _get_encoded_value_or_default(self, name: str, value: Any) -> Any:
"""Encode argument value if necessary, or throw an exception if encoding is not supported.
Args:
name: The name of the property/argument.
value: The value of the property/argument.
Returns:
The encoded value or the original value if encoding is not needed.
Raises:
NotImplementedError: If the value is a complex type and allow_dangerously_set_content is False.
"""
if self.allow_dangerously_set_content or self.prompt_template_config.allow_dangerously_set_content:
return value
# Check if this variable allows dangerous content
for variable in self.prompt_template_config.input_variables:
if variable.name == name and variable.allow_dangerously_set_content:
return value
if isinstance(value, str):
return escape(value)
if self._is_safe_type(value):
return value
# For complex types, throw an exception if dangerous content is not allowed
raise NotImplementedError(
f"Argument '{name}' has a value that doesn't support automatic encoding. "
f"Set allow_dangerously_set_content to 'True' for this argument and implement custom encoding, "
"or provide the value as a string."
)
def _is_safe_type(self, value: Any) -> bool:
"""Determine if a type is considered safe and doesn't require encoding.
Args:
value: The value to check.
Returns:
True if the type is safe, False otherwise.
"""
from datetime import datetime, timedelta
from enum import Enum
from uuid import UUID
# Check for primitive types
if isinstance(value, (int, float, bool, bytes)):
return True
# Check for date/time types
if isinstance(value, (datetime, timedelta)):
return True
# Check for UUID
if isinstance(value, UUID):
return True
# Check for enums
if isinstance(value, Enum):
return True
# Check for None
return value is None
@@ -0,0 +1,138 @@
# Copyright (c) Microsoft. All rights reserved.
import logging
from collections.abc import MutableMapping, MutableSequence, Sequence
from typing import TypeVar
from pydantic import Field, field_validator, model_validator
from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings
from semantic_kernel.const import DEFAULT_SERVICE_NAME
from semantic_kernel.functions.kernel_parameter_metadata import KernelParameterMetadata
from semantic_kernel.kernel_pydantic import KernelBaseModel
from semantic_kernel.prompt_template.const import KERNEL_TEMPLATE_FORMAT_NAME, TEMPLATE_FORMAT_TYPES
from semantic_kernel.prompt_template.input_variable import InputVariable
PromptExecutionSettingsT = TypeVar("PromptExecutionSettingsT", bound=PromptExecutionSettings)
logger: logging.Logger = logging.getLogger(__name__)
_T = TypeVar("_T", bound="PromptTemplateConfig")
class PromptTemplateConfig(KernelBaseModel):
"""Configuration for a prompt template.
Args:
name: The name of the prompt template.
description: The description of the prompt template.
template: The template for the prompt.
template_format: The format of the template, should be 'semantic-kernel', 'jinja2' or 'handlebars'.
input_variables: The input variables for the prompt.
allow_dangerously_set_content (bool = False): Allow content without encoding throughout, this overrides
the same settings in the prompt template config and input variables.
This reverts the behavior to unencoded input.
execution_settings: The execution settings for the prompt.
"""
name: str = ""
description: str | None = ""
template: str | None = None
template_format: TEMPLATE_FORMAT_TYPES = KERNEL_TEMPLATE_FORMAT_NAME
input_variables: MutableSequence[InputVariable] = Field(default_factory=list)
allow_dangerously_set_content: bool = False
execution_settings: MutableMapping[str, PromptExecutionSettings] = Field(default_factory=dict)
@model_validator(mode="after")
def check_input_variables(self):
"""Verify that input variable default values are string only."""
for variable in self.input_variables:
if variable.default and not isinstance(variable.default, str):
raise TypeError(f"Default value for input variable {variable.name} must be a string.")
return self
@field_validator("execution_settings", mode="before")
@classmethod
def rewrite_execution_settings(
cls: type[_T],
settings: PromptExecutionSettings
| Sequence[PromptExecutionSettings]
| MutableMapping[str, PromptExecutionSettings]
| None,
) -> MutableMapping[str, PromptExecutionSettings]:
"""Rewrite execution settings to a dictionary."""
if not settings:
return {}
if isinstance(settings, PromptExecutionSettings):
return {settings.service_id or DEFAULT_SERVICE_NAME: settings}
if isinstance(settings, Sequence):
return {s.service_id or DEFAULT_SERVICE_NAME: s for s in settings}
return settings
def add_execution_settings(self, settings: PromptExecutionSettings, overwrite: bool = True) -> None:
"""Add execution settings to the prompt template."""
if settings.service_id in self.execution_settings and not overwrite:
return
self.execution_settings[settings.service_id or DEFAULT_SERVICE_NAME] = settings
logger.warning("Execution settings already exist and overwrite is set to False")
def get_kernel_parameter_metadata(self) -> Sequence[KernelParameterMetadata]:
"""Get the kernel parameter metadata for the input variables."""
return [
KernelParameterMetadata(
name=variable.name,
description=variable.description,
default_value=variable.default,
type_=variable.json_schema, # TODO (moonbox3): update to handle complex JSON schemas # type: ignore
is_required=variable.is_required,
)
for variable in self.input_variables
]
@classmethod
def from_json(cls: type[_T], json_str: str) -> _T:
"""Create a PromptTemplateConfig instance from a JSON string."""
if not json_str:
raise ValueError("json_str is empty")
try:
return cls.model_validate_json(json_str)
except Exception as exc:
raise ValueError(
"Unable to deserialize PromptTemplateConfig from the "
f"specified JSON string: {json_str} with exception: {exc}"
) from exc
@classmethod
def restore(
cls: type[_T],
name: str,
description: str,
template: str,
template_format: TEMPLATE_FORMAT_TYPES = KERNEL_TEMPLATE_FORMAT_NAME,
input_variables: MutableSequence[InputVariable] | None = None,
execution_settings: MutableMapping[str, PromptExecutionSettings] | None = None,
allow_dangerously_set_content: bool = False,
) -> _T:
"""Restore a PromptTemplateConfig instance from the specified parameters.
Args:
name: The name of the prompt template.
description: The description of the prompt template.
template: The template for the prompt.
template_format: The format of the template, should be 'semantic-kernel', 'jinja2' or 'handlebars'.
input_variables: The input variables for the prompt.
execution_settings: The execution settings for the prompt.
allow_dangerously_set_content: Allow content without encoding.
Returns:
A new PromptTemplateConfig instance.
"""
return cls(
name=name,
description=description,
template=template,
template_format=template_format,
input_variables=input_variables or [],
execution_settings=execution_settings or {},
allow_dangerously_set_content=allow_dangerously_set_content,
)
@@ -0,0 +1,11 @@
# Copyright (c) Microsoft. All rights reserved.
from semantic_kernel.prompt_template.utils.handlebars_system_helpers import HANDLEBAR_SYSTEM_HELPERS
from semantic_kernel.prompt_template.utils.jinja2_system_helpers import JINJA2_SYSTEM_HELPERS
from semantic_kernel.prompt_template.utils.template_function_helpers import create_template_helper_from_function
__all__ = [
"HANDLEBAR_SYSTEM_HELPERS",
"JINJA2_SYSTEM_HELPERS",
"create_template_helper_from_function",
]
@@ -0,0 +1,186 @@
# Copyright (c) Microsoft. All rights reserved.
import json
import logging
import re
from collections.abc import Callable
from enum import Enum
from xml.etree.ElementTree import Element, SubElement, tostring # nosec B405
logger: logging.Logger = logging.getLogger(__name__)
def _messages(this, options, *args, **kwargs):
from semantic_kernel.contents.chat_history import ChatHistory
if not isinstance(this.context["chat_history"], ChatHistory):
return ""
return this.context["chat_history"].to_prompt()
def _message_to_prompt(this, *args, **kwargs):
from semantic_kernel.contents.chat_message_content import ChatMessageContent
if isinstance(this.context, ChatMessageContent):
return str(this.context.to_prompt())
return str(this.context)
def _message(this, options, *args, **kwargs):
from semantic_kernel.contents.chat_message_content import ChatMessageContent
from semantic_kernel.contents.const import CHAT_MESSAGE_CONTENT_TAG
# When the context is a ChatMessageContent, delegate to to_element() so that
# the XML contract is consistent with the Jinja2 path.
if isinstance(this.context, ChatMessageContent):
message = this.context.to_element()
return tostring(message, encoding="unicode", short_empty_elements=False)
# Fallback: build the element manually from kwargs and block content.
from semantic_kernel.contents.const import TEXT_CONTENT_TAG
message = Element(CHAT_MESSAGE_CONTENT_TAG)
for key, value in kwargs.items():
if isinstance(value, Enum):
value = value.value
if value is not None:
message.set(key, str(value))
try:
content = str(options["fn"](this))
except Exception: # pragma: no cover
content = ""
if content:
text_elem = SubElement(message, TEXT_CONTENT_TAG)
text_elem.text = content
return tostring(message, encoding="unicode", short_empty_elements=False)
def _set(this, *args, **kwargs):
if "name" in kwargs and "value" in kwargs:
this.context[kwargs["name"]] = kwargs["value"]
if len(args) == 2 and isinstance(args[0], str):
this.context[args[0]] = args[1]
return ""
def _get(this, *args, **kwargs):
if len(args) == 0:
return ""
return this.context.get(args[0], "")
def _array(this, *args, **kwargs):
return list(args)
def _range(this, *args, **kwargs):
args = list(args)
for index, arg in enumerate(args):
if not isinstance(arg, int):
try:
args[index] = int(arg)
except ValueError:
args.pop(index)
if len(args) == 1:
return list(range(args[0]))
if len(args) == 2:
return list(range(args[0], args[1]))
if len(args) == 3:
return list(range(args[0], args[1], args[2]))
return []
def _concat(this, *args, **kwargs):
return "".join([str(value) for value in args])
def _or(this, *args, **kwargs):
return any(args)
def _add(this, *args, **kwargs):
return sum([float(value) for value in args])
def _subtract(this, *args, **kwargs):
return float(args[0]) - sum([float(value) for value in args[1:]])
def _equals(this, *args, **kwargs):
return args[0] == args[1]
def _less_than(this, *args, **kwargs):
return float(args[0]) < float(args[1])
def _greater_than(this, *args, **kwargs):
return float(args[0]) > float(args[1])
def _less_than_or_equal(this, *args, **kwargs):
return float(args[0]) <= float(args[1])
def _greater_than_or_equal(this, *args, **kwargs):
return float(args[0]) >= float(args[1])
def _json(this, *args, **kwargs):
if not args:
return ""
return json.dumps(args[0])
def _double_open(this, *args, **kwargs):
return "{{"
def _double_close(this, *args, **kwargs):
return "}}"
def _camel_case(this, *args, **kwargs):
return "".join([word.capitalize() for word in args[0].split("_")])
def _snake_case(this, *args, **kwargs):
arg = args[0]
arg = re.sub(r"([A-Z]+)([A-Z][a-z])", r"\1_\2", arg)
arg = re.sub(r"([a-z\d])([A-Z])", r"\1_\2", arg)
arg = arg.replace("-", "_")
return arg.lower()
HANDLEBAR_SYSTEM_HELPERS: dict[str, Callable] = {
"set": _set,
"get": _get,
"array": _array,
"range": _range,
"concat": _concat,
"or": _or,
"add": _add,
"subtract": _subtract,
"equals": _equals,
"less_than": _less_than,
"lessThan": _less_than,
"greater_than": _greater_than,
"greaterThan": _greater_than,
"less_than_or_equal": _less_than_or_equal,
"lessThanOrEqual": _less_than_or_equal,
"greater_than_or_equal": _greater_than_or_equal,
"greaterThanOrEqual": _greater_than_or_equal,
"json": _json,
"double_open": _double_open,
"doubleOpen": _double_open,
"double_close": _double_close,
"doubleClose": _double_close,
"camel_case": _camel_case,
"camelCase": _camel_case,
"snake_case": _snake_case,
"snakeCase": _snake_case,
"message": _message,
"message_to_prompt": _message_to_prompt,
"messageToPrompt": _message_to_prompt,
"messages": _messages,
}
@@ -0,0 +1,85 @@
# Copyright (c) Microsoft. All rights reserved.
import logging
import re
from collections.abc import Callable
from xml.etree.ElementTree import tostring # nosec B405
logger: logging.Logger = logging.getLogger(__name__)
def _messages(chat_history):
from semantic_kernel.contents.chat_history import ChatHistory
if not isinstance(chat_history, ChatHistory):
return ""
return chat_history.to_prompt()
def _message_to_prompt(context):
from semantic_kernel.contents.chat_message_content import ChatMessageContent
if isinstance(context, ChatMessageContent):
return str(context.to_prompt())
return str(context)
def _message(item):
message = item.to_element()
return tostring(message, encoding="unicode", short_empty_elements=False)
# Wrap the _get function to safely handle calls without arguments
def _safe_get_wrapper(context=None, name=None, default=""):
if context is None or name is None:
return default
return _get(context, name, default)
def _get(context, name, default=""):
"""Retrieves a value from the context, with a default if not found."""
return context.get(name, default)
def _double_open():
"""Returns the string representing double open braces."""
return "{{"
def _double_close():
"""Returns the string representing double close braces."""
return "}}"
def _array(*args, **kwargs):
return list(args)
def _camel_case(*args, **kwargs):
return "".join([word.capitalize() for word in args[0].split("_")])
def _snake_case(*args, **kwargs):
arg = args[0]
arg = re.sub(r"([A-Z]+)([A-Z][a-z])", r"\1_\2", arg)
arg = re.sub(r"([a-z\d])([A-Z])", r"\1_\2", arg)
arg = arg.replace("-", "_")
return arg.lower()
JINJA2_SYSTEM_HELPERS: dict[str, Callable] = {
"get": _safe_get_wrapper,
"double_open": _double_open,
"doubleOpen": _double_open,
"double_close": _double_close,
"doubleClose": _double_close,
"message": _message,
"message_to_prompt": _message_to_prompt,
"messages": _messages,
"messageToPrompt": _message_to_prompt,
"array": _array,
"camel_case": _camel_case,
"camelCase": _camel_case,
"snake_case": _snake_case,
"snakeCase": _snake_case,
}
@@ -0,0 +1,141 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import logging
from collections.abc import Callable
from html import escape
from typing import TYPE_CHECKING, Any
import nest_asyncio
from semantic_kernel.functions.kernel_arguments import KernelArguments
from semantic_kernel.prompt_template.const import (
HANDLEBARS_TEMPLATE_FORMAT_NAME,
JINJA2_TEMPLATE_FORMAT_NAME,
TEMPLATE_FORMAT_TYPES,
)
if TYPE_CHECKING:
from semantic_kernel.functions.kernel_function import KernelFunction
from semantic_kernel.kernel import Kernel
logger: logging.Logger = logging.getLogger(__name__)
def create_template_helper_from_function(
function: "KernelFunction",
kernel: "Kernel",
base_arguments: "KernelArguments",
template_format: TEMPLATE_FORMAT_TYPES,
allow_dangerously_set_content: bool = False,
enable_async: bool = False,
) -> Callable[..., Any]:
"""Create a helper function for both the Handlebars and Jinja2 templating engines from a kernel function.
Args:
function (KernelFunction): The kernel function to create a helper for.
kernel (Kernel): The kernel to use for invoking the function.
base_arguments (KernelArguments): The base arguments to use when invoking the function.
template_format (TEMPLATE_FORMAT_TYPES): The template format to create the helper for.
allow_dangerously_set_content (bool, optional): Return the content of the function result
without encoding it or not.
enable_async (bool, optional): Enable async helper function. Defaults to False.
Currently only works for Jinja2 templates.
Returns:
The function with args that are callable by the different templates.
Raises:
ValueError: If the template format is not supported.
"""
if enable_async:
return _create_async_template_helper_from_function(
function=function,
kernel=kernel,
base_arguments=base_arguments,
template_format=template_format,
allow_dangerously_set_content=allow_dangerously_set_content,
)
return _create_sync_template_helper_from_function(
function=function,
kernel=kernel,
base_arguments=base_arguments,
template_format=template_format,
allow_dangerously_set_content=allow_dangerously_set_content,
)
def _create_sync_template_helper_from_function(
function: "KernelFunction",
kernel: "Kernel",
base_arguments: "KernelArguments",
template_format: TEMPLATE_FORMAT_TYPES,
allow_dangerously_set_content: bool = False,
) -> Callable[..., Any]:
"""Create a helper function for both the Handlebars and Jinja2 templating engines from a kernel function."""
if template_format not in [JINJA2_TEMPLATE_FORMAT_NAME, HANDLEBARS_TEMPLATE_FORMAT_NAME]:
raise ValueError(f"Invalid template format: {template_format}")
if not getattr(asyncio, "_nest_patched", False):
nest_asyncio.apply()
def func(*args, **kwargs):
arguments = KernelArguments()
if base_arguments and base_arguments.execution_settings:
arguments.execution_settings = base_arguments.execution_settings # pragma: no cover
arguments.update(base_arguments)
arguments.update(kwargs)
if len(args) > 0 and template_format == HANDLEBARS_TEMPLATE_FORMAT_NAME:
this = args[0]
actual_args = args[1:]
else:
this = None
actual_args = args
if this is not None:
logger.debug(f"Handlebars context with `this`: {this}")
else:
logger.debug("Jinja2 context or Handlebars context without `this`")
logger.debug(
f"Invoking function {function.metadata.fully_qualified_name} "
f"with args: {actual_args} and kwargs: {kwargs} and this: {this}."
)
result = asyncio.run(function.invoke(kernel=kernel, arguments=arguments))
if allow_dangerously_set_content:
return result
return escape(str(result))
return func
def _create_async_template_helper_from_function(
function: "KernelFunction",
kernel: "Kernel",
base_arguments: "KernelArguments",
template_format: TEMPLATE_FORMAT_TYPES,
allow_dangerously_set_content: bool = False,
) -> Callable[..., Any]:
"""Create a async helper function for Jinja2 templating engines from a kernel function."""
if template_format not in [JINJA2_TEMPLATE_FORMAT_NAME]:
raise ValueError(f"Invalid template format: {template_format}")
async def func(*args, **kwargs):
arguments = KernelArguments()
if base_arguments and base_arguments.execution_settings:
arguments.execution_settings = base_arguments.execution_settings # pragma: no cover
arguments.update(base_arguments)
arguments.update(kwargs)
logger.debug(
f"Invoking function {function.metadata.fully_qualified_name} with args: {arguments} and kwargs: {kwargs}."
)
result = await function.invoke(kernel=kernel, arguments=arguments)
if allow_dangerously_set_content:
return result
return escape(str(result))
return func