Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
72 lines
3.7 KiB
Python
72 lines
3.7 KiB
Python
"""Prompt-cache ordering tests: stable instructions first, time in user messages."""
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
BASE = Path(__file__).resolve().parents[1]
|
|
sys.path[:0] = [str(BASE / "bot-utility"), str(BASE)]
|
|
|
|
from agent import ToolContext
|
|
from chat import engine
|
|
|
|
|
|
class PromptCacheOrderTests(unittest.IsolatedAsyncioTestCase):
|
|
def build_private(self):
|
|
channel = SimpleNamespace(name="tg", is_admin=lambda uid: True)
|
|
ctx = ToolContext(channel=channel, msg_type="private", user_id=42, target_id=10001, is_admin=True)
|
|
with patch.object(engine, "user_persona_prompt", return_value="PERSONA"), \
|
|
patch.object(engine, "load_user_status", return_value={"pause": False, "sleep": False}), \
|
|
patch.object(engine.memory_store, "get_relation", return_value=None), \
|
|
patch.object(engine.memory_store, "list_entities", return_value=[]), \
|
|
patch.object(engine.memory_store, "recall_facts", return_value=[(1, "长期事实", 5)]), \
|
|
patch.object(engine.memory_store, "migrate_legacy"), \
|
|
patch.object(engine, "get_history_within_budget", return_value=["最近历史"]), \
|
|
patch.object(engine.notebook.notebook, "context_for_chat", return_value="笔记本背景"), \
|
|
patch.object(engine, "_sticker_prompt_summary", return_value="贴纸场景"):
|
|
return engine.build_system_prompt(42, query="当前问题", tool_context=ctx)
|
|
|
|
def test_prompt_budget_is_128k(self):
|
|
self.assertEqual(engine.PROMPT_TOKEN_BUDGET, 131072)
|
|
|
|
def test_private_stable_instructions_precede_dynamic_sections(self):
|
|
prompt = self.build_private()
|
|
self.assertNotIn("现在是 ", prompt)
|
|
agent = prompt.index("当你回复时")
|
|
memory = prompt.index("【长期记忆】")
|
|
status = prompt.index("当前用户状态")
|
|
history = prompt.index("以下是最近你和用户的聊天记录")
|
|
self.assertLess(agent, memory)
|
|
self.assertLess(memory, status)
|
|
self.assertLess(status, history)
|
|
self.assertIn("你还可以调用以下 utility 工具", prompt[:history])
|
|
|
|
def test_sticker_prompt_prefers_meaning_over_description(self):
|
|
expected = "理解其情绪、含义和使用场景即可,不要主动描述画面内容"
|
|
self.assertIn(expected, engine.load_instruction("agent_instructions.txt"))
|
|
self.assertIn(expected, engine.load_instruction("group_instructions.txt"))
|
|
|
|
def test_time_is_prepended_to_user_message(self):
|
|
message = engine._message_with_time("用户输入")
|
|
self.assertTrue(message.startswith("现在是 "))
|
|
self.assertTrue(message.endswith("\n用户输入"))
|
|
|
|
async def test_group_prompt_has_stable_prefix_and_time_in_user_message(self):
|
|
response = '{"message":[{"content":"ok"}],"tools":[],"need_reasoning":false}'
|
|
model = AsyncMock(return_value=response)
|
|
with patch.object(engine, "get_recent_group_history", return_value=["群历史"]), \
|
|
patch.object(engine, "call_deepseek", new=model), \
|
|
patch.object(engine, "send_group_message", new=AsyncMock(return_value=True)), \
|
|
patch.object(engine, "append_group_history"):
|
|
await engine._call_group_model_and_reply_impl(None, 1, "群消息", 0)
|
|
system, user = model.await_args.args[:2]
|
|
self.assertNotIn("现在是 ", system)
|
|
self.assertTrue(user.startswith("现在是 "))
|
|
self.assertLess(system.index("当你回复时"), system.index("以下是本群的最近聊天记录"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|