构建LangChain应用程序的示例代码：56、如何实现一个多智能体模拟,其中没有固定的发言顺序。智能体自行决定谁来发言,通过竞价机制实现

多智能体分散式发言人选择

示例展示了如何实现一个多智能体模拟,其中没有固定的发言顺序。智能体自行决定谁来发言,通过竞价机制实现。

我们将在下面的示例中展示一场虚构的总统辩论来演示这一过程。

导入LangChain相关模块

from typing import Callable, Listimport tenacity
from langchain.output_parsers import RegexParser
from langchain.prompts import PromptTemplate
from langchain.schema import (HumanMessage,SystemMessage,
)
from langchain_openai import ChatOpenAI# 导入所需的模块和类
# typing: 用于类型注解
# tenacity: 用于实现重试机制
# langchain相关模块: 用于构建对话系统

`DialogueAgent` 和 `DialogueSimulator` 类

我们将使用在 Multi-Player Dungeons & Dragons 中定义的相同 DialogueAgent 和 DialogueSimulator 类。

class DialogueAgent:def __init__(self,name: str,system_message: SystemMessage,model: ChatOpenAI,) -> None:self.name = nameself.system_message = system_messageself.model = modelself.prefix = f"{self.name}: "self.reset()def reset(self):self.message_history = ["Here is the conversation so far."]def send(self) -> str:"""将聊天模型应用于消息历史记录并返回消息字符串"""message = self.model.invoke([self.system_message,HumanMessage(content="\n".join(self.message_history + [self.prefix])),])return message.contentdef receive(self, name: str, message: str) -> None:"""将{name}说的{message}连接到消息历史记录中"""self.message_history.append(f"{name}: {message}")class DialogueSimulator:def __init__(self,agents: List[DialogueAgent],selection_function: Callable[[int, List[DialogueAgent]], int],) -> None:self.agents = agentsself._step = 0self.select_next_speaker = selection_functiondef reset(self):for agent in self.agents:agent.reset()def inject(self, name: str, message: str):"""用{name}的{message}开始对话"""for agent in self.agents:agent.receive(name, message)# 增加时间步self._step += 1def step(self) -> tuple[str, str]:# 1. 选择下一个发言者speaker_idx = self.select_next_speaker(self._step, self.agents)speaker = self.agents[speaker_idx]# 2. 下一个发言者发送消息message = speaker.send()# 3. 所有人接收消息for receiver in self.agents:receiver.receive(speaker.name, message)# 4. 增加时间步self._step += 1return speaker.name, message# DialogueAgent类: 表示对话中的一个智能体
# DialogueSimulator类: 用于模拟多个智能体之间的对话

`BiddingDialogueAgent` 类

我们定义了 DialogueAgent 的一个子类,它有一个 bid() 方法,根据消息历史和最近的消息产生一个出价。

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_templatedef bid(self) -> str:"""要求聊天模型输出一个发言出价"""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.invoke([SystemMessage(content=prompt)]).contentreturn bid_string# BiddingDialogueAgent类: DialogueAgent的子类,增加了竞价功能

定义参与者和辩论主题

character_names = ["Donald Trump", "Kanye West", "Elizabeth Warren"]
topic = "transcontinental high speed rail"
word_limit = 50# 定义参与辩论的人物和辩论主题
# character_names: 参与者姓名列表
# topic: 辩论主题
# word_limit: 回答字数限制

生成系统消息

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).contentreturn character_descriptiondef 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)
]# 生成系统消息和角色描述
# generate_character_description: 生成角色描述
# generate_character_header: 生成角色头部信息
# generate_character_system_message: 生成角色系统消息

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}")# 打印生成的角色描述、头部信息和系统消息

出价的输出解析器

我们要求智能体输出一个发言出价。但由于智能体是输出字符串的LLM,我们需要:

定义他们将产生输出的格式
解析他们的输出

我们可以继承 RegexParser 来实现我们自己的自定义出价输出解析器。

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"
)# BidOutputParser类: 自定义的出价输出解析器
# bid_parser: 实例化的出价解析器

生成竞价系统消息

这受到 Generative Agents 中使用LLM确定记忆重要性的提示的启发。这将使用我们的 BidOutputParser 的格式指令。

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_templatecharacter_bidding_templates = [generate_character_bidding_template(character_header)for character_header in character_headers
]# generate_character_bidding_template: 生成角色竞价模板
# character_bidding_templates: 所有角色的竞价模板列表

for character_name, bidding_template in zip(character_names, character_bidding_templates
):print(f"{character_name} Bidding Template:")print(bidding_template)# 打印生成的竞价模板

使用LLM详细阐述辩论主题

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).contentprint(f"Original topic:\n{topic}\n")
print(f"Detailed topic:\n{specified_topic}\n")# 使用LLM生成更详细的辩论主题

定义发言人选择函数

最后,我们将定义一个发言人选择函数 select_next_speaker,它接受每个智能体的出价并选择出价最高的智能体(同分随机打破平局)。

我们将定义一个 ask_for_bid 函数,使用我们之前定义的 bid_parser 来解析智能体的出价。我们将使用 tenacity 来装饰 ask_for_bid,在智能体的出价无法正确解析时多次重试,并在达到最大尝试次数后生成默认出价0。

@tenacity.retry(stop=tenacity.stop_after_attempt(2),wait=tenacity.wait_none(),  # 重试之间没有等待时间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,
)  # 当所有重试都用尽时的默认值
def ask_for_bid(agent) -> str:"""请求智能体出价并将出价解析为正确的格式。"""bid_string = agent.bid()bid = int(bid_parser.parse(bid_string)["bid"])return bid# ask_for_bid: 请求智能体出价并解析
# 使用tenacity装饰器处理可能的错误和重试

import numpy as npdef select_next_speaker(step: int, agents: List[DialogueAgent]) -> int:bids = []for agent in agents:bid = ask_for_bid(agent)bids.append(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.nameprint(f"Selected: {selected_name}")print("\n")return idx# select_next_speaker: 选择下一个发言者
# 根据智能体的出价选择出价最高的智能体

主循环

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,))# 创建BiddingDialogueAgent实例列表

max_iters = 10
n = 0simulator = 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# 主循环
# max_iters: 最大对话轮数
# simulator: 对话模拟器实例
# 循环执行对话步骤,每步选择一个发言者并打印其消息