diff --git a/prompts/action_prompt.prompt b/prompts/action_prompt.prompt new file mode 100644 index 00000000..91831b2a --- /dev/null +++ b/prompts/action_prompt.prompt @@ -0,0 +1,5 @@ +{action_name} +动作描述：{action_description} +使用条件{parallel_text}： +{action_require} +{{"action":"{action_name}",{action_parameters}, "target_message_id":"消息id(m+数字)"}} \ No newline at end of file diff --git a/prompts/jargon_explainer_summarize_prompt.prompt b/prompts/jargon_explainer_summarize_prompt.prompt new file mode 100644 index 00000000..427d9f05 --- /dev/null +++ b/prompts/jargon_explainer_summarize_prompt.prompt @@ -0,0 +1,11 @@ +上下文聊天内容: +{chat_context} + +在上下文中提取到的黑话及其含义: +{jargon_explanations} + +请根据上述信息，对黑话解释进行概括和整理。 +- 如果上下文中有黑话出现，请简要说明这些黑话在上下文中的使用情况 +- 将所有黑话解释整理成简洁、易读的一段话 +- 输出格式要自然，适合作为回复参考信息 +请输出概括后的黑话解释（直接输出一段平文本，不要标题，无特殊格式或markdown格式，不要使用JSON格式）： \ No newline at end of file diff --git a/prompts/planner_prompt.prompt b/prompts/planner_prompt.prompt new file mode 100644 index 00000000..d6bf0de7 --- /dev/null +++ b/prompts/planner_prompt.prompt @@ -0,0 +1,44 @@ +{time_block} +{name_block} +{chat_context_description}，以下是具体的聊天内容 +**聊天内容** +{chat_content_block} + +**可选的action** +reply +动作描述： +1.你可以选择呼叫了你的名字，但是你没有做出回应的消息进行回复 +2.你可以自然的顺着正在进行的聊天内容进行回复或自然的提出一个问题 +3.最好一次对一个话题进行回复，免得啰嗦或者回复内容太乱。 +4.不要选择回复你自己发送的消息 +5.不要单独对表情包进行回复 +6.将上下文中所有含义不明的，疑似黑话的，缩写词均写入unknown_words中 +{reply_action_example} + +no_reply +动作描述： +保持沉默，不回复直到有新消息 +控制聊天频率，不要太过频繁的发言 +{{"action":"no_reply"}} + +{action_options_text} + +**你之前的action执行和思考记录** +{actions_before_now_block} + +请选择**可选的**且符合使用条件的action，并说明触发action的消息id(消息id格式:m+数字) +先输出你的简短的选择思考理由，再输出你选择的action，理由不要分点，精简。 +**动作选择要求** +请你根据聊天内容,用户的最新消息和以下标准选择合适的动作: +{plan_style} +{moderation_prompt} + +target_message_id为必填，表示触发消息的id +请选择所有符合使用要求的action，每个动作最多选择一次，但是可以选择多个动作； +动作用json格式输出，用```json包裹，如果输出多个json，每个json都要单独一行放在同一个```json代码块内: +**示例** +// 理由文本（简短） +```json +{{"action":"动作名", "target_message_id":"m123", .....}} +{{"action":"动作名", "target_message_id":"m456", .....}} +``` \ No newline at end of file diff --git a/prompts/replyer_prompt.prompt b/prompts/replyer_prompt.prompt new file mode 100644 index 00000000..4da5c062 --- /dev/null +++ b/prompts/replyer_prompt.prompt @@ -0,0 +1,18 @@ +{knowledge_prompt}{tool_info_block}{extra_info_block} +{expression_habits_block}{memory_retrieval}{jargon_explanation} + +你正在qq群里聊天，下面是群里正在聊的内容，其中包含聊天记录和聊天中的图片 +其中标注 {bot_name}(你) 的发言是你自己的发言，请注意区分: +{time_block} +{dialogue_prompt} + +{reply_target_block}。 +{planner_reasoning} +{identity} +{chat_prompt}你正在群里聊天,现在请你读读之前的聊天记录，把握当前的话题，然后给出日常且简短的回复。 +最好一次对一个话题进行回复，免得啰嗦或者回复内容太乱。 +{keywords_reaction_prompt} +请注意把握聊天内容。 +{reply_style} +请注意不要输出多余内容(包括不必要的前后缀，冒号，括号，at或 @等 )，只输出发言内容就好。 +现在，你说： \ No newline at end of file diff --git a/prompts/replyer_prompt_0.prompt b/prompts/replyer_prompt_0.prompt new file mode 100644 index 00000000..8e3a425a --- /dev/null +++ b/prompts/replyer_prompt_0.prompt @@ -0,0 +1,18 @@ +{knowledge_prompt}{tool_info_block}{extra_info_block} +{expression_habits_block}{memory_retrieval}{jargon_explanation} + +你正在qq群里聊天，下面是群里正在聊的内容，其中包含聊天记录和聊天中的图片 +其中标注 {bot_name}(你) 的发言是你自己的发言，请注意区分: +{time_block} +{dialogue_prompt} + +{reply_target_block}。 +{planner_reasoning} +{identity} +{chat_prompt}你正在群里聊天,现在请你读读之前的聊天记录，然后给出日常且口语化的回复， +尽量简短一些。{keywords_reaction_prompt} +请注意把握聊天内容，不要回复的太有条理。 +{reply_style} +请注意不要输出多余内容(包括不必要的前后缀，冒号，括号，表情包，at或 @等 )，只输出发言内容就好。 +最好一次对一个话题进行回复，免得啰嗦或者回复内容太乱。 +现在，你说： \ No newline at end of file diff --git a/pytests/prompt_test/test_prompt_manager.py b/pytests/prompt_test/test_prompt_manager.py index e4852936..f16d3dfe 100644 --- a/pytests/prompt_test/test_prompt_manager.py +++ b/pytests/prompt_test/test_prompt_manager.py @@ -1,8 +1,10 @@ +# File: tests/test_prompt_manager.py + import asyncio -import sys -from collections.abc import Callable +import inspect from pathlib import Path -from typing import Optional +from typing import Any +import sys import pytest @@ -10,598 +12,596 @@ PROJECT_ROOT: Path = Path(__file__).parent.parent.parent.absolute().resolve() sys.path.insert(0, str(PROJECT_ROOT)) sys.path.insert(0, str(PROJECT_ROOT / "src" / "config")) -from src.prompt.prompt_manager import PromptManager - -# --- Minimal stubs / constants matching the production module --- - -# These imports/definitions are here only to make the tests self‑contained -# In the real project, they already exist in `prompt_manager.py`'s module. -# We mirror them here to control behavior via monkeypatch. - - -class Prompt: - def __init__(self, prompt_name: str, template: str, prompt_render_context: Optional[dict[str, Callable]] = None): - self.prompt_name = prompt_name - self.template = template - self.prompt_render_context = prompt_render_context or {} - - -class DummyLogger: - def __init__(self): - self.errors: list[str] = [] - self.warnings: list[str] = [] - - def error(self, msg: str) -> None: - self.errors.append(msg) - - def warning(self, msg: str) -> None: - self.warnings.append(msg) - - -# --- Fixtures to patch module-level objects in prompt_manager --- - - -@pytest.fixture -def dummy_logger(monkeypatch): - from src.prompt import prompt_manager as pm - - logger = DummyLogger() - monkeypatch.setattr(pm, "logger", logger, raising=False) - return logger - - -@pytest.fixture -def temp_prompts_dir(tmp_path, monkeypatch): - from src.prompt import prompt_manager as pm - - prompts_dir = tmp_path / "prompts" - monkeypatch.setattr(pm, "PROMPTS_DIR", prompts_dir, raising=False) - return prompts_dir - - -@pytest.fixture -def brace_constants(monkeypatch): - from src.prompt import prompt_manager as pm - - # emulate the placeholders used in the manager - monkeypatch.setattr(pm, "_LEFT_BRACE", "__LEFT__", raising=False) - monkeypatch.setattr(pm, "_RIGHT_BRACE", "__RIGHT__", raising=False) - - -@pytest.fixture -def suffix_prompt(monkeypatch): - from src.prompt import prompt_manager as pm - - monkeypatch.setattr(pm, "SUFFIX_PROMPT", ".prompt", raising=False) - - -@pytest.fixture -def manager(temp_prompts_dir, brace_constants, suffix_prompt): - # PromptManager.__init__ uses patched PROMPTS_DIR - return PromptManager() - - -# --- Helper to run async methods in tests (for non-async pytest) --- - - -def run(coro): - return asyncio.get_event_loop().run_until_complete(coro) - - -# --- add_prompt tests -------------------------------------------------------- +from src.prompt.prompt_manager import SUFFIX_PROMPT, Prompt, PromptManager, prompt_manager # noqa @pytest.mark.parametrize( - "existing_prompts, existing_funcs, name_to_add, need_save, expect_in_save", + "prompt_name, template", [ - pytest.param( - {}, - {}, - "greeting", - False, - False, - id="add_prompt_simple_not_saved", - ), - pytest.param( - {}, - {}, - "system", - True, - True, - id="add_prompt_marked_for_save", - ), - pytest.param( - {"existing": Prompt("existing", "tmpl")}, - {}, - "new", - True, - True, - id="add_prompt_with_existing_other_prompt", - ), + pytest.param("simple", "Hello {name}", id="simple-template-with-field"), + pytest.param("no-fields", "Just a static template", id="template-without-fields"), + pytest.param("brace-escaping", "Use {{ and }} around {field}", id="template-with-escaped-braces"), ], ) -def test_add_prompt_happy_path(manager, existing_prompts, existing_funcs, name_to_add, need_save, expect_in_save): - # Arrange - - manager.prompts.update(existing_prompts) - manager._context_construct_functions.update(existing_funcs) - prompt = Prompt(name_to_add, "template") - +def test_prompt_init_happy_paths(prompt_name: str, template: str): # Act - - manager.add_prompt(prompt, need_save=need_save) + prompt = Prompt(prompt_name=prompt_name, template=template) # Assert - - assert manager.prompts[name_to_add] is prompt - assert (name_to_add in manager._prompt_to_save) is expect_in_save + assert prompt.prompt_name == prompt_name + assert prompt.template == template @pytest.mark.parametrize( - "existing_prompts, existing_funcs, new_name, conflict_type", + "prompt_name, template, expected_exception, expected_msg_substring", [ + pytest.param("", "Hello {name}", ValueError, "prompt_name 不能为空", id="empty-prompt-name"), + pytest.param("valid-name", "", ValueError, "template 不能为空", id="empty-template"), pytest.param( - {"dup": Prompt("dup", "tmpl")}, - {}, - "dup", - "prompt_conflict", - id="add_prompt_conflict_with_existing_prompt", + "unnamed-placeholder", + "Hello {}", + ValueError, + "模板中不允许使用未命名的占位符", + id="unnamed-placeholder-not-allowed", ), pytest.param( - {}, - {"dup": (lambda x: x, "mod")}, - "dup", - "func_conflict", - id="add_prompt_conflict_with_existing_context_function", + "unnamed-placeholder-with-escaped-brace", + "Value {{}} and {}", + ValueError, + "模板中不允许使用未命名的占位符", + id="unnamed-placeholder-mixed-with-escaped", ), ], ) -def test_add_prompt_conflict_raises_key_error(manager, existing_prompts, existing_funcs, new_name, conflict_type): - # Arrange - - manager.prompts.update(existing_prompts) - manager._context_construct_functions.update(existing_funcs) - prompt = Prompt(new_name, "template") - +def test_prompt_init_error_cases(prompt_name, template, expected_exception, expected_msg_substring): # Act / Assert - - with pytest.raises(KeyError) as exc: - manager.add_prompt(prompt) - - assert new_name in str(exc.value) - - -# --- add_context_construct_function tests ----------------------------------- - - -def test_add_context_construct_function_happy_path(manager): - # Arrange - - def builder(prompt_name: str) -> str: - return f"ctx_for_{prompt_name}" - - # Act - - manager.add_context_construct_function("ctx", builder) + with pytest.raises(expected_exception) as exc_info: + Prompt(prompt_name=prompt_name, template=template) # Assert - - assert "ctx" in manager._context_construct_functions - stored_func, module = manager._context_construct_functions["ctx"] - assert stored_func is builder - # module is caller's module name - assert isinstance(module, str) - assert module != "" - - -def test_add_context_construct_function_logs_unknown_module(manager, dummy_logger, monkeypatch): - # Arrange - - def builder(prompt_name: str) -> str: - return f"v_{prompt_name}" - - def fake_currentframe(): - class FakeCallerFrame: - f_globals = {"__name__": "unknown"} - - class FakeFrame: - f_back = FakeCallerFrame() - - return FakeFrame() - - from src.prompt import prompt_manager as pm - - monkeypatch.setattr(pm.inspect, "currentframe", fake_currentframe) - - # Act - manager.add_context_construct_function("unknown_ctx", builder) - - # Assert - - assert any("无法获取调用函数的模块名" in msg for msg in dummy_logger.warnings) - assert "unknown_ctx" in manager._context_construct_functions + assert expected_msg_substring in str(exc_info.value) @pytest.mark.parametrize( - "existing_prompts, existing_funcs, name_to_add", + "initial_context, name, func, expected_value, expected_exception, expected_msg_substring, case_id", [ - pytest.param( - {"p": Prompt("p", "tmpl")}, + ( {}, - "p", - id="add_context_construct_function_conflict_with_prompt", + "const_str", + "constant", + "constant", + None, + None, + "add-context-from-string-creates-wrapper", ), - pytest.param( + ( {}, - {"f": (lambda x: x, "mod")}, - "f", - id="add_context_construct_function_conflict_with_existing_func", + "callable_str", + lambda prompt_name: f"hello-{prompt_name}", + "hello-my_prompt", + None, + None, + "add-context-from-callable", + ), + ( + {"dup": lambda _: "x"}, + "dup", + "y", + None, + KeyError, + "Context function name 'dup' 已存在于 Prompt 'my_prompt' 中", + "add-context-duplicate-key-error", ), ], ) -def test_add_context_construct_function_conflict_raises_key_error( - manager, existing_prompts, existing_funcs, name_to_add +def test_prompt_add_context( + initial_context, + name, + func, + expected_value, + expected_exception, + expected_msg_substring, + case_id, ): # Arrange + prompt = Prompt(prompt_name="my_prompt", template="template") + prompt.prompt_render_context = dict(initial_context) - manager.prompts.update(existing_prompts) - manager._context_construct_functions.update(existing_funcs) + # Act + if expected_exception: + with pytest.raises(expected_exception) as exc_info: + prompt.add_context(name, func) - def func(prompt_name: str) -> str: + # Assert + assert expected_msg_substring in str(exc_info.value) + else: + prompt.add_context(name, func) + + # Assert + assert name in prompt.prompt_render_context + result = prompt.prompt_render_context[name]("my_prompt") + assert result == expected_value + + +def test_prompt_manager_add_prompt_happy_and_error(): + # Arrange + manager = PromptManager() + prompt1 = Prompt(prompt_name="p1", template="T1") + manager.add_prompt(prompt1, need_save=True) + + # Act + prompt2 = Prompt(prompt_name="p2", template="T2") + manager.add_prompt(prompt2, need_save=False) + + # Assert + assert "p1" in manager._prompt_to_save + assert "p2" not in manager._prompt_to_save + + # Arrange + prompt_dup = Prompt(prompt_name="p1", template="T-dup") + + # Act / Assert + with pytest.raises(KeyError) as exc_info: + manager.add_prompt(prompt_dup) + + # Assert + assert "Prompt name 'p1' 已存在" in str(exc_info.value) + +def test_prompt_manager_get_prompt_is_copy(): + # Arrange + manager = PromptManager() + prompt = Prompt(prompt_name="original", template="T") + manager.add_prompt(prompt) + + # Act + retrieved_prompt = manager.get_prompt("original") + + # Assert + assert retrieved_prompt is not prompt + assert retrieved_prompt.prompt_name == prompt.prompt_name + assert retrieved_prompt.template == prompt.template + assert retrieved_prompt.prompt_render_context == prompt.prompt_render_context + +def test_prompt_manager_add_prompt_conflict_with_context_name(): + # Arrange + manager = PromptManager() + manager.add_context_construct_function("ctx_name", lambda _: "value") + prompt_conflict = Prompt(prompt_name="ctx_name", template="T") + + # Act / Assert + with pytest.raises(KeyError) as exc_info: + manager.add_prompt(prompt_conflict) + + # Assert + assert "Prompt name 'ctx_name' 已存在" in str(exc_info.value) + + +def test_prompt_manager_add_context_construct_function_happy(): + # Arrange + manager = PromptManager() + + def ctx_func(prompt_name: str) -> str: + return f"ctx-{prompt_name}" + + # Act + manager.add_context_construct_function("ctx", ctx_func) + + # Assert + assert "ctx" in manager._context_construct_functions + stored_func, module = manager._context_construct_functions["ctx"] + assert stored_func is ctx_func + assert module == __name__ + + +def test_prompt_manager_add_context_construct_function_duplicate(): + # Arrange + manager = PromptManager() + + def f(_): + return "x" + + manager.add_context_construct_function("dup", f) + manager.add_prompt(Prompt(prompt_name="dup_prompt", template="T")) + + # Act / Assert + with pytest.raises(KeyError) as exc_info1: + manager.add_context_construct_function("dup", f) + + # Assert + assert "Construct function name 'dup' 已存在" in str(exc_info1.value) + + # Act / Assert + with pytest.raises(KeyError) as exc_info2: + manager.add_context_construct_function("dup_prompt", f) + + # Assert + assert "Construct function name 'dup_prompt' 已存在" in str(exc_info2.value) + + +def test_prompt_manager_get_prompt_not_exist(): + # Arrange + manager = PromptManager() + + # Act / Assert + with pytest.raises(KeyError) as exc_info: + manager.get_prompt("no_such_prompt") + + # Assert + assert "Prompt name 'no_such_prompt' 不存在" in str(exc_info.value) + + +@pytest.mark.parametrize( + "template, inner_context, global_context, expected, case_id", + [ + pytest.param( + "Hello {name}", + {"name": lambda p: f"name-for-{p}"}, + {}, + "Hello name-for-main", + "render-with-inner-context", + ), + pytest.param( + "Global {block}", + {}, + {"block": lambda p: f"block-{p}"}, + "Global block-main", + "render-with-global-context", + ), + pytest.param( + "Mix {inner} and {global}", + {"inner": lambda p: f"inner-{p}"}, + {"global": lambda p: f"global-{p}"}, + "Mix inner-main and global-main", + "render-with-inner-and-global-context", + ), + pytest.param( + "Escaped {{ and }} and {field}", + {"field": lambda _: "X"}, + {}, + "Escaped { and } and X", + "render-with-escaped-braces", + ), + ], +) +@pytest.mark.asyncio +async def test_prompt_manager_render_contexts(template, inner_context, global_context, expected, case_id): + # Arrange + manager = PromptManager() + tmp_prompt = Prompt(prompt_name="main", template=template) + manager.add_prompt(tmp_prompt) + prompt = manager.get_prompt("main") + for name, fn in inner_context.items(): + prompt.add_context(name, fn) + for name, fn in global_context.items(): + manager.add_context_construct_function(name, fn) + + + # Act + rendered = await manager.render_prompt(prompt) + + # Assert + assert rendered == expected + + +@pytest.mark.asyncio +async def test_prompt_manager_render_nested_prompts(): + # Arrange + manager = PromptManager() + p1 = Prompt(prompt_name="p1", template="P1-{x}") + p2 = Prompt(prompt_name="p2", template="P2-{p1}") + p3_tmp = Prompt(prompt_name="p3", template="{p2}-end") + manager.add_prompt(p1) + manager.add_prompt(p2) + manager.add_prompt(p3_tmp) + p3 = manager.get_prompt("p3") + p3.add_context("x", lambda _: "X") + + # Act + rendered = await manager.render_prompt(p3) + + # Assert + assert rendered == "P2-P1-X-end" + + +@pytest.mark.asyncio +async def test_prompt_manager_render_recursive_limit(): + # Arrange + manager = PromptManager() + p1_tmp = Prompt(prompt_name="p1", template="{p2}") + p2_tmp = Prompt(prompt_name="p2", template="{p1}") + manager.add_prompt(p1_tmp) + manager.add_prompt(p2_tmp) + p1 = manager.get_prompt("p1") + + # Act / Assert + with pytest.raises(RecursionError) as exc_info: + await manager.render_prompt(p1) + + # Assert + assert "递归层级过深" in str(exc_info.value) + + +@pytest.mark.asyncio +async def test_prompt_manager_render_missing_field_error(): + # Arrange + manager = PromptManager() + tmp_prompt = Prompt(prompt_name="main", template="Hello {missing}") + manager.add_prompt(tmp_prompt) + prompt = manager.get_prompt("main") + + # Act / Assert + with pytest.raises(KeyError) as exc_info: + await manager.render_prompt(prompt) + + # Assert + assert "Prompt 'main' 中缺少必要的内容块或构建函数: 'missing'" in str(exc_info.value) + + +@pytest.mark.asyncio +async def test_prompt_manager_render_prefers_inner_context_over_global(): + # Arrange + manager = PromptManager() + tmp_prompt = Prompt(prompt_name="main", template="{field}") + manager.add_context_construct_function("field", lambda _: "global") + manager.add_prompt(tmp_prompt) + prompt = manager.get_prompt("main") + prompt.add_context("field", lambda _: "inner") + + # Act + rendered = await manager.render_prompt(prompt) + + # Assert + assert rendered == "inner" + + +@pytest.mark.asyncio +async def test_prompt_manager_render_with_coroutine_context_function(): + # Arrange + manager = PromptManager() + + async def async_inner(prompt_name: str) -> str: + await asyncio.sleep(0) + return f"async-{prompt_name}" + + tmp_prompt = Prompt(prompt_name="main", template="{inner}") + manager.add_prompt(tmp_prompt) + prompt = manager.get_prompt("main") + prompt.add_context("inner", async_inner) + + # Act + rendered = await manager.render_prompt(prompt) + + # Assert + assert rendered == "async-main" + + +@pytest.mark.asyncio +async def test_prompt_manager_render_with_coroutine_global_context_function(): + # Arrange + manager = PromptManager() + + async def async_global(prompt_name: str) -> str: + await asyncio.sleep(0) + return f"g-{prompt_name}" + + tmp_prompt = Prompt(prompt_name="main", template="{g}") + manager.add_context_construct_function("g", async_global) + manager.add_prompt(tmp_prompt) + prompt = manager.get_prompt("main") + + # Act + rendered = await manager.render_prompt(prompt) + + # Assert + assert rendered == "g-main" + + +@pytest.mark.parametrize( + "is_prompt_context, use_coroutine, case_id", + [ + pytest.param(True, False, "prompt-context-sync-error"), + pytest.param(False, False, "global-context-sync-error"), + pytest.param(True, True, "prompt-context-async-error"), + pytest.param(False, True, "global-context-async-error"), + ], +) +@pytest.mark.asyncio +async def test_prompt_manager_get_function_result_error_logging(monkeypatch, is_prompt_context, use_coroutine, case_id): + # Arrange + manager = PromptManager() + + class DummyError(Exception): + pass + + def sync_func(_name: str) -> str: + raise DummyError("sync-error") + + async def async_func(_name: str) -> str: + await asyncio.sleep(0) + raise DummyError("async-error") + + func = async_func if use_coroutine else sync_func + logged_messages: list[str] = [] + + def fake_error(msg: Any) -> None: + logged_messages.append(str(msg)) + + fake_logger = type("FakeLogger", (), {"error": staticmethod(fake_error)}) + + monkeypatch.setattr("src.prompt.prompt_manager.logger", fake_logger) + + # Act / Assert + with pytest.raises(DummyError): + await manager._get_function_result( + func=func, + prompt_name="P", + field_name="field", + is_prompt_context=is_prompt_context, + module="mod", + ) + + # Assert + assert logged_messages + log = logged_messages[0] + if is_prompt_context: + assert "调用 Prompt 'P' 内部上下文构造函数 'field' 时出错" in log + else: + assert "调用上下文构造函数 'field' 时出错，所属模块: 'mod'" in log + + +def test_prompt_manager_add_context_construct_function_unknown_frame(monkeypatch): + # Arrange + manager = PromptManager() + + def fake_currentframe() -> None: + return None + + monkeypatch.setattr("inspect.currentframe", fake_currentframe) + + def f(_): return "x" # Act / Assert + with pytest.raises(RuntimeError) as exc_info: + manager.add_context_construct_function("x", f) - with pytest.raises(KeyError) as exc: - manager.add_context_construct_function(name_to_add, func) - - assert name_to_add in str(exc.value) + # Assert + assert "无法获取调用栈" in str(exc_info.value) -def test_add_context_construct_function_no_frame_raises_runtime_error(manager, monkeypatch): +def test_prompt_manager_add_context_construct_function_unknown_caller_frame(monkeypatch): # Arrange - - from src.prompt import prompt_manager as pm - - monkeypatch.setattr(pm.inspect, "currentframe", lambda: None) - - def func(prompt_name: str) -> str: - return "x" - - # Act / Assert - - with pytest.raises(RuntimeError) as exc: - manager.add_context_construct_function("ctx", func) - - assert "无法获取调用栈" in str(exc.value) - - -def test_add_context_construct_function_no_caller_frame_raises_runtime_error(manager, monkeypatch): - # Arrange - - from src.prompt import prompt_manager as pm + manager = PromptManager() + real_currentframe = inspect.currentframe class FakeFrame: f_back = None - monkeypatch.setattr(pm.inspect, "currentframe", lambda: FakeFrame()) + def fake_currentframe(): + return FakeFrame() - def func(prompt_name: str) -> str: + monkeypatch.setattr("inspect.currentframe", fake_currentframe) + + def f(_): return "x" # Act / Assert - - with pytest.raises(RuntimeError) as exc: - manager.add_context_construct_function("ctx", func) - - assert "无法获取调用栈的上一级" in str(exc.value) - - -# --- get_prompt tests -------------------------------------------------------- - - -@pytest.mark.parametrize( - "existing_name, requested_name, should_raise", - [ - pytest.param("p1", "p1", False, id="get_existing_prompt"), - pytest.param("p1", "missing", True, id="get_missing_prompt_raises"), - ], -) -def test_get_prompt(manager, existing_name, requested_name, should_raise): - # Arrange - - manager.prompts[existing_name] = Prompt(existing_name, "tmpl") - - # Act / Assert - - if should_raise: - with pytest.raises(KeyError) as exc: - manager.get_prompt(requested_name) - assert requested_name in str(exc.value) - else: - prompt = manager.get_prompt(requested_name) - assert prompt.prompt_name == existing_name - - -# --- render_prompt and _render tests ---------------------------------------- - - -@pytest.mark.parametrize( - "template, prompts_setup, ctx_funcs_setup, prompt_ctx, expected", - [ - pytest.param( - "Hello {name}", - {}, - {}, - {"name": lambda p: "World"}, - "Hello World", - id="render_with_prompt_context_sync", - ), - pytest.param( - "Hello {name}", - {}, - {}, - { - "name": lambda p: asyncio.sleep(0, result=f"Async-{p}"), - }, - "Hello Async-main", - id="render_with_prompt_context_async", - ), - pytest.param( - "Outer {inner}", - { - "inner": Prompt("inner", "Inner {value}", {"value": lambda p: "42"}), - }, - {}, - {}, - "Outer Inner 42", - id="render_with_nested_prompt_reference", - ), - pytest.param( - "Module says {ext}", - {}, - { - "ext": (lambda p: f"external-{p}", "test_module"), - }, - {}, - "Module says external-main", - id="render_with_external_context_function_sync", - ), - pytest.param( - "Module async {ext}", - {}, - { - "ext": (lambda p: asyncio.sleep(0, result=f"ext_async-{p}"), "test_module"), - }, - {}, - "Module async ext_async-main", - id="render_with_external_context_function_async", - ), - pytest.param( - "Escaped {{ and }} literal plus {value}", - {}, - {}, - {"value": lambda p: "X"}, - "Escaped { and } literal plus X", - id="render_with_escaped_braces", - ), - ], -) -def test_render_prompt_happy_path( - manager, - template, - prompts_setup, - ctx_funcs_setup, - prompt_ctx, - expected, -): - # Arrange - - main_prompt = Prompt("main", template, prompt_ctx) - manager.add_prompt(main_prompt) - for name, prompt in prompts_setup.items(): - manager.add_prompt(prompt) - manager._context_construct_functions.update(ctx_funcs_setup) - - # Act - - rendered = run(manager.render_prompt(main_prompt)) + with pytest.raises(RuntimeError) as exc_info: + manager.add_context_construct_function("x", f) # Assert + assert "无法获取调用栈的上一级" in str(exc_info.value) - assert rendered == expected + # Cleanup + monkeypatch.setattr("inspect.currentframe", real_currentframe) -def test_render_prompt_missing_field_raises_key_error(manager): +def test_prompt_manager_save_and_load_prompts(tmp_path, monkeypatch): # Arrange + test_dir = tmp_path / "prompts_dir" + test_dir.mkdir() - prompt = Prompt("main", "Hello {missing}") - manager.add_prompt(prompt) + monkeypatch.setattr("src.prompt.prompt_manager.PROMPTS_DIR", test_dir, raising=False) - # Act / Assert - - with pytest.raises(KeyError) as exc: - run(manager.render_prompt(prompt)) - - assert "缺少必要的内容块或构建函数" in str(exc.value) - assert "missing" in str(exc.value) - - -def test_render_prompt_recursion_limit_exceeded(manager): - # Arrange - - # Create mutual recursion between two prompts - p1 = Prompt("p1", "P1 uses {p2}") - p2 = Prompt("p2", "P2 uses {p1}") - manager.add_prompt(p1) - manager.add_prompt(p2) - - # Act / Assert - - with pytest.raises(RecursionError): - run(manager.render_prompt(p1)) - - -# --- _get_function_result tests --------------------------------------------- - - -@pytest.mark.parametrize( - "func, is_prompt_context, expect_async", - [ - pytest.param( - lambda p: f"sync_{p}", - True, - False, - id="get_function_result_sync_prompt_context", - ), - pytest.param( - lambda p: asyncio.sleep(0, result=f"async_{p}"), - False, - True, - id="get_function_result_async_external_context", - ), - ], -) -def test_get_function_result_happy_path(manager, dummy_logger, func, is_prompt_context, expect_async): - # Act - - res = run( - manager._get_function_result( - func=func, - prompt_name="prompt", - field_name="f", - is_prompt_context=is_prompt_context, - module="mod", - ) - ) - - # Assert - - assert res in {"sync_prompt", "async_prompt"} - - -@pytest.mark.parametrize( - "is_prompt_context, expected_message_part", - [ - pytest.param(True, "内部上下文构造函数", id="get_function_result_error_prompt_context_logs_internal_msg"), - pytest.param(False, "上下文构造函数", id="get_function_result_error_external_logs_external_msg"), - ], -) -def test_get_function_result_error_logging(manager, dummy_logger, is_prompt_context, expected_message_part): - # Arrange - - def bad_func(prompt_name: str) -> str: - raise ValueError("bad") - - # Act / Assert - - with pytest.raises(ValueError): - run( - manager._get_function_result( - func=bad_func, - prompt_name="promptX", - field_name="fieldX", - is_prompt_context=is_prompt_context, - module="modX", - ) - ) - - # Assert - - assert any(expected_message_part in msg for msg in dummy_logger.errors) - assert any("promptX" in msg for msg in dummy_logger.errors) ^ (not is_prompt_context) - assert any("modX" in msg for msg in dummy_logger.errors) ^ is_prompt_context - assert any("fieldX" in msg for msg in dummy_logger.errors) - - -# --- save_prompts tests ------------------------------------------------------ - - -def test_save_prompts_happy_path(manager, temp_prompts_dir): - # Arrange - - p1 = Prompt("p1", "Hello {{name}}") - p2 = Prompt("p2", "Bye {{value}}") + manager = PromptManager() + p1 = Prompt(prompt_name="save_me", template="Template {x}") + p1.add_context("x", "X") manager.add_prompt(p1, need_save=True) - manager.add_prompt(p2, need_save=True) # Act - manager.save_prompts() # Assert + saved_file = test_dir / f"save_me{SUFFIX_PROMPT}" + assert saved_file.exists() + assert saved_file.read_text(encoding="utf-8") == "Template {x}" - files = sorted(temp_prompts_dir.glob("*.prompt")) - assert len(files) == 2 - contents = {f.stem: f.read_text(encoding="utf-8") for f in files} - assert contents["p1"] == "Hello {{name}}" - assert contents["p2"] == "Bye {{value}}" - - -def test_save_prompts_io_error(manager, temp_prompts_dir, dummy_logger, monkeypatch): # Arrange + new_manager = PromptManager() - prompt = Prompt("p1", "Hi") - manager.add_prompt(prompt, need_save=True) + # Act + new_manager.load_prompts() - def bad_open(*args, **kwargs): - raise OSError("disk full") + # Assert + loaded = new_manager.get_prompt("save_me") + assert loaded.template == "Template {x}" + assert "save_me" in new_manager._prompt_to_save - monkeypatch.setattr("builtins.open", bad_open) + +def test_prompt_manager_save_prompts_io_error(tmp_path, monkeypatch): + # Arrange + test_dir = tmp_path / "prompts_dir" + test_dir.mkdir() + monkeypatch.setattr("src.prompt.prompt_manager.PROMPTS_DIR", test_dir, raising=False) + manager = PromptManager() + p1 = Prompt(prompt_name="save_error", template="T") + manager.add_prompt(p1, need_save=True) + + class FakeFile: + def __enter__(self): + raise OSError("disk error") + + def __exit__(self, exc_type, exc, tb): + return False + + def fake_open(*_args, **_kwargs): + return FakeFile() + + monkeypatch.setattr("builtins.open", fake_open) # Act / Assert - - with pytest.raises(OSError): + with pytest.raises(OSError) as exc_info: manager.save_prompts() # Assert - - assert any("保存 Prompt 'p1' 时出错" in msg for msg in dummy_logger.errors) + assert "disk error" in str(exc_info.value) -# --- load_prompts tests ------------------------------------------------------ - - -def test_load_prompts_happy_path(manager, temp_prompts_dir): +def test_prompt_manager_load_prompts_io_error(tmp_path, monkeypatch): # Arrange + test_dir = tmp_path / "prompts_dir" + test_dir.mkdir() + monkeypatch.setattr("src.prompt.prompt_manager.PROMPTS_DIR", test_dir, raising=False) + prompt_file = test_dir / f"bad{SUFFIX_PROMPT}" + prompt_file.write_text("content", encoding="utf-8") - file1 = temp_prompts_dir / "greet.prompt" - file2 = temp_prompts_dir / "farewell.prompt" - temp_prompts_dir.mkdir(parents=True, exist_ok=True) - file1.write_text("Hello {{name}}", encoding="utf-8") - file2.write_text("Bye {{name}}", encoding="utf-8") + class FakeFile: + def __enter__(self): + raise OSError("read error") - # Act + def __exit__(self, exc_type, exc, tb): + return False - manager.load_prompts() + def fake_open(*_args, **_kwargs): + return FakeFile() - # Assert - - assert "greet" in manager.prompts - assert "farewell" in manager.prompts - assert "greet" in manager._prompt_to_save - assert "farewell" in manager._prompt_to_save - assert manager.prompts["greet"].template == "Hello {{name}}" - assert manager.prompts["farewell"].template == "Bye {{name}}" - - -def test_load_prompts_error(manager, temp_prompts_dir, dummy_logger, monkeypatch): - # Arrange - - file1 = temp_prompts_dir / "broken.prompt" - temp_prompts_dir.mkdir(parents=True, exist_ok=True) - file1.write_text("whatever", encoding="utf-8") - - def bad_open(*args, **kwargs): - raise OSError("cannot read") - - monkeypatch.setattr("builtins.open", bad_open) + monkeypatch.setattr("builtins.open", fake_open) + manager = PromptManager() # Act / Assert - - with pytest.raises(OSError): + with pytest.raises(OSError) as exc_info: manager.load_prompts() # Assert + assert "read error" in str(exc_info.value) - assert any("加载 Prompt 文件" in msg for msg in dummy_logger.errors) + +def test_prompt_manager_global_instance_access(): + # Act + pm = prompt_manager + + # Assert + assert isinstance(pm, PromptManager) + + +def test_formatter_parsing_named_fields_only(): + # Arrange + manager = PromptManager() + prompt = Prompt(prompt_name="main", template="A {x} B {y} C") + manager.add_prompt(prompt) + + # Act + fields = {field_name for _, field_name, _, _ in manager._formatter.parse(prompt.template) if field_name} + + # Assert + assert fields == {"x", "y"} diff --git a/src/bw_learner/jargon_explainer.py b/src/bw_learner/jargon_explainer.py index 252d1e40..4ee828f6 100644 --- a/src/bw_learner/jargon_explainer.py +++ b/src/bw_learner/jargon_explainer.py @@ -6,7 +6,7 @@ from src.common.logger import get_logger from src.common.database.database_model import Jargon from src.llm_models.utils_model import LLMRequest from src.config.config import model_config, global_config -from src.chat.utils.prompt_builder import Prompt, global_prompt_manager +from src.prompt.prompt_manager import prompt_manager from src.bw_learner.jargon_miner import search_jargon from src.bw_learner.learner_utils import ( is_bot_message, @@ -17,28 +17,6 @@ from src.bw_learner.learner_utils import ( logger = get_logger("jargon") - -def _init_explainer_prompts() -> None: - """初始化黑话解释器相关的prompt""" - # Prompt：概括黑话解释结果 - summarize_prompt_str = """上下文聊天内容: -{chat_context} - -在上下文中提取到的黑话及其含义: -{jargon_explanations} - -请根据上述信息，对黑话解释进行概括和整理。 -- 如果上下文中有黑话出现，请简要说明这些黑话在上下文中的使用情况 -- 将所有黑话解释整理成简洁、易读的一段话 -- 输出格式要自然，适合作为回复参考信息 -请输出概括后的黑话解释（直接输出一段平文本，不要标题，无特殊格式或markdown格式，不要使用JSON格式）： -""" - Prompt(summarize_prompt_str, "jargon_explainer_summarize_prompt") - - -_init_explainer_prompts() - - class JargonExplainer: """黑话解释器，用于在回复前识别和解释上下文中的黑话""" @@ -222,11 +200,15 @@ class JargonExplainer: explanations_text = "\n".join(jargon_explanations) # 使用LLM概括黑话解释 - summarize_prompt = await global_prompt_manager.format_prompt( - "jargon_explainer_summarize_prompt", - chat_context=chat_context, - jargon_explanations=explanations_text, - ) + # summarize_prompt = await global_prompt_manager.format_prompt( + # "jargon_explainer_summarize_prompt", + # chat_context=chat_context, + # jargon_explanations=explanations_text, + # ) + prompt_of_summarize = prompt_manager.get_prompt("jargon_explainer_summarize_prompt") + prompt_of_summarize.add_context("chat_context", lambda _: chat_context) + prompt_of_summarize.add_context("jargon_explanations", lambda _: explanations_text) + summarize_prompt = await prompt_manager.render_prompt(prompt_of_summarize) summary, _ = await self.llm.generate_response_async(summarize_prompt, temperature=0.3) if not summary: diff --git a/src/chat/planner_actions/planner.py b/src/chat/planner_actions/planner.py index ca907406..fc487efc 100644 --- a/src/chat/planner_actions/planner.py +++ b/src/chat/planner_actions/planner.py @@ -13,7 +13,8 @@ from src.config.config import global_config, model_config from src.common.logger import get_logger from src.chat.logger.plan_reply_logger import PlanReplyLogger from src.common.data_models.info_data_model import ActionPlannerInfo -from src.chat.utils.prompt_builder import Prompt, global_prompt_manager +# from src.chat.utils.prompt_builder import Prompt, global_prompt_manager +from src.prompt.prompt_manager import prompt_manager from src.chat.utils.chat_message_builder import ( build_readable_messages_with_id, get_raw_msg_before_timestamp_with_chat, @@ -35,69 +36,6 @@ logger = get_logger("planner") install(extra_lines=3) - -def init_prompt(): - Prompt( - """ -{time_block} -{name_block} -{chat_context_description}，以下是具体的聊天内容 -**聊天内容** -{chat_content_block} - -**可选的action** -reply -动作描述： -1.你可以选择呼叫了你的名字，但是你没有做出回应的消息进行回复 -2.你可以自然的顺着正在进行的聊天内容进行回复或自然的提出一个问题 -3.最好一次对一个话题进行回复，免得啰嗦或者回复内容太乱。 -4.不要选择回复你自己发送的消息 -5.不要单独对表情包进行回复 -6.将上下文中所有含义不明的，疑似黑话的，缩写词均写入unknown_words中 -{reply_action_example} - -no_reply -动作描述： -保持沉默，不回复直到有新消息 -控制聊天频率，不要太过频繁的发言 -{{"action":"no_reply"}} - -{action_options_text} - -**你之前的action执行和思考记录** -{actions_before_now_block} - -请选择**可选的**且符合使用条件的action，并说明触发action的消息id(消息id格式:m+数字) -先输出你的简短的选择思考理由，再输出你选择的action，理由不要分点，精简。 -**动作选择要求** -请你根据聊天内容,用户的最新消息和以下标准选择合适的动作: -{plan_style} -{moderation_prompt} - -target_message_id为必填，表示触发消息的id -请选择所有符合使用要求的action，每个动作最多选择一次，但是可以选择多个动作； -动作用json格式输出，用```json包裹，如果输出多个json，每个json都要单独一行放在同一个```json代码块内: -**示例** -// 理由文本（简短） -```json -{{"action":"动作名", "target_message_id":"m123", .....}} -{{"action":"动作名", "target_message_id":"m456", .....}} -```""", - "planner_prompt", - ) - - Prompt( - """ -{action_name} -动作描述：{action_description} -使用条件{parallel_text}： -{action_require} -{{"action":"{action_name}",{action_parameters}, "target_message_id":"消息id(m+数字)"}} -""", - "action_prompt", - ) - - class ActionPlanner: def __init__(self, chat_id: str, action_manager: ActionManager): self.chat_id = chat_id @@ -663,19 +601,31 @@ class ActionPlanner: reply_action_example += ', "quote":"如果需要引用该message，设置为true"' reply_action_example += "}" - planner_prompt_template = await global_prompt_manager.get_prompt_async("planner_prompt") - prompt = planner_prompt_template.format( - time_block=time_block, - chat_context_description=chat_context_description, - chat_content_block=chat_content_block, - actions_before_now_block=actions_before_now_block, - action_options_text=action_options_block, - moderation_prompt=moderation_prompt_block, - name_block=name_block, - interest=interest, - plan_style=global_config.personality.plan_style, - reply_action_example=reply_action_example, - ) + # planner_prompt_template = await global_prompt_manager.get_prompt_async("planner_prompt") + # prompt = planner_prompt_template.format( + # time_block=time_block, + # chat_context_description=chat_context_description, + # chat_content_block=chat_content_block, + # actions_before_now_block=actions_before_now_block, + # action_options_text=action_options_block, + # moderation_prompt=moderation_prompt_block, + # name_block=name_block, + # interest=interest, + # plan_style=global_config.personality.plan_style, + # reply_action_example=reply_action_example, + # ) + planner_prompt_template = prompt_manager.get_prompt("planner_prompt") + planner_prompt_template.add_context("time_block", time_block) + planner_prompt_template.add_context("chat_context_description", chat_context_description) + planner_prompt_template.add_context("chat_content_block", chat_content_block) + planner_prompt_template.add_context("actions_before_now_block", actions_before_now_block) + planner_prompt_template.add_context("action_options_text", action_options_block) + planner_prompt_template.add_context("moderation_prompt", moderation_prompt_block) + planner_prompt_template.add_context("name_block", name_block) + planner_prompt_template.add_context("interest", interest) + planner_prompt_template.add_context("plan_style", global_config.personality.plan_style) + planner_prompt_template.add_context("reply_action_example", reply_action_example) + prompt = await prompt_manager.render_prompt(planner_prompt_template) return prompt, message_id_list except Exception as e: @@ -759,16 +709,23 @@ class ActionPlanner: parallel_text = "" # 获取动作提示模板并填充 - using_action_prompt = await global_prompt_manager.get_prompt_async("action_prompt") - using_action_prompt = using_action_prompt.format( - action_name=action_name, - action_description=action_info.description, - action_parameters=param_text, - action_require=require_text, - parallel_text=parallel_text, - ) + # using_action_prompt = await global_prompt_manager.get_prompt_async("action_prompt") + # using_action_prompt = using_action_prompt.format( + # action_name=action_name, + # action_description=action_info.description, + # action_parameters=param_text, + # action_require=require_text, + # parallel_text=parallel_text, + # ) + using_action_prompt = prompt_manager.get_prompt("action_prompt") + using_action_prompt.add_context("action_name", action_name) + using_action_prompt.add_context("action_description", action_info.description) + using_action_prompt.add_context("action_parameters", param_text) + using_action_prompt.add_context("action_require", require_text) + using_action_prompt.add_context("parallel_text", parallel_text) + using_action_rendered_prompt = await prompt_manager.render_prompt(using_action_prompt) - action_options_block += using_action_prompt + action_options_block += using_action_rendered_prompt return action_options_block @@ -994,6 +951,3 @@ class ActionPlanner: logger.debug(f"处理不完整的JSON代码块时出错: {e}") return json_objects, reasoning_content - - -init_prompt() diff --git a/src/chat/replyer/group_generator.py b/src/chat/replyer/group_generator.py index 360ab088..054fb101 100644 --- a/src/chat/replyer/group_generator.py +++ b/src/chat/replyer/group_generator.py @@ -17,6 +17,7 @@ from src.chat.message_receive.chat_stream import ChatStream from src.chat.message_receive.uni_message_sender import UniversalMessageSender from src.chat.utils.timer_calculator import Timer # <--- Import Timer from src.chat.utils.utils import get_chat_type_and_target_info, is_bot_self +from src.prompt.prompt_manager import prompt_manager from src.chat.utils.prompt_builder import global_prompt_manager from src.chat.utils.chat_message_builder import ( build_readable_messages, @@ -33,14 +34,12 @@ from src.plugin_system.apis import llm_api from src.chat.logger.plan_reply_logger import PlanReplyLogger from src.chat.replyer.prompt.lpmm_prompt import init_lpmm_prompt -from src.chat.replyer.prompt.replyer_prompt import init_replyer_prompt from src.chat.replyer.prompt.rewrite_prompt import init_rewrite_prompt from src.memory_system.memory_retrieval import init_memory_retrieval_prompt, build_memory_retrieval_prompt from src.bw_learner.jargon_explainer import explain_jargon_in_context, retrieve_concepts_with_jargon from src.chat.utils.common_utils import TempMethodsExpression init_lpmm_prompt() -init_replyer_prompt() init_rewrite_prompt() init_memory_retrieval_prompt() @@ -947,6 +946,7 @@ class DefaultReplyer: else: reply_target_block = "" + dialogue_prompt: str = "" if message_list_before_now_long: latest_msgs = message_list_before_now_long[-int(global_config.chat.max_context_size) :] dialogue_prompt = build_readable_messages( @@ -980,33 +980,55 @@ class DefaultReplyer: # 兜底：即使 multiple_reply_style 配置异常也不影响正常回复 reply_style = global_config.personality.reply_style - return ( - await global_prompt_manager.format_prompt( - prompt_name, - expression_habits_block=expression_habits_block, - tool_info_block=tool_info, - bot_name=global_config.bot.nickname, - knowledge_prompt=prompt_info, - # relation_info_block=relation_info, - extra_info_block=extra_info_block, - jargon_explanation=jargon_explanation, - identity=personality_prompt, - action_descriptions=actions_info, - sender_name=sender, - dialogue_prompt=dialogue_prompt, - time_block=time_block, - reply_target_block=reply_target_block, - reply_style=reply_style, - keywords_reaction_prompt=keywords_reaction_prompt, - moderation_prompt=moderation_prompt_block, - memory_retrieval=memory_retrieval, - chat_prompt=chat_prompt_block, - planner_reasoning=planner_reasoning, - ), - selected_expressions, - timing_logs, - almost_zero_str, - ) + # return ( + # await global_prompt_manager.format_prompt( + # prompt_name, + # expression_habits_block=expression_habits_block, + # tool_info_block=tool_info, + # bot_name=global_config.bot.nickname, + # knowledge_prompt=prompt_info, + # # relation_info_block=relation_info, + # extra_info_block=extra_info_block, + # jargon_explanation=jargon_explanation, + # identity=personality_prompt, + # action_descriptions=actions_info, + # sender_name=sender, + # dialogue_prompt=dialogue_prompt, + # time_block=time_block, + # reply_target_block=reply_target_block, + # reply_style=reply_style, + # keywords_reaction_prompt=keywords_reaction_prompt, + # moderation_prompt=moderation_prompt_block, + # memory_retrieval=memory_retrieval, + # chat_prompt=chat_prompt_block, + # planner_reasoning=planner_reasoning, + # ), + # selected_expressions, + # timing_logs, + # almost_zero_str, + # ) + prompt = prompt_manager.get_prompt(prompt_name) + prompt.add_context("expression_habits_block", expression_habits_block) + prompt.add_context("tool_info_block", tool_info) + prompt.add_context("bot_name", global_config.bot.nickname) + prompt.add_context("knowledge_prompt", prompt_info) + # prompt.add_context("relation_info_block", relation_info) + prompt.add_context("extra_info_block", extra_info_block) + prompt.add_context("jargon_explanation", jargon_explanation) + prompt.add_context("identity", personality_prompt) + prompt.add_context("action_descriptions", actions_info) + prompt.add_context("sender_name", sender) + prompt.add_context("dialogue_prompt", dialogue_prompt) + prompt.add_context("time_block", time_block) + prompt.add_context("reply_target_block", reply_target_block) + prompt.add_context("reply_style", reply_style) + prompt.add_context("keywords_reaction_prompt", keywords_reaction_prompt) + prompt.add_context("moderation_prompt", moderation_prompt_block) + prompt.add_context("memory_retrieval", memory_retrieval) + prompt.add_context("chat_prompt", chat_prompt_block) + prompt.add_context("planner_reasoning", planner_reasoning) + formatted_prompt = await prompt_manager.render_prompt(prompt) + return (formatted_prompt, selected_expressions, timing_logs, almost_zero_str) async def build_prompt_rewrite_context( self, diff --git a/src/chat/replyer/prompt/replyer_prompt.py b/src/chat/replyer/prompt/replyer_prompt.py deleted file mode 100644 index 6b92a3a9..00000000 --- a/src/chat/replyer/prompt/replyer_prompt.py +++ /dev/null @@ -1,48 +0,0 @@ -from src.chat.utils.prompt_builder import Prompt -# from src.chat.memory_system.memory_activator import MemoryActivator - - -def init_replyer_prompt(): - Prompt( - """{knowledge_prompt}{tool_info_block}{extra_info_block} -{expression_habits_block}{memory_retrieval}{jargon_explanation} - -你正在qq群里聊天，下面是群里正在聊的内容，其中包含聊天记录和聊天中的图片 -其中标注 {bot_name}(你) 的发言是你自己的发言，请注意区分: -{time_block} -{dialogue_prompt} - -{reply_target_block}。 -{planner_reasoning} -{identity} -{chat_prompt}你正在群里聊天,现在请你读读之前的聊天记录，然后给出日常且口语化的回复， -尽量简短一些。{keywords_reaction_prompt} -请注意把握聊天内容，不要回复的太有条理。 -{reply_style} -请注意不要输出多余内容(包括不必要的前后缀，冒号，括号，表情包，at或 @等 )，只输出发言内容就好。 -最好一次对一个话题进行回复，免得啰嗦或者回复内容太乱。 -现在，你说：""", - "replyer_prompt_0", - ) - - Prompt( - """{knowledge_prompt}{tool_info_block}{extra_info_block} -{expression_habits_block}{memory_retrieval}{jargon_explanation} - -你正在qq群里聊天，下面是群里正在聊的内容，其中包含聊天记录和聊天中的图片 -其中标注 {bot_name}(你) 的发言是你自己的发言，请注意区分: -{time_block} -{dialogue_prompt} - -{reply_target_block}。 -{planner_reasoning} -{identity} -{chat_prompt}你正在群里聊天,现在请你读读之前的聊天记录，把握当前的话题，然后给出日常且简短的回复。 -最好一次对一个话题进行回复，免得啰嗦或者回复内容太乱。 -{keywords_reaction_prompt} -请注意把握聊天内容。 -{reply_style} -请注意不要输出多余内容(包括不必要的前后缀，冒号，括号，at或 @等 )，只输出发言内容就好。 -现在，你说：""", - "replyer_prompt", - ) diff --git a/src/main.py b/src/main.py index aa985ce0..8abb34c6 100644 --- a/src/main.py +++ b/src/main.py @@ -26,6 +26,8 @@ from src.common.message import get_global_api from src.dream.dream_agent import start_dream_scheduler from src.bw_learner.expression_auto_check_task import ExpressionAutoCheckTask +from src.prompt.prompt_manager import prompt_manager + # 插件系统现在使用统一的插件加载器 install(extra_lines=3) @@ -119,6 +121,8 @@ class MainSystem: # 将bot.py中的chat_bot.message_process消息处理函数注册到api.py的消息处理基类中 self.app.register_message_handler(chat_bot.message_process) self.app.register_custom_message_handler("message_id_echo", chat_bot.echo_message_process) + + prompt_manager.load_prompts() # 触发 ON_START 事件 from src.plugin_system.core.events_manager import events_manager diff --git a/src/prompt/prompt_manager.py b/src/prompt/prompt_manager.py index cef2114e..a3d21f44 100644 --- a/src/prompt/prompt_manager.py +++ b/src/prompt/prompt_manager.py @@ -19,19 +19,28 @@ SUFFIX_PROMPT = ".prompt" class Prompt: - prompt_name: str - template: str - prompt_render_context: dict[str, Callable[[str], str | Coroutine[Any, Any, str]]] = {} - def __init__(self, prompt_name: str, template: str) -> None: self.prompt_name = prompt_name self.template = template + self.prompt_render_context: dict[str, Callable[[str], str | Coroutine[Any, Any, str]]] = {} + self._is_cloned = False self.__post_init__() - def add_context(self, name: str, func: Callable[[str], str | Coroutine[Any, Any, str]]) -> None: + def add_context(self, name: str, func_or_str: Callable[[str], str | Coroutine[Any, Any, str]] | str) -> None: if name in self.prompt_render_context: raise KeyError(f"Context function name '{name}' 已存在于 Prompt '{self.prompt_name}' 中") - self.prompt_render_context[name] = func + if isinstance(func_or_str, str): + + def tmp_func(_: str) -> str: + return func_or_str + + render_function = tmp_func + else: + render_function = func_or_str + self.prompt_render_context[name] = render_function + + def clone(self) -> "Prompt": + return Prompt(self.prompt_name, self.template) def __post_init__(self): if not self.prompt_name: @@ -47,7 +56,7 @@ class PromptManager: def __init__(self) -> None: PROMPTS_DIR.mkdir(parents=True, exist_ok=True) # 确保提示词目录存在 self.prompts: dict[str, Prompt] = {} - """存储 Prompt 实例""" + """存储 Prompt 实例，禁止直接从外部访问，否则将引起不可知后果""" self._context_construct_functions: dict[str, tuple[Callable[[str], str | Coroutine[Any, Any, str]], str]] = {} """存储上下文构造函数及其所属模块""" self._formatter = Formatter() # 仅用来解析模板 @@ -57,6 +66,7 @@ class PromptManager: def add_prompt(self, prompt: Prompt, need_save: bool = False) -> None: if prompt.prompt_name in self.prompts or prompt.prompt_name in self._context_construct_functions: + # 确保名称无冲突 raise KeyError(f"Prompt name '{prompt.prompt_name}' 已存在") self.prompts[prompt.prompt_name] = prompt if need_save: @@ -81,14 +91,26 @@ class PromptManager: self._context_construct_functions[name] = func, caller_module def get_prompt(self, prompt_name: str) -> Prompt: + """获取指定名称的 Prompt 实例的克隆""" if prompt_name not in self.prompts: raise KeyError(f"Prompt name '{prompt_name}' 不存在") - return self.prompts[prompt_name] + prompt = self.prompts[prompt_name].clone() + prompt._is_cloned = True + return prompt async def render_prompt(self, prompt: Prompt) -> str: + if not prompt._is_cloned: + raise ValueError( + "只能渲染通过 PromptManager.get_prompt 方法获取的 Prompt 实例，你可能对原始实例进行了修改和渲染操作" + ) return await self._render(prompt) - async def _render(self, prompt: Prompt, recursive_level: int = 0) -> str: + async def _render( + self, + prompt: Prompt, + recursive_level: int = 0, + additional_construction_function_dict: dict[str, Callable[[str], str | Coroutine[Any, Any, str]]] = {}, # noqa: B006 + ) -> str: prompt.template = prompt.template.replace("{{", _LEFT_BRACE).replace("}}", _RIGHT_BRACE) if recursive_level > 10: raise RecursionError("递归层级过深，可能存在循环引用") @@ -96,16 +118,40 @@ class PromptManager: rendered_fields: dict[str, str] = {} for field_name in field_block: if field_name in self.prompts: - rendered_fields[field_name] = await self._render(self.prompts[field_name], recursive_level + 1) + nested_prompt = self.get_prompt(field_name) + additional_construction_function_dict |= prompt.prompt_render_context + rendered_fields[field_name] = await self._render( + nested_prompt, + recursive_level + 1, + additional_construction_function_dict, + ) elif field_name in prompt.prompt_render_context: + # 优先使用内部构造函数 func = prompt.prompt_render_context[field_name] rendered_fields[field_name] = await self._get_function_result( - func, prompt.prompt_name, field_name, is_prompt_context=True + func, + prompt.prompt_name, + field_name, + is_prompt_context=True, ) elif field_name in self._context_construct_functions: + # 随后查找全局构造函数 func, module = self._context_construct_functions[field_name] rendered_fields[field_name] = await self._get_function_result( - func, prompt.prompt_name, field_name, is_prompt_context=False, module=module + func, + prompt.prompt_name, + field_name, + is_prompt_context=False, + module=module, + ) + elif field_name in additional_construction_function_dict: + # 最后查找额外传入的构造函数 + func = additional_construction_function_dict[field_name] + rendered_fields[field_name] = await self._get_function_result( + func, + prompt.prompt_name, + field_name, + is_prompt_context=True, ) else: raise KeyError(f"Prompt '{prompt.prompt_name}' 中缺少必要的内容块或构建函数: '{field_name}'")