lakehouse/scripts/process_knowledge.py

#!/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")