* fix: filter empty content in rebuild embeddings queries Update collect_items_for_rebuild() to properly filter out items with empty or whitespace-only content before submitting embedding jobs. Changes: - Sources: add string::trim(full_text) != '' filter - Notes: add string::trim(content) != '' filter - Insights: add content != none AND string::trim(content) != '' filter (previously had no content filter at all) This prevents unnecessary job submissions that would fail validation in the individual embed commands. Ref #513 * feat: add command_id to embedding error logs Add get_command_id() helper to extract command_id from execution context. Include command_id in error logs for all embedding commands: - embed_note_command - embed_insight_command - embed_source_command - create_insight_command This makes it easier to trace failed embedding jobs back to specific command records in the database. Ref #513 * fix: improve logging for embedding commands Log improvements: - Add command_id to all embedding error logs for traceability - Transaction conflicts in repo_insert now log at DEBUG (not ERROR) - Embedding API errors log at DEBUG, only ERROR when retries exhausted - Friendlier retry messages: "This will be retried automatically" - Include model name and command_id in generate_embeddings errors Files changed: - commands/embedding_commands.py: command_id in logs, friendlier messages - open_notebook/database/repository.py: DEBUG for transaction conflicts - open_notebook/utils/embedding.py: DEBUG logging, pass-through command_id Ref #513 * fix: correct field names in rebuild embeddings status endpoint The API status endpoint was looking for wrong field names: - sources_processed → sources_submitted - notes_processed → notes_submitted - insights_processed → insights_submitted - processed_items → jobs_submitted - failed_items → failed_submissions The command outputs "_submitted" because embedding happens async (we count jobs submitted, not items processed). Ref #513 * fix: update rebuild UI text to reflect async job submission Changed terminology from "Completed/processed" to "Jobs Submitted" since the rebuild command submits embedding jobs for async processing, not completing them synchronously. Updated in all locales: en-US, pt-BR, zh-CN, zh-TW, ja-JP Ref #513 * refactor: migrate retry strategy from allowlist to blocklist - Change from `retry_on: [RuntimeError, ...]` to `stop_on: [ValueError]` - This is more resilient: new exception types auto-retry by default - Simplified exception handling: ValueError = permanent, else = retry - Transient errors logged at DEBUG (surreal-commands logs final failure) - Permanent errors (ValueError) logged at ERROR Ref #513
5.4 KiB
5.4 KiB
Commands Module
Purpose: Defines async command handlers for long-running operations via surreal-commands job queue system.
Key Components
Embedding Commands
embed_note_command: Embeds a single note using unified embedding pipeline with content-type aware processing. Uses MARKDOWN content type detection. Retry: 5 attempts, exponential jitter 1-60s.embed_insight_command: Embeds a single source insight. Uses MARKDOWN content type. Retry: 5 attempts, exponential jitter 1-60s.embed_source_command: Embeds a source by chunking full_text with content-type aware splitters (HTML, Markdown, plain), then batch embedding all chunks. Uses single Esperanto API call. Retry: 5 attempts, exponential jitter 1-60s.create_insight_command: Creates a source insight with automatic retry on transaction conflicts. Creates the DB record, then submitsembed_insightcommand (fire-and-forget). Retry: 5 attempts, exponential jitter 1-60s. Used bySource.add_insight().rebuild_embeddings_command: Submits individual embed_* commands for all sources/notes/insights. Returns immediately; actual embedding happens async. No retry (coordinator only).
Other Commands
process_source_command: Ingests content throughsource_graph, creates embeddings (optional), and generates insights. Retries on transaction conflicts (exp. jitter, max 15×, 1-120s).run_transformation_command: Runs a transformation on an existing source to generate an insight. Executes the transformation graph (LLM call) then creates insight viacreate_insight_command. Used byPOST /sources/{id}/insightsAPI endpoint. Retry: 5 attempts, exponential jitter 1-60s.generate_podcast_command: Creates podcasts viapodcast-creatorlibrary using stored episode/speaker profiles.process_text_command(example): Test fixture for text operations (uppercase, lowercase, reverse, word_count).analyze_data_command(example): Test fixture for numeric aggregations.
Important Patterns
- Pydantic I/O: All commands use
CommandInput/CommandOutputsubclasses for type safety and serialization. - Error handling: Permanent errors (ValueError) return failure output; all other exceptions auto-retry via surreal-commands.
- Retry configuration: Uses
stop_on: [ValueError](blocklist approach) - retries all exceptions EXCEPT ValueError. This is more resilient than allowlist as new exception types auto-retry. - Fire-and-forget embedding: Domain models submit embed_* commands via
submit_command()without waiting. Commands process asynchronously. - Content-type aware chunking:
embed_source_commanduseschunk_text()with automatic content type detection (HTML, Markdown, plain text) for optimal text splitting. Default: 1500 char chunks with 225 char overlap. - Batch embedding:
embed_source_commandusesgenerate_embeddings()for single API call efficiency instead of per-chunk calls. - Mean pooling for large content:
embed_note_commandandembed_insight_commandusegenerate_embedding()which handles content larger than chunk size via mean pooling. - Model dumping: Recursive
full_model_dump()utility converts Pydantic models → dicts for DB/API responses. - Logging: Uses
loguru.loggerthroughout; logs execution start/end and key metrics (processing time, counts). - Time tracking: All commands measure
start_time→processing_timefor monitoring.
Dependencies
External: surreal_commands (command decorator, job queue, submit_command), loguru, pydantic, podcast_creator
Internal: open_notebook.domain.notebook (Source, Note, SourceInsight), open_notebook.utils.chunking (chunk_text, detect_content_type), open_notebook.utils.embedding (generate_embedding, generate_embeddings), open_notebook.database.repository (repo_query, repo_insert)
Quirks & Edge Cases
- source_commands:
ensure_record_id()wraps command IDs for DB storage; transaction conflicts trigger exponential backoff retry. ValueError exceptions are permanent (not retried). - embedding_commands: Content type detection uses file extension as primary source, heuristics as fallback. Chunks >1800 chars trigger secondary splitting. Empty/whitespace-only content returns ValueError (not retried).
- rebuild_embeddings_command: Returns "jobs_submitted" not "processed_items" - embedding is async. Individual commands handle failures with their own retries.
- podcast_commands: Profiles loaded from SurrealDB by name (must exist); briefing can be extended with suffix. Episode records created mid-execution.
- Example commands: Accept optional
delay_secondsfor testing async behavior; not for production.
Code Example
@command("process_source", app="open_notebook", retry={
"max_attempts": 5,
"wait_strategy": "exponential_jitter",
"stop_on": [ValueError], # Don't retry validation errors
})
async def process_source_command(input_data: SourceProcessingInput) -> SourceProcessingOutput:
start_time = time.time()
try:
transformations = [await Transformation.get(id) for id in input_data.transformations]
source = await Source.get(input_data.source_id)
result = await source_graph.ainvoke({...})
return SourceProcessingOutput(success=True, ...)
except ValueError as e:
return SourceProcessingOutput(success=False, error_message=str(e)) # No retry
except Exception as e:
raise # Retry all other exceptions