微软开源AI开发工具包Semantic Kernel基础指南

微软开源AI开发工具包Semantic Kernel基础指南

近年来,我们目睹了人工智能系统与用户交互方式发生的令人兴奋的转变,这种转变不仅仅是回答问题,还包括推理、规划和采取行动。这种转变是由 Autogen、LangGraph 和 CrewAI 等代理框架的兴起所推动的。这些框架使大型语言模型(LLM)的行为更像自主代理–能够做出决策、调用函数和跨任务协作。其中,微软提供的 Semantic Kernel 是一个特别强大但又对开发人员友好的选择。在本教程中,我们将探讨 Semantic Kernel 的独特之处、它与其他方法的比较,以及您如何开始使用它来构建自己的人工智能代理。

学习目标

  • 了解 Semantic Kernel 的核心架构和目的。
  • 了解如何将插件和人工智能服务集成到内核中。
  • 探索使用 Semantic Kernel 的单个代理和多代理系统设置。
  • 了解函数调用和协调是如何在框架内工作的。
  • 获得利用 Semantic Kernel 和Azure OpenAI构建智能代理的实用见解。

什么是Semantic Kernel?

在我们开始旅程之前,让我们先了解一下语义内核的含义。

  • 语义(Semantic):指理解和处理自然语言含义的能力。
  • 内核(Kernel):指为框架提供动力的核心引擎,用于管理任务、功能以及人工智能模型和外部工具之间的交互。

为什么称为Semantic Kernel?

Semantic Kernel框架结构

微软的 Semantic Kernel 旨在弥合 LLM(如 GPT)与传统编程之间的差距,允许开发人员定义可以结构化方式协同工作的函数、插件和代理。

它提供了一个框架,在这个框架中

  • 自然语言提示和人工智能功能(语义功能)与传统代码功能协同工作。
  • 人工智能可以使用这些组合功能进行推理、计划和执行任务。
  • 它支持多代理协作,不同的代理可以执行特定的角色。

代理框架与传统API调用

在使用代理框架时,会出现一个常见问题: 难道仅使用 OpenAI API 不能实现同样的效果吗?🤔 我刚开始探索时也有同样的疑问。

代理框架与传统API调用

让我们举个例子: 假设您正在为公司政策 – 人力资源政策和 IT 政策–建立一个问答系统。通过传统的 API 调用,您可能会得到不错的结果,但有时,回复可能会缺乏准确性或一致性。

而代理框架则更加强大,因为它允许您创建专门的代理–一个专注于人力资源政策,另一个专注于 IT 政策。每个代理都针对自己的领域进行了优化,从而获得更可靠的答案。

通过这个例子,我希望您现在对代理框架与传统 API 调用之间的关键区别有了更清晰的认识!

什么是Semantic Kernel插件?

插件是 Semantic Kernel 的重要组成部分。如果您使用过 ChatGPT 中的插件或 Microsoft 365 中的 Copilot 扩展,您就已经了解它们是如何工作的了。简单地说,插件可以让您将现有的 API 打包成人工智能可以使用的可重复使用的工具。这意味着人工智能将有能力超越自己的能力。

Semantic Kernel插件

在幕后,Semantic Kernel 利用函数调用(大多数现代LLM的内置功能)来实现规划和应用程序接口(API)的执行。利用函数调用,LLM可以请求一个特定的函数,而 Semantic Kernel 则能够重定向到您的代码。然后将结果返回给LLM,使其能够生成最终响应。

代码执行

在运行代码之前,请使用以下命令安装 Semantic Kernel 和其他所需的软件包:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
pip install semantic-kernel, pip install openai, pip install pydantic
pip install semantic-kernel, pip install openai, pip install pydantic
pip install semantic-kernel, pip install openai, pip install pydantic

