wehub-resource-sync
142 lines
5.1 KiB
Python
142 lines
5.1 KiB
Python
# 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
|