ankn
a5f8fb49ab
Bahn: aisupport, Analyse-O2C-C2S, awesome-bahn-mcp-servers, beam-mcp,
Confluence_Bot, db-planet-mcp-server, O2C-Harness, project-audit,
Projekt-KIQ-HP, teamlandkarte-mcp
Dhive: Jury-Voting
Privat: CV, NoteGraph (NOTE: NoteGraph needs complete redo after consolidation)
Shared: AI-Orchestrator, OrgMyLife, power_skills_and_more
Shared/references: symphony (read-only)
Bahn repos remain available as independent remotes - this monorepo
pulls them in via subtree, the originals are untouched.
90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
"""
|
|
Query - Kontextsuche im Knowledge Graph
|
|
Wird als Skill genutzt um relevanten Kontext zu finden.
|
|
|
|
Usage:
|
|
python project-audit/knowledge/query.py "Suchbegriff"
|
|
python project-audit/knowledge/query.py --node "team-404"
|
|
python project-audit/knowledge/query.py --facts "Spring Boot"
|
|
"""
|
|
import sys
|
|
import os
|
|
import json
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
|
|
from db import get_db, search_summaries, search_facts, get_node_context
|
|
|
|
|
|
def format_results(summaries, facts, node_context=None):
|
|
"""Formatiert Ergebnisse als kompakten Kontext-String."""
|
|
output = []
|
|
|
|
if node_context:
|
|
n = node_context["node"]
|
|
output.append(f"=== {n['name']} ({n['type']}) ===")
|
|
if n["summary"]:
|
|
output.append(f" {n['summary']}")
|
|
if n["properties"]:
|
|
props = json.loads(n["properties"]) if isinstance(n["properties"], str) else n["properties"]
|
|
for k, v in props.items():
|
|
output.append(f" {k}: {v}")
|
|
if node_context["outgoing"]:
|
|
output.append(" Beziehungen (ausgehend):")
|
|
for e in node_context["outgoing"]:
|
|
output.append(f" -> {e['relation']} -> {e['name']} ({e['type']})")
|
|
if node_context["incoming"]:
|
|
output.append(" Beziehungen (eingehend):")
|
|
for e in node_context["incoming"]:
|
|
output.append(f" <- {e['relation']} <- {e['name']} ({e['type']})")
|
|
output.append("")
|
|
|
|
if summaries:
|
|
output.append("=== Relevante Zusammenfassungen ===")
|
|
for s in summaries:
|
|
output.append(f"[{s['topic']}] {s['content']}")
|
|
if s.get("detail_path"):
|
|
output.append(f" -> Detail: {s['detail_path']}")
|
|
output.append("")
|
|
|
|
if facts:
|
|
output.append("=== Relevante Fakten ===")
|
|
for f in facts:
|
|
conf = f" ({f['confidence']})" if f["confidence"] != "confirmed" else ""
|
|
date = f" [{f['date']}]" if f.get("date") else ""
|
|
output.append(f" {f['subject']}.{f['predicate']} = {f['value']}{conf}{date}")
|
|
|
|
return "\n".join(output)
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print("Usage: python query.py <suchbegriff>")
|
|
print(" python query.py --node <node-id>")
|
|
print(" python query.py --facts <subject>")
|
|
sys.exit(1)
|
|
|
|
db = get_db()
|
|
|
|
if sys.argv[1] == "--node" and len(sys.argv) > 2:
|
|
node_id = sys.argv[2]
|
|
ctx = get_node_context(db, node_id)
|
|
if ctx:
|
|
print(format_results([], [], ctx))
|
|
else:
|
|
print(f"Node '{node_id}' nicht gefunden.")
|
|
elif sys.argv[1] == "--facts" and len(sys.argv) > 2:
|
|
query = " ".join(sys.argv[2:])
|
|
facts = search_facts(db, query, limit=15)
|
|
print(format_results([], facts))
|
|
else:
|
|
query = " ".join(sys.argv[1:])
|
|
summaries = search_summaries(db, query, limit=5)
|
|
facts = search_facts(db, query, limit=10)
|
|
print(format_results(summaries, facts))
|
|
|
|
db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|