From bc33c720d62156a362095acddba8c918c54e9a04 Mon Sep 17 00:00:00 2001 From: Lukas Schaefer Date: Wed, 15 Jul 2026 15:40:10 -0400 Subject: [PATCH] feat: support multimodal chat inputs and outputs Signed-off-by: Lukas Schaefer --- ex_app/lib/agent.py | 29 +++++++++++---- ex_app/lib/main.py | 23 +++++++++--- ex_app/lib/nc_model.py | 82 ++++++++++++++++++++++++++++++++++++++---- ex_app/lib/provider.py | 45 ++++++++++++++--------- 4 files changed, 146 insertions(+), 33 deletions(-) diff --git a/ex_app/lib/agent.py b/ex_app/lib/agent.py index a1183b2..48dad93 100644 --- a/ex_app/lib/agent.py +++ b/ex_app/lib/agent.py @@ -17,7 +17,13 @@ from ex_app.lib.signature import verify_signature from ex_app.lib.signature import add_signature from ex_app.lib.graph import AgentState, get_graph -from ex_app.lib.nc_model import model +from ex_app.lib.nc_model import ( + model, + MULTIMODAL_INTERACTION, + extract_text_content, + extract_file_ids, + build_multimodal_content, +) from ex_app.lib.tools import get_tools from ex_app.lib.memorysaver import MemorySaver from ex_app.lib.jsonplus import JsonPlusSerializer @@ -102,9 +108,11 @@ async def react( nc: AsyncNextcloudApp, stream_output: Callable[[dict[str, Any]], Awaitable[None]] | None = None, ): - safe_tools, dangerous_tools = await get_tools(nc) - + multimodal = task.get('type') == MULTIMODAL_INTERACTION model.bind_nextcloud(nc) + model.multimodal = multimodal + + safe_tools, dangerous_tools = await get_tools(nc) tools = dangerous_tools + safe_tools @@ -188,7 +196,13 @@ async def call_model( else: new_input = None else: - new_input = {"messages": [("user", task['input']['input'])]} + input_attachments = task['input'].get('input_attachments') or [] + user_text = task['input']['input'] + if multimodal and input_attachments: + user_content = build_multimodal_content(user_text, [int(file_id) for file_id in input_attachments]) + new_input = {"messages": [HumanMessage(content=user_content)]} + else: + new_input = {"messages": [("user", user_text)]} snapshot_messages = state_snapshot.values.get('messages', []) last_message: AIMessage = AIMessage("") @@ -252,9 +266,12 @@ async def report_stream_state(force: bool = False): if state_snapshot.next == ('dangerous_tools', ): actions = json.dumps(last_message.tool_calls) - return { - 'output': last_message.content, + result = { + 'output': extract_text_content(last_message.content), 'actions': actions, 'conversation_token': export_conversation(checkpointer), 'sources': source_list, } + if multimodal: + result['output_attachments'] = extract_file_ids(last_message.content) + return result diff --git a/ex_app/lib/main.py b/ex_app/lib/main.py index 086fed4..591f1f2 100644 --- a/ex_app/lib/main.py +++ b/ex_app/lib/main.py @@ -24,9 +24,13 @@ from ex_app.lib.agent import react from ex_app.lib.logger import log from ex_app.lib.mcp_server import UserAuthMiddleware, ToolListMiddleware -from ex_app.lib.provider import provider +from ex_app.lib.provider import provider, multimodal_provider from ex_app.lib.tools import get_categories +PROVIDERS = [provider, multimodal_provider] +PROVIDER_IDS = [p.id for p in PROVIDERS] +TASK_TYPES = [p.task_type for p in PROVIDERS] + from contextvars import ContextVar from gettext import translation from fastmcp import FastMCP @@ -113,7 +117,8 @@ async def enabled_handler(enabled: bool, nc: AsyncNextcloudApp) -> str: # NOTE: `user` is unavailable on this step, so all NC API calls that require it will fail as unauthorized. await log(nc, LogLvl.INFO, f"enabled={enabled}") if enabled: - await nc.providers.task_processing.register(provider) + for p in PROVIDERS: + await nc.providers.task_processing.register(p) app_enabled.set() await log(nc, LogLvl.WARNING, f"App enabled: {nc.app_cfg.app_name}") @@ -125,7 +130,8 @@ async def enabled_handler(enabled: bool, nc: AsyncNextcloudApp) -> str: await nc.appconfig_ex.set_value('tool_status', json.dumps(pref_settings)) else: - await nc.providers.task_processing.unregister(provider.id) + for p in PROVIDERS: + await nc.providers.task_processing.unregister(p.id) app_enabled.clear() await log(nc, LogLvl.WARNING, f"App disabled: {nc.app_cfg.app_name}") # In case of an error, a non-empty short string should be returned, which will be shown to the NC administrator. @@ -142,7 +148,7 @@ async def background_thread_task(): continue try: - response = await nc.providers.task_processing.next_task([provider.id], [provider.task_type]) + response = await nc.providers.task_processing.next_task(PROVIDER_IDS, TASK_TYPES) if not response or not 'task' in response: async with NUM_RUNNING_TASKS_LOCK: no_tasks_running = NUM_RUNNING_TASKS == 0 @@ -162,7 +168,14 @@ async def background_thread_task(): task = response["task"] await log(nc, LogLvl.INFO, 'New Task incoming') await log(nc, LogLvl.DEBUG, str(task)) - await log(nc, LogLvl.INFO, str({'input': task['input']['input'], 'confirmation': task['input']['confirmation'], 'conversation_token': '', 'memories': task['input'].get('memories', None)})) + await log(nc, LogLvl.INFO, str({ + 'type': task.get('type'), + 'input': task['input']['input'], + 'confirmation': task['input']['confirmation'], + 'conversation_token': '', + 'memories': task['input'].get('memories', None), + 'input_attachments': task['input'].get('input_attachments', None), + })) tg.create_task(handle_task(task, nc)) NUM_RUNNING_TASKS_LOCK = asyncio.Lock() diff --git a/ex_app/lib/nc_model.py b/ex_app/lib/nc_model.py index 5d7653f..76915f5 100644 --- a/ex_app/lib/nc_model.py +++ b/ex_app/lib/nc_model.py @@ -24,6 +24,50 @@ from ex_app.lib.logger import log +TEXT_CHAT_WITH_TOOLS = "core:text2text:chatwithtools" +MULTIMODAL_CHAT_WITH_TOOLS = "core:text2text:multimodal-chatwithtools" +MULTIMODAL_INTERACTION = "core:contextagent:multimodal-interaction" + + +def extract_text_content(content: Any) -> str: + """Extract plain text from a string or multimodal content-part list.""" + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [] + for part in content: + if isinstance(part, dict) and part.get("type") == "text": + parts.append(part.get("text", "")) + elif isinstance(part, str): + parts.append(part) + return "".join(parts) + return "" + + +def extract_file_ids(content: Any) -> list[int]: + """Extract file IDs from a multimodal content-part list.""" + if not isinstance(content, list): + return [] + file_ids = [] + for part in content: + if isinstance(part, dict) and part.get("type") == "file" and "file_id" in part: + file_ids.append(int(part["file_id"])) + return file_ids + + +def build_multimodal_content(text: str, file_ids: list[int] | None = None, task_id: int | None = None) -> list[dict[str, Any]] | str: + """Build multimodal content parts, or plain text when there are no files.""" + if not file_ids: + return text + content: list[dict[str, Any]] = [ + ({"type": "file", "file_id": file_id} if task_id is None else {"type": "file", "file_id": file_id, "ocp_task_id": task_id}) + for file_id in file_ids + ] + if text: + content.append({"type": "text", "text": text}) + return content + + class Task(BaseModel): id: int status: str @@ -45,6 +89,7 @@ class ChatWithNextcloud(BaseChatModel): TOOL_OUTPUT_MAX_LENGTH: int = 2000 POLL_WAIT_TIME: int = 5 STREAMING_POLL_WAIT_TIME: int = 1 + multimodal: bool = False def _generate(self, messages: list[BaseMessage], stop: Optional[list[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any): raise Exception("Use _agenerate instead") @@ -77,6 +122,8 @@ def _build_task_input(self, messages: list[BaseMessage]) -> dict[str, typing.Any task_input['input'] = '' task_input['tool_message'] = [] task_input['tools'] = json.dumps(self.tools) + if self.multimodal: + task_input['input_attachments'] = [] history = [] for i, message in enumerate(messages): @@ -87,9 +134,13 @@ def _build_task_input(self, messages: list[BaseMessage]) -> dict[str, typing.Any history.append(json.dumps(msg)) elif message.type == 'human': if len(messages)-1 != i: + # Earlier turns keep file parts in history content. history.append(json.dumps({"role": "human", "content": message.content})) else: - task_input['input'] = message.content + # Current turn: text → input, files → input_attachments. + task_input['input'] = extract_text_content(message.content) + if self.multimodal: + task_input['input_attachments'] = extract_file_ids(message.content) elif message.type == 'tool': content = message.content age = len(messages) - 1 - i @@ -116,6 +167,8 @@ async def _schedule_task(self, task_input: dict[str, typing.Any], prefer_streami await log(nc, LogLvl.DEBUG, task_input) + task_type = MULTIMODAL_CHAT_WITH_TOOLS if self.multimodal else TEXT_CHAT_WITH_TOOLS + i = 0 while i < 20: try: @@ -123,7 +176,7 @@ async def _schedule_task(self, task_input: dict[str, typing.Any], prefer_streami "POST", "/ocs/v1.php/taskprocessing/schedule", json={ - "type": "core:text2text:chatwithtools", + "type": task_type, "appId": "context_agent", "input": task_input, "preferStreaming": prefer_streaming, @@ -180,7 +233,6 @@ def _task_output_text(self, task: Task) -> str | None: return None output = task.output.get('output') return output if isinstance(output, str) else None - def _raw_task_tool_calls(self, task: Task) -> list[dict[str, typing.Any]]: if not isinstance(task.output, dict): return [] @@ -237,11 +289,15 @@ def _task_to_message(self, task: Task) -> AIMessage: if not isinstance(task.output, dict) or "output" not in task.output: raise Exception('"output" key not found in Nextcloud TaskProcessing task result') + output_text = task.output['output'] + output_attachments = task.output.get('output_attachments', []) + content = build_multimodal_content(output_text, output_attachments, task.id) + tool_calls, invalid_tool_calls = self._task_tool_calls(task) if len(tool_calls) > 0 or len(invalid_tool_calls) > 0: - message = AIMessage(task.output['output'], tool_calls=tool_calls, invalid_tool_calls=invalid_tool_calls) + message = AIMessage(content, tool_calls=tool_calls, invalid_tool_calls=invalid_tool_calls) else: - message = AIMessage(task.output['output']) + message = AIMessage(content) return message @@ -345,7 +401,21 @@ async def _astream( yield ChatGenerationChunk(message=AIMessageChunk(content=final_delta)) tool_calls, invalid_tool_calls = self._task_tool_calls(task) - if len(tool_calls) > 0 or len(invalid_tool_calls) > 0: + output_attachments = [] + if isinstance(task.output, dict): + output_attachments = task.output.get('output_attachments') or [] + + if output_attachments: + # Append file parts after streamed text so LangChain merges them + # into the final AIMessage content (attachments are not streamed). + yielded_chunk = True + content = [{"type": "file", "file_id": int(file_id), "ocp_task_id": task.id} for file_id in output_attachments] + yield ChatGenerationChunk(message=AIMessageChunk( + content=content, + tool_calls=tool_calls, + invalid_tool_calls=invalid_tool_calls, + )) + elif len(tool_calls) > 0 or len(invalid_tool_calls) > 0: yielded_chunk = True yield ChatGenerationChunk(message=AIMessageChunk(content='', tool_calls=tool_calls, invalid_tool_calls=invalid_tool_calls)) diff --git a/ex_app/lib/provider.py b/ex_app/lib/provider.py index 1a502c4..f6e1ac2 100644 --- a/ex_app/lib/provider.py +++ b/ex_app/lib/provider.py @@ -1,25 +1,38 @@ # SPDX-FileCopyrightText: 2024 Nextcloud GmbH and Nextcloud contributors # SPDX-License-Identifier: AGPL-3.0-or-later -from nc_py_api.ex_app.providers.task_processing import TaskProcessingProvider, TaskType, ShapeDescriptor, ShapeType +from nc_py_api.ex_app.providers.task_processing import TaskProcessingProvider, ShapeDescriptor, ShapeType +_optional_output_shape = [ + ShapeDescriptor( + name="sources", + description="Used tools", + shape_type=ShapeType.LIST_OF_TEXTS + ) +] + +_optional_input_shape = [ + ShapeDescriptor( + name="memories", + description="Injected memories", + shape_type=ShapeType.LIST_OF_TEXTS + ) +] + provider = TaskProcessingProvider( id='context_agent:agent', name='ContextAgent Provider', task_type='core:contextagent:interaction', expected_runtime=60, - optional_output_shape= [ - ShapeDescriptor( - name="sources", - description="Used tools", - shape_type=ShapeType.LIST_OF_TEXTS - ) - ], - optional_input_shape= [ - ShapeDescriptor( - name="memories", - description="Injected memories", - shape_type=ShapeType.LIST_OF_TEXTS - ) - ] -) \ No newline at end of file + optional_output_shape=_optional_output_shape, + optional_input_shape=_optional_input_shape, +) + +multimodal_provider = TaskProcessingProvider( + id='context_agent:agent_multimodal', + name='ContextAgent Multimodal Provider', + task_type='core:contextagent:multimodal-interaction', + expected_runtime=60, + optional_output_shape=_optional_output_shape, + optional_input_shape=_optional_input_shape, +)