Add harvest_input and knowledge_queue models with pipeline support#989
Add harvest_input and knowledge_queue models with pipeline support#989manshusainishab wants to merge 6 commits into
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yml Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
Summary by CodeRabbit
WalkthroughChangesModule B noise-filter pipeline
Estimated code review effort: 3 (Moderate) | ~25 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
application/utils/noise_filter/pipeline.py (1)
34-47: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
RunSummary.statusis hardcoded to"ok"and never reflects run outcome.The field is initialized once and never updated (e.g., to signal partial failure when
parse_errors > 0), so callers parsing the JSON summary can't distinguish a clean run from one with parse errors except by separately checkingparse_errors. Either drop the unused field or set it meaningfully (e.g.,"partial"whenparse_errorsor dropped rows are non-zero).🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@application/utils/noise_filter/pipeline.py` around lines 34 - 47, Update RunSummary.status so it reflects the run outcome instead of remaining hardcoded to "ok"; set it to the established partial-failure value when parse_errors or dropped_noise is non-zero, while preserving "ok" for clean runs, or remove the field entirely if no meaningful status handling exists.cre.py (1)
304-318: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winInconsistent CLI validation UX vs.
--export/--csv.
--exportwithout--csvis validated inline withparser.error(...)(clean usage message, exit 2).--run_noise_filterwithout--run_idis instead validated insidecre_main.run()by raisingValueError, which surfaces as an unhandled traceback. Consider mirroring the--exportpattern here for a consistent CLI experience.♻️ Suggested fix
args = parser.parse_args() if args.export and not args.csv: parser.error("--export requires --csv <path>") + if args.run_noise_filter and not args.run_id.strip(): + parser.error("--run_noise_filter requires --run_id <pipeline_run_id>")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@cre.py` around lines 304 - 318, Validate the --run_noise_filter and --run_id dependency during CLI argument parsing, alongside the existing --export/--csv validation, using parser.error(...) when --run_noise_filter is set without a nonempty --run_id. Remove or bypass the corresponding ValueError validation in cre_main.run() so invalid CLI input produces the standard usage error instead of an unhandled traceback.application/database/db.py (1)
275-277: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winStatus comment omits the
errorstate actually used by the pipeline.
pipeline.pysetsrow.status = "error"for rows that failChangeRecordvalidation, but this comment only documentspending | processed. Update the comment (and ideally add a comment/constraint enumerating all three values) to keep the state machine documented accurately.📝 Suggested fix
status = sqla.Column( sqla.String, nullable=False, default="pending" - ) # pending | processed + ) # pending | processed | error🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@application/database/db.py` around lines 275 - 277, Update the status documentation on the status column in the database model to enumerate all pipeline states: pending, processed, and error. If an existing constraint or state declaration is present nearby, keep it synchronized with these three values without changing the pipeline behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@application/utils/noise_filter/queue_writer.py`:
- Around line 81-104: Update write_verdicts() to use database-level idempotent
insertion for content_hash, such as PostgreSQL ON CONFLICT DO NOTHING, instead
of relying on the existing pre-query and in-memory deduplication. Ensure
concurrent duplicate hashes are skipped without aborting the transaction or
losing unrelated batch inserts, while preserving the existing WriteStats counts.
---
Nitpick comments:
In `@application/database/db.py`:
- Around line 275-277: Update the status documentation on the status column in
the database model to enumerate all pipeline states: pending, processed, and
error. If an existing constraint or state declaration is present nearby, keep it
synchronized with these three values without changing the pipeline behavior.
In `@application/utils/noise_filter/pipeline.py`:
- Around line 34-47: Update RunSummary.status so it reflects the run outcome
instead of remaining hardcoded to "ok"; set it to the established
partial-failure value when parse_errors or dropped_noise is non-zero, while
preserving "ok" for clean runs, or remove the field entirely if no meaningful
status handling exists.
In `@cre.py`:
- Around line 304-318: Validate the --run_noise_filter and --run_id dependency
during CLI argument parsing, alongside the existing --export/--csv validation,
using parser.error(...) when --run_noise_filter is set without a nonempty
--run_id. Remove or bypass the corresponding ValueError validation in
cre_main.run() so invalid CLI input produces the standard usage error instead of
an unhandled traceback.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yml
Review profile: CHILL
Plan: Pro Plus
Run ID: efb18730-0a64-41ab-87d7-358a43ab2162
📒 Files selected for processing (8)
application/cmd/cre_main.pyapplication/database/db.pyapplication/tests/noise_filter/pipeline_test.pyapplication/tests/noise_filter/queue_writer_test.pyapplication/utils/noise_filter/pipeline.pyapplication/utils/noise_filter/queue_writer.pycre.pymigrations/versions/d4e5f6a7b8c9_add_module_b_tables.py
Summary
Wires Module B (Noise/Relevance Filter) into the orchestrated pipeline: it reads
Module A's harvested chunks from a database table, classifies them (regex →
sanitize → LLM), writes the security-knowledge keepers to a queue for Module C,
and reports a JSON summary. Builds on the Stage 1/1.5/2 code from Weeks 1–4.
What's added
application/database/db.py):harvest_input— Module A writes harvested chunks here (JSONBpayload+pipeline_run_id+status); Module B reads.knowledge_queue— Module B writes classified keepers here (deduped oncontent_hash); Module C reads.queue_writer.py— maps a verdict to aknowledge_queuerow, source-typeaware, deduped on
content_hash; NOISE dropped, KNOWLEDGE/UNCERTAIN kept.pipeline.py(run_noise_filter) — for apipeline_run_id: read pendingharvest_inputrows → regex → sanitize → LLM classify → write keepers → markrows processed → return a
RunSummary.cre.py/cre_main.py) —--run_noise_filter --run_id <id> [--noise_filter_dry_run], prints the summary as JSON and exits (the entrypoint the orchestrator invokes; exit code = completion signal).
d4e5f6a7b8c9— creates the two tables (down_revisionc7d8e9f0a1b2).Behavior
UNCERTAIN always reach the queue.
processed/error; re-running arun_idissafe;
UNIQUE(content_hash)collapses duplicate content.error(not fatal); failed LLM batch→ those chunks become UNCERTAIN (never dropped); infra failure → non-zero exit.
Testing
suite 95/95,
black --checkclean.flask db upgradefrom an empty DBbuilds the schema through our head; a real run (
cre.py --run_noise_filter)classifies a mixed batch, writes KNOWLEDGE rows to
knowledge_queue, and exits0 with a JSON summary.
Notes
queue_writeris the only DB boundary.litellm(slim prod reqs).