Compare commits

...
10 Commits
Author SHA1 Message Date
rootandClaude Opus 4.6 4d64e4a618 Observer KB UI: now shows LLM Team extracted facts + entities
- Knowledge tab: stats (categories, verified facts, entities, decisions, insights, questions),
  auto-generated categories, key decisions, key insights
- Facts & Entities tab (NEW): 279 extracted facts from LLM Team extract mode,
  264 entities as clickable chips (click → search), verified entries from secondary model
- Search tab: semantic HNSW search across all knowledge
- Ask AI tab: RAG with cited sources
- Ingest tab: upload with progress
- Entity chips link to semantic search
- Category drill-down shows related content via HNSW

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-28 00:25:39 -05:00
rootandClaude Opus 4.6 3f19ad5d93 LLM Team integration: extract mode processes 8 conversations
- process_knowledge.py: routes PDF content through devop.live /api/run
- Uses extract mode: LLM extracts facts/entities/relationships, verifier checks
- 543 extractions (279 facts, 264 entities) from 8 conversations
- 60 verified entries from gemma2 cross-checking
- All stored as queryable Parquet (processed_extractions, processed_validated)
- Streaming SSE parsing for LLM Team responses
- This proves the architecture: Lakehouse storage + devop.live processing

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-28 00:22:29 -05:00
rootandClaude Opus 4.6 bd68b662b7 Update PRD + memory: Observer KB needs devop.live LLM Team mode integration
Observer KB PRD v2:
- Multi-pass pipeline using devop.live's 20 analysis modes
- extract → validate → hallucination-check → consensus → organize → render
- Q&A chain extraction (questions as anchors)
- Curated document output (hand-it-to-someone quality)
- Integration architecture with LLM Team API on :5000

Memory updated with:
- Two-project structure (Lakehouse core + Observer KB)
- What's built vs what's next
- Critical: devop.live modes ARE the multi-pass pipeline
- Next session: integrate Observer KB with LLM Team modes

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-27 23:35:15 -05:00
rootandClaude Opus 4.6 3cf000ccf3 Observer KB: knowledge-first interface (not a PDF viewer)
Rebuilt from scratch around processed/enriched knowledge:

Knowledge tab (default):
- Stats: topic count, decision count, insight count, question count
- Auto-generated categories with mention counts and cross-conversation coverage
- Click category → drill into semantic matches + related decisions
- Extracted decisions (color-coded badges)
- Key insights
- Open questions (unresolved)

Explore tab:
- Semantic search across all extracted knowledge
- Results show relevance scores and source documents

Ask tab:
- RAG query: question → search → context → LLM answer with cited sources
- Full provenance chain

Ingest tab:
- Drag & drop upload with progress
- Note about running extraction pipeline for new files

This is an editorial workflow, not a file browser.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-27 23:26:05 -05:00
rootandClaude Opus 4.6 1dcbe8e04c Fix Observer KB: filter knowledge tables from sidebar, improve error handling
- Sidebar now excludes knowledge_topics/decisions/insights/questions tables
- Better error display when conversation load fails
- Prevents JSON parse errors from unfiltered dataset names

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-27 23:17:16 -05:00
rootandClaude Opus 4.6 2189a6dbee Multi-pass deep analysis workflow: dynamic knowledge discovery
5-pass analysis over extracted knowledge:
1. Corpus overview: LLM characterizes what the data is about
2. Cross-conversation patterns: recurring themes found via SQL + semantic search
3. Decision evolution: contradictions, progressive refinement, consistent principles
4. Gap analysis: recovered knowledge (questions answered elsewhere),
   real gaps (unresolved), blind spots (never discussed)
5. Synthesis: actionable recommendations based on patterns

Demonstrates: SQL (DataFusion) + Vector Search (HNSW) + Local LLM (qwen2.5)
working together for structured analytical workflow.
All local, no data leaves the machine.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-27 23:12:13 -05:00
rootandClaude Opus 4.6 bf033b2d9d 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>
2026-03-27 23:06:36 -05:00
rootandClaude Opus 4.6 0756075c6d Observer KB: full knowledge base interface
- Conversations tab: chat-style view with user/AI/page messages, page numbers
- Search tab: semantic search via HNSW, save findings to journal
- Ask AI tab: RAG query against all conversations, shows sources with scores
- Map tab: visual heatmap of conversations by size, click to open
- Upload tab: drag & drop PDFs
- Sidebar: conversation list with page counts
- Save to KB: bookmark search results and AI answers to event journal

Pure HTML/JS, no WASM. Loads instantly.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-27 22:57:19 -05:00
rootandClaude Opus 4.6 8ca4ba5717 Observer KB: 477 pages from 8 ChatGPT PDFs indexed and searchable
- Fixed Unicode crash in chunker (ligatures like fi causing byte boundary panic)
- Chunker now works with char indices, not byte indices
- 8 PDFs ingested: hybrid_ai_data_lake (133p), spark_vs_flink_iceberg (124p),
  code_sorting (70p), hybrid_indexing (69p), rust_in_tech_stack (49p), etc.
