Update app.py
Browse files
app.py
CHANGED
|
@@ -1,7 +1,7 @@
|
|
| 1 |
import gradio as gr
|
| 2 |
import json
|
| 3 |
from dataclasses import dataclass, field, asdict
|
| 4 |
-
from typing import List, Dict, Any, Tuple, TypedDict # <--- 导入
|
| 5 |
|
| 6 |
# --- 1. 数据结构与核心逻辑 (无变化) ---
|
| 7 |
@dataclass
|
|
@@ -41,51 +41,43 @@ class KnowledgeGraphManager:
|
|
| 41 |
|
| 42 |
kg_manager = KnowledgeGraphManager()
|
| 43 |
|
| 44 |
-
# --- 2. 为每个工具的 payload 定义 TypedDict ---
|
| 45 |
-
class CreateEntitiesPayload(TypedDict):
|
| 46 |
-
|
| 47 |
-
class
|
| 48 |
-
|
| 49 |
-
class
|
| 50 |
-
|
| 51 |
-
class
|
| 52 |
-
|
| 53 |
-
|
| 54 |
-
|
| 55 |
-
|
| 56 |
-
relations: List[Dict]
|
| 57 |
-
class SearchNodesPayload(TypedDict):
|
| 58 |
-
query: str
|
| 59 |
-
class OpenNodesPayload(TypedDict):
|
| 60 |
-
names: List[str]
|
| 61 |
-
|
| 62 |
-
# --- 3. API 工具函数定义 (使用 TypedDict 更新了函数签名) ---
|
| 63 |
-
def create_entities(payload: CreateEntitiesPayload) -> str:
|
| 64 |
"""在知识图谱中创建多个新实体。"""
|
| 65 |
new_entities = kg_manager.create_entities(payload['entities'])
|
| 66 |
return json.dumps([asdict(e) for e in new_entities], indent=2)
|
| 67 |
|
| 68 |
-
def create_relations(payload: CreateRelationsPayload) -> str:
|
| 69 |
"""在知识图谱中的实体之间创建多个新关系。"""
|
| 70 |
new_relations = kg_manager.create_relations(payload['relations'])
|
| 71 |
return json.dumps([{"from": r.from_entity, "to": r.to_entity, "relationType": r.relationType} for r in new_relations], indent=2)
|
| 72 |
|
| 73 |
-
def add_observations(payload: AddObservationsPayload) -> str:
|
| 74 |
"""向知识图谱中已存在的实体添加新的观察记录。"""
|
| 75 |
results = kg_manager.add_observations(payload['observations'])
|
| 76 |
return json.dumps(results, indent=2)
|
| 77 |
|
| 78 |
-
def delete_entities(payload: DeleteEntitiesPayload) -> str:
|
| 79 |
"""从知识图谱中删除多个实体及其相关联的关系。"""
|
| 80 |
kg_manager.delete_entities(payload['entityNames'])
|
| 81 |
return "Entities and their relations deleted successfully."
|
| 82 |
|
| 83 |
-
def delete_observations(payload: DeleteObservationsPayload) -> str:
|
| 84 |
"""从知识图谱中的实体删除特定的观察记录。"""
|
| 85 |
kg_manager.delete_observations(payload['deletions'])
|
| 86 |
return "Observations deleted successfully."
|
| 87 |
|
| 88 |
-
def delete_relations(payload: DeleteRelationsPayload) -> str:
|
| 89 |
"""从知识图谱中删除多个关系。"""
|
| 90 |
kg_manager.delete_relations(payload['relations'])
|
| 91 |
return "Relations deleted successfully."
|
|
@@ -94,23 +86,19 @@ def read_graph() -> Dict:
|
|
| 94 |
"""读取并返回整个知识图谱。此函数不需要参数。"""
|
| 95 |
return kg_manager.read_graph()
|
| 96 |
|
| 97 |
-
def search_nodes(payload: SearchNodesPayload) -> Dict:
|
| 98 |
"""根据查询词在知识图谱中搜索节点。"""
|
| 99 |
return kg_manager.search_nodes(payload['query'])
|
| 100 |
|
| 101 |
-
def open_nodes(payload: OpenNodesPayload) -> Dict:
|
| 102 |
"""通过名称打开知识图谱中的特定节点。"""
|
| 103 |
return kg_manager.open_nodes(payload['names'])
|
| 104 |
|
| 105 |
-
# --- 4. Gradio Headless 应用构建与 API 注册 ---
|
| 106 |
with gr.Blocks() as app:
|
| 107 |
gr.Markdown("MCP Server is running. This UI is intentionally blank.", visible=True)
|
| 108 |
-
|
| 109 |
with gr.Row(visible=False):
|
| 110 |
-
json_input = gr.JSON(value={})
|
| 111 |
-
generic_output = gr.JSON()
|
| 112 |
-
dummy_btn = gr.Button()
|
| 113 |
-
|
| 114 |
dummy_btn.click(fn=create_entities, inputs=json_input, outputs=generic_output, api_name="create_entities")
|
| 115 |
dummy_btn.click(fn=create_relations, inputs=json_input, outputs=generic_output, api_name="create_relations")
|
| 116 |
dummy_btn.click(fn=add_observations, inputs=json_input, outputs=generic_output, api_name="add_observations")
|
|
|
|
| 1 |
import gradio as gr
|
| 2 |
import json
|
| 3 |
from dataclasses import dataclass, field, asdict
|
| 4 |
+
from typing import List, Dict, Any, Tuple, TypedDict, Annotated # <--- 导入 Annotated
|
| 5 |
|
| 6 |
# --- 1. 数据结构与核心逻辑 (无变化) ---
|
| 7 |
@dataclass
|
|
|
|
| 41 |
|
| 42 |
kg_manager = KnowledgeGraphManager()
|
| 43 |
|
| 44 |
+
# --- 2. 为每个工具的 payload 定义 TypedDict (无变化) ---
|
| 45 |
+
class CreateEntitiesPayload(TypedDict): entities: List[Dict]
|
| 46 |
+
class CreateRelationsPayload(TypedDict): relations: List[Dict]
|
| 47 |
+
class AddObservationsPayload(TypedDict): observations: List[Dict]
|
| 48 |
+
class DeleteEntitiesPayload(TypedDict): entityNames: List[str]
|
| 49 |
+
class DeleteObservationsPayload(TypedDict): deletions: List[Dict]
|
| 50 |
+
class DeleteRelationsPayload(TypedDict): relations: List[Dict]
|
| 51 |
+
class SearchNodesPayload(TypedDict): query: str
|
| 52 |
+
class OpenNodesPayload(TypedDict): names: List[str]
|
| 53 |
+
|
| 54 |
+
# --- 3. API 工具函数定义 (使用 Annotated[TypedDict, "description"]) ---
|
| 55 |
+
def create_entities(payload: Annotated[CreateEntitiesPayload, "一个包含'entities'键的JSON对象。'entities'的值是一个实体对象的列表。"]) -> str:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 56 |
"""在知识图谱中创建多个新实体。"""
|
| 57 |
new_entities = kg_manager.create_entities(payload['entities'])
|
| 58 |
return json.dumps([asdict(e) for e in new_entities], indent=2)
|
| 59 |
|
| 60 |
+
def create_relations(payload: Annotated[CreateRelationsPayload, "一个包含'relations'键的JSON对象。'relations'的值是一个关系对象的列表。"]) -> str:
|
| 61 |
"""在知识图谱中的实体之间创建多个新关系。"""
|
| 62 |
new_relations = kg_manager.create_relations(payload['relations'])
|
| 63 |
return json.dumps([{"from": r.from_entity, "to": r.to_entity, "relationType": r.relationType} for r in new_relations], indent=2)
|
| 64 |
|
| 65 |
+
def add_observations(payload: Annotated[AddObservationsPayload, "一个包含'observations'键的JSON对象。'observations'的值是一个观察记录对象的列表。"]) -> str:
|
| 66 |
"""向知识图谱中已存在的实体添加新的观察记录。"""
|
| 67 |
results = kg_manager.add_observations(payload['observations'])
|
| 68 |
return json.dumps(results, indent=2)
|
| 69 |
|
| 70 |
+
def delete_entities(payload: Annotated[DeleteEntitiesPayload, "一个包含'entityNames'键的JSON对象。'entityNames'的值是一个实体名称的列表。"]) -> str:
|
| 71 |
"""从知识图谱中删除多个实体及其相关联的关系。"""
|
| 72 |
kg_manager.delete_entities(payload['entityNames'])
|
| 73 |
return "Entities and their relations deleted successfully."
|
| 74 |
|
| 75 |
+
def delete_observations(payload: Annotated[DeleteObservationsPayload, "一个包含'deletions'键的JSON对象。'deletions'的值是一个删除指令的列表。"]) -> str:
|
| 76 |
"""从知识图谱中的实体删除特定的观察记录。"""
|
| 77 |
kg_manager.delete_observations(payload['deletions'])
|
| 78 |
return "Observations deleted successfully."
|
| 79 |
|
| 80 |
+
def delete_relations(payload: Annotated[DeleteRelationsPayload, "一个包含'relations'键的JSON对象。'relations'的值是一个要删除的关系对象的列表。"]) -> str:
|
| 81 |
"""从知识图谱中删除多个关系。"""
|
| 82 |
kg_manager.delete_relations(payload['relations'])
|
| 83 |
return "Relations deleted successfully."
|
|
|
|
| 86 |
"""读取并返回整个知识图谱。此函数不需要参数。"""
|
| 87 |
return kg_manager.read_graph()
|
| 88 |
|
| 89 |
+
def search_nodes(payload: Annotated[SearchNodesPayload, "一个包含'query'键的JSON对象。'query'的值是用于搜索的字符串。"]) -> Dict:
|
| 90 |
"""根据查询词在知识图谱中搜索节点。"""
|
| 91 |
return kg_manager.search_nodes(payload['query'])
|
| 92 |
|
| 93 |
+
def open_nodes(payload: Annotated[OpenNodesPayload, "一个包含'names'键的JSON对象。'names'的值是一个要检索的实体名称的列表。"]) -> Dict:
|
| 94 |
"""通过名称打开知识图谱中的特定节点。"""
|
| 95 |
return kg_manager.open_nodes(payload['names'])
|
| 96 |
|
| 97 |
+
# --- 4. Gradio Headless 应用构建与 API 注册 (无变化) ---
|
| 98 |
with gr.Blocks() as app:
|
| 99 |
gr.Markdown("MCP Server is running. This UI is intentionally blank.", visible=True)
|
|
|
|
| 100 |
with gr.Row(visible=False):
|
| 101 |
+
json_input = gr.JSON(value={}); generic_output = gr.JSON(); dummy_btn = gr.Button()
|
|
|
|
|
|
|
|
|
|
| 102 |
dummy_btn.click(fn=create_entities, inputs=json_input, outputs=generic_output, api_name="create_entities")
|
| 103 |
dummy_btn.click(fn=create_relations, inputs=json_input, outputs=generic_output, api_name="create_relations")
|
| 104 |
dummy_btn.click(fn=add_observations, inputs=json_input, outputs=generic_output, api_name="add_observations")
|