play7284 commited on
Commit
c50c4ad
·
verified ·
1 Parent(s): 54f551c

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +21 -80
app.py CHANGED
@@ -1,9 +1,9 @@
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
8
  class Entity: name: str; entityType: str; observations: List[str] = field(default_factory=list)
9
  @dataclass
@@ -12,7 +12,6 @@ class Relation: from_entity: str; to_entity: str; relationType: str
12
  class KnowledgeGraph: entities: Dict[str, Entity] = field(default_factory=dict); relations: List[Relation] = field(default_factory=list)
13
 
14
  class KnowledgeGraphManager:
15
- # ... (Manager 的所有内部方法保持不变) ...
16
  def __init__(self): self.graph = KnowledgeGraph(); print("Initialized a new, empty in-memory knowledge graph.")
17
  def create_entities(self, entities_data: List[Dict]) -> List[Entity]:
18
  new_entities = []; [ (new_entity := Entity(**e_data), self.graph.entities.update({e_data['name']: new_entity}), new_entities.append(new_entity)) for e_data in entities_data if e_data['name'] not in self.graph.entities]; return new_entities
@@ -37,7 +36,6 @@ class KnowledgeGraphManager:
37
 
38
  kg_manager = KnowledgeGraphManager()
39
 
40
- # --- 2. 为每个工具的 payload 定义 TypedDict (无变化) ---
41
  class CreateEntitiesPayload(TypedDict): entities: List[Dict]
42
  class CreateRelationsPayload(TypedDict): relations: List[Dict]
43
  class AddObservationsPayload(TypedDict): observations: List[Dict]
@@ -47,82 +45,44 @@ class DeleteRelationsPayload(TypedDict): relations: List[Dict]
47
  class SearchNodesPayload(TypedDict): query: str
48
  class OpenNodesPayload(TypedDict): names: List[str]
49
 
