構建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: 對話模擬器實例
# 循環執行對話步驟,每步選擇一個發言者并打印其消息