Guard skill and continuation loops
This commit is contained in:
武阳
+172
-1
@@ -8,7 +8,7 @@ from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from src.agent_session import AgentMessage
|
||||
from src.agent_runtime import LocalCodingAgent
|
||||
from src.agent_runtime import LocalCodingAgent, MAX_AUTO_CONTINUATIONS
|
||||
from src.agent_tools import build_tool_context, default_tool_registry, execute_tool
|
||||
from src.agent_types import (
|
||||
AgentPermissions,
|
||||
@@ -1215,6 +1215,177 @@ class AgentRuntimeTests(unittest.TestCase):
|
||||
)
|
||||
)
|
||||
|
||||
def test_agent_stops_when_truncated_response_is_empty(self) -> None:
|
||||
responses = [
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': '',
|
||||
},
|
||||
'finish_reason': 'length',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 10, 'completion_tokens': 0},
|
||||
},
|
||||
]
|
||||
recorded_payloads: list[dict[str, object]] = []
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
with patch(
|
||||
'src.openai_compat.request.urlopen',
|
||||
side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
|
||||
):
|
||||
agent = LocalCodingAgent(
|
||||
model_config=ModelConfig(
|
||||
model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
|
||||
base_url='http://127.0.0.1:8000/v1',
|
||||
),
|
||||
runtime_config=AgentRuntimeConfig(cwd=workspace),
|
||||
)
|
||||
result = agent.run('Give me a long answer')
|
||||
|
||||
self.assertEqual(len(recorded_payloads), 1)
|
||||
self.assertEqual(result.stop_reason, 'empty_truncated_response')
|
||||
self.assertIn('空的截断响应', result.final_output)
|
||||
self.assertFalse(
|
||||
any(event.get('type') == 'continuation_request' for event in result.events)
|
||||
)
|
||||
self.assertEqual(result.transcript[-1]['stop_reason'], 'empty_truncated_response')
|
||||
|
||||
def test_agent_limits_automatic_truncated_response_continuations(self) -> None:
|
||||
responses = [
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': f'Part {index} ',
|
||||
},
|
||||
'finish_reason': 'length',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 4, 'completion_tokens': 2},
|
||||
}
|
||||
for index in range(MAX_AUTO_CONTINUATIONS + 1)
|
||||
]
|
||||
recorded_payloads: list[dict[str, object]] = []
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
with patch(
|
||||
'src.openai_compat.request.urlopen',
|
||||
side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
|
||||
):
|
||||
agent = LocalCodingAgent(
|
||||
model_config=ModelConfig(
|
||||
model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
|
||||
base_url='http://127.0.0.1:8000/v1',
|
||||
),
|
||||
runtime_config=AgentRuntimeConfig(cwd=workspace),
|
||||
)
|
||||
result = agent.run('Give me a very long answer')
|
||||
|
||||
continuation_events = [
|
||||
event for event in result.events if event.get('type') == 'continuation_request'
|
||||
]
|
||||
self.assertEqual(len(recorded_payloads), MAX_AUTO_CONTINUATIONS + 1)
|
||||
self.assertEqual(len(continuation_events), MAX_AUTO_CONTINUATIONS)
|
||||
self.assertEqual(result.stop_reason, 'length')
|
||||
self.assertIn(f'Part {MAX_AUTO_CONTINUATIONS}', result.final_output)
|
||||
self.assertIn('自动续写上限', result.final_output)
|
||||
self.assertTrue(
|
||||
any(event.get('type') == 'continuation_limit_reached' for event in result.events)
|
||||
)
|
||||
|
||||
def test_agent_short_circuits_duplicate_skill_invocation_in_one_run(self) -> None:
|
||||
responses = [
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': 'I will load the verification skill.',
|
||||
'tool_calls': [
|
||||
{
|
||||
'id': 'call_1',
|
||||
'type': 'function',
|
||||
'function': {
|
||||
'name': 'Skill',
|
||||
'arguments': '{"skill": "verify"}',
|
||||
},
|
||||
}
|
||||
],
|
||||
},
|
||||
'finish_reason': 'tool_calls',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
|
||||
},
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': 'I accidentally load it again.',
|
||||
'tool_calls': [
|
||||
{
|
||||
'id': 'call_2',
|
||||
'type': 'function',
|
||||
'function': {
|
||||
'name': 'Skill',
|
||||
'arguments': '{"skill": "verify"}',
|
||||
},
|
||||
}
|
||||
],
|
||||
},
|
||||
'finish_reason': 'tool_calls',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
|
||||
},
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': 'Done.',
|
||||
},
|
||||
'finish_reason': 'stop',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 8, 'completion_tokens': 1},
|
||||
},
|
||||
]
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
with patch(
|
||||
'src.openai_compat.request.urlopen',
|
||||
side_effect=make_urlopen_side_effect(responses),
|
||||
):
|
||||
agent = LocalCodingAgent(
|
||||
model_config=ModelConfig(
|
||||
model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
|
||||
base_url='http://127.0.0.1:8000/v1',
|
||||
),
|
||||
runtime_config=AgentRuntimeConfig(cwd=workspace),
|
||||
)
|
||||
result = agent.run('Use verify skill')
|
||||
|
||||
self.assertEqual(result.final_output, 'Done.')
|
||||
self.assertEqual(result.tool_calls, 2)
|
||||
tool_messages = [
|
||||
message for message in result.transcript if message.get('role') == 'tool'
|
||||
]
|
||||
self.assertEqual(len(tool_messages), 2)
|
||||
self.assertEqual(tool_messages[0].get('metadata', {}).get('action'), 'skill')
|
||||
self.assertEqual(
|
||||
tool_messages[1].get('metadata', {}).get('action'),
|
||||
'skill_duplicate',
|
||||
)
|
||||
duplicate_payload = json.loads(str(tool_messages[1].get('content', '{}')))
|
||||
self.assertIn('已在本轮激活', duplicate_payload.get('content', ''))
|
||||
|
||||
def test_agent_records_file_history_for_write_tool(self) -> None:
|
||||
responses = [
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user