diff --git a/context_chat_backend/chain/types.py b/context_chat_backend/chain/types.py index 3afdf297..f995e493 100644 --- a/context_chat_backend/chain/types.py +++ b/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 diff --git a/context_chat_backend/controller.py b/context_chat_backend/controller.py index 12d86fa3..61d53e66 100644 --- a/context_chat_backend/controller.py +++ b/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: diff --git a/context_chat_backend/task_fetcher.py b/context_chat_backend/task_fetcher.py index 69523979..6643c82c 100644 --- a/context_chat_backend/task_fetcher.py +++ b/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,