의미 체계 커널 ChatCompletionAgent 탐색

프로젝트에 Microsoft.SemanticKernel.Agents.Core 패키지를 추가합니다.

dotnet add package Microsoft.SemanticKernel.Agents.Core --prerelease

semantic-kernel 패키지를 설치합니다.

pip install semantic-kernel

<dependency>
    <groupId>com.microsoft.semantic-kernel</groupId>
    <artifactId>semantickernel-agents-core</artifactId>
    <version>[LATEST]</version>
</dependency>

// Initialize a Kernel with a chat-completion service
IKernelBuilder builder = Kernel.CreateBuilder();

builder.AddAzureOpenAIChatCompletion(/*<...configuration parameters>*/);

Kernel kernel = builder.Build();

// Create the agent
ChatCompletionAgent agent =
    new()
    {
        Name = "SummarizationAgent",
        Instructions = "Summarize user input",
        Kernel = kernel
    };

ChatCompletionAgent만드는 방법에는 두 가지가 있습니다.

1. 채팅 완료 서비스를 직접 제공하여

from semantic_kernel.agents import ChatCompletionAgent

# Create the agent by directly providing the chat completion service
agent = ChatCompletionAgent(
    service=AzureChatCompletion(),  # your chat completion service instance
    name="<agent name>",
    instructions="<agent instructions>",
)

2. 커널을 먼저 만들고 서비스를 추가한 다음 커널을 제공합니다.

# Define the kernel
kernel = Kernel()

# Add the chat completion service to the kernel
kernel.add_service(AzureChatCompletion())

# Create the agent using the kernel
agent = ChatCompletionAgent(
  kernel=kernel, 
  name="<agent name>", 
  instructions="<agent instructions>",
)

첫 번째 방법은 채팅 완료 서비스가 이미 준비되었을 때 유용합니다. 두 번째 방법은 여러 서비스 또는 추가 기능을 관리하는 커널이 필요한 경우에 유용합니다.

// Initialize a Kernel with a chat-completion service
var chatCompletion = OpenAIChatCompletion.builder()
        .withOpenAIAsyncClient(client) // OpenAIAsyncClient with configuration parameters
        .withModelId(MODEL_ID)
        .build();

var kernel = Kernel.builder()
        .withAIService(ChatCompletionService.class, chatCompletion)
        .build();

// Create the agent
var agent = ChatCompletionAgent.builder()
        .withKernel(kernel)
        .build();

IKernelBuilder builder = Kernel.CreateBuilder();

// Initialize multiple chat-completion services.
builder.AddAzureOpenAIChatCompletion(/*<...service configuration>*/, serviceId: "service-1");
builder.AddAzureOpenAIChatCompletion(/*<...service configuration>*/, serviceId: "service-2");

Kernel kernel = builder.Build();

ChatCompletionAgent agent =
    new()
    {
        Name = "<agent name>",
        Instructions = "<agent instructions>",
        Kernel = kernel,
        Arguments = // Specify the service-identifier via the KernelArguments
          new KernelArguments(
            new OpenAIPromptExecutionSettings() 
            { 
              ServiceId = "service-2" // The target service-identifier.
            })
    };

from semantic_kernel.connectors.ai.open_ai import (
    AzureChatCompletion,
    AzureChatPromptExecutionSettings,
)

# Define the Kernel
kernel = Kernel()

# Add the AzureChatCompletion AI Service to the Kernel
kernel.add_service(AzureChatCompletion(service_id="service1"))
kernel.add_service(AzureChatCompletion(service_id="service2"))

settings = AzureChatPromptExecutionSettings(service_id="service2")

# Create the agent
agent = ChatCompletionAgent(
  kernel=kernel, 
  name="<agent name>", 
  instructions="<agent instructions>",
  arguments=KernelArguments(settings=settings)
)

현재 Java에서 기능을 사용할 수 없습니다.

ChatCompletionAgent와 대화하는 것은, 채팅 완료 ChatHistory와 상호 작용하는 인스턴스를 기반으로 한다는 점에서 다르지 않습니다.

사용자 메시지를 사용하여 에이전트를 호출하기만 하면 됩니다.

// Define agent
ChatCompletionAgent agent = ...;

// Generate the agent response(s)
await foreach (ChatMessageContent response in agent.InvokeAsync(new ChatMessageContent(AuthorRole.User, "<user input>")))
{
  // Process agent response(s)...
}

에이전트와 대화를 나누기 위해 AgentThread을 사용할 수도 있습니다. 여기서는 .를 사용하고 있습니다 ChatHistoryAgentThread.

ChatHistoryAgentThread 또한 이전 대화를 다시 시작하면 선택적 ChatHistory 개체를 해당 생성자를 통해 입력으로 사용할 수 있습니다. (표시되지 않음)

