3、单智能体开发_单个智能体功能

写在开头：本系列跟随datewhale的 《MetaGPT多智能体课程》，深入理解并实践多智能体系统的开发，感谢datewhale的开源与组织学习！

概述

在metagpt看来，agent = LLM+观察+思考+行动+记忆。一个agent启动后会观察获取信息，加入记忆，进行多轮思考和行动后（ReAct）后输出结果。

RoleContext

在metagpt中，role是智能体agent的逻辑抽象。role在与环境交互时，通过RoleContext对象实现。

class RoleContext(BaseModel):
    """Role Runtime Context"""

    model_config = ConfigDict(arbitrary_types_allowed=True) #意味着这个字典可以存储任意类型的值

    # # env exclude=True to avoid `RecursionError: maximum recursion depth exceeded in comparison`
    #指向 "Environment" 类型的引用。它被标记为 exclude=True,这意味着在序列化和反序列化过程中,这个属性将被排除。这样做是为了避免由于环境对象中可能存在的循环引用导致的递归错误。
    env: "Environment" = Field(default=None, exclude=True)  # # avoid circular import
    #msg_buffer 是一个 MessageQueue 通过该对象与其他role交互
    msg_buffer: MessageQueue = Field(
        default_factory=MessageQueue, exclude=True
    )  # Memory记忆对象，当role执行_act后的结果以message对象保存在memory里；当role执行_observe的时候会把 msg_buffer所有消息转移到memory
    memory: Memory = Field(default_factory=Memory)
    # long_term_memory: LongTermMemory = Field(default_factory=LongTermMemory)
    state: int = Field(default=-1)  # -1 indicates initial or termination state where todo is None
    todo: Action = Field(default=None, exclude=True) #下一个待执行的action
    watch: set[str] = Field(default_factory=set) #str是当前role观察的action列表用着_observe获得的news进行消息过滤
    news: list[Type[Message]] = Field(default=[], exclude=True)  # TODO not used _observe的时候与当前role相关的消息
    react_mode: RoleReactMode = (
        RoleReactMode.REACT
    )  # see `Role._set_react_mode` for definitions of the following two attributes,还有其他两个：BY_ORDER顺序执行，PLAN_AND_ACT一次计划，后面依次执行
    max_react_loop: int = 1 #react最大循环次数
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

实现一个单动作智能体

需求分析

在metagpt中，action是动作的逻辑抽象。通过self._aask来获取大模型的回答

async def _aask(self, prompt: str, system_msgs: Optional[list[str]] = None) -> str:
    """Append default prefix"""
    return await self.llm.aask(prompt, system_msgs) #self.llm 是一个 Language Model 对象,它提供了异步的 aask 方法来处理提示并返回响应。
1
2
3

编写SimpleWriteCode动作

import re
import asyncio
from metagpt.actions import Action

class SimpleWriteCode(Action):   #继承Action类

    PROMPT_TEMPLATE: str = """
    Write a python function that can {instruction} and provide two runnnable test cases.
    Return ```python your_code_here ```with NO other texts,
    your code:
    """    #一个类属性,存储了一个多行的字符串模板，含有一个占位符{instruction}，提供一个标准化的格式
    
    name: str = "SimpleWriteCode"   #类属性
#定义了一个名为run的异步方法,它接受一个instruction参数,并返回一个生成的包含代码的文本
    async def run(self, instruction: str):
        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction) 
        #使用format()方法将PROMPT_TEMPLATE中的{instruction}占位符替换为传入的instruction参数。
        rsp = await self._aask(prompt) #名为_aask的异步方法,并传入了生成的prompt。await关键字确保方法等待_aask的结果
        code_text = SimpleWriteCode.parse_code(rsp) #提取python代码
        return code_text

    @staticmethod 
    #这是一个静态方法，不需要用self就能访问，不需要访问类的实例属性和方法，一般用于实现辅助功能，比如这里的提取大模型返回的python代码
    #通过正则表达式提取python部分
    def parse_code(rsp):
        pattern = r'```python(.*)```'    #以```python开头，.*除了换行符外的任意字符，又是以```结尾
        match = re.search(pattern, rsp, re.DOTALL)   #匹配，re.DOTALL，re.DOTALL 标志确保 . 字符可以匹配任何字符,包括换行符
        code_text = match.group(1) if match else rsp  #group(0)是整个匹配的字符 1是第一个匹配的字符组
        return code_text
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

