Merge branch 'main' into main

tests/metagpt/memory/__init__.py

@@ -0,0 +1,3 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# @Desc   :

tests/metagpt/memory/test_longterm_memory.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# @Desc   : unittest of `metagpt/memory/longterm_memory.py`
+
+from metagpt.config import CONFIG
+from metagpt.schema import Message
+from metagpt.actions import BossRequirement
+from metagpt.roles.role import RoleContext
+from metagpt.memory import LongTermMemory
+
+
+def test_ltm_search():
+    assert hasattr(CONFIG, "long_term_memory") is True
+    openai_api_key = CONFIG.openai_api_key
+    assert len(openai_api_key) > 20
+
+    role_id = 'UTUserLtm(Product Manager)'
+    rc = RoleContext(watch=[BossRequirement])
+    ltm = LongTermMemory()
+    ltm.recover_memory(role_id, rc)
+
+    idea = 'Write a cli snake game'
+    message = Message(role='BOSS', content=idea, cause_by=BossRequirement)
+    news = ltm.remember([message])
+    assert len(news) == 1
+    ltm.add(message)
+
+    sim_idea = 'Write a game of cli snake'
+    sim_message = Message(role='BOSS', content=sim_idea, cause_by=BossRequirement)
+    news = ltm.remember([sim_message])
+    assert len(news) == 0
+    ltm.add(sim_message)
+
+    new_idea = 'Write a 2048 web game'
+    new_message = Message(role='BOSS', content=new_idea, cause_by=BossRequirement)
+    news = ltm.remember([new_message])
+    assert len(news) == 1
+    ltm.add(new_message)
+
+    # restore from local index
+    ltm_new = LongTermMemory()
+    ltm_new.recover_memory(role_id, rc)
+    news = ltm_new.remember([message])
+    assert len(news) == 0
+
+    ltm_new.recover_memory(role_id, rc)
+    news = ltm_new.remember([sim_message])
+    assert len(news) == 0
+
+    new_idea = 'Write a Battle City'
+    new_message = Message(role='BOSS', content=new_idea, cause_by=BossRequirement)
+    news = ltm_new.remember([new_message])
+    assert len(news) == 1
+
+    ltm_new.clear()

tests/metagpt/memory/test_memory_storage.py

@@ -0,0 +1,82 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# @Desc   : the unittests of metagpt/memory/memory_storage.py
+
+from typing import List
+
+from metagpt.memory.memory_storage import MemoryStorage
+from metagpt.schema import Message
+from metagpt.actions import BossRequirement
+from metagpt.actions import WritePRD
+from metagpt.actions.action_output import ActionOutput
+
+
+def test_idea_message():
+    idea = 'Write a cli snake game'
+    role_id = 'UTUser1(Product Manager)'
+    message = Message(role='BOSS', content=idea, cause_by=BossRequirement)
+
+    memory_storage: MemoryStorage = MemoryStorage()
+    messages = memory_storage.recover_memory(role_id)
+    assert len(messages) == 0
+
+    memory_storage.add(message)
+    assert memory_storage.is_initialized is True
+
+    sim_idea = 'Write a game of cli snake'
+    sim_message = Message(role='BOSS', content=sim_idea, cause_by=BossRequirement)
+    new_messages = memory_storage.search(sim_message)
+    assert len(new_messages) == 0   # similar, return []
+
+    new_idea = 'Write a 2048 web game'
+    new_message = Message(role='BOSS', content=new_idea, cause_by=BossRequirement)
+    new_messages = memory_storage.search(new_message)
+    assert new_messages[0].content == message.content
+
+    memory_storage.clean()
+    assert memory_storage.is_initialized is False
+
+
+def test_actionout_message():
+    out_mapping = {
+        'field1': (str, ...),
+        'field2': (List[str], ...)
+    }
+    out_data = {
+        'field1': 'field1 value',
+        'field2': ['field2 value1', 'field2 value2']
+    }
+    ic_obj = ActionOutput.create_model_class('prd', out_mapping)
+
+    role_id = 'UTUser2(Architect)'
+    content = 'The boss has requested the creation of a command-line interface (CLI) snake game'
+    message = Message(content=content,
+                      instruct_content=ic_obj(**out_data),
+                      role='user',
+                      cause_by=WritePRD)  # WritePRD as test action
+
+    memory_storage: MemoryStorage = MemoryStorage()
+    messages = memory_storage.recover_memory(role_id)
+    assert len(messages) == 0
+
+    memory_storage.add(message)
+    assert memory_storage.is_initialized is True
+
+    sim_conent = 'The request is command-line interface (CLI) snake game'
+    sim_message = Message(content=sim_conent,
+                          instruct_content=ic_obj(**out_data),
+                          role='user',
+                          cause_by=WritePRD)
+    new_messages = memory_storage.search(sim_message)
+    assert len(new_messages) == 0   # similar, return []
+
+    new_conent = 'Incorporate basic features of a snake game such as scoring and increasing difficulty'
+    new_message = Message(content=new_conent,
+                          instruct_content=ic_obj(**out_data),
+                          role='user',
+                          cause_by=WritePRD)
+    new_messages = memory_storage.search(new_message)
+    assert new_messages[0].content == message.content
+
+    memory_storage.clean()
+    assert memory_storage.is_initialized is False

