microsoft/semantic-kernel
1
0
Fork 0
mirror of https://github.com/microsoft/semantic-kernel.git synced 2026-10-12 02:19:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Python: mypy coverage enhancement (#6250)

Browse Source 
### Motivation and Context

<!-- Thank you for your contribution to the semantic-kernel repo!
Please help reviewers and future users, providing the following
information:
  1. Why is this change required?
  2. What problem does it solve?
  3. What scenario does it contribute to?
  4. If it fixes an open issue, please link to the issue here.
-->
extend mypy coverage for prompt_templates, template_engine and some
other smaller folders.

still todo: connectors, memory and planner folders.

### Description

<!-- Describe your changes, the overall approach, the underlying design.
These notes will help understanding how your code works. Thanks! -->

### Contribution Checklist

<!-- Before submitting this PR, please make sure: -->

- [x] The code builds clean without any errors or warnings
- [x] The PR follows the [SK Contribution
Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md)
and the [pre-submission formatting
script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts)
raises no violations
- [x] All unit tests pass, and I have added new tests where possible
- [x] I didn't break anyone 😄
This commit is contained in:
Eduard van Valkenburg
2024-05-31 16:37:19 +00:00
committed by GitHub
parent 4508e8d966
commit ba9af0e07b
28 changed files with 182 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/semantic_kernel/const.py
+2
View File
@@ -1,4 +1,6 @@
# Copyright (c) Microsoft. All rights reserved.
from typing import Final
METADATA_EXCEPTION_KEY: Final[str] = "exception"
DEFAULT_SERVICE_NAME: Final[str] = "default"
python/semantic_kernel/core_plugins/conversation_summary_plugin.py
+18 -10
View File
@@ -1,8 +1,11 @@
# Copyright (c) Microsoft. All rights reserved.
from typing import TYPE_CHECKING, Annotated
from semantic_kernel.functions.kernel_function_decorator import kernel_function
if TYPE_CHECKING:
from semantic_kernel.functions.kernel_arguments import KernelArguments
from semantic_kernel.functions.kernel_function import KernelFunction
from semantic_kernel.kernel import Kernel
from semantic_kernel.prompt_template.prompt_template_config import PromptTemplateConfig
@@ -10,8 +13,6 @@ if TYPE_CHECKING:
class ConversationSummaryPlugin:
"""Semantic plugin that enables conversations summarization."""
from semantic_kernel.functions.kernel_function_decorator import kernel_function
# The max tokens to process in a single semantic function call.
_max_tokens = 1024
@@ -30,17 +31,21 @@ class ConversationSummaryPlugin:
) -> None:
"""Initializes a new instance of the ConversationSummaryPlugin class.
:param kernel: The kernel instance.
:param prompt_template_config: The prompt template configuration.
:param return_key: The key to use for the return value.
Args:
kernel (Kernel): The kernel instance to use for the function.
prompt_template_config (PromptTemplateConfig): The configuration to use for functions.
return_key (str): The key to use for the return value.
"""
self.return_key = return_key
self._summarizeConversationFunction = kernel.add_function(
kernel.add_function(
prompt=ConversationSummaryPlugin._summarize_conversation_prompt_template,
plugin_name=ConversationSummaryPlugin.__name__,
function_name="SummarizeConversation",
prompt_template_config=prompt_template_config,
)
self._summarizeConversationFunction: "KernelFunction" = kernel.get_function(
plugin_name=ConversationSummaryPlugin.__name__, function_name="SummarizeConversation"
)
@kernel_function(
description="Given a long conversation transcript, summarize the conversation.",
@@ -56,10 +61,13 @@ class ConversationSummaryPlugin:
]:
"""Given a long conversation transcript, summarize the conversation.
:param input: A long conversation transcript.
:param kernel: The kernel for function execution.
:param arguments: Arguments used by the kernel.
:return: KernelArguments with the summarized conversation result in key self.return_key.
Args:
input (str): A long conversation transcript.
kernel (Kernel): The kernel for function execution.
arguments (KernelArguments): Arguments used by the kernel.
Returns:
KernelArguments with the summarized conversation result in key self.return_key.
"""
from semantic_kernel.text import text_chunker
from semantic_kernel.text.function_extension import aggregate_chunked_results
python/semantic_kernel/core_plugins/sessions_python_tool/sessions_python_plugin.py
+7 -11
View File
@@ -32,9 +32,9 @@ class SessionsPythonTool(KernelBaseModel):
"""A plugin for running Python code in an Azure Container Apps dynamic sessions code interpreter."""
pool_management_endpoint: str
settings: SessionsPythonSettings | None = None
settings: SessionsPythonSettings
auth_callback: Callable[..., Awaitable[Any]]
http_client: httpx.AsyncClient | None = None
http_client: httpx.AsyncClient
def __init__(
self,
@@ -60,12 +60,10 @@ class SessionsPythonTool(KernelBaseModel):
logger.error(f"Failed to load the ACASessionsSettings with message: {e!s}")
raise FunctionExecutionException(f"Failed to load the ACASessionsSettings with message: {e!s}") from e
endpoint = pool_management_endpoint or aca_settings.pool_management_endpoint
super().__init__(
pool_management_endpoint=endpoint,
auth_callback=auth_callback,
pool_management_endpoint=aca_settings.pool_management_endpoint,
settings=settings,
auth_callback=auth_callback,
http_client=http_client,
**kwargs,
)
@@ -100,7 +98,7 @@ class SessionsPythonTool(KernelBaseModel):
Remove whitespace, backtick & python (if llm mistakes python console as terminal).
Args:
code: The query to sanitize
code (str): The query to sanitize
Returns:
str: The sanitized query
"""
@@ -201,7 +199,7 @@ class SessionsPythonTool(KernelBaseModel):
response = await self.http_client.post(
url=f"{self.pool_management_endpoint}python/uploadFile?identifier={self.settings.session_id}",
json={},
files=files,
files=files, # type: ignore
)
response.raise_for_status()
@@ -232,9 +230,7 @@ class SessionsPythonTool(KernelBaseModel):
response_json = response.json()
return [SessionsRemoteFileMetadata.from_dict(entry) for entry in response_json["$values"]]
async def download_file(
self, *, remote_file_path: str, local_file_path: str | None = None
) -> BufferedReader | None:
async def download_file(self, *, remote_file_path: str, local_file_path: str | None = None) -> BytesIO | None:
"""Download a file from the session pool.
Args:
python/semantic_kernel/core_plugins/text_memory_plugin.py
+1 -1
View File
@@ -68,7 +68,7 @@ class TextMemoryPlugin(KernelBaseModel):
logger.warning(f"Memory not found in collection: {collection}")
return ""
return results[0].text if limit == 1 else json.dumps([r.text for r in results])
return results[0].text if limit == 1 else json.dumps([r.text for r in results]) # type: ignore
@kernel_function(
description="Save information to semantic memory",
python/semantic_kernel/core_plugins/web_search_engine_plugin.py
+2 -2
View File
@@ -31,8 +31,8 @@ class WebSearchEnginePlugin:
async def search(
self,
query: Annotated[str, "The search query"],
num_results: Annotated[int | None, "The number of search results to return"] = 1,
offset: Annotated[int | None, "The number of search results to skip"] = 0,
num_results: Annotated[int, "The number of search results to return"] = 1,
offset: Annotated[int, "The number of search results to skip"] = 0,
) -> list[str]:
"""Returns the search results of the query provided."""
return await self._connector.search(query, num_results, offset)
python/semantic_kernel/functions/kernel_arguments.py
+4 -2
View File
@@ -2,6 +2,8 @@
from typing import TYPE_CHECKING, Any
from semantic_kernel.const import DEFAULT_SERVICE_NAME
if TYPE_CHECKING:
from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings
@@ -35,7 +37,7 @@ class KernelArguments(dict):
if isinstance(settings, dict):
settings_dict = settings
elif isinstance(settings, list):
settings_dict = {s.service_id or "default": s for s in settings}
settings_dict = {s.service_id or DEFAULT_SERVICE_NAME: s for s in settings}
else:
settings_dict = {settings.service_id or "default": settings}
settings_dict = {settings.service_id or DEFAULT_SERVICE_NAME: settings}
self.execution_settings: dict[str, "PromptExecutionSettings"] | None = settings_dict
python/semantic_kernel/functions/kernel_function_from_prompt.py
+14 -6
View File
@@ -4,7 +4,7 @@ import logging
import os
from collections.abc import AsyncGenerator
from html import unescape
from typing import Any
from typing import TYPE_CHECKING, Any
import yaml
from pydantic import Field, ValidationError, model_validator
@@ -12,6 +12,7 @@ from pydantic import Field, ValidationError, model_validator
from semantic_kernel.connectors.ai.chat_completion_client_base import ChatCompletionClientBase
from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings
from semantic_kernel.connectors.ai.text_completion_client_base import TextCompletionClientBase
from semantic_kernel.const import DEFAULT_SERVICE_NAME
from semantic_kernel.contents.chat_history import ChatHistory
from semantic_kernel.contents.chat_message_content import ChatMessageContent
from semantic_kernel.contents.text_content import TextContent
@@ -31,6 +32,9 @@ from semantic_kernel.prompt_template.const import KERNEL_TEMPLATE_FORMAT_NAME, T
from semantic_kernel.prompt_template.prompt_template_base import PromptTemplateBase
from semantic_kernel.prompt_template.prompt_template_config import PromptTemplateConfig
if TYPE_CHECKING:
from semantic_kernel.services.ai_service_client_base import AIServiceClientBase
logger: logging.Logger = logging.getLogger(__name__)
PROMPT_FILE_NAME = "skprompt.txt"
@@ -146,10 +150,12 @@ through prompt_template_config or in the prompt_template."
return data
if isinstance(prompt_execution_settings, PromptExecutionSettings):
data["prompt_execution_settings"] = {
prompt_execution_settings.service_id or "default": prompt_execution_settings
prompt_execution_settings.service_id or DEFAULT_SERVICE_NAME: prompt_execution_settings
}
if isinstance(prompt_execution_settings, list):
data["prompt_execution_settings"] = {s.service_id or "default": s for s in prompt_execution_settings}
data["prompt_execution_settings"] = {
s.service_id or DEFAULT_SERVICE_NAME: s for s in prompt_execution_settings
}
return data
async def _invoke_internal(self, context: FunctionInvocationContext) -> None:
@@ -232,7 +238,6 @@ through prompt_template_config or in the prompt_template."
async def _render_prompt(self, context: FunctionInvocationContext) -> PromptRenderingResult:
"""Render the prompt and apply the prompt rendering filters."""
self.update_arguments_with_defaults(context.arguments)
service, execution_settings = context.kernel.select_ai_service(self, context.arguments)
_rebuild_prompt_render_context()
prompt_render_context = PromptRenderContext(function=self, kernel=context.kernel, arguments=context.arguments)
@@ -245,10 +250,13 @@ through prompt_template_config or in the prompt_template."
if prompt_render_context.rendered_prompt is None:
raise PromptRenderingException("Prompt rendering failed, no rendered prompt was returned.")
selected_service: tuple["AIServiceClientBase", PromptExecutionSettings] = context.kernel.select_ai_service(
function=self, arguments=context.arguments
)
return PromptRenderingResult(
rendered_prompt=prompt_render_context.rendered_prompt,
ai_service=service,
execution_settings=execution_settings,
ai_service=selected_service[0],
execution_settings=selected_service[1],
)
async def _inner_render_prompt(self, context: PromptRenderContext) -> None:
python/semantic_kernel/kernel.py
+5 -2
View File
@@ -3,7 +3,7 @@
import logging
from collections.abc import AsyncGenerator, AsyncIterable
from copy import copy
from typing import TYPE_CHECKING, Any, Literal
from typing import TYPE_CHECKING, Any, Literal, TypeVar
from semantic_kernel.const import METADATA_EXCEPTION_KEY
from semantic_kernel.contents.streaming_content_mixin import StreamingContentMixin
@@ -19,14 +19,17 @@ from semantic_kernel.functions.kernel_arguments import KernelArguments
from semantic_kernel.functions.kernel_function_extension import KernelFunctionExtension
from semantic_kernel.functions.kernel_function_from_prompt import KernelFunctionFromPrompt
from semantic_kernel.functions.kernel_plugin import KernelPlugin
from semantic_kernel.kernel_types import AI_SERVICE_CLIENT_TYPE
from semantic_kernel.prompt_template.const import KERNEL_TEMPLATE_FORMAT_NAME
from semantic_kernel.reliability.kernel_reliability_extension import KernelReliabilityExtension
from semantic_kernel.services.ai_service_selector import AIServiceSelector
from semantic_kernel.services.kernel_services_extension import AI_SERVICE_CLIENT_TYPE, KernelServicesExtension
from semantic_kernel.services.kernel_services_extension import KernelServicesExtension
if TYPE_CHECKING:
from semantic_kernel.functions.kernel_function import KernelFunction
T = TypeVar("T")
logger: logging.Logger = logging.getLogger(__name__)
python/semantic_kernel/kernel_types.py
+7
View File
@@ -0,0 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.
from typing import TypeVar
from semantic_kernel.services.ai_service_client_base import AIServiceClientBase
AI_SERVICE_CLIENT_TYPE = TypeVar("AI_SERVICE_CLIENT_TYPE", bound=AIServiceClientBase)
python/semantic_kernel/planners/function_calling_stepwise_planner/function_calling_stepwise_planner.py
+3 -2
View File
@@ -15,6 +15,7 @@ from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.open_ai_pro
from semantic_kernel.connectors.ai.open_ai.services.azure_chat_completion import AzureChatCompletion
from semantic_kernel.connectors.ai.open_ai.services.open_ai_chat_completion import OpenAIChatCompletion
from semantic_kernel.connectors.ai.open_ai.services.utils import kernel_function_metadata_to_openai_tool_format
from semantic_kernel.const import DEFAULT_SERVICE_NAME
from semantic_kernel.contents.chat_history import ChatHistory
from semantic_kernel.contents.function_call_content import FunctionCallContent
from semantic_kernel.contents.function_result_content import FunctionResultContent
@@ -256,8 +257,8 @@ class FunctionCallingStepwisePlanner(KernelBaseModel):
"""
data = yaml.safe_load(self.generate_plan_yaml)
prompt_template_config = PromptTemplateConfig(**data)
if "default" in prompt_template_config.execution_settings:
settings = prompt_template_config.execution_settings.pop("default")
if DEFAULT_SERVICE_NAME in prompt_template_config.execution_settings:
settings = prompt_template_config.execution_settings.pop(DEFAULT_SERVICE_NAME)
prompt_template_config.execution_settings[self.service_id] = settings
return kernel.add_function(
function_name="create_plan",
python/semantic_kernel/planners/sequential_planner/sequential_planner.py
+5 -5
View File
@@ -3,7 +3,7 @@
import os
from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings
from semantic_kernel.const import METADATA_EXCEPTION_KEY
from semantic_kernel.const import DEFAULT_SERVICE_NAME, METADATA_EXCEPTION_KEY
from semantic_kernel.exceptions import PlannerCreatePlanError, PlannerException, PlannerInvalidGoalError
from semantic_kernel.functions.function_result import FunctionResult
from semantic_kernel.functions.kernel_arguments import KernelArguments
@@ -67,8 +67,8 @@ class SequentialPlanner:
prompt_template = prompt or read_file(PROMPT_TEMPLATE_FILE_PATH)
if service_id in prompt_config.execution_settings:
prompt_config.execution_settings[service_id].extension_data["max_tokens"] = self.config.max_tokens
elif "default" in prompt_config.execution_settings:
prompt_config.execution_settings["default"].extension_data["max_tokens"] = self.config.max_tokens
elif DEFAULT_SERVICE_NAME in prompt_config.execution_settings:
prompt_config.execution_settings[DEFAULT_SERVICE_NAME].extension_data["max_tokens"] = self.config.max_tokens
else:
prompt_config.execution_settings[service_id] = PromptExecutionSettings(
service_id=service_id, max_tokens=self.config.max_tokens
@@ -79,9 +79,9 @@ class SequentialPlanner:
if (
service_id
and service_id not in prompt_config.execution_settings
and "default" in prompt_config.execution_settings
and DEFAULT_SERVICE_NAME in prompt_config.execution_settings
):
settings = prompt_config.execution_settings.pop("default")
settings = prompt_config.execution_settings.pop(DEFAULT_SERVICE_NAME)
prompt_config.execution_settings[service_id] = settings
return self._kernel.add_function(
python/semantic_kernel/prompt_template/jinja2_prompt_template.py
+4 -1
View File
@@ -9,6 +9,7 @@ from jinja2.sandbox import ImmutableSandboxedEnvironment
from pydantic import PrivateAttr, field_validator
from semantic_kernel.exceptions import Jinja2TemplateRenderException
from semantic_kernel.exceptions.template_engine_exceptions import TemplateRenderException
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
@@ -47,7 +48,7 @@ class Jinja2PromptTemplate(PromptTemplateBase):
Jinja2TemplateSyntaxError: If there is a syntax error in the Jinja2 template.
"""
_env: ImmutableSandboxedEnvironment = PrivateAttr()
_env: ImmutableSandboxedEnvironment | None = PrivateAttr()
@field_validator("prompt_template_config")
@classmethod
@@ -101,6 +102,8 @@ class Jinja2PromptTemplate(PromptTemplateBase):
}
)
try:
if self.prompt_template_config.template is None:
raise TemplateRenderException("Template is None")
template = self._env.from_string(self.prompt_template_config.template, globals=helpers)
return template.render(**arguments)
python/semantic_kernel/prompt_template/kernel_prompt_template.py
+9 -5
View File
@@ -12,7 +12,9 @@ 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.block_types import BlockTypes
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:
@@ -56,17 +58,17 @@ class KernelPromptTemplate(PromptTemplateBase):
# Enumerate every block in the template, adding any variables that are referenced.
for block in self._blocks:
if block.type == BlockTypes.VARIABLE:
if isinstance(block, VarBlock):
# Add all variables from variable blocks, e.g. "{{$a}}".
self._add_if_missing(block.name, seen)
continue
if block.type == BlockTypes.CODE:
if isinstance(block, CodeBlock):
for sub_block in block.tokens:
if sub_block.type == BlockTypes.VARIABLE:
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 sub_block.type == BlockTypes.NAMED_ARG and sub_block.variable:
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
@@ -75,6 +77,8 @@ class KernelPromptTemplate(PromptTemplateBase):
def _add_if_missing(self, variable_name: str, seen: set | None = None):
# Convert variable_name to lower case to handle case-insensitivity
if not seen:
seen = set()
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))
python/semantic_kernel/prompt_template/prompt_template_config.py
+4 -3
View File
@@ -5,6 +5,7 @@ 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
@@ -57,16 +58,16 @@ class PromptTemplateConfig(KernelBaseModel):
if not settings:
return {}
if isinstance(settings, PromptExecutionSettings):
return {settings.service_id or "default": settings}
return {settings.service_id or DEFAULT_SERVICE_NAME: settings}
if isinstance(settings, list):
return {s.service_id or "default": s for s in settings}
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"] = settings
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) -> list[KernelParameterMetadata]:
python/semantic_kernel/prompt_template/utils/template_function_helpers.py
+10 -3
View File
@@ -4,12 +4,16 @@ import asyncio
import logging
from collections.abc import Callable
from html import escape
from typing import TYPE_CHECKING, Any, Literal
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
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
@@ -23,10 +27,13 @@ def create_template_helper_from_function(
function: "KernelFunction",
kernel: "Kernel",
base_arguments: "KernelArguments",
template_format: Literal["handlebars", "jinja2"],
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()
python/semantic_kernel/reliability/pass_through_without_retry.py
+1 -1
View File
@@ -25,7 +25,7 @@ class PassThroughWithoutRetry(RetryMechanismBase, KernelBaseModel):
Awaitable[T]: An awaitable that will return the result of the action.
"""
try:
await action()
return action()
except Exception as e:
logger.warning(e, "Error executing action, not retrying")
raise e
python/semantic_kernel/services/ai_service_client_base.py
+1 -1
View File
@@ -28,7 +28,7 @@ class AIServiceClientBase(KernelBaseModel, ABC):
if not self.service_id:
self.service_id = self.ai_model_id
def get_prompt_execution_settings_class(self) -> "PromptExecutionSettings":
def get_prompt_execution_settings_class(self) -> type["PromptExecutionSettings"]:
"""Get the request settings class."""
return PromptExecutionSettings # pragma: no cover
python/semantic_kernel/services/ai_service_selector.py
+10 -7
View File
@@ -2,13 +2,16 @@
from typing import TYPE_CHECKING
from semantic_kernel.const import DEFAULT_SERVICE_NAME
from semantic_kernel.exceptions import KernelServiceNotFoundError
from semantic_kernel.kernel_types import AI_SERVICE_CLIENT_TYPE
if TYPE_CHECKING:
from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings
from semantic_kernel.functions.kernel_arguments import KernelArguments
from semantic_kernel.functions.kernel_function import KernelFunction
from semantic_kernel.kernel import AI_SERVICE_CLIENT_TYPE, Kernel
from semantic_kernel.services.ai_service_client_base import AIServiceClientBase
from semantic_kernel.services.kernel_services_extension import KernelServicesExtension
class AIServiceSelector:
@@ -20,11 +23,11 @@ class AIServiceSelector:
def select_ai_service(
self,
kernel: "Kernel",
kernel: "KernelServicesExtension",
function: "KernelFunction",
arguments: "KernelArguments",
type_: type["AI_SERVICE_CLIENT_TYPE"] | None = None,
) -> tuple["AI_SERVICE_CLIENT_TYPE", "PromptExecutionSettings"]:
type_: type[AI_SERVICE_CLIENT_TYPE] | tuple[type[AI_SERVICE_CLIENT_TYPE], ...] | None = None,
) -> tuple["AIServiceClientBase", "PromptExecutionSettings"]:
"""Select an AI Service on a first come, first served basis.
Starts with execution settings in the arguments,
@@ -35,7 +38,7 @@ class AIServiceSelector:
from semantic_kernel.connectors.ai.chat_completion_client_base import ChatCompletionClientBase
from semantic_kernel.connectors.ai.text_completion_client_base import TextCompletionClientBase
type_ = (TextCompletionClientBase, ChatCompletionClientBase)
type_ = (TextCompletionClientBase, ChatCompletionClientBase) # type: ignore
execution_settings_dict = arguments.execution_settings or {}
if func_exec_settings := getattr(function, "prompt_execution_settings", None):
@@ -45,13 +48,13 @@ class AIServiceSelector:
if not execution_settings_dict:
from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings
execution_settings_dict = {"default": PromptExecutionSettings()}
execution_settings_dict = {DEFAULT_SERVICE_NAME: PromptExecutionSettings()}
for service_id, settings in execution_settings_dict.items():
try:
service = kernel.get_service(service_id, type=type_)
except KernelServiceNotFoundError:
continue
if service:
if service is not None:
service_settings = service.get_prompt_execution_settings_from_settings(settings)
return service, service_settings
raise KernelServiceNotFoundError("No service found.")
python/semantic_kernel/services/kernel_services_extension.py
+38 -47
View File
@@ -2,32 +2,22 @@
import logging
from abc import ABC
from typing import TYPE_CHECKING, TypeVar, Union
from typing import TYPE_CHECKING
from pydantic import Field, field_validator
from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings
from semantic_kernel.exceptions import (
KernelFunctionAlreadyExistsError,
KernelServiceNotFoundError,
ServiceInvalidTypeError,
)
from semantic_kernel.functions.kernel_arguments import KernelArguments
from semantic_kernel.const import DEFAULT_SERVICE_NAME
from semantic_kernel.exceptions import KernelFunctionAlreadyExistsError, KernelServiceNotFoundError
from semantic_kernel.kernel_pydantic import KernelBaseModel
from semantic_kernel.kernel_types import AI_SERVICE_CLIENT_TYPE
from semantic_kernel.services.ai_service_client_base import AIServiceClientBase
from semantic_kernel.services.ai_service_selector import AIServiceSelector
if TYPE_CHECKING:
from semantic_kernel.connectors.ai.chat_completion_client_base import ChatCompletionClientBase
from semantic_kernel.connectors.ai.embeddings.embedding_generator_base import EmbeddingGeneratorBase
from semantic_kernel.connectors.ai.text_completion_client_base import TextCompletionClientBase
from semantic_kernel.functions.kernel_arguments import KernelArguments
from semantic_kernel.functions.kernel_function import KernelFunction
T = TypeVar("T")
AI_SERVICE_CLIENT_TYPE = TypeVar("AI_SERVICE_CLIENT_TYPE", bound=AIServiceClientBase)
ALL_SERVICE_TYPES = Union["TextCompletionClientBase", "ChatCompletionClientBase", "EmbeddingGeneratorBase"]
logger: logging.Logger = logging.getLogger(__name__)
@@ -48,22 +38,22 @@ class KernelServicesExtension(KernelBaseModel, ABC):
if not services:
return {}
if isinstance(services, AIServiceClientBase):
return {services.service_id if services.service_id else "default": services} # type: ignore
return {services.service_id if services.service_id else DEFAULT_SERVICE_NAME: services} # type: ignore
if isinstance(services, list):
return {s.service_id if s.service_id else "default": s for s in services}
return {s.service_id if s.service_id else DEFAULT_SERVICE_NAME: s for s in services}
return services
def select_ai_service(
self, function: "KernelFunction", arguments: KernelArguments
) -> tuple[ALL_SERVICE_TYPES, PromptExecutionSettings]:
self, function: "KernelFunction", arguments: "KernelArguments"
) -> tuple[AIServiceClientBase, PromptExecutionSettings]:
"""Uses the AI service selector to select a service for the function."""
return self.ai_service_selector.select_ai_service(self, function, arguments)
def get_service(
self,
service_id: str | None = None,
type: type[ALL_SERVICE_TYPES] | None = None,
) -> "AIServiceClientBase":
type: type[AI_SERVICE_CLIENT_TYPE] | tuple[type[AI_SERVICE_CLIENT_TYPE], ...] | None = None,
) -> AIServiceClientBase:
"""Get a service by service_id and type.
Type is optional and when not supplied, no checks are done.
@@ -71,46 +61,47 @@ class KernelServicesExtension(KernelBaseModel, ABC):
TextCompletionClientBase, ChatCompletionClientBase, EmbeddingGeneratorBase
or a subclass of one.
You can also check for multiple types in one go,
by using TextCompletionClientBase | ChatCompletionClientBase.
by using a tuple: (TextCompletionClientBase, ChatCompletionClientBase).
If type and service_id are both None, the first service is returned.
Args:
service_id (str | None): The service id,
if None, the default service is returned or the first service is returned.
type (Type[ALL_SERVICE_TYPES] | None): The type of the service, if None, no checks are done.
type (Type[AI_SERVICE_CLIENT_TYPE] | tuple[type[AI_SERVICE_CLIENT_TYPE], ...] | None):
The type of the service, if None, no checks are done on service type.
Returns:
ALL_SERVICE_TYPES: The service.
AIServiceClientBase: The service, should be a class derived from AIServiceClientBase.
Raises:
ValueError: If no service is found that matches the type.
KernelServiceNotFoundError: If no service is found that matches the type or id.
"""
service: "AIServiceClientBase | None" = None
if not service_id or service_id == "default":
if not type:
if default_service := self.services.get("default"):
return default_service
return next(iter(self.services.values()))
if (default_service := self.services.get("default")) and isinstance(default_service, type):
return default_service
for service in self.services.values():
if isinstance(service, type):
return service
raise KernelServiceNotFoundError(f"No service found of type {type}")
if not (service := self.services.get(service_id)):
raise KernelServiceNotFoundError(f"Service with service_id '{service_id}' does not exist")
if type and not isinstance(service, type):
raise ServiceInvalidTypeError(f"Service with service_id '{service_id}' is not of type {type}")
return service
services = self.get_services_by_type(type)
if not services:
raise KernelServiceNotFoundError(f"No services found of type {type}.")
if not service_id:
service_id = DEFAULT_SERVICE_NAME
def get_services_by_type(self, type: type[ALL_SERVICE_TYPES]) -> dict[str, ALL_SERVICE_TYPES]:
if service_id not in services:
if service_id == DEFAULT_SERVICE_NAME:
return next(iter(services.values()))
raise KernelServiceNotFoundError(
f"Service with service_id '{service_id}' does not exist or has a different type."
)
return services[service_id]
def get_services_by_type(
self, type: type[AI_SERVICE_CLIENT_TYPE] | tuple[type[AI_SERVICE_CLIENT_TYPE], ...] | None
) -> dict[str, AIServiceClientBase]:
"""Get all services of a specific type."""
return {service.service_id: service for service in self.services.values() if isinstance(service, type)} # type: ignore
if type is None:
return self.services
return {service.service_id: service for service in self.services.values() if isinstance(service, type)}
def get_prompt_execution_settings_from_service_id(
self, service_id: str, type: type[ALL_SERVICE_TYPES] | None = None
self, service_id: str, type: type[AI_SERVICE_CLIENT_TYPE] | None = None
) -> PromptExecutionSettings:
"""Get the specific request settings from the service, instantiated with the service_id and ai_model_id."""
service = self.get_service(service_id, type=type)
@@ -128,8 +119,8 @@ class KernelServicesExtension(KernelBaseModel, ABC):
"""
if service.service_id not in self.services or overwrite:
self.services[service.service_id] = service
else:
raise KernelFunctionAlreadyExistsError(f"Service with service_id '{service.service_id}' already exists")
return
raise KernelFunctionAlreadyExistsError(f"Service with service_id '{service.service_id}' already exists")
def remove_service(self, service_id: str) -> None:
"""Delete a single service from the Kernel."""
python/semantic_kernel/template_engine/blocks/code_block.py
+10 -6
View File
@@ -11,6 +11,8 @@ from semantic_kernel.exceptions.kernel_exceptions import KernelFunctionNotFoundE
from semantic_kernel.functions.kernel_function_metadata import KernelFunctionMetadata
from semantic_kernel.template_engine.blocks.block import Block
from semantic_kernel.template_engine.blocks.block_types import BlockTypes
from semantic_kernel.template_engine.blocks.function_id_block import FunctionIdBlock
from semantic_kernel.template_engine.blocks.named_arg_block import NamedArgBlock
from semantic_kernel.template_engine.code_tokenizer import CodeTokenizer
if TYPE_CHECKING:
@@ -107,13 +109,15 @@ these will be ignored."
Otherwise, it is a value or variable and those are then rendered directly.
"""
logger.debug(f"Rendering code: `{self.content}`")
if self.tokens[0].type == BlockTypes.FUNCTION_ID:
if isinstance(self.tokens[0], FunctionIdBlock):
return await self._render_function_call(kernel, arguments)
# validated that if the first token is not a function_id, it is a value or variable
return self.tokens[0].render(kernel, arguments)
return self.tokens[0].render(kernel, arguments) # type: ignore
async def _render_function_call(self, kernel: "Kernel", arguments: "KernelArguments"):
function_block = self.tokens[0]
if not isinstance(self.tokens[0], FunctionIdBlock):
raise CodeBlockRenderException("The first token should be a function_id")
function_block: FunctionIdBlock = self.tokens[0]
try:
function = kernel.get_function(function_block.plugin_name, function_block.function_name)
except (KernelFunctionNotFoundError, KernelPluginNotFoundError) as exc:
@@ -145,10 +149,10 @@ these will be ignored."
)
for index, token in enumerate(self.tokens[1:], start=1):
logger.debug(f"Parsing variable/value: `{self.tokens[1].content}`")
rendered_value = token.render(kernel, arguments)
if token.type != BlockTypes.NAMED_ARG and index == 1:
rendered_value = token.render(kernel, arguments) # type: ignore
if not isinstance(token, NamedArgBlock) and index == 1:
arguments[function_metadata.parameters[0].name] = rendered_value
continue
arguments[token.name] = rendered_value
arguments[token.name] = rendered_value # type: ignore
return arguments
python/semantic_kernel/template_engine/blocks/function_id_block.py
+3 -3
View File
@@ -2,7 +2,7 @@
import logging
from re import compile
from typing import TYPE_CHECKING, Any, ClassVar, Optional
from typing import TYPE_CHECKING, Any, ClassVar
from pydantic import model_validator
@@ -39,7 +39,7 @@ class FunctionIdBlock(Block):
"""
type: ClassVar[BlockTypes] = BlockTypes.FUNCTION_ID
function_name: str | None = ""
function_name: str = ""
plugin_name: str | None = None
@model_validator(mode="before")
@@ -61,6 +61,6 @@ class FunctionIdBlock(Block):
fields["function_name"] = matches.group("function")
return fields
def render(self, *_: tuple["Kernel", Optional["KernelArguments"]]) -> str:
def render(self, *_: "Kernel | KernelArguments | None") -> str:
"""Render the function id block."""
return self.content
python/semantic_kernel/template_engine/blocks/named_arg_block.py
+1 -1
View File
@@ -90,7 +90,7 @@ class NamedArgBlock(Block):
def render(self, kernel: "Kernel", arguments: Optional["KernelArguments"] = None) -> Any:
"""Render the named argument block."""
if self.value:
return self.value.render(kernel, arguments)
return self.value.render()
if arguments is None:
return ""
if self.variable:
python/semantic_kernel/template_engine/blocks/val_block.py
+3 -3
View File
@@ -2,7 +2,7 @@
import logging
from re import S, compile
from typing import TYPE_CHECKING, Any, ClassVar, Optional
from typing import TYPE_CHECKING, Any, ClassVar
from pydantic import model_validator
@@ -69,6 +69,6 @@ class ValBlock(Block):
fields["quote"] = quote
return fields
def render(self, *_: tuple["Kernel", Optional["KernelArguments"]]) -> str:
def render(self, *_: "Kernel | KernelArguments | None") -> str:
"""Render the value block."""
return self.value
return self.value or ""
python/semantic_kernel/template_engine/blocks/var_block.py
+1 -1
View File
@@ -45,7 +45,7 @@ class VarBlock(Block):
"""
type: ClassVar[BlockTypes] = BlockTypes.VARIABLE
name: str | None = ""
name: str = ""
@model_validator(mode="before")
@classmethod
python/semantic_kernel/text/text_chunker.py
+8 -8
View File
@@ -12,7 +12,7 @@ from collections.abc import Callable
NEWLINE = os.linesep
TEXT_SPLIT_OPTIONS = [
TEXT_SPLIT_OPTIONS: list[list[str] | None] = [
["\n", "\r"],
["."],
["?", "!"],
@@ -25,7 +25,7 @@ TEXT_SPLIT_OPTIONS = [
None,
]
MD_SPLIT_OPTIONS = [
MD_SPLIT_OPTIONS: list[list[str] | None] = [
["."],
["?", "!"],
[";"],
@@ -112,8 +112,8 @@ def _split_text_paragraph(text: list[str], max_tokens: int, token_counter: Calla
if not text:
return []
paragraphs = []
current_paragraph = []
paragraphs: list[str] = []
current_paragraph: list[str] = []
for line in text:
num_tokens_line = token_counter(line)
@@ -187,7 +187,7 @@ def _split_text_lines(
def _split_str_lines(
text: str,
max_tokens: int,
separators: list[list[str]],
separators: list[list[str] | None],
trim: bool,
token_counter: Callable = _token_counter,
) -> list[str]:
@@ -196,7 +196,7 @@ def _split_str_lines(
return []
text = text.replace("\r\n", "\n")
lines = []
lines: list[str] = []
was_split = False
for split_option in separators:
if not lines:
@@ -224,7 +224,7 @@ def _split_str_lines(
def _split_str(
text: str,
max_tokens: int,
separators: list[str],
separators: list[str] | None,
trim: bool,
token_counter: Callable = _token_counter,
) -> tuple[list[str], bool]:
@@ -283,7 +283,7 @@ def _split_str(
def _split_list(
text: list[str],
max_tokens: int,
separators: list[str],
separators: list[str] | None,
trim: bool,
token_counter: Callable = _token_counter,
) -> tuple[list[str], bool]:
python/semantic_kernel/utils/experimental_decorator.py
+3 -4
View File
@@ -1,18 +1,17 @@
# Copyright (c) Microsoft. All rights reserved.
import types
from collections.abc import Callable
def experimental_function(func: Callable) -> Callable:
"""Decorator to mark a function as experimental."""
if isinstance(func, types.FunctionType):
if callable(func):
if func.__doc__:
func.__doc__ += "\n\nNote: This function is experimental and may change in the future."
else:
func.__doc__ = "Note: This function is experimental and may change in the future."
func.is_experimental = True
setattr(func, "is_experimental", True)
return func
@@ -25,6 +24,6 @@ def experimental_class(cls: type) -> type:
else:
cls.__doc__ = "Note: This class is experimental and may change in the future."
cls.is_experimental = True
setattr(cls, "is_experimental", True)
return cls