设计SimpleCoder角色

在metagpt中message类是最基本的信息类型，组成如下：

在这个角色中实现把用户输入传递给动作，调用动作

from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger #logger 用于记录代理的日志信息

class SimpleCoder(Role):
	#都是用于标志代理，profile更适合用于描述代理的能力，更适合用于与用户的直接交互中,使对话更加自然和个性化。
    name: str = "Alice"
    profile: str = "SimpleCoder"
	
	#__init__方法初始化，匹配写好的动作
    def __init__(self, **kwargs):   #self 代表当前类的实例对象，
    #"基类先,派生类后"，子类可能依赖于父类的一些属性和方法,如果父类没有初始化好,子类的行为可能会出现问题。
    # super() 可以访问父类的方法和属性，**kwargs 表示接受任意数量的关键字参数，这些参数会被收集到一个字典中,通过 **kwargs 传递给方法。
        super().__init__(**kwargs)   
        self._init_action([SimpleWriteCode]) 
        #_init_actions是role中定义的，接受一个动作列表进行初始化
        #写好后自定义动作SimpleWriteCode被加入代办self.rc.todo

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self.rc.todo}")  #f-string 字符串格式，在字符串内插入表达式，动态的替换为对应值
        todo = self.rc.todo # 先赋值，可以确保只访问一次
		
		# find the most recent messages，如果变成[1]就是找到的第2个
		#role通过初始化memory对象作为self.rc.memory属性，存储_observe之后的的每个message,也就是说，他是一个message列表
		'''函数定义如下
		def get_memories(self, k=0) -> list[Message]:
       			 return self.rc.memory.get(k=k) #返回top-k,k=0时返回所有
		'''
        msg = self.get_memories(k=1)[0]  #返回的是最近的一条

        code_text = await todo.run(msg.content)
        msg = Message(content=code_text, role=self.profile, cause_by=type(todo)) # type(todo) (可能是 SimpleWriteCode 类)

        return msg

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

• 使用属性装饰器:
  1.允许你控制属性的读写访问,提高了类的封装性
  2.使代码更加清晰和易读,因为属性的访问和设置看起来-
  和普通属性一样

运行角色

• _act() 和 run() 这两个方法之间的主要区别如下:
  1._act() 方法是实际执行操作的方法,它负责根据当前的任务(self.rc.todo)来生成一个 Message 对象作为响应。
  2.run() 方法是整个行为的入口点,它负责观察环境,并根据观察结果决定是否需要采取行动(调用 _act() 方法)。

import asyncio

async def main():
    msg = "write a function that calculates the sum of a list"
    role = SimpleCoder()
    logger.info(msg)
    result = await role.run(msg) 
    logger.info(result)

asyncio.run(main())
1
2
3
4
5
6
7
8
9
10

报错：AttributeError: ‘SimpleCoder’ object has no attribute ‘_init_actions’. Did you mean: ‘_init_action’?
上述解决办法：将’_init_actions’改为’set_actions’ （版本问题）
输出：

def create_sum_function():
    def sum_of_list(lst):
        return sum(lst)
    return sum_of_list

# Test cases
sum_function = create_sum_function()

# Test case 1
assert sum_function([1, 2, 3, 4, 5]) == 15
print("Test case 1 passed.")

# Test case 2
assert sum_function([-1, 0, 1, 2, 3]) == 5
print("Test case 2 passed.")
# 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

实现多动作agent

需求分析

不仅希望编写代码，还能代码执行。需要一个多动作智能体，暂且叫RunnableCoder,一个又能写又能运行的role,有两个动作：SimpleWriteCode和SimpleRunCode.
SimpleWriteCode同前。

SimpleRunCode动作

