{"id":"cea1209c-cf84-489d-9bd4-756224e2c25e","slug":"poknowledgebase-etl","kind":"mcp","name":"POKnowledgeBase-ETL","summary":"Monthly automated CDC ETL pipeline using Airflow local runner, to occupy, monitor, and fix knowledge drift in Neo4j graph and Qdrant vector databases using local SQL database as central backend.","owner":"yoyo4581","repoUrl":"https://github.com/yoyo4581/POKnowledgeBase_ETL","license":"Apache-2.0","licenseTier":"permissive","trust":"community","stars":1,"match":null,"fewStars":true,"format":"mcp","createdAt":"2025-07-27T19:59:07.000Z","addedAt":"2026-10-21T15:38:14.634Z","logoUrl":"https://avatars.githubusercontent.com/u/88361925?u=f95ceb2dbbad9c5196b0a85ecb0734e1cca70f03&v=4","linkOnly":false,"collection":null,"skillPath":null,"categories":["databases","data-analytics","memory-knowledge"],"tags":[],"runtime":"binary","local":true,"cloudCapable":false,"cloudCapability":"unknown","cloudReadyOnPlatform":false,"needsApiKey":false,"reviewed":false,"syncedAt":"2026-10-26T08:49:07.433Z","sample":false,"url":"https://conduits.dev/tools/poknowledgebase-etl","language":"Python","transports":["stdio"],"install":{},"requirements":{"env":[],"upstreamMayCharge":false},"tools":null,"readme":"# POKnowledgeBase_ETL\n\n[![Python](https://img.shields.io/badge/Python-3.12-3776AB?logo=python&logoColor=white)](https://www.python.org/)\n[![Airflow](https://img.shields.io/badge/Apache%20Airflow-3.3.0-017CEE?logo=apacheairflow&logoColor=white)](https://airflow.apache.org/)\n[![Neo4j](https://img.shields.io/badge/Neo4j-5.19%20Community-4581C3?logo=neo4j&logoColor=white)](https://neo4j.com/)\n[![Qdrant](https://img.shields.io/badge/Qdrant-1.19.1-DC244C?logo=qdrant&logoColor=white)](https://qdrant.tech/)\n[![Docker](https://img.shields.io/badge/Docker-Compose-2496ED?logo=docker&logoColor=white)](https://docs.docker.com/compose/)\n[![License](https://img.shields.io/badge/License-Apache%202.0-D22128?logo=apache&logoColor=white)](LICENSE)\n[![nbviewer](https://img.shields.io/badge/notebook-nbviewer-F37726?logo=jupyter&logoColor=white)](https://nbviewer.org/github/yoyo4581/POKnowledgeBase_ETL/blob/main/Finetune_BioBERT_Colab.ipynb)\n\nThe knowledge-base ETL behind **PathwayOracle**. It builds and maintains a Neo4j\nknowledge graph from Reactome, Gene Ontology and UniProt, plus a Qdrant vector\nstore of protein-function text, and keeps both in step with their upstream\nsources using change-data-capture.\n\nThe design question it answers is not \"how do I load this data once\" but **\"how\ndo I tell what changed since last time, and prove it.\"**\n\n---\n\n## Run Current Database Builds\n\n\n```bash\ncurl -sLO https://raw.githubusercontent.com/yoyo4581/POKnowledgeBase_ETL/main/compose.serve.yaml\ndocker compose -f compose.serve.yaml up -d\n```\n\nFirst start pulls ~440 MB of snapshots and restores them; it takes a few\nminutes. After that:\n\n| | |\n|---|---|\n| Qdrant | — REST, and `/dashboard` in a browser |\n| Neo4j Browser | — no login |\n| Neo4j Bolt | `bolt://localhost:7687` |\n\nWhat you get: **122,347 nodes / 602,142 relationships** in the graph, and\n**10,204 protein records / 42,714 chunks** in the vector store.\n\n
\nWhat it is doing\n\nFive services, three of which exit once their job is done:\n\n| service | role |\n|---|---|\n| `fetch` | downloads the release assets, verifies `SHA256SUMS` |\n| `qdrant` / `neo4j` | the databases |\n| `qdrant-load` | uploads both snapshots, skipping collections already present |\n| `neo4j-load` | restores the dump **before** the server starts — `neo4j-admin` needs the store to itself, and Community has no online restore |\n\nOrdering is `depends_on: { condition: service_completed_successfully }`, so\n`up` blocks until the data is in.\n\nEverything is idempotent: a second `up` prints `have SHA256SUMS`, `already\npresent`, `graph already loaded` and finishes in seconds. Both ports bind to\n`127.0.0.1` only, which is the reason Neo4j ships with `NEO4J_AUTH: none` —\nnothing off the machine can reach it. Override `QDRANT_REST_PORT`,\n`NEO4J_HTTP_PORT` or `NEO4J_BOLT_PORT` if something already owns them.\n\n
\n\n> [!TIP]\n> `docker compose -f compose.serve.yaml down` stops everything and keeps the\n> data. Only `down -v` discards it — which is also how you upgrade to a newer\n> release.\n\nThis is the serving path. To **rebuild** the knowledge base from Reactome, GO\nand UniProt, you need the repo, Airflow and SQL Server — see\n[Quickstart](#quickstart).\n\n---\n\n## Contents\n\n- [Run it](#run-it) — serving stack, one command\n- [Overview](#overview)\n- [Quickstart](#quickstart) — building it yourself\n- [Exploring the data](#exploring-the-data)\n- [Architecture](#architecture)\n- [BioBERT fine-tuning](#biobert-fine-tuning)\n- [Repository layout](#repository-layout)\n- [Known gaps](#known-gaps)\n\n---\n\n## Overview\n\nUpstream biology sources are not static. Reactome re-releases quarterly, GO\nupdates continuously, UniProt re-annotates. A graph rebuilt from scratch each\ntime tells you nothing about *what moved*; a graph updated in place drifts\nsilently away from its sources. This repo treats every load as a diff.\n\n```mermaid\nflowchart LR\n subgraph SRC[\"upstream sources\"]\n RX[\"Reactome
SBML + ContentService\"]\n GO[\"Gene Ontology
OBO + GOA\"]\n UP[\"UniProt
function text\"]\n end\n\n subgraph CDC[\"SQL Server — change data capture\"]\n ST[\"staging
this run's rows\"]\n DF[\"diff
INSERT / UPDATE / DELETE\"]\n DB[\"dbo
current truth\"]\n end\n\n subgraph SERVE[\"serving layer\"]\n N4[\"Neo4j
knowledge graph\"]\n QD[\"Qdrant
function vectors\"]\n end\n\n RX --> ST\n GO --> ST\n UP --> ST\n ST -- \"MERGE\" --> DB\n ST -. \"OUTPUT clause\" .-> DF\n DF -- \"replay only what changed\" --> N4\n DB --> QD\n N4 --> MCP[\"PathwayOracle
MCP server\"]\n QD --> MCP\n\n classDef src fill:#e8f0fe,stroke:#4285f4\n classDef cdc fill:#fef7e0,stroke:#f9ab00\n classDef srv fill:#e6f4ea,stroke:#34a853\n class RX,GO,UP src\n class ST,DF,DB cdc\n class N4,QD,MCP srv\n```\n\n### Why CDC, concretely\n\nEvery table is loaded through the same three-schema cycle. Rows for the current\nrun land in `staging`. A single `MERGE` reconciles them into `dbo`, and SQL\nServer's `OUTPUT` clause captures what that merge actually did into `diff`,\ntagged with the action and the run id. The graph loaders then read `diff`, not\n`dbo` — so a run that changes nothing writes nothing to Neo4j.\n\nWhat makes the diff trustworthy is how a row's identity is defined:\n\n| Sync strategy | Identity | Deletes | Used by |\n|---|---|---|---|\n| `DiffSync` | the table's primary key | no | keyed annotation tables (`GeneData`, `reactions`, …) |\n| `IdentityHashSync` | `HASHBYTES` over named columns | yes, **coverage-scoped** | every edge table (`entity_membership`, `gene_edges`, …) |\n| `CompositeKeySync` | a declared composite key | yes, coverage-scoped | multi-column keyed tables |\n| `DefaultSync` | none — append only | no | `FunctionData` |\n\n\"Coverage-scoped\" is the part that makes drift detectable. An edge table\ndeclares `coverage_scope_columns=(\"pathway_id\",)`. When the merge runs, any row\nin `dbo` whose `pathway_id` appears in staging but whose identity hash does\n*not* is deleted and recorded as a `DELETE` in the diff. That is how an edge\nReactome *removed* propagates — without it, retractions would be invisible and\nthe graph would only ever grow.\n\n> [!NOTE]\n> The scope column is what bounds the delete. Without it a partial load would\n> delete everything absent from this batch; with it, only the pathways actually\n> re-derived are in scope.\n\n### Graph at a glance\n\nLive figures from the build of **2026-10-01**, `pathway_source=reactome`:\n\n| | count |\n|---|---:|\n| Nodes | 84,255 |\n| Relationships (structural) | 193,078 |\n| Pathways ingested | 2,035 (leaf pathways of 2,883) |\n| Reactions | 13,652 |\n| Genes (UniProt-keyed) | 10,561 |\n| SQL tables under CDC | 19 |\n| Airflow DAGs | 16 |\n\n\n---\n\n## Quickstart\n\nThis is the **build** path — the one that regenerates the knowledge base from\nsource. If you only want to query it, use [Run it](#run-it) instead: one file,\none command, no clone.\n\n`docker-compose.yml` is not the same file as `compose.serve.yaml`. This one\nbuilds databases and produces snapshots, so it bind-mounts repo directories\nand carries the dump/load one-shots. The ETL itself does **not** run in\nDocker: Airflow and SQL Server run on the host.\n\n### Prerequisites\n\n| | |\n|---|---|\n| Docker Compose | v2 |\n| Disk | ~1 GB for the graph volume, ~100 MB for Qdrant storage |\n| Memory | Neo4j is configured for **2 GB heap + 1 GB pagecache**, so give Docker ≥ 4 GB |\n\n### 1. Configure\n\n```bash\ncp .env.example .env\n```\n\nBoth containers ship with **no Neo4j password** (`NEO4J_AUTH: none`), which is\nsafe only because every port binds to `127.0.0.1`. Leave it alone unless you\nwiden the ports — and if you do, set it *before* the first start.\n\n> [!IMPORTANT]\n> `NEO4J_AUTH` is read **only when the data volume is empty**. After the first\n> start, auth is whatever it was then — including after loading a dump, since a\n> dump covers the `neo4j` database and not `system`. Changing it later takes\n> `neo4j-admin dbms set-initial-password` or `down -v`.\n\nOverride the ports in `.env` if something already owns them. A local Neo4j\nDesktop install holds **7474 and 7687**, and a second Desktop instance takes\n7688 — in which case pick free ones and point `NEO4J_CONTAINER_URI` at the\nsame place:\n\n```bash\nNEO4J_HTTP_PORT=\"7475\"\nNEO4J_BOLT_PORT=\"7689\"\nNEO4J_CONTAINER_URI=\"bolt://localhost:7689\"\n```\n\n### 2. Start the services\n\n```bash\ndocker compose up -d qdrant neo4j\ncurl localhost:6333/readyz # Qdrant\ncurl -s localhost:7474 >/dev/null && echo \"neo4j up\"\n```\n\n| Service | Image | Ports | Volume |\n|---|---|---|---|\n| `qdrant` | `qdrant/qdrant:v1.19.1` | `6333` REST, `6334` gRPC | `qdrant_storage:/qdrant/storage` |\n| `neo4j` | `neo4j:5.19.0-community` | `7474` HTTP/Browser, `7687` Bolt | `neo4j_data:/data`, `neo4j_logs:/logs`, `./data/neo4j/dumps:/dumps` |\n\nAll ports bind to `127.0.0.1`, not `0.0.0.0` — a bare `\"6333:6333\"` publishes\nto every interface, which with no API key means an unauthenticated store\nanyone routable can read and write.\n\nBoth declare healthchecks (`interval 10s`, `retries 12`); Neo4j allows a 40 s\nstart period. Qdrant probes `/readyz` rather than `/healthz` deliberately —\n`healthz` answers as soon as the process is alive, `readyz` waits until\ncollections are loadable, which is the condition a restore needs.\n\nNeo4j loads the **APOC** plugin (`NEO4J_PLUGINS: '[\"apoc\"]'`). The ETL's\nontology annotation uses `apoc.periodic.iterate`, so writes require it; serving\nreads do not.\n\n### 3. Load a graph dump\n\nPut `neo4j.dump` in `data/neo4j/dumps/`, then:\n\n```bash\ndocker compose stop neo4j\ndocker compose --profile load run --rm neo4j-load\ndocker compose start neo4j\n```\n\nThe loader is behind a Compose **profile** so a plain `up` can never re-import\nand undo whatever the server has done since. The server must be stopped — both\nmount `/data`, and `neo4j-admin` needs the store to itself. Dumping back out is\nsymmetric, with `--profile dump`.\n\n> [!CAUTION]\n> The graph lives in the `neo4j_data` named volume, not in the dump. It survives\n> `stop`, `start`, `restart`, `down` and reboots. **`docker compose down -v`\n> destroys it.** That is the command never to type by reflex.\n\n### Running the ETL\n\nAirflow and SQL Server are host services. Required `.env` keys:\n\n| Key | Purpose |\n|---|---|\n| `pathway_source` | `reactome` (default) or `kegg` — selects the schema and DAG lineage |\n| `sql_server`, `gene_database`, `sql_uid`, `sql_pwd` | SQL Server connection (needs ODBC Driver 18) |\n| `uuid` | run id stamped onto every staged and diffed row |\n| `NEO4J_URI`, `NEO4J_USERNAME`, `NEO4J_PASSWORD` | graph target, e.g. `bolt://localhost:7687` |\n| `GITHUB_TOKEN`, `GITHUB_DATASET_REPO` | publishing the training dataset as a release asset |\n| `QDRANT_URL` *or* `QDRANT_PATH` | server mode vs embedded folder |\n| `BIOBERT_MODEL_PATH` | encoder for the vector store |\n\n```bash\npip install -r requirements.txt # or requirements.lock.txt for exact pins\nairflow dags trigger reactome_meta_build\n```\n\nOnly `reactome_meta_build` is triggered by hand; everything downstream fires on\nassets. See [the DAG graph](#dag-graph).\n\n> [!TIP]\n> `pathway_source` and `gene_database` must agree. Six table names exist in both\n> the KEGG and Reactome schemas with different columns, so pointing Reactome\n> code at a KEGG database fails at upsert, not at import.\n\n---\n\n## Exploring the data\n\n### Neo4j Browser\n\nOpen , connect over `bolt://localhost:7687`.\n\nEvery node the ETL writes carries a shared **`:Entity`** label alongside its\nclass label (`:Protein`, `:Complex`, `:DefinedSet`, `:SmallMolecule`,\n`:Gene`, `:Reaction`, `:Pathway`, …), because an edge endpoint can be any of a\ndozen classes and the edge row does not say which.\n\nA protein's identity and its states are deliberately separate nodes — `WEE1` and\n`p-WEE1` are two `:Protein` entities, both `IS_FORM_OF` the one `:Gene`:\n\n```cypher\n// TP53's physical states across the graph\nMATCH (e:Entity)-[:IS_FORM_OF]->(g:Gene {id: 'P04637'})\nRETURN e.id, e.name, e.compartment\nLIMIT 25;\n```\n\n```cypher\n// Which complexes is TP53 a component of? -> 106\nMATCH (:Gene {id: 'P04637'})<-[:IS_FORM_OF]-(e:Entity)\n -[:HAS_COMPONENT*1..3]->(c:Complex)\nRETURN count(DISTINCT c) AS complexes;\n```\n\n```cypher\n// Read a reaction in one traversal: reactants in, products out\nMATCH (r:Reaction {id: 'R-HSA-6804879'})\nOPTIONAL MATCH (sub:Entity)-[:REACTANT]->(r)\nOPTIONAL MATCH (r)-[:PRODUCT]->(prod:Entity)\nOPTIONAL MATCH (cat:Entity)-[:CATALYST]->(r)\nRETURN r.name, collect(DISTINCT sub.name) AS reactants,\n collect(DISTINCT prod.name) AS products,\n collect(DISTINCT cat.name) AS catalysts;\n```\n\n```cypher\n// Pathway membership, by size\nMATCH (e:Entity)-[:IN_PATHWAY]->(p:Pathway)\nRETURN p.name, count(e) AS entities\nORDER BY entities DESC LIMIT 10;\n```\n\nRelationship types currently in the graph:\n\n| Type | Count | Meaning |\n|---|---:|---|\n| `IN_PATHWAY` | 36,085 | entity → pathway |\n| `HAS_COMPONENT` | 35,054 | complex/polymer → part |\n| `IS_FORM_OF` | 34,069 | physical state → gene/compound identity |\n| `REACTANT` | 26,395 | entity → reaction |\n| `HAS_MEMBER` | 23,388 | defined set → member |\n| `PRODUCT` | 21,022 | reaction → entity |\n| `HAS_CANDIDATE` | 7,960 | candidate set → member |\n| `CATALYST` | 5,803 | enzyme → reaction |\n| `STIMULATOR` | 1,907 | positive regulator → reaction |\n| `INHIBITOR` | 894 | negative regulator → reaction |\n| `HAS_MOIETY` | 501 | entity → attached moiety |\n\nLayer-2 gene–gene edges (`ACTS_ON`, `ACTS_ON_CHEMICAL`, `ASSOCIATED_WITH`) are\nderived from the above and are currently being rebuilt — see\n[Known gaps](#known-gaps).\n\n### Qdrant\n\nDashboard: . Two collections:\n\n| Collection | Granularity | Point id |\n|---|---|---|\n| `function_records` | one point per UniProt accession | `uuid5(NAMESPACE_URL, uniprot_id)` |\n| `function_chunks` | sliding windows over the function text | `uuid5` of accession + chunk index |\n\nBoth carry a **dense** vector (768-d, cosine, from the fine-tuned BioBERT) and a\n**sparse** BM25 vector (`Qdrant/bm25`, IDF modifier, `k=1.2`, `b=0.75`). Sparse\nvectors are re-derived from payload text on restore, since BM25 term frequency\nis corpus-independent.\n\n```bash\ncurl -s localhost:6333/collections | jq '.result.collections[].name'\ncurl -s localhost:6333/collections/function_records | jq '.result.points_count'\n```\n\n### Through the MCP server\n\n`src/builders/Qdrant/mcp_server.py` exposes the store as an MCP toolkit shaped\naround how an agent actually works a question — a funnel, not a single lookup:\n\n| Tool | Use |\n|---|---|\n| `search_proteins` | shortlist accessions, each with its best snippet as evidence |\n| `read_snippets` | more windows from one protein, re-scored against a *different* question |\n| `expand_snippet` | widen a promising window outward in sentences |\n| `read_record` | give up on windows, read the whole function text |\n| `collection_info` | what is actually indexed |\n\n```jsonc\n{\"mcpServers\": {\"protein-retrieval\": {\n \"command\": \"python\",\n \"args\": [\"src/builders/Qdrant/mcp_server.py\"],\n \"env\": {\"QDRANT_URL\": \"http://localhost:6333\",\n \"BIOBERT_MODEL_PATH\": \"EmbeddingModel/biobert-go-retrieval\"}}}}\n```\n\nEvery tool speaks UniProt accessions, so a result can be fed straight into the\nnext call.\n\n> [!NOTE]\n> Embedded mode (`QDRANT_PATH`) takes an exclusive lock on the storage folder —\n> one process at a time. Point `QDRANT_URL` at the container for concurrent use.\n\n---\n\n## Architecture\n\n### The sync core\n\nTwo orthogonal strategy hierarchies. A `SyncStrategy` decides **what a row's\nidentity is** and what the diff should say; a `DbOpsStrategy` decides **what SQL\nto emit**. `TableManager` binds one of each to a `TableSchema`.\n\n```mermaid\nclassDiagram\n class TableSchema {\n +Mapping columns\n +Optional key\n +tuple~Constraint~ constraints\n +bool auto_id\n +sync\n }\n class SyncStrategy {\n <>\n +match_clause(schema, target_table)* str\n +diff_output_clause(schema, target_table, run_id)* str\n +key_clause(schema) str\n +dedup_match_columns(schema) tuple\n }\n class DbOpsStrategy {\n <>\n +stage_data(schema)* StagingQuery\n +upsert_data(schema, run_id)* list~str~\n +build_merge_query(schema, run_id) str\n +compute_identity_hashes(schema, sync) str\n +create_table(schema, kind) str\n }\n class TableManager {\n +TableSchema schema\n +DbOpsStrategy strategy\n +create_production_environment() list~str~\n +create_staging_environment() list~str~\n +wipe_staging_environment() list~str~\n }\n\n SyncStrategy <|-- DiffSync\n SyncStrategy <|-- IdentityHashSync\n SyncStrategy <|-- CompositeKeySync\n SyncStrategy <|-- DefaultSync\n DbOpsStrategy <|-- SQLStageUpsert\n DbOpsStrategy <|-- SQLStageUpsertDiff\n DbOpsStrategy <|-- SQLStageUpsertConstraints\n TableSchema --> SyncStrategy : sync\n TableManager --> TableSchema : schema\n TableManager --> DbOpsStrategy : strategy\n```\n\nStrategy selection is automatic, in `table_registry._select_strategy`:\nconstraints win over sync type, because `SQLStageUpsertConstraints` degrades\ngracefully for tables with plain foreign keys.\n\n
\nFull UML — generated with pyreverse -o mmd\n\nRegenerate with:\n\n```bash\npyreverse -o mmd -p SQLCore -d docs/assets \\\n src/builders/SQL/schema/types.py \\\n src/builders/SQL/schema/strategies \\\n src/builders/SQL/schema/table_registry.py\n```\n\n```mermaid\nclassDiagram\n class StagingQuery {\n sql : str\n insert_cols : tuple~str~\n match_cols : tuple~str~\n num_params : int\n }\n class SyncStrategy {\n dedup_match_columns(schema) tuple~str~\n diff_output_clause(schema, target_table, run_id)* str\n insert_columns(schema) list~str~\n insert_values(schema) list~str~\n key_clause(schema) str\n match_clause(schema, target_table)* str\n }\n class IdentityHashSync {\n coverage_scope_columns : tuple~str~\n identity_hash : tuple~str~\n dedup_match_columns(schema) tuple~str~\n diff_output_clause(schema, target_table, run_id) str\n insert_columns(schema) list~str~\n insert_values(schema) list~str~\n key_clause(schema) str\n match_clause(schema, target_table) str\n }\n class CompositeKeySync {\n coverage_scope_columns : tuple~str~\n dedup_match_columns(schema) tuple~str~\n diff_output_clause(schema, target_table, run_id) str\n key_clause(schema) str\n match_clause(schema, target_table) str\n }\n class DiffSync {\n diff_columns : tuple~str~\n diff_output_clause(schema, target_table, run_id) str\n match_clause(schema, target_table) str\n }\n class DefaultSync {\n diff_output_clause(schema, target_table, run_id) str\n match_clause(schema, target_table) str\n }\n class DbOpsStrategy {\n build_merge_query(schema, run_id) str\n compute_identity_hashes(schema, sync) str\n create_indexes(schema, kind) list~str~\n create_table(schema, kind) str\n stage_data(schema)* StagingQuery\n upsert_data(schema, run_id)* list~str~\n wipe_data(schema, kind) str\n }\n class SQLStageUpsert {\n stage_data(schema) StagingQuery\n upsert_data(schema, run_id) list~str~\n }\n class SQLStageUpsertDiff {\n stage_data(schema) StagingQuery\n upsert_data(schema, run_id) list~str~\n }\n class SQLStageUpsertConstraints {\n stage_data(schema) StagingQuery\n upsert_data(schema, run_id) list~str~\n }\n class TableManager {\n schema : TableSchema\n strategy : DbOpsStrategy\n create_diff_table() str\n create_production_environment() list~str~\n create_staging_environment() list~str~\n create_staging_table() str\n wipe_staging_environment() list~str~\n }\n class TableSchema {\n auto_id : bool\n columns : Mapping\n constraints : tuple~Constraint~\n key : Optional\n sync\n }\n class ForeignKey {\n name : str\n columns : tuple~str~\n ref_table : str\n ref_columns : tuple~str~\n deferred : Optional~DeferredResolution~\n on_delete : Optional~str~\n to_sql() str\n }\n class DeferredResolution {\n match_columns : tuple~str~\n staging_columns : tuple~str~\n }\n class UniqueConstraint {\n name : str\n columns : tuple~str~\n to_sql() str\n }\n class CheckConstraint {\n name : str\n expression : str\n to_sql() str\n }\n class PrimaryCompositeKey {\n columns : tuple~str~\n }\n class ColumnPlan {\n production_col : str\n staging_col : str\n staging_type : str\n reused_column : bool\n deffered_fk : Optional~ForeignKey~\n }\n class TableClearOrder {\n order : list~str~\n levels : list\n cycles : list~set~\n self_refs : set~str~\n }\n class DBSchema {\n <>\n PRODUCTION\n STAGING\n DIFF\n }\n\n SyncStrategy <|-- IdentityHashSync\n SyncStrategy <|-- CompositeKeySync\n SyncStrategy <|-- DiffSync\n SyncStrategy <|-- DefaultSync\n DbOpsStrategy <|-- SQLStageUpsert\n DbOpsStrategy <|-- SQLStageUpsertDiff\n DbOpsStrategy <|-- SQLStageUpsertConstraints\n TableSchema --> SyncStrategy : sync\n TableSchema --> PrimaryCompositeKey : key\n TableSchema --> ForeignKey : constraints\n ForeignKey --> DeferredResolution : deferred\n TableManager --> TableSchema : schema\n TableManager --> DbOpsStrategy : strategy\n DbOpsStrategy --> StagingQuery : returns\n DbOpsStrategy --> DBSchema : kind\n DbOpsStrategy --> ColumnPlan : plans\n```\n\n
\n\n### Special case: deferred FK resolution\n\nSelf-referencing hierarchy tables cannot be staged with their foreign key\nalready resolved, because the parent's surrogate id does not exist until the\nparent row is inserted. `DeferredResolution` declares the natural key to match\non instead, and `SQLStageUpsertConstraints` rewrites the `USING` clause to join\nstaging against production on that natural key at merge time:\n\n```python\nForeignKey(\n name=\"fk_parent\",\n columns=(\"parent_id\",),\n ref_table=\"kegg_class\",\n ref_columns=(\"class_id\",),\n deferred=DeferredResolution(\n match_columns=(\"name\",), # production column to match\n staging_columns=(\"parent_name\",), # staging column holding the value\n ),\n)\n```\n\nThe Reactome port sidesteps this: `pathway_class` is keyed on Reactome's own\n`stId` rather than a surrogate id, so there is nothing to defer. Reactome event\nnames are not unique (≈19k events share ≈18.8k names), so a name match would\nbind some children to the wrong parent.\n\n### Table inventory\n\n19 tables under CDC with `pathway_source=reactome`:\n\n| Table | Sync | Executor | Rows |\n|---|---|---|---:|\n| `entities` | `DiffSync` | `SQLStageUpsertDiff` | 84,253 |\n| `gene_xref` | `IdentityHashSync` | `SQLStageUpsertDiff` | 112,721 |\n| `entity_membership` | `IdentityHashSync` | `SQLStageUpsertDiff` | 109,654 |\n| `reaction_participants` | `IdentityHashSync` | `SQLStageUpsertDiff` | 58,575 |\n| `EntityData` | `DiffSync` | `SQLStageUpsertDiff` | 55,214 |\n| `EntityPathMem` | `IdentityHashSync` | `SQLStageUpsertDiff` | 36,085 |\n| `entity_identity` | `IdentityHashSync` | `SQLStageUpsertDiff` | 34,069 |\n| `reactions` | `DiffSync` | `SQLStageUpsertDiff` | 13,652 |\n| `GeneData` | `DiffSync` | `SQLStageUpsertDiff` | 10,561 |\n| `pathway_class` | `IdentityHashSync` | `SQLStageUpsertDiff` | 2,883 |\n| `PathwayIds` | `DiffSync` | `SQLStageUpsertDiff` | 2,035 |\n| `PathwayData` | `DiffSync` | **`SQLStageUpsertConstraints`** | 2,035 |\n| `PathwaySBMLMeta` | `DiffSync` | `SQLStageUpsertDiff` | 2,035 |\n| `CompoundData` | `DiffSync` | `SQLStageUpsertDiff` | 1,818 |\n| `DrugData` | `DiffSync` | `SQLStageUpsertDiff` | 975 |\n| `entity_moiety` | `IdentityHashSync` | `SQLStageUpsertDiff` | 501 |\n| `gene_edges` | `IdentityHashSync` | `SQLStageUpsertDiff` | 142547 |\n| `FunctionData` | `DefaultSync` | `SQLStageUpsert` | — |\n| `GOOntologyMeta` | `IdentityHashSync` | `SQLStageUpsertDiff` | 0 |\n\nSchemas are selected, not merged: `definitions.py` picks `reactome_definitions`\nor `kegg_definitions` from `pathway_source` and merges in `shared_definitions`\n(the source-neutral spine: `PathwayIds`, `entities`, `EntityPathMem`,\n`FunctionData`, `GOOntologyMeta`). Six table names exist in both source sets\nwith different columns, which is why they replace rather than combine.\n\n### DAG graph\n\n```mermaid\nflowchart TD\n MB[\"reactome_meta_build
manual trigger\"]\n ST[\"reactome_structure\"]\n GE[\"reactome_gene_edges\"]\n OA[\"reactome_ontology_annotation\"]\n GON[\"go_ontology_network
@monthly\"]\n SNAP[\"neo4j_snapshot\"]\n EDS[\"embedding_dataset_export\"]\n EDC[\"embedding_drift_check\"]\n QCB[\"qdrant_collection_build
manual\"]\n FDB[\"function_data_build
manual\"]\n\n MB -- \"reactome://pathway_sbml_batched\" --> ST\n ST -- \"reactome://structure_complete\" --> GE\n GE -- \"reactome://gene_edges_complete\" --> OA\n GON -. \"triggered + awaited\" .-> OA\n OA -- \"neo4j://kgml_complete\" --> SNAP\n OA -- \"neo4j://kgml_complete\" --> EDS\n EDS -- \"qdrant://dataset_export\" --> EDC\n EDC -- \"qdrant://retrain_needed\" --> NB([\"Colab fine-tune\"])\n NB --> QCB\n FDB --> EDS\n\n classDef manual fill:#fef7e0,stroke:#f9ab00,stroke-width:2px\n classDef auto fill:#e8f0fe,stroke:#4285f4\n classDef ext fill:#f3e8fd,stroke:#a142f4\n class MB,QCB,FDB manual\n class ST,GE,OA,GON,SNAP,EDS,EDC auto\n class NB ext\n```\n\nOnly `reactome_meta_build` is triggered by hand; the rest fire on Airflow\nassets. The four `kegg_*` DAGs plus `go_ontology_annotation` form the legacy\nlineage — still present and runnable via `pathway_source=kegg`, but superseded.\nPause them in the UI to avoid writing KEGG-shaped rows into a Reactome database.\n\n`entrez_uniprot_annotation` serves both sources and is **not** legacy. It owns\n`dbo.EntrezUniprotMap` and pushes the opposite id form onto each Gene node:\n`uniprot_ids` under KEGG (which keys genes by Entrez id), `entrez_ids` under\nReactome (which keys them by accession). Under Reactome that id is a lookup\nqualifier rather than a join key, and it rides down to the Qdrant record\npayload so a caller holding either id form can reach the same record. About\n95% of genes carry one; the rest have no Entrez mapping and omit the field.\n\n
\nAll 16 DAGs\n\n| DAG | Schedule | Emits |\n|---|---|---|\n| `reactome_meta_build` | manual | `reactome://pathway_sbml_batched` |\n| `reactome_structure` | asset | `reactome://structure_complete` |\n| `reactome_gene_edges` | asset | `reactome://gene_edges_complete` |\n| `reactome_ontology_annotation` | asset | `neo4j://kgml_complete` |\n| `go_ontology_network` | `@monthly` | `go://structure_complete` |\n| `neo4j_snapshot` | asset | `neo4j://container_live` |\n| `embedding_dataset_export` | asset | `qdrant://dataset_export` |\n| `embedding_drift_check` | asset | `qdrant://retrain_needed` |\n| `qdrant_collection_build` | manual | `qdrant://collections_live` |\n| `function_data_build` | manual | — |\n| `wipe_sql_environment` | manual | — |\n| `kegg_meta_build` *(legacy)* | manual | `kegg://pathway_kgml_batched` |\n| `kgml_structure_annotation` *(legacy)* | asset | `kegg://structure_complete` |\n| `kgml_entity_annotation` *(legacy)* | asset | — |\n| `go_ontology_annotation` *(legacy)* | asset | `neo4j://kgml_complete` |\n| `entrez_uniprot_annotation` | manual | — |\n\n
\n\n### One CDC cycle\n\n```mermaid\nsequenceDiagram\n participant P as producer\n participant C as consumer\n participant E as SQL_ETL\n participant M as TableManager\n participant S as SQL Server\n participant N as Neo4j\n\n P->>C: stream of model objects\n C->>E: stage_data(table, rows)\n E->>M: strategy.stage_data(schema)\n M-->>E: StagingQuery\n E->>S: INSERT INTO staging.\n C->>E: upsert_data(table)\n E->>M: strategy.upsert_data(schema, run_id)\n M-->>E: [compute_identity_hashes, build_merge_query]\n E->>S: UPDATE staging SET identity_hash = HASHBYTES(...)\n E->>S: MERGE dbo.
USING staging ... OUTPUT INTO diff.
\n Note over S: coverage-scoped DELETE for rows
in scope but absent from staging\n C->>E: _clear_table_rows(staging)\n N-->>S: fetch_diff_batches(table)\n S-->>N: rows tagged INSERT / UPDATE / DELETE\n N->>N: MERGE (a)-[r]->(b) / DELETE r\n```\n\nThe staging table is cleared after every batch, so it never holds more than one\nbatch's rows. The diff accumulates across the run and is read once at the end by\nthe graph loaders.\n\n> [!WARNING]\n> A staging batch must never split a coverage scope. If half a pathway's edges\n> are staged in one batch and half in the next, the second batch's\n> coverage-scoped delete retracts what the first just wrote. Consumers therefore\n> batch by *record* (one whole pathway), never by row.\n\n---\n\n## BioBERT fine-tuning\n\nThe vector store is searched with a BioBERT encoder fine-tuned on Gene Ontology\nannotations, so that \"what does this protein do\" retrieves on function rather\nthan on lexical overlap.\n\n[![nbviewer](https://img.shields.io/badge/open-Finetune__BioBERT__Colab.ipynb-F37726?logo=jupyter&logoColor=white)](https://nbviewer.org/github/yoyo4581/POKnowledgeBase_ETL/blob/main/Finetune_BioBERT_Colab.ipynb)\n\n```mermaid\nflowchart LR\n A[\"embedding_dataset_export
Airflow\"] -->|GitHub release asset| B[\"Colab notebook\"]\n B -->|fine-tune| C[\"biobert-go-retrieval\"]\n C -->|build| D[\"function_records
function_chunks\"]\n D -->|3 zips via Drive| E[\"data/qdrant/incoming/\"]\n E --> F[\"qdrant_collection_build
Airflow\"]\n F -->|gate passes| G[\"Qdrant server\"]\n```\n\n### Supervision from the ontology\n\nTraining pairs are built from GO annotations, not from free text. An **anchor**\nis a GO term; its **positives** are the proteins annotated to it; **negatives**\nare drawn three ways, which is what stops the model collapsing onto topic\nsimilarity:\n\n| Negative source | Count | Rationale |\n|---|---:|---|\n| `neg_sibling` | 34,671 | sibling GO terms — hard negatives, close in the ontology |\n| `neg_curated_NOT` | 2,734 | curated `NOT` qualifiers — the ontology asserting non-membership |\n| `neg_random` | 1,575 | random proteins — easy negatives for stability |\n\nFiltering, from the exported `stats.json`:\n\n| | |\n|---|---:|\n| Dropped: weak evidence codes | 59,650 |\n| Dropped: indirect qualifiers | 1,601 |\n| Dropped: `NOT` conflicts | 234 |\n| Eligible GO terms | 12,153 |\n| Rejected — fewer than 3 positives | 5,335 |\n| Rejected — more than 50 positives | 305 |\n| **Anchors total** | **6,513** |\n| → train | 3,441 |\n| → held out | 651 |\n| → lost to the buffer | 2,421 |\n| **Training triplets** | **38,980** |\n| Corpus accessions | 6,019 |\n\nHeld-out GO terms are excluded entirely from training, so evaluation measures\ngeneralisation to unseen function categories rather than memorisation.\n\n### Training setup\n\n| | |\n|---|---|\n| Base model | `dmis-lab/biobert-base-cased-v1.2` |\n| Hardware | 1 × NVIDIA A100-SXM4-40GB |\n| Epochs / steps | 10 / 390 |\n| Batch size | 1,024 (`mini_batch_num_tokens=65536`) |\n| Wall clock | 1,330 s (≈22 min) |\n| Peak GPU memory | 27.52 GB |\n| Final train loss | 4.546 |\n| Best checkpoint | `checkpoint-390`, by `heldout_go_cosine_ndcg@10` |\n| Embedding dim | 768, cosine |\n\n### Results\n\nHeld-out retrieval, 651 unseen GO terms against the 6,019-accession corpus:\n\n![Held-out GO retrieval by epoch](docs/assets/heldout_metrics_by_epoch.png)\n\n| Metric | @1 | @3 | @5 | @10 |\n|---|---:|---:|---:|---:|\n| Accuracy | 0.2289 | 0.4178 | 0.4854 | 0.6329 |\n| Precision | 0.2289 | 0.2207 | 0.1985 | 0.1697 |\n| Recall | 0.0290 | 0.0856 | 0.1191 | 0.1888 |\n| nDCG@10 | | | | **0.2308** |\n| MRR@10 | | | | 0.3470 |\n| MAP@100 | | | | 0.1637 |\n\nThen the end-to-end retrieval comparison over the same 651 queries, run through\nthe MCP server — a one-stage hybrid search versus a two-stage chunk re-rank:\n\n![Retrieval arms](docs/assets/retrieval_arms.png)\n\n| Metric | `record_hybrid` | `chunk_rerank` | Δ |\n|---|---:|---:|---:|\n| accuracy@1 | 0.2826 | 0.2857 | +0.0031 |\n| ndcg@3 | 0.2519 | 0.2528 | +0.0009 |\n| ndcg@5 | 0.2471 | 0.2472 | +0.0001 |\n| ndcg@10 | **0.2526** | 0.2497 | −0.0029 |\n| recall@10 | 0.2054 | 0.1993 | −0.0062 |\n| map@100 | 0.1761 | 0.1716 | −0.0045 |\n| median latency | **244 ms** | 2,500 ms | 10× |\n\nPaired bootstrap on nDCG@10, 10,000 resamples:\n\n$$\\Delta_{\\text{ndcg@10}} = -0.0029,\\quad 95\\%\\ \\text{CI}\\ [-0.0125,\\ +0.0067],\\quad p = 0.5508$$\n\n217 queries better, 232 worse, 202 unchanged. **The interval spans zero**: this\neval set cannot distinguish the two arms, so the 10× cheaper `record_hybrid` arm\nis the one to serve.\n\n`qdrant_collection_build` gates promotion on exactly one number —\n`summary[record_hybrid][\"ndcg@10\"]` — against `QDRANT_GATE_FLOOR`:\n\n| Gate setting | Env var | Default | This build |\n|---|---|---|---|\n| Arm | `QDRANT_GATE_ARM` | `record_hybrid` | `record_hybrid` |\n| Metric | `QDRANT_GATE_METRIC` | `ndcg@10` | 0.2526 |\n| Floor | `QDRANT_GATE_FLOOR` | **0.15** | **passes** |\n\n> [!CAUTION]\n> The notebook prints *\"refuses to promote a build below QDRANT_GATE_FLOOR\n> (default 0.35) … if this is under that, do not ship it.\"* That message is\n> **stale**: [`dags/qdrant_build.py:94`](dags/qdrant_build.py) reads\n> `float(os.getenv(\"QDRANT_GATE_FLOOR\", \"0.15\"))`. At 0.2526 the build clears\n> the real floor comfortably. Trust the DAG, not the notebook's prose.\n\nThere is **no pre-fine-tune baseline** in the notebook, so the gain attributable\nto fine-tuning is not measured — only the trajectory across epochs, which rises\nfrom nDCG@10 0.1716 (epoch 1) to 0.2305 (epoch 10).\n\n### Chunking\n\n| | |\n|---|---|\n| Corpus | 6,019 accessions → 19,883 chunks |\n| Window / stride | 2 sentences / 1 (3.3 chunks per record) |\n| Artifacts | `biobert-go-retrieval.zip` 171.5 MB, `qdrant_store.zip` 55.7 MB, `qdrant_export.zip` 38.4 MB |\n\n---\n\n## Repository layout\n\n```text\n├── dags/ 16 Airflow DAGs\n│ ├── reactome_routine.py Reactome lineage (current)\n│ ├── routine.py KEGG lineage (legacy)\n│ ├── ontology_build.py GO structure + annotation\n│ ├── embedding_dataset.py training-set export + drift check\n│ ├── qdrant_build.py collection build, gate, promote\n│ └── neo4j_snapshot.py dump → container\n├── src/\n│ ├── parsers/ source clients: Reactome, KEGG, GO, UniProt\n│ ├── models/ dataclasses on BaseSQLObject, one per table\n│ ├── builders/\n│ │ ├── SQL/ SQLCaller, SQLState, schema/ + strategies/\n│ │ ├── Neo4j/ Neo4jCaller, node/edge schema, ontology sync\n│ │ └── Qdrant/ core/, build/, analysis/, mcp_server.py\n│ └── workflow/ producers + consumers wiring parse → SQL\n├── EmbeddingModel/\n│ ├── BioBERT_Files/ train.py, dataset.py\n│ └── biobert-go-retrieval/ fine-tuned encoder\n├── notebooks/ stack_smoke_test.ipynb + scratch\n├── scripts/ one-off probes, not part of any DAG\n├── docs/assets/ generated figures\n├── data/ gitignored, but snapshots/ + dumps/ are\n│ tracked-empty so Docker does not create\n│ them as root\n├── docker-compose.yml qdrant + neo4j (+ load profiles)\n├── .env.example copy to .env\n└── Finetune_BioBERT_Colab.ipynb\n```\n\n"}