Add Context Chat Multi: process a batch of questions in one task

context_chat_backend/chain/types.py

@@ -12,6 +12,7 @@
 	'ContextException',
 	'InDocument',
 	'LLMOutput',
+	'MultiOutput',
 	'ScopeType',
 ]
 
@@ -43,6 +44,12 @@ class LLMOutput(TypedDict):
 	sources: list[SearchResult]
 
 
+class MultiOutput(TypedDict):
+	questions: list[str]
+	answers: list[str]
+	sources_per_question: list[list[SearchResult]]
+
+
 class EnrichedSource(BaseModel):
 	id: str
 	label: str

context_chat_backend/controller.py

@@ -37,7 +37,7 @@
 from .models.types import LlmException
 from nc_py_api.ex_app import AppAPIAuthMiddleware
 from .utils import JSONResponse, exec_in_proc, get_app_role, is_k8s_env
-from .task_fetcher import THREAD_STOP_EVENT, start_bg_threads, trigger_handler, wait_for_bg_threads
+from .task_fetcher import MAX_MULTI_QUESTIONS, THREAD_STOP_EVENT, start_bg_threads, trigger_handler, wait_for_bg_threads
 from .vectordb.service import count_documents_by_provider
 
 # setup
@@ -101,11 +101,15 @@ def get_enabled_state() -> bool:
 		return app_enabled.is_set()
 
 
+
+
 def enabled_handler(enabled: bool, nc: NextcloudApp | AsyncNextcloudApp) -> str:
 	SEARCH_TASKTYPE_ID = 'context_chat:context_chat_search'
 	SEARCH_PROVIDER_ID = 'context_chat-context_chat_search'
 	CC_TASKTYPE_ID = 'context_chat:context_chat'
 	CC_PROVIDER_ID = 'context_chat-context_chat'
+	MULTI_TASKTYPE_ID = 'context_chat:context_chat_multi'
+	MULTI_PROVIDER_ID = 'context_chat-context_chat_multi'
 
 	# todo: translate user-facing texts
 	try:
@@ -209,6 +213,63 @@ def enabled_handler(enabled: bool, nc: NextcloudApp | AsyncNextcloudApp) -> str:
 			)
 			nc.providers.task_processing.register(cc_provider, cc_tasktype)
 
+			multi_tasktype = TaskType(
+				id=MULTI_TASKTYPE_ID,
+				name='Context Chat Multi',
+				description='Ask several questions about your data at once.',
+				input_shape=[
+					ShapeDescriptor(
+						name='questions',
+						description='Ask questions about your documents, files and more.',
+						shape_type=ShapeType.LIST_OF_TEXTS,
+					),
+					ShapeDescriptor(
+						name='scopeType',
+						description='Any of the following values: "none", "source", "provider".',
+						shape_type=ShapeType.TEXT,
+					),
+					ShapeDescriptor(
+						name='scopeList',
+						description='List of sources or providers',
+						shape_type=ShapeType.LIST_OF_TEXTS,
+					),
+					ShapeDescriptor(
+						name='scopeListMeta',
+						description='Required to nicely render the scope list in assistant',
+						shape_type=ShapeType.TEXT,
+					),
+				],
+				output_shape=[
+					ShapeDescriptor(
+						name='questions',
+						description='The individual questions that were asked, in order',
+						shape_type=ShapeType.LIST_OF_TEXTS,
+					),
+					ShapeDescriptor(
+						name='answers',
+						description='The answers generated by the model, in the same order as the questions',
+						shape_type=ShapeType.LIST_OF_TEXTS,
+					),
+					ShapeDescriptor(
+						name='sources',
+						description='The sources referenced to generate this answer',
+						shape_type=ShapeType.LIST_OF_TEXTS,
+					),
+				],
+			)
+
+			multi_provider = TaskProcessingProvider(
+				id=MULTI_PROVIDER_ID,
+				name='Context Chat',
+				task_type=MULTI_TASKTYPE_ID,
+				# multiple sequential LLM calls, so allow more time than a single question
+				expected_runtime=30 * MAX_MULTI_QUESTIONS,
+				input_shape_defaults={
+					'maxQuestions': MAX_MULTI_QUESTIONS,
+				},
+			)
+			nc.providers.task_processing.register(multi_provider, multi_tasktype)
+
 			app_enabled.set()
 			if THREAD_STOP_EVENT.is_set():
 				# If the threads were previously stopped, we start them again
@@ -219,6 +280,7 @@ def enabled_handler(enabled: bool, nc: NextcloudApp | AsyncNextcloudApp) -> str:
 			app_enabled.clear()
 			nc.providers.task_processing.unregister(SEARCH_PROVIDER_ID)
 			nc.providers.task_processing.unregister(CC_PROVIDER_ID)
+			nc.providers.task_processing.unregister(MULTI_PROVIDER_ID)
 
 			wait_for_bg_threads()
 	except Exception as e:

context_chat_backend/task_fetcher.py

@@ -22,7 +22,7 @@
 from .chain.context import do_doc_search
 from .chain.ingest.injest import embed_sources
 from .chain.one_shot import process_context_query