class SimpleRunCode(Action):

    name: str = "SimpleRunCode"

    async def run(self, code_text: str):
        # 在Windows环境下，result可能无法正确返回生成结果，在windows中在终端中输入python3可能会导致打开微软商店
        #subprocess.run() 函数来执行外部命令和程序，"python3" 是 Python 解释器的路径，-c 选项告诉 Python 执行后面的代码
        #capture_output=True 选项用于捕获执行命令的输出,包括标准输出和标准错误，text=True 选项告诉 subprocess.run() 以字符串形式返回捕获的输出,而不是字节类型。
        result = subprocess.run(["python3", "-c", code_text], capture_output=True, text=True)
        # 采用下面的可选代码来替换上面的代码
        # result = subprocess.run(["python", "-c", code_text], capture_output=True, text=True)
        # import sys
        # result = subprocess.run([sys.executable, "-c", code_text], capture_output=True, text=True)
        code_result = result.stdout
        logger.info(f"{code_result=}")
        return code_result
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

RunnableCoder角色

class RunnableCoder(Role):

    name: str = "Alice"
    profile: str = "RunnableCoder"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteCode, SimpleRunCode])
        self._set_react_mode(react_mode="by_order")

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: 准备 {self.rc.todo}")
        # 通过在底层按顺序选择动作
        # todo 首先是 SimpleWriteCode() 然后是 SimpleRunCode()
        todo = self.rc.todo

        msg = self.get_memories(k=1)[0] # 得到最近的一条消息
        result = await todo.run(msg.content)

        msg = Message(content=result, role=self.profile, cause_by=type(todo))
        self.rc.memory.add(msg)
        return msg
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

对比上一代单动作的角色，同样是先初始化super().init（），再覆盖_act（）函数，区别在于self.set_actions()传入了两个动作, self._set_react_mode(react_mode=“by_order”)是将模式定义成顺序执行，后面还可以选择ReAct。

运行角色

import asyncio

async def main():
    msg = "write a function that calculates the sum of a list"
    role = RunnableCoder()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)

asyncio.run(main())
1
2
3
4
5
6
7
8
9
10

把role改成新角色即可。

• 格式规范
  #注释后面先空格再加内容；一个模块完成空两行；最后结尾再空一行。
  结果如下：

2024-05-27 14:57:42.367 | INFO     | metagpt.const:get_metagpt_package_root:29 - Package root set to /hpc2hdd/home/gbian883/MetaGPT/MetaGPT-main
2024-05-27 14:57:45.699 | INFO     | __main__:main:90 - write a function that calculates the sum of a list
2024-05-27 14:57:45.700 | INFO     | __main__:_act:74 - Alice(RunnableCoder): 准备 SimpleWriteCode
1
2
3

def generate_sum_function():
    def sum_of_list(lst):
        return sum(lst)
    return sum_of_list

# Test cases
def test_sum_of_list():
    sum_function = generate_sum_function()
    assert sum_function([1, 2, 3, 4, 5]) == 15
    assert sum_function([-1, 2, -3, 4, -5]) == -3

test_sum_of_list()
1
2
3
4
5
6
7
8
9
10
11
12

2024-05-27 14:57:49.989 | INFO     | metagpt.utils.cost_manager:update_cost:57 - Total running cost: $0.002 | Max budget: $10.000 | Current cost: $0.002, prompt_tokens: 65, completion_tokens: 98
2024-05-27 14:57:49.991 | INFO     | __main__:_act:74 - Alice(RunnableCoder): 准备 SimpleRunCode
2024-05-27 14:57:50.040 | INFO     | __main__:run:59 - code_result=''
2024-05-27 14:57:50.041 | INFO     | __main__:main:92 - RunnableCoder: 

Process finished with exit code 0
1
2
3
4
5
6

实现一个更复杂的智能体：技术文档助手

需求分析：由于大模型token的限制，无法一次性的输出我们希望的技术文档。如果把需求拆解成一个一个的去提问，不仅费时，还需要人工交互，增加麻烦。

WriteDirectory动作