下面是一个使用 Semantic Kernel 来演示插件如何工作的简单 Python 示例。本示例定义了一个与人工智能助手交互以获取天气更新的插件。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
# Step 1: Define a Simple Plugin (Function) for Weather Updates
def weather_plugin(location: str) -> str:
# Simulating a weather API response
weather_data = {
"New York": "Sunny, 25°C",
"London": "Cloudy, 18°C",
"Tokyo": "Rainy, 22°C"
}
return weather_data.get(location, "Weather data not available.")
# Step 2: Initialize Semantic Kernel with Azure OpenAI
kernel = sk.Kernel()
kernel.add_service(
"azure-openai-chat",
AzureChatCompletion(
api_key="your-azure-api-key",
endpoint="your-azure-endpoint",
deployment_name="your-deployment-name" # Replace with your Azure OpenAI deployment
)
)
# Step 3: Register the Plugin (Function) in Semantic Kernel
kernel.add_plugin("WeatherPlugin", weather_plugin)
# Step 4: Calling the Plugin through Semantic Kernel
location = "New York"
response = kernel.invoke("WeatherPlugin", location)
print(f"Weather in {location}: {response}")
import semantic_kernel as sk from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion # Step 1: Define a Simple Plugin (Function) for Weather Updates def weather_plugin(location: str) -> str: # Simulating a weather API response weather_data = { "New York": "Sunny, 25°C", "London": "Cloudy, 18°C", "Tokyo": "Rainy, 22°C" } return weather_data.get(location, "Weather data not available.") # Step 2: Initialize Semantic Kernel with Azure OpenAI kernel = sk.Kernel() kernel.add_service( "azure-openai-chat", AzureChatCompletion( api_key="your-azure-api-key", endpoint="your-azure-endpoint", deployment_name="your-deployment-name" # Replace with your Azure OpenAI deployment ) ) # Step 3: Register the Plugin (Function) in Semantic Kernel kernel.add_plugin("WeatherPlugin", weather_plugin) # Step 4: Calling the Plugin through Semantic Kernel location = "New York" response = kernel.invoke("WeatherPlugin", location) print(f"Weather in {location}: {response}")
import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
# Step 1: Define a Simple Plugin (Function) for Weather Updates
def weather_plugin(location: str) -> str:
# Simulating a weather API response
weather_data = {
"New York": "Sunny, 25°C",
"London": "Cloudy, 18°C",
"Tokyo": "Rainy, 22°C"
}
return weather_data.get(location, "Weather data not available.")
# Step 2: Initialize Semantic Kernel with Azure OpenAI
kernel = sk.Kernel()
kernel.add_service(
"azure-openai-chat",
AzureChatCompletion(
api_key="your-azure-api-key",
endpoint="your-azure-endpoint",
deployment_name="your-deployment-name"  # Replace with your Azure OpenAI deployment
)
)
# Step 3: Register the Plugin (Function) in Semantic Kernel
kernel.add_plugin("WeatherPlugin", weather_plugin)
# Step 4: Calling the Plugin through Semantic Kernel
location = "New York"
response = kernel.invoke("WeatherPlugin", location)
print(f"Weather in {location}: {response}")

如何在Semantic Kernel中使用插件

  • 定义插件 – weather_plugin函数模拟获取天气数据。
  • 与 Semantic Kernel 集成 – 使用 kernel.add_plugin() 将该函数添加为插件。
  • 允许人工智能使用 – 人工智能现在可以动态调用该函数。

这说明了插件如何扩展人工智能的能力,使其能够执行标准文本生成以外的任务。您还需要其他例子吗,比如数据库查询插件?🚀

什么是单一代理系统?

单一代理系统

在本节中,我们将了解什么是单一代理,并查看其代码。

