【AI Agent系列】【阿里AgentScope框架】1. 深入源码：详细解读AgentScope中的智能体定义以及模型配置的流程

class AgentBase(Operator, metaclass=_RecordInitSettingMeta):
    """Base class for all agents.
    All agents should inherit from this class and implement the `reply`
    function.
    """

def __init__(
    self,
    name: str,
    sys_prompt: Optional[str] = None,
    model_config_name: str = None,
    use_memory: bool = True,
    memory_config: Optional[dict] = None,
) -> None:
    r"""Initialize an agent from the given arguments.
    Args:
        name (`str`):
            The name of the agent.
        sys_prompt (`Optional[str]`):
            The system prompt of the agent, which can be passed by args
            or hard-coded in the agent.
        model_config_name (`str`, defaults to None):
            The name of the model config, which is used to load model from
            configuration.
        use_memory (`bool`, defaults to `True`):
            Whether the agent has memory.
        memory_config (`Optional[dict]`):
            The config of memory.
    """
    self.name = name
    self.memory_config = memory_config
    if sys_prompt is not None:
        self.sys_prompt = sys_prompt
    # TODO: support to receive a ModelWrapper instance
    if model_config_name is not None:
        self.model = load_model_by_config_name(model_config_name)
    if use_memory:
        self.memory = TemporaryMemory(memory_config)
    else:
        self.memory = None
    # The global unique id of this agent
    self._agent_id = self.__class__.generate_agent_id()
    # The audience of this agent, which means if this agent generates a
    # response, it will be passed to all agents in the audience.
    self._audience = None

def _get_model_wrapper(model_type: str) -> Type[ModelWrapperBase]:
    ......
    if model_type in ModelWrapperBase.type_registry:
        return ModelWrapperBase.type_registry[  # type: ignore[return-value]
            model_type
        ]
    elif model_type in ModelWrapperBase.registry:
        return ModelWrapperBase.registry[  # type: ignore[return-value]
            model_type
        ]
    elif model_type in ModelWrapperBase.deprecated_type_registry:
        cls = ModelWrapperBase.deprecated_type_registry[model_type]
        ......
        return cls  # type: ignore[return-value]
    else:
        ......
        return PostAPIModelWrapperBase
def load_model_by_config_name(config_name: str) -> ModelWrapperBase:
    """Load the model by config name."""
    ......
    return _get_model_wrapper(model_type=model_type)(**kwargs)

class DialogAgent(AgentBase):
    """A simple agent used to perform a dialogue. Your can set its role by
    `sys_prompt`."""

def __init__(
    self,
    name: str,
    sys_prompt: str,
    model_config_name: str,
    use_memory: bool = True,
    memory_config: Optional[dict] = None,
    prompt_type: Optional[PromptType] = None,
) -> None:
    """Initialize the dialog agent.
    Arguments:
        name (`str`):
            The name of the agent.
        sys_prompt (`Optional[str]`):
            The system prompt of the agent, which can be passed by args
            or hard-coded in the agent.
        model_config_name (`str`):
            The name of the model config, which is used to load model from
            configuration.
        use_memory (`bool`, defaults to `True`):
            Whether the agent has memory.
        memory_config (`Optional[dict]`):
            The config of memory.
        prompt_type (`Optional[PromptType]`, defaults to
        `PromptType.LIST`):
            The type of the prompt organization, chosen from
            `PromptType.LIST` or `PromptType.STRING`.
    """

def reply(self, x: dict = None) -> dict:
    """Reply function of the agent. Processes the input data,
    generates a prompt using the current dialogue memory and system
    prompt, and invokes the language model to produce a response. The
    response is then formatted and added to the dialogue memory.
    Args:
        x (`dict`, defaults to `None`):
            A dictionary representing the user's input to the agent. This
            input is added to the dialogue memory if provided. Defaults to
            None.
    Returns:
        A dictionary representing the message generated by the agent in
        response to the user's input.
    """
    # record the input if needed
    if self.memory:
        self.memory.add(x)
    # prepare prompt
    prompt = self.model.format(
        Msg("system", self.sys_prompt, role="system"),
        self.memory and self.memory.get_memory(),  # type: ignore[arg-type]
    )
    # call llm and generate response
    response = self.model(prompt).text
    msg = Msg(self.name, response, role="assistant")
    # Print/speak the message in this agent's voice
    self.speak(msg)
    # Record the message in memory
    if self.memory:
        self.memory.add(msg)
    return msg

if self.memory:
    self.memory.add(x)
......
if self.memory:
    self.memory.add(msg)

def format(
    self,
    *args: Union[Msg, Sequence[Msg]],
) -> List[dict]:
......
    messages = []
    for arg in args:
        if arg is None:
            continue
        if isinstance(arg, Msg):
            messages.append(
                {
                    "role": arg.role,
                    "name": arg.name,
                    "content": _convert_to_str(arg.content),
                },
            )
        elif isinstance(arg, list):
            messages.extend(self.format(*arg))
        ......
    return messages
prompt = self.model.format(
    Msg("system", self.sys_prompt, role="system"),
    self.memory and self.memory.get_memory(),  # type: ignore[arg-type]
)

response = self.model(prompt).text

class UserAgent(AgentBase):
    """User agent class"""
    def __init__(self, name: str = "User", require_url: bool = False) -> None:
        """Initialize a UserAgent object.
        Arguments:
            name (`str`, defaults to `"User"`):
                The name of the agent. Defaults to "User".
            require_url (`bool`, defaults to `False`):
                Whether the agent requires user to input a URL. Defaults to
                False. The URL can lead to a website, a file,
                or a directory. It will be added into the generated message
                in field `url`.
        """
        super().__init__(name=name)
        self.name = name
        self.require_url = require_url

def user_input(
    prefix: str = "User input: ",
    timeout: Optional[int] = None,
) -> str:
    """get user input"""
    if hasattr(thread_local_data, "uid"):
        content = get_player_input(
            timeout=timeout,
            uid=thread_local_data.uid,
        )
    else:
        if timeout:
            from inputimeout import inputimeout, TimeoutOccurred
            try:
                content = inputimeout(prefix, timeout=timeout)
            except TimeoutOccurred as exc:
                raise TimeoutError("timed out") from exc
        else:
            content = input(prefix)
    return content
def reply(
    self,
    x: dict = None,
    required_keys: Optional[Union[list[str], str]] = None,
    timeout: Optional[int] = None,
) -> dict:
    ......
    if self.memory:
        self.memory.add(x)
    ......
    content = user_input(timeout=timeout)
    ......
    # Add to memory
    if self.memory:
        self.memory.add(msg)
    return msg