#!/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")