diff --git a/data/_catalog/manifests/24a05512-dc06-4884-9889-2ea978abdaec.json b/data/_catalog/manifests/24a05512-dc06-4884-9889-2ea978abdaec.json new file mode 100644 index 0000000..ebc0cd9 --- /dev/null +++ b/data/_catalog/manifests/24a05512-dc06-4884-9889-2ea978abdaec.json @@ -0,0 +1,58 @@ +{ + "id": "24a05512-dc06-4884-9889-2ea978abdaec", + "name": "processed_extractions", + "schema_fingerprint": "c2853ba4c0cc5a897b37680fefea2708f51d51774226b00b72444e79d594c874", + "objects": [ + { + "bucket": "data", + "key": "datasets/processed_extractions.parquet", + "size_bytes": 28175, + "created_at": "2026-03-28T05:16:57.437041979Z" + } + ], + "created_at": "2026-03-28T05:16:57.437042648Z", + "updated_at": "2026-03-28T05:16:57.437403737Z", + "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": "type", + "data_type": "Utf8", + "sensitivity": null, + "description": "", + "is_pii": false + }, + { + "name": "content", + "data_type": "Utf8", + "sensitivity": null, + "description": "", + "is_pii": false + } + ], + "lineage": { + "source_system": "csv", + "source_file": "processed_extractions.csv", + "ingest_job": "ingest-1774675017437", + "ingest_timestamp": "2026-03-28T05:16:57.437041979Z", + "parent_datasets": [] + }, + "freshness": null, + "tags": [], + "row_count": 543 +} \ No newline at end of file diff --git a/data/_catalog/manifests/3ead1a7a-810b-49d1-add8-752bab52dfae.json b/data/_catalog/manifests/3ead1a7a-810b-49d1-add8-752bab52dfae.json new file mode 100644 index 0000000..4ff0533 --- /dev/null +++ b/data/_catalog/manifests/3ead1a7a-810b-49d1-add8-752bab52dfae.json @@ -0,0 +1,58 @@ +{ + "id": "3ead1a7a-810b-49d1-add8-752bab52dfae", + "name": "processed_validated", + "schema_fingerprint": "8078de214059a2a4582203b32c1172fcc311c2d3c762a3b075c07268308f1841", + "objects": [ + { + "bucket": "data", + "key": "datasets/processed_validated.parquet", + "size_bytes": 15271, + "created_at": "2026-03-28T05:16:57.439472748Z" + } + ], + "created_at": "2026-03-28T05:16:57.439473552Z", + "updated_at": "2026-03-28T05:16:57.439749766Z", + "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": "type", + "data_type": "Utf8", + "sensitivity": null, + "description": "", + "is_pii": false + }, + { + "name": "content", + "data_type": "Utf8", + "sensitivity": null, + "description": "", + "is_pii": false + } + ], + "lineage": { + "source_system": "csv", + "source_file": "processed_validated.csv", + "ingest_job": "ingest-1774675017439", + "ingest_timestamp": "2026-03-28T05:16:57.439472748Z", + "parent_datasets": [] + }, + "freshness": null, + "tags": [], + "row_count": 60 +} \ No newline at end of file diff --git a/data/datasets/processed_extractions.parquet b/data/datasets/processed_extractions.parquet new file mode 100644 index 0000000..3574aef Binary files /dev/null and b/data/datasets/processed_extractions.parquet differ diff --git a/data/datasets/processed_validated.parquet b/data/datasets/processed_validated.parquet new file mode 100644 index 0000000..8e89c5e Binary files /dev/null and b/data/datasets/processed_validated.parquet differ diff --git a/scripts/process_knowledge.py b/scripts/process_knowledge.py new file mode 100644 index 0000000..2805858 --- /dev/null +++ b/scripts/process_knowledge.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +""" +Process uploaded conversations through devop.live LLM Team modes. +Runs multi-pass pipeline: extract → validate → organize. +""" + +import json, urllib.request, time, sys + +LAKEHOUSE = "http://localhost:3100" +LLM_TEAM = "http://localhost:5000" + +def sql(query): + req = urllib.request.Request(f"{LAKEHOUSE}/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 run_mode(mode, prompt, models=None): + """Run a devop.live LLM Team mode and collect all responses.""" + if models is None: + models = ["qwen2.5:latest"] + + config = {"mode": mode, "prompt": prompt, "models": models} + req = urllib.request.Request(f"{LLM_TEAM}/api/run", + data=json.dumps(config).encode(), + headers={"Content-Type": "application/json"}, method="POST") + + resp = urllib.request.urlopen(req, timeout=300) + + # Parse SSE stream + responses = [] + final = None + for line in resp: + line = line.decode().strip() + if line.startswith("data: "): + try: + event = json.loads(line[6:]) + if event.get("type") == "response": + responses.append({ + "model": event.get("model", ""), + "role": event.get("role", ""), + "text": event.get("text", ""), + }) + if event.get("role") == "final": + final = event.get("text", "") + except json.JSONDecodeError: + pass + + return {"responses": responses, "final": final} + +def ingest_csv(name, csv_text): + boundary = "----proc" + 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"{LAKEHOUSE}/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 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','knowledge_questions', + 'processed_extractions','processed_qa','processed_validated'] +ds = json.loads(urllib.request.urlopen(f"{LAKEHOUSE}/catalog/datasets").read()) +convs = [d for d in ds if d['name'] not in old and d.get('row_count',0) and d['row_count'] > 0] + +print(f"{'='*60}") +print(f"KNOWLEDGE PROCESSING PIPELINE") +print(f"Using devop.live LLM Team modes") +print(f"{'='*60}") +print(f"\n{len(convs)} conversations to process") + +all_extractions = [] +all_qa = [] +all_validated = [] + +for conv in convs: + name = conv['name'] + pages = conv.get('row_count', 0) or 0 + if pages == 0: continue + + print(f"\n{'='*60}") + print(f"Processing: {name} ({pages} pages)") + print(f"{'='*60}") + + # Sample pages (every Nth for large docs) + step = max(1, pages // 10) + r = sql(f"SELECT turn_number, content FROM {name} WHERE turn_number % {step} = 1 ORDER BY turn_number LIMIT 10") + + for row in r['rows']: + page_num = row.get('turn_number', 0) + content = (row.get('content', '') or '')[:3000] + if len(content) < 100: continue + + print(f"\n --- Page {page_num} ---") + + # PASS 1: Extract (uses devop.live extract mode) + print(f" [EXTRACT] Running extraction...", end=" ", flush=True) + try: + result = run_mode("extract", content) + final = result.get("final", "") + + # Parse the extraction response + for resp in result["responses"]: + if resp["role"].startswith("extraction"): + text = resp["text"] + # Try to find JSON in the response + if "{" in text: + start = text.find("{") + end = text.rfind("}") + 1 + if end > start: + try: + data = json.loads(text[start:end]) + facts = data.get("facts", []) + entities = data.get("entities", []) + relationships = data.get("relationships", []) + + for fact in facts[:5]: + claim = fact if isinstance(fact, str) else fact.get("claim", str(fact)) + all_extractions.append(f'"{name}",{page_num},"fact","{claim[:200].replace(chr(34), chr(39))}"') + + for ent in entities[:5]: + ent_name = ent if isinstance(ent, str) else ent.get("name", str(ent)) + all_extractions.append(f'"{name}",{page_num},"entity","{ent_name[:200].replace(chr(34), chr(39))}"') + + print(f"{len(facts)} facts, {len(entities)} entities, {len(relationships)} relationships") + except json.JSONDecodeError: + print("no JSON found") + else: + print("no structured output") + + # Check verifier response + if resp["role"] == "verifier": + verified_text = resp["text"][:500] + if verified_text.strip(): + all_validated.append(f'"{name}",{page_num},"verification","{verified_text[:200].replace(chr(34), chr(39))}"') + + except Exception as e: + print(f"error: {str(e)[:60]}") + + # Small delay between pages to not overwhelm + time.sleep(0.5) + +# Store results +print(f"\n{'='*60}") +print(f"RESULTS") +print(f"{'='*60}") +print(f"Extractions: {len(all_extractions)}") +print(f"Validated: {len(all_validated)}") + +if all_extractions: + csv = "conversation,page,type,content\n" + "\n".join(all_extractions) + print(f"\nIngesting processed_extractions...", end=" ") + print(ingest_csv("processed_extractions", csv)) + +if all_validated: + csv = "conversation,page,type,content\n" + "\n".join(all_validated) + print(f"Ingesting processed_validated...", end=" ") + print(ingest_csv("processed_validated", csv)) + +print(f"\nDone. LLM Team modes processed {len(convs)} conversations.") +print(f"Query: SELECT * FROM processed_extractions")