Open Dialogue Chit-Chat

windows
echo. > chitchat.py

chitchat.py
 
# Import required libraries
 
from typing import Type
from uagents import Model
from uagents.experimental.dialogues import Dialogue, Edge, Node
 
 

chitchat.py
 
# Define node of dialogue
default_state = Node(
    name="Default State",
    description=(
        "This is the default state of the dialogue. Every session starts in "
        "this state and is automatically updated once the dialogue starts."
    )
)
init_state = Node(
    name="Initiated",
    description=(
        "This is the initial state of the dialogue that is only available at "
        "the receiving agent."
    ),
)
chatting_state = Node(
    name="Chit Chatting",
    description="This is the state in which messages are exchanged.",
)
end_state = Node(
    name="Concluded",
    description="This is the state after the dialogue has been concluded.",
)
 
 

chitchat.py
 
# Edge definition for the dialogue transitions
init_session = Edge(
    name="initiate_session",
    description="Every dialogue starts with this transition.",
    parent=None,
    child=init_state,
)
reject_session = Edge(
    name="reject_session",
    description=("This is the transition for when the dialogue is rejected"),
    parent=init_state,
    child=end_state,
)
start_dialogue = Edge(
    name="start_dialogue",
    description="This is the transition from initiated to chit chatting.",
    parent=init_state,
    child=chatting_state,
)
cont_dialogue = Edge(
    name="continue_dialogue",
    description=(
        "This is the transition from one dialogue message to the next, "
        "i.e. for when the dialogue continues."
    ),
    parent=chatting_state,
    child=chatting_state,
)
end_session = Edge(
    name="end_session",
    description="This is the transition for when the session is ended.",
    parent=chatting_state,
    child=end_state,
)
 
 

chitchat.py
 
 
class ChitChatDialogue(Dialogue):
    """
    This is the specific definition of the rules for the chit-chat dialogue
    The rules will be predefined and the actual messages will be passed into it
    """
 
    def __init__(
        self,
        version: str | None = None,
        agent_address: str | None = None,
    ) -> None:
        super().__init__(
            name="ChitChatDialogue",
            version=version,
            nodes=[
                default_state,
                init_state,
                chatting_state,
                end_state,
            ],
            edges=[
                init_session,
                reject_session,
                start_dialogue,
                cont_dialogue,
                end_session,
            ],
        )
 
    def on_initiate_session(self, model: Type[Model]):
        """
        This handler is triggered when the initial message of the
        dialogue is received. From here you can either accept or reject.
        Logic that is needed to complete any kind of handshake or considers
        global agent state should go here.
        """
        return super()._on_state_transition(init_session.name, model)
 
    def on_reject_session(self, model: Type[Model]):
        """
        This handler is triggered when a reject message is returned on
        the initial message.
        Implement this if you need to clean up session data.
        """
        return super()._on_state_transition(reject_session.name, model)
 
    def on_start_dialogue(self, model: Type[Model]):
        """
        This handler is triggered when an accept message is returned on
        the initial message.
        Include logic to complete any handshake on the sender side and
        prepare the actual message exchange.
        """
        return super()._on_state_transition(start_dialogue.name, model)
 
    def on_continue_dialogue(self, model: Type[Model]):
        """
        This handler is triggered for every incoming "chitchat" message
        once the session has been accepted.
        Any additional stateful information within a dialogue needs to be
        persisted explicitly to access it at a later point in the dialogue.
        """
        return super()._on_state_transition(cont_dialogue.name, model)
 
    def on_end_session(self, model: Type[Model]):
        """
        This handler is triggered once the other party has ended the dialogue.
        Any final conclusion or cleanup goes here.
        """
        return super()._on_state_transition(end_session.name, model)
 

windows
echo. > agent1.py

agent1.py
 
# Import required libraries
import json
 
from uagents import Agent, Context, Model
from uagents.setup import fund_agent_if_low
from dialogues.chitchat import ChitChatDialogue
 
CHAT_AGENT_ADDRESS = "<your_agent_2_address>"
 
agent = Agent(
    name="chit_agent",
    seed="<random_string_of_choice>",
    port=8001,
    endpoint="http://127.0.0.1:8001/submit",
)
 
fund_agent_if_low(agent.wallet.address())
 
# Define dialogue messages; each transition needs a separate message
class InitiateChitChatDialogue(Model):
    pass
 
class AcceptChitChatDialogue(Model):
    pass
 
class ChitChatDialogueMessage(Model):
    text: str
 
class ConcludeChitChatDialogue(Model):
    pass
 
class RejectChitChatDialogue(Model):
    pass
 
# Instantiate the dialogues
chitchat_dialogue = ChitChatDialogue(
    version="0.1",
    agent_address=agent.address,
)
 
# Get an overview of the dialogue structure
print("Dialogue overview:")
print(json.dumps(chitchat_dialogue.get_overview(), indent=4))
print("---")
 
@chitchat_dialogue.on_initiate_session(InitiateChitChatDialogue)
async def start_chitchat(
    ctx: Context,
    sender: str,
    _msg: InitiateChitChatDialogue,
):
    ctx.logger.info(f"Received init message from {sender}")
    # Do something when the dialogue is initiated
    await ctx.send(sender, AcceptChitChatDialogue())
 
@chitchat_dialogue.on_start_dialogue(AcceptChitChatDialogue)
async def accept_chitchat(
    ctx: Context,
    sender: str,
    _msg: AcceptChitChatDialogue,
):
    ctx.logger.info(
        f"session with {sender} was accepted. I'll say 'Hello!' to start the ChitChat"
    )
    # Do something after the dialogue is started; e.g. send a message
    await ctx.send(sender, ChitChatDialogueMessage(text="Hello!"))
 
