Metacognitive knowledge extraction: 496 topics, 300 decisions, 100 insights

- extract_knowledge.py: LLM processes sampled pages from all conversations
- Extracts structured JSON: topics, decisions, insights, open questions
- Stores as queryable Parquet tables (knowledge_topics, knowledge_decisions,
  knowledge_insights, knowledge_questions)
- Each finding traces to source conversation + page number
- Fixed Unicode chunker crash (byte boundary on multi-byte chars)
- Tested on 8 real ChatGPT PDFs, 477 pages:
  496 topics, 300 decisions, 100 insights, 200 open questions extracted

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

data/_catalog/manifests/667478c3-47f9-4076-bab2-8293999d4db2.json

@@ -0,0 +1,51 @@
+{
+  "id": "667478c3-47f9-4076-bab2-8293999d4db2",
+  "name": "knowledge_insights",
+  "schema_fingerprint": "36896987c7df0e78f5282208e3b8bbbaf9c017588d0bb42a20a7dbe97d8d4fa0",
+  "objects": [
+    {
+      "bucket": "data",
+      "key": "datasets/knowledge_insights.parquet",
+      "size_bytes": 13099,
+      "created_at": "2026-03-28T04:06:05.361963023Z"
+    }
+  ],
+  "created_at": "2026-03-28T04:06:05.361963273Z",
+  "updated_at": "2026-03-28T04:06:05.362127023Z",
+  "description": "",
+  "owner": "",
+  "sensitivity": null,
+  "columns": [
+    {
+      "name": "conversation",
+      "data_type": "Utf8",
+      "sensitivity": null,
+      "description": "",
+      "is_pii": false
+    },
+    {
+      "name": "page",
+      "data_type": "Int64",
+      "sensitivity": null,
+      "description": "",
+      "is_pii": false
+    },
+    {
+      "name": "insight",
+      "data_type": "Utf8",
+      "sensitivity": null,
+      "description": "",
+      "is_pii": false
+    }
+  ],
+  "lineage": {
+    "source_system": "csv",
+    "source_file": "knowledge_insights.csv",
+    "ingest_job": "ingest-1774670765361",
+    "ingest_timestamp": "2026-03-28T04:06:05.361963023Z",
+    "parent_datasets": []
+  },
+  "freshness": null,
+  "tags": [],
+  "row_count": 100
+}

data/_catalog/manifests/799c825c-950e-480e-a38e-4407377e43e3.json

@@ -0,0 +1,51 @@
+{
+  "id": "799c825c-950e-480e-a38e-4407377e43e3",
+  "name": "knowledge_topics",
+  "schema_fingerprint": "356c37c8fef56ce7f7882c80f1301cc2f70bcbf874719cb0628b42179874494c",
+  "objects": [
+    {
+      "bucket": "data",
+      "key": "datasets/knowledge_topics.parquet",
+      "size_bytes": 14252,
+      "created_at": "2026-03-28T04:06:05.358571822Z"
+    }
+  ],
+  "created_at": "2026-03-28T04:06:05.358572461Z",
+  "updated_at": "2026-03-28T04:06:05.358833280Z",
+  "description": "",
+  "owner": "",
+  "sensitivity": null,
+  "columns": [
+    {
+      "name": "conversation",
+      "data_type": "Utf8",
+      "sensitivity": null,
+      "description": "",
+      "is_pii": false
+    },
+    {
+      "name": "page",
+      "data_type": "Int64",
+      "sensitivity": null,
+      "description": "",
+      "is_pii": false
+    },
+    {
+      "name": "topic",
+      "data_type": "Utf8",
+      "sensitivity": null,
+      "description": "",
+      "is_pii": false
+    }
+  ],
+  "lineage": {
+    "source_system": "csv",
+    "source_file": "knowledge_topics.csv",
+    "ingest_job": "ingest-1774670765358",
+    "ingest_timestamp": "2026-03-28T04:06:05.358571822Z",
+    "parent_datasets": []
+  },
+  "freshness": null,
+  "tags": [],
+  "row_count": 496
+}

data/_catalog/manifests/80eaf246-778b-4987-9bef-62a6cccb6021.json