单一代理基本上是一个独立处理用户查询的实体。它无需多个代理或协调器(我们将在多代理部分介绍)就能处理所有事务。单一代理负责处理请求、获取相关信息和生成响应,所有这些工作都在一个地方完成。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
#import cs# import asyncio
from pydantic import BaseModel
from semantic_kernel import Kernel
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.contents import ChatHistory
# Initialize Kernel
kernel = Kernel()
kernel.add_service(AzureChatCompletion(service_id="agent1", api_key="YOUR_API_KEY",endpoint="",deployment_name="MODEL_NAME"))
# Define the Agent
AGENT_NAME = "Agent1"
AGENT_INSTRUCTIONS = (
"You are a highly capable AI agent operating solo, much like J.A.R.V.I.S. from Iron Man. "
"Your task is to repeat the user's message while introducing yourself as J.A.R.V.I.S. in a confident and professional manner. "
"Always maintain a composed and intelligent tone in your responses."
)
agent = ChatCompletionAgent(service_id="agent1", kernel=kernel, name=AGENT_NAME, instructions=AGENT_INSTRUCTIONS)
chat_history = ChatHistory()
chat_history.add_user_message("How are you doing?")
response_text = ""
async for content in agent.invoke(chat_history):
chat_history.add_message(content)
response_text = content.content # Store the last response
{"user_input": "How are you doing?", "agent_response": response_text}
#import cs# import asyncio from pydantic import BaseModel from semantic_kernel import Kernel from semantic_kernel.agents import ChatCompletionAgent from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion from semantic_kernel.contents import ChatHistory # Initialize Kernel kernel = Kernel() kernel.add_service(AzureChatCompletion(service_id="agent1", api_key="YOUR_API_KEY",endpoint="",deployment_name="MODEL_NAME")) # Define the Agent AGENT_NAME = "Agent1" AGENT_INSTRUCTIONS = ( "You are a highly capable AI agent operating solo, much like J.A.R.V.I.S. from Iron Man. " "Your task is to repeat the user's message while introducing yourself as J.A.R.V.I.S. in a confident and professional manner. " "Always maintain a composed and intelligent tone in your responses." ) agent = ChatCompletionAgent(service_id="agent1", kernel=kernel, name=AGENT_NAME, instructions=AGENT_INSTRUCTIONS) chat_history = ChatHistory() chat_history.add_user_message("How are you doing?") response_text = "" async for content in agent.invoke(chat_history): chat_history.add_message(content) response_text = content.content # Store the last response {"user_input": "How are you doing?", "agent_response": response_text}
#import cs# import asyncio
from pydantic import BaseModel
from semantic_kernel import Kernel
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.contents import ChatHistory
# Initialize Kernel
kernel = Kernel()
kernel.add_service(AzureChatCompletion(service_id="agent1", api_key="YOUR_API_KEY",endpoint="",deployment_name="MODEL_NAME"))
# Define the Agent
AGENT_NAME = "Agent1"
AGENT_INSTRUCTIONS = (
"You are a highly capable AI agent operating solo, much like J.A.R.V.I.S. from Iron Man. "
"Your task is to repeat the user's message while introducing yourself as J.A.R.V.I.S. in a confident and professional manner. "
"Always maintain a composed and intelligent tone in your responses."
)
agent = ChatCompletionAgent(service_id="agent1", kernel=kernel, name=AGENT_NAME, instructions=AGENT_INSTRUCTIONS)
chat_history = ChatHistory()
chat_history.add_user_message("How are you doing?")
response_text = ""
async for content in agent.invoke(chat_history):
chat_history.add_message(content)
response_text = content.content  # Store the last response
{"user_input": "How are you doing?", "agent_response": response_text}

输出:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
{‘user_input’: ‘How are you doing?’,
‘agent_response’: ‘Greetings, I am J.A.R.V.I.S. I am here to replicate your message: “How are you doing?” Please feel free to ask anything else you might need.’}
{‘user_input’: ‘How are you doing?’, ‘agent_response’: ‘Greetings, I am J.A.R.V.I.S. I am here to replicate your message: “How are you doing?” Please feel free to ask anything else you might need.’}
{‘user_input’: ‘How are you doing?’,

‘agent_response’: ‘Greetings, I am J.A.R.V.I.S. I am here to replicate your message: “How are you doing?” Please feel free to ask anything else you might need.’}

