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>
This commit is contained in:
root
@@ -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)
|
||||
Reference in New Issue
Block a user