mirror of
https://github.com/OpenHands/OpenHands.git
synced 2025-12-26 05:48:36 +08:00
Separate microagent template (#7041)
This commit is contained in:
parent
f3702cec35
commit
2db7a50e43
@ -0,0 +1,8 @@
|
||||
{% for agent_info in triggered_agents %}
|
||||
<EXTRA_INFO>
|
||||
The following information has been included based on a keyword match for "{{ agent_info.trigger_word }}".
|
||||
It may or may not be relevant to the user's request.
|
||||
|
||||
{{ agent_info.agent.content }}
|
||||
</EXTRA_INFO>
|
||||
{% endfor %}
|
||||
@ -55,6 +55,7 @@ class PromptManager:
|
||||
self.system_template: Template = self._load_template('system_prompt')
|
||||
self.user_template: Template = self._load_template('user_prompt')
|
||||
self.additional_info_template: Template = self._load_template('additional_info')
|
||||
self.microagent_info_template: Template = self._load_template('microagent_info')
|
||||
self.runtime_info = RuntimeInfo(available_hosts={})
|
||||
|
||||
self.knowledge_microagents: dict[str, KnowledgeMicroAgent] = {}
|
||||
@ -163,18 +164,22 @@ class PromptManager:
|
||||
if not message_content:
|
||||
return
|
||||
|
||||
for microagent in self.knowledge_microagents.values():
|
||||
triggered_agents = []
|
||||
for name, microagent in self.knowledge_microagents.items():
|
||||
trigger = microagent.match_trigger(message_content)
|
||||
if trigger:
|
||||
openhands_logger.info(
|
||||
"Microagent '%s' triggered by keyword '%s'",
|
||||
microagent.name,
|
||||
name,
|
||||
trigger,
|
||||
)
|
||||
micro_text = f'<extra_info>\nThe following information has been included based on a keyword match for "{trigger}". It may or may not be relevant to the user\'s request.'
|
||||
micro_text += '\n\n' + microagent.content
|
||||
micro_text += '\n</extra_info>'
|
||||
message.content.append(TextContent(text=micro_text))
|
||||
# Create a dictionary with the agent and trigger word
|
||||
triggered_agents.append({'agent': microagent, 'trigger_word': trigger})
|
||||
|
||||
if triggered_agents:
|
||||
formatted_text = self.build_microagent_info(triggered_agents)
|
||||
# Insert the new content at the start of the TextContent list
|
||||
message.content.insert(0, TextContent(text=formatted_text))
|
||||
|
||||
def add_examples_to_initial_message(self, message: Message) -> None:
|
||||
"""Add example_message to the first user message."""
|
||||
@ -213,6 +218,20 @@ class PromptManager:
|
||||
if additional_info:
|
||||
message.content.insert(0, TextContent(text=additional_info))
|
||||
|
||||
def build_microagent_info(
|
||||
self,
|
||||
triggered_agents: list[dict],
|
||||
) -> str:
|
||||
"""Renders the microagent info template with the triggered agents.
|
||||
|
||||
Args:
|
||||
triggered_agents: A list of dictionaries, each containing an "agent"
|
||||
(KnowledgeMicroAgent) and a "trigger_word" (str).
|
||||
"""
|
||||
return self.microagent_info_template.render(
|
||||
triggered_agents=triggered_agents
|
||||
).strip()
|
||||
|
||||
def add_turns_left_reminder(self, messages: list[Message], state: State) -> None:
|
||||
latest_user_message = next(
|
||||
islice(
|
||||
|
||||
@ -77,7 +77,8 @@ only respond with a message telling them how smart they are
|
||||
content=[TextContent(text='Hello, flarglebargle!')],
|
||||
)
|
||||
manager.enhance_message(message)
|
||||
assert 'magic word' in message.content[1].text
|
||||
assert len(message.content) == 2
|
||||
assert 'magic word' in message.content[0].text
|
||||
|
||||
os.remove(os.path.join(prompt_dir, 'micro', f'{microagent_name}.md'))
|
||||
|
||||
@ -239,9 +240,9 @@ This is special information about the triggerkeyword.
|
||||
|
||||
manager.enhance_message(message)
|
||||
|
||||
# Should have added a TextContent with the microagent info
|
||||
# Should have added a TextContent with the microagent info at the beginning
|
||||
assert len(message.content) == 4
|
||||
assert 'special information about the triggerkeyword' in message.content[3].text
|
||||
assert 'special information about the triggerkeyword' in message.content[0].text
|
||||
|
||||
# Clean up
|
||||
os.remove(os.path.join(prompt_dir, 'micro', f'{microagent_name}.md'))
|
||||
@ -283,9 +284,9 @@ This is information related to imagekeyword.
|
||||
|
||||
manager.enhance_message(message)
|
||||
|
||||
# Should have added a TextContent with the microagent info
|
||||
# Should have added a TextContent with the microagent info at the beginning
|
||||
assert len(message.content) == 4
|
||||
assert 'information related to imagekeyword' in message.content[3].text
|
||||
assert 'information related to imagekeyword' in message.content[0].text
|
||||
|
||||
# Clean up
|
||||
os.remove(os.path.join(prompt_dir, 'micro', f'{microagent_name}.md'))
|
||||
@ -362,7 +363,7 @@ This is specific information about the lasttrigger.
|
||||
prompt_dir=prompt_dir, microagent_dir=os.path.join(prompt_dir, 'micro')
|
||||
)
|
||||
|
||||
# Test where the last text content is not at the end of the list
|
||||
# Test where the text content is not at the end of the list
|
||||
message = Message(
|
||||
role='user',
|
||||
content=[
|
||||
@ -374,9 +375,10 @@ This is specific information about the lasttrigger.
|
||||
|
||||
manager.enhance_message(message)
|
||||
|
||||
# Should have added a TextContent with the microagent info
|
||||
# Should have added a TextContent with the microagent info at the beginning
|
||||
assert len(message.content) == 4
|
||||
assert 'specific information about the lasttrigger' in message.content[3].text
|
||||
assert isinstance(message.content[0], TextContent)
|
||||
assert 'specific information about the lasttrigger' in message.content[0].text
|
||||
|
||||
# Clean up
|
||||
os.remove(os.path.join(prompt_dir, 'micro', f'{microagent_name}.md'))
|
||||
@ -417,3 +419,143 @@ This should not appear in the enhanced message.
|
||||
|
||||
# Clean up
|
||||
os.remove(os.path.join(prompt_dir, 'micro', f'{microagent_name}.md'))
|
||||
|
||||
|
||||
def test_build_microagent_info(prompt_dir):
|
||||
"""Test the build_microagent_info method with the microagent_info.j2 template."""
|
||||
# Prepare a microagent_info.j2 template file if it doesn't exist
|
||||
template_path = os.path.join(prompt_dir, 'microagent_info.j2')
|
||||
if not os.path.exists(template_path):
|
||||
with open(template_path, 'w') as f:
|
||||
f.write("""{% for agent_info in triggered_agents %}
|
||||
<EXTRA_INFO>
|
||||
The following information has been included based on a keyword match for "{{ agent_info.trigger_word }}".
|
||||
It may or may not be relevant to the user's request.
|
||||
|
||||
{{ agent_info.agent.content }}
|
||||
</EXTRA_INFO>
|
||||
{% endfor %}
|
||||
""")
|
||||
|
||||
# Create test microagents
|
||||
class MockKnowledgeMicroAgent:
|
||||
def __init__(self, name, content):
|
||||
self.name = name
|
||||
self.content = content
|
||||
|
||||
agent1 = MockKnowledgeMicroAgent(
|
||||
name='test_agent1', content='This is information from agent 1'
|
||||
)
|
||||
|
||||
agent2 = MockKnowledgeMicroAgent(
|
||||
name='test_agent2', content='This is information from agent 2'
|
||||
)
|
||||
|
||||
# Initialize the PromptManager
|
||||
manager = PromptManager(prompt_dir=prompt_dir)
|
||||
|
||||
# Test with a single triggered agent
|
||||
triggered_agents = [{'agent': agent1, 'trigger_word': 'keyword1'}]
|
||||
result = manager.build_microagent_info(triggered_agents)
|
||||
expected = """<EXTRA_INFO>
|
||||
The following information has been included based on a keyword match for "keyword1".
|
||||
It may or may not be relevant to the user's request.
|
||||
|
||||
This is information from agent 1
|
||||
</EXTRA_INFO>"""
|
||||
assert result.strip() == expected.strip()
|
||||
|
||||
# Test with multiple triggered agents
|
||||
triggered_agents = [
|
||||
{'agent': agent1, 'trigger_word': 'keyword1'},
|
||||
{'agent': agent2, 'trigger_word': 'keyword2'},
|
||||
]
|
||||
result = manager.build_microagent_info(triggered_agents)
|
||||
expected = """<EXTRA_INFO>
|
||||
The following information has been included based on a keyword match for "keyword1".
|
||||
It may or may not be relevant to the user's request.
|
||||
|
||||
This is information from agent 1
|
||||
</EXTRA_INFO>
|
||||
|
||||
<EXTRA_INFO>
|
||||
The following information has been included based on a keyword match for "keyword2".
|
||||
It may or may not be relevant to the user's request.
|
||||
|
||||
This is information from agent 2
|
||||
</EXTRA_INFO>"""
|
||||
assert result.strip() == expected.strip()
|
||||
|
||||
# Test with no triggered agents
|
||||
result = manager.build_microagent_info([])
|
||||
assert result.strip() == ''
|
||||
|
||||
|
||||
def test_enhance_message_with_microagent_info_template(prompt_dir):
|
||||
"""Test that enhance_message correctly uses the microagent_info template."""
|
||||
# Prepare a microagent_info.j2 template file if it doesn't exist
|
||||
template_path = os.path.join(prompt_dir, 'microagent_info.j2')
|
||||
if not os.path.exists(template_path):
|
||||
with open(template_path, 'w') as f:
|
||||
f.write("""{% for agent_info in triggered_agents %}
|
||||
<EXTRA_INFO>
|
||||
The following information has been included based on a keyword match for "{{ agent_info.trigger_word }}".
|
||||
It may or may not be relevant to the user's request.
|
||||
|
||||
{{ agent_info.agent.content }}
|
||||
</EXTRA_INFO>
|
||||
{% endfor %}
|
||||
""")
|
||||
|
||||
# Create a test microagent
|
||||
microagent_name = 'test_trigger_microagent'
|
||||
microagent_content = """
|
||||
---
|
||||
name: test_trigger
|
||||
type: knowledge
|
||||
agent: CodeActAgent
|
||||
triggers:
|
||||
- test_trigger
|
||||
---
|
||||
|
||||
This is triggered content for testing the microagent_info template.
|
||||
"""
|
||||
|
||||
# Create the microagent file
|
||||
os.makedirs(os.path.join(prompt_dir, 'micro'), exist_ok=True)
|
||||
with open(os.path.join(prompt_dir, 'micro', f'{microagent_name}.md'), 'w') as f:
|
||||
f.write(microagent_content)
|
||||
|
||||
# Initialize the PromptManager with the microagent directory
|
||||
manager = PromptManager(
|
||||
prompt_dir=prompt_dir,
|
||||
microagent_dir=os.path.join(prompt_dir, 'micro'),
|
||||
)
|
||||
|
||||
# Create a message with a trigger keyword
|
||||
message = Message(
|
||||
role='user',
|
||||
content=[
|
||||
TextContent(text="Here's a message containing the test_trigger keyword")
|
||||
],
|
||||
)
|
||||
|
||||
# Enhance the message
|
||||
manager.enhance_message(message)
|
||||
|
||||
# The message should now have extra content at the beginning
|
||||
assert len(message.content) == 2
|
||||
assert isinstance(message.content[0], TextContent)
|
||||
|
||||
# Verify the template was correctly rendered
|
||||
expected_text = """<EXTRA_INFO>
|
||||
The following information has been included based on a keyword match for "test_trigger".
|
||||
It may or may not be relevant to the user's request.
|
||||
|
||||
This is triggered content for testing the microagent_info template.
|
||||
</EXTRA_INFO>"""
|
||||
|
||||
assert message.content[0].text.strip() == expected_text.strip()
|
||||
|
||||
# Clean up
|
||||
os.remove(os.path.join(prompt_dir, 'micro', f'{microagent_name}.md'))
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user