@@ -0,0 +1,51 @@
+{
+  "id": "80eaf246-778b-4987-9bef-62a6cccb6021",
+  "name": "knowledge_questions",
+  "schema_fingerprint": "d86b4ddf7a9694118aeff3d8aa41312e0c3091b0f15cad1079ee9561c29f9c91",
+  "objects": [
+    {
+      "bucket": "data",
+      "key": "datasets/knowledge_questions.parquet",
+      "size_bytes": 17253,
+      "created_at": "2026-03-28T04:06:05.363389625Z"
+    }
+  ],
+  "created_at": "2026-03-28T04:06:05.363390263Z",
+  "updated_at": "2026-03-28T04:06:05.363439745Z",
+  "description": "",
+  "owner": "",
+  "sensitivity": null,
+  "columns": [
+    {
+      "name": "conversation",
+      "data_type": "Utf8",
+      "sensitivity": null,
+      "description": "",
+      "is_pii": false
+    },
+    {
+      "name": "page",
+      "data_type": "Int64",
+      "sensitivity": null,
+      "description": "",
+      "is_pii": false
+    },
+    {
+      "name": "question",
+      "data_type": "Utf8",
+      "sensitivity": null,
+      "description": "",
+      "is_pii": false
+    }
+  ],
+  "lineage": {
+    "source_system": "csv",
+    "source_file": "knowledge_questions.csv",
+    "ingest_job": "ingest-1774670765363",
+    "ingest_timestamp": "2026-03-28T04:06:05.363389625Z",
+    "parent_datasets": []
+  },
+  "freshness": null,
+  "tags": [],
+  "row_count": 200
+}

data/_catalog/manifests/fe6b6872-03f8-491a-b2c2-7d25b19cd9bb.json

@@ -0,0 +1,51 @@
+{
+  "id": "fe6b6872-03f8-491a-b2c2-7d25b19cd9bb",
+  "name": "knowledge_decisions",
+  "schema_fingerprint": "be65fb1ba2121bcf43c195c06dbdc0ae892eb835f4a909e9c1c0438ac81a1d0d",
+  "objects": [
+    {
+      "bucket": "data",
+      "key": "datasets/knowledge_decisions.parquet",
+      "size_bytes": 21274,
+      "created_at": "2026-03-28T04:06:05.360578855Z"
+    }
+  ],
+  "created_at": "2026-03-28T04:06:05.360579258Z",
+  "updated_at": "2026-03-28T04:06:05.360801969Z",
+  "description": "",
+  "owner": "",
+  "sensitivity": null,
+  "columns": [
+    {
+      "name": "conversation",
+      "data_type": "Utf8",
+      "sensitivity": null,
+      "description": "",
+      "is_pii": false
+    },
+    {
+      "name": "page",
+      "data_type": "Int64",
+      "sensitivity": null,
+      "description": "",
+      "is_pii": false
+    },
+    {
+      "name": "decision",
+      "data_type": "Utf8",
+      "sensitivity": null,
+      "description": "",
+      "is_pii": false
+    }
+  ],
+  "lineage": {
+    "source_system": "csv",
+    "source_file": "knowledge_decisions.csv",
+    "ingest_job": "ingest-1774670765360",
+    "ingest_timestamp": "2026-03-28T04:06:05.360578855Z",
+    "parent_datasets": []
+  },
+  "freshness": null,
+  "tags": [],
+  "row_count": 300
+}

