autogen_ext.models.semantic_kernel#
- class SKChatCompletionAdapter(sk_client: ChatCompletionClientBase, kernel: Kernel | 无 = None, prompt_settings: PromptExecutionSettings | 无 = None, model_info: ModelInfo | 无 = None, service_id: str | 无 = None)[源代码]#
-
SKChatCompletionAdapter 是一个适配器,允许将语义内核模型客户端用作 Autogen ChatCompletion 客户端。这使得可以将语义内核连接器(例如 Azure OpenAI、Google Gemini、Ollama 等)无缝集成到依赖 ChatCompletionClient 接口的 Autogen 代理中。
通过利用这个适配器,你可以:
传入一个Kernel和任何支持的Semantic KernelChatCompletionClientBase连接器。
提供工具(通过 Autogen 工具 或 工具Schema)用于在聊天完成期间进行函数调用。
流式响应或在单个请求中检索它们。
- 通过构造函数全局提供提示设置以控制聊天完成行为
或者通过 extra_create_args 字典在每次请求基础上提供。
可以安装的额外功能列表:
semantic-kernel-anthropic: 安装此扩展以使用Anthropic模型。
semantic-kernel-google: 安装此扩展以使用Google Gemini模型。
semantic-kernel-ollama: 安装此扩展以使用Ollama模型。
semantic-kernel-mistralai: 安装此额外组件以使用 MistralAI 模型。
semantic-kernel-aws: 安装此扩展以使用AWS模型。
semantic-kernel-hugging-face: 安装此额外组件以使用Hugging Face模型。
- Parameters:
sk_client (ChatCompletionClientBase) – 要包装的Semantic Kernel客户端(例如,AzureChatCompletion、GoogleAIChatCompletion、OllamaChatCompletion)。
kernel (可选[Kernel]) – 用于执行请求的Semantic Kernel实例。如果未提供,则必须在每次请求的extra_create_args中传递一个。
prompt_settings (可选[PromptExecutionSettings]) – 默认使用的提示执行设置。可以在每个请求中覆盖。
model_info (可选[ModelInfo]) – 关于模型能力的信息。
service_id (可选[str]) – 可选的服务标识符。
示例
具有函数调用功能的Anthropic模型:
pip install "autogen-ext[semantic-kernel-anthropic]"
import asyncio import os from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.ui import Console from autogen_core.models import ModelFamily, UserMessage from autogen_ext.models.semantic_kernel import SKChatCompletionAdapter from semantic_kernel import Kernel from semantic_kernel.connectors.ai.anthropic import AnthropicChatCompletion, AnthropicChatPromptExecutionSettings from semantic_kernel.memory.null_memory import NullMemory async def get_weather(city: str) -> str: """Get the weather for a city.""" return f"The weather in {city} is 75 degrees." async def main() -> None: sk_client = AnthropicChatCompletion( ai_model_id="claude-3-5-sonnet-20241022", api_key=os.environ["ANTHROPIC_API_KEY"], service_id="my-service-id", # Optional; for targeting specific services within Semantic Kernel ) settings = AnthropicChatPromptExecutionSettings( temperature=0.2, ) model_client = SKChatCompletionAdapter( sk_client, kernel=Kernel(memory=NullMemory()), prompt_settings=settings, model_info={ "function_calling": True, "json_output": True, "vision": True, "family": ModelFamily.CLAUDE_3_5_SONNET, }, ) # Call the model directly. response = await model_client.create([UserMessage(content="What is the capital of France?", source="test")]) print(response) # Create an assistant agent with the model client. assistant = AssistantAgent( "assistant", model_client=model_client, system_message="You are a helpful assistant.", tools=[get_weather] ) # Call the assistant with a task. await Console(assistant.run_stream(task="What is the weather in Paris and London?")) asyncio.run(main())
Google Gemini 模型带有函数调用功能:
pip install "autogen-ext[semantic-kernel-google]"
import asyncio import os from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.ui import Console from autogen_core.models import UserMessage, ModelFamily from autogen_ext.models.semantic_kernel import SKChatCompletionAdapter from semantic_kernel import Kernel from semantic_kernel.connectors.ai.google.google_ai import ( GoogleAIChatCompletion, GoogleAIChatPromptExecutionSettings, ) from semantic_kernel.memory.null_memory import NullMemory def get_weather(city: str) -> str: """Get the weather for a city.""" return f"The weather in {city} is 75 degrees." async def main() -> None: sk_client = GoogleAIChatCompletion( gemini_model_id="gemini-2.0-flash", api_key=os.environ["GEMINI_API_KEY"], ) settings = GoogleAIChatPromptExecutionSettings( temperature=0.2, ) kernel = Kernel(memory=NullMemory()) model_client = SKChatCompletionAdapter( sk_client, kernel=kernel, prompt_settings=settings, model_info={ "family": ModelFamily.GEMINI_2_0_FLASH, "function_calling": True, "json_output": True, "vision": True, }, ) # Call the model directly. model_result = await model_client.create( messages=[UserMessage(content="What is the capital of France?", source="User")] ) print(model_result) # Create an assistant agent with the model client. assistant = AssistantAgent( "assistant", model_client=model_client, tools=[get_weather], system_message="You are a helpful assistant." ) # Call the assistant with a task. stream = assistant.run_stream(task="What is the weather in Paris and London?") await Console(stream) asyncio.run(main())
Ollama 模型:
pip install "autogen-ext[semantic-kernel-ollama]"
import asyncio from autogen_agentchat.agents import AssistantAgent from autogen_core.models import UserMessage from autogen_ext.models.semantic_kernel import SKChatCompletionAdapter from semantic_kernel import Kernel from semantic_kernel.connectors.ai.ollama import OllamaChatCompletion, OllamaChatPromptExecutionSettings from semantic_kernel.memory.null_memory import NullMemory async def main() -> None: sk_client = OllamaChatCompletion( host="http://localhost:11434", ai_model_id="llama3.2:latest", ) ollama_settings = OllamaChatPromptExecutionSettings( options={"temperature": 0.5}, ) model_client = SKChatCompletionAdapter( sk_client, kernel=Kernel(memory=NullMemory()), prompt_settings=ollama_settings ) # Call the model directly. model_result = await model_client.create( messages=[UserMessage(content="What is the capital of France?", source="User")] ) print(model_result) # Create an assistant agent with the model client. assistant = AssistantAgent("assistant", model_client=model_client) # Call the assistant with a task. result = await assistant.run(task="What is the capital of France?") print(result) asyncio.run(main())
- actual_usage() RequestUsage [源代码]#
- count_tokens(messages: Sequence[已注解[系统消息 | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[工具 | 工具模式] = []) int [源代码]#
- async create(messages: Sequence[已注解[系统消息 | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[工具 | 工具模式] = [], json_output: bool | 无 = None, extra_create_args: 映射[str, 任何] = {}, cancellation_token: CancellationToken | 无 = None) CreateResult [源代码]#
使用Semantic Kernel客户端创建聊天完成。
extra_create_args 字典可以包含两个特殊的键:
- “kernel” (可选):
用于执行请求的
semantic_kernel.Kernel
实例。 如果在构造函数或extra_create_args中未提供,则会引发ValueError。
- “prompt_execution_settings” (可选):
一个对应于底层 Semantic Kernel 客户端的
PromptExecutionSettings
子类的实例(例如,AzureChatPromptExecutionSettings, GoogleAIChatPromptExecutionSettings)。如果未提供,将使用适配器的默认 prompt 设置。
- Parameters:
messages – 要发送的LLM消息列表。
tools – 聊天过程中可能被调用的工具。
json_output – 模型是否期望返回JSON。
extra_create_args – 控制聊天完成行为的额外参数。
cancellation_token – 允许取消请求的令牌。
- Returns:
CreateResult – 聊天完成的结果。
- async create_stream(messages: Sequence[已注解[系统消息 | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[工具 | 工具模式] = [], json_output: bool | 无 = None, extra_create_args: 映射[str, 任何] = {}, cancellation_token: CancellationToken | 无 = None) AsyncGenerator[str | CreateResult, 无] [源代码]#
使用语义内核客户端创建一个流式聊天完成。
extra_create_args 字典可以包含两个特殊的键:
- “kernel” (可选):
用于执行请求的
semantic_kernel.Kernel
实例。 如果在构造函数或extra_create_args中未提供,则会引发ValueError。
- “prompt_execution_settings” (可选):
一个对应于底层 Semantic Kernel 客户端的
PromptExecutionSettings
子类的实例(例如,AzureChatPromptExecutionSettings, GoogleAIChatPromptExecutionSettings)。如果未提供,将使用适配器的默认 prompt 设置。
- Parameters:
messages – 要发送的LLM消息列表。
工具 – 聊天过程中可能调用的工具。
json_output – 模型是否期望返回JSON。
extra_create_args – 控制聊天完成行为的额外参数。
cancellation_token – 允许取消请求的令牌。
- Yields:
Union[str, CreateResult] – 可以是响应的字符串片段,也可以是包含函数调用的CreateResult。
- remaining_tokens(messages: Sequence[已注解[系统消息 | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[工具 | 工具模式] = []) int [源代码]#
- total_usage() RequestUsage [源代码]#