diff --git a/agentmain.py b/agentmain.py index 6551b995b..99e9b215e 100644 --- a/agentmain.py +++ b/agentmain.py @@ -160,6 +160,7 @@ def run(self): if raw_query is None: self.task_queue.task_done(); continue self.is_running = True; self._current_queue = display_queue + is_autonomous = raw_query.lstrip().startswith('[AUTO]') if len(raw_query) > 2000: task_file = os.path.join(script_dir, 'temp', f'user_prompt_{os.getpid()}_{time.time_ns()}.md') with open(task_file, 'w', encoding='utf-8') as f: f.write(raw_query) @@ -170,7 +171,9 @@ def run(self): self.history.append(f"[USER]: {rquery}") sys_prompt = get_system_prompt() + '\n'.join(self.extra_sys_prompts) + getattr(self.llmclient.backend, 'extra_sys_prompt', '') if self.peer_hint: sys_prompt += f"\n[Peer] 用户提及其他会话/后台任务状态时: temp/model_responses/ (只找近期修改的文件尾部)\n" - handler = GenericAgentHandler(self, self.history, os.path.join(script_dir, 'temp')) + long_term_tool_enabled = any(t.get('function', {}).get('name') == 'start_long_term_update' for t in TOOLS_SCHEMA) + handler = GenericAgentHandler(self, self.history, os.path.join(script_dir, 'temp'), + long_term_update_pending=long_term_tool_enabled and not is_autonomous) if getattr(self, 'no_print', False): handler.print = lambda *a, **k: None if self.handler and 'key_info' in self.handler.working: ki = re.sub(r'\n\[SYSTEM\] 此为.*?工作记忆[。\n]*', '', self.handler.working['key_info']) # 去旧 diff --git a/assets/tools_schema.json b/assets/tools_schema.json index 81c52437a..1c8aa1377 100644 --- a/assets/tools_schema.json +++ b/assets/tools_schema.json @@ -66,7 +66,7 @@ }}, {"type": "function", "function": { "name": "start_long_term_update", - "description": "After a long task completes, distill long-term memory. Call when discovering info worth remembering (env facts/user prefs/lessons learned). Skip if memory already updated or in autonomous flow. Must call when a task that took 15+ turns is completed.", + "description": "After a long task completes, distill long-term memory. Call when discovering info worth remembering (env facts/user prefs/lessons learned). Skip if memory already updated or in autonomous flow. Must call when a task that took 15 or more turns is completed.", "parameters": {"type": "object", "properties": {}}} } ] diff --git a/assets/tools_schema_cn.json b/assets/tools_schema_cn.json index 98fbc211b..9a6107045 100644 --- a/assets/tools_schema_cn.json +++ b/assets/tools_schema_cn.json @@ -67,7 +67,7 @@ }}, {"type": "function", "function": { "name": "start_long_term_update", - "description": "准备开始提炼记忆。发现值得长期记忆的信息(环境事实/用户偏好/避坑经验)时调用此工具。已记忆更新或在自主流程内时无需调用。超15轮完成的任务必须调用以沉淀经验", + "description": "准备开始提炼记忆。发现值得长期记忆的信息(环境事实/用户偏好/避坑经验)时调用此工具。已记忆更新或在自主流程内时无需调用。达到15轮后完成的任务必须调用以沉淀经验", "parameters": {"type": "object", "properties": {}}} } -] \ No newline at end of file +] diff --git a/ga.py b/ga.py index e1ee909f2..13bbf4d29 100644 --- a/ga.py +++ b/ga.py @@ -279,10 +279,13 @@ def consume_file(dr, file): class GenericAgentHandler(BaseHandler): '''Generic Agent 工具库,包含多种工具的实现。工具函数自动加上了 do_ 前缀。实际工具名没有前缀。''' - def __init__(self, parent, last_history=None, cwd='./temp'): + LONG_TERM_UPDATE_MIN_TURN = 15 + + def __init__(self, parent, last_history=None, cwd='./temp', long_term_update_pending=True): self.parent = parent self.working = {} self.cwd = cwd; self.current_turn = 0 + self.long_term_update_pending = long_term_update_pending self.history_info = last_history if last_history else [] self.code_stop_signal = [] self._done_hooks = [] @@ -520,12 +523,18 @@ def do_no_tool(self, args, response): remaining = self._check_plan_completion() if remaining == 0: self._exit_plan_mode(); yield "[Info] Plan完成:plan.md中0个[ ]残留,退出plan模式。\n" + + if (self.current_turn >= self.LONG_TERM_UPDATE_MIN_TURN + and self.long_term_update_pending): + self.long_term_update_pending = False + yield "[Info] Auto-calling start_long_term_update before exit.\n" + outcome = yield from self.do_start_long_term_update({}, response) + return StepOutcome(None, next_prompt=f'{outcome.data}\n{outcome.next_prompt}') #yield "[Info] Final response to user.\n" return StepOutcome(response, next_prompt=None) - - def do_start_long_term_update(self, args, response): - '''Agent觉得当前任务完成后有重要信息需要记忆时调用此工具。''' + + def _begin_long_term_update(self): prompt = '''### [总结提炼经验] 既然你觉得当前任务有重要信息需要记忆,请提取最近一次任务中【事实验证成功且长期有效】的环境事实、用户偏好、重要步骤,更新记忆。 本工具是标记开启结算过程,若已在更新记忆过程或没有值得记忆的点,忽略本次调用。 **如果没有经验证的,未来能用上的信息,忽略本次调用!** @@ -535,13 +544,19 @@ def do_start_long_term_update(self, args, response): **禁止**:临时变量、具体推理过程、未验证信息、通用常识、你可以轻松复现的细节、只是做了但没有验证的信息 **操作**:严格遵循提供的L0的记忆更新SOP。先 `file_read` 看现有 → 判断类型 → 最小化更新 → 无新内容跳过,保证对记忆库最小局部修改。\n ''' + get_global_memory() - yield "[Info] Start distilling good memory for long-term storage.\n" path = './memory/memory_management_sop.md' if os.path.exists(path): result = 'This is L0:\n' + file_read(path, show_linenos=False) else: result = "Memory Management SOP not found. Do not update memory." - if self.current_turn < 10: result, prompt = 'start_long_term_update is only used after completing a long turn task!', '\n' return StepOutcome(result, next_prompt=prompt) + def do_start_long_term_update(self, args, response): + '''Agent觉得当前任务完成后有重要信息需要记忆时调用此工具。''' + if self.current_turn < 10: + return StepOutcome('start_long_term_update is only used after completing a long turn task!', next_prompt='\n') + yield "[Info] Start distilling good memory for long-term storage.\n" + self.long_term_update_pending = False + return self._begin_long_term_update() + def _fold_earlier(self, lines): FALLBACK = '直接回答了用户问题' parts, cnt, last = [], 0, '' diff --git a/tests/test_long_term_update_gate.py b/tests/test_long_term_update_gate.py new file mode 100644 index 000000000..56c141503 --- /dev/null +++ b/tests/test_long_term_update_gate.py @@ -0,0 +1,103 @@ +import importlib.util +import sys +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] + + +def load_module(name, path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +previous_agent_loop = sys.modules.get('agent_loop') +real_agent_loop = load_module('_long_term_gate_agent_loop', ROOT / 'agent_loop.py') +sys.modules['agent_loop'] = real_agent_loop +try: + ga = load_module('_long_term_gate_ga', ROOT / 'ga.py') +finally: + if previous_agent_loop is None: + del sys.modules['agent_loop'] + else: + sys.modules['agent_loop'] = previous_agent_loop + +GenericAgentHandler = ga.GenericAgentHandler + + +class Parent: + def get_ctx_multiplier(self): + return 1 + + +def exhaust(generator): + try: + while True: + next(generator) + except StopIteration as exc: + return exc.value + + +class LongTermUpdateGateTests(unittest.TestCase): + def handler(self, enabled=True): + return GenericAgentHandler(Parent(), cwd='./temp', long_term_update_pending=enabled) + + def finish(self, handler, turn): + handler.current_turn = turn + response = SimpleNamespace(content='Task complete.', thinking='') + return exhaust(handler.do_no_tool({}, response)) + + def test_turn_before_threshold_exits_normally(self): + handler = self.handler() + outcome = self.finish(handler, 14) + self.assertIsNone(outcome.next_prompt) + self.assertTrue(handler.long_term_update_pending) + + @patch.object(ga, 'file_read', return_value='memory SOP') + @patch.object(ga, 'get_global_memory', return_value='memory index') + @patch.object(ga.os.path, 'exists', return_value=True) + def test_threshold_starts_evaluation_once(self, _exists, _memory, _read): + handler = self.handler() + outcome = self.finish(handler, 15) + self.assertIn('memory SOP', outcome.next_prompt) + self.assertIn('总结提炼经验', outcome.next_prompt) + self.assertFalse(handler.long_term_update_pending) + + second = self.finish(handler, 16) + self.assertIsNone(second.next_prompt) + + def test_disabled_gate_exits_normally(self): + handler = self.handler(enabled=False) + outcome = self.finish(handler, 20) + self.assertIsNone(outcome.next_prompt) + self.assertFalse(handler.long_term_update_pending) + + @patch.object(ga, 'file_read', return_value='memory SOP') + @patch.object(ga, 'get_global_memory', return_value='memory index') + @patch.object(ga.os.path, 'exists', return_value=True) + def test_explicit_update_consumes_completion_gate(self, _exists, _memory, _read): + handler = self.handler() + handler.current_turn = 10 + exhaust(handler.do_start_long_term_update({}, SimpleNamespace())) + self.assertFalse(handler.long_term_update_pending) + + outcome = self.finish(handler, 20) + self.assertIsNone(outcome.next_prompt) + self.assertFalse(handler.long_term_update_pending) + + def test_rejected_early_call_keeps_completion_gate_pending(self): + handler = self.handler() + handler.current_turn = 9 + outcome = exhaust(handler.do_start_long_term_update({}, SimpleNamespace())) + self.assertEqual('\n', outcome.next_prompt) + self.assertTrue(handler.long_term_update_pending) + + +if __name__ == '__main__': + unittest.main()