【AI Agent系列】【MetaGPT】总结这段时间学习MetaGPT的一些学习方法和感悟

@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.info(f"{self._setting}: no news. waiting.")
        return
    rsp = await self.react()
    # Reset the next action to be taken.
    self.rc.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

self.put_message(msg）

def put_message(self, message):
        """Place the message into the Role object's private message buffer."""
        if not message:
            return
        self.rc.msg_buffer.push(message)

async def _observe(self, ignore_memory=False) -> int:
        """Prepare new messages for processing from the message buffer and other sources."""
        # Read unprocessed messages from the msg buffer.
        news = []
        if self.recovered: 
            news = [self.latest_observed_msg] if self.latest_observed_msg else []
        if not news:
            news = self.rc.msg_buffer.pop_all()
        # Store the read messages in your own memory to prevent duplicate processing.
        old_messages = [] if ignore_memory else self.rc.memory.get()
        self.rc.memory.add_batch(news)
        # Filter out messages of interest.
        self.rc.news = [
            n for n in news if (n.cause_by in self.rc.watch or self.name in n.send_to) and n not in old_messages
        ]
        self.latest_observed_msg = self.rc.news[-1] if self.rc.news else None  # record the latest observed msg
        # Design Rules:
        # If you need to further categorize Message objects, you can do so using the Message.set_meta function.
        # msg_buffer is a receiving buffer, avoid adding message data and operations to msg_buffer.
        news_text = [f"{i.role}: {i.content[:20]}..." for i in self.rc.news]
        if news_text:
            logger.debug(f"{self._setting} observed: {news_text}")
        return len(self.rc.news)

self.rc.news = [
         n for n in news if (n.cause_by in self.rc.watch or self.name in n.send_to) and n not in old_messages
     ]

def _watch(self, actions: Iterable[Type[Action]] | Iterable[Action]):
        """Watch Actions of interest. Role will select Messages caused by these Actions from its personal message
        buffer during _observe.
        """
        self.rc.watch = {any_to_str(t) for t in actions}

async def react(self) -> Message:
        """Entry to one of three strategies by which Role reacts to the observed Message"""
        if self.rc.react_mode == RoleReactMode.REACT:
            rsp = await self._react()
        elif self.rc.react_mode == RoleReactMode.BY_ORDER:
            rsp = await self._act_by_order()
        elif self.rc.react_mode == RoleReactMode.PLAN_AND_ACT:
            rsp = await self._plan_and_act()
        self._set_state(state=-1)  # current reaction is complete, reset state to -1 and todo back to None
        return rsp

class SubscriptionAssistant(Role):
    """Analyze user subscription requirements."""
    name: str = "同学小张的订阅助手"
    profile: str = "Subscription Assistant"
    goal: str = "analyze user subscription requirements to provide personalized subscription services."
    constraints: str = "utilize the same language as the User Requirement"
    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)
        self._init_actions([ParseSubRequirement, RunSubscription]) ## 2. 先解析用户需求，然后运行订阅
        self._watch([UserRequirement, WriteCrawlerCode])  ## 触发