Update logging

service/README.md

@@ -45,6 +45,10 @@ The file path can be overridden globally via `AI_DOCUMENT_PLUGIN_CONFIG_PATH`. S
 Prompt templates and LLM parameters for each step. In `config.yaml`, `files.prompts_path` must be a relative path and
 is resolved relative to the active config file.
 
+## Logging
+
+The service logs only to standard output. Configure the verbosity under `logging.level`.
+
 ## Make commands
 
 The project `Makefile` provides a few shortcuts for common development tasks:

service/config.template.yaml

@@ -29,4 +29,4 @@ files:
 # This specifies how many DMPs can be processed at the same time.
 # Each DMP process can create further parallel lines of execution depending on the tenants setting.
 # This limit is server-wide, that means it has the same counter for all tenants.
-max_parallel_executions: 2
+max_parallel_executions: 2

service/pyproject.toml

@@ -34,7 +34,6 @@ dependencies = [
     "sqlalchemy[asyncio]",
     "tabulate",
     "tiktoken",
-    "tqdm",
 ]
 
 [project.scripts]

service/requirements.txt

@@ -157,7 +157,6 @@ tiktoken==0.13.0
     # via ai-document-plugin-service (pyproject.toml)
 tqdm==4.70.0
     # via
-    #   ai-document-plugin-service (pyproject.toml)
     #   haystack-ai
     #   openai
 truststore==0.10.4

service/src/ai_document_plugin_service/ai/assignment/assignment_component.py

@@ -1,6 +1,7 @@
 import asyncio
 import itertools
 import logging
+import time
 from collections.abc import Callable
 from typing import Any, TypedDict
 
@@ -73,10 +74,19 @@ async def run_async(
         km: dict[str, Any],
         on_progress: Callable[[str], None] | None = None,
     ) -> AssignmentComponentResult:
+        started = time.perf_counter()
         logger.debug('Step 1: Assigning questions to sections...')
 
         sections = build_section_records(template_data)
         question_chunks, question_id_to_path = build_question_chunks(data)
+        logger.info(
+            'Starting question-to-section assignment',
+            extra={
+                'section_count': len(sections),
+                'question_chunk_count': len(question_chunks),
+                'question_path_count': len(question_id_to_path),
+            },
+        )
         stats = AssignmentStats()
 
         section_formatter = SectionFormatter(sections)
@@ -119,6 +129,15 @@ async def match_chunk(question_chunk: str) -> dict[str, list[str]]:
             result_mapping,
             km,
         )
