Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 34 additions & 0 deletions ex_app/lib/agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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)
Expand Down
2 changes: 2 additions & 0 deletions ex_app/lib/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -275,6 +275,8 @@ async def background_thread_task():
'confirmation': task['input']['confirmation'],
'conversation_token': '<skipped>',
'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
Expand Down
12 changes: 11 additions & 1 deletion ex_app/lib/provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
56 changes: 56 additions & 0 deletions tests/test_user_instructions.py
Original file line number Diff line number Diff line change
@@ -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}")
Loading