|
| 1 | +from typing import AsyncIterator |
| 2 | +from aact import Message, NodeFactory |
| 3 | +from aact.messages import Text, Tick, DataModel, DataModelFactory |
| 4 | +from sotopia.agents.llm_agent import ainput |
| 5 | +from sotopia.experimental.agents import BaseAgent |
| 6 | + |
| 7 | +from sotopia.generation_utils import agenerate |
| 8 | +from sotopia.generation_utils.generate import StrOutputParser |
| 9 | +from sotopia.messages import ActionType |
| 10 | + |
| 11 | +from pydantic import Field |
| 12 | + |
| 13 | + |
| 14 | +@DataModelFactory.register("agent_action") |
| 15 | +class AgentAction(DataModel): |
| 16 | + agent_name: str = Field(description="the name of the agent") |
| 17 | + action_type: ActionType = Field( |
| 18 | + description="whether to speak at this turn or choose to not do anything" |
| 19 | + ) |
| 20 | + argument: str = Field( |
| 21 | + description="the utterance if choose to speak, the expression or gesture if choose non-verbal communication, or the physical action if choose action" |
| 22 | + ) |
| 23 | + |
| 24 | + def to_natural_language(self) -> str: |
| 25 | + match self.action_type: |
| 26 | + case "none": |
| 27 | + return "did nothing" |
| 28 | + case "speak": |
| 29 | + return f'said: "{self.argument}"' |
| 30 | + case "non-verbal communication": |
| 31 | + return f"[{self.action_type}] {self.argument}" |
| 32 | + case "action": |
| 33 | + return f"[{self.action_type}] {self.argument}" |
| 34 | + case "leave": |
| 35 | + return "left the conversation" |
| 36 | + |
| 37 | + |
| 38 | +def _format_message_history(message_history: list[tuple[str, str]]) -> str: |
| 39 | + return "\n".join( |
| 40 | + (f"{speaker} said {message}") for speaker, message in message_history |
| 41 | + ) |
| 42 | + |
| 43 | + |
| 44 | +@NodeFactory.register("llm_agent") |
| 45 | +class LLMAgent(BaseAgent[AgentAction | Tick, AgentAction]): |
| 46 | + def __init__( |
| 47 | + self, |
| 48 | + input_text_channels: list[str], |
| 49 | + input_tick_channel: str, |
| 50 | + output_channel: str, |
| 51 | + query_interval: int, |
| 52 | + agent_name: str, |
| 53 | + goal: str, |
| 54 | + model_name: str, |
| 55 | + redis_url: str, |
| 56 | + ): |
| 57 | + super().__init__( |
| 58 | + [ |
| 59 | + (input_text_channel, AgentAction) |
| 60 | + for input_text_channel in input_text_channels |
| 61 | + ] |
| 62 | + + [ |
| 63 | + (input_tick_channel, Tick), |
| 64 | + ], |
| 65 | + [(output_channel, AgentAction)], |
| 66 | + redis_url, |
| 67 | + ) |
| 68 | + self.output_channel = output_channel |
| 69 | + self.query_interval = query_interval |
| 70 | + self.count_ticks = 0 |
| 71 | + self.message_history: list[tuple[str, str]] = [] |
| 72 | + self.name = agent_name |
| 73 | + self.model_name = model_name |
| 74 | + self.goal = goal |
| 75 | + |
| 76 | + async def send(self, message: AgentAction) -> None: |
| 77 | + if message.action_type == "speak": |
| 78 | + await self.r.publish( |
| 79 | + self.output_channel, |
| 80 | + Message[AgentAction](data=message).model_dump_json(), |
| 81 | + ) |
| 82 | + |
| 83 | + async def aact(self, message: AgentAction | Tick) -> AgentAction: |
| 84 | + match message: |
| 85 | + case Tick(): |
| 86 | + self.count_ticks += 1 |
| 87 | + if self.count_ticks % self.query_interval == 0: |
| 88 | + agent_action: str = await agenerate( |
| 89 | + model_name=self.model_name, |
| 90 | + template="Imagine that you are a friend of the other persons. Here is the " |
| 91 | + "conversation between you and them.\n" |
| 92 | + "You are {agent_name} in the conversation.\n" |
| 93 | + "{message_history}\n" |
| 94 | + "and you plan to {goal}.\n" |
| 95 | + "You can choose to interrupt the other person " |
| 96 | + "by saying something or not to interrupt by outputting notiong. What would you say? " |
| 97 | + "Please only output a sentence or not outputting anything." |
| 98 | + "{format_instructions}", |
| 99 | + input_values={ |
| 100 | + "message_history": _format_message_history( |
| 101 | + self.message_history |
| 102 | + ), |
| 103 | + "goal": self.goal, |
| 104 | + "agent_name": self.name, |
| 105 | + }, |
| 106 | + temperature=0.7, |
| 107 | + output_parser=StrOutputParser(), |
| 108 | + ) |
| 109 | + if agent_action != "none" and agent_action != "": |
| 110 | + self.message_history.append((self.name, agent_action)) |
| 111 | + return AgentAction( |
| 112 | + agent_name=self.name, |
| 113 | + action_type="speak", |
| 114 | + argument=agent_action, |
| 115 | + ) |
| 116 | + else: |
| 117 | + return AgentAction( |
| 118 | + agent_name=self.name, action_type="none", argument="" |
| 119 | + ) |
| 120 | + else: |
| 121 | + return AgentAction( |
| 122 | + agent_name=self.name, action_type="none", argument="" |
| 123 | + ) |
| 124 | + case AgentAction( |
| 125 | + agent_name=agent_name, action_type=action_type, argument=text |
| 126 | + ): |
| 127 | + if action_type == "speak": |
| 128 | + self.message_history.append((agent_name, text)) |
| 129 | + return AgentAction( |
| 130 | + agent_name=self.name, action_type="none", argument="" |
| 131 | + ) |
| 132 | + case _: |
| 133 | + raise ValueError(f"Unexpected message type: {type(message)}") |
| 134 | + |
| 135 | + |
| 136 | +@NodeFactory.register("input_node") |
| 137 | +class InputNode(BaseAgent[AgentAction, AgentAction]): |
| 138 | + def __init__( |
| 139 | + self, |
| 140 | + input_channel: str, |
| 141 | + output_channel: str, |
| 142 | + agent_name: str, |
| 143 | + redis_url: str = "redis://localhost:6379/0", |
| 144 | + ): |
| 145 | + super().__init__( |
| 146 | + input_channel_types=[(input_channel, AgentAction)], |
| 147 | + output_channel_types=[(output_channel, AgentAction)], |
| 148 | + redis_url=redis_url, |
| 149 | + ) |
| 150 | + self.input_channel = input_channel |
| 151 | + self.agent_name = agent_name |
| 152 | + |
| 153 | + async def event_handler( |
| 154 | + self, channel: str, message: Message[AgentAction] |
| 155 | + ) -> AsyncIterator[tuple[str, Message[AgentAction]]]: |
| 156 | + if channel == self.input_channel: |
| 157 | + print(f"Received message: {message}") |
| 158 | + else: |
| 159 | + raise ValueError(f"Unexpected channel: {channel}") |
| 160 | + yield self.output_channel, Text(text=message.data.argument) |
| 161 | + |
| 162 | + async def _task_scheduler(self) -> None: |
| 163 | + while not self.shutdown_event.is_set(): |
| 164 | + text_input = await ainput() |
| 165 | + await self.send( |
| 166 | + AgentAction( |
| 167 | + agent_name=self.agent_name, action_type="speak", argument=text_input |
| 168 | + ) |
| 169 | + ) |
0 commit comments