# How to Debug Semantic Processing Pipeline Issues in OpenViking

> Debug semantic processing pipeline issues in OpenViking by enabling DEBUG logging, checking worker threads, inspecting DagStats, and instrumenting key methods to pinpoint failures.

- Repository: [Volcengine/OpenViking](https://github.com/volcengine/OpenViking)
- Tags: how-to-guide
- Published: 2026-03-08

---

**To debug semantic processing pipeline issues in OpenViking, enable DEBUG logging to trace message flow, verify QueueManager worker thread health, inspect DagStats for stuck nodes, and instrument the `_file_summary_task` and `_overview_task` methods in [`semantic_dag.py`](https://github.com/volcengine/OpenViking/blob/main/semantic_dag.py) to isolate failures.**

The semantic processing pipeline in the **volcengine/OpenViking** repository automatically generates [`.abstract.md`](https://github.com/volcengine/OpenViking/blob/main/.abstract.md) and [`.overview.md`](https://github.com/volcengine/OpenViking/blob/main/.overview.md) files for directories and pushes them to the embedding queue. Because the system relies on asynchronous workers and a lazy DAG executor, failures often manifest as stalled queues, missing summaries, or silently skipped directories. This guide walks you through systematic debugging using the actual source code paths and diagnostic hooks exposed by the `SemanticQueue`, `SemanticProcessor`, and `SemanticDagExecutor` components.

## Understanding the Pipeline Architecture

Before debugging, map the three core components that handle every semantic message:

- **SemanticQueue** ([`openviking/storage/queuefs/semantic_queue.py`](https://github.com/volcengine/OpenViking/blob/main/openviking/storage/queuefs/semantic_queue.py)): Wraps `NamedQueue` to enqueue `SemanticMsg` objects containing target URIs and recursion flags.
- **SemanticProcessor** ([`openviking/storage/queuefs/semantic_processor.py`](https://github.com/volcengine/OpenViking/blob/main/openviking/storage/queuefs/semantic_processor.py)): The dequeue handler that receives messages and initiates the DAG executor.
- **SemanticDagExecutor** ([`openviking/storage/queuefs/semantic_dag.py`](https://github.com/volcengine/OpenViking/blob/main/openviking/storage/queuefs/semantic_dag.py)): Executes a lazy DAG where directory overviews wait for child file summaries to complete.

The flow is orchestrated by `QueueManager` ([`openviking/storage/queuefs/queue_manager.py`](https://github.com/volcengine/OpenViking/blob/main/openviking/storage/queuefs/queue_manager.py)), which creates the queues and manages worker threads. When a message enters `SemanticQueue.enqueue()`, a worker eventually triggers `SemanticProcessor.on_dequeue()`, launching the recursive walk via `SemanticDagExecutor.run()`.

## Enable Verbose Logging

OpenViking uses the standard Python `logging` module centralized in [`openviking_cli/utils/logger.py`](https://github.com/volcengine/OpenViking/blob/main/openviking_cli/utils/logger.py). Switch from `INFO` to `DEBUG` to see state transitions in `SemanticProcessor` and `SemanticDagExecutor`.

```python
from openviking_cli.utils.logger import get_logger

logger = get_logger("openviking")
logger.setLevel("DEBUG")

```

With debug logging active, you will see entries such as `Processing semantic generation for: viking://user/1234/project (recursive=True)` in [`semantic_processor.py`](https://github.com/volcengine/OpenViking/blob/main/semantic_processor.py), plus `Dispatching directory …` and `File summary completed for …` entries from the DAG executor. If these logs stop appearing mid-stream, the worker thread likely encountered an unhandled exception.

## Verify Queue Health and Worker Status

### Inspect Queue Depth

If enqueued messages never process, check whether the queue is actually draining:

```python
import asyncio
from openviking.storage.queuefs import get_queue_manager

qm = get_queue_manager()
semantic_q = qm.get_queue(qm.SEMANTIC)
print("Queue size:", asyncio.run(semantic_q.size()))

```

A static size after enqueueing indicates the worker is stalled or dead.

### Check Worker Thread Vitality

Each queue type has a dedicated worker thread managed by `QueueManager._queue_worker_loop`. Verify liveness directly:

```python
import threading

for name, thread in qm._queue_threads.items():
    print(f"{name} alive: {thread.is_alive()}")

```

If the `SEMANTIC` thread is not alive, inspect the logs for stack traces originating from [`semantic_processor.py`](https://github.com/volcengine/OpenViking/blob/main/semantic_processor.py) or [`semantic_dag.py`](https://github.com/volcengine/OpenViking/blob/main/semantic_dag.py).

## Examine DAG Statistics for Stuck Nodes

`SemanticDagExecutor` exposes runtime statistics via `get_stats()`, returning a `DagStats` dataclass defined in [`semantic_dag.py`](https://github.com/volcengine/OpenViking/blob/main/semantic_dag.py). Access this during a debugging session to pinpoint bottlenecks:

```python

# Assuming you have a reference to the running processor

executor = processor._dag_executor  # Set during on_dequeue

stats = executor.get_stats()
print(f"Total: {stats.total_nodes}, Pending: {stats.pending_nodes}, "
      f"In-progress: {stats.in_progress_nodes}, Done: {stats.done_nodes}")

```

Interpret the metrics as follows:

- **`pending_nodes` > 0 indefinitely**: A child directory never reported completion. Check the `_overview_task` in [`semantic_dag.py`](https://github.com/volcengine/OpenViking/blob/main/semantic_dag.py) (lines 40–73) for the parent URI.
- **`in_progress_nodes` stuck**: A file summary task raised an exception without calling `_on_file_done`, leaving the DAG waiting forever.

## Validate Individual Task Execution

### Instrument File Summary Tasks

The `_file_summary_task` method (lines 44–58 in [`semantic_dag.py`](https://github.com/volcengine/OpenViking/blob/main/semantic_dag.py)) generates LLM summaries for individual files. Wrap it to capture exceptions:

```python
async def _file_summary_task_debug(self, parent_uri, file_path):
    try:
        await self._file_summary_task(parent_uri, file_path)
    except Exception as e:
        from openviking_cli.utils.logger import get_logger
        logger = get_logger("openviking")
        logger.error(f"DEBUG: file summary failed for {file_path}: {e}", exc_info=True)
        raise

# Patch at runtime for debugging

processor._file_summary_task = _file_summary_task_debug.__get__(processor)

```

### Monitor Overview Generation

Similarly, instrument `_overview_task` (lines 40–73 in [`semantic_dag.py`](https://github.com/volcengine/OpenViking/blob/main/semantic_dag.py)) to log the exact file lists and child URIs it receives before building the directory overview. This reveals when the LLM receives empty or malformed context.

## Check LLM Concurrency and Configuration

The pipeline throttles LLM calls via a semaphore initialized from `max_concurrent_llm` in [`openviking_cli/utils/config/vlm_config.py`](https://github.com/volcengine/OpenViking/blob/main/openviking_cli/utils/config/vlm_config.py). If this value is too low for large directories, tasks block indefinitely waiting for the semaphore (`self._llm_sem` in the executor).

Verify the current limit:

```python
from openviking_cli.utils.config import get_openviking_config
cfg = get_openviking_config()
print(f"Max concurrent LLM calls: {cfg.vlm.max_concurrent_llm}")

```

Rate-limit errors or network timeouts surface as warnings inside `_generate_single_file_summary` in [`semantic_processor.py`](https://github.com/volcengine/OpenViking/blob/main/semantic_processor.py) (lines 46–55). Ensure the LLM client in [`openviking/utils/llm.py`](https://github.com/volcengine/OpenViking/blob/main/openviking/utils/llm.py) is correctly configured and responsive.

## Isolate with a Minimal Test Case

Create a tiny directory with two or three text files and trigger the pipeline manually to rule out data-scale issues:

```python
import asyncio
from openviking.storage.queuefs.semantic_msg import SemanticMsg
from openviking.storage.queuefs.semantic_queue import SemanticQueue
from openviking_cli.utils import VikingURI

async def minimal_test():
    uri = VikingURI("viking://user/1234/tmp_test").uri
    msg = SemanticMsg(
        uri=uri,
        recursive=True,
        account_id="1234",
        user_id="1234",
        agent_id="",
        role="ROOT",
        context_type="default",
    )
    q = SemanticQueue(agfs=None)  # In production, QueueManager injects agfs

    await q.enqueue(msg)

asyncio.run(minimal_test())

```

Observe the logs and `DagStats`—if this succeeds while your production batch fails, the issue is likely resource exhaustion or a specific malformed file in the larger dataset.

## Key Source Files and Debug Entry Points

Keep these files open when tracing failures:

| File | Key Function / Class | Debug Purpose |
|------|---------------------|---------------|
| [`openviking/storage/queuefs/semantic_queue.py`](https://github.com/volcengine/OpenViking/blob/main/openviking/storage/queuefs/semantic_queue.py) | `SemanticQueue.enqueue()` | Verify message enqueue logic |
| [`openviking/storage/queuefs/semantic_processor.py`](https://github.com/volcengine/OpenViking/blob/main/openviking/storage/queuefs/semantic_processor.py) | `on_dequeue()` (lines 53–84) | Entry point for DAG creation |
| [`openviking/storage/queuefs/semantic_dag.py`](https://github.com/volcengine/OpenViking/blob/main/openviking/storage/queuefs/semantic_dag.py) | `run()`, `_dispatch_dir()` (lines 71–98) | Recursive directory dispatch |
| [`openviking/storage/queuefs/semantic_dag.py`](https://github.com/volcengine/OpenViking/blob/main/openviking/storage/queuefs/semantic_dag.py) | `_file_summary_task()` (lines 44–58) | File-level LLM summary |
| [`openviking/storage/queuefs/semantic_dag.py`](https://github.com/volcengine/OpenViking/blob/main/openviking/storage/queuefs/semantic_dag.py) | `_overview_task()` (lines 40–73) | Directory overview generation |
| [`openviking/storage/queuefs/semantic_dag.py`](https://github.com/volcengine/OpenViking/blob/main/openviking/storage/queuefs/semantic_dag.py) | `DagStats`, `get_stats()` | Health metrics |
| [`openviking/storage/queuefs/queue_manager.py`](https://github.com/volcengine/OpenViking/blob/main/openviking/storage/queuefs/queue_manager.py) | `_queue_worker_loop()` (lines 64–84) | Worker thread management |
| [`openviking_cli/utils/logger.py`](https://github.com/volcengine/OpenViking/blob/main/openviking_cli/utils/logger.py) | `get_logger()` | Logging configuration |
| [`openviking_cli/utils/config/vlm_config.py`](https://github.com/volcengine/OpenViking/blob/main/openviking_cli/utils/config/vlm_config.py) | `max_concurrent_llm` | Concurrency limits |

## Summary

- **Enable DEBUG logging** via `openviking_cli.utils.logger` to trace the enqueue-to-completion lifecycle in [`semantic_processor.py`](https://github.com/volcengine/OpenViking/blob/main/semantic_processor.py).
- **Verify worker health** by checking `QueueManager._queue_threads` liveness and `SemanticQueue.size()` trends.
- **Inspect DagStats** from `SemanticDagExecutor.get_stats()` to identify stuck pending or in-progress nodes.
- **Instrument tasks** by wrapping `_file_summary_task` and `_overview_task` in [`semantic_dag.py`](https://github.com/volcengine/OpenViking/blob/main/semantic_dag.py) to catch silent exceptions.
- **Validate LLM configuration** in [`vlm_config.py`](https://github.com/volcengine/OpenViking/blob/main/vlm_config.py) to ensure the semaphore does not block execution.
- **Test minimally** with a single small directory to isolate data-specific corruption from systemic failures.

## Frequently Asked Questions

### Why does my semantic queue size never decrease after enqueueing?

The worker thread responsible for the `SEMANTIC` queue may be dead or blocked. Check `qm._queue_threads` for thread liveness and review logs for unhandled exceptions in `SemanticProcessor.on_dequeue()`. If the thread is alive but idle, the DAG executor may be waiting on a semaphore or stuck LLM call.

### How do I identify which file is causing the pipeline to hang?

Access the running `SemanticDagExecutor` instance via `processor._dag_executor` and call `get_stats()`. If `in_progress_nodes` is non-zero, the stuck node corresponds to a file currently inside `_file_summary_task`. Temporarily patch that method to log every `file_path` entry before awaiting the LLM call.

### What causes empty [`.overview.md`](https://github.com/volcengine/OpenViking/blob/main/.overview.md) files in large directories?

Empty overviews usually indicate that `_overview_task` ran before child file summaries completed, or the LLM returned an empty response. Check `DagStats.pending_nodes`—if the count dropped to zero prematurely, a child task may have crashed without signaling completion. Also verify `max_concurrent_llm` is sufficient for your directory size to prevent timeouts.

### Where is the LLM client configured for the semantic pipeline?

The LLM client is configured in [`openviking/utils/llm.py`](https://github.com/volcengine/OpenViking/blob/main/openviking/utils/llm.py) and consumed by [`semantic_processor.py`](https://github.com/volcengine/OpenViking/blob/main/semantic_processor.py). Concurrency is controlled by `max_concurrent_llm` defined in [`openviking_cli/utils/config/vlm_config.py`](https://github.com/volcengine/OpenViking/blob/main/openviking_cli/utils/config/vlm_config.py). Reduce this value if you hit rate limits, or increase it if the pipeline stalls on IO-bound LLM calls.