50
- # --- 3. API 工具函数定义 (同时使用 TypedDict 和 完整 Docstring) ---
51
- def create_entities(payload: CreateEntitiesPayload) -> str:
52
- """在知识图谱中创建多个新实体。
53
-
54
- Args:
55
- payload (CreateEntitiesPayload): 一个包含'entities'键的JSON对象。'entities'的值是一个实体对象的列表。
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: CreateRelationsPayload) -> str:
61
- """在知识图谱中的实体之间创建多个新关系。
62
-
63
- Args:
64
- payload (CreateRelationsPayload): 一个包含'relations'键的JSON对象。'relations'的值是一个关系对象的列表。
65
- """
66
  new_relations = kg_manager.create_relations(payload['relations'])
67
  return json.dumps([{"from": r.from_entity, "to": r.to_entity, "relationType": r.relationType} for r in new_relations], indent=2)
68
 
69
- def add_observations(payload: AddObservationsPayload) -> str:
70
- """向知识图谱中已存在实体添加新的观察记录。
71
-
72
- Args:
73
- payload (AddObservationsPayload): 一个包含'observations'键的JSON对象。'observations'的值是一个观察记录对象的列表。
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
-
81
- Args:
82
- payload (DeleteEntitiesPayload): 一个包含'entityNames'键的JSON对象。'entityNames'的值是一个实体名称的列表。
83
- """
84
  kg_manager.delete_entities(payload['entityNames'])
85
  return "Entities and their relations deleted successfully."
86
-
87
- def delete_observations(payload: DeleteObservationsPayload) -> str:
88
- """从知识图谱中的实体删除特定的观察记录。
89
-
90
- Args:
91
- payload (DeleteObservationsPayload): 一个包含'deletions'键的JSON对象。'deletions'的值是一个删除指令的列表。
92
- """
93
  kg_manager.delete_observations(payload['deletions'])
94
  return "Observations deleted successfully."
95
-
96
- def delete_relations(payload: DeleteRelationsPayload) -> str:
97
- """从知识图谱中删除多个关系。
98
-
99
- Args:
100
- payload (DeleteRelationsPayload): 一个包含'relations'键的JSON对象。'relations'的值是一个要删除的关系对象的列表。
101
- """
102
  kg_manager.delete_relations(payload['relations'])
103
  return "Relations deleted successfully."
104
-
105
  def read_graph() -> Dict:
106
  """读取并返回整个知识图谱。此函数不需要参数。"""
107
  return kg_manager.read_graph()
108
-
109
- def search_nodes(payload: SearchNodesPayload) -> Dict:
110
- """根据查询词在知识图谱中搜索节点。
111
-
112
- Args:
113
- payload (SearchNodesPayload): 一个包含'query'键的JSON对象。'query'的值是用于搜索的字符串。
114
- """
115
  return kg_manager.search_nodes(payload['query'])
116
-
117
- def open_nodes(payload: OpenNodesPayload) -> Dict:
118
- """通过名称打开知识图谱中的特定节点。
119
-
120
- Args:
121
- payload (OpenNodesPayload): 一个包含'names'键的JSON对象。'names'的值是一个要检索的实体名称的列表。
122
- """
123
  return kg_manager.open_nodes(payload['names'])
124
 
125
- # --- 4. Gradio Headless 应用构建与 API 注册 (无变化) ---
126
  with gr.Blocks() as app:
127
  gr.Markdown("MCP Server is running. This UI is intentionally blank.", visible=True)
128
  with gr.Row(visible=False):
@@ -138,23 +98,4 @@ with gr.Blocks() as app:
138
  dummy_btn.click(fn=open_nodes, inputs=json_input, outputs=generic_output, api_name="open_nodes")
139
 
140
  if __name__ == "__main__":
141
- app.launch(mcp_server=True)
142
- ```### 核心修改
143
-
144
- 1. **移除了 `Annotated`**: 我们不再需要它,因为它与你环境中的 Gradio 版本不完全兼容。
145
- 2. **为每个函数添加了完整的文档字符串**: 这是最关键的修改。现在,每个需要参数的函数都包含一个 `Args:` 部分,明确地描述了 `payload` 参数。
146
- ```python
147
- def create_entities(payload: CreateEntitiesPayload) -> str:
148
- """在知识图谱中创建多个新实体。
149
-
150
- Args:
151
- payload (CreateEntitiesPayload): 一个包含'entities'键的JSON对象。'entities'的值是一个实体对象的列表。
152
- """
153
- # ...
154
- ```
155
-
156
- 现在,当你运行这个版本的代码时,Gradio 的 MCP 服务器将会:
157
- 1. **看到 `payload: CreateEntitiesPayload`**,并生成正确的、严格的验证模式,确保工具可以正常工作。
158
- 2. **解析文档字符串**,找到 `Args:` 部分中对 `payload` 的描述,并将其显示在工具面板中。
159
-
160
- 这下就真的万无一失了。再次为之前的曲折过程道歉。
 
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
5
 
6
+ # ... (所有数据结构 Manager 类的代码保持不变) ...
7
  @dataclass
8
  class Entity: name: str; entityType: str; observations: List[str] = field(default_factory=list)
9
  @dataclass
 
12
  class KnowledgeGraph: entities: Dict[str, Entity] = field(default_factory=dict); relations: List[Relation] = field(default_factory=list)
13
 
14
  class KnowledgeGraphManager:
 
15
  def __init__(self): self.graph = KnowledgeGraph(); print("Initialized a new, empty in-memory knowledge graph.")
16
  def create_entities(self, entities_data: List[Dict]) -> List[Entity]:
17
  new_entities = []; [ (new_entity := Entity(**e_data), self.graph.entities.update({e_data['name']: new_entity}), new_entities.append(new_entity)) for e_data in entities_data if e_data['name'] not in self.graph.entities]; return new_entities
 
36
 
37
  kg_manager = KnowledgeGraphManager()
38
 
 
39
  class CreateEntitiesPayload(TypedDict): entities: List[Dict]
40
  class CreateRelationsPayload(TypedDict): relations: List[Dict]
41
  class AddObservationsPayload(TypedDict): observations: List[Dict]
 
45
  class SearchNodesPayload(TypedDict): query: str
46
  class OpenNodesPayload(TypedDict): names: List[str]
47
 
48
+ def create_entities(payload: Annotated[CreateEntitiesPayload, "一个包含'entities'键JSON对象。'entities'的值是一个实体对象的列表。"]) -> str:
49
+ """在知识图谱中创建多个新实体。"""
 
 
 
 
 
50
  new_entities = kg_manager.create_entities(payload['entities'])
51
  return json.dumps([asdict(e) for e in new_entities], indent=2)
52
 
53
+ def create_relations(payload: Annotated[CreateRelationsPayload, "一个包含'relations'键的JSON对象。'relations'的值是一个关系对象的列表。"]) -> str:
54
+ """在知识图谱中的实体之间创建多个新关系。"""
 
 
 
 
55
  new_relations = kg_manager.create_relations(payload['relations'])
56
  return json.dumps([{"from": r.from_entity, "to": r.to_entity, "relationType": r.relationType} for r in new_relations], indent=2)
57
 
58
+ # ... (所有其他API函数的定义也使用 Annotated[TypedDict, "description"] 格式) ...
59
+ def add_observations(payload: Annotated[AddObservationsPayload, "一个包含'observations'键JSON对象。'observations'值是一个观察记录对象的列表"]) -> str:
60
+ """向知识图ppu中已存在的实体添加新的观察记录。"""
 
 
 
61
  results = kg_manager.add_observations(payload['observations'])
62
  return json.dumps(results, indent=2)
63
+ def delete_entities(payload: Annotated[DeleteEntitiesPayload, "一个包含'entityNames'键的JSON对象。'entityNames'的值是一个实体名称的列表。"]) -> str:
64
+ """从知识图谱中删除多个实体及其相关联的关系。"""
 
 
 
 
 
65
  kg_manager.delete_entities(payload['entityNames'])
66
  return "Entities and their relations deleted successfully."
67
+ def delete_observations(payload: Annotated[DeleteObservationsPayload, "一个包含'deletions'键的JSON对象。'deletions'的值是一个删除指令的列表。"]) -> str:
68
+ """从知识图谱中的实体删除特定的观察记录。"""
 
 
 
 
 
69
  kg_manager.delete_observations(payload['deletions'])
70
  return "Observations deleted successfully."
71
+ def delete_relations(payload: Annotated[DeleteRelationsPayload, "一个包含'relations'键的JSON对象。'relations'的值是一个要删除的关系对象的列表。"]) -> str:
72
+ """从知识图谱中删除多个关系。"""
 
 
 
 
 
73
  kg_manager.delete_relations(payload['relations'])
74
  return "Relations deleted successfully."
 
75
  def read_graph() -> Dict:
76
  """读取并返回整个知识图谱。此函数不需要参数。"""
77
  return kg_manager.read_graph()
78
+ def search_nodes(payload: Annotated[SearchNodesPayload, "一个包含'query'键的JSON对象。'query'的值是用于搜索的字符串。"]) -> Dict:
79
+ """根据查询词在知识图谱中搜索节点。"""
 
 
 
 
 
80
  return kg_manager.search_nodes(payload['query'])
81
+ def open_nodes(payload: Annotated[OpenNodesPayload, "一个包含'names'键的JSON对象。'names'的值是一个要检索的实体名称的列表。"]) -> Dict:
82
+ """通过名称打开知识图谱中的特定节点。"""
 
 
 
 
 
83
  return kg_manager.open_nodes(payload['names'])
84
 
85
+
86
  with gr.Blocks() as app:
87
  gr.Markdown("MCP Server is running. This UI is intentionally blank.", visible=True)
88
  with gr.Row(visible=False):
 
98
  dummy_btn.click(fn=open_nodes, inputs=json_input, outputs=generic_output, api_name="open_nodes")
99
 
100
  if __name__ == "__main__":
101
+ app.launch(mcp_server=True)