PlanOpticon

feat(graph): add shortest_path, clusters, and enhanced agentic queries Add BFS-based shortest path finding between entities and connected component detection for graph clustering. Wire path and clusters commands into CLI and agentic query mode.

lmata 2026-03-08 00:58 trunk
Commit 7bdcd8912bb89e5545df0b8c297f494a9eec070b010996b7e20077477a6c6917
Parent 6febc3f76decc47…
3 files changed +62 +10 -1 +137
~ tests/test_graph_query.py ~ video_processor/cli/commands.py ~ video_processor/integrators/graph_query.py
M tests/test_graph_query.py
+62
--- tests/test_graph_query.py
+++ tests/test_graph_query.py
@@ -271,5 +271,67 @@
271271
 mock_pm.chat.return_value = "I don't understand"
272272
 engine = GraphQueryEngine(store, provider_manager=mock_pm)
273273
 result = engine.ask("Gibberish?")
274274
 assert result.data is None
275275
 assert "parse" in result.explanation.lower() or "could not" in result.explanation.lower()
276
+
277
+
278
+class TestGraphAlgorithms:
279
+ def test_shortest_path(self):
280
+ store = InMemoryStore()
281
+ store.merge_entity("A", "concept", [])
282
+ store.merge_entity("B", "concept", [])
283
+ store.merge_entity("C", "concept", [])
284
+ store.add_relationship("A", "B", "connects")
285
+ store.add_relationship("B", "C", "connects")
286
+ engine = GraphQueryEngine(store)
287
+
288
+ result = engine.shortest_path("A", "C")
289
+ assert "Path found" in result.explanation
290
+ assert len(result.data) > 0
291
+
292
+ def test_shortest_path_same_entity(self):
293
+ store = InMemoryStore()
294
+ store.merge_entity("X", "concept", [])
295
+ engine = GraphQueryEngine(store)
296
+
297
+ result = engine.shortest_path("X", "X")
298
+ assert "same entity" in result.explanation.lower()
299
+
300
+ def test_shortest_path_not_found(self):
301
+ store = InMemoryStore()
302
+ store.merge_entity("A", "concept", [])
303
+ store.merge_entity("Z", "concept", [])
304
+ engine = GraphQueryEngine(store)
305
+
306
+ result = engine.shortest_path("A", "Z")
307
+ assert "No path found" in result.explanation
308
+
309
+ def test_shortest_path_entity_missing(self):
310
+ store = InMemoryStore()
311
+ engine = GraphQueryEngine(store)
312
+
313
+ result = engine.shortest_path("Missing", "Also Missing")
314
+ assert "not found" in result.explanation
315
+
316
+ def test_clusters(self):
317
+ store = InMemoryStore()
318
+ store.merge_entity("A", "concept", [])
319
+ store.merge_entity("B", "concept", [])
320
+ store.add_relationship("A", "B", "connected")
321
+
322
+ store.merge_entity("X", "concept", [])
323
+ store.merge_entity("Y", "concept", [])
324
+ store.add_relationship("X", "Y", "connected")
325
+
326
+ store.merge_entity("Lone", "concept", [])
327
+
328
+ engine = GraphQueryEngine(store)
329
+ result = engine.clusters()
330
+ assert "3 clusters" in result.explanation
331
+ assert result.data[0]["size"] == 2
332
+
333
+ def test_clusters_empty(self):
334
+ store = InMemoryStore()
335
+ engine = GraphQueryEngine(store)
336
+ result = engine.clusters()
337
+ assert result.data == []
276338
--- tests/test_graph_query.py
+++ tests/test_graph_query.py
@@ -271,5 +271,67 @@
271 mock_pm.chat.return_value = "I don't understand"
272 engine = GraphQueryEngine(store, provider_manager=mock_pm)
273 result = engine.ask("Gibberish?")
274 assert result.data is None
275 assert "parse" in result.explanation.lower() or "could not" in result.explanation.lower()
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
276
--- tests/test_graph_query.py
+++ tests/test_graph_query.py
@@ -271,5 +271,67 @@
271 mock_pm.chat.return_value = "I don't understand"
272 engine = GraphQueryEngine(store, provider_manager=mock_pm)
273 result = engine.ask("Gibberish?")
274 assert result.data is None
275 assert "parse" in result.explanation.lower() or "could not" in result.explanation.lower()
276
277
278 class TestGraphAlgorithms:
279 def test_shortest_path(self):
280 store = InMemoryStore()
281 store.merge_entity("A", "concept", [])
282 store.merge_entity("B", "concept", [])
283 store.merge_entity("C", "concept", [])
284 store.add_relationship("A", "B", "connects")
285 store.add_relationship("B", "C", "connects")
286 engine = GraphQueryEngine(store)
287
288 result = engine.shortest_path("A", "C")
289 assert "Path found" in result.explanation
290 assert len(result.data) > 0
291
292 def test_shortest_path_same_entity(self):
293 store = InMemoryStore()
294 store.merge_entity("X", "concept", [])
295 engine = GraphQueryEngine(store)
296
297 result = engine.shortest_path("X", "X")
298 assert "same entity" in result.explanation.lower()
299
300 def test_shortest_path_not_found(self):
301 store = InMemoryStore()
302 store.merge_entity("A", "concept", [])
303 store.merge_entity("Z", "concept", [])
304 engine = GraphQueryEngine(store)
305
306 result = engine.shortest_path("A", "Z")
307 assert "No path found" in result.explanation
308
309 def test_shortest_path_entity_missing(self):
310 store = InMemoryStore()
311 engine = GraphQueryEngine(store)
312
313 result = engine.shortest_path("Missing", "Also Missing")
314 assert "not found" in result.explanation
315
316 def test_clusters(self):
317 store = InMemoryStore()
318 store.merge_entity("A", "concept", [])
319 store.merge_entity("B", "concept", [])
320 store.add_relationship("A", "B", "connected")
321
322 store.merge_entity("X", "concept", [])
323 store.merge_entity("Y", "concept", [])
324 store.add_relationship("X", "Y", "connected")
325
326 store.merge_entity("Lone", "concept", [])
327
328 engine = GraphQueryEngine(store)
329 result = engine.clusters()
330 assert "3 clusters" in result.explanation
331 assert result.data[0]["size"] == 2
332
333 def test_clusters_empty(self):
334 store = InMemoryStore()
335 engine = GraphQueryEngine(store)
336 result = engine.clusters()
337 assert result.data == []
338
M video_processor/cli/commands.py
+10 -1
--- video_processor/cli/commands.py
+++ video_processor/cli/commands.py
@@ -851,11 +851,12 @@
851851
 @click.pass_context