// Define agent
ChatCompletionAgent agent = ...;

AgentThread thread = new ChatHistoryAgentThread();

// Generate the agent response(s)
await foreach (ChatMessageContent response in agent.InvokeAsync(new ChatMessageContent(AuthorRole.User, "<user input>"), thread))
{
  // Process agent response(s)...
}

ChatCompletionAgent와 여러 가지 방법으로 대화할 수 있습니다.

가장 쉬운 방법은 get_response호출하고 기다리는 것입니다.

# Define agent
agent = ChatCompletionAgent(...)

# Generate the agent response
response = await agent.get_response(messages="user input")
# response is an `AgentResponseItem[ChatMessageContent]` object

에이전트가 호출 간에 대화 기록을 유지하도록 하려면 다음과 같이 전달할 ChatHistoryAgentThread 수 있습니다.

# Define agent
agent = ChatCompletionAgent(...)

# Generate the agent response(s)
response = await agent.get_response(messages="user input")

# Generate another response, continuing the conversation thread from the first response.
response2 = await agent.get_response(messages="user input", thread=response.thread)
# process agent response(s)

invoke 메서드를 호출하면 AsyncIterable의 AgentResponseItem[ChatMessageContent]가 반환됩니다.

# Define agent
agent = ChatCompletionAgent(...)

# Define the thread
thread = ChatHistoryAgentThread()

# Generate the agent response(s)
async for response in agent.invoke(messages="user input", thread=thread):
  # process agent response(s)

ChatCompletionAgent는 스트리밍을 지원하며, 이때 invoke_stream 메서드는 AsyncIterableStreamingChatMessageContent 반환합니다.

# Define agent
agent = ChatCompletionAgent(...)

# Define the thread
thread = ChatHistoryAgentThread()

# Generate the agent response(s)
async for response in agent.invoke_stream(messages="user input", thread=thread):
  # process agent response(s)

ChatCompletionAgent agent = ...;

// Generate the agent response(s)
agent.invokeAsync(new ChatMessageContent<>(AuthorRole.USER, "<user input>")).block();

에이전트와 대화를 나누기 위해 AgentThread을 사용할 수도 있습니다. 여기서는 .를 사용하고 있습니다 ChatHistoryAgentThread.

이전 대화를 다시 시작하는 경우, ChatHistoryAgentThread 생성자를 통해 ChatHistory 개체를 입력으로 사용할 수 있습니다. (표시되지 않음)

// Define agent
ChatCompletionAgent agent = ...;

AgentThread thread = new ChatHistoryAgentThread();

// Generate the agent response(s)
agent.invokeAsync(new ChatMessageContent<>(AuthorRole.USER, "<user input>"), thread).block();

ChatCompletionAgent에 대한 콜백 설명서는 곧 제공될 예정입니다.

on_intermediate_message 또는 agent.invoke(...) 안에서 agent.invoke_stream(...) 콜백을 구성하면, 호출자는 에이전트의 최종 응답을 작성하는 과정에서 생성된 중간 메시지를 수신할 수 있습니다.

import asyncio
from typing import Annotated

from semantic_kernel.agents.chat_completion.chat_completion_agent import ChatCompletionAgent, ChatHistoryAgentThread
from semantic_kernel.connectors.ai.open_ai.services.azure_chat_completion import AzureChatCompletion
from semantic_kernel.contents import FunctionCallContent, FunctionResultContent
from semantic_kernel.contents.chat_message_content import ChatMessageContent
from semantic_kernel.functions import kernel_function


# Define a sample plugin for the sample
class MenuPlugin:
    """A sample Menu Plugin used for the concept sample."""

    @kernel_function(description="Provides a list of specials from the menu.")
    def get_specials(self) -> Annotated[str, "Returns the specials from the menu."]:
        return """
        Special Soup: Clam Chowder
        Special Salad: Cobb Salad
        Special Drink: Chai Tea
        """

    @kernel_function(description="Provides the price of the requested menu item.")
    def get_item_price(
        self, menu_item: Annotated[str, "The name of the menu item."]
    ) -> Annotated[str, "Returns the price of the menu item."]:
        return "$9.99"


# This callback function will be called for each intermediate message
# Which will allow one to handle FunctionCallContent and FunctionResultContent
# If the callback is not provided, the agent will return the final response
# with no intermediate tool call steps.
async def handle_intermediate_steps(message: ChatMessageContent) -> None:
    for item in message.items or []:
        if isinstance(item, FunctionCallContent):
            print(f"Function Call:> {item.name} with arguments: {item.arguments}")
        elif isinstance(item, FunctionResultContent):
            print(f"Function Result:> {item.result} for function: {item.name}")
        else:
            print(f"{message.role}: {message.content}")


