Spaces:
Runtime error
Runtime error
File size: 9,996 Bytes
4962437 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 |
from typing import Callable, List
import numpy as np
import tenacity
from langchain.chat_models import ChatOpenAI
from langchain.output_parsers import RegexParser
from langchain.prompts import PromptTemplate
from langchain.schema import (
HumanMessage,
SystemMessage,
)
from swarms import Worker
class DialogueAgent:
def __init__(
self,
name: str,
system_message: SystemMessage,
model: ChatOpenAI,
) -> None:
self.name = name
self.system_message = system_message
self.model = model
self.prefix = f"{self.name}: "
self.reset()
def reset(self):
self.message_history = ["Here is the conversation so far."]
def send(self) -> str:
"""
Applies the chatmodel to the message history
and returns the message string
"""
message = self.model(
[
self.system_message,
HumanMessage(content="\n".join(self.message_history + [self.prefix])),
]
)
return message.content
def receive(self, name: str, message: str) -> None:
"""
Concatenates {message} spoken by {name} into message history
"""
self.message_history.append(f"{name}: {message}")
class DialogueSimulator:
def __init__(
self,
agents: List[Worker],
selection_function: Callable[[int, List[Worker]], int],
) -> None:
self.agents = agents
self._step = 0
self.select_next_speaker = selection_function
def reset(self):
for agent in self.agents:
agent.reset()
def inject(self, name: str, message: str):
"""
Initiates the conversation with a {message} from {name}
"""
for agent in self.agents:
agent.receive(name, message)
# increment time
self._step += 1
def step(self) -> tuple[str, str]:
# 1. choose the next speaker
speaker_idx = self.select_next_speaker(self._step, self.agents)
speaker = self.agents[speaker_idx]
# 2. next speaker sends message
message = speaker.send()
# 3. everyone receives message
for receiver in self.agents:
receiver.receive(speaker.name, message)
# 4. increment time
self._step += 1
return speaker.name, message
class BiddingDialogueAgent(DialogueAgent):
def __init__(
self,
name,
system_message: SystemMessage,
bidding_template: PromptTemplate,
model: ChatOpenAI,
) -> None:
super().__init__(name, system_message, model)
self.bidding_template = bidding_template
def bid(self) -> str:
"""
Asks the chat model to output a bid to speak
"""
prompt = PromptTemplate(
input_variables=["message_history", "recent_message"],
template=self.bidding_template,
).format(
message_history="\n".join(self.message_history),
recent_message=self.message_history[-1],
)
bid_string = self.model([SystemMessage(content=prompt)]).content
return bid_string
character_names = ["Donald Trump", "Kanye West", "Elizabeth Warren"]
topic = "transcontinental high speed rail"
word_limit = 50
game_description = f"""Here is the topic for the presidential debate: {topic}.
The presidential candidates are: {', '.join(character_names)}."""
player_descriptor_system_message = SystemMessage(
content="You can add detail to the description of each presidential candidate."
)
def generate_character_description(character_name):
character_specifier_prompt = [
player_descriptor_system_message,
HumanMessage(
content=f"""{game_description}
Please reply with a creative description of the presidential candidate, {character_name}, in {word_limit} words or less, that emphasizes their personalities.
Speak directly to {character_name}.
Do not add anything else."""
),
]
character_description = ChatOpenAI(temperature=1.0)(
character_specifier_prompt
).content
return character_description
def generate_character_header(character_name, character_description):
return f"""{game_description}
Your name is {character_name}.
You are a presidential candidate.
Your description is as follows: {character_description}
You are debating the topic: {topic}.
Your goal is to be as creative as possible and make the voters think you are the best candidate.
"""
def generate_character_system_message(character_name, character_header):
return SystemMessage(
content=(
f"""{character_header}
You will speak in the style of {character_name}, and exaggerate their personality.
You will come up with creative ideas related to {topic}.
Do not say the same things over and over again.
Speak in the first person from the perspective of {character_name}
For describing your own body movements, wrap your description in '*'.
Do not change roles!
Do not speak from the perspective of anyone else.
Speak only from the perspective of {character_name}.
Stop speaking the moment you finish speaking from your perspective.
Never forget to keep your response to {word_limit} words!
Do not add anything else.
"""
)
)
character_descriptions = [
generate_character_description(character_name) for character_name in character_names
]
character_headers = [
generate_character_header(character_name, character_description)
for character_name, character_description in zip(
character_names, character_descriptions
)
]
character_system_messages = [
generate_character_system_message(character_name, character_headers)
for character_name, character_headers in zip(character_names, character_headers)
]
for (
character_name,
character_description,
character_header,
character_system_message,
) in zip(
character_names,
character_descriptions,
character_headers,
character_system_messages,
):
print(f"\n\n{character_name} Description:")
print(f"\n{character_description}")
print(f"\n{character_header}")
print(f"\n{character_system_message.content}")
class BidOutputParser(RegexParser):
def get_format_instructions(self) -> str:
return "Your response should be an integer delimited by angled brackets, like this: <int>."
bid_parser = BidOutputParser(
regex=r"<(\d+)>", output_keys=["bid"], default_output_key="bid"
)
def generate_character_bidding_template(character_header):
bidding_template = f"""{character_header}
{{message_history}}
On the scale of 1 to 10, where 1 is not contradictory and 10 is extremely contradictory, rate how contradictory the following message is to your ideas.
{{recent_message}}
{bid_parser.get_format_instructions()}
Do nothing else.
"""
return bidding_template
character_bidding_templates = [
generate_character_bidding_template(character_header)
for character_header in character_headers
]
for character_name, bidding_template in zip(
character_names, character_bidding_templates
):
print(f"{character_name} Bidding Template:")
print(bidding_template)
topic_specifier_prompt = [
SystemMessage(content="You can make a task more specific."),
HumanMessage(
content=f"""{game_description}
You are the debate moderator.
Please make the debate topic more specific.
Frame the debate topic as a problem to be solved.
Be creative and imaginative.
Please reply with the specified topic in {word_limit} words or less.
Speak directly to the presidential candidates: {*character_names,}.
Do not add anything else."""
),
]
specified_topic = ChatOpenAI(temperature=1.0)(topic_specifier_prompt).content
print(f"Original topic:\n{topic}\n")
print(f"Detailed topic:\n{specified_topic}\n")
@tenacity.retry(
stop=tenacity.stop_after_attempt(2),
wait=tenacity.wait_none(), # No waiting time between retries
retry=tenacity.retry_if_exception_type(ValueError),
before_sleep=lambda retry_state: print(
f"ValueError occurred: {retry_state.outcome.exception()}, retrying..."
),
retry_error_callback=lambda retry_state: 0,
) # Default value when all retries are exhausted
def ask_for_bid(agent) -> str:
"""
Ask for agent bid and parses the bid into the correct format.
"""
bid_string = agent.bid()
bid = int(bid_parser.parse(bid_string)["bid"])
return bid
def select_next_speaker(step: int, agents: List[DialogueAgent]) -> int:
bids = []
for agent in agents:
bid = ask_for_bid(agent)
bids.append(bid)
# randomly select among multiple agents with the same bid
max_value = np.max(bids)
max_indices = np.where(bids == max_value)[0]
idx = np.random.choice(max_indices)
print("Bids:")
for i, (bid, agent) in enumerate(zip(bids, agents)):
print(f"\t{agent.name} bid: {bid}")
if i == idx:
selected_name = agent.name
print(f"Selected: {selected_name}")
print("\n")
return idx
characters = []
for character_name, character_system_message, bidding_template in zip(
character_names, character_system_messages, character_bidding_templates
):
characters.append(
BiddingDialogueAgent(
name=character_name,
system_message=character_system_message,
model=ChatOpenAI(temperature=0.2),
bidding_template=bidding_template,
)
)
max_iters = 10
n = 0
simulator = DialogueSimulator(agents=characters, selection_function=select_next_speaker)
simulator.reset()
simulator.inject("Debate Moderator", specified_topic)
print(f"(Debate Moderator): {specified_topic}")
print("\n")
while n < max_iters:
name, message = simulator.step()
print(f"({name}): {message}")
print("\n")
n += 1 |