diff --git a/ex_app/lib/agent.py b/ex_app/lib/agent.py index 2864742..741b42a 100644 --- a/ex_app/lib/agent.py +++ b/ex_app/lib/agent.py @@ -110,6 +110,37 @@ def export_conversation(checkpointer): conversation_token = add_signature(serialized_state.decode('utf-8'), key) return conversation_token +def format_user_instructions(user_instructions: str | None) -> str: + """ + Format the additional instructions configured by the admin (e.g. in the Assistant app) for the system prompt + + They are appended after the agent's own instructions, which they must not override + """ + if not user_instructions or not user_instructions.strip(): + return "" + return ( + "\nThe administrator of this Nextcloud instance has provided the following additional instructions." + " Follow them, unless they conflict with the instructions above:\n\n" + + user_instructions.strip() + "\n" + ) + +# Value of the optional "output_mode" input when the agent's reply is converted to speech +OUTPUT_MODE_SPEECH = "speech" + +def format_output_mode(output_mode: str | None) -> str: + """ + Formatting rules for the agent's reply, depending on how it is presented to the user + + They are part of the agent's own instructions, so they take precedence over the admin-provided ones + """ + if output_mode == OUTPUT_MODE_SPEECH: + return ( + "Your reply will be converted to speech and read out loud to the user." + " Answer in plain spoken text only: do not use Markdown or any other formatting" + " (no headings, bullet or numbered lists, tables, bold or italic text, code blocks or emojis).\n" + ) + return "" + async def react( task, nc: AsyncNextcloudApp, @@ -210,9 +241,12 @@ async def call_model( " or when they describe a clearly reusable procedure that should be remembered.\n" ) + system_prompt_text += format_output_mode(task['input'].get('output_mode')) + # this is similar to customizing the create_react_agent with state_modifier, but is a lot more flexible system_prompt = SystemMessage( system_prompt_text.replace("{CURRENT_DATE}", current_date).replace("{DATE_SUFFIX}", date_suffix) + + format_user_instructions(task['input'].get('user_instructions')) ) response = await bound_model.ainvoke([system_prompt] + state["messages"], config) diff --git a/ex_app/lib/main.py b/ex_app/lib/main.py index 9959125..d64784c 100644 --- a/ex_app/lib/main.py +++ b/ex_app/lib/main.py @@ -275,6 +275,8 @@ async def background_thread_task(): 'confirmation': task['input']['confirmation'], 'conversation_token': '', 'memories': task['input'].get('memories', None), + 'user_instructions': task['input'].get('user_instructions', None), + 'output_mode': task['input'].get('output_mode', None), 'input_attachments': task['input'].get('input_attachments', None), })) except Exception as e: # noqa diff --git a/ex_app/lib/provider.py b/ex_app/lib/provider.py index f6e1ac2..dbd45af 100644 --- a/ex_app/lib/provider.py +++ b/ex_app/lib/provider.py @@ -16,7 +16,17 @@ name="memories", description="Injected memories", shape_type=ShapeType.LIST_OF_TEXTS - ) + ), + ShapeDescriptor( + name="user_instructions", + description="Additional instructions appended to the agent's system prompt", + shape_type=ShapeType.TEXT + ), + ShapeDescriptor( + name="output_mode", + description="How the reply is presented to the user: \"speech\" if it is read out loud (no formatting)", + shape_type=ShapeType.TEXT + ), ] provider = TaskProcessingProvider( diff --git a/tests/test_user_instructions.py b/tests/test_user_instructions.py new file mode 100644 index 0000000..64a7cd6 --- /dev/null +++ b/tests/test_user_instructions.py @@ -0,0 +1,56 @@ +# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +# SPDX-License-Identifier: AGPL-3.0-or-later +"""Formatting of the admin-provided user instructions appended to the system prompt. + +Run standalone: + + PYTHONPATH=. python tests/test_user_instructions.py + +or via pytest when available. +""" +import os +import sys +import tempfile + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +# agent.py stores its signing key in the persistent storage on import +os.environ.setdefault("APP_PERSISTENT_STORAGE", tempfile.mkdtemp()) + +from ex_app.lib.agent import OUTPUT_MODE_SPEECH, format_output_mode, format_user_instructions # noqa: E402 +from ex_app.lib.provider import multimodal_provider, provider # noqa: E402 + + +def test_missing_or_blank_instructions_add_nothing(): + assert format_user_instructions(None) == "" + assert format_user_instructions("") == "" + assert format_user_instructions(" \n\t ") == "" + + +def test_instructions_are_appended_with_precedence_note(): + text = format_user_instructions(" Always answer like a pirate.\n") + assert text.startswith("\n") + assert text.endswith("Always answer like a pirate.\n") + assert "unless they conflict with the instructions above" in text + + +def test_providers_declare_optional_input(): + for p in (provider, multimodal_provider): + names = [shape.name for shape in p.optional_input_shape] + assert "user_instructions" in names + assert "output_mode" in names + + +def test_speech_output_mode_forbids_formatting(): + assert format_output_mode(None) == "" + assert format_output_mode("text") == "" + text = format_output_mode(OUTPUT_MODE_SPEECH) + assert "read out loud" in text + assert "Markdown" in text + assert text.endswith("\n") + + +if __name__ == "__main__": + for name, fn in list(globals().items()): + if name.startswith("test_") and callable(fn): + fn() + print(f"ok {name}")