determine the content to be saved to the experience pool using cmd_prompt_exp_part.

This commit is contained in:
seehi 2024-07-12 19:30:33 +08:00
parent 39360b41c5
commit bf21bbf12e
13 changed files with 113 additions and 139 deletions

View file

@ -25,27 +25,31 @@ class TestRoleZeroContextBuilder:
async def test_build_with_experiences(self, context_builder, mocker):
mocker.patch.object(BaseContextBuilder, "format_exps", return_value="Formatted experiences")
mocker.patch.object(RoleZeroContextBuilder, "replace_example_content", return_value="Updated content")
req = [{"content": "Original content"}]
req = [{"content": "Original content 1"}, {"content": "Original content exp part"}]
result = await context_builder.build(req=req)
assert result == [{"content": "Updated content"}]
assert result == [{"content": "Updated content"}, {"content": "Original content exp part"}]
def test_replace_example_content(self, context_builder, mocker):
mocker.patch.object(RoleZeroContextBuilder, "replace_content_between_markers", return_value="Replaced content")
result = context_builder.replace_example_content("Original text", "New example content")
assert result == "Replaced content"
context_builder.replace_content_between_markers.assert_called_once_with(
"Original text", "# Example", "# Instruction", "New example content"
"Original text", "# Example", "# Available Commands", "New example content"
)
def test_replace_content_between_markers(self):
text = "Start\n# Example\nOld content\n# Instruction\nEnd"
text = "Start\n# Example\nOld content\n# Available Commands\nEnd"
new_content = "New content"
result = RoleZeroContextBuilder.replace_content_between_markers(text, "# Example", "# Instruction", new_content)
expected = "Start\n# Example\nNew content\n\n# Instruction\nEnd"
result = RoleZeroContextBuilder.replace_content_between_markers(
text, "# Example", "# Available Commands", new_content
)
expected = "Start\n# Example\nNew content\n\n# Available Commands\nEnd"
assert result == expected
def test_replace_content_between_markers_no_match(self):
text = "Start\nNo markers\nEnd"
new_content = "New content"
result = RoleZeroContextBuilder.replace_content_between_markers(text, "# Example", "# Instruction", new_content)
result = RoleZeroContextBuilder.replace_content_between_markers(
text, "# Example", "# Available Commands", new_content
)
assert result == text

View file

@ -13,7 +13,7 @@ class TestSimpleContextBuilder:
return SimpleContextBuilder()
@pytest.mark.asyncio
async def test_build_with_experiences(self, context_builder, mocker):
async def test_build_with_experiences(self, mocker, context_builder: SimpleContextBuilder):
# Mock the format_exps method
mock_exps = "Mocked experiences"
mocker.patch.object(BaseContextBuilder, "format_exps", return_value=mock_exps)
@ -25,7 +25,7 @@ class TestSimpleContextBuilder:
assert result == expected
@pytest.mark.asyncio
async def test_build_without_experiences(self, context_builder, mocker):
async def test_build_without_experiences(self, mocker, context_builder: SimpleContextBuilder):
# Mock the format_exps method to return an empty string
mocker.patch.object(BaseContextBuilder, "format_exps", return_value="")
@ -36,12 +36,12 @@ class TestSimpleContextBuilder:
assert result == expected
@pytest.mark.asyncio
async def test_build_without_req(self, context_builder, mocker):
async def test_build_without_req(self, mocker, context_builder: SimpleContextBuilder):
# Mock the format_exps method
mock_exps = "Mocked experiences"
mocker.patch.object(BaseContextBuilder, "format_exps", return_value=mock_exps)
result = await context_builder.build()
result = await context_builder.build(req="")
expected = SIMPLE_CONTEXT_TEMPLATE.format(req="", exps=mock_exps)
assert result == expected

View file

@ -7,71 +7,42 @@ from metagpt.exp_pool.serializers import RoleZeroSerializer
class TestRoleZeroSerializer:
@pytest.fixture
def serializer(self):
def serializer(self) -> RoleZeroSerializer:
return RoleZeroSerializer()
@pytest.fixture
def last_item(self) -> dict:
return {
"role": "user",
"content": "# Current Plan\nsome plan\n# Current Plan\nsome plan\n# Instruction\nsome instruction",
}
@pytest.fixture
def sample_req(self):
return [{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]
def test_serialize_req_empty_input(self, serializer: RoleZeroSerializer):
assert serializer.serialize_req([]) == ""
def test_serialize_req_with_content(self, serializer: RoleZeroSerializer):
def test_serialize_req_with_content(self, serializer: RoleZeroSerializer, last_item: dict):
req = [
{"content": "Command Editor.read executed: file_path=test.py"},
{"content": "Some other content"},
{
"content": "# Data Structure\nsome data\n# Current Plan\nsome plan\n# Example\nsome example\n# Instruction\nsome instruction"
},
{"role": "user", "content": "Command Editor.read executed: file_path=test.py"},
{"role": "assistant", "content": "Some other content"},
last_item,
]
expected_output = json.dumps(
[
{"content": "Command Editor.read executed: file_path=test.py"},
{
"content": "# Data Structure\n\n\n# Current Plan\nsome plan\n# Example\n\n\n# Instruction\nsome instruction"
},
]
[{"role": "user", "content": "Command Editor.read executed: file_path=test.py"}, last_item]
)
assert serializer.serialize_req(req) == expected_output
def test_filter_req(self, serializer: RoleZeroSerializer):
req = [
{"content": "Command Editor.read executed: file_path=test1.py"},
{"content": "Some other content"},
{"content": "Command Editor.read executed: file_path=test2.py"},
{"content": "Final content"},
{"role": "user", "content": "Command Editor.read executed: file_path=test1.py"},
{"role": "assistant", "content": "Some other content"},
{"role": "user", "content": "Command Editor.read executed: file_path=test2.py"},
{"role": "assistant", "content": "Final content"},
]
filtered_req = serializer._filter_req(req)
assert len(filtered_req) == 3
assert len(filtered_req) == 2
assert filtered_req[0]["content"] == "Command Editor.read executed: file_path=test1.py"
assert filtered_req[1]["content"] == "Command Editor.read executed: file_path=test2.py"
assert filtered_req[2]["content"] == "Final content"
def test_clean_last_entry_content(self, serializer: RoleZeroSerializer):
req = [
{"content": "Some content"},
{
"content": "# Data Structure\nsome data\n# Current Plan\nsome plan\n# Example\nsome example\n# Instruction\nsome instruction"
},
]
serializer._clean_last_entry_content(req)
expected_content = (
"# Data Structure\n\n\n# Current Plan\nsome plan\n# Example\n\n\n# Instruction\nsome instruction"
)
assert req[-1]["content"] == expected_content
def test_integration(self, serializer: RoleZeroSerializer):
req = [
{"content": "Command Editor.read executed: file_path=test.py"},
{"content": "Some other content"},
{
"content": "# Data Structure\nsome data\n# Current Plan\nsome plan\n# Example\nsome example\n# Instruction\nsome instruction"
},
]
result = serializer.serialize_req(req)
expected_output = json.dumps(
[
{"content": "Command Editor.read executed: file_path=test.py"},
{
"content": "# Data Structure\n\n\n# Current Plan\nsome plan\n# Example\n\n\n# Instruction\nsome instruction"
},
]
)
assert result == expected_output