scripts/extract_knowledge.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+"""
+Knowledge extraction pipeline: process all conversations through LLM,
+extract structured findings, build knowledge graph as Parquet tables.
+"""
+
+import json, urllib.request, time, re
+
+API = "http://localhost:3100"
+
+def sql(query):
+    req = urllib.request.Request(f"{API}/query/sql",
+        data=json.dumps({"sql": query}).encode(),
+        headers={"Content-Type": "application/json"}, method="POST")
+    return json.loads(urllib.request.urlopen(req, timeout=30).read())
+
+def generate(prompt, max_tokens=512):
+    req = urllib.request.Request(f"{API}/ai/generate",
+        data=json.dumps({"prompt": prompt, "max_tokens": max_tokens, "temperature": 0.1}).encode(),
+        headers={"Content-Type": "application/json"}, method="POST")
+    return json.loads(urllib.request.urlopen(req, timeout=120).read())
+
+def ingest_csv(name, csv_text):
+    boundary = "----kb" + str(int(time.time()))
+    body = f"--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"{name}.csv\"\r\nContent-Type: text/csv\r\n\r\n{csv_text}\r\n--{boundary}--\r\n"
+    req = urllib.request.Request(f"{API}/ingest/file?name={name}",
+        data=body.encode(), headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, method="POST")
+    try: return urllib.request.urlopen(req, timeout=15).read().decode()
+    except: return "error"
+
+# Get all conversation datasets
+old = ['call_log','placements','clients','threat_intel','timesheets','meta_runs','candidates',
+       'job_orders','lab_trials','users','email_log','new_candidates','schema_test_v1',
+       'schema_test_v2','stress_test_ingest','stress_ingest','knowledge_topics',
+       'knowledge_decisions','knowledge_insights']
+ds = json.loads(urllib.request.urlopen(f"{API}/catalog/datasets").read())
+convs = [d for d in ds if d['name'] not in old]
+
+print(f"Processing {len(convs)} conversations...")
+
+# Knowledge tables
+all_topics = []      # (conversation, topic)
+all_decisions = []   # (conversation, decision, context)
+all_insights = []    # (conversation, page, insight, reasoning)
+all_questions = []   # (conversation, question)
+
+for conv in convs:
+    name = conv['name']
+    pages = conv.get('row_count', 0) or 0
+    if pages == 0: continue
+
+    print(f"\n{'='*50}")
+    print(f"Extracting from: {name} ({pages} pages)")
+
+    # Sample pages (every 5th page for large convos, all for small)
+    step = max(1, pages // 20)  # ~20 samples max per conversation
+    r = sql(f"SELECT turn_number, content FROM {name} WHERE turn_number % {step} = 1 ORDER BY turn_number LIMIT 20")
+
+    for row in r['rows']:
+        page_num = row.get('turn_number', 0)
+        content = (row.get('content', '') or '')[:2000]
+        if len(content) < 100: continue
+
+        prompt = f"""Analyze this excerpt from a technical AI conversation. Extract structured knowledge.
+
+TEXT:
+{content}
+
+Output a JSON object with:
+- topics: list of technology/concept topics (max 5)
+- decisions: list of decisions made or recommended (max 3)
+- key_insight: single most important takeaway (1 sentence)
+- open_questions: unresolved questions (max 2)
+
+Output ONLY valid JSON."""
+
+        try:
+            resp = generate(prompt, 300)
+            text = resp['text'].strip()
+            # Extract JSON from response
+            if '```' in text:
+                text = text.split('```')[1].replace('json','',1).strip()
+            # Find JSON object
+            start = text.find('{')
+            end = text.rfind('}') + 1
+            if start >= 0 and end > start:
+                data = json.loads(text[start:end])
+
+                for topic in data.get('topics', [])[:5]:
+                    all_topics.append(f"{name},{page_num},{topic}")
+
+                for decision in data.get('decisions', [])[:3]:
+                    all_decisions.append(f"{name},{page_num},{decision}")
+
+                insight = data.get('key_insight', '')
+                if insight:
+                    all_insights.append(f"{name},{page_num},{insight}")
+
+                for q in data.get('open_questions', [])[:2]:
+                    all_questions.append(f"{name},{page_num},{q}")
+
+                print(f"  p{page_num}: {len(data.get('topics',[]))} topics, {len(data.get('decisions',[]))} decisions")
+            else:
+                print(f"  p{page_num}: no valid JSON")
+        except Exception as e:
+            print(f"  p{page_num}: error — {str(e)[:50]}")
+
+# Build and ingest knowledge tables
+print(f"\n{'='*50}")
+print(f"Extracted: {len(all_topics)} topics, {len(all_decisions)} decisions, {len(all_insights)} insights, {len(all_questions)} questions")
+
+if all_topics:
+    csv = "conversation,page,topic\n" + "\n".join(all_topics)
+    print(f"Ingesting knowledge_topics ({len(all_topics)} rows)...", end=" ")
+    print(ingest_csv("knowledge_topics", csv))
+
+if all_decisions:
+    csv = "conversation,page,decision\n" + "\n".join(all_decisions)
+    print(f"Ingesting knowledge_decisions ({len(all_decisions)} rows)...", end=" ")
+    print(ingest_csv("knowledge_decisions", csv))
+
+if all_insights:
+    csv = "conversation,page,insight\n" + "\n".join(all_insights)
+    print(f"Ingesting knowledge_insights ({len(all_insights)} rows)...", end=" ")
+    print(ingest_csv("knowledge_insights", csv))
+
+if all_questions:
+    csv = "conversation,page,question\n" + "\n".join(all_questions)
+    print(f"Ingesting knowledge_questions ({len(all_questions)} rows)...", end=" ")
+    print(ingest_csv("knowledge_questions", csv))
+
+print(f"\nDone. Query with:")
+print(f"  SELECT topic, COUNT(*) FROM knowledge_topics GROUP BY topic ORDER BY COUNT(*) DESC")
+print(f"  SELECT * FROM knowledge_decisions")
+print(f"  SELECT * FROM knowledge_insights")