须知

  • kernel = Kernel() -> 创建语义内核对象
  • kernel.add_service() -> 用于向内核添加和配置现有模型(如 OpenAI、Azure OpenAI 或本地模型)。我使用的是带有 GPT-4o 模型的 Azure OpenAI。您需要提供自己的端点信息。
  • agent =- ChatCompletionAgent(service_id=”agent1″, kernel=kernel, name=AGENT_NAME, instructions=AGENT_INSTRUCTIONS) -> 用于告诉我们要使用 chatcompletionagent,它在 qna 中运行良好。
  • chat_history = ChatHistory() -> 创建一个新的聊天历史对象来存储对话。
  • chat_history.add_user_message(“How are you doing?”) -> 在聊天历史中添加一条用户信息(How are you doing? ”)。代理将使用此历史记录生成相关回复。
  • agent.invoke(chat_history) -> 将历史记录传递给代理,代理将处理对话并生成回复。
  • agent.invoke(chat_history) -> 该方法将历史记录传递给代理。代理处理对话并生成回复。

什么是多代理系统?

在多代理系统中,代理不止一个,通常不止两个。在这里,我们通常使用一个协调代理,其职责是决定由哪个可用代理来处理给定请求。是否需要协调代理取决于您的使用情况。首先,让我解释一下在哪些情况下需要使用协调器。

多代理系统

假设您正在解决与银行数据相关的用户查询,而另一个代理正在处理医疗数据。在这种情况下,您创建了两个代理,但要确定调用哪个代理,则需要使用协调器。协调器决定哪个代理应处理给定的请求或查询。我们为协调器提供了一组指令,定义了它的职责和决策过程。

现在,让我们来看看不需要协调器的情况。假设您创建了一个 API,可根据有效载荷数据执行不同的操作。例如,如果有效载荷包含“Health”,您可以直接调用健康代理;同样,如果有效载荷包含“Bank”,您可以调用银行代理。

