- 大家好，我是 同学小张，日常分享AI知识和实战案例
- 欢迎 点赞 + 关注 👏，持续学习，持续干货输出。
- + 一起交流💬，一起进步💪。
- 微信公众号也可搜【同学小张】 🙏
本站文章一览：
书接上篇文章，上篇文章我们运行了第一个AgentScope程序，里面我们创建了两个智能体：DialogAgent对话智能体 和 UserAgent用户智能体。本文我们来深入源码，看AgentScope框架内，这些智能体是如何实现的。
0. 智能体基类 - AgentBase
class AgentBase(Operator, metaclass=_RecordInitSettingMeta): """Base class for all agents. All agents should inherit from this class and implement the `reply` function. """
这是AgentScope中智能体的基类，所有智能体的定义都需要继承它，并且重写其 reply 函数。
0.1 初始化过程
0.1.1 总过程
初始化实现的功能：
（1）初始化的过程，name是必须参数。其余都可选。
（2）load_model_by_config_name 来加载模型配置，后面细说。
（3）use_memory 默认为True，使用memory时，memory是TemporaryMemory类型，并接收一个可选的 memory_config 参数。
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
0.1.2 加载模型配置 - load_model_by_config_name
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)
从源码看，这个函数的作用是根据你配置中的模型名称来加载模型的APIWrapper，例如我们上篇文章中的"model_type": "openai"。理论上，这里的"model_type"的值，只要是下面图中的任意一个即可：
0.1.3 模型配置的一些细节
在上篇文章中的配置部分我有一些疑问：
今天通过源码知道了其中的门道，看下面关于OpenAI接口的封装：
model_name是在外面赋值的，所以必须要有。
api_key是通过OpenAI接口传递进去的参数，根据我们先前的使用经验，这里可以不传，只要环境变量中存在 OPENAI_API_KEY 和 OPENAI_BASE_URL 即可使用。
1. 对话智能体 - DialogAgent
定义如下：通过 sys_prompt 参数指定其承担的角色，例如 “你是一个Python专家” 让其充当Python专家给你解决问题。
class DialogAgent(AgentBase): """A simple agent used to perform a dialogue. Your can set its role by `sys_prompt`."""
1.1 初始化过程
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`. """
初始化参数中 name、sys_prompt、model_config_name是必须设置的。
1.2 重写 reply 函数
这个reply函数是必须重写的。下面来看源码：
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
（1）首先是 memory 的操作，开始时将输入信息添加到 memory 中，结束前将大模型的回复写入到 memory 中：
if self.memory: self.memory.add(x) ...... if self.memory: self.memory.add(msg)
（2）prompt组装部分：将 system_prompt 和 memory的内容组装成OpenAI需要的message列表。
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] )
（3）调用大模型获取结果
response = self.model(prompt).text
1.3 小结
所以，对话智能体的实现就是将 system_prompt 和 对话的 memory 组装成prompt，然后让大模型回复。
2. 用户智能体 - UserAgent
2.1 初始化过程
所有参数都是可选的，默认名称是 “User” :
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
2.2 重写 reply 函数
实现的功能就是接收用户输入：content = user_input(timeout=timeout)
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
2.3 小结
所以，UserAgent的作用就是接收用户输入，让人参与到多智能体的交互中。
3. 总结
本文主要看了下AgentScope中智能体agent的定义源码，深入学习了agent初始化的过程，配置的加载等流程。挑选了两个简单的agent - DialogAgent 和 UserAgent进行了详细学习。个性化Agent的定义，需要继承AgentBase基类，重写其reply函数，在reply函数中，定义个性化Agent的个性化动作。
AgentScope目前还有一些其它的个性化Agent，例如react_agent、rpc_agent等，咱们后面用到再学。
如果觉得本文对你有帮助，麻烦点个赞和关注呗 ~~~
- 大家好，我是 同学小张，日常分享AI知识和实战案例
- 欢迎 点赞 + 关注 👏，持续学习，持续干货输出。
- 一起交流💬，一起进步💪。
- 微信公众号也可搜【同学小张】 🙏
本站文章一览：