class WriteDirectory(Action):
    """Action class for writing tutorial directories.
    Args:
        name: The name of the action.
        language: The language to output, default is "Chinese".
        用于编写教程目录的动作类。
        参数：
        name：动作的名称。
        language：输出的语言，默认为"Chinese"。
    """

    name: str = "WriteDirectory"
    language: str = "Chinese"

    async def run(self, topic: str, *args, **kwargs) -> Dict:
        """
        根据主题执行生成教程目录的格式化定义。
            参数：
            topic：教程主题。
            返回：
            教程目录信息，包括{"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}.
        """
        COMMON_PROMPT = """
        You are now a seasoned technical professional in the field of the internet. 
        We need you to write a technical tutorial with the topic "{topic}".
        您现在是互联网领域的经验丰富的技术专业人员。
        我们需要您撰写一个关于"{topic}"的技术教程。
        """

        DIRECTORY_PROMPT = COMMON_PROMPT + """
        Please provide the specific table of contents for this tutorial, strictly following the following requirements:
        1. The output must be strictly in the specified language, {language}.
        2. Answer strictly in the dictionary format like {{"title": "xxx", "directory": [{{"dir 1": ["sub dir 1", "sub dir 2"]}}, {{"dir 2": ["sub dir 3", "sub dir 4"]}}]}}.
        3. The directory should be as specific and sufficient as possible, with a primary and secondary directory.The secondary directory is in the array.
        4. Do not have extra spaces or line breaks.
        5. Each directory title has practical significance.
        请按照以下要求提供本教程的具体目录：
        1. 输出必须严格符合指定语言，{language}。
        2. 回答必须严格按照字典格式，如{{"title": "xxx", "directory": [{{"dir 1": ["sub dir 1", "sub dir 2"]}}, {{"dir 2": ["sub dir 3", "sub dir 4"]}}]}}。
        3. 目录应尽可能具体和充分，包括一级和二级目录。二级目录在数组中。
        4. 不要有额外的空格或换行符。
        5. 每个目录标题都具有实际意义。
        """
        prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language) 
        # Python 的字符串格式化功能,将 topic 和 self.language 变量的值插入到预定义的字符串中,生成新的提示字符串并赋值给 prompt。
        resp = await self._aask(prompt=prompt)
        return OutputParser.extract_struct(resp, dict) # 提取为字典类型
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

得到结果后做内容解析，数据格式化得到字典类型的格式。

def extract_struct(cls, text: str, data_type: Union[type(list), type(dict)]) -> Union[list, dict]:
    """Extracts and parses a specified type of structure (dictionary or list) from the given text.
    The text only contains a list or dictionary, which may have nested structures.

    Args:
        text: The text containing the structure (dictionary or list).
        data_type: The data type to extract, can be "list" or "dict".

    Returns:
        - If extraction and parsing are successful, it returns the corresponding data structure (list or dictionary).
        - If extraction fails or parsing encounters an error, it throw an exception.
    返回：
    - 如果提取和解析成功，它将返回相应的数据结构（列表或字典）。
    - 如果提取失败或解析遇到错误，则抛出异常。
    Examples:
        >>> text = 'xxx [1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}] xxx'
        >>> result_list = OutputParser.extract_struct(text, "list")
        >>> print(result_list)
        >>> # Output: [1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}]

        >>> text = 'xxx {"x": 1, "y": {"a": 2, "b": {"c": 3}}} xxx'
        >>> result_dict = OutputParser.extract_struct(text, "dict")
        >>> print(result_dict)
        >>> # Output: {"x": 1, "y": {"a": 2, "b": {"c": 3}}}
    """
    # Find the first "[" or "{" and the last "]" or "}"
    start_index = text.find("[" if data_type is list else "{")
    end_index = text.rfind("]" if data_type is list else "}")

    if start_index != -1 and end_index != -1:
        # Extract the structure part
        structure_text = text[start_index : end_index + 1]  #切片左闭右开

        try:
            # structure_text 转换为 Python 的数据类型，将字符串形式的数据转为python对象
            result = ast.literal_eval(structure_text)

            # isinstance是内置函数，用于检查一个对象是否是指定类型
            if isinstance(result, list) or isinstance(result, dict):
                return result  #解析成功且是正确类型

            raise ValueError(f"The extracted structure is not a {data_type}.") # 如果不是指定类型就抛出一个ValueError
		#一旦上面抛出一个ValueError,下面就进入except，抛出一个更通用具体的的Exception
		#SyntaxError是structure_text格式不对无法解析，所以有两个
        except (ValueError, SyntaxError) as e:
            raise Exception(f"Error while extracting and parsing the {data_type}: {e}")
    else:
        logger.error(f"No {data_type} found in the text.") #没有找到正确的位置
        return [] if data_type is list else {} #也能返回一个符合要求的值
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