852852
 def query(ctx, question, db_path, mode, output_format, interactive, provider, chat_model):
853853
 """Query a knowledge graph. Runs stats if no question given.
854854
855855
 Direct commands recognized in QUESTION: stats, entities, relationships,
856
- neighbors, sources, provenance, sql. Natural language questions use agentic mode.
856
+ neighbors, path, clusters, sources, provenance, sql.
857
+ Natural language questions use agentic mode.
857858
858859
 Examples:
859860
860861
 planopticon query
861862
 planopticon query stats
@@ -950,10 +951,18 @@
950951
951952
 if cmd == "provenance":
952953
 entity_name = " ".join(parts[1:]) if len(parts) > 1 else ""
953954
 return engine.provenance(entity_name)
954955
956
+ if cmd == "path":
957
+ if len(parts) < 3:
958
+ return engine.stats()
959
+ return engine.shortest_path(start=parts[1], end=parts[2])
960
+
961
+ if cmd == "clusters":
962
+ return engine.clusters()
963
+
955964
 if cmd == "sql":
956965
 sql_query = " ".join(parts[1:])
957966
 return engine.sql(sql_query)
958967
959968
 # Natural language → agentic (or fallback to entity search in direct mode)
960969
--- video_processor/cli/commands.py
+++ video_processor/cli/commands.py
@@ -851,11 +851,12 @@
851 @click.pass_context
852 def query(ctx, question, db_path, mode, output_format, interactive, provider, chat_model):
853 """Query a knowledge graph. Runs stats if no question given.
854
855 Direct commands recognized in QUESTION: stats, entities, relationships,
856 neighbors, sources, provenance, sql. Natural language questions use agentic mode.
 