-from .chain.types import ContextException, EnrichedSourceList, LLMOutput, ScopeList, SearchResult
+from .chain.types import ContextException, EnrichedSourceList, LLMOutput, MultiOutput, ScopeList, SearchResult
 from .dyn_loader import LLMModelLoader, VectorDBLoader
 from .network_em import NetworkEmbeddings
 from .types import (
@@ -62,6 +62,7 @@
 TP_CHECK_INTERVAL_WITH_TRIGGER = 5 * 60
 TP_CHECK_INTERVAL_ON_ERROR = 15
 CONTEXT_LIMIT = 30
+MAX_MULTI_QUESTIONS = 20
 
 
 class ThreadType(Enum):
@@ -513,8 +514,16 @@ def request_processing_thread(app_config: TConfig, get_enabled_state) -> None:
 			# Fetch pending task
 			try:
 				response = nc.providers.task_processing.next_task(
-					['context_chat-context_chat', 'context_chat-context_chat_search'],
-					['context_chat:context_chat', 'context_chat:context_chat_search'],
+					[
+						'context_chat-context_chat',
+						'context_chat-context_chat_search',
+						'context_chat-context_chat_multi',
+					],
+					[
+						'context_chat:context_chat',
+						'context_chat:context_chat_search',
+						'context_chat:context_chat_multi',
+					],
 				)
 				if not response:
 					wait_for_tasks()
@@ -548,6 +557,25 @@ def request_processing_thread(app_config: TConfig, get_enabled_state) -> None:
 					success = return_result_to_nextcloud(task['id'], userId, {
 						'sources': enrich_sources(search_result, userId),
 					})
+				elif task['type'] == 'context_chat:context_chat_multi':
+					multi_result = process_multi_task(task, vectordb_loader, llm, app_config)
+					# enrich every source in a single API call, then regroup the
+					# enriched results back into one list per question
+					sources_per_question = multi_result['sources_per_question']
+					counts = [len(group) for group in sources_per_question]
+					flat_sources = [source for group in sources_per_question for source in group]
+					enriched_flat = enrich_sources(flat_sources, userId)
+					grouped_sources: list[str] = []
+					i = 0
+					for count in counts:
+						grouped_sources.append('[' + ','.join(enriched_flat[i:i + count]) + ']')
+						i += count
+					# Return result to Nextcloud
+					success = return_result_to_nextcloud(task['id'], userId, {
+						'questions': multi_result['questions'],
+						'answers': multi_result['answers'],
+						'sources': grouped_sources,
+					})
 				else:
 					LOGGER.error(f'Unknown task type {task["type"]}')
 					success = return_error_to_nextcloud(task['id'], Exception(f'Unknown task type {task["type"]}'))
@@ -693,6 +721,81 @@ def process_normal_task(
 		)
 	)
 
+
+
+
+def _normalize_questions(raw_questions: list[str] | None) -> list[str]:
+	"""Clean up a list of questions: strip whitespace, drop empty entries, cap the count."""
+	questions = [q.strip() for q in (raw_questions or [])]
+	questions = [q for q in questions if q]
+	return questions[:MAX_MULTI_QUESTIONS]
+
+def process_multi_task(
+	task: dict[str, Any],
+	vectordb_loader: VectorDBLoader,
+	llm: LLM,
+	app_config: TConfig,
+) -> MultiOutput:
+	"""
+	Process a task containing several questions (one per line), answering each
+	sequentially against the same scope, and collecting all results.
+
+	Args:
+		task: Task dictionary from fetch_query_tasks_from_nextcloud
+		vectordb_loader: Vector database loader instance
+		llm: Language model instance
+		app_config: Application configuration
+
+	Returns:
+		dict with 'questions' (as asked), 'answers' (in the same order) and 'sources' (deduplicated across all answers)
+
+	Raises:
+		ValueError: if no valid question was found in the prompt
+		Various exceptions from query execution
+	"""
+	user_id = task['userId']
+	task_input = task['input']
+	if task_input.get('scopeType') == 'none':
+		task_input['scopeType'] = None
+
+	questions = _normalize_questions(task_input.get('questions'))
+	if not questions:
+		raise ValueError('No questions found. Please provide at least one questione.')
+
+	answers: list[str] = []
+	sources_per_question: list[list[SearchResult]] = []
+
+	for question in questions:
+		result: LLMOutput = exec_in_proc(target=process_context_query,
+			args=(
+				user_id,
+				vectordb_loader,
+				llm,
+				app_config,
+				question,
+				CONTEXT_LIMIT,
+				task_input.get('scopeType'),
+				task_input.get('scopeList'),
+				app_config.llm[1].get('template'),
+			)
+		)
+		answers.append(result['output'])
+		# de-duplicate sources within this single question's own answer
+		seen_sources: set[str] = set()
+		question_sources: list[SearchResult] = []
+		for source in result['sources']:
+			key = getattr(source, 'id', None) or str(source)
+			if key not in seen_sources:
+				seen_sources.add(key)
+				question_sources.append(source)
+		sources_per_question.append(question_sources)
+
+	return {
+		'questions': questions,
+		'answers': answers,
+		'sources_per_question': sources_per_question,
+	}
+
 def process_search_task(
 	task: dict[str, Any],
 	vectordb_loader: VectorDBLoader,