From de6bb12ae58f9d9db9fbb8d529616324d04dc66f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=AD=A6=E9=98=B3?= Date: Thu, 7 May 2026 20:26:59 +0800 Subject: [PATCH] Guard skill and continuation loops --- src/agent_runtime.py | 185 ++++++++++++++++++++++++++++++++++-- tests/test_agent_runtime.py | 173 ++++++++++++++++++++++++++++++++- 2 files changed, 347 insertions(+), 11 deletions(-) diff --git a/src/agent_runtime.py b/src/agent_runtime.py index ed6e43c..7a09f4e 100644 --- a/src/agent_runtime.py +++ b/src/agent_runtime.py @@ -71,6 +71,10 @@ from .session_env_vars import clear_session_env_vars RuntimeEventSink = Callable[[dict[str, object]], None] +# 防止模型在截断续写时无限自我延长，保留少量自动续写空间即可。 +MAX_AUTO_CONTINUATIONS = 3 +CONTINUATION_FINISH_REASONS = {'length', 'max_tokens'} + class _RuntimeEventBuffer(list[dict[str, object]]): def __init__(self, event_sink: RuntimeEventSink | None = None) -> None: @@ -573,6 +577,8 @@ class LocalCodingAgent: file_history = list(existing_file_history) stream_events: list[dict[str, object]] = _RuntimeEventBuffer(event_sink) assistant_response_segments: list[str] = [] + continuation_count = 0 + active_skill_names: set[str] = set() delegated_tasks = sum( 1 for entry in file_history if entry.get('action') in ('delegate_agent', 'Agent') ) @@ -743,12 +749,42 @@ class LocalCodingAgent: if not turn.tool_calls: assistant_response_segments.append(turn.content) - if self._should_continue_response(turn): + if self._is_empty_truncated_response(turn): + final_output = self._build_empty_truncated_response_output() + session.append_assistant( + final_output, + message_id=f'assistant_{len(session.messages)}', + stop_reason='empty_truncated_response', + ) + _append_final_text_stream_events(stream_events, final_output) + result = AgentRunResult( + final_output=final_output, + turns=turn_index, + tool_calls=tool_calls, + transcript=session.transcript(), + events=tuple(stream_events), + usage=total_usage, + total_cost_usd=total_cost_usd, + stop_reason='empty_truncated_response', + file_history=tuple(file_history), + session_id=session_id, + scratchpad_directory=( + str(scratchpad_directory) if scratchpad_directory is not None else None + ), + ) + result = self._persist_session(session, result) + self.last_run_result = result + return result + if self._should_continue_response( + turn, + continuation_count=continuation_count, + ): + continuation_count += 1 session.append_user( self._build_continuation_prompt(), metadata={ 'kind': 'continuation_request', - 'continuation_index': len(assistant_response_segments), + 'continuation_index': continuation_count, }, message_id=f'continuation_{turn_index}', ) @@ -756,12 +792,24 @@ class LocalCodingAgent: { 'type': 'continuation_request', 'reason': turn.finish_reason, - 'continuation_index': len(assistant_response_segments), + 'continuation_index': continuation_count, } ) last_content = ''.join(assistant_response_segments) continue final_output = ''.join(assistant_response_segments) + if self._reached_auto_continuation_limit( + turn, + continuation_count=continuation_count, + ): + stream_events.append( + { + 'type': 'continuation_limit_reached', + 'reason': turn.finish_reason, + 'max_continuations': MAX_AUTO_CONTINUATIONS, + } + ) + final_output = self._append_continuation_limit_note(final_output) _append_final_text_stream_events(stream_events, final_output) result = AgentRunResult( final_output=final_output, @@ -846,12 +894,47 @@ class LocalCodingAgent: if not turn.tool_calls: assistant_response_segments.append(turn.content) - if self._should_continue_response(turn): + if self._is_empty_truncated_response(turn): + final_output = self._build_empty_truncated_response_output() + session.append_assistant( + final_output, + message_id=f'assistant_{len(session.messages)}', + stop_reason='empty_truncated_response', + ) + _append_final_text_stream_events(stream_events, final_output) + result = AgentRunResult( + final_output=final_output, + turns=turn_index, + tool_calls=tool_calls, + transcript=session.transcript(), + events=tuple(stream_events), + usage=total_usage, + total_cost_usd=total_cost_usd, + stop_reason='empty_truncated_response', + file_history=tuple(file_history), + session_id=session_id, + scratchpad_directory=( + str(scratchpad_directory) if scratchpad_directory is not None else None + ), + ) + result = self._append_runtime_after_turn_events( + result, + prompt=effective_prompt, + turn_index=turn_index, + ) + result = self._persist_session(session, result) + self.last_run_result = result + return result + if self._should_continue_response( + turn, + continuation_count=continuation_count, + ): + continuation_count += 1 session.append_user( self._build_continuation_prompt(), metadata={ 'kind': 'continuation_request', - 'continuation_index': len(assistant_response_segments), + 'continuation_index': continuation_count, }, message_id=f'continuation_{turn_index}', ) @@ -859,12 +942,24 @@ class LocalCodingAgent: { 'type': 'continuation_request', 'reason': turn.finish_reason, - 'continuation_index': len(assistant_response_segments), + 'continuation_index': continuation_count, } ) last_content = ''.join(assistant_response_segments) continue final_output = ''.join(assistant_response_segments) + if self._reached_auto_continuation_limit( + turn, + continuation_count=continuation_count, + ): + stream_events.append( + { + 'type': 'continuation_limit_reached', + 'reason': turn.finish_reason, + 'max_continuations': MAX_AUTO_CONTINUATIONS, + } + ) + final_output = self._append_continuation_limit_note(final_output) _append_final_text_stream_events(stream_events, final_output) result = AgentRunResult( final_output=final_output, @@ -892,6 +987,7 @@ class LocalCodingAgent: for tool_call in turn.tool_calls: assistant_response_segments.clear() + continuation_count = 0 tool_calls += 1 if tool_call.name in ('Agent', 'delegate_agent'): delegated_tasks += self._delegated_task_units(tool_call.arguments) @@ -1032,7 +1128,10 @@ class LocalCodingAgent: tool_result = self._execute_delegate_agent(tool_call.arguments) elif tool_call.name == 'Skill': if tool_result is None: - tool_result = self._execute_skill(tool_call.arguments) + tool_result = self._execute_skill( + tool_call.arguments, + active_skill_names=active_skill_names, + ) elif tool_result is None: for update in execute_tool_streaming( self.tool_registry, @@ -1431,8 +1530,52 @@ class LocalCodingAgent: ) return tuple(parsed) - def _should_continue_response(self, turn: AssistantTurn) -> bool: - return turn.finish_reason in {'length', 'max_tokens'} + def _should_continue_response( + self, + turn: AssistantTurn, + *, + continuation_count: int = 0, + ) -> bool: + if turn.finish_reason not in CONTINUATION_FINISH_REASONS: + return False + if continuation_count >= MAX_AUTO_CONTINUATIONS: + return False + return bool((turn.content or '').strip()) + + def _is_empty_truncated_response(self, turn: AssistantTurn) -> bool: + return ( + turn.finish_reason in CONTINUATION_FINISH_REASONS + and not (turn.content or '').strip() + and not turn.tool_calls + ) + + def _reached_auto_continuation_limit( + self, + turn: AssistantTurn, + *, + continuation_count: int, + ) -> bool: + return ( + turn.finish_reason in CONTINUATION_FINISH_REASONS + and bool((turn.content or '').strip()) + and continuation_count >= MAX_AUTO_CONTINUATIONS + ) + + def _append_continuation_limit_note(self, text: str) -> str: + note = ( + f'[系统提示] 已达到自动续写上限 {MAX_AUTO_CONTINUATIONS} 次，' + '本轮先暂停，避免模型无限续写。你可以回复“继续”再接着处理。' + ) + if not text.strip(): + return note + return f'{text.rstrip()}\n\n{note}' + + def _build_empty_truncated_response_output(self) -> str: + return ( + '模型返回了空的截断响应，本轮已暂停。\n\n' + '这通常表示模型后端输出被截断但没有返回可展示内容。' + '你可以回复“继续”让我接着处理，或补充更具体的指令重新发起。' + ) def _build_continuation_prompt(self) -> str: return ( @@ -2347,6 +2490,8 @@ class LocalCodingAgent: def _execute_skill( self, arguments: dict[str, object], + *, + active_skill_names: set[str] | None = None, ) -> ToolExecutionResult: """Execute a skill through the Skill tool. @@ -2372,6 +2517,26 @@ class LocalCodingAgent: # 1. Check bundled skills first bundled = find_bundled_skill(skill_name, cwd=self.runtime_config.cwd) if bundled is not None: + skill_key = bundled.name.strip().lower() + if active_skill_names is not None and skill_key in active_skill_names: + return ToolExecutionResult( + name='Skill', + ok=True, + content=( + f'Skill "{bundled.name}" 已在本轮激活。' + '请直接依据前面注入的 Skill 指南继续执行，不要重复调用 Skill 工具。' + ), + metadata={ + 'action': 'skill_duplicate', + 'skill_name': bundled.name, + 'source': bundled.source, + 'skill_path': bundled.path, + 'should_query': True, + 'duplicate': True, + }, + ) + if active_skill_names is not None: + active_skill_names.add(skill_key) prompt = bundled.get_prompt(self, args.strip()) return ToolExecutionResult( name='Skill', @@ -2379,7 +2544,7 @@ class LocalCodingAgent: content=prompt, metadata={ 'action': 'skill', - 'skill_name': skill_name, + 'skill_name': bundled.name, 'source': bundled.source, 'skill_path': bundled.path, 'should_query': True, diff --git a/tests/test_agent_runtime.py b/tests/test_agent_runtime.py index 26bfad2..1ef4526 100644 --- a/tests/test_agent_runtime.py +++ b/tests/test_agent_runtime.py @@ -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 = [ {