• 注意
  Union表示其中任选一种，Union[type(list), type(dict)]表示两种类型的的对象，而Union[list, dict]是两种数据结构

这样我们就得到了字典类型的对象–大模型输出的目录结构。下面需要生成每个章节的内容。

WriteContent动作

class WriteContent(Action):
    """Action class for writing tutorial content.
    Args:
        name: The name of the action.
        directory: The content to write.
        language: The language to output, default is "Chinese".
    """

    name: str = "WriteContent"
    directory: dict = dict() #属性声明
    language: str = "Chinese"

    async def run(self, topic: str, *args, **kwargs) -> str:
        """Execute the action to write document content according to the directory and topic.
        Args:
            topic: The tutorial topic.
        Returns:
            The written tutorial content.
        """
        COMMON_PROMPT = """
        You are now a seasoned technical professional in the field of the internet. 
        We need you to write a technical tutorial with the topic "{topic}".
        """
        CONTENT_PROMPT = COMMON_PROMPT + """
        现在我将为您提供该主题的模块目录标题。请详细输出此标题的详细原理内容。如果有代码示例，请按照标准代码规范提供。没有代码示例则不需要提供。     
        该主题的模块目录标题如下：
        {directory}
        
        严格按照以下要求限制输出：
        1. 遵循Markdown语法格式进行布局。
        2. 如果有代码示例，必须遵循标准语法规范，具备文档注释，并以代码块形式显示。
        3. 输出必须严格使用指定语言{language}。
        4. 不得有冗余输出，包括总结性陈述。
        5. 严禁输出主题"{topic}"。
        """
        prompt = CONTENT_PROMPT.format(
            topic=topic, language=self.language, directory=self.directory)
        return await self._aask(prompt=prompt)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

• 关键字和位置参数
  *args 和 **kwargs都是用来处理可变长度的参数，*args 是位置参数，比如：print_info(Alice, 25, New York, Engineer)，**kwargs是关键字参数，在函数内部通过字典访问，比如：print_info(name=“Alice”, age=25, city=“New York”, occupation=“Engineer”)

根据目录，生成内容主要是 directory: dict = dict()，传入目录，以及在prompt中添加你的要求。

TutorialAssistant角色

class TutorialAssistant(Role):
    """Tutorial assistant, input one sentence to generate a tutorial document in markup format.

    Args:
        name: The name of the role.
        profile: The role profile description.
        goal: The goal of the role.
        constraints: Constraints or requirements for the role.
        language: The language in which the tutorial documents will be generated.
    """

    name: str = "Stitch"
    profile: str = "Tutorial Assistant"
    goal: str = "Generate tutorial documents"
    constraints: str = "Strictly follow Markdown's syntax, with neat and standardized layout"
    language: str = "Chinese"

    topic: str = ""
    main_title: str = ""
    total_content: str = ""

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._init_actions([WriteDirectory(language=self.language)])
        self._set_react_mode(react_mode=RoleReactMode.REACT.value)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