@chitchat_dialogue.on_reject_session(RejectChitChatDialogue)
async def reject_chitchat(
    ctx: Context,
    sender: str,
    _msg: RejectChitChatDialogue,
):
    # Do something when the dialogue is rejected and nothing has been sent yet
    ctx.logger.info(f"Received reject message from: {sender}")
 
@chitchat_dialogue.on_continue_dialogue(ChitChatDialogueMessage)
async def continue_chitchat(
    ctx: Context,
    sender: str,
    msg: ChitChatDialogueMessage,
):
    # Do something when the dialogue continues
    ctx.logger.info(f"Received message: {msg.text}")
    try:
        my_msg = input("Please enter your message:\n> ")
        if my_msg != "exit":
            await ctx.send(sender, ChitChatDialogueMessage(text=my_msg))
        else:
            await ctx.send(sender, ConcludeChitChatDialogue())
            ctx.logger.info(
                f"Received conclude message from: {sender}; accessing history:"
            )
            ctx.logger.info(chitchat_dialogue.get_conversation(ctx.session))
    except EOFError:
        await ctx.send(sender, ConcludeChitChatDialogue())
 
@chitchat_dialogue.on_end_session(ConcludeChitChatDialogue)
async def conclude_chitchat(
    ctx: Context,
    sender: str,
    _msg: ConcludeChitChatDialogue,
):
    # Do something when the dialogue is concluded after messages have been exchanged
    ctx.logger.info(f"Received conclude message from: {sender}; accessing history:")
    ctx.logger.info(chitchat_dialogue.get_conversation(ctx.session))
 
agent.include(chitchat_dialogue)
 
if __name__ == "__main__":
    print(f"Agent address: {agent.address}")
    agent.run()
 

windows
echo. > agent2.py

agent2.py
 
"""Chit chat dialogue example"""
 
from asyncio import sleep
 
from uagents import Agent, Context, Model
from uagents.setup import fund_agent_if_low
from dialogues.chitchat import ChitChatDialogue
 
CHIT_AGENT_ADDRESS = "<your_agent_1_address>"
 
agent = Agent(
    name="chat_agent",
    seed="<random_string_of_choice>",
    port=8002,
    endpoint="http://127.0.0.1:8002/submit",
)
 
fund_agent_if_low(agent.wallet.address())
 
# Define dialogue messages; each transition needs a separate message
class InitiateChitChatDialogue(Model):
    pass
 
class AcceptChitChatDialogue(Model):
    pass
 
class ChitChatDialogueMessage(Model):
    text: str
 
class ConcludeChitChatDialogue(Model):
    pass
 
class RejectChitChatDialogue(Model):
    pass
 
# Instantiate the dialogues
chitchat_dialogue = ChitChatDialogue(
    version="0.1",
    agent_address=agent.address,
)
 
@chitchat_dialogue.on_initiate_session(InitiateChitChatDialogue)
async def start_chitchat(
    ctx: Context,
    sender: str,
    _msg: InitiateChitChatDialogue,
):
    ctx.logger.info(f"Received init message from {sender}")
    # Do something when the dialogue is initiated
    await ctx.send(sender, AcceptChitChatDialogue())
 
@chitchat_dialogue.on_start_dialogue(AcceptChitChatDialogue)
async def accept_chitchat(
    ctx: Context,
    sender: str,
    _msg: AcceptChitChatDialogue,
):
    ctx.logger.info(
        f"session with {sender} was accepted. I'll say 'Hello!' to start the ChitChat"
    )
    # Do something after the dialogue is started; e.g. send a message
    await ctx.send(sender, ChitChatDialogueMessage(text="Hello!"))
 
@chitchat_dialogue.on_reject_session(RejectChitChatDialogue)
async def reject_chitchat(
    ctx: Context,
    sender: str,
    _msg: RejectChitChatDialogue,
):
    # Do something when the dialogue is rejected and nothing has been sent yet
    ctx.logger.info(f"Received reject message from: {sender}")
 
@chitchat_dialogue.on_continue_dialogue(ChitChatDialogueMessage)
async def continue_chitchat(
    ctx: Context,
    sender: str,
    msg: ChitChatDialogueMessage,
):
    ctx.logger.info(f"Received message: {msg.text}")
    try:
        my_msg = input("Please enter your message:\n> ")
        if my_msg != "exit":
            await ctx.send(sender, ChitChatDialogueMessage(text=my_msg))
        else:
            await ctx.send(sender, ConcludeChitChatDialogue())
            ctx.logger.info(
                f"Received conclude message from: {sender}; accessing history:"
            )
            ctx.logger.info(chitchat_dialogue.get_conversation(ctx.session))
    except EOFError:
        await ctx.send(sender, ConcludeChitChatDialogue())
 
@chitchat_dialogue.on_end_session(ConcludeChitChatDialogue)
async def conclude_chitchat(
    ctx: Context,
    sender: str,
    _msg: ConcludeChitChatDialogue,
):
    # Do something when the dialogue is concluded after messages have been exchanged
    ctx.logger.info(f"Received conclude message from: {sender}; accessing history:")
    ctx.logger.info(chitchat_dialogue.get_conversation(ctx.session))
 
agent.include(chitchat_dialogue)
 
# Initiate dialogue by sending message to agent1
@agent.on_event("startup")
async def start_cycle(ctx: Context):
    await sleep(5)
    await chitchat_dialogue.start_dialogue(
        ctx, CHIT_AGENT_ADDRESS, InitiateChitChatDialogue()
    )
 
if __name__ == "__main__":
    print(f"Agent address: {agent.address}")
    agent.run()