mirror of
https://github.com/microsoft/semantic-kernel.git
synced 2026-10-12 02:19:06 +00:00
### Motivation and Context We've been working on increasing the test coverage. These tests aren't specific to the CQ spike, but I noticed that there was covering lacking for them, and they were "low hanging fruit." <!-- 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. --> ### Description Increasing the test coverage for some classes that have current test coverage but were missing some line coverage here and there. At this point in time, the test coverage report is showing 95% covered now. <!-- 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 😄 --------- Co-authored-by: Gil LaHaye <[email protected]>
15 KiB
15 KiB
In [ ]:
# Note: if using a Poetry virtual environment, do not run this cell
%pip install semantic-kernel==1.2.0In [ ]:
# Make sure paths are correct for the imports
import os
import sys
notebook_dir = os.path.abspath("")
parent_dir = os.path.dirname(notebook_dir)
grandparent_dir = os.path.dirname(parent_dir)
sys.path.append(grandparent_dir)In [ ]:
from services import Service
from samples.service_settings import ServiceSettings
service_settings = ServiceSettings()
# Select a service to use for this notebook (available services: OpenAI, AzureOpenAI, HuggingFace)
selectedService = (
Service.AzureOpenAI
if service_settings.global_llm_service is None
else Service(service_settings.global_llm_service.lower())
)
print(f"Using service type: {selectedService}")In [ ]:
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import (
AzureChatCompletion,
AzureChatPromptExecutionSettings, # noqa: F401
AzureTextCompletion,
OpenAIChatCompletion,
OpenAIChatPromptExecutionSettings, # noqa: F401
OpenAITextCompletion,
OpenAITextPromptExecutionSettings, # noqa: F401
)
kernel = Kernel()
# Configure Azure LLM service
service_id = None
if selectedService == Service.OpenAI:
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
service_id = "default"
oai_chat_service = OpenAIChatCompletion(
service_id="oai_chat",
)
oai_text_service = OpenAITextCompletion(
service_id="oai_text",
)
elif selectedService == Service.AzureOpenAI:
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
service_id = "default"
aoai_chat_service = AzureChatCompletion(
service_id="aoai_chat",
)
aoai_text_service = AzureTextCompletion(
service_id="aoai_text",
)
# Configure Hugging Face service
if selectedService == Service.HuggingFace:
from semantic_kernel.connectors.ai.hugging_face import ( # noqa: F401
HuggingFacePromptExecutionSettings,
HuggingFaceTextCompletion,
)
hf_text_service = HuggingFaceTextCompletion(service_id="hf_text", ai_model_id="distilgpt2", task="text-generation")In [ ]:
oai_text_prompt_execution_settings = OpenAITextPromptExecutionSettings(
service="oai_text",
extension_data={
"max_tokens": 80,
"temperature": 0.7,
"top_p": 1,
"frequency_penalty": 0.5,
"presence_penalty": 0.5,
"number_of_responses": 3,
},
)In [ ]:
if selectedService == Service.OpenAI:
prompt = "What is the purpose of a rubber duck?"
results = await oai_text_service.get_text_contents(prompt=prompt, settings=oai_text_prompt_execution_settings)
for i, result in enumerate(results):
print(f"Result {i + 1}: {result}")In [ ]:
if selectedService == Service.AzureOpenAI:
prompt = "provide me a list of possible meanings for the acronym 'ORLD'"
results = await aoai_text_service.get_text_contents(prompt=prompt, settings=oai_text_prompt_execution_settings)
for i, result in enumerate(results):
print(f"Result {i + 1}: {result}")In [ ]:
if selectedService == Service.HuggingFace:
hf_prompt_execution_settings = HuggingFacePromptExecutionSettings(
service_id="hf_text",
extension_data={"max_new_tokens": 80, "temperature": 0.7, "top_p": 1, "num_return_sequences": 3},
)In [ ]:
if selectedService == Service.HuggingFace:
prompt = "The purpose of a rubber duck is"
results = await hf_text_service.get_text_contents(prompt=prompt, settings=hf_prompt_execution_settings)
for i, result in enumerate(results):
print(f"Result {i + 1}: {result}")In [ ]:
oai_chat_prompt_execution_settings = OpenAIChatPromptExecutionSettings(
service_id="oai_chat",
max_tokens=80,
temperature=0.7,
top_p=1,
frequency_penalty=0.5,
presence_penalty=0.5,
number_of_responses=3,
)In [ ]:
from semantic_kernel.contents import ChatHistory
if selectedService == Service.OpenAI:
chat = ChatHistory()
chat.add_user_message(
"It's a beautiful day outside, birds are singing, flowers are blooming. On days like these, kids like you..."
)
results = await oai_chat_service.get_chat_message_contents(
chat_history=chat, settings=oai_chat_prompt_execution_settings
)
for i, result in enumerate(results):
print(f"Result {i + 1}: {result!s}")In [ ]:
az_oai_prompt_execution_settings = AzureChatPromptExecutionSettings(
service_id="aoai_chat",
max_tokens=80,
temperature=0.7,
top_p=1,
frequency_penalty=0.5,
presence_penalty=0.5,
number_of_responses=3,
)In [ ]:
if selectedService == Service.AzureOpenAI:
content = (
"Tomorrow is going to be a great day, I can feel it. I'm going to wake up early, go for a run, and then..."
)
chat = ChatHistory()
chat.add_user_message(content)
results = await aoai_chat_service.get_chat_message_contents(
chat_history=chat, settings=az_oai_prompt_execution_settings
)
for i, result in enumerate(results):
print(f"Result {i + 1}: {result!s}")In [ ]:
if selectedService == Service.OpenAI:
import os
import time
from IPython.display import clear_output
# Determine the clear command based on OS
clear_command = "cls" if os.name == "nt" else "clear"
chat = ChatHistory()
chat.add_user_message("what is the purpose of a rubber duck?")
stream = oai_chat_service.get_streaming_chat_message_contents(
chat_history=chat, settings=oai_chat_prompt_execution_settings
)
number_of_responses = oai_chat_prompt_execution_settings.number_of_responses
texts = [""] * number_of_responses
last_clear_time = time.time()
clear_interval = 0.5 # seconds
# Note: there are some quirks with displaying the output, which sometimes flashes and disappears.
# This could be influenced by a few factors specific to Jupyter notebooks and asynchronous processing.
# The following code attempts to buffer the results to avoid the output flashing on/off the screen.
async for results in stream:
current_time = time.time()
# Update texts with new results
for result in results:
texts[result.choice_index] += str(result)
# Clear and display output at intervals
if current_time - last_clear_time > clear_interval:
clear_output(wait=True)
for idx, text in enumerate(texts):
print(f"Result {idx + 1}: {text}")
last_clear_time = current_time
print("----------------------------------------")