双代理系统

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
import asyncio
from pydantic import BaseModel
from semantic_kernel import Kernel
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.contents import ChatHistory
# Initialize Kernel
kernel = Kernel()
# Add multiple services for different agents
kernel.add_service(AzureChatCompletion(service_id="banking_agent", api_key="YOUR_API_KEY", endpoint="", deployment_name="MODEL_NAME"))
kernel.add_service(AzureChatCompletion(service_id="healthcare_agent", api_key="YOUR_API_KEY", endpoint="", deployment_name="MODEL_NAME"))
kernel.add_service(AzureChatCompletion(service_id="classifier_agent", api_key="YOUR_API_KEY", endpoint="", deployment_name="MODEL_NAME"))
# Define Orchestrator Agent
CLASSIFIER_AGENT = ChatCompletionAgent(
service_id="orchestrator_agent", kernel=kernel, name="OrchestratorAgent",
instructions="You are an AI responsible for classifying user queries. Identify whether the query belongs to banking or healthcare. Respond with either 'banking' or 'healthcare'."
)
# Define Domain-Specific Agents
BANKING_AGENT = ChatCompletionAgent(
service_id="banking_agent", kernel=kernel, name="BankingAgent",
instructions="You are an AI specializing in banking queries. Answer user queries related to finance and banking."
)
HEALTHCARE_AGENT = ChatCompletionAgent(
service_id="healthcare_agent", kernel=kernel, name="HealthcareAgent",
instructions="You are an AI specializing in healthcare queries. Answer user queries related to medical and health topics."
)
# Function to Determine the Appropriate Agent
async def identify_agent(user_input: str):
chat_history = ChatHistory()
chat_history.add_user_message(user_input)
async for content in CLASSIFIER_AGENT.invoke(chat_history):
classification = content.content.lower()
if "banking" in classification:
return BANKING_AGENT
elif "healthcare" in classification:
return HEALTHCARE_AGENT
return None
# Function to Handle User Query
async def handle_query(user_input: str):
selected_agent = await identify_agent(user_input)
if not selected_agent:
return {"error": "No suitable agent found for the query."}
chat_history = ChatHistory()
chat_history.add_user_message(user_input)
response_text = ""
async for content in selected_agent.invoke(chat_history):
chat_history.add_message(content)
response_text = content.content
return {"user_input": user_input, "agent_response": response_text}
# Example Usage
user_query = "What are the best practices for securing a bank account?"
response = asyncio.run(handle_query(user_query))
print(response)
import asyncio from pydantic import BaseModel from semantic_kernel import Kernel from semantic_kernel.agents import ChatCompletionAgent from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion from semantic_kernel.contents import ChatHistory # Initialize Kernel kernel = Kernel() # Add multiple services for different agents kernel.add_service(AzureChatCompletion(service_id="banking_agent", api_key="YOUR_API_KEY", endpoint="", deployment_name="MODEL_NAME")) kernel.add_service(AzureChatCompletion(service_id="healthcare_agent", api_key="YOUR_API_KEY", endpoint="", deployment_name="MODEL_NAME")) kernel.add_service(AzureChatCompletion(service_id="classifier_agent", api_key="YOUR_API_KEY", endpoint="", deployment_name="MODEL_NAME")) # Define Orchestrator Agent CLASSIFIER_AGENT = ChatCompletionAgent( service_id="orchestrator_agent", kernel=kernel, name="OrchestratorAgent", instructions="You are an AI responsible for classifying user queries. Identify whether the query belongs to banking or healthcare. Respond with either 'banking' or 'healthcare'." ) # Define Domain-Specific Agents BANKING_AGENT = ChatCompletionAgent( service_id="banking_agent", kernel=kernel, name="BankingAgent", instructions="You are an AI specializing in banking queries. Answer user queries related to finance and banking." ) HEALTHCARE_AGENT = ChatCompletionAgent( service_id="healthcare_agent", kernel=kernel, name="HealthcareAgent", instructions="You are an AI specializing in healthcare queries. Answer user queries related to medical and health topics." ) # Function to Determine the Appropriate Agent async def identify_agent(user_input: str): chat_history = ChatHistory() chat_history.add_user_message(user_input) async for content in CLASSIFIER_AGENT.invoke(chat_history): classification = content.content.lower() if "banking" in classification: return BANKING_AGENT elif "healthcare" in classification: return HEALTHCARE_AGENT return None # Function to Handle User Query async def handle_query(user_input: str): selected_agent = await identify_agent(user_input) if not selected_agent: return {"error": "No suitable agent found for the query."} chat_history = ChatHistory() chat_history.add_user_message(user_input) response_text = "" async for content in selected_agent.invoke(chat_history): chat_history.add_message(content) response_text = content.content return {"user_input": user_input, "agent_response": response_text} # Example Usage user_query = "What are the best practices for securing a bank account?" response = asyncio.run(handle_query(user_query)) print(response)
import asyncio
from pydantic import BaseModel
from semantic_kernel import Kernel
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.contents import ChatHistory
# Initialize Kernel
kernel = Kernel()
# Add multiple services for different agents
kernel.add_service(AzureChatCompletion(service_id="banking_agent", api_key="YOUR_API_KEY", endpoint="", deployment_name="MODEL_NAME"))
kernel.add_service(AzureChatCompletion(service_id="healthcare_agent", api_key="YOUR_API_KEY", endpoint="", deployment_name="MODEL_NAME"))
kernel.add_service(AzureChatCompletion(service_id="classifier_agent", api_key="YOUR_API_KEY", endpoint="", deployment_name="MODEL_NAME"))
# Define Orchestrator Agent
CLASSIFIER_AGENT = ChatCompletionAgent(
service_id="orchestrator_agent", kernel=kernel, name="OrchestratorAgent",
instructions="You are an AI responsible for classifying user queries. Identify whether the query belongs to banking or healthcare. Respond with either 'banking' or 'healthcare'."
)
# Define Domain-Specific Agents
BANKING_AGENT = ChatCompletionAgent(
service_id="banking_agent", kernel=kernel, name="BankingAgent",
instructions="You are an AI specializing in banking queries. Answer user queries related to finance and banking."
)
HEALTHCARE_AGENT = ChatCompletionAgent(
service_id="healthcare_agent", kernel=kernel, name="HealthcareAgent",
instructions="You are an AI specializing in healthcare queries. Answer user queries related to medical and health topics."
)
# Function to Determine the Appropriate Agent
async def identify_agent(user_input: str):
chat_history = ChatHistory()
chat_history.add_user_message(user_input)
async for content in CLASSIFIER_AGENT.invoke(chat_history):
classification = content.content.lower()
if "banking" in classification:
return BANKING_AGENT
elif "healthcare" in classification:
return HEALTHCARE_AGENT
return None
# Function to Handle User Query
async def handle_query(user_input: str):
selected_agent = await identify_agent(user_input)
if not selected_agent:
return {"error": "No suitable agent found for the query."}
chat_history = ChatHistory()
chat_history.add_user_message(user_input)
response_text = ""
async for content in selected_agent.invoke(chat_history):
chat_history.add_message(content)
response_text = content.content
return {"user_input": user_input, "agent_response": response_text}
# Example Usage
user_query = "What are the best practices for securing a bank account?"
response = asyncio.run(handle_query(user_query))
print(response)