def _init_actions(self, actions):
    self._reset()
    for idx, action in enumerate(actions):
        if not isinstance(action, Action):
            i = action("", llm=self._llm)  # 如果不是,则创建一个新的 Action 实例,并将 llm 参数设置为 self._llm
        else:
        	# 如果 self._setting.is_human 为 True，但 action.llm 不是 HumanProvider 的实例,则会触发这个 if 语句块
            if self._setting.is_human and not isinstance(action.llm, HumanProvider):
                logger.warning(f"is_human attribute does not take effect,"
                    f"as Role's {str(action)} was initialized using LLM, try passing in Action classes instead of initialized instances")
            i = action
        i.set_prefix(self._get_prefix(), self.profile)
        self.actions.append(i)
        self.states.append(f"{idx}. {action}")
        # 最后输出的样例 ['0. WriteContent', '1. WriteContent', '2. WriteContent', '3. WriteContent', '4. WriteContent', '5. WriteContent', '6. WriteContent', '7. WriteContent', '8. WriteContent']
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

当初始化一个动作，动作会加入到self.action列表，列表里面存储了所有动作。
接下来查看run方法的实现，当我们启动一个角色run时如何工作，有message就添加到记忆，没有就观察环境。

async def run(self, message=None):
    """Observe, and think and act based on the results of the observation
        观察，并根据观察结果进行思考和行动。"""
    if message:
        if isinstance(message, str):
            message = Message(message)
        if isinstance(message, Message):
            self.recv(message)  # recv 方法获取消息内容并添加到memory
        if isinstance(message, list):
            self.recv(Message("\n".join(message)))
         '''如果message存在，它会检查message的类型，
            如果是字符串，则将其转换为Message对象；
            如果是Message对象，则直接调用recv方法；
            如果是列表，则将列表中的消息合并成一个新的消息，然后再调用recv方法。
            相当于预处理将入参转化为Message对象并添加到role的记忆中'''
    elif 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()
    # Publish the reply to the environment, waiting for the next subscriber to process
    self._publish_message(rsp)
    return rsp

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

• logger方法
  debug() 用于记录调试级别的日志信息，info() 用于记录信息级别的日志信息，比如正确的输入输出。

看看rev()方法的定义：

def recv(self, message: Message) -> None:
    """add message to history."""
    # self._history += f"\n{message}"
    # self._context = self._history
    if message in self.rc.memory.get():
        return
    self.rc.memory.add(message)
1
2
3
4
5
6
7

看看react()方法定义

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
1
2
3
4
5
6
7
8
9
10
11

state=-1,表示始末，因为state是当前需要执行动作的下标，-1就表示当前没有执行动作，self.rc.todo就为空

def _set_state(self, state: int):
    """Update the current state."""
    self.rc.state = state
    logger.debug(self.actions)
    self.rc.todo = self.actions[self.rc.state] if state >= 0 else None
1
2
3
4
5

self._react基本确定了agent的行动路线

async def _react(self) -> Message:
    """Think first, then act, until the Role _think it is time to stop and requires no more todo.
    This is the standard think-act loop in the ReAct paper, which alternates thinking and acting in task solving, i.e. _think -> _act -> _think -> _act -> ... 
    Use llm to select actions in _think dynamically
    """
    actions_taken = 0
    rsp = Message("No actions taken yet") # will be overwritten after Role _act
    while actions_taken < self.rc.max_react_loop:
        # think
        await self._think()
        if self.rc.todo is None:
            break
        # act
        logger.debug(f"{self._setting}: {self.rc.state=}, will do {self.rc.todo}") #好奇self._setting是什么
        rsp = await self._act()
        actions_taken += 1
    return rsp # return output from the last action
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

现在需要重写_react,先思考，再执行，直到没有todo

async def _react(self) -> Message:
    """Execute the assistant's think and actions.

    Returns:
        A message containing the final result of the assistant's actions.
    执行助手的思考和行动。
    返回：
    包含助手行动最终结果的消息。
    """
    while True:
        await self._think()
        if self.rc.todo is None:
            break
        msg = await self._act()
    root_path = TUTORIAL_PATH / datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
    await File.write(root_path, f"{self.main_title}.md", self.total_content.encode('utf-8'))
    return msg
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

_rethink方法负责更新当前需要触发的流程。当self.rc.todo为none时，设置初始化_set_state为第一个action,对应从0开始，此时self.rc.todo也会变成第一个action;当还有下一个状态时，更新_set_state；当没有任何处理事项了self.rc.todo又被置为none

