diff --git a/data/_catalog/manifests/667478c3-47f9-4076-bab2-8293999d4db2.json b/data/_catalog/manifests/667478c3-47f9-4076-bab2-8293999d4db2.json new file mode 100644 index 0000000..68d7e90 --- /dev/null +++ b/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 +} \ No newline at end of file diff --git a/data/_catalog/manifests/799c825c-950e-480e-a38e-4407377e43e3.json b/data/_catalog/manifests/799c825c-950e-480e-a38e-4407377e43e3.json new file mode 100644 index 0000000..7aa1852 --- /dev/null +++ b/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 +} \ No newline at end of file diff --git a/data/_catalog/manifests/80eaf246-778b-4987-9bef-62a6cccb6021.json b/data/_catalog/manifests/80eaf246-778b-4987-9bef-62a6cccb6021.json new file mode 100644 index 0000000..89c6769 --- /dev/null +++ b/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 +} \ No newline at end of file diff --git a/data/_catalog/manifests/fe6b6872-03f8-491a-b2c2-7d25b19cd9bb.json b/data/_catalog/manifests/fe6b6872-03f8-491a-b2c2-7d25b19cd9bb.json new file mode 100644 index 0000000..1b81976 --- /dev/null +++ b/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 +} \ No newline at end of file diff --git a/data/datasets/knowledge_decisions.parquet b/data/datasets/knowledge_decisions.parquet new file mode 100644 index 0000000..20ec9b8 Binary files /dev/null and b/data/datasets/knowledge_decisions.parquet differ diff --git a/data/datasets/knowledge_insights.parquet b/data/datasets/knowledge_insights.parquet new file mode 100644 index 0000000..bc35aec Binary files /dev/null and b/data/datasets/knowledge_insights.parquet differ diff --git a/data/datasets/knowledge_questions.parquet b/data/datasets/knowledge_questions.parquet new file mode 100644 index 0000000..84d3e47 Binary files /dev/null and b/data/datasets/knowledge_questions.parquet differ diff --git a/data/datasets/knowledge_topics.parquet b/data/datasets/knowledge_topics.parquet new file mode 100644 index 0000000..00e0ace Binary files /dev/null and b/data/datasets/knowledge_topics.parquet differ diff --git a/scripts/extract_knowledge.py b/scripts/extract_knowledge.py new file mode 100644 index 0000000..de66265 --- /dev/null +++ b/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")