diff --git a/apps/observer-kb/docs/PRD.md b/apps/observer-kb/docs/PRD.md new file mode 100644 index 0000000..7004967 --- /dev/null +++ b/apps/observer-kb/docs/PRD.md @@ -0,0 +1,159 @@ +# PRD: Observer Knowledge Base + +**Status:** Active +**Created:** 2026-03-28 +**Owner:** J +**Branch:** `observer-kb` +**Parent:** Lakehouse core (crates/) + +--- + +## Problem + +AI conversations (ChatGPT, Claude, etc.) generate massive amounts of knowledge — decisions, research, code, strategies, troubleshooting — that disappear into PDF exports and chat logs. There's no way to: +- Search across hundreds of conversations by meaning +- Find "what did I discuss about X six months ago?" +- Compare approaches across different conversations +- Build a reusable knowledge base from past AI interactions +- Track which ideas worked and which didn't + +## Solution + +An upload portal + knowledge extraction pipeline built on the Lakehouse infrastructure. Upload ChatGPT/Claude PDFs → extract conversations → chunk and embed → searchable knowledge base with comparison tools. + +--- + +## Architecture + +``` +Upload Portal (web UI) + │ + ├── Upload large PDFs (chunked upload for big files) + ├── Progress tracking per file + └── Visual extraction preview + │ + ▼ +Extraction Pipeline (uses Lakehouse ingestd) + │ + ├── PDF text extraction (lopdf) + ├── Conversation turn parsing (user/assistant) + ├── Topic detection + ├── Metadata extraction (dates, titles, participants) + └── Quality scoring (how useful was this conversation?) + │ + ▼ +Lakehouse Storage + │ + ├── conversations table (id, title, source, date, turns, topic) + ├── turns table (conversation_id, turn_num, role, content) + ├── topics table (topic, conversation_ids, summary) + └── vectors/ (embedded turns for semantic search) + │ + ▼ +Knowledge Interface + │ + ├── Search: "what did I discuss about Kubernetes?" + ├── Compare: conversation A vs B on same topic + ├── Timeline: when did I first explore this idea? + ├── Extract: pull key decisions/code/strategies + └── Playbook: which approaches worked? +``` + +## Tech Stack + +Uses Lakehouse core — NOT a separate codebase: +- **Storage:** Lakehouse storaged (Parquet on object storage) +- **Catalog:** Lakehouse catalogd (metadata, PII, lineage) +- **Query:** Lakehouse queryd (DataFusion SQL) +- **Vectors:** Lakehouse vectord (HNSW search, RAG) +- **AI:** Lakehouse aibridge → Ollama +- **New:** Upload portal (standalone HTML/JS — NOT Dioxus WASM) + +## Why HTML/JS for the Upload Portal + +- No 39MB WASM payload +- Chunked file upload (XMLHttpRequest with progress) +- Works on any browser +- Fast to iterate +- The Dioxus UI can coexist for the SQL/Ask features + +--- + +## Phases + +### Phase 1: Upload Portal +- Chunked file upload (handles 100MB+ PDFs) +- Upload progress bar per file +- Batch upload (drag & drop multiple) +- File list with status (uploading, processing, done, failed) +- Served as static HTML from the gateway + +### Phase 2: Extraction Pipeline +- ChatGPT PDF parser (conversation turns, titles, roles) +- Claude PDF parser (similar structure) +- Generic PDF fallback (page-based) +- Extraction preview: see what was extracted before committing +- Auto-embed extracted turns + +### Phase 3: Knowledge Browser +- Search across all conversations by meaning +- Filter by date, topic, source +- View individual conversations with turns +- Highlight matching content in search results + +### Phase 4: Comparison Tools +- Side-by-side conversation comparison +- Topic tracking across conversations over time +- "Best approach" extraction from successful conversations +- Playbook builder: curate successful patterns + +--- + +## Data Model + +```sql +-- Conversations (one per PDF/chat export) +conversations: + conversation_id TEXT PRIMARY KEY + title TEXT + source TEXT -- "chatgpt", "claude", "other" + source_file TEXT + date TEXT + turn_count INTEGER + total_chars INTEGER + topics TEXT -- pipe-separated + summary TEXT -- AI-generated + +-- Individual turns +turns: + turn_id TEXT PRIMARY KEY + conversation_id TEXT + turn_number INTEGER + role TEXT -- "user", "assistant" + content TEXT + char_count INTEGER + has_code BOOLEAN + page_number INTEGER + +-- Vector embeddings (via vectord) +-- Stored in vectors/observer_kb.parquet +``` + +--- + +## Upload Requirements + +- Max file size: 500MB per PDF +- Chunked upload: 5MB chunks with resume on failure +- Batch: up to 50 files at once +- Progress: per-file percentage + overall +- Dedup: skip already-ingested files (content hash) + +--- + +## Non-Goals (for now) + +- Live chat ingestion (only PDF exports) +- Editing conversations +- Multi-user access control +- Real-time collaboration diff --git a/apps/observer-kb/index.html b/apps/observer-kb/index.html new file mode 100644 index 0000000..34f1170 --- /dev/null +++ b/apps/observer-kb/index.html @@ -0,0 +1,178 @@ + + + + + +Observer KB + + + +
+

OBSERVER KB

loading...
+
+ + + +
+ +
+
+

Drop PDF files here

+

ChatGPT exports, Claude logs, any PDF — or click to browse

+ +
+
+
+ + + + +
+ + + + diff --git a/crates/ingestd/src/chatgpt_parser.rs b/crates/ingestd/src/chatgpt_parser.rs new file mode 100644 index 0000000..270941f --- /dev/null +++ b/crates/ingestd/src/chatgpt_parser.rs @@ -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, Vec), 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 { + 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, Vec), 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 = page_texts.iter().map(|_| filename.replace(".pdf", "")).collect(); + let turn_nums: Vec = (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 = page_texts.iter().map(|(p, _)| *p).collect(); + + let arrays: Vec = vec![ + Arc::new(StringArray::from(sources)), + Arc::new(StringArray::from(titles.iter().map(|s| s.as_str()).collect::>())), + 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, Vec), 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 = vec![ + Arc::new(StringArray::from(turns.iter().map(|t| t.source_file.as_str()).collect::>())), + Arc::new(StringArray::from(turns.iter().map(|t| t.conversation_title.as_str()).collect::>())), + Arc::new(Int32Array::from(turns.iter().map(|t| t.turn_number).collect::>())), + Arc::new(StringArray::from(turns.iter().map(|t| t.role.as_str()).collect::>())), + Arc::new(StringArray::from(turns.iter().map(|t| t.content.as_str()).collect::>())), + Arc::new(Int32Array::from(turns.iter().map(|t| t.page_number).collect::>())), + ]; + + let batch = RecordBatch::try_new(schema.clone(), arrays) + .map_err(|e| format!("RecordBatch error: {e}"))?; + + Ok((schema, vec![batch])) +} diff --git a/crates/ingestd/src/lib.rs b/crates/ingestd/src/lib.rs index e0068b0..159e13e 100644 --- a/crates/ingestd/src/lib.rs +++ b/crates/ingestd/src/lib.rs @@ -1,3 +1,4 @@ +pub mod chatgpt_parser; pub mod db_ingest; pub mod detect; pub mod csv_ingest; diff --git a/crates/ingestd/src/pipeline.rs b/crates/ingestd/src/pipeline.rs index 65c056f..76939ae 100644 --- a/crates/ingestd/src/pipeline.rs +++ b/crates/ingestd/src/pipeline.rs @@ -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}'")), };