Create app.py

app.py

@@ -0,0 +1,266 @@
+import gradio as gr
+import os
+import json
+from dataclasses import dataclass, field, asdict
+from typing import List, Dict, Any
+
+# --- 1. 数据结构定义 (与之前相同) ---
+
+@dataclass
+class Entity:
+    name: str
+    entityType: str
+    observations: List[str] = field(default_factory=list)
+
+@dataclass
+class Relation:
+    from_entity: str
+    to_entity: str
+    relationType: str
+
+@dataclass
+class KnowledgeGraph:
+    entities: Dict[str, Entity] = field(default_factory=dict)
+    relations: List[Relation] = field(default_factory=list)
+
+
+# --- 2. 核心逻辑: KnowledgeGraphManager (修改为纯内存) ---
+
+class KnowledgeGraphManager:
+    """
+    这个版本的 Manager 将知识图谱完全保存在内存中。
+    服务器重启后，所有数据都会丢失。
+    """
+    def __init__(self):
+        # 服务器启动时，在内存中初始化一个空的知识图谱
+        self.graph = KnowledgeGraph()
+        print("Initialized a new, empty in-memory knowledge graph.")
+
+    def _reset_graph(self):
+        """提供一个方法来手动重置图谱，主要用于UI。"""
+        self.graph = KnowledgeGraph()
+        return "In-memory knowledge graph has been reset to an empty state."
+
+    def create_entities(self, entities_data: List[Dict]) -> List[Entity]:
+        new_entities = []
+        for e_data in entities_data:
+            if e_data['name'] not in self.graph.entities:
+                new_entity = Entity(**e_data)
+                self.graph.entities[e_data['name']] = new_entity
+                new_entities.append(new_entity)
+        return new_entities
+
+    def create_relations(self, relations_data: List[Dict]) -> List[Relation]:
+        new_relations = []
+        existing_relations_set = {
+            (r.from_entity, r.to_entity, r.relationType) for r in self.graph.relations
+        }
+        for r_data in relations_data:
+            r_data['from_entity'] = r_data.pop('from')
+            r_data['to_entity'] = r_data.pop('to')
+            if (r_data['from_entity'], r_data['to_entity'], r_data['relationType']) not in existing_relations_set:
+                new_relation = Relation(**r_data)
+                self.graph.relations.append(new_relation)
+                new_relations.append(new_relation)
+        return new_relations
+
+    def add_observations(self, observations_data: List[Dict]) -> List[Dict]:
+        results = []
+        for o_data in observations_data:
+            entity = self.graph.entities.get(o_data['entityName'])
+            if not entity:
+                raise ValueError(f"Entity with name {o_data['entityName']} not found")
+            
+            new_observations = [
+                obs for obs in o_data['contents'] if obs not in entity.observations
+            ]
+            entity.observations.extend(new_observations)
+            results.append({"entityName": entity.name, "addedObservations": new_observations})
+        return results
+
+    def delete_entities(self, entity_names: List[str]) -> None:
+        names_to_delete = set(entity_names)
+        self.graph.entities = {name: entity for name, entity in self.graph.entities.items() if name not in names_to_delete}
+        self.graph.relations = [
+            r for r in self.graph.relations 
+            if r.from_entity not in names_to_delete and r.to_entity not in names_to_delete
+        ]
+
+    def delete_observations(self, deletions_data: List[Dict]) -> None:
+        for d_data in deletions_data:
+            entity = self.graph.entities.get(d_data['entityName'])
+            if entity:
+                obs_to_delete = set(d_data['observations'])
+                entity.observations = [obs for obs in entity.observations if obs not in obs_to_delete]
+
+    def delete_relations(self, relations_data: List[Dict]) -> None:
+        relations_to_delete = {
+            (r.get('from'), r.get('to'), r.get('relationType')) for r in relations_data
+        }
+        self.graph.relations = [
+            r for r in self.graph.relations 
+            if (r.from_entity, r.to_entity, r.relationType) not in relations_to_delete
+        ]
+        
+    def read_graph(self) -> Dict:
+        return {
+            "entities": [asdict(e) for e in self.graph.entities.values()],
+            "relations": [
+                {"from": r.from_entity, "to": r.to_entity, "relationType": r.relationType} 
+                for r in self.graph.relations
+            ]
+        }
+        
+    def search_nodes(self, query: str) -> Dict:
+        query_lower = query.lower()
+        
+        filtered_entities = [
+            e for e in self.graph.entities.values() if
+            query_lower in e.name.lower() or
+            query_lower in e.entityType.lower() or
+            any(query_lower in obs.lower() for obs in e.observations)
+        ]
+        
+        filtered_entity_names = {e.name for e in filtered_entities}
+        
+        filtered_relations = [
+            r for r in self.graph.relations if
+            r.from_entity in filtered_entity_names and r.to_entity in filtered_entity_names
+        ]
+        
+        return {
+            "entities": [asdict(e) for e in filtered_entities],
+            "relations": [
+                {"from": r.from_entity, "to": r.to_entity, "relationType": r.relationType} 
+                for r in filtered_relations
+            ]
+        }
+        
+    def open_nodes(self, names: List[str]) -> Dict:
+        names_set = set(names)
+        
+        filtered_entities = [e for name, e in self.graph.entities.items() if name in names_set]
+        filtered_entity_names = {e.name for e in filtered_entities}
+        filtered_relations = [
+            r for r in self.graph.relations if
+            r.from_entity in filtered_entity_names and r.to_entity in filtered_entity_names
+        ]
+
+        return {
+            "entities": [asdict(e) for e in filtered_entities],
+            "relations": [
+                {"from": r.from_entity, "to": r.to_entity, "relationType": r.relationType} 
+                for r in filtered_relations
+            ]
+        }
+
+# 初始化一个全局的、单例的 Manager 实例
+# 所有API调用都将共享这一个实例
+kg_manager = KnowledgeGraphManager()
+
+
+# --- 3. Gradio MCP 工具函数定义 (与之前相同, 只调用 manager) ---
+# ... (所有工具函数如 create_entities, read_graph 等保持不变) ...
+def create_entities(entities: List[Dict]) -> str:
+    try:
+        new_entities = kg_manager.create_entities(entities)
+        return json.dumps([asdict(e) for e in new_entities], indent=2)
+    except Exception as e: return f"Error: {e}"
+def create_relations(relations: List[Dict]) -> str:
+    try:
+        new_relations = kg_manager.create_relations(relations)
+        return json.dumps([{"from": r.from_entity, "to": r.to_entity, "relationType": r.relationType} for r in new_relations], indent=2)
+    except Exception as e: return f"Error: {e}"
+def add_observations(observations: List[Dict]) -> str:
+    try:
+        results = kg_manager.add_observations(observations)
+        return json.dumps(results, indent=2)
+    except Exception as e: return f"Error: {e}"
+def delete_entities(entityNames: List[str]) -> str:
+    try:
+        kg_manager.delete_entities(entityNames)
+        return "Entities and their relations deleted successfully."
+    except Exception as e: return f"Error: {e}"
+def delete_observations(deletions: List[Dict]) -> str:
+    try:
+        kg_manager.delete_observations(deletions)
+        return "Observations deleted successfully."
+    except Exception as e: return f"Error: {e}"
+def delete_relations(relations: List[Dict]) -> str:
+    try:
+        kg_manager.delete_relations(relations)
+        return "Relations deleted successfully."
+    except Exception as e: return f"Error: {e}"
+def read_graph() -> Dict: return kg_manager.read_graph()
+def search_nodes(query: str) -> Dict: return kg_manager.search_nodes(query)
+def open_nodes(names: List[str]) -> Dict: return kg_manager.open_nodes(names)
+def reset_graph_ui() -> Tuple[str, Dict]:
+    message = kg_manager._reset_graph()
+    empty_graph = kg_manager.read_graph()
+    return message, empty_graph
+
+
+# --- 4. Gradio 界面构建 (略作调整) ---
+
+with gr.Blocks(theme=gr.themes.Soft()) as app:
+    gr.Markdown("# 知识图谱记忆工具 (纯内存模式)")
+    gr.Markdown("所有数据都保存在服务器内存中。**关闭或重启服务器将清空所有数据。**")
+
+    with gr.Tabs():
+        # --- 查看/搜索 Tab ---
+        with gr.TabItem("查看与搜索"):
+            graph_output = gr.JSON(label="知识图谱内容")
+            with gr.Row():
+                read_btn = gr.Button("读取/刷新图谱")
+                reset_btn = gr.Button("清空并重置图谱", variant="stop")
+            with gr.Row():
+                search_query = gr.Textbox(label="搜索查询", placeholder="输入实体名称、类型或观察内容中的关键词...")
+                search_btn = gr.Button("搜索节点")
+            with gr.Row():
+                open_names = gr.Textbox(label="打开节点", placeholder="输入实体名称，用逗号分隔...")
+                open_btn = gr.Button("打开指定节点")
+        
+        # ... 其他Tab (创建/添加, 删除) 保持不变 ...
+        with gr.TabItem("创建与添加"):
+            status_output_create = gr.Textbox(label="操作结果", interactive=False)
+            with gr.Accordion("创建实体", open=True):
+                create_entities_input = gr.Code(label="实体 (JSON格式)", language="json", value='[\n  {\n    "name": "Claude 3",\n    "entityType": "AI Model",\n    "observations": ["Developed by Anthropic", "Is a large language model"]\n  }\n]')
+                create_entities_btn = gr.Button("创建实体")
+            with gr.Accordion("创建关系", open=False):
+                create_relations_input = gr.Code(label="关系 (JSON格式)", language="json", value='[\n  {\n    "from": "Claude 3",\n    "to": "Anthropic",\n    "relationType": "developed by"\n  }\n]')
+                create_relations_btn = gr.Button("创建关系")
+            with gr.Accordion("添加观察记录", open=False):
+                add_obs_input = gr.Code(label="观察记录 (JSON格式)", language="json", value='[\n  {\n    "entityName": "Claude 3",\n    "contents": ["Released in 2024"]\n  }\n]')
+                add_obs_btn = gr.Button("添加观察记录")
+        with gr.TabItem("删除"):
+            status_output_delete = gr.Textbox(label="操作结果", interactive=False)
+            with gr.Accordion("删除实体", open=True):
+                delete_entities_input = gr.Textbox(label="实体名称 (逗号分隔)", placeholder="Claude 3,Anthropic")
+                delete_entities_btn = gr.Button("删除实体")
+            with gr.Accordion("删除关系", open=False):
+                delete_relations_input = gr.Code(label="关系 (JSON格式)", language="json", value='[\n  {\n    "from": "Claude 3",\n    "to": "Anthropic",\n    "relationType": "developed by"\n  }\n]')
+                delete_relations_btn = gr.Button("删除关系")
+            with gr.Accordion("删除观察记录", open=False):
+                delete_obs_input = gr.Code(label="要删除的观察记录 (JSON格式)", language="json", value='[\n  {\n    "entityName": "Claude 3",\n    "observations": ["Released in 2024"]\n  }\n]')
+                delete_obs_btn = gr.Button("删除观察记录")
+
+
+    # --- 5. 事件与API绑定 (与之前相同) ---
+    # ... (所有 .click 事件绑定保持不变, 添加 reset_btn 的绑定) ...
+    read_btn.click(fn=read_graph, outputs=graph_output, api_name="read_graph")
+    search_btn.click(fn=search_nodes, inputs=search_query, outputs=graph_output, api_name="search_nodes")
+    open_btn.click(fn=lambda names_str: open_nodes(names=[name.strip() for name in names_str.split(',')]), inputs=open_names, outputs=graph_output, api_name="open_nodes")
+    create_entities_btn.click(fn=lambda entities_json: create_entities(json.loads(entities_json)), inputs=create_entities_input, outputs=status_output_create, api_name="create_entities")
+    create_relations_btn.click(fn=lambda relations_json: create_relations(json.loads(relations_json)), inputs=create_relations_input, outputs=status_output_create, api_name="create_relations")
+    add_obs_btn.click(fn=lambda obs_json: add_observations(json.loads(obs_json)), inputs=add_obs_input, outputs=status_output_create, api_name="add_observations")
+    delete_entities_btn.click(fn=lambda names_str: delete_entities(entityNames=[name.strip() for name in names_str.split(',')]), inputs=delete_entities_input, outputs=status_output_delete, api_name="delete_entities")
+    delete_relations_btn.click(fn=lambda relations_json: delete_relations(json.loads(relations_json)), inputs=delete_relations_input, outputs=status_output_delete, api_name="delete_relations")
+    delete_obs_btn.click(fn=lambda obs_json: delete_observations(json.loads(obs_json)), inputs=delete_obs_input, outputs=status_output_delete, api_name="delete_observations")
+    
+    # UI-only reset button
+    # We need a status output for the reset button, let's reuse one.
+    reset_btn.click(fn=reset_graph_ui, outputs=[status_output_create, graph_output])
+
+
+if __name__ == "__main__":
+    app.launch(mcp_server=True)