'commit'

AI/Neo4j/K1_KnowledgeGraph.py

@@ -1,15 +1,13 @@
 # -*- coding: utf-8 -*-
-import re
-import time
 import hashlib
-from typing import Iterator, Tuple, Dict
-from py2neo import Graph
+import json
+import time
+from typing import Iterator, Dict
+
 from openai import OpenAI
 from openai.types.chat import ChatCompletionChunk
-from Config import *
 from K2_Neo4jExecutor import *
 
-
 class KnowledgeGraph:
     def __init__(self, content: str):
         self.content = content
@@ -18,6 +16,9 @@ class KnowledgeGraph:
         self.existing_knowledge = self._fetch_existing_nodes("KnowledgePoint")
         self.existing_ability = self._fetch_existing_nodes("AbilityPoint")
         self.client = OpenAI(api_key=MODEL_API_KEY, base_url=MODEL_API_URL)
+        # 在__init__方法中添加检查
+        print(f"现有知识点示例：{list(self.existing_knowledge.items())[:3]}")
+        print(f"现有能力点示例：{list(self.existing_ability.items())[:3]}")
 
     def _init_graph_connection(self) -> Graph:
         """初始化并测试数据库连接"""
@@ -47,18 +48,20 @@ class KnowledgeGraph:
         """生成限制性提示词"""
         # 修改提示词中的Cypher示例部分
         system_prompt = f'''
-        // 修改后的正确示例
+        将题目中涉及到的小学数学知识点、能力点进行总结，并且按照以下格式生成在neo4j-community-5.26.2上的语句：
+        // 在示例中强调WITH的必要性
         MERGE (q:Question {{id: "{self.question_id}"}})
-        SET q.content = "题目内容摘要"
+        SET q.content = "..."
 
-        // 必须添加WITH子句
-        WITH q
+        // 必须使用WITH传递上下文
+        WITH q  
         MATCH (kp:KnowledgePoint {{id: "KP_xxxxxx"}})
-        MERGE (q)-[:TESTS_KNOWLEDGE {{weight: 0.8}}]->(kp)
+        MERGE (q)-[:TESTS_KNOWLEDGE]->(kp)
 
-        WITH q  // 多个关系需要继续传递
+        // 多个关系需要继续使用WITH
+        WITH q  
         MATCH (ab:AbilityPoint {{id: "AB_xxxxxx"}})
-        MERGE (q)-[:REQUIRES_ABILITY {{weight: 0.7}}]->(ab)
+        MERGE (q)-[:REQUIRES_ABILITY]->(ab)
         '''
 
         return self.client.chat.completions.create(
@@ -93,57 +96,6 @@ class KnowledgeGraph:
                 safe_blocks.append(cleaned)
         return ';\n\n'.join(safe_blocks) if safe_blocks else ""
 
-    def _sanitize_cypher(self, cypher: str) -> str:
-        """消毒Cypher语句"""
-        valid_lines = []
-        for line in cypher.split('\n'):
-            line = line.split('//')[0].strip()
-            if not line:
-                continue
-
-            # 检查非法操作
-            if re.search(r'\bCREATE\b', line, re.IGNORECASE):
-                continue
-
-            # 验证节点ID
-            if not self._validate_ids(line):
-                continue
-
-            # 验证权重范围
-            if not self._validate_weight(line):
-                continue
-
-            valid_lines.append(line)
-
-        return '\n'.join(valid_lines) if valid_lines else ''
-
-    def _validate_ids(self, line: str) -> bool:
-        """验证行内的所有ID"""
-        kp_ids = {id_.upper() for id_ in re.findall(r'kp_[\da-f]{6}', line, re.IGNORECASE)}
-        ab_ids = {id_.upper() for id_ in re.findall(r'ab_[\da-f]{6}', line, re.IGNORECASE)}
-
-        valid_kp = all(kp in self.existing_knowledge for kp in kp_ids)
-        valid_ab = all(ab in self.existing_ability for ab in ab_ids)
-
-        return valid_kp and valid_ab
-
-    def _validate_weight(self, line: str) -> bool:
-        """验证关系权重是否合法"""
-        weight_match = re.search(r"weight\s*:\s*([0-9.]+)", line)
-        if weight_match:
-            try:
-                weight = float(weight_match.group(1))
-                return 0.0 <= weight <= 1.0
-            except ValueError:
-                return False
-        return True  # 没有weight属性时视为合法
-
-    def _validate_cypher_structure(self, cypher: str) -> bool:
-        """验证WITH子句存在性"""
-        has_merge = re.search(r'\bMERGE\s*\(q:Question\b', cypher, re.IGNORECASE)
-        has_with = re.search(r'\bWITH\s+q\b', cypher, re.IGNORECASE)
-        return not has_merge or (has_merge and has_with)
-
     def run(self) -> Tuple[bool, str, str]:
         """执行安全生成流程"""
         if not self.existing_knowledge or not self.existing_ability:
@@ -174,19 +126,13 @@ class KnowledgeGraph:
                 cypher_script = self._extract_cypher(full_content)
 
                 print(f"\n\n✅ 生成完成！耗时 {int(time.time() - start_time)}秒")
-                print("\n================ 安全Cypher ===============")
-                print(cypher_script if cypher_script else "未通过安全检查")
-                print("==========================================")
                 return True, full_content, cypher_script
 
             print("\n⚠️ 生成完成但未获取到有效内容")
             return False, "空内容", ""
 
-
         # 修改run方法中的异常处理
         except Exception as e:
-            # 原错误代码
-            # print(f"\n\n❌ 生成失败：{str(e)}")
             # 修正后的代码
             error_msg = str(e) if not isinstance(e, dict) else json.dumps(e)
             print(f"\n\n❌ 生成失败：{error_msg}")
@@ -207,9 +153,7 @@ if __name__ == '__main__':
     try:
         kg = KnowledgeGraph(test_content)
         success, result, cypher = kg.run()
-
-        if success and cypher:
-            res = executor.execute_cypher_text(cypher)
-            print("\n\n执行结果：" + res)
+        #res = executor.execute_cypher_text(cypher)
+        #print("\n\n执行结果：" + res)
     except Exception as e:
         print(f"程序初始化失败: {str(e)}")

AI/Neo4j/output.cypher

@@ -1,7 +0,0 @@
-MERGE (q:Question {id: "cccea161"})
-SET q.content = "一个长方体的长是8厘米，宽是5厘米，高是3厘米，求它的表面积是多少平方厘米？"
-WITH q
-MATCH (kp:KnowledgePoint {id: "KP_7i8j9k"})
-MERGE (q)-[:TESTS_KNOWLEDGE {weight: 0.8}]->(kp)
-MATCH (ab:AbilityPoint {id: "AB_j1k2l3"})
-MERGE (q)-[:REQUIRES_ABILITY {weight: 0.7}]->(ab)