DataOps-platform/docs/superpowers/plans/2026-07-16-governance-knowledge-base.md

Governance Knowledge Base Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:test-driven-development and superpowers:executing-plans to implement this plan task-by-task.

Goal: Build a permission-aware data governance knowledge base from business domains, data flows, metadata, standards and labels, with incremental synchronization and daily full consistency audits.

Architecture: Governance changes emit outbox events. A deterministic document builder produces versioned text, Qwen generates embeddings stored in PostgreSQL/pgvector, and DeepSeek answers only from permission-filtered retrieved sources. A daily audit compares source versions/hashes and repairs drift.

Tech Stack: Flask, PostgreSQL 16 + pgvector, Neo4j, Qwen Embedding API, DeepSeek chat API, MinIO, pytest.

---

Task 1: Add pgvector and knowledge schema

Files:

• Modify: deploy/docker/postgres/init/000-init.sql
• Create: migrations/versions/20260716_40_governance_knowledge.py
• Create: app/core/knowledge/models.py

• Create: tests/test_knowledge_schema.py

• [ ] Enable vector extension using a pinned pgvector-compatible PostgreSQL image in local Docker.

• [ ] Create governance_documents, governance_chunks and governance_sync_jobs with stable object UID/version/content hash.

• [ ] Add unique document version and chunk number constraints plus a vector index selected from measured corpus size.

• [ ] Test idempotent upsert, superseded versions, deletion tombstones and permission-scope metadata.

Task 2: Build deterministic governance documents

Files:

• Create: app/core/knowledge/document_builder.py

• Create: tests/test_governance_document_builder.py

• [ ] Add golden tests for BusinessDomain, DataFlow, DataMeta, data standard and label documents.

• [ ] Normalize property order, timestamps and relation sorting so identical source state has identical SHA-256.

• [ ] Include object type, UID, version, business domain, ownership, definitions, relationships and source update time.

• [ ] Exclude secrets, n8n credentials, raw connection passwords and unauthorized attachment text.

Task 3: Implement Qwen embedding adapter

Files:

• Create: app/services/embedding/qwen.py
• Modify: app/config/config.py

• Create: tests/test_qwen_embeddings.py

• [ ] Require explicit Qwen embedding model/base URL/API key and fail closed on missing configuration.

• [ ] Test batching, timeout, rate limit, retry, dimension mismatch, deterministic cache keys and secret-free logs.

• [ ] Store provider/model/dimension with every chunk so a model change creates a controlled reindex job.

• [ ] Keep DeepSeek as the only default generative provider; do not send chat generation to Qwen.

Task 4: Implement incremental synchronization

Files:

• Create: app/core/knowledge/sync.py
• Create: app/commands/sync_governance_knowledge.py

• Create: tests/test_knowledge_incremental_sync.py

• [ ] Consume governance outbox events by stable object UID and source version.

• [ ] Skip unchanged hashes, rebuild changed documents, chunk deterministically, embed missing chunks and atomically publish the new version.

• [ ] Test duplicate/out-of-order events, object deletion, embedding failure and restart midway through a document.

• [ ] Expose job counts, lag and last error to administrator health diagnostics.

Task 5: Add daily full consistency audit

Files:

• Create: app/core/knowledge/audit.py
• Create: app/commands/audit_governance_knowledge.py
• Create: tests/test_knowledge_full_audit.py

• Modify: deploy/docker/docker-compose.yml

• [ ] Enumerate source objects with pagination and compare UID/version/hash against indexed documents.

• [ ] Record missing, stale, orphaned and dimension-mismatch findings before repair.

• [ ] Repair only differences and produce a persisted audit summary; never delete source governance data.

• [ ] Schedule once daily in deployment orchestration and add a manual local command for acceptance.

Task 6: Implement permission-aware retrieval and DeepSeek answers

Files:

• Create: app/core/knowledge/retrieval.py
• Create: app/core/knowledge/qa.py

• Create: tests/test_knowledge_qa.py

• [ ] Filter candidate chunks by user role/business-domain scope before vector similarity ranking.

• [ ] Add lexical fallback and return no-answer when evidence is insufficient.

• [ ] Send DeepSeek only selected chunks plus question; require answers to cite object UID, name and version.

• [ ] Test prompt injection in source text, cross-domain data leakage, stale versions, model timeout and unsupported questions.

Task 7: Add APIs and retained frontend page

Files:

• Create: app/api/knowledge_base/
• Modify: app/__init__.py
• Modify: frontend/src/views/knowledgeBaseProduct/index.vue
• Create: frontend/src/api/governanceKnowledge.js
• Create: tests/test_knowledge_api.py

• Create: tests/e2e/governance-knowledge.spec.js

• [ ] Rename all user-facing copy to“数据治理知识库”, not product knowledge base.

• [ ] Add search, ask, source detail, sync status and administrator audit endpoints.

• [ ] Render source citations and object versions; make degraded embedding/model states explicit.

• [ ] Verify administrator/editor/viewer access scopes and prevent direct source-detail bypass.

Task 8: Run end-to-end acceptance

Files:

• Modify: docs/architecture/OPENAPI.yaml
• Modify: docs/architecture/DATA_MODEL.md

• Modify: deploy/docker/README.md

• [ ] Seed a small governance graph, trigger an incremental change, search it, ask a sourced question and confirm role filtering.

• [ ] Deliberately remove one chunk, run the daily full audit and verify it detects and repairs the difference.

• [ ] Record embedding latency, retrieval quality fixtures, token usage and failure behavior.

• [ ] Run the complete local Docker, Python, frontend and browser acceptance suite.