tests/metagpt/utils/test_serialize.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# @Desc   : the unittest of serialize
+
+from typing import List, Tuple
+import pytest
+
+from pydantic import create_model
+
+from metagpt.actions.action_output import ActionOutput
+from metagpt.actions import WritePRD
+from metagpt.schema import Message
+from metagpt.utils.serialize import actionoutout_schema_to_mapping, serialize_message, deserialize_message
+
+
+def test_actionoutout_schema_to_mapping():
+    schema = {
+        'title': 'test',
+        'type': 'object',
+        'properties': {
+            'field': {
+                'title': 'field',
+                'type': 'string'
+            }
+        }
+    }
+    mapping = actionoutout_schema_to_mapping(schema)
+    assert mapping['field'] == (str, ...)
+
+    schema = {
+        'title': 'test',
+        'type': 'object',
+        'properties': {
+            'field': {
+                'title': 'field',
+                'type': 'array',
+                'items': {
+                    'type': 'string'
+                }
+            }
+        }
+    }
+    mapping = actionoutout_schema_to_mapping(schema)
+    assert mapping['field'] == (List[str], ...)
+
+    schema = {
+        'title': 'test',
+        'type': 'object',
+        'properties': {
+            'field': {
+                'title': 'field',
+                'type': 'array',
+                'items': {
+                    'type': 'array',
+                    'minItems': 2,
+                    'maxItems': 2,
+                    'items': [
+                        {
+                            'type': 'string'
+                        },
+                        {
+                            'type': 'string'
+                        }
+                    ]
+                }
+            }
+        }
+    }
+    mapping = actionoutout_schema_to_mapping(schema)
+    assert mapping['field'] == (List[Tuple[str, str]], ...)
+
+    assert True, True
+
+
+def test_serialize_and_deserialize_message():
+    out_mapping = {
+        'field1': (str, ...),
+        'field2': (List[str], ...)
+    }
+    out_data = {
+        'field1': 'field1 value',
+        'field2': ['field2 value1', 'field2 value2']
+    }
+    ic_obj = ActionOutput.create_model_class('prd', out_mapping)
+
+    message = Message(content='prd demand',
+                      instruct_content=ic_obj(**out_data),
+                      role='user',
+                      cause_by=WritePRD)  # WritePRD as test action
+
+    message_ser = serialize_message(message)
+
+    new_message = deserialize_message(message_ser)
+    assert new_message.content == message.content
+    assert new_message.cause_by == message.cause_by
+    assert new_message.instruct_content.field1 == out_data['field1']