async def main() -> None:
    agent = ChatCompletionAgent(
        service=AzureChatCompletion(),
        name="Assistant",
        instructions="Answer questions about the menu.",
        plugins=[MenuPlugin()],
    )

    # Create a thread for the agent
    # If no thread is provided, a new thread will be
    # created and returned with the initial response
    thread: ChatHistoryAgentThread = None

    user_inputs = [
        "Hello",
        "What is the special soup?",
        "How much does that cost?",
        "Thank you",
    ]

    for user_input in user_inputs:
        print(f"# User: '{user_input}'")
        async for response in agent.invoke(
            messages=user_input,
            thread=thread,
            on_intermediate_message=handle_intermediate_steps,
        ):
            print(f"# {response.role}: {response}")
            thread = response.thread


if __name__ == "__main__":
    asyncio.run(main())

다음은 에이전트 호출 프로세스의 샘플 출력을 보여 줍니다.

User: 'Hello'
AuthorRole.ASSISTANT: Hi there! How can I assist you today?
User: 'What is the special soup?'
Function Call:> MenuPlugin-get_specials with arguments: {}
Function Result:> 
        Special Soup: Clam Chowder
        Special Salad: Cobb Salad
        Special Drink: Chai Tea
        for function: MenuPlugin-get_specials
AuthorRole.ASSISTANT: The special soup today is Clam Chowder. Would you like to know anything else from the menu?
User: 'How much does that cost?'
Function Call:> MenuPlugin-get_item_price with arguments: {"menu_item":"Clam Chowder"}
Function Result:> $9.99 for function: MenuPlugin-get_item_price
AuthorRole.ASSISTANT: The Clam Chowder costs $9.99. Would you like to know more about the menu or anything else?
User: 'Thank you'
AuthorRole.ASSISTANT: You're welcome! If you have any more questions, feel free to ask. Enjoy your day!

현재 Java에서 기능을 사용할 수 없습니다.

선언적 사양 사용에 대한 설명서는 곧 제공될 예정입니다.

YAML ChatCompletionAgent 선언적 사양에서 직접 인스턴스화할 수 있습니다. 이 방법을 사용하면 에이전트의 핵심 속성, 지침 및 사용 가능한 함수(플러그 인)를 구조화되고 이식 가능한 방식으로 정의할 수 있습니다. YAML을 사용하면 에이전트의 이름, 설명, 명령 프롬프트, 도구 집합 및 모델 매개 변수를 단일 문서에서 설명할 수 있으므로 에이전트의 구성을 쉽게 감사하고 재현할 수 있습니다.

예: YAML 사양을 기반으로 ChatCompletionAgent 만들기

import asyncio
from typing import Annotated

from semantic_kernel import Kernel
from semantic_kernel.agents import AgentRegistry, ChatHistoryAgentThread
from semantic_kernel.agents.chat_completion.chat_completion_agent import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.functions import kernel_function

# Define a plugin with kernel functions
class MenuPlugin:
    @kernel_function(description="Provides a list of specials from the menu.")
    def get_specials(self) -> Annotated[str, "Returns the specials from the menu."]:
        return """
        Special Soup: Clam Chowder
        Special Salad: Cobb Salad
        Special Drink: Chai Tea
        """

    @kernel_function(description="Provides the price of the requested menu item.")
    def get_item_price(
        self, menu_item: Annotated[str, "The name of the menu item."]
    ) -> Annotated[str, "Returns the price of the menu item."]:
        return "$9.99"

# YAML spec for the agent
AGENT_YAML = """
type: chat_completion_agent
name: Assistant
description: A helpful assistant.
instructions: Answer the user's questions using the menu functions.
tools:
  - id: MenuPlugin.get_specials
    type: function
  - id: MenuPlugin.get_item_price
    type: function
model:
  options:
    temperature: 0.7
"""

USER_INPUTS = [
    "Hello",
    "What is the special soup?",
    "What does that cost?",
    "Thank you",
]

async def main():
    kernel = Kernel()
    kernel.add_plugin(MenuPlugin(), plugin_name="MenuPlugin")

    agent: ChatCompletionAgent = await AgentRegistry.create_from_yaml(
        AGENT_YAML, kernel=kernel, service=OpenAIChatCompletion()
    )

    thread: ChatHistoryAgentThread | None = None

    for user_input in USER_INPUTS:
        print(f"# User: {user_input}")
        response = await agent.get_response(user_input, thread=thread)
        print(f"# {response.name}: {response}")
        thread = response.thread

    await thread.delete() if thread else None

if __name__ == "__main__":
    asyncio.run(main())

이 기능을 사용할 수 없습니다.