figmar
df148f5da8
查 Graphiti Neo4j → 标准 {nodes,edges} 供 graph-viewer 组件渲染; 六组group_id;
含部署配置(unit/nginx)与 viz.env.example(密钥走⟨secdb⟩不入库); .gitignore 排除 env/vendor/备份
208 lines
7.1 KiB
Python
208 lines
7.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Graphiti 时序知识图谱 · 可视化展示层
|
|
- FastAPI 单进程: /api/groups, /api/graph?group_id=&limit=, 静态前端(static/)
|
|
- 只读 Neo4j (graphiti-neo4j, bolt://127.0.0.1:7687), 绝不写库
|
|
- 输出契约对齐 DeepData KnowledgeGraphView: nodes[{id,labels,properties}], edges[{id,type,source,target,properties}]
|
|
"""
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, HTTPException, Query
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from neo4j import GraphDatabase
|
|
|
|
NEO4J_URI = os.environ.get("NEO4J_URI", "bolt://127.0.0.1:7687")
|
|
NEO4J_USER = os.environ.get("NEO4J_USER", "neo4j")
|
|
NEO4J_PASSWORD = os.environ.get("NEO4J_PASSWORD", "")
|
|
BIND_HOST = os.environ.get("VIZ_BIND", "127.0.0.1")
|
|
BIND_PORT = int(os.environ.get("VIZ_PORT", "8096"))
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
STATIC_DIR = BASE_DIR / "static"
|
|
|
|
GROUPS = ["infra", "customers", "org", "sop", "conversations", "knowledge"]
|
|
MAX_NODES, MAX_EDGES = 500, 800
|
|
|
|
app = FastAPI(title="Graphiti Viz", docs_url=None, redoc_url=None)
|
|
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
|
|
|
|
|
|
# ---------- entity_type 启发式推断 ----------
|
|
# labels 属性在 graphiti 里恒为 ["Entity"], 故从名称模式 + 情景共现上下文推断类型
|
|
def infer_type(name: str, ctx: dict) -> str:
|
|
"""ctx: {deleg, res, deg} — 情景共现统计"""
|
|
n = name.lower()
|
|
if n.startswith("agent_preset"): return "预设数字员工"
|
|
if n.startswith("agent_"): return "数字员工"
|
|
if n.startswith("apijob"): return "委托任务"
|
|
if n.startswith("session"): return "会话"
|
|
if n.startswith("channel"): return "渠道"
|
|
if n == "public_api": return "公共接口"
|
|
if n.startswith("http") or "://" in name or re.match(r"^[\w.-]+\.[a-z]{2,}(/|$)", n):
|
|
return "URL/域名"
|
|
if re.match(r"^\d+\.\d+\.\d+\.\d+", name):
|
|
return "IP地址"
|
|
if name.startswith("/") or name.startswith("~"):
|
|
return "文件路径"
|
|
if re.match(r"^[0-9a-f]{8}-[0-9a-f]{4}", n):
|
|
return "UUID标识"
|
|
if any(k in name for k in ("阅读笔记", "笔记", "手册", "章程", "方案")):
|
|
return "文档"
|
|
if "部署" in name or name.startswith(("服务-", "svc-")):
|
|
return "服务部署"
|
|
if re.match(r"^\d+-p$", n):
|
|
return "设备编号"
|
|
if n.endswith(".sh"):
|
|
return "脚本"
|
|
# 兜底: 用情景共现推断
|
|
deleg, res = ctx.get("deleg", 0), ctx.get("res", 0)
|
|
if deleg > 0 and deleg >= res:
|
|
return "客户业务对象"
|
|
if res > 0 and res > deleg:
|
|
return "基础设施对象"
|
|
if all(c.isascii() and (c.isalnum() or c in "_-.:") for c in name) and re.search(r"\d", name):
|
|
return "系统标识"
|
|
if not re.search(r"[A-Za-z]", name):
|
|
return "人物/组织"
|
|
if n.startswith("mag"):
|
|
return "人物/组织"
|
|
return "概念"
|
|
|
|
|
|
def _fmt(v):
|
|
return None if v is None else str(v)
|
|
|
|
|
|
# ---------- Cypher ----------
|
|
Q_GROUPS = """
|
|
UNWIND $groups AS g
|
|
OPTIONAL MATCH (n {group_id: g})
|
|
WITH g, count(n) AS nodes
|
|
OPTIONAL MATCH ()-[r]->() WHERE r.group_id = g
|
|
RETURN g, nodes, count(r) AS rels
|
|
ORDER BY g
|
|
"""
|
|
|
|
Q_ENTITIES = """
|
|
MATCH (e:Entity {group_id: $g})
|
|
RETURN e.uuid AS uuid, e.name AS name, e.labels AS labels,
|
|
e.summary AS summary, e.created_at AS created_at, e.group_id AS group_id
|
|
LIMIT $limit
|
|
"""
|
|
|
|
Q_EDGE_TYPES = """
|
|
MATCH (ep:Episodic {group_id: $g})-[m:MENTIONS]->(e:Entity)
|
|
WHERE e.uuid IN $uuids
|
|
WITH e.uuid AS u, count(m) AS deg,
|
|
sum(CASE WHEN ep.name STARTS WITH '委托-' THEN 1 ELSE 0 END) AS deleg,
|
|
sum(CASE WHEN ep.content CONTAINS '/' OR ep.content CONTAINS ':' THEN 1 ELSE 0 END) AS res
|
|
RETURN u, deg, deleg, res
|
|
"""
|
|
|
|
Q_EDGES = """
|
|
MATCH (a:Entity {group_id: $g})-[r:RELATES_TO]->(b:Entity {group_id: $g})
|
|
WHERE a.uuid IN $uuids AND b.uuid IN $uuids
|
|
RETURN elementId(r) AS id, r.name AS name, r.fact AS fact,
|
|
r.valid_at AS valid_at, r.invalid_at AS invalid_at,
|
|
r.created_at AS created_at, r.episodes AS episodes,
|
|
a.uuid AS src, b.uuid AS dst
|
|
LIMIT $limit
|
|
"""
|
|
|
|
|
|
@app.get("/api/groups")
|
|
def api_groups():
|
|
with driver.session(database="neo4j") as s:
|
|
rows = s.run(Q_GROUPS, groups=GROUPS).data()
|
|
erows = s.run(
|
|
"UNWIND $groups AS g MATCH (e:Entity {group_id:g}) RETURN g, count(e) AS c",
|
|
groups=GROUPS).data()
|
|
emap = {r["g"]: r["c"] for r in erows}
|
|
out = [{
|
|
"group_id": r["g"],
|
|
"nodes": r["nodes"], # 全部带 group_id 的节点(Entity+Episodic)
|
|
"edges": r["rels"], # 该组全部关系
|
|
"entities": emap.get(r["g"], 0),
|
|
} for r in rows]
|
|
return {"groups": out}
|
|
|
|
|
|
@app.get("/api/graph")
|
|
def api_graph(group_id: str = Query(...), limit: int = Query(200)):
|
|
if group_id not in GROUPS:
|
|
raise HTTPException(400, f"group_id must be one of {GROUPS}")
|
|
nl = max(10, min(int(limit), MAX_NODES))
|
|
el = max(10, min(int(limit * 1.5), MAX_EDGES))
|
|
|
|
with driver.session(database="neo4j") as s:
|
|
ents = s.run(Q_ENTITIES, g=group_id, limit=nl).data()
|
|
uuids = [e["uuid"] for e in ents if e["uuid"]]
|
|
ctx_map = {}
|
|
if uuids:
|
|
try:
|
|
for r in s.run(Q_EDGE_TYPES, g=group_id, uuids=uuids):
|
|
ctx_map[r["u"]] = {"deg": r["deg"], "deleg": r["deleg"], "res": r["res"]}
|
|
except Exception:
|
|
pass
|
|
edge_rows = s.run(Q_EDGES, g=group_id, uuids=uuids, limit=el).data() if uuids else []
|
|
|
|
nodes = []
|
|
for e in ents:
|
|
t = infer_type(e["name"] or "", ctx_map.get(e["uuid"], {}))
|
|
nodes.append({
|
|
"id": e["uuid"] or e["name"],
|
|
"labels": ["Entity", t],
|
|
"properties": {
|
|
"entity_type": t,
|
|
"description": e["summary"] or "",
|
|
"name": e["name"],
|
|
"created_at": _fmt(e["created_at"]),
|
|
"group_id": e["group_id"],
|
|
},
|
|
})
|
|
|
|
edges = []
|
|
for r in edge_rows:
|
|
edges.append({
|
|
"id": r["id"],
|
|
"type": r["name"] or "RELATES_TO",
|
|
"source": r["src"],
|
|
"target": r["dst"],
|
|
"properties": {
|
|
"keywords": r["fact"] or r["name"] or "",
|
|
"relation": r["name"],
|
|
"valid_at": _fmt(r["valid_at"]),
|
|
"invalid_at": _fmt(r["invalid_at"]),
|
|
"bi_temporal_invalid": bool(r["invalid_at"]),
|
|
"created_at": _fmt(r["created_at"]),
|
|
"episodes": r["episodes"] or [],
|
|
},
|
|
})
|
|
|
|
return {
|
|
"nodes": nodes,
|
|
"edges": edges,
|
|
"is_truncated": len(ents) >= nl or len(edge_rows) >= el,
|
|
"requested_node_limit": nl,
|
|
"requested_edge_limit": el,
|
|
}
|
|
|
|
|
|
@app.get("/healthz")
|
|
def healthz():
|
|
with driver.session(database="neo4j") as s:
|
|
s.run("RETURN 1").consume()
|
|
return {"ok": True}
|
|
|
|
|
|
# ---------- static frontend ----------
|
|
app.mount("/", StaticFiles(directory=str(STATIC_DIR), html=True), name="static")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host=BIND_HOST, port=BIND_PORT, log_level="info")
|