+        logger.info(
+            'Completed question-to-section assignment',
+            extra={
+                'assignment_count': len(assignments),
+                'mapped_question_count': len(result_mapping),
+                'llm_call_count': stats.total_calls,
+            },
+        )
+        stats.set_duration_ms(round((time.perf_counter() - started) * 1000, 3))
 
         return {
             'assignments': assignments,

service/src/ai_document_plugin_service/ai/assignment/llm.py

@@ -1,6 +1,5 @@
 import json
 import logging
-import pathlib
 from abc import ABC, abstractmethod
 from json import JSONDecodeError
 from typing import TYPE_CHECKING
@@ -78,16 +77,17 @@ async def match_questions_to_sections(
 
         async def call_and_parse() -> dict[str, list[str]]:
             response = await self.client.completion(
+                stats=stats,
                 messages=messages,
                 temperature=self.config.assignment.temperature,
                 max_tokens=self.config.assignment.max_tokens,
                 reasoning_effort='low',
             )
             choice = response.choices[0]
             if choice.finish_reason != 'stop':
-                logger.debug(
-                    'Model did not stop generating naturally: %s',
-                    choice,
+                logger.error(
+                    'Model did not stop generating naturally',
+                    extra={'finish_reason': choice.finish_reason},
                 )
                 msg = 'Model did not stop generating naturally.'
                 raise ModelDidNotStopError(msg)
@@ -97,6 +97,7 @@ async def call_and_parse() -> dict[str, list[str]]:
                 return self._parse_json_question_to_sections(content)
             except JSONDecodeError as e:
                 msg = 'Unable to parse: ' + content
+                logger.exception('Unable to parse LLM assignment response as JSON', exc_info=e)
                 raise UnableToParseResponseError(msg) from e
 
         return await call_with_retry(call_and_parse)
@@ -113,6 +114,7 @@ def _parse_json_question_to_sections(content: str) -> dict[str, list[str]]:
 
         if not isinstance(data, dict):
             msg = 'LLM response is not a JSON object.'
+            logger.error('LLM assignment response is not a JSON object')
             raise TypeError(msg)
         result: dict[str, list[str]] = {}
         for id_str, section_list in data.items():
@@ -123,31 +125,3 @@ def _parse_json_question_to_sections(content: str) -> dict[str, list[str]]:
             else:
                 result[str(id_str)] = []
         return result
-
-
-class LoggingNoopLayerMatcher(LayerMatcher):
-    """Logs assignment inputs to logger and a JSONL file; returns no mappings."""
-
-    def __init__(self, log_path: str | pathlib.Path) -> None:
-        self._log_path = pathlib.Path(log_path)
-
-    async def match_questions_to_sections(
-        self,
-        sections_xml: str,
-        question_chunk_xml: str,
-        stats: AssignmentStats,
-    ) -> dict[str, list[str]]:
-        record = {
-            'sections_xml': sections_xml,
-            'question_chunk_xml': question_chunk_xml,
-            'stats': stats.to_dict(),
-        }
-        self._log_path.parent.mkdir(parents=True, exist_ok=True)
-        with self._log_path.open('a', encoding='utf-8') as handle:
-            handle.write(json.dumps(record, ensure_ascii=False) + '\n')
-        logger.info(
-            'LoggingNoopLayerMatcher appended inputs to %s stats=%s',
-            self._log_path,
-            stats,
-        )
-        return {}

service/src/ai_document_plugin_service/ai/assignment/sections/llm.py

@@ -1,8 +1,8 @@
+import json
+import logging
 import typing
 from abc import ABC, abstractmethod
 
-from tqdm import tqdm
-
 from ai_document_plugin_service.ai.assignment.types import LeafSection
 from ai_document_plugin_service.ai.common.config import Config
 from ai_document_plugin_service.ai.common.llm_client import (
@@ -12,6 +12,8 @@
 )
 from ai_document_plugin_service.ai.common.types import AssignmentStats
 
+logger = logging.getLogger(__name__)
+
 
 class SectionIdGenerator(ABC):
     @abstractmethod
@@ -47,8 +49,9 @@ async def generate_leaf_section_ids(
         user_tpl = self.config.section_id.user_message
         result: dict[str, str] = {}
         used_ids: set[str] = set()
+        total_sections = len(leaf_sections)
 
-        for leaf in tqdm(leaf_sections):
+        for index, leaf in enumerate(leaf_sections, start=1):
             existing_str = ', '.join(sorted(used_ids)) if used_ids else '(none yet)'
             content_block = leaf.text.strip()
             user_msg = (
@@ -77,6 +80,17 @@ async def generate_leaf_section_ids(
                 sid = f'{sid}_{len(used_ids)}'
             used_ids.add(sid)
             result[leaf.id] = sid
+            logger.info(
+                json.dumps(
+                    {
+                        'event': 'generating_section_identifiers_progress',
+                        'completed_sections': index,
+                        'total_sections': total_sections,
+                    },
+                    ensure_ascii=False,
+                    sort_keys=True,
+                ),
+            )
 
         return result
 
@@ -98,6 +112,18 @@ async def generate_leaf_section_ids(  # ty: ignore[invalid-method-override]
         _: AssignmentStats,
     ) -> dict[str, str]:
         res = {}
-        for i, leaf in tqdm(enumerate(leaf_sections)):
-            res[leaf.id] = f'{leaf.id}_{i}'
+        total_sections = len(leaf_sections)
+        for index, leaf in enumerate(leaf_sections, start=1):
+            res[leaf.id] = f'{leaf.id}_{index - 1}'
+            logger.info(
+                json.dumps(
+                    {
+                        'event': 'generating_section_identifiers_progress',
+                        'completed_sections': index,
+                        'total_sections': total_sections,
+                    },
+                    ensure_ascii=False,
+                    sort_keys=True,
+                ),
+            )
         return res

service/src/ai_document_plugin_service/ai/common/config.py

@@ -79,7 +79,7 @@ def _normalize_path(path: str) -> str:
     return str(pathlib.Path(_expand_env_vars(path).strip()).expanduser())
 
 
-def _get(config: dict[str, Any], *path: str) -> Any:  # noqa: ANN401
+def _get(config: dict[str, Any], *path: str, allow_empty_string: bool = False) -> Any:  # noqa: ANN401
     current = config
     for key in path:
         if not isinstance(current, dict) or key not in current:
@@ -92,7 +92,7 @@ def _get(config: dict[str, Any], *path: str) -> Any:  # noqa: ANN401
     if current is None:
         msg = f"Missing required config value: '{'.'.join(path)}'"
         raise ValueError(msg)
-    if isinstance(current, str) and not current.strip():
+    if isinstance(current, str) and not current.strip() and not allow_empty_string:
         msg = f"Missing required config value: '{'.'.join(path)}'"
         raise ValueError(msg)
     return current
@@ -204,7 +204,6 @@ def resolve_config_path(config_path: str | None = None) -> str:
 def load_config(config_path: str | None = None) -> Config:
     resolved_config_path = _resolve_existing_path(resolve_config_path(config_path))
     config_dir = pathlib.Path(resolved_config_path).parent
-
     with pathlib.Path(resolved_config_path).open(encoding='utf-8') as handle:
         config = yaml.safe_load(handle)
 

service/src/ai_document_plugin_service/ai/common/dynamic_semaphore.py

@@ -57,6 +57,14 @@ def set_limit(self, new_limit: int) -> None:
         self.limit = new_limit
         self._wake_waiters()
 
+    @property
+    def active_count(self) -> int:
+        return self._active_count
+
+    @property
+    def queued_count(self) -> int:
+        return len(self._waiters)
+
     async def acquire(self) -> None:
         loop = self._bind_loop()
         if self._active_count < self.limit:

service/src/ai_document_plugin_service/ai/common/execution_logging.py

@@ -0,0 +1,76 @@
+import logging
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+_run_log_context: ContextVar['RunLogContext | None'] = ContextVar('run_log_context', default=None)
+_LLM_EVENT_LOGGER = logging.getLogger('ai_document_plugin_service.execution.llm')
+_PIPELINE_EVENT_LOGGER = logging.getLogger('ai_document_plugin_service.execution.pipeline')
+
+
+@dataclass(frozen=True)
+class RunLogContext:
+    run_id: str
+    questionnaire_uuid: str | None = None
+    template_uuid: str | None = None
+    template_title: str | None = None
+    knowledge_model_uuid: str | None = None
+    user_uuid: str | None = None
+    tenant_uuid: str | None = None
+
+
+@contextmanager
+def run_log_context(context: RunLogContext) -> 'Iterator[None]':
+    token = _run_log_context.set(context)
+    try:
+        yield
+    finally:
+        _run_log_context.reset(token)
+
+
+def get_run_log_context() -> RunLogContext | None:
+    return _run_log_context.get()
+
+
+def utc_now_iso() -> str:
+    return datetime.now(tz=UTC).isoformat()
+
+
+def log_llm_event(record: dict[str, Any]) -> None:
+    _emit_event(_LLM_EVENT_LOGGER, record)
+
+
+def log_timing_event(event: str, **fields: Any) -> None:  # noqa: ANN401
+    _emit_event(_PIPELINE_EVENT_LOGGER, {'event': event, **fields})
+
+
+def log_semaphore_event(event: str, **fields: Any) -> None:  # noqa: ANN401
+    _emit_event(_LLM_EVENT_LOGGER, {'event': event, **fields})
+
+
+def _with_context(record: dict[str, Any]) -> dict[str, Any]:
+    payload: dict[str, Any] = {
+        **record,
+    }
+    context = get_run_log_context()
+    if context is None:
+        return payload
+
+    payload.update({field: value for field, value in context.__dict__.items() if value is not None})
+    return payload
+
+
+def _emit_event(logger: logging.Logger, record: dict[str, Any]) -> None:
+    payload = _with_context(record)
+    logger.log(_event_log_level(payload), 'Execution event', extra=payload)
+
+
+def _event_log_level(record: dict[str, Any]) -> int:
+    if record.get('status') in {'error', 'failed'} or record.get('state') == 'failed' or 'error_type' in record:
+        return logging.ERROR
+    return logging.INFO

service/src/ai_document_plugin_service/ai/common/json_log_writer.py

@@ -0,0 +1,33 @@
+import json
+import threading
+from collections.abc import Mapping
+from pathlib import Path
+
+from pydantic import BaseModel
+
+type JsonScalar = str | int | float | bool | None
+type JsonValue = JsonScalar | list[JsonValue] | dict[str, JsonValue]
+
+_WRITE_LOCK = threading.Lock()
+
+
+def append_jsonl(path: Path, payload: dict[str, JsonValue]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    serialized = json.dumps(payload, ensure_ascii=False)
+    with _WRITE_LOCK, path.open('a', encoding='utf-8') as handle:
+        handle.write(serialized + '\n')
+
+
+def make_json_safe(value: object) -> JsonValue:
+    result: JsonValue
+    if value is None or isinstance(value, str | int | float | bool):
+        result = value
+    elif isinstance(value, BaseModel):
+        result = make_json_safe(value.model_dump(mode='json'))
+    elif isinstance(value, Mapping):
+        result = {str(key): make_json_safe(item) for key, item in value.items()}
+    elif isinstance(value, list | tuple | set):
+        result = [make_json_safe(item) for item in value]
+    else:
+        result = repr(value)
+    return result