- 477 chunks embedded via supervisor (43/sec)
- HNSW index built in 0.44s
- Search results <200ms, finding correct conversations by meaning
- Tested: "S3 architecture", "CRM WordPress", "Rust vs Python" — all correct

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-27 22:53:27 -05:00
rootandClaude Opus 4.6 c49b30053b Observer KB: new app for ChatGPT/Claude PDF knowledge extraction
- New branch: observer-kb
- PRD: apps/observer-kb/docs/PRD.md
- ChatGPT PDF parser: extracts conversation turns (user/assistant)
- Upload portal: pure HTML/JS (no WASM), drag & drop, progress bars
- Browse: view ingested datasets with preview
- Search: semantic search across embedded conversations (HNSW)
- Served on :3400 via systemd (observer-kb.service)
- Uses Lakehouse core for storage, query, vectors, AI

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-27 22:23:09 -05:00
45 changed files with 2160 additions and 25 deletions
+183
View File
@@ -0,0 +1,183 @@
# PRD: Observer Knowledge Base
**Status:** Active — extraction pipeline proven, needs multi-pass processing via LLM Team modes
**Created:** 2026-03-28
**Owner:** J
**Branch:** `observer-kb`
**Parent:** Lakehouse core (crates/)
---
## Problem
AI conversations generate massive amounts of knowledge that's lost to context windows, hallucination, and scope drift. Current approach (single-pass extraction into categories) produces disorganized, hard-to-validate output. Reading raw pages is not useful — the system needs to process, clean, structure, and curate knowledge like an editorial team.
## Vision
A publishing pipeline that takes raw AI conversation exports and produces well-curated, validated knowledge documents. The output should be something you hand to a team — organized, filtered, and clearly presented. Questions anchor the content, with branching connections showing what was asked, what was answered, and how reasoning was directed.
---
## Architecture
### Multi-Pass Processing via LLM Team Modes
The key insight: devop.live already has 20 analysis modes that process content through structured workflows. The Observer KB should route uploaded data through these modes automatically.
```
Raw PDF Upload
│
▼ Mode: EXTRACT
Structured facts, entities, relationships
│
▼ Mode: VALIDATOR
Fact-check claims, score 1-10 confidence
│
▼ Mode: HALLUCINATION
Verify each claim, flag likely hallucinations
│
▼ Mode: CONSENSUS
Multiple passes iterate until extraction converges
│
▼ Mode: REDTEAM
Find flaws in knowledge, patch them
│
▼ Mode: PIPELINE (custom)
Clean → Q&A structure → topic grouping → value ranking
│
▼ RENDER
Well-curated document: questions as anchors, branching connections,
what was asked → answered → directed
```
### Integration with devop.live (LLM Team)
The Observer KB calls the LLM Team API to run modes against extracted content:
| Mode | Purpose in KB |
|---|---|
| **extract** | Structured facts, entities, relationships from raw text |
| **validator** | Fact-check extracted claims, confidence scores |
| **hallucination** | Verify each factual claim independently |
| **drift** | Process N times, find consistent vs uncertain content |
| **redteam** | Find flaws in extracted knowledge, improve |
| **consensus** | Iterate until multiple passes agree on correct extraction |
| **pipeline** | Chain: clean → extract → validate → organize → render |
| **research** | Generate follow-up questions, research in parallel |
| **blindassembly** | Split content into parts, process independently, reassemble |
### Data Flow
```
Upload Portal (Observer KB :3400)
│
├── PDF → Lakehouse ingestd (text extraction, Parquet storage)
│
▼
Processing Pipeline
│
├── Call devop.live API (:5000) with each mode
├── Each mode processes content through local Ollama models
├── Results stored back in Lakehouse (Parquet + catalog)
│
▼
Knowledge Store
│
├── knowledge_facts: verified claims with confidence scores
├── knowledge_qa: question → answer pairs with context
├── knowledge_topics: auto-generated categories
├── knowledge_connections: how topics relate across conversations
├── knowledge_contradictions: flagged conflicts between conversations
├── knowledge_validated: claims that passed hallucination check
│
▼
Rendering
│
├── Structured knowledge view (web UI)
├── Exportable curated document (PDF/markdown)
└── Question-anchored navigation (Q → A → follow-up → related)
```
---
## Processing Pipeline Detail
### Pass 1: Clean + Structure
- Strip PDF artifacts (headers, footers, page numbers)
- Identify conversation turns (user/AI)
- Fix encoding issues (ligatures, special chars)
- Normalize formatting
### Pass 2: Q&A Extraction
- Identify questions asked by user
- Map each question to its answer
- Track follow-up questions and how they branch
- Build question→answer→follow-up chains
### Pass 3: Fact Extraction + Validation
- Run EXTRACT mode: pull structured facts, entities, relationships
- Run VALIDATOR mode: score each fact 1-10 confidence
- Run HALLUCINATION mode: verify claims independently
- Flag contradictions between conversations
### Pass 4: Organize + Curate
- Run CONSENSUS mode: multiple passes converge on correct categorization
- Auto-generate topic categories from content (not predefined)
- Rank content by value (decisions > insights > discussion > noise)
- Connect related content across conversations
### Pass 5: Render
- Question-anchored view: each question is a node, answers branch from it
- Topic-grouped view: knowledge organized by auto-generated categories
- Timeline view: how thinking evolved across conversations
- Export as clean markdown/PDF
---
## Current State (as of 2026-03-28)
### What's Built
- PDF ingestion (8 PDFs, 477 pages extracted)
- Single-pass knowledge extraction (496 topics, 300 decisions, 100 insights)
- Semantic search via HNSW (27ms over 477 chunks)
- RAG (question → search → LLM answer with sources)
- 5-pass deep analysis (corpus overview, patterns, evolution, gaps, synthesis)
- Observer KB web UI (knowledge-first, not page-first)
- Upload portal with drag & drop
### What's Not Built Yet
- Integration with devop.live LLM Team modes
- Multi-pass processing pipeline (extract → validate → hallucination check)
- Q&A chain extraction (question → answer → follow-up mapping)
- Contradiction detection between conversations
- Curated document export (PDF/markdown)
- Question-anchored navigation
- Auto-processing on upload (currently manual script)
### What Needs to Happen Next Session
1. **Integrate with devop.live API** — call modes programmatically
2. **Build the multi-pass pipeline** — extract → validate → hallucinate-check → organize
3. **Q&A chain extraction** — questions as anchors, answers branching
4. **Auto-process on upload** — no manual script, pipeline runs automatically
5. **Render curated output** — readable, hand-it-to-someone quality
---
## Tech Stack
- **Storage:** Lakehouse storaged (Parquet on object storage)
- **Catalog:** Lakehouse catalogd (metadata, PII, lineage)
- **Query:** Lakehouse queryd (DataFusion SQL, hot cache, paged results)
- **Vectors:** Lakehouse vectord (HNSW search, RAG)
- **AI Processing:** devop.live LLM Team API (20 modes, multiple models)
- **AI Runtime:** Ollama (local models — qwen2.5, mistral, gemma2, nomic-embed-text)
- **UI:** Pure HTML/JS on :3400
---
## Non-Goals (for now)
- Real-time conversation ingestion (PDF exports only)
- Multi-user collaboration
- Cloud deployment
- Editing extracted knowledge from UI (read-only for now)
+286
View File
@@ -0,0 +1,286 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Observer KB</title>
<style>
:root{--bg:#0a0a0f;--s1:#12121a;--s2:#1a1a28;--b:#2a2a3a;--t:#e0e0e8;--d:#888898;--a:#6c5ce7;--ok:#00d2a0;--err:#ff6b6b;--warn:#ffa94d;--f:'JetBrains Mono','Fira Code',monospace}
*{margin:0;padding:0;box-sizing:border-box}body{font-family:var(--f);background:var(--bg);color:var(--t);font-size:13px;min-height:100vh}
.app{max-width:1200px;margin:0 auto;padding:20px}
header{display:flex;align-items:center;gap:20px;padding:16px 0;border-bottom:1px solid var(--b);margin-bottom:20px}
header h1{font-size:20px;color:var(--a)}
.tabs{display:flex;gap:2px}.tab{background:none;border:none;color:var(--d);font-family:var(--f);font-size:12px;padding:10px 16px;cursor:pointer;border-bottom:2px solid transparent}
.tab:hover{color:var(--t)}.tab.active{color:var(--a);border-bottom-color:var(--a)}
.hdr-stats{margin-left:auto;font-size:11px;color:var(--d)}
.btn{background:var(--a);color:white;border:none;padding:8px 16px;border-radius:6px;font-family:var(--f);font-size:12px;cursor:pointer;font-weight:600}
.btn:hover{opacity:.9}.btn:disabled{opacity:.5}.btn-sm{padding:5px 10px;font-size:11px}
.hidden{display:none}
.stats-bar{display:grid;grid-template-columns:repeat(auto-fill,minmax(130px,1fr));gap:8px;margin-bottom:20px}
.stat{background:var(--s1);border:1px solid var(--b);border-radius:8px;padding:12px;text-align:center}
.stat-val{font-size:22px;font-weight:700;color:var(--a)}.stat-label{font-size:10px;color:var(--d);text-transform:uppercase;margin-top:2px}
.section{margin-bottom:24px}.section h2{font-size:13px;color:var(--d);text-transform:uppercase;letter-spacing:.08em;margin-bottom:10px;border-bottom:1px solid var(--b);padding-bottom:6px}
.entry{background:var(--s1);border:1px solid var(--b);border-radius:8px;padding:14px;margin-bottom:8px}
.badge{font-size:9px;text-transform:uppercase;letter-spacing:.08em;padding:2px 8px;border-radius:3px;display:inline-block;margin-bottom:6px;font-weight:600}
.badge-fact{background:var(--a);color:white}.badge-entity{background:var(--ok);color:white}
.badge-decision{background:#e67e22;color:white}.badge-insight{background:#3498db;color:white}
.badge-question{background:var(--warn);color:#000}.badge-verified{background:var(--ok);color:white}
.entry-text{font-size:12px;line-height:1.6}.entry-src{font-size:10px;color:var(--d);margin-top:4px}
.cat-grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(250px,1fr));gap:8px;margin-bottom:16px}
.cat-card{background:var(--s1);border:1px solid var(--b);border-radius:8px;padding:14px;cursor:pointer;border-left:3px solid var(--a);transition:all .15s}
.cat-card:hover{border-color:var(--a)}.cat-title{font-weight:600;font-size:13px}.cat-count{font-size:10px;color:var(--d);margin-top:2px}
.search-row{display:flex;gap:8px;margin-bottom:16px}
.search-row input{flex:1;background:var(--s1);border:2px solid var(--b);color:var(--t);font-family:var(--f);font-size:14px;padding:12px;border-radius:8px;outline:none}
.search-row input:focus{border-color:var(--a)}
.ask-input{width:100%;background:var(--s1);border:2px solid var(--b);color:var(--t);font-family:var(--f);font-size:13px;padding:12px;border-radius:8px;resize:vertical;min-height:80px;outline:none;margin-bottom:8px}
.ask-input:focus{border-color:var(--a)}
.ai-answer{background:var(--s2);border:1px solid var(--b);border-radius:8px;padding:16px;margin-top:12px;line-height:1.7;white-space:pre-wrap}
.ai-meta{font-size:10px;color:var(--d);margin-top:6px}
.upload-zone{border:2px dashed var(--b);border-radius:12px;padding:36px;text-align:center;cursor:pointer;margin-bottom:16px}
.upload-zone:hover{border-color:var(--a);background:rgba(108,92,231,.03)}
.upload-zone h3{margin-bottom:4px}.upload-zone p{color:var(--d);font-size:11px}
.file-item{display:flex;align-items:center;justify-content:space-between;padding:8px 12px;border-bottom:1px solid var(--b);font-size:12px}
.done{color:var(--ok)}.failed{color:var(--err)}.uploading{color:var(--a)}
.result{background:var(--s2);border:1px solid var(--b);border-radius:8px;padding:12px;margin-bottom:6px;border-left:3px solid var(--a)}
.result-score{font-size:10px;color:var(--a);margin-bottom:2px}.result-text{font-size:12px;line-height:1.5}
.empty{color:var(--d);text-align:center;padding:32px}
</style>
</head>
<body>
<div class="app">
<header>
<h1>OBSERVER KB</h1>
<div class="tabs">
<button class="tab active" onclick="showTab('knowledge')">Knowledge</button>
<button class="tab" onclick="showTab('facts')">Facts &amp; Entities</button>
<button class="tab" onclick="showTab('explore')">Search</button>
<button class="tab" onclick="showTab('ask')">Ask AI</button>
<button class="tab" onclick="showTab('upload')">Ingest</button>
</div>
<div class="hdr-stats" id="hdr-stats"></div>
</header>
<div id="tab-knowledge"></div>
<div id="tab-facts" class="hidden"></div>
<div id="tab-explore" class="hidden"></div>
<div id="tab-ask" class="hidden"></div>
<div id="tab-upload" class="hidden"></div>
</div>
<script>
const API=`http://${location.hostname}:3100`;
const $=id=>document.getElementById(id);
async function q(sql){return(await(await fetch(API+'/query/sql',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({sql})})).json())}
async function searchHNSW(query,k=10){try{return(await(await fetch(API+'/vectors/hnsw/search',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({index_name:'observer_kb',query,top_k:k})})).json()).results||[]}catch{return[]}}
function showTab(name){
['knowledge','facts','explore','ask','upload'].forEach((t,i)=>{
$('tab-'+t).classList.toggle('hidden',t!==name);
document.querySelectorAll('.tab')[i].classList.toggle('active',t===name);
});
({knowledge:loadKnowledge,facts:loadFacts,explore:loadExplore,ask:loadAsk,upload:loadUpload})[name]();
}
function makeEntry(type,text,source){
const e=document.createElement('div');e.className='entry';
const b=document.createElement('span');b.className='badge badge-'+type;b.textContent=type;
const t=document.createElement('div');t.className='entry-text';t.textContent=text;
const s=document.createElement('div');s.className='entry-src';s.textContent=source;
e.append(b,t,s);return e;
}
// === KNOWLEDGE (categories + decisions + insights) ===
async function loadKnowledge(){
const el=$('tab-knowledge');el.textContent='';
try{
const[topics,decisions,insights,questions,extractions]=await Promise.all([
q("SELECT topic,COUNT(*) as n,COUNT(DISTINCT conversation) as c FROM knowledge_topics GROUP BY topic ORDER BY n DESC LIMIT 15"),
q("SELECT conversation,page,decision FROM knowledge_decisions LIMIT 12"),
q("SELECT conversation,page,insight FROM knowledge_insights LIMIT 8"),
q("SELECT COUNT(*) as n FROM knowledge_questions"),
q("SELECT type,COUNT(*) as n FROM processed_extractions GROUP BY type"),
]);
const factCount=extractions.rows.find(r=>r.type==='fact')?.n||0;
const entityCount=extractions.rows.find(r=>r.type==='entity')?.n||0;
// Stats
const sb=document.createElement('div');sb.className='stats-bar';
[{v:topics.row_count,l:'Categories'},{v:factCount,l:'Verified Facts'},{v:entityCount,l:'Entities'},{v:decisions.row_count,l:'Decisions'},{v:insights.row_count,l:'Insights'},{v:questions.rows[0]?.n||0,l:'Open Questions'}].forEach(s=>{
const d=document.createElement('div');d.className='stat';
const v=document.createElement('div');v.className='stat-val';v.textContent=s.v;
const l=document.createElement('div');l.className='stat-label';l.textContent=s.l;
d.append(v,l);sb.appendChild(d);
});
el.appendChild(sb);
$('hdr-stats').textContent=factCount+' facts · '+entityCount+' entities · '+decisions.row_count+' decisions';
// Categories
const cs=document.createElement('div');cs.className='section';
cs.insertAdjacentHTML('beforeend','<h2>Auto-Generated Categories</h2>');
const cg=document.createElement('div');cg.className='cat-grid';
topics.rows.forEach(t=>{
const c=document.createElement('div');c.className='cat-card';
const n=document.createElement('div');n.className='cat-title';n.textContent=t.topic;
const m=document.createElement('div');m.className='cat-count';m.textContent=t.n+' mentions · '+t.c+' conversations';
c.append(n,m);c.onclick=()=>drillTopic(t.topic);cg.appendChild(c);
});
cs.appendChild(cg);el.appendChild(cs);
// Decisions
const ds=document.createElement('div');ds.className='section';
ds.insertAdjacentHTML('beforeend','<h2>Key Decisions</h2>');
decisions.rows.forEach(d=>ds.appendChild(makeEntry('decision',d.decision,d.conversation+' p'+d.page)));
el.appendChild(ds);
// Insights
const is=document.createElement('div');is.className='section';
is.insertAdjacentHTML('beforeend','<h2>Key Insights</h2>');
insights.rows.forEach(d=>is.appendChild(makeEntry('insight',d.insight,d.conversation+' p'+d.page)));
el.appendChild(is);
}catch(e){el.textContent='Error: '+e}
}
async function drillTopic(topic){
const el=$('tab-knowledge');el.textContent='';
const back=document.createElement('button');back.className='btn btn-sm';back.textContent='← Back';back.onclick=loadKnowledge;
const h=document.createElement('h2');h.style.cssText='font-size:18px;margin:12px 0';h.textContent=topic;
el.append(back,h);
const results=await searchHNSW(topic,10);
const s=document.createElement('div');s.className='section';
s.insertAdjacentHTML('beforeend','<h2>Related Content</h2>');
results.forEach(r=>{
const e=document.createElement('div');e.className='result';
const sc=document.createElement('div');sc.className='result-score';sc.textContent='['+r.score.toFixed(3)+'] '+r.doc_id;
const t=document.createElement('div');t.className='result-text';t.textContent=r.chunk_text.substring(0,300);
e.append(sc,t);s.appendChild(e);
});
el.appendChild(s);
}
// === FACTS & ENTITIES (from LLM Team extract mode) ===
async function loadFacts(){
const el=$('tab-facts');el.textContent='';
try{
const[facts,entities,verified]=await Promise.all([
q("SELECT conversation,page,content FROM processed_extractions WHERE type='fact' ORDER BY conversation,page LIMIT 30"),
q("SELECT conversation,content,COUNT(*) as n FROM processed_extractions WHERE type='entity' GROUP BY conversation,content ORDER BY n DESC LIMIT 30"),
q("SELECT conversation,page,content FROM processed_validated LIMIT 10"),
]);
const fs=document.createElement('div');fs.className='section';
fs.insertAdjacentHTML('beforeend','<h2>Extracted Facts ('+facts.row_count+' total — via LLM Team Extract Mode)</h2>');
facts.rows.forEach(r=>fs.appendChild(makeEntry('fact',r.content,r.conversation+' p'+r.page)));
el.appendChild(fs);
const es=document.createElement('div');es.className='section';
es.insertAdjacentHTML('beforeend','<h2>Entities ('+entities.row_count+' unique)</h2>');
const eg=document.createElement('div');eg.style.cssText='display:flex;flex-wrap:wrap;gap:6px';
entities.rows.forEach(r=>{
const chip=document.createElement('span');
chip.style.cssText='background:var(--s2);border:1px solid var(--b);padding:4px 10px;border-radius:4px;font-size:11px;cursor:pointer';
chip.textContent=r.content+(r.n>1?' ('+r.n+')':'');
chip.onclick=()=>{showTab('explore');setTimeout(()=>{document.getElementById('searchInput').value=r.content;doSearch()},100)};
eg.appendChild(chip);
});
es.appendChild(eg);el.appendChild(es);
if(verified.rows.length){
const vs=document.createElement('div');vs.className='section';
vs.insertAdjacentHTML('beforeend','<h2>Verified by Secondary Model</h2>');
verified.rows.forEach(r=>vs.appendChild(makeEntry('verified',r.content,r.conversation+' p'+r.page)));
el.appendChild(vs);
}
}catch(e){el.textContent='Error: '+e}
}
// === SEARCH ===
function loadExplore(){
const el=$('tab-explore');el.textContent='';
const row=document.createElement('div');row.className='search-row';
const input=document.createElement('input');input.id='searchInput';input.placeholder='Search by meaning across all knowledge...';
const btn=document.createElement('button');btn.className='btn';btn.textContent='Search';
btn.onclick=doSearch;input.onkeydown=e=>{if(e.key==='Enter')doSearch()};
row.append(input,btn);
const res=document.createElement('div');res.id='searchResults';
const hint=document.createElement('div');hint.className='empty';hint.textContent='Try: "object storage architecture", "data ingestion trade-offs", "CRM options"';
el.append(row,res,hint);
}
async function doSearch(){
const query=document.getElementById('searchInput').value;if(!query)return;
const el=$('searchResults');el.textContent='searching...';
const results=await searchHNSW(query,15);
el.textContent='';
if(!results.length){el.textContent='No results';return}
results.forEach(r=>{
const e=document.createElement('div');e.className='result';
const s=document.createElement('div');s.className='result-score';s.textContent='['+r.score.toFixed(3)+'] '+r.doc_id.replace(/_/g,' ');
const t=document.createElement('div');t.className='result-text';t.textContent=r.chunk_text.substring(0,400);
e.append(s,t);el.appendChild(e);
});
}
// === ASK AI ===
function loadAsk(){
const el=$('tab-ask');el.textContent='';
const ta=document.createElement('textarea');ta.className='ask-input';ta.id='askInput';
ta.placeholder='Ask a question about your knowledge base...\n\nExamples:\n- What architecture decisions were made about Iceberg?\n- Compare Spark vs Flink approaches\n- What approaches were abandoned and why?';
const btn=document.createElement('button');btn.className='btn';btn.id='askBtn';btn.textContent='Ask AI';
btn.onclick=doAsk;
const resp=document.createElement('div');resp.id='askResp';
el.append(ta,btn,resp);
}
async function doAsk(){
const question=$('askInput').value;if(!question)return;
$('askBtn').disabled=true;$('askBtn').textContent='thinking...';
const el=$('askResp');el.textContent='searching knowledge...';
try{
const r=await(await fetch(API+'/vectors/rag',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({index_name:'observer_kb',question,top_k:5})})).json();
el.textContent='';
const ans=document.createElement('div');ans.className='ai-answer';ans.textContent=r.answer;
const meta=document.createElement('div');meta.className='ai-meta';meta.textContent='Model: '+r.model+(r.tokens_generated?' · '+r.tokens_generated+' tokens':'');
el.append(ans,meta);
if(r.sources?.length){
const sh=document.createElement('h2');sh.style.cssText='font-size:11px;color:var(--d);margin-top:14px';sh.textContent='SOURCES';el.appendChild(sh);
r.sources.forEach(s=>{
const e=document.createElement('div');e.className='result';e.style.marginTop='4px';
const sc=document.createElement('div');sc.className='result-score';sc.textContent='['+s.score.toFixed(3)+'] '+s.doc_id;
const t=document.createElement('div');t.className='result-text';t.style.fontSize='11px';t.textContent=s.chunk_text.substring(0,200);
e.append(sc,t);el.appendChild(e);
});
}
}catch(e){el.textContent='Error: '+e}
$('askBtn').disabled=false;$('askBtn').textContent='Ask AI';
}
// === UPLOAD ===
function loadUpload(){
const el=$('tab-upload');el.textContent='';
const zone=document.createElement('div');zone.className='upload-zone';
zone.onclick=()=>{const i=document.createElement('input');i.type='file';i.multiple=true;i.accept='.pdf,.txt,.csv,.json';i.onchange=e=>[...e.target.files].forEach(uploadFile);i.click()};
zone.ondragover=e=>{e.preventDefault();zone.style.borderColor='var(--a)'};
zone.ondragleave=()=>zone.style.borderColor='';
zone.ondrop=e=>{e.preventDefault();zone.style.borderColor='';[...e.dataTransfer.files].forEach(uploadFile)};
const h=document.createElement('h3');h.textContent='Drop files to ingest';
const p=document.createElement('p');p.textContent='PDF, CSV, JSON, Text — auto-extracted and indexed';
zone.append(h,p);
const list=document.createElement('div');list.id='upList';
el.append(zone,list);
}
function uploadFile(file){
const item=document.createElement('div');item.className='file-item';
const name=document.createElement('span');name.textContent=file.name;
const status=document.createElement('span');status.className='uploading';status.textContent='uploading...';
item.append(name,status);$('upList').prepend(item);
const fd=new FormData();fd.append('file',file);
const xhr=new XMLHttpRequest();xhr.open('POST',API+'/ingest/file');
xhr.upload.onprogress=e=>{if(e.lengthComputable)status.textContent=Math.round(e.loaded/e.total*100)+'%'};
xhr.onload=()=>{status.className=xhr.status<300?'done':'failed';status.textContent=xhr.status<300?'ingested':'failed'};
xhr.onerror=()=>{status.className='failed';status.textContent='error'};
xhr.send(fd);
}
loadKnowledge();
</script>
</body>
</html>
+235
View File
@@ -0,0 +1,235 @@
/// ChatGPT PDF conversation parser.
/// Extracts structured conversation turns from ChatGPT export PDFs.
/// Handles: user/assistant turns, timestamps, conversation titles.
use arrow::array::{ArrayRef, Int32Array, RecordBatch, StringArray};
use arrow::datatypes::{DataType, Field, Schema};
use std::sync::Arc;
/// A single conversation turn.
#[derive(Debug, Clone)]
pub struct ConversationTurn {
pub source_file: String,
pub conversation_title: String,
pub turn_number: i32,
pub role: String, // "user" or "assistant"
pub content: String,
pub page_number: i32,
}
/// Parse a ChatGPT PDF into structured conversation turns.
pub fn parse_chatgpt_pdf(content: &[u8], filename: &str) -> Result<(Arc<Schema>, Vec<RecordBatch>), String> {
let doc = lopdf::Document::load_mem(content)
.map_err(|e| format!("PDF load error: {e}"))?;
// Extract all text
let pages = doc.get_pages();
let mut all_text = String::new();
let mut page_texts: Vec<(i32, String)> = Vec::new();
for (&page_num, _) in pages.iter() {
let text = doc.extract_text(&[page_num]).unwrap_or_default();
let text = text.trim().to_string();
if !text.is_empty() {
page_texts.push((page_num as i32, text.clone()));
all_text.push_str(&text);
all_text.push('\n');
}
}
if all_text.trim().is_empty() {
return Err("PDF contains no extractable text".into());
}
// Try to parse as ChatGPT conversation format
let turns = extract_turns(&all_text, filename, &page_texts);
if turns.is_empty() {
// Fall back to page-based extraction if we can't detect conversation structure
return parse_as_raw_pages(&page_texts, filename);
}
tracing::info!("parsed ChatGPT PDF '{}': {} turns", filename, turns.len());
turns_to_batch(&turns)
}
/// Extract conversation turns from ChatGPT PDF text.
/// ChatGPT PDFs typically have patterns like:
/// - "You said:" / "ChatGPT said:" (newer format)
/// - "User:" / "Assistant:"
/// - Or just alternating blocks separated by formatting
fn extract_turns(text: &str, filename: &str, page_texts: &[(i32, String)]) -> Vec<ConversationTurn> {
let mut turns = Vec::new();
let mut turn_num = 0;
// Try to extract conversation title from first line
let title = text.lines().next()
.map(|l| l.trim().to_string())
.filter(|l| l.len() < 200 && !l.is_empty())
.unwrap_or_else(|| filename.replace(".pdf", ""));
// Strategy 1: Look for "You said:" / "ChatGPT said:" markers (common format)
let you_markers = ["You said:", "You:", "User:", "Human:"];
let ai_markers = ["ChatGPT said:", "ChatGPT:", "Assistant:", "AI:"];
let has_markers = you_markers.iter().any(|m| text.contains(m))
&& ai_markers.iter().any(|m| text.contains(m));
if has_markers {
let lines: Vec<&str> = text.lines().collect();
let mut current_role = String::new();
let mut current_content = String::new();
let mut current_page = 1i32;
for line in &lines {
let trimmed = line.trim();
// Check for role markers
let mut found_marker = false;
for marker in &you_markers {
if trimmed.starts_with(marker) || trimmed == *marker {
// Save previous turn
if !current_content.trim().is_empty() && !current_role.is_empty() {
turn_num += 1;
turns.push(ConversationTurn {
source_file: filename.to_string(),
conversation_title: title.clone(),
turn_number: turn_num,
role: current_role.clone(),
content: current_content.trim().to_string(),
page_number: current_page,
});
}
current_role = "user".to_string();
current_content = trimmed.strip_prefix(marker).unwrap_or("").trim().to_string();
current_content.push('\n');
found_marker = true;
break;
}
}
if !found_marker {
for marker in &ai_markers {
if trimmed.starts_with(marker) || trimmed == *marker {
if !current_content.trim().is_empty() && !current_role.is_empty() {
turn_num += 1;
turns.push(ConversationTurn {
source_file: filename.to_string(),
conversation_title: title.clone(),
turn_number: turn_num,
role: current_role.clone(),
content: current_content.trim().to_string(),
page_number: current_page,
});
}
current_role = "assistant".to_string();
current_content = trimmed.strip_prefix(marker).unwrap_or("").trim().to_string();
current_content.push('\n');
found_marker = true;
break;
}
}
}
if !found_marker && !current_role.is_empty() {
current_content.push_str(trimmed);
current_content.push('\n');
}
}
// Save last turn
if !current_content.trim().is_empty() && !current_role.is_empty() {
turn_num += 1;
turns.push(ConversationTurn {
source_file: filename.to_string(),
conversation_title: title.clone(),
turn_number: turn_num,
role: current_role,
content: current_content.trim().to_string(),
page_number: current_page,
});
}
return turns;
}
// Strategy 2: Split by paragraph blocks, alternate user/assistant
// This is a heuristic fallback for PDFs without clear markers
let paragraphs: Vec<&str> = text.split("\n\n")
.map(|p| p.trim())
.filter(|p| p.len() > 20) // skip short fragments
.collect();
if paragraphs.len() >= 2 {
for (i, para) in paragraphs.iter().enumerate() {
turn_num += 1;
turns.push(ConversationTurn {
source_file: filename.to_string(),
conversation_title: title.clone(),
turn_number: turn_num,
role: if i % 2 == 0 { "user" } else { "assistant" }.to_string(),
content: para.to_string(),
page_number: 1,
});
}
}
turns
}
/// Fallback: parse as raw pages with full text.
fn parse_as_raw_pages(page_texts: &[(i32, String)], filename: &str) -> Result<(Arc<Schema>, Vec<RecordBatch>), String> {
let schema = Arc::new(Schema::new(vec![
Field::new("source_file", DataType::Utf8, false),
Field::new("conversation_title", DataType::Utf8, false),
Field::new("turn_number", DataType::Int32, false),
Field::new("role", DataType::Utf8, false),
Field::new("content", DataType::Utf8, false),
Field::new("page_number", DataType::Int32, false),
]));
let sources: Vec<&str> = page_texts.iter().map(|_| filename).collect();
let titles: Vec<String> = page_texts.iter().map(|_| filename.replace(".pdf", "")).collect();
let turn_nums: Vec<i32> = (1..=page_texts.len() as i32).collect();
let roles: Vec<&str> = page_texts.iter().map(|_| "page").collect();
let contents: Vec<&str> = page_texts.iter().map(|(_, t)| t.as_str()).collect();
let pages: Vec<i32> = page_texts.iter().map(|(p, _)| *p).collect();
let arrays: Vec<ArrayRef> = vec![
Arc::new(StringArray::from(sources)),
Arc::new(StringArray::from(titles.iter().map(|s| s.as_str()).collect::<Vec<_>>())),
Arc::new(Int32Array::from(turn_nums)),
Arc::new(StringArray::from(roles)),
Arc::new(StringArray::from(contents)),
Arc::new(Int32Array::from(pages)),
];
let batch = RecordBatch::try_new(schema.clone(), arrays)
.map_err(|e| format!("RecordBatch error: {e}"))?;
Ok((schema, vec![batch]))
}
fn turns_to_batch(turns: &[ConversationTurn]) -> Result<(Arc<Schema>, Vec<RecordBatch>), String> {
let schema = Arc::new(Schema::new(vec![
Field::new("source_file", DataType::Utf8, false),
Field::new("conversation_title", DataType::Utf8, false),
Field::new("turn_number", DataType::Int32, false),
Field::new("role", DataType::Utf8, false),
Field::new("content", DataType::Utf8, false),
Field::new("page_number", DataType::Int32, false),
]));
let arrays: Vec<ArrayRef> = vec![
Arc::new(StringArray::from(turns.iter().map(|t| t.source_file.as_str()).collect::<Vec<_>>())),
Arc::new(StringArray::from(turns.iter().map(|t| t.conversation_title.as_str()).collect::<Vec<_>>())),
Arc::new(Int32Array::from(turns.iter().map(|t| t.turn_number).collect::<Vec<_>>())),
Arc::new(StringArray::from(turns.iter().map(|t| t.role.as_str()).collect::<Vec<_>>())),
Arc::new(StringArray::from(turns.iter().map(|t| t.content.as_str()).collect::<Vec<_>>())),
Arc::new(Int32Array::from(turns.iter().map(|t| t.page_number).collect::<Vec<_>>())),
];
let batch = RecordBatch::try_new(schema.clone(), arrays)
.map_err(|e| format!("RecordBatch error: {e}"))?;
Ok((schema, vec![batch]))
}
+1
View File
@@ -1,3 +1,4 @@
pub mod chatgpt_parser;
pub mod db_ingest;
pub mod detect;
pub mod csv_ingest;
+7 -1
View File
@@ -43,7 +43,13 @@ pub async fn ingest_file(
let (schema, batches) = match file_type {
FileType::Csv => csv_ingest::parse_csv(content)?,
FileType::Json | FileType::NdJson => json_ingest::parse_json(content)?,
FileType::Pdf => pdf_ingest::parse_pdf(content, filename)?,
FileType::Pdf => {
// Try ChatGPT parser first, fall back to generic PDF
match crate::chatgpt_parser::parse_chatgpt_pdf(content, filename) {
Ok(result) => result,
Err(_) => pdf_ingest::parse_pdf(content, filename)?,
}
}
FileType::Text => parse_text_file(content, filename)?,
FileType::Unknown => return Err(format!("unknown file type for '{filename}'")),
};
+21 -24
View File
@@ -40,48 +40,45 @@ pub fn chunk_text(
}];
}
// Work with char indices, not byte indices, to handle Unicode safely
let chars: Vec<char> = text.chars().collect();
let char_len = chars.len();
let mut chunks = Vec::new();
let mut start = 0;
let mut start = 0usize; // char index
let mut idx = 0u32;
while start < text.len() {
let end = (start + chunk_size).min(text.len());
while start < char_len {
let end = (start + chunk_size).min(char_len);
let chunk_str: String = chars[start..end].iter().collect();
// Try to break at a sentence or paragraph boundary
let chunk_text = &text[start..end];
let actual_end = if end < text.len() {
// Look for last sentence boundary in the chunk
if let Some(pos) = chunk_text.rfind(". ") {
start + pos + 2
} else if let Some(pos) = chunk_text.rfind('\n') {
start + pos + 1
let actual_end = if end < char_len {
// Look for last sentence boundary
if let Some(pos) = chunk_str.rfind(". ") {
start + chunk_str[..pos + 2].chars().count()
} else if let Some(pos) = chunk_str.rfind('\n') {
start + chunk_str[..pos + 1].chars().count()
} else if let Some(pos) = chunk_str.rfind(' ') {
start + chunk_str[..pos + 1].chars().count()
} else {
// Fall back to word boundary
if let Some(pos) = chunk_text.rfind(' ') {
start + pos + 1
} else {
end
}
end
}
} else {
end
};
let chunk = text[start..actual_end].trim();
if !chunk.is_empty() {
let chunk: String = chars[start..actual_end].iter().collect();
let trimmed = chunk.trim();
if !trimmed.is_empty() {
chunks.push(TextChunk {
source: source.to_string(),
doc_id: doc_id.to_string(),
chunk_idx: idx,
text: chunk.to_string(),
text: trimmed.to_string(),
});
idx += 1;
}
// Advance with overlap
if actual_end >= text.len() {
break;
}
if actual_end >= char_len { break; }
start = if actual_end > overlap { actual_end - overlap } else { actual_end };
}
@@ -0,0 +1,72 @@
{
"id": "0ae8f29b-bf70-4c9d-8b64-6ea7cb7f76a1",
"name": "rust_in_tech_stack",
"schema_fingerprint": "314ee8ede68287240d7cec1403d252787c8ceb4ee2afa72cffcb0853ac3ac889",
"objects": [
{
"bucket": "data",
"key": "datasets/rust_in_tech_stack.parquet",
"size_bytes": 99623,
"created_at": "2026-03-28T03:32:50.769271381Z"
}
],
"created_at": "2026-03-28T03:32:50.769272195Z",
"updated_at": "2026-03-28T03:32:50.769710478Z",
"description": "",
"owner": "",
"sensitivity": null,
"columns": [
{
"name": "source_file",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "conversation_title",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "turn_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "role",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "content",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "page_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
}
],
"lineage": {
"source_system": "pdf",
"source_file": "Rust in Tech Stack.pdf",
"ingest_job": "ingest-1774668770769",
"ingest_timestamp": "2026-03-28T03:32:50.769271381Z",
"parent_datasets": []
},
"freshness": null,
"tags": [],
"row_count": 49
}
@@ -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
}
@@ -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
}
@@ -0,0 +1,72 @@
{
"id": "54cc1ae9-14fe-40f6-88e4-7c623f57b577",
"name": "crm_development_options",
"schema_fingerprint": "bb04328fa99161403fa0993f59bf14adcf08eb551c7b0c5e9b9e305e5e0c8556",
"objects": [
{
"bucket": "data",
"key": "datasets/crm_development_options.parquet",
"size_bytes": 50528,
"created_at": "2026-03-28T03:30:29.036504470Z"
}
],
"created_at": "2026-03-28T03:30:29.036504841Z",
"updated_at": "2026-03-28T03:30:29.036642992Z",
"description": "",
"owner": "",
"sensitivity": null,
"columns": [
{
"name": "source_file",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "conversation_title",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "turn_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "role",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "content",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "page_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
}
],
"lineage": {
"source_system": "pdf",
"source_file": "CRM Development Options.pdf",
"ingest_job": "ingest-1774668629036",
"ingest_timestamp": "2026-03-28T03:30:29.036504470Z",
"parent_datasets": []
},
"freshness": null,
"tags": [],
"row_count": 25
}
@@ -0,0 +1,72 @@
{
"id": "5d4bbdb1-fb53-4cea-86d5-309fe067111c",
"name": "legacy_staffing_crm_search",
"schema_fingerprint": "87b3fa58cae74468bf7fbbe32397b1a88bc281832bb0226b9eae5ce2501cda4e",
"objects": [
{
"bucket": "data",
"key": "datasets/legacy_staffing_crm_search.parquet",
"size_bytes": 2722188,
"created_at": "2026-03-28T03:34:14.444384147Z"
}
],
"created_at": "2026-03-28T03:34:14.444384964Z",
"updated_at": "2026-03-28T03:34:14.444631183Z",
"description": "",
"owner": "",
"sensitivity": null,
"columns": [
{
"name": "source_file",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "conversation_title",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "turn_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "role",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "content",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "page_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
}
],
"lineage": {
"source_system": "pdf",
"source_file": "Legacy Staffing CRM Search.pdf",
"ingest_job": "ingest-1774668854444",
"ingest_timestamp": "2026-03-28T03:34:14.444384147Z",
"parent_datasets": []
},
"freshness": null,
"tags": [],
"row_count": 1
}
@@ -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
}
@@ -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
}
@@ -0,0 +1,72 @@
{
"id": "7ca19535-838e-4f79-aa6f-d8ebaa2e6edc",
"name": "code_sorting_assistance",
"schema_fingerprint": "47e317d38bbad23688f6c274cbecc3dfd7076e42a6a0d09002a0bef835789bc2",
"objects": [
{
"bucket": "data",
"key": "datasets/code_sorting_assistance.parquet",
"size_bytes": 122034,
"created_at": "2026-03-28T03:31:08.585681049Z"
}
],
"created_at": "2026-03-28T03:31:08.585682007Z",
"updated_at": "2026-03-28T03:31:08.585978498Z",
"description": "",
"owner": "",
"sensitivity": null,
"columns": [
{
"name": "source_file",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "conversation_title",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "turn_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "role",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "content",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "page_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
}
],
"lineage": {
"source_system": "pdf",
"source_file": "Code Sorting Assistance.pdf",
"ingest_job": "ingest-1774668668585",
"ingest_timestamp": "2026-03-28T03:31:08.585681049Z",
"parent_datasets": []
},
"freshness": null,
"tags": [],
"row_count": 70
}
@@ -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
}
@@ -0,0 +1,72 @@
{
"id": "8f5c82bf-93ec-4ce7-b2d1-61e559242add",
"name": "llms3_update_prompt",
"schema_fingerprint": "77cfcdb5e22f095302bb816abd4e33380a6b357cd7d23efee5076083020fbf32",
"objects": [
{
"bucket": "data",
"key": "datasets/llms3_update_prompt.parquet",
"size_bytes": 39396,
"created_at": "2026-03-28T03:28:04.752204191Z"
}
],
"created_at": "2026-03-28T03:28:04.752204734Z",
"updated_at": "2026-03-28T03:28:04.752281926Z",
"description": "",
"owner": "",
"sensitivity": null,
"columns": [
{
"name": "source_file",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "conversation_title",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "turn_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "role",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "content",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "page_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
}
],
"lineage": {
"source_system": "pdf",
"source_file": "LLMS3 Update Prompt.pdf",
"ingest_job": "ingest-1774668484752",
"ingest_timestamp": "2026-03-28T03:28:04.752204191Z",
"parent_datasets": []
},
"freshness": null,
"tags": [],
"row_count": 6
}
@@ -0,0 +1,72 @@
{
"id": "d609d775-1a8e-4114-8aee-3238017229a7",
"name": "hybrid_indexing_setup",
"schema_fingerprint": "dfb09f4395d131e5fda5dab65f8c9dbe395d46d0e828f504e053cdb1cf23f24b",
"objects": [
{
"bucket": "data",
"key": "datasets/hybrid_indexing_setup.parquet",
"size_bytes": 131530,
"created_at": "2026-03-28T03:30:28.978970613Z"
}
],
"created_at": "2026-03-28T03:30:28.978971805Z",
"updated_at": "2026-03-28T03:30:28.979244113Z",
"description": "",
"owner": "",
"sensitivity": null,
"columns": [
{
"name": "source_file",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "conversation_title",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "turn_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "role",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "content",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "page_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
}
],
"lineage": {
"source_system": "pdf",
"source_file": "Hybrid Indexing Setup.pdf",
"ingest_job": "ingest-1774668628978",
"ingest_timestamp": "2026-03-28T03:30:28.978970613Z",
"parent_datasets": []
},
"freshness": null,
"tags": [],
"row_count": 69
}
@@ -0,0 +1,72 @@
{
"id": "f9055d87-c7c5-4109-b311-11fab418541f",
"name": "spark_vs_flink_iceberg",
"schema_fingerprint": "e3f74df7c9db965c184ff5f5ec92b0f68168a7f6e367e8253d28fca571a11688",
"objects": [
{
"bucket": "data",
"key": "datasets/spark_vs_flink_iceberg.parquet",
"size_bytes": 163762,
"created_at": "2026-03-28T03:35:18.667429709Z"
}
],
"created_at": "2026-03-28T03:35:18.667430594Z",
"updated_at": "2026-03-28T03:35:18.667593640Z",
"description": "",
"owner": "",
"sensitivity": null,
"columns": [
{
"name": "source_file",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "conversation_title",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "turn_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "role",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "content",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "page_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
}
],
"lineage": {
"source_system": "pdf",
"source_file": "Spark vs Flink Iceberg.pdf",
"ingest_job": "ingest-1774668918667",
"ingest_timestamp": "2026-03-28T03:35:18.667429709Z",
"parent_datasets": []
},
"freshness": null,
"tags": [],
"row_count": 124
}
@@ -0,0 +1,72 @@
{
"id": "faef511e-4f34-4002-b2f4-8aecb8cfb80e",
"name": "hybrid_ai_data_lake",
"schema_fingerprint": "1fa158168cf5f542e9cdeec30c9e9aad9b12285b66e6fbdb7321e5d9f7039515",
"objects": [
{
"bucket": "data",
"key": "datasets/hybrid_ai_data_lake.parquet",
"size_bytes": 220352,
"created_at": "2026-03-28T03:35:16.027063136Z"
}
],
"created_at": "2026-03-28T03:35:16.027064491Z",
"updated_at": "2026-03-28T03:35:16.027333389Z",
"description": "",
"owner": "",
"sensitivity": null,
"columns": [
{
"name": "source_file",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "conversation_title",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "turn_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "role",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "content",
"data_type": "Utf8",
"sensitivity": null,
"description": "",
"is_pii": false
},
{
"name": "page_number",
"data_type": "Int32",
"sensitivity": null,
"description": "",
"is_pii": false
}
],
"lineage": {
"source_system": "pdf",
"source_file": "Hybrid AI Data Lake.pdf",
"ingest_job": "ingest-1774668916027",
"ingest_timestamp": "2026-03-28T03:35:16.027063136Z",
"parent_datasets": []
},
"freshness": null,
"tags": [],
"row_count": 133
}
@@ -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
}
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+15
View File
@@ -0,0 +1,15 @@
{
"index_name": "observer_kb",
"source": "chatgpt_pdfs",
"model_name": "nomic-embed-text",
"model_version": "latest",
"dimensions": 768,
"chunk_count": 477,
"doc_count": 477,
"chunk_size": 500,
"overlap": 50,
"storage_key": "vectors/observer_kb.parquet",
"created_at": "2026-03-28T03:52:23.689193948Z",
"build_time_secs": 11.022282,
"chunks_per_sec": 43.275978
}
+15
View File
@@ -0,0 +1,15 @@
{
"index_name": "observer_kb_test",
"source": "chatgpt",
"model_name": "nomic-embed-text",
"model_version": "latest",
"dimensions": 768,
"chunk_count": 100,
"doc_count": 100,
"chunk_size": 500,
"overlap": 50,
"storage_key": "vectors/observer_kb_test.parquet",
"created_at": "2026-03-28T03:52:02.861946987Z",
"build_time_secs": 1.8791063,
"chunks_per_sec": 53.21679
}
+15
View File
@@ -0,0 +1,15 @@
{
"index_name": "observer_test",
"source": "test",
"model_name": "nomic-embed-text",
"model_version": "latest",
"dimensions": 768,
"chunk_count": 15,
"doc_count": 5,
"chunk_size": 500,
"overlap": 50,
"storage_key": "vectors/observer_test.parquet",
"created_at": "2026-03-28T03:46:22.891189902Z",
"build_time_secs": 2.113568,
"chunks_per_sec": 7.0970035
}
+15
View File
@@ -0,0 +1,15 @@
{
"index_name": "quick_test",
"source": "test",
"model_name": "nomic-embed-text",
"model_version": "latest",
"dimensions": 768,
"chunk_count": 5,
"doc_count": 5,
"chunk_size": 500,
"overlap": 50,
"storage_key": "vectors/quick_test.parquet",
"created_at": "2026-03-28T03:51:49.689474954Z",
"build_time_secs": 1.3572433,
"chunks_per_sec": 3.6839378
}
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+173
View File
@@ -0,0 +1,173 @@
#!/usr/bin/env python3
"""
Multi-pass deep analysis workflow.
Demonstrates the architecture's ability to dynamically discover and organize
knowledge from large-scale conversation data using local LLM processing.
Pass 1: Corpus overview — what's in the data
Pass 2: Pattern detection — recurring themes across conversations
Pass 3: Evolution tracking — how thinking changed over time
Pass 4: Gap analysis — what's missing or unresolved
Pass 5: Synthesis — actionable knowledge report
"""
import json, urllib.request, time
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 ask(prompt, max_tokens=800):
req = urllib.request.Request(f"{API}/ai/generate",
data=json.dumps({"prompt": prompt, "max_tokens": max_tokens, "temperature": 0.2}).encode(),
headers={"Content-Type": "application/json"}, method="POST")
return json.loads(urllib.request.urlopen(req, timeout=180).read())['text'].strip()
def search(query, top_k=10):
req = urllib.request.Request(f"{API}/vectors/hnsw/search",
data=json.dumps({"index_name": "observer_kb", "query": query, "top_k": top_k}).encode(),
headers={"Content-Type": "application/json"}, method="POST")
return json.loads(urllib.request.urlopen(req, timeout=30).read())['results']
print("=" * 70)
print("DEEP ANALYSIS: Multi-Pass Knowledge Discovery")
print("=" * 70)
# ============================================================
print("\n" + "=" * 70)
print("PASS 1: Corpus Overview")
print("=" * 70)
# Get all conversation stats
convs = sql("SELECT conversation, COUNT(*) as mentions FROM knowledge_topics GROUP BY conversation ORDER BY mentions DESC")
topic_stats = sql("SELECT topic, COUNT(*) as n FROM knowledge_topics GROUP BY topic ORDER BY n DESC LIMIT 20")
decision_count = sql("SELECT COUNT(*) as n FROM knowledge_decisions")['rows'][0]['n']
insight_count = sql("SELECT COUNT(*) as n FROM knowledge_insights")['rows'][0]['n']
question_count = sql("SELECT COUNT(*) as n FROM knowledge_questions")['rows'][0]['n']
print(f"\nCorpus: {len(convs['rows'])} conversations")
print(f"Extracted: {decision_count} decisions, {insight_count} insights, {question_count} open questions")
print(f"\nTop topics:")
for r in topic_stats['rows'][:10]:
print(f" {r['n']:>3}x {r['topic']}")
# Feed corpus overview to LLM for characterization
overview_text = "\n".join([f" {r['topic']} ({r['n']}x)" for r in topic_stats['rows']])
corpus_analysis = ask(f"""You are analyzing a knowledge base extracted from technical AI conversations.
Here are the most frequent topics discussed:
{overview_text}
The corpus contains {decision_count} decisions, {insight_count} insights, and {question_count} open questions across {len(convs['rows'])} conversations.
Provide a 3-sentence characterization of what this person is building and what domains they're focused on. Be specific about the technical direction.""")
print(f"\nAI Characterization:")
print(f" {corpus_analysis}")
# ============================================================
print("\n" + "=" * 70)
print("PASS 2: Cross-Conversation Pattern Detection")
print("=" * 70)
# Find topics that appear across MULTIPLE conversations
cross_topics = sql("SELECT topic, COUNT(DISTINCT conversation) as convs, COUNT(*) as total FROM knowledge_topics GROUP BY topic HAVING COUNT(DISTINCT conversation) > 1 ORDER BY convs DESC LIMIT 15")
print("\nTopics spanning multiple conversations (recurring themes):")
for r in cross_topics['rows']:
print(f" {r['topic']}: {r['convs']} conversations, {r['total']} mentions")
# Use semantic search to find related content across conversations
print("\nSemantic pattern discovery:")
patterns_to_check = [
"architecture decisions about data storage and object storage",
"trade-offs between different technologies and frameworks",
"problems that were identified but not solved",
"approaches that were tried and abandoned",
]
for pattern in patterns_to_check:
results = search(pattern, 3)
if results:
sources = set(r['doc_id'].split('_p')[0] for r in results)
print(f"\n Pattern: \"{pattern}\"")
print(f" Found across: {', '.join(sources)}")
print(f" Top match [{results[0]['score']:.3f}]: {results[0]['chunk_text'][:100]}...")
# ============================================================
print("\n" + "=" * 70)
print("PASS 3: Decision Evolution")
print("=" * 70)
# Get all decisions and analyze for contradictions/evolution
decisions_sample = sql("SELECT conversation, page, decision FROM knowledge_decisions LIMIT 30")
decisions_text = "\n".join([f"[{r['conversation']} p{r['page']}] {r['decision']}" for r in decisions_sample['rows']])
evolution = ask(f"""Analyze these technical decisions from multiple conversations over time.
Identify:
1. Decisions that CONTRADICT each other (thinking changed)
2. Decisions that BUILD on each other (progressive refinement)
3. Decisions that remain CONSISTENT (core principles)
Decisions:
{decisions_text}
Format your response as three sections: CONTRADICTIONS, PROGRESSIVE, CONSISTENT. Be specific about which decisions you're comparing.""")
print(evolution)
# ============================================================
print("\n" + "=" * 70)
print("PASS 4: Gap Analysis")
print("=" * 70)
questions_sample = sql("SELECT conversation, question FROM knowledge_questions LIMIT 20")
questions_text = "\n".join([f"[{r['conversation']}] {r['question']}" for r in questions_sample['rows']])
insights_sample = sql("SELECT conversation, insight FROM knowledge_insights LIMIT 20")
insights_text = "\n".join([f"[{r['conversation']}] {r['insight']}" for r in insights_sample['rows']])
gaps = ask(f"""You are analyzing a technical knowledge base.
OPEN QUESTIONS (things that were asked but not resolved):
{questions_text}
KEY INSIGHTS (things that were learned):
{insights_text}
Identify:
1. Which open questions have actually been ANSWERED by insights in other conversations? (recovered knowledge)
2. Which questions remain genuinely UNRESOLVED? (real gaps)
3. What important topics are MISSING entirely? (blind spots)
Be specific.""")
print(gaps)
# ============================================================
print("\n" + "=" * 70)
print("PASS 5: Synthesis — Actionable Knowledge Report")
print("=" * 70)
synthesis = ask(f"""Based on a deep analysis of {len(convs['rows'])} technical conversations containing {decision_count} decisions and {insight_count} insights, synthesize the following:
The person is building: {corpus_analysis[:200]}
Top recurring topics: {', '.join([r['topic'] for r in cross_topics['rows'][:8]])}
Key question: Given all this work, what are the 3 most important things this person should focus on NEXT? What patterns suggest they keep revisiting the same problems? What knowledge has been recovered that should be applied immediately?
Be direct and actionable. No filler.""")
print(synthesis)
print("\n" + "=" * 70)
print("ANALYSIS COMPLETE")
print(f"Processed: {len(convs['rows'])} conversations, {decision_count} decisions, {insight_count} insights")
print(f"Infrastructure: SQL (DataFusion) + Vector Search (HNSW) + Local LLM (qwen2.5)")
print(f"All local. No data left the machine.")
print("=" * 70)
+135
View File
@@ -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")
+163
View File
@@ -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")