nandi.uk
This repository has no description
16 kB
448 lines
1#!/usr/bin/env python3
2"""Strata analysis: fetch islands, produce prose with semantic arrows, push to PDS as org.latha.strata records.
3
4Produces an org.latha.strata record for each island containing:
5- source: lexmin vertex (canonical component reference)
6- analysis: themes, connections, tensions, open questions, synthesis
7- createdAt: timestamp
8
9Usage: python3 scripts/strata-analyze.py [--force] [--island ID]
10"""
11
12import json
13import os
14import subprocess
15import sys
16import time
17import urllib.parse
18import urllib.request
19
20API_BASE = "https://stigmergic.latha.org"
21LETTA_BIN = "/home/nandi/.local/npm-global/bin/letta"
22VENV_PYTHON = "/home/nandi/code/stigmergic/.venv/bin/python3"
23
24
25def fetch_islands():
26 req = urllib.request.Request(
27 f"{API_BASE}/xrpc/org.latha.strata.getIslands",
28 headers={"User-Agent": "stigmergic-strata/1.0"},
29 )
30 with urllib.request.urlopen(req) as resp:
31 return json.loads(resp.read())["islands"]
32
33
34def build_island_context(island):
35 """Build a rich context for the LLM including vertex titles and edge relationships."""
36 vertices = island["vertices"]
37 edges = island["edges"]
38 meta = island.get("vertexMeta", {})
39
40 # Vertex summaries with titles
41 vertex_lines = []
42 for v in vertices[:30]:
43 m = meta.get(v, {})
44 title = m.get("title", v)
45 desc = m.get("description", "")
46 vtype = m.get("type", "unknown")
47 line = f" - [{vtype}] {title}"
48 if desc:
49 line += f": {desc[:100]}"
50 vertex_lines.append(line)
51
52 # Edge relationships as arrows
53 edge_lines = []
54 for e in edges[:30]:
55 source_meta = meta.get(e.get("source", ""), {})
56 target_meta = meta.get(e.get("target", ""), {})
57 source_title = source_meta.get("title", e.get("source", ""))[:60]
58 target_title = target_meta.get("title", e.get("target", ""))[:60]
59 ctype = e.get("connectionType", "relates")
60 # Normalize connection type
61 raw = ctype.split("#")[-1] if "#" in ctype else ctype
62 label_map = {
63 "RELATED": "relates to",
64 "LEADS_TO": "leads to",
65 "SUPPLEMENT": "supplements",
66 "SUPPLEMENTS": "supplements",
67 "OPPOSES": "opposes",
68 "SUPPORTS": "supports",
69 "ADDRESSES": "addresses",
70 "EXPLAINER": "explains",
71 "HELPFUL": "helpful for",
72 "CONTAINS": "contains",
73 "CITES": "cites",
74 "INSPIRED": "inspired by",
75 "CONTRADICTS": "contradicts",
76 "EXTENDS": "extends",
77 "FORKS": "forks from",
78 "ANNOTATES": "annotates",
79 "RELATES": "relates to",
80 }
81 relation = label_map.get(raw, raw.lower().replace("_", " "))
82 note = e.get("note", "")
83 line = f" {source_title} → {relation} → {target_title}"
84 if note:
85 line += f" ({note})"
86 edge_lines.append(line)
87
88 # Domains
89 domains = {}
90 for v in vertices:
91 if v.startswith("http"):
92 try:
93 from urllib.parse import urlparse
94 h = urlparse(v).hostname.replace("www.", "")
95 domains[h] = domains.get(h, 0) + 1
96 except Exception:
97 pass
98 top_domains = sorted(domains.items(), key=lambda x: -x[1])[:5]
99
100 return {
101 "vertex_count": len(vertices),
102 "edge_count": len(edges),
103 "top_domains": [f"{d[0]} ({d[1]})" for d in top_domains],
104 "vertices": vertex_lines,
105 "edges": edge_lines,
106 }
107
108
109def analyze_with_letta(context):
110 """Ask letta to produce strata prose with semantic arrows + discourse graph."""
111 prompt = """You are a knowledge synthesis engine. Given a knowledge graph island (connected component), produce a strata analysis as structured prose and a discourse graph.
112
113Rules:
114- Use arrows to show semantic relationships: → (forward), ← (backward), ↔ (bidirectional), ⇏ (blocked/failed)
115- Each relationship line should be: Source → relation → Target
116- CRITICAL: You MUST use the EXACT vertex titles from the island data below — do not paraphrase, abbreviate, or rename them. If a title is long, use it anyway. This is required so the titles can be linked back to their source pages.
117- Group related vertices into thematic clusters
118- Note tensions with ⚠
119- End with open questions marked with ?
120- Be concise and specific — reference actual titles, not generic descriptions
121- No preamble, no markdown, just the prose
122
123Format:
124[Title]
125A single concise sentence that captures the core argument of this island — not a list of topics, but a cohesive claim or narrative arc.
126
127[Themes]
128theme1, theme2, ...
129
130[Relationships]
131Source → relation → Target
132Source ← relation ← Target
133...
134
135[Tensions]
136⚠ tension description
137
138[Open Questions]
139? question
140
141[Synthesis]
1422-3 sentence synthesis paragraph
143
144[DiscourseGraph]
145Classify each vertex as a discourse node type:
146- QUE: poses a question, identifies a knowledge gap, or asks "what about..."
147- CLM: makes a claim, argument, or takes a position
148- EVD: provides empirical evidence, data, or concrete observations
149- SRC: is a source document, reference, or background material
150
151Then identify discourse relations between classified nodes:
152- informs: EVD → QUE (evidence informs a question)
153- supports: EVD → CLM (evidence supports a claim)
154- opposes: EVD → CLM (evidence opposes/contradicts a claim)
155- addresses: CLM → QUE (claim addresses a question)
156
157Format — one per line:
158NODE: title_or_short_id : TYPE
159REL: source → relation → target
160
161Island data:
162""" + json.dumps(context, indent=2)
163
164 try:
165 result = subprocess.run(
166 [LETTA_BIN, "-p", prompt, "--output-format", "json", "--system", "letta", "--memfs"],
167 capture_output=True,
168 text=True,
169 timeout=120,
170 )
171 except subprocess.TimeoutExpired:
172 return None
173 if result.returncode != 0:
174 print(f" letta error: {result.stderr[:200]}", file=sys.stderr)
175 return None
176 try:
177 data = json.loads(result.stdout)
178 if data.get("type") == "result":
179 return data["result"].strip()
180 except Exception:
181 pass
182 return result.stdout.strip()[:5000] or None
183
184
185def parse_strata(raw_prose):
186 """Parse the LLM output into structured sections."""
187 if not raw_prose:
188 return None
189
190 result = {
191 "prose": raw_prose,
192 "title": "",
193 "themes": [],
194 "relationships": [],
195 "tensions": [],
196 "open_questions": [],
197 "synthesis": "",
198 "discourse_graph": {"nodes": [], "edges": []},
199 }
200
201 current_section = None
202 for line in raw_prose.split("\n"):
203 line = line.strip()
204 if not line:
205 continue
206
207 if line.startswith("[Title]"):
208 current_section = "title"
209 continue
210 elif line.startswith("[Themes]"):
211 current_section = "themes"
212 continue
213 elif line.startswith("[Relationships]"):
214 current_section = "relationships"
215 continue
216 elif line.startswith("[Tensions]"):
217 current_section = "tensions"
218 continue
219 elif line.startswith("[Open Questions]"):
220 current_section = "questions"
221 continue
222 elif line.startswith("[Synthesis]"):
223 current_section = "synthesis"
224 continue
225 elif line.startswith("[DiscourseGraph]"):
226 current_section = "discourse"
227 continue
228
229 if current_section == "title":
230 result["title"] = line.strip()
231 elif current_section == "themes":
232 result["themes"].extend([t.strip() for t in line.split(",") if t.strip()])
233 elif current_section == "relationships":
234 result["relationships"].append(line)
235 elif current_section == "tensions":
236 result["tensions"].append(line.lstrip("⚠ ").strip())
237 elif current_section == "questions":
238 result["open_questions"].append(line.lstrip("? ").strip())
239 elif current_section == "synthesis":
240 result["synthesis"] += line + " "
241 elif current_section == "discourse":
242 if line.startswith("NODE:"):
243 # NODE: title : TYPE
244 parts = line[5:].rsplit(":", 2)
245 if len(parts) >= 2:
246 title = parts[-2].strip()
247 ntype = parts[-1].strip().upper()
248 if ntype in ("QUE", "CLM", "EVD", "SRC"):
249 result["discourse_graph"]["nodes"].append({
250 "id": title,
251 "type": ntype,
252 })
253 elif line.startswith("REL:"):
254 # REL: source → relation → target
255 rel_text = line[4:].strip()
256 parts = [p.strip() for p in rel_text.split("→")]
257 if len(parts) == 3:
258 relation = parts[1].strip().lower()
259 if relation in ("informs", "supports", "opposes", "addresses"):
260 result["discourse_graph"]["edges"].append({
261 "source": parts[0],
262 "relation": relation,
263 "target": parts[2],
264 })
265
266 result["synthesis"] = result["synthesis"].strip()
267 return result
268
269
270def push_strata_record(island, strata_data):
271 """Create an org.latha.strata record on the PDS."""
272 # Build the lexmin from sorted vertices
273 lexmin = sorted(island["vertices"])[0]
274
275 # Map relationship lines to #connection format
276 # We need to match relationship text to vertex URIs
277 meta = island.get("vertexMeta", {})
278 title_to_uri = {}
279 for uri in island["vertices"]:
280 m = meta.get(uri, {})
281 title = m.get("title", "")
282 if title and len(title) > 3 and not title.startswith("http") and not title.startswith("at://"):
283 title_to_uri[title] = uri
284
285 connections = []
286 for rel_line in strata_data.get("relationships", []):
287 # Parse "Source → relation → Target"
288 parts = [p.strip() for p in rel_line.split("→")]
289 if len(parts) < 3:
290 continue
291 source_title = parts[0]
292 target_title = parts[-1]
293 relation_text = "→".join(parts[1:-1]).strip()
294
295 # Map relation text to lexicon tokens
296 relation_map = {
297 "leads to": "org.latha.strata#extends",
298 "supports": "org.latha.strata#supports",
299 "opposes": "org.latha.strata#contradicts",
300 "contradicts": "org.latha.strata#contradicts",
301 "extends": "org.latha.strata#extends",
302 "contextualizes": "org.latha.strata#contextualizes",
303 "exemplifies": "org.latha.strata#exemplifies",
304 "relates to": "org.latha.strata#contextualizes",
305 "supplements": "org.latha.strata#extends",
306 "explains": "org.latha.strata#contextualizes",
307 "cites": "org.latha.strata#contextualizes",
308 "addresses": "org.latha.strata#contextualizes",
309 }
310 relation = relation_map.get(relation_text.lower(), "org.latha.strata#contextualizes")
311
312 # Find target URI by fuzzy matching
313 target_uri = None
314 sorted_titles = sorted(title_to_uri.items(), key=lambda x: -len(x[0]))
315 for title, uri in sorted_titles:
316 if target_title in title or title in target_title:
317 target_uri = uri
318 break
319
320 if target_uri:
321 connections.append({
322 "description": rel_line,
323 "targetUri": target_uri,
324 "relation": relation,
325 })
326
327 # Build the AT record
328 from datetime import datetime, timezone
329 now = datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
330
331 record = {
332 "$type": "org.latha.strata",
333 "source": {
334 "uri": lexmin,
335 "collection": "network.cosmik.connection",
336 },
337 "analysis": {
338 "title": strata_data.get("title", ""),
339 "themes": strata_data.get("themes", []),
340 "connections": connections[:20],
341 "tensions": strata_data.get("tensions", []),
342 "openQuestions": strata_data.get("open_questions", []),
343 "synthesis": strata_data.get("synthesis", ""),
344 },
345 "createdAt": now,
346 }
347
348 # Push to PDS via atproto
349 script = f"""
350import json
351from atproto import Client
352
353client = Client()
354client.login(
355 {json.dumps(os.environ.get("BLUESKY_IDENTIFIER", "nandi.latha.org"))},
356 {json.dumps(os.environ.get("BLUESKY_APP_PASSWORD", ""))},
357)
358
359record = {json.dumps(record)}
360
361result = client.com.atproto.repo.create_record({{
362 "repo": client.me.did,
363 "collection": "org.latha.strata",
364 "record": record,
365}})
366print(json.dumps({{"uri": result.uri, "cid": result.cid}}))
367"""
368
369 try:
370 result = subprocess.run(
371 [VENV_PYTHON, "-c", script],
372 capture_output=True,
373 text=True,
374 timeout=30,
375 )
376 if result.returncode != 0:
377 print(f" PDS push failed: {result.stderr[:300]}", file=sys.stderr)
378 return None
379 data = json.loads(result.stdout.strip())
380 return data.get("uri")
381 except Exception as e:
382 print(f" PDS push error: {e}", file=sys.stderr)
383 return None
384
385
386def main():
387 force = "--force" in sys.argv
388 filter_island = None
389 for arg in sys.argv[1:]:
390 if arg.startswith("--island="):
391 filter_island = arg.split("=", 1)[1]
392
393 # Load credentials from swarm .env if not in env
394 env_path = "/home/nandi/code/swarm/.env"
395 if not os.environ.get("BLUESKY_APP_PASSWORD"):
396 with open(env_path) as f:
397 for line in f:
398 line = line.strip()
399 if "=" in line and not line.startswith("#"):
400 key, _, value = line.partition("=")
401 os.environ.setdefault(key.strip(), value.strip())
402
403 islands = fetch_islands()
404 total = len(islands)
405 print(f"Strata analysis: {total} islands")
406
407 for i, island in enumerate(islands):
408 island_id = island["id"]
409
410 if filter_island and filter_island not in island_id:
411 continue
412
413 if island.get("strata") and not force:
414 print(f"[{i+1}/{total}] Skip (has strata): {island_id[:60]}...")
415 continue
416
417 print(f"[{i+1}/{total}] Analyzing: {island_id[:60]}...")
418 context = build_island_context(island)
419 raw_prose = analyze_with_letta(context)
420
421 if not raw_prose:
422 print(" Skipping (no output)")
423 continue
424
425 strata_data = parse_strata(raw_prose)
426 if not strata_data:
427 print(" Skipping (parse failed)")
428 continue
429
430 print(f" Themes: {', '.join(strata_data['themes'][:5])}")
431 print(f" Relationships: {len(strata_data['relationships'])}")
432 print(f" Tensions: {len(strata_data['tensions'])}")
433 dg = strata_data.get("discourse_graph", {})
434 print(f" Discourse nodes: {len(dg.get('nodes', []))}, edges: {len(dg.get('edges', []))}")
435
436 uri = push_strata_record(island, strata_data)
437 if uri:
438 print(f" Pushed: {uri}")
439 else:
440 print(" Push failed.")
441
442 time.sleep(2)
443
444 print("Done.")
445
446
447if __name__ == "__main__":
448 main()