857
858 Examples:
859
860 planopticon query
861 planopticon query stats
@@ -950,10 +951,18 @@
950
951 if cmd == "provenance":
952 entity_name = " ".join(parts[1:]) if len(parts) > 1 else ""
953 return engine.provenance(entity_name)
954
 
 
 
 
 
 
 
 
955 if cmd == "sql":
956 sql_query = " ".join(parts[1:])
957 return engine.sql(sql_query)
958
959 # Natural language → agentic (or fallback to entity search in direct mode)
960
--- video_processor/cli/commands.py
+++ video_processor/cli/commands.py
@@ -851,11 +851,12 @@
851 @click.pass_context
852 def query(ctx, question, db_path, mode, output_format, interactive, provider, chat_model):
853 """Query a knowledge graph. Runs stats if no question given.
854
855 Direct commands recognized in QUESTION: stats, entities, relationships,
856 neighbors, path, clusters, sources, provenance, sql.
857 Natural language questions use agentic mode.
858
859 Examples:
860
861 planopticon query
862 planopticon query stats
@@ -950,10 +951,18 @@
951
952 if cmd == "provenance":
953 entity_name = " ".join(parts[1:]) if len(parts) > 1 else ""
954 return engine.provenance(entity_name)
955
956 if cmd == "path":
957 if len(parts) < 3:
958 return engine.stats()
959 return engine.shortest_path(start=parts[1], end=parts[2])
960
961 if cmd == "clusters":
962 return engine.clusters()
963
964 if cmd == "sql":
965 sql_query = " ".join(parts[1:])
966 return engine.sql(sql_query)
967
968 # Natural language → agentic (or fallback to entity search in direct mode)
969
M video_processor/integrators/graph_query.py
+137
--- video_processor/integrators/graph_query.py
+++ video_processor/integrators/graph_query.py
@@ -310,10 +310,138 @@
310310
 data=locations,
311311
 query_type="filter",
312312
 raw_query=f"provenance({entity_name!r})",
313313
 explanation=f"Found {len(locations)} provenance records for '{entity_name}'",
314314
 )
315
+
316
+ def shortest_path(self, start: str, end: str, max_depth: int = 6) -> QueryResult:
317
+ """Find the shortest path between two entities via BFS."""
318
+ start_entity = self.store.get_entity(start)
319
+ end_entity = self.store.get_entity(end)
320
+ if not start_entity:
321
+ return QueryResult(
322
+ data=[],
323
+ query_type="filter",
324
+ raw_query=f"shortest_path({start!r}, {end!r})",
325
+ explanation=f"Entity '{start}' not found",
326
+ )
327
+ if not end_entity:
328
+ return QueryResult(
329
+ data=[],
330
+ query_type="filter",
331
+ raw_query=f"shortest_path({start!r}, {end!r})",
332
+ explanation=f"Entity '{end}' not found",
333
+ )
334
+
335
+ all_rels = self.store.get_all_relationships()
336
+ # Build adjacency list
337
+ adj: dict[str, list[tuple[str, dict]]] = {}
338
+ for rel in all_rels:
339
+ src_l = rel["source"].lower()
340
+ tgt_l = rel["target"].lower()
341
+ adj.setdefault(src_l, []).append((tgt_l, rel))
342
+ adj.setdefault(tgt_l, []).append((src_l, rel))
343
+
344
+ # BFS
345
+ start_l = start.lower()
346
+ end_l = end.lower()
347
+ if start_l == end_l:
348
+ return QueryResult(
349
+ data=[start_entity],
350
+ query_type="filter",
351
+ raw_query=f"shortest_path({start!r}, {end!r})",
352
+ explanation="Start and end are the same entity",
353
+ )
354
+
355
+ from collections import deque
356
+
357
+ queue: deque[tuple[str, list[dict]]] = deque([(start_l, [])])
358
+ visited = {start_l}
359
+
360
+ while queue:
361
+ current, path = queue.popleft()
362
+ if len(path) >= max_depth:
363
+ continue
364
+ for neighbor, rel in adj.get(current, []):
365
+ if neighbor in visited:
366
+ continue
367
+ new_path = path + [rel]
368
+ if neighbor == end_l:
369
+ # Build result: entities + relationships along path
370
+ path_entities = [start_entity]
371
+ for r in new_path:
372
+ path_entities.append(r)
373
+ tgt_name = r["target"] if r["source"].lower() == current else r["source"]
374
+ e = self.store.get_entity(tgt_name)
375
+ if e:
376
+ path_entities.append(e)
377
+ path_entities.append(end_entity)
378
+ # Deduplicate
379
+ seen = set()
380
+ deduped = []
381
+ for item in path_entities:
382
+ key = str(item)
383
+ if key not in seen:
384
+ seen.add(key)
385
+ deduped.append(item)
386
+ return QueryResult(
387
+ data=deduped,
388
+ query_type="filter",
389
+ raw_query=f"shortest_path({start!r}, {end!r})",
390
+ explanation=f"Path found: {len(new_path)} hops",
391
+ )
392
+ visited.add(neighbor)
393
+ queue.append((neighbor, new_path))
394
+
395
+ return QueryResult(
396
+ data=[],
397
+ query_type="filter",
398
+ raw_query=f"shortest_path({start!r}, {end!r})",
399
+ explanation=f"No path found between '{start}' and '{end}' within {max_depth} hops",
400
+ )
401
+
402
+ def clusters(self) -> QueryResult:
403
+ """Find connected components (clusters) in the graph."""
404
+ all_entities = self.store.get_all_entities()
405
+ all_rels = self.store.get_all_relationships()
406
+
407
+ # Build adjacency
408
+ adj: dict[str, set[str]] = {}
409
+ for e in all_entities:
410
+ adj.setdefault(e["name"].lower(), set())
411
+ for r in all_rels:
412
+ adj.setdefault(r["source"].lower(), set()).add(r["target"].lower())
413
+ adj.setdefault(r["target"].lower(), set()).add(r["source"].lower())
414
+
415
+ visited: set[str] = set()
416
+ components: list[list[str]] = []
417
+
418
+ for node in adj:
419
+ if node in visited:
420
+ continue
421
+ component: list[str] = []
422
+ stack = [node]
423
+ while stack:
424
+ n = stack.pop()
425
+ if n in visited:
426
+ continue
427
+ visited.add(n)
428
+ component.append(n)
429
+ stack.extend(adj.get(n, set()) - visited)
430
+ components.append(sorted(component))
431
+
432
+ # Sort by size descending
433
+ components.sort(key=len, reverse=True)
434
+
435
+ result = [{"cluster_id": i, "size": len(c), "members": c} for i, c in enumerate(components)]
436
+
437
+ return QueryResult(
438
+ data=result,
439
+ query_type="filter",
440
+ raw_query="clusters()",
441
+ explanation=f"Found {len(components)} clusters",
442
+ )
315443
316444
 def sql(self, query: str) -> QueryResult:
317445
 """Execute a raw SQL query (SQLite only)."""
318446
 result = self.store.raw_query(query)
319447
 return QueryResult(
@@ -350,10 +478,12 @@
350478
 f"Graph stats: {json.dumps(stats)}\n\n"
351479
 "Available actions (pick exactly one):\n"
352480
 '- {{"action": "entities", "name": "...", "entity_type": "..."}}\n'
353481
 '- {{"action": "relationships", "source": "...", "target": "...", "rel_type": "..."}}\n'
354482
 '- {{"action": "neighbors", "entity_name": "...", "depth": 1}}\n'
483
+ '- {{"action": "shortest_path", "start": "...", "end": "..."}}\n'
484
+ '- {{"action": "clusters"}}\n'
355485
 '- {{"action": "stats"}}\n\n'
356486
 f"User question: {question}\n\n"
357487
 "Return ONLY a JSON object with the action. Omit optional fields you don't need."
358488
 )
359489
@@ -398,10 +528,17 @@
398528
 elif action == "neighbors":
399529
 result = self.neighbors(
400530
 entity_name=plan.get("entity_name", ""),
401531
 depth=plan.get("depth", 1),
402532
 )
533
+ elif action == "shortest_path":
534
+ result = self.shortest_path(
535
+ start=plan.get("start", ""),
536
+ end=plan.get("end", ""),
537
+ )
538
+ elif action == "clusters":
539
+ result = self.clusters()
403540
 elif action == "stats":
404541
 result = self.stats()
405542
 else:
406543
 return QueryResult(
407544
 data=None,
408545
--- video_processor/integrators/graph_query.py
+++ video_processor/integrators/graph_query.py
@@ -310,10 +310,138 @@
310 data=locations,
311 query_type="filter",
312 raw_query=f"provenance({entity_name!r})",
313 explanation=f"Found {len(locations)} provenance records for '{entity_name}'",
314 )
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
315
316 def sql(self, query: str) -> QueryResult:
317 """Execute a raw SQL query (SQLite only)."""
318 result = self.store.raw_query(query)
319 return QueryResult(
@@ -350,10 +478,12 @@
350 f"Graph stats: {json.dumps(stats)}\n\n"
351 "Available actions (pick exactly one):\n"
352 '- {{"action": "entities", "name": "...", "entity_type": "..."}}\n'
353 '- {{"action": "relationships", "source": "...", "target": "...", "rel_type": "..."}}\n'
354 '- {{"action": "neighbors", "entity_name": "...", "depth": 1}}\n'
 
 
355 '- {{"action": "stats"}}\n\n'
356 f"User question: {question}\n\n"
357 "Return ONLY a JSON object with the action. Omit optional fields you don't need."
358 )
359
@@ -398,10 +528,17 @@
398 elif action == "neighbors":
399 result = self.neighbors(
400 entity_name=plan.get("entity_name", ""),
401 depth=plan.get("depth", 1),
402 )
 
 
 
 
 
 
 
403 elif action == "stats":
404 result = self.stats()
405 else:
406 return QueryResult(
407 data=None,
408
--- video_processor/integrators/graph_query.py
+++ video_processor/integrators/graph_query.py
@@ -310,10 +310,138 @@
310 data=locations,
311 query_type="filter",
312 raw_query=f"provenance({entity_name!r})",
313 explanation=f"Found {len(locations)} provenance records for '{entity_name}'",
314 )
315
316 def shortest_path(self, start: str, end: str, max_depth: int = 6) -> QueryResult:
317 """Find the shortest path between two entities via BFS."""
318 start_entity = self.store.get_entity(start)
319 end_entity = self.store.get_entity(end)
320 if not start_entity:
321 return QueryResult(
322 data=[],
323 query_type="filter",
324 raw_query=f"shortest_path({start!r}, {end!r})",
325 explanation=f"Entity '{start}' not found",
326 )
327 if not end_entity:
328 return QueryResult(
329 data=[],
330 query_type="filter",
331 raw_query=f"shortest_path({start!r}, {end!r})",
332 explanation=f"Entity '{end}' not found",
333 )
334
335 all_rels = self.store.get_all_relationships()
336 # Build adjacency list
337 adj: dict[str, list[tuple[str, dict]]] = {}
338 for rel in all_rels:
339 src_l = rel["source"].lower()
340 tgt_l = rel["target"].lower()
341 adj.setdefault(src_l, []).append((tgt_l, rel))
342 adj.setdefault(tgt_l, []).append((src_l, rel))
343
344 # BFS
345 start_l = start.lower()
346 end_l = end.lower()
347 if start_l == end_l:
348 return QueryResult(
349 data=[start_entity],
350 query_type="filter",
351 raw_query=f"shortest_path({start!r}, {end!r})",
352 explanation="Start and end are the same entity",
353 )
354
355 from collections import deque
356
357 queue: deque[tuple[str, list[dict]]] = deque([(start_l, [])])
358 visited = {start_l}
359
360 while queue:
361 current, path = queue.popleft()
362 if len(path) >= max_depth:
363 continue
364 for neighbor, rel in adj.get(current, []):
365 if neighbor in visited:
366 continue
367 new_path = path + [rel]
368 if neighbor == end_l:
369 # Build result: entities + relationships along path
370 path_entities = [start_entity]
371 for r in new_path:
372 path_entities.append(r)
373 tgt_name = r["target"] if r["source"].lower() == current else r["source"]
374 e = self.store.get_entity(tgt_name)
375 if e:
376 path_entities.append(e)
377 path_entities.append(end_entity)
378 # Deduplicate
379 seen = set()
380 deduped = []
381 for item in path_entities:
382 key = str(item)
383 if key not in seen:
384 seen.add(key)
385 deduped.append(item)
386 return QueryResult(
387 data=deduped,
388 query_type="filter",
389 raw_query=f"shortest_path({start!r}, {end!r})",
390 explanation=f"Path found: {len(new_path)} hops",
391 )
392 visited.add(neighbor)
393 queue.append((neighbor, new_path))
394
395 return QueryResult(
396 data=[],
397 query_type="filter",
398 raw_query=f"shortest_path({start!r}, {end!r})",
399 explanation=f"No path found between '{start}' and '{end}' within {max_depth} hops",
400 )
401
402 def clusters(self) -> QueryResult:
403 """Find connected components (clusters) in the graph."""
404 all_entities = self.store.get_all_entities()
405 all_rels = self.store.get_all_relationships()
406
407 # Build adjacency
408 adj: dict[str, set[str]] = {}
409 for e in all_entities:
410 adj.setdefault(e["name"].lower(), set())
411 for r in all_rels:
412 adj.setdefault(r["source"].lower(), set()).add(r["target"].lower())
413 adj.setdefault(r["target"].lower(), set()).add(r["source"].lower())
414
415 visited: set[str] = set()
416 components: list[list[str]] = []
417
418 for node in adj:
419 if node in visited:
420 continue
421 component: list[str] = []
422 stack = [node]
423 while stack:
424 n = stack.pop()
425 if n in visited:
426 continue
427 visited.add(n)
428 component.append(n)
429 stack.extend(adj.get(n, set()) - visited)
430 components.append(sorted(component))
431
432 # Sort by size descending
433 components.sort(key=len, reverse=True)
434
435 result = [{"cluster_id": i, "size": len(c), "members": c} for i, c in enumerate(components)]
436
437 return QueryResult(
438 data=result,
439 query_type="filter",
440 raw_query="clusters()",
441 explanation=f"Found {len(components)} clusters",
442 )
443
444 def sql(self, query: str) -> QueryResult:
445 """Execute a raw SQL query (SQLite only)."""
446 result = self.store.raw_query(query)
447 return QueryResult(
@@ -350,10 +478,12 @@
478 f"Graph stats: {json.dumps(stats)}\n\n"
479 "Available actions (pick exactly one):\n"
480 '- {{"action": "entities", "name": "...", "entity_type": "..."}}\n'
481 '- {{"action": "relationships", "source": "...", "target": "...", "rel_type": "..."}}\n'
482 '- {{"action": "neighbors", "entity_name": "...", "depth": 1}}\n'
483 '- {{"action": "shortest_path", "start": "...", "end": "..."}}\n'
484 '- {{"action": "clusters"}}\n'
485 '- {{"action": "stats"}}\n\n'
486 f"User question: {question}\n\n"
487 "Return ONLY a JSON object with the action. Omit optional fields you don't need."
488 )
489
@@ -398,10 +528,17 @@
528 elif action == "neighbors":
529 result = self.neighbors(
530 entity_name=plan.get("entity_name", ""),
531 depth=plan.get("depth", 1),
532 )
533 elif action == "shortest_path":
534 result = self.shortest_path(
535 start=plan.get("start", ""),
536 end=plan.get("end", ""),
537 )
538 elif action == "clusters":
539 result = self.clusters()
540 elif action == "stats":
541 result = self.stats()
542 else:
543 return QueryResult(
544 data=None,
545

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button