test_micro_agents.py 4.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134
  1. import json
  2. import os
  3. from unittest.mock import MagicMock
  4. import pytest
  5. import yaml
  6. from pytest import TempPathFactory
  7. from openhands.agenthub.micro.registry import all_microagents
  8. from openhands.controller.agent import Agent
  9. from openhands.controller.state.state import State
  10. from openhands.core.config import AgentConfig
  11. from openhands.events.action import MessageAction
  12. from openhands.events.stream import EventStream
  13. from openhands.storage import get_file_store
  14. @pytest.fixture
  15. def temp_dir(tmp_path_factory: TempPathFactory) -> str:
  16. return str(tmp_path_factory.mktemp('test_micro_agents'))
  17. @pytest.fixture
  18. def event_stream(temp_dir):
  19. file_store = get_file_store('local', temp_dir)
  20. event_stream = EventStream('asdf', file_store)
  21. yield event_stream
  22. @pytest.fixture
  23. def agent_configs():
  24. return {
  25. 'CoderAgent': AgentConfig(memory_enabled=True),
  26. 'PlannerAgent': AgentConfig(memory_enabled=True),
  27. }
  28. def test_all_agents_are_loaded():
  29. assert all_microagents is not None
  30. assert len(all_microagents) > 1
  31. base = os.path.join('openhands', 'agenthub', 'micro')
  32. full_path = os.path.dirname(__file__) + '/../../' + base
  33. agent_names = set()
  34. for root, _, files in os.walk(full_path):
  35. for file in files:
  36. if file == 'agent.yaml':
  37. file_path = os.path.join(root, file)
  38. with open(file_path, 'r') as yaml_file:
  39. data = yaml.safe_load(yaml_file)
  40. agent_names.add(data['name'])
  41. assert agent_names == set(all_microagents.keys())
  42. def test_coder_agent_with_summary(event_stream: EventStream, agent_configs: dict):
  43. """Coder agent should render code summary as part of prompt"""
  44. mock_llm = MagicMock()
  45. content = json.dumps({'action': 'finish', 'args': {}})
  46. mock_llm.completion.return_value = {'choices': [{'message': {'content': content}}]}
  47. mock_llm.format_messages_for_llm.return_value = [
  48. {
  49. 'role': 'user',
  50. 'content': "This is a dummy task. This is a dummy summary about this repo. Here's a summary of the codebase, as it relates to this task.",
  51. }
  52. ]
  53. coder_agent = Agent.get_cls('CoderAgent')(
  54. llm=mock_llm, config=agent_configs['CoderAgent']
  55. )
  56. assert coder_agent is not None
  57. # give it some history
  58. task = 'This is a dummy task'
  59. history = list()
  60. history.append(MessageAction(content=task))
  61. summary = 'This is a dummy summary about this repo'
  62. state = State(history=history, inputs={'summary': summary})
  63. coder_agent.step(state)
  64. mock_llm.completion.assert_called_once()
  65. _, kwargs = mock_llm.completion.call_args
  66. prompt_element = kwargs['messages'][0]['content']
  67. if isinstance(prompt_element, dict):
  68. prompt = prompt_element['content']
  69. else:
  70. prompt = prompt_element
  71. assert task in prompt
  72. assert "Here's a summary of the codebase, as it relates to this task" in prompt
  73. assert summary in prompt
  74. def test_coder_agent_without_summary(event_stream: EventStream, agent_configs: dict):
  75. """When there's no codebase_summary available, there shouldn't be any prompt
  76. about 'code summary'
  77. """
  78. mock_llm = MagicMock()
  79. content = json.dumps({'action': 'finish', 'args': {}})
  80. mock_llm.completion.return_value = {'choices': [{'message': {'content': content}}]}
  81. mock_llm.format_messages_for_llm.return_value = [
  82. {
  83. 'role': 'user',
  84. 'content': [
  85. {
  86. 'type': 'text',
  87. 'text': "This is a dummy task. This is a dummy summary about this repo. Here's a summary of the codebase, as it relates to this task.",
  88. }
  89. ],
  90. }
  91. ]
  92. coder_agent = Agent.get_cls('CoderAgent')(
  93. llm=mock_llm, config=agent_configs['CoderAgent']
  94. )
  95. assert coder_agent is not None
  96. # give it some history
  97. task = 'This is a dummy task'
  98. history = list()
  99. history.append(MessageAction(content=task))
  100. # set state without codebase summary
  101. state = State(history=history)
  102. coder_agent.step(state)
  103. mock_llm.completion.assert_called_once()
  104. _, kwargs = mock_llm.completion.call_args
  105. prompt_element = kwargs['messages'][0]['content']
  106. if isinstance(prompt_element, dict):
  107. prompt = prompt_element['content']
  108. else:
  109. prompt = prompt_element
  110. print(f'\n{prompt_element}\n')
  111. assert "Here's a summary of the codebase, as it relates to this task" not in prompt