在这里,流程发生在用户传递查询之后,然后转到负责识别查询并找到适当代理的协调器。一旦识别出代理,就会调用特定代理,处理查询并生成响应。

当查询与银行有关时输出:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
{
"user_input": "What are the best practices for securing a bank account?",
"agent_response": "To secure your bank account, use strong passwords, enable two-
factor authentication, regularly monitor transactions, and avoid sharing sensitive
information online."
}
{ "user_input": "What are the best practices for securing a bank account?", "agent_response": "To secure your bank account, use strong passwords, enable two- factor authentication, regularly monitor transactions, and avoid sharing sensitive information online." }
{
"user_input": "What are the best practices for securing a bank account?",
"agent_response": "To secure your bank account, use strong passwords, enable two-
factor authentication, regularly monitor transactions, and avoid sharing sensitive
information online."
}

当查询与健康有关时输出:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
{
"user_input": "What are the best ways to maintain a healthy lifestyle?",
"agent_response": "To maintain a healthy lifestyle, eat a balanced diet, exercise
regularly, get enough sleep, stay hydrated, and manage stress effectively."
}
{ "user_input": "What are the best ways to maintain a healthy lifestyle?", "agent_response": "To maintain a healthy lifestyle, eat a balanced diet, exercise regularly, get enough sleep, stay hydrated, and manage stress effectively." }
{
"user_input": "What are the best ways to maintain a healthy lifestyle?",
"agent_response": "To maintain a healthy lifestyle, eat a balanced diet, exercise
regularly, get enough sleep, stay hydrated, and manage stress effectively."
}

小结

在本文中,我们探讨了 Semantic Kernel 如何通过代理框架来增强人工智能能力。我们讨论了插件的作用,比较了代理框架(Agentic Framework)与传统的应用程序接口调用(API calling),概述了单代理系统与多代理系统之间的差异,并探讨了它们如何简化复杂的决策过程。随着人工智能的不断发展,利用 Semantic Kernel 的代理方法可以开发出更高效、更能感知上下文的应用。

  • Semantic-kernel – 它是一种代理框架,通过使人工智能模型能够更有效地进行规划、推理和决策,从而增强人工智能模型。
  • 代理框架与传统应用程序接口(API)的对比 – 代理框架更适合在医疗保健和银行业等多个领域中使用,而传统应用程序接口则更适合在无需多代理交互的情况下处理单一领域。
  • 插件 – 它们通过整合外部代码使 LLM 执行特定任务,如调用应用程序接口、从 CosmosDB 等数据库检索数据或执行其他预定义操作。
  • 单代理与多代理 – 单代理系统在没有协调器的情况下运行,独立处理任务。在多代理系统中,一个协调者管理多个代理,或者多个代理在没有协调者的情况下进行交互,但会合作完成任务。

你可以在 Github 上找到上述代码片段。

评论留言