 |
| https://www.pexels.com/photo/blond-women-having-a-conversation-8373834/ |
After we have looked at multi agent frameworks like LangGraph LangGraph Supervisor mode, LangGraph Network Mode) and AutoGen, we look at Semantic Kernel's Agent Chat. This how to set it up.
We have 3 LLM Agents (Customer, Investment and Saving Account) and we have one tool (or plugin in Semantic Kernel's term) assigned to each one of them.
Next, we have a selector which helps to decide which LLM Agent is to go next by looking at the last chat response in the chat history.
Finally, we have the terminator which decides when to terminate the chat.
Let's look at the source code. It is available at github.
Firstly, the tools. We have Python classes and function is annotated with @kernel_function.
import logging
import random
from semantic_kernel.functions import kernel_function
from sk_agentchat.hosting import container
logger = container[logging.Logger]
class AccountIdPlugin:
@kernel_function(
name="get_account_id", description="Get the account id for all accounts"
)
def get_account_id(self) -> str:
"""
Get the account id.
"""
id = random.choice(["123", "456", "789"])
logger.info(f"[TOOL] Bank account ID: {id}")
return id
class AccountSavingPlugin:
@kernel_function(
name="get_saving_balance", description="Get the balance of the saving account"
)
def get_saving_balance(self, account_id: str) -> float:
"""
Get the balance of the saving account.
"""
balance = {
"123": 10000.10,
"456": 20000.20,
"789": 30000.30,
}[account_id]
logger.info(f"[TOOL] Saving balance: {balance}")
return balance
class AccountInvestmentPlugin:
@kernel_function(
name="get_investment_balance",
description="Get the balance of the investment account",
)
def get_investment_balance(self, account_id: str) -> float:
"""
Get the balance of the investment account.
"""
balance = {
"123": 100000.10,
"456": 200000.20,
"789": 300000.30,
}[account_id]
logger.info(f"[TOOL] Investment balance: {balance}")
return balanceNext, we have the Agents that use these tools. Look for kernel.add_plugin. This is very simple.
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.function_choice_behavior import (
FunctionChoiceBehavior,
)
from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.azure_chat_prompt_execution_settings import ( # noqa: E501
AzureChatPromptExecutionSettings,
)
from semantic_kernel.kernel import Kernel
from sk_agentchat.hosting import container
from sk_agentchat.protocols.i_azure_chat_completion_service import (
IAzureChatCompletionService,
)
from sk_agentchat.tools.tools import (
AccountIdPlugin,
AccountInvestmentPlugin,
AccountSavingPlugin,
)
execution_settings = AzureChatPromptExecutionSettings()
execution_settings.function_choice_behavior = FunctionChoiceBehavior.Auto()
execution_settings.temperature = 0
AGENT_NAME_SAVING_ACCOUNT = "saving_account_agent"
AGENT_NAME_INVESTMENT_ACCOUNT = "investment_account_agent"
AGENT_NAME_CUSTOMER = "customer_agent"
azure_chat_completion_service = container[IAzureChatCompletionService]
def create_kernel_with_chat_completion(
service_id: str,
) -> Kernel:
kernel = Kernel()
kernel.add_service(
azure_chat_completion_service.create_chat_completion(service_id=service_id),
overwrite=True,
)
return kernel
def create_saving_account_agent() -> ChatCompletionAgent:
kernel = create_kernel_with_chat_completion(AGENT_NAME_SAVING_ACCOUNT)
kernel.add_plugin(plugin_name="AccountSavingPlugin", plugin=AccountSavingPlugin())
agent_saving_account = ChatCompletionAgent(
service_id=AGENT_NAME_SAVING_ACCOUNT,
kernel=kernel,
name=AGENT_NAME_SAVING_ACCOUNT,
execution_settings=execution_settings,
instructions="You are the bank saving account agent who can provide the saving "
"account balance.",
)
return agent_saving_account
def create_investment_account_agent() -> ChatCompletionAgent:
kernel = create_kernel_with_chat_completion(AGENT_NAME_INVESTMENT_ACCOUNT)
kernel.add_plugin(
plugin_name="AccountInvestmentPlugin", plugin=AccountInvestmentPlugin()
)
agent_investment_account = ChatCompletionAgent(
service_id=AGENT_NAME_INVESTMENT_ACCOUNT,
kernel=kernel,
name=AGENT_NAME_INVESTMENT_ACCOUNT,
execution_settings=execution_settings,
instructions="You are the bank investment account agent who can provide the "
"investment account balance.",
)
return agent_investment_account
def create_customer_agent() -> ChatCompletionAgent:
kernel = create_kernel_with_chat_completion(AGENT_NAME_CUSTOMER)
kernel.add_plugin(plugin_name="AccountIdPlugin", plugin=AccountIdPlugin())
agent_customer = ChatCompletionAgent(
service_id=AGENT_NAME_CUSTOMER,
kernel=kernel,
name=AGENT_NAME_CUSTOMER,
execution_settings=execution_settings,
instructions="You are the bank customer agent who can provide the account id.",
)
return agent_customerNext, we look at the selector and terminator.
from semantic_kernel.functions.kernel_function_from_prompt import (
KernelFunctionFromPrompt,
)
from sk_agentchat.tools.agents import (
AGENT_NAME_CUSTOMER,
AGENT_NAME_INVESTMENT_ACCOUNT,
AGENT_NAME_SAVING_ACCOUNT,
)
TERMINATION_KEYWORD = "done"
selection_function = KernelFunctionFromPrompt(
function_name="selection",
prompt=f"""
You are a helpful bank customer agent, collaborating with other product agents.
Determine which agent takes the next turn in a conversation based on the the most recent conversation.
State only the name of the participant to take the next turn.
No participant should take more than one turn in a row.
Choose only from these agents:
- {AGENT_NAME_CUSTOMER}
- {AGENT_NAME_INVESTMENT_ACCOUNT}
- {AGENT_NAME_SAVING_ACCOUNT}
Always follow these rules when selecting the next agent:
- After user input, it is {AGENT_NAME_CUSTOMER}'s turn.
- After {AGENT_NAME_CUSTOMER} replies, {AGENT_NAME_INVESTMENT_ACCOUNT}'s turn.
- After {AGENT_NAME_INVESTMENT_ACCOUNT} replies, {AGENT_NAME_SAVING_ACCOUNT}'s turn.
History:
{{{{$history}}}}
""", # noqa E501
)
termination_function = KernelFunctionFromPrompt(
function_name="termination",
prompt=f"""
Examine the RESPONSE and determine whether the account balances are returned in the conversation.
If content is satisfactory, respond with a single word without explanation: {TERMINATION_KEYWORD}.
If specific suggestions are being provided, it is not satisfactory.
If no correction is suggested, it is satisfactory.
RESPONSE:
{{{{$history}}}}
""", # noqa E501
)Lastly, we bring all of these together.
import asyncio
from typing import Any
from semantic_kernel.agents import AgentGroupChat
from semantic_kernel.agents.strategies.selection.kernel_function_selection_strategy import ( # noqa: E501
KernelFunctionSelectionStrategy,
)
from semantic_kernel.agents.strategies.termination.kernel_function_termination_strategy import ( # noqa: E501
KernelFunctionTerminationStrategy,
)
from semantic_kernel.contents.chat_message_content import ChatMessageContent
from semantic_kernel.contents.utils.author_role import AuthorRole
from sk_agentchat.tools.agents import (
AGENT_NAME_CUSTOMER,
create_customer_agent,
create_investment_account_agent,
create_kernel_with_chat_completion,
create_saving_account_agent,
)
from sk_agentchat.tools.functions import (
TERMINATION_KEYWORD,
selection_function,
termination_function,
)
async def main():
agent_customer = create_customer_agent()
agent_saving_account = create_saving_account_agent()
agent_investment_account = create_investment_account_agent()
def fn(results: Any) -> bool:
return TERMINATION_KEYWORD in str(results.value[0]).lower()
chat = AgentGroupChat(
agents=[agent_customer, agent_saving_account, agent_investment_account],
selection_strategy=KernelFunctionSelectionStrategy(
function=selection_function,
kernel=create_kernel_with_chat_completion("selection"),
result_parser=lambda result: (
str(result.value[0])
if result.value is not None
else AGENT_NAME_CUSTOMER
),
agent_variable_name="agents",
history_variable_name="history",
),
termination_strategy=KernelFunctionTerminationStrategy(
agents=[agent_saving_account],
function=termination_function,
kernel=create_kernel_with_chat_completion("termination"),
result_parser=fn,
history_variable_name="history",
maximum_iterations=10,
),
)
user_input = "Get the investment and saving account balance."
# this can be a while loop to get user input
await chat.add_chat_message(
ChatMessageContent(role=AuthorRole.USER, content=user_input)
)
async for response in chat.invoke():
print(f"# {response.role} - {response.name or '*'}: '{response.content}'")
if __name__ == "__main__":
asyncio.run(main())So, we have LLM agents, selector and terminator to get this agent chat to work. The output looks like this
# AuthorRole.ASSISTANT - customer_agent: 'The account IDs for your investment
and savings accounts are 456 and 789, respectively. Please let me know if you
need further assistance with these accounts!'
# AuthorRole.ASSISTANT - investment_account_agent: 'The balance for your investment account
is $200,000.20, and the balance for your savings account is $300,000.30. If you
need any more information, feel free to ask!'
# AuthorRole.ASSISTANT - saving_account_agent: 'The balance for your savings account is
$30,000.30. If you need any more information, feel free to ask!'
Comments
Post a Comment