async def _think(self) -> None:
    """Determine the next action to be taken by the role."""
    if self.rc.todo is None:
        self._set_state(0)
        return

    if self.rc.state + 1 < len(self.states):
        self._set_state(self.rc.state + 1)
    else:
        self.rc.todo = None
1
2
3
4
5
6
7
8
9
10

重写_act(),负责执行角色(role)确定的下一个行动,根据上面的Action可以分为WriteDirectory和WriteContent，在需要生成目录的时候读取memory的要求

async def _act(self) -> Message:
    """Perform an action as determined by the role.

    Returns:
            A message containing the result of the action.
    """
    todo = self.rc.todo
    if type(todo) is WriteDirectory:
        msg = self.rc.memory.get(k=1)[0]
        self.topic = msg.content
        resp = await todo.run(topic=self.topic)
        logger.info(resp)
        return await self._handle_directory(resp)#将writedirector生成的目录一级标题actions添加到actions列表中。
    resp = await todo.run(topic=self.topic)
    logger.info(resp)
    if self.total_content != "":
        self.total_content += "\n\n\n"
    self.total_content += resp
    return Message(content=resp, role=self.profile)

async def _handle_directory(self, titles: Dict) -> Message:
    """Handle the directories for the tutorial document.

    Args:
        titles: A dictionary containing the titles and directory structure,
                such as {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}

    Returns:
        A message containing information about the directory.
        处理教程文档的目录。
        参数：
        titles：包含标题和目录结构的字典，
        例如{"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}。
        返回值：
        包含目录信息的消息。
    """
    # 当生成目录后记录目录标题（因为最后要输出完整文档）
    self.main_title = titles.get("title")
    directory = f"{self.main_title}\n"
    # self.total_content用来存储最好要输出的所有内容
    self.total_content += f"# {self.main_title}"
    actions = list()
    for first_dir in titles.get("directory"):
        # 根据目录结构来生成新的需要行动的action（目前只设计了两级目录）
        actions.append(WriteContent(language=self.language, directory=first_dir))
        key = list(first_dir.keys())[0]
        directory += f"- {key}\n"  # 添加到 directory 变量的末尾，-是无序列表
        for second_dir in first_dir[key]:
            directory += f"  - {second_dir}\n"
    self._init_actions(actions)
    self.rc.todo = None
    return Message(content=directory)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

这里一级二级标题的工作流程
_act( ):
1、如果现在执行的动作是写目录WriteDirectory，就先读取用户输入的问题，执行WriteDirectory动作，得到字典结构的目录例如{“title”:“写一份大模型产品的需求分析报告”,“directory”:[ {‘‘dir1’’:[“subdir11”,“subdir12”] } , {‘‘dir2’’:[“subdir21”,“subdir22”] } ] }，通过_handle_directory方法输出的Message对象directory 是：

写一份大模型产品的需求分析报告
	-dir1
		-subdir11
		-subdir12
	-dir2
		-subdir21
		-subdir22
1
2
3
4
5
6
7

2、如果不是，就是生成内容动作，直接执行，不空，末尾填添加换行标志，并且把生成的内容合并到total_content中，输出的Message对象就是完整的内容。
_handle_directory( ):
先从字典中得到标题title，放入到directory 中，并且在total_content中写上标题#(markdown)形式。读取动作，字典"directory"的值又是一个字典包含一级标题（key）和二级标题（value）,将一级标题（key）+二级标题（value）喂进去写内容，再把一级标题（加-）添加到directory 中（就是此时的key），添加二级标题（加 -）添加到directory 中（就是此时的value）.。初始化role的action,这次就该是生成内容啦，因为一开始在role的__init__时候action是WriteDirectory。

运行角色

import asyncio

async def main():
    msg = "Git 教程"
    role = TutorialAssistant()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)

asyncio.run(main())
1
2
3
4
5
6
7
8
9
10

参考文献：https://blog.csdn.net/Attitude93/article/details/136314642