Multi-Agent-Task05: Multi-Agent Development
输入“/”快速插入
Multi-Agent-Task05: Multi-Agent Development
飞书用户68503月4日创建
Multi-Agent Components
Environment
Environment 是一个在 RL 领域更常用的概念，在 RL 中，Agent 需要学着在 Environment 中采取最好的 Action 来最大化一些累积奖励。
Agent 的 Action Space 往往受到 Environment 的限制，Environment 通常具有一定的规则，Agents 必须按照规则进行 Act。
MetaGPT 提供了一个标准的 Environment Component 来管理 Agent 的 Action 与 Message 传递。
MetaGPT 的源码对 Environment 介绍如下
环境，承载一批角色，角色可以向环境发布消息，可以被其他角色观察到
Environment, hosting a batch of roles, roles can publish messages to the environment, and can be observed by other roles
打个比方，env 提供的是一个让 Agents 都能上桌讨论的环境，例如圆桌会议。
源码中，成员变量部分定义如下
class Environment(ExtEnv):
"""环境，承载一批角色，角色可以向环境发布消息，可以被其他角色观察到
Environment, hosting a batch of roles, roles can publish messages to the environment, and can be observed by other roles
"""
model_config = ConfigDict(arbitrary_types_allowed=True)
desc: str = Field(default="") # 环境描述
roles: dict[str, SerializeAsAny["Role"]] = Field(default_factory=dict, validate_default=True)
member_addrs: Dict["Role", Set] = Field(default_factory=dict, exclude=True)
history: str = "" # For debug
context: Context = Field(default_factory=Context, exclude=True)
其中，
desc 变量用于描述当前的 Environment
role 变量用来指定当前 Environment 中的 Roles
member_adds 变量表示当前 Environment 中的 Roles 以及他们可以接收消息的地址
history 变量用于记录 Environment 中的 Message 记录
context 变量用于存储环境的上下文信息
下面看 MetaGPT 中的 Environment 机制是如何运作的
首先，当 Environment 运行时，调用 Environment 的 run 方法，让所有 Role 运行一次。
async def run(self, k=1):
"""处理一次所有信息的运行
Process all Role runs at once
"""
for _ in range(k):
futures = []
for role in self.roles.values():
future = role.run()
futures.append(future)
await asyncio.gather(*futures)
logger.debug(f"is idle: {self.is_idle}")
跳转到 Role 的 run 方法，如下
@role_raise_decorator
async def run(self, with_message=None) -> Message | None:
"""Observe, and think and act based on the results of the observation"""
if with_message:
msg = None
if isinstance(with_message, str):
msg = Message(content=with_message)
elif isinstance(with_message, Message):
msg = with_message
elif isinstance(with_message, list):
msg = Message(content="\n".join(with_message))
if not msg.cause_by:
msg.cause_by = UserRequirement
self.put_message(msg)
if not await self._observe():
# If there is no new information, suspend and wait
logger.debug(f"{self._setting}: no news. waiting.")
return
rsp = await self.react()
# Reset the next action to be taken.
self.set_todo(None)
# Send the response message to the Environment object to have it relay the message to the subscribers.
self.publish_message(rsp)
return rsp