imbajin commented on code in PR #372: URL: https://github.com/apache/hugegraph-ai/pull/372#discussion_r4304284392
########## hugegraph-llm/src/hugegraph_llm/extraction_runtime/property_graph.py: ########## @@ -0,0 +1,397 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Property-graph extraction and review using the configured model client.""" + +import json +import math +import re +from copy import deepcopy +from dataclasses import asdict +from datetime import datetime +from decimal import Decimal + +from hugegraph_llm.config import prompt +from hugegraph_llm.extraction_runtime.v1.contracts import ChunkInput, Issue, ReviewResult +from hugegraph_llm.models.llms.base import BaseLLM +from hugegraph_llm.operators.llm_op.property_graph_extract import ( + PropertyGraphExtract, + generate_extract_property_graph_prompt, +) + +_JAVA_HEX_LETTERS = str.maketrans("ＡＢＣＤＥＦａｂｃｄｅｆ", "ABCDEFabcdef") +# The midpoint above the largest finite float32 rounds to infinity, not to FLOAT_MAX. +_FLOAT32_OVERFLOW_MIDPOINT = Decimal(2**128 - 2**103) + + +def _invalid_constant(value: str): + raise ValueError(f"Invalid JSON constant: {value}") + + +def _parse_float(value: str) -> float: + number = float(value) + if not math.isfinite(number): + raise ValueError(f"Non-finite JSON number: {value}") + return number + + +def _parse_json(response: str): + text = response.strip() + fenced = re.fullmatch(r"```(?:json)?\s*\n?(.*?)\s*```", text, re.DOTALL | re.IGNORECASE) + if fenced: + text = fenced.group(1) + return json.loads(text, parse_constant=_invalid_constant, parse_float=_parse_float) + + +def _parse_graph(response: str) -> dict: + graph = _parse_json(response) + if isinstance(graph, list): + grouped = {"vertices": [], "edges": []} + for item in graph: + if not isinstance(item, dict) or item.get("type") not in ("vertex", "edge"): + raise ValueError("Flat graph entries must declare type vertex or edge") + grouped["vertices" if item["type"] == "vertex" else "edges"].append(item) + graph = grouped + if not isinstance(graph, dict) or any(not isinstance(graph.get(key), list) for key in ("vertices", "edges")): + raise ValueError("Graph response must contain vertices and edges arrays") + for key, kind in (("vertices", "vertex"), ("edges", "edge")): + for item in graph[key]: + if isinstance(item, dict): + item.setdefault("type", kind) + return graph + + +def _valid_id(value) -> bool: + return (isinstance(value, str) and bool(value)) or (isinstance(value, int) and not isinstance(value, bool)) + + +def _valid_uuid(value) -> bool: + """Match HugeGraph StringEncoding.uuid without rewriting the supplied value.""" + if not isinstance(value, str): + return False + if len(value) == 36 and "-" in value: + # Java UUID.fromString also accepts nonstandard widths and leading '+'. + parts, bits = value.split("-"), 63 + if len(parts) != 5: + return False + elif len(value) == 32: + parts, bits = [value[:16], value[16:]], 64 + else: + return False + for part in parts: + # Java Character.digit(char, 16) includes BMP decimal digits and fullwidth + # A-F. Python int also accepts underscores and non-BMP digits; exclude them. + part = part.translate(_JAVA_HEX_LETTERS) + if re.fullmatch(r"\+?[\da-fA-F]+", part) is None or any(ord(char) > 0xFFFF for char in part): + return False + if int(part, 16) >= 2**bits: + return False + return True + + +def _uuid_identity(value: str) -> int: + """Compare accepted spellings as StringEncoding.uuid/UUID.fromString does.""" + if len(value) == 32: + return (int(value[:16].translate(_JAVA_HEX_LETTERS), 16) << 64) | int( + value[16:].translate(_JAVA_HEX_LETTERS), 16 + ) + parts = [int(part.translate(_JAVA_HEX_LETTERS), 16) for part in value.split("-")] + identity = 0 + for part, bits in zip(parts, (32, 16, 16, 16, 48)): + identity = (identity << bits) | (part & (2**bits - 1)) + return identity + + +def _float_identity(value) -> tuple[bool, int]: + """Round the decimal Number.toString value directly to IEEE float32, ties to even.""" + decimal = Decimal(str(value)) + numerator, denominator = decimal.copy_abs().as_integer_ratio() + if not numerator: + return decimal.is_signed(), 0 + exponent = numerator.bit_length() - denominator.bit_length() + below_power = (numerator < denominator << exponent) if exponent >= 0 else (numerator << -exponent < denominator) + if below_power: + exponent -= 1 + # One integer unit below is the least positive float32 subnormal, 2**-149. + shift = max(exponent - 23, -149) + if shift >= 0: + denominator <<= shift + else: + numerator <<= -shift + rounded, remainder = divmod(numerator, denominator) + if 2 * remainder > denominator or (2 * remainder == denominator and rounded % 2): + rounded += 1 + return decimal.is_signed(), rounded << (shift + 149) + + +def _valid_scalar(value, data_type: str) -> bool: + if data_type == "TEXT": + return isinstance(value, str) + if data_type == "UUID": + return _valid_uuid(value) + if data_type == "BOOLEAN": + return isinstance(value, bool) + if data_type in ("BYTE", "INT", "LONG"): + bits = {"BYTE": 8, "INT": 32, "LONG": 64}[data_type] + return isinstance(value, int) and not isinstance(value, bool) and -(2 ** (bits - 1)) <= value < 2 ** (bits - 1) + if data_type in ("FLOAT", "DOUBLE"): + if not isinstance(value, (int, float)) or isinstance(value, bool): + return False + try: + if not math.isfinite(value): + return False + except OverflowError: + return False + # HugeGraph uses Float.valueOf(Number.toString()). Compare decimal values + # to avoid binary64 double rounding; preserve normal rounding and underflow. + return data_type == "DOUBLE" or Decimal(str(value)).copy_abs() < _FLOAT32_OVERFLOW_MIDPOINT + if data_type == "DATE": + if isinstance(value, int): + # HugeGraph accepts Integer/Long epoch milliseconds, but not floats or booleans. + return _valid_scalar(value, "LONG") + if not isinstance(value, str): + return False + match = re.fullmatch(r"(\d{4}-\d{1,2}-\d{1,2})(?:\s(\d{2}:\d{2}:\d{2})(\.\d{1,3})?)?", value, re.ASCII) + if match is None: + return False + # Match HugeGraph DateUtil's date, second and millisecond formats without normalizing the value. + date_format = "%Y-%m-%d" + (" %H:%M:%S" if match[2] else "") + (".%f" if match[3] else "") + try: + datetime.strptime(value, date_format) + return True + except ValueError: + return False + # Leave unmodeled types (including OBJECT/BLOB) to HugeGraph instead of inventing restrictions. + return value is not None + + +class PropertyGraphBundle: + """Keep invalid graph entries available for repair instead of filtering them.""" + + def __init__(self, *, llm: BaseLLM, schema: dict, example_prompt: str | None = None): + self.llm = llm + self.schema = deepcopy(schema) + self.example_prompt = prompt.extract_graph_prompt if example_prompt is None else example_prompt + self.vertex_labels = {item["name"]: item for item in self.schema["vertexlabels"]} + self.edge_labels = {item["name"]: item for item in self.schema["edgelabels"]} + self.property_keys = {item["name"]: item for item in self.schema.get("propertykeys", [])} + + def extract(self, chunk: ChunkInput) -> dict: + request = self.example_prompt + generate_extract_property_graph_prompt(chunk.text, self.schema) + return _parse_graph(self.llm.generate(prompt=request)) + + def review(self, graph: dict, chunk: ChunkInput) -> ReviewResult: + response = _parse_json(self.llm.generate(prompt=self._context(prompt.review_graph_prompt, graph, chunk))) + if not isinstance(response, dict) or response.get("decision") not in ("pass", "fix", "block"): + raise ValueError("Review decision must be pass, fix, or block") + if not isinstance(response.get("issues"), list): + raise ValueError("Review issues must be an array") + issues = [] + for item in response["issues"]: + if ( + not isinstance(item, dict) + or not isinstance(item.get("path"), str) + or not item["path"].strip() + or not isinstance(item.get("message"), str) + or not item["message"].strip() + or (item.get("evidence") is not None and not isinstance(item["evidence"], str)) + ): + raise ValueError("Review issues require path/message strings and optional evidence text") + issues.append(Issue(path=item["path"], message=item["message"], evidence=item.get("evidence"))) + if (response["decision"] == "pass") != (not issues): + raise ValueError("A pass review must have no issues; fix/block must explain at least one issue") + return ReviewResult(decision=response["decision"], issues=issues) + + def repair(self, graph: dict, chunk: ChunkInput, issues: list[Issue]) -> dict: + request = self._context(prompt.repair_graph_prompt, graph, chunk, issues=[asdict(issue) for issue in issues]) + return _parse_graph(self.llm.generate(prompt=request)) + + def _context(self, instructions: str, graph: dict, chunk: ChunkInput, **extra) -> str: + context = {"text": chunk.text, "schema": self.schema, "graph": graph, **extra} + return instructions + "\n## Input\n" + json.dumps(context, ensure_ascii=False, allow_nan=False) + + def _primary_identity(self, key: str, value) -> tuple: + data_type = self.property_keys.get(key, {}).get("data_type") + if data_type and _valid_scalar(value, data_type): + if data_type == "UUID": + return data_type, _uuid_identity(value) + if data_type == "FLOAT": + return data_type, _float_identity(value) + if data_type == "DOUBLE": + return data_type, float(value).hex() + if data_type == "DATE" and isinstance(value, str): + # DateUtil uses Joda's fraction-of-second SSS, so .1 means 100ms. + date_format = "%Y-%m-%d" + (" %H:%M:%S" if re.search(r"\s", value) else "") + date_format += ".%f" if "." in value else "" + parsed = datetime.strptime(value, date_format) + # SafeDateFormat uses the server's default zone, which this schema + # does not expose. Compare wall times; never guess an epoch/text offset. + return data_type, "wall", parsed + return data_type, json.dumps(value, sort_keys=True) + + def validate(self, graph: dict, chunk: ChunkInput) -> list[Issue]: + """Normalize this chunk's private graph in place and report all remaining problems.""" + del chunk + issues = self._normalize(graph) + vertices_by_id = {} + uuid_ids_seen = set() + primary_keys_seen = set() + for key, kind, labels in ( + ("vertices", "vertex", self.vertex_labels), + ("edges", "edge", self.edge_labels), + ): + for index, item in enumerate(graph[key]): + path = f"{key}[{index}]" + if not isinstance(item, dict): + issues.append(Issue(path, "Entry must be an object")) + continue + if item.get("type", kind) != kind: + issues.append(Issue(path + ".type", f"Expected {kind}")) + label = item.get("label") + definition = labels.get(label) if isinstance(label, str) else None + if definition is None: + issues.append(Issue(path + ".label", f"Unknown {kind} label: {label!r}")) + properties = item.get("properties") + if not isinstance(properties, dict): + issues.append(Issue(path + ".properties", "Properties must be an object")) + elif definition is not None: + issues.extend(self._validate_properties(properties, definition, path)) + keys = definition.get("primary_keys", []) if kind == "vertex" else [] + if keys and all(key in properties and properties[key] not in (None, "") for key in keys): + identity = (label, tuple(self._primary_identity(key, properties[key]) for key in keys)) + if identity in primary_keys_seen: + issues.append(Issue(path + ".properties", "Duplicate vertex primary key")) + primary_keys_seen.add(identity) + if kind == "vertex": + vertex_id = item.get("id") + strategy = definition.get("id_strategy") if definition is not None else None + uuid_identity = ( + _uuid_identity(vertex_id) if strategy == "CUSTOMIZE_UUID" and _valid_uuid(vertex_id) else None + ) + if not _valid_id(vertex_id): + issues.append(Issue(path + ".id", "Vertex id must be a non-empty string or integer")) + elif vertex_id in vertices_by_id: Review Comment: ⚠️ Important — keep string IDs and UUID IDs in separate identity namespaces. A `CUSTOMIZE_STRING` vertex and a `CUSTOMIZE_UUID` vertex can legally have the same UUID text as their ID: HugeGraph stores them as `StringId` and `UuidId`, whose equality checks only match their own type. This global raw-string lookup instead reports `Duplicate vertex id`. I reproduced a graph with those two labels and identical valid UUID text becoming `blocked` with `max_fixes=0`, so neither valid vertex reaches the aggregate result. Please key uniqueness and endpoint resolution by the actual HugeGraph ID type plus value, using the endpoint label to determine that type; keep detecting collisions across labels within the same ID type and add a mixed-strategy regression. ########## hugegraph-llm/src/hugegraph_llm/extraction_runtime/property_graph.py: ########## @@ -0,0 +1,397 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Property-graph extraction and review using the configured model client.""" + +import json +import math +import re +from copy import deepcopy +from dataclasses import asdict +from datetime import datetime +from decimal import Decimal + +from hugegraph_llm.config import prompt +from hugegraph_llm.extraction_runtime.v1.contracts import ChunkInput, Issue, ReviewResult +from hugegraph_llm.models.llms.base import BaseLLM +from hugegraph_llm.operators.llm_op.property_graph_extract import ( + PropertyGraphExtract, + generate_extract_property_graph_prompt, +) + +_JAVA_HEX_LETTERS = str.maketrans("ＡＢＣＤＥＦａｂｃｄｅｆ", "ABCDEFabcdef") +# The midpoint above the largest finite float32 rounds to infinity, not to FLOAT_MAX. +_FLOAT32_OVERFLOW_MIDPOINT = Decimal(2**128 - 2**103) + + +def _invalid_constant(value: str): + raise ValueError(f"Invalid JSON constant: {value}") + + +def _parse_float(value: str) -> float: + number = float(value) + if not math.isfinite(number): + raise ValueError(f"Non-finite JSON number: {value}") + return number + + +def _parse_json(response: str): + text = response.strip() + fenced = re.fullmatch(r"```(?:json)?\s*\n?(.*?)\s*```", text, re.DOTALL | re.IGNORECASE) + if fenced: + text = fenced.group(1) + return json.loads(text, parse_constant=_invalid_constant, parse_float=_parse_float) + + +def _parse_graph(response: str) -> dict: + graph = _parse_json(response) + if isinstance(graph, list): + grouped = {"vertices": [], "edges": []} + for item in graph: + if not isinstance(item, dict) or item.get("type") not in ("vertex", "edge"): + raise ValueError("Flat graph entries must declare type vertex or edge") + grouped["vertices" if item["type"] == "vertex" else "edges"].append(item) + graph = grouped + if not isinstance(graph, dict) or any(not isinstance(graph.get(key), list) for key in ("vertices", "edges")): + raise ValueError("Graph response must contain vertices and edges arrays") + for key, kind in (("vertices", "vertex"), ("edges", "edge")): + for item in graph[key]: + if isinstance(item, dict): + item.setdefault("type", kind) + return graph + + +def _valid_id(value) -> bool: + return (isinstance(value, str) and bool(value)) or (isinstance(value, int) and not isinstance(value, bool)) + + +def _valid_uuid(value) -> bool: + """Match HugeGraph StringEncoding.uuid without rewriting the supplied value.""" + if not isinstance(value, str): + return False + if len(value) == 36 and "-" in value: + # Java UUID.fromString also accepts nonstandard widths and leading '+'. + parts, bits = value.split("-"), 63 + if len(parts) != 5: + return False + elif len(value) == 32: + parts, bits = [value[:16], value[16:]], 64 + else: + return False + for part in parts: + # Java Character.digit(char, 16) includes BMP decimal digits and fullwidth + # A-F. Python int also accepts underscores and non-BMP digits; exclude them. + part = part.translate(_JAVA_HEX_LETTERS) + if re.fullmatch(r"\+?[\da-fA-F]+", part) is None or any(ord(char) > 0xFFFF for char in part): + return False + if int(part, 16) >= 2**bits: + return False + return True + + +def _uuid_identity(value: str) -> int: + """Compare accepted spellings as StringEncoding.uuid/UUID.fromString does.""" + if len(value) == 32: + return (int(value[:16].translate(_JAVA_HEX_LETTERS), 16) << 64) | int( + value[16:].translate(_JAVA_HEX_LETTERS), 16 + ) + parts = [int(part.translate(_JAVA_HEX_LETTERS), 16) for part in value.split("-")] + identity = 0 + for part, bits in zip(parts, (32, 16, 16, 16, 48)): + identity = (identity << bits) | (part & (2**bits - 1)) + return identity + + +def _float_identity(value) -> tuple[bool, int]: + """Round the decimal Number.toString value directly to IEEE float32, ties to even.""" + decimal = Decimal(str(value)) + numerator, denominator = decimal.copy_abs().as_integer_ratio() + if not numerator: + return decimal.is_signed(), 0 + exponent = numerator.bit_length() - denominator.bit_length() + below_power = (numerator < denominator << exponent) if exponent >= 0 else (numerator << -exponent < denominator) + if below_power: + exponent -= 1 + # One integer unit below is the least positive float32 subnormal, 2**-149. + shift = max(exponent - 23, -149) + if shift >= 0: + denominator <<= shift + else: + numerator <<= -shift + rounded, remainder = divmod(numerator, denominator) + if 2 * remainder > denominator or (2 * remainder == denominator and rounded % 2): + rounded += 1 + return decimal.is_signed(), rounded << (shift + 149) + + +def _valid_scalar(value, data_type: str) -> bool: + if data_type == "TEXT": + return isinstance(value, str) + if data_type == "UUID": + return _valid_uuid(value) + if data_type == "BOOLEAN": + return isinstance(value, bool) + if data_type in ("BYTE", "INT", "LONG"): + bits = {"BYTE": 8, "INT": 32, "LONG": 64}[data_type] + return isinstance(value, int) and not isinstance(value, bool) and -(2 ** (bits - 1)) <= value < 2 ** (bits - 1) + if data_type in ("FLOAT", "DOUBLE"): + if not isinstance(value, (int, float)) or isinstance(value, bool): + return False + try: + if not math.isfinite(value): + return False + except OverflowError: + return False + # HugeGraph uses Float.valueOf(Number.toString()). Compare decimal values + # to avoid binary64 double rounding; preserve normal rounding and underflow. + return data_type == "DOUBLE" or Decimal(str(value)).copy_abs() < _FLOAT32_OVERFLOW_MIDPOINT + if data_type == "DATE": + if isinstance(value, int): + # HugeGraph accepts Integer/Long epoch milliseconds, but not floats or booleans. + return _valid_scalar(value, "LONG") + if not isinstance(value, str): + return False + match = re.fullmatch(r"(\d{4}-\d{1,2}-\d{1,2})(?:\s(\d{2}:\d{2}:\d{2})(\.\d{1,3})?)?", value, re.ASCII) + if match is None: + return False + # Match HugeGraph DateUtil's date, second and millisecond formats without normalizing the value. + date_format = "%Y-%m-%d" + (" %H:%M:%S" if match[2] else "") + (".%f" if match[3] else "") + try: + datetime.strptime(value, date_format) + return True + except ValueError: + return False + # Leave unmodeled types (including OBJECT/BLOB) to HugeGraph instead of inventing restrictions. + return value is not None + + +class PropertyGraphBundle: + """Keep invalid graph entries available for repair instead of filtering them.""" + + def __init__(self, *, llm: BaseLLM, schema: dict, example_prompt: str | None = None): + self.llm = llm + self.schema = deepcopy(schema) + self.example_prompt = prompt.extract_graph_prompt if example_prompt is None else example_prompt + self.vertex_labels = {item["name"]: item for item in self.schema["vertexlabels"]} + self.edge_labels = {item["name"]: item for item in self.schema["edgelabels"]} + self.property_keys = {item["name"]: item for item in self.schema.get("propertykeys", [])} + + def extract(self, chunk: ChunkInput) -> dict: + request = self.example_prompt + generate_extract_property_graph_prompt(chunk.text, self.schema) + return _parse_graph(self.llm.generate(prompt=request)) + + def review(self, graph: dict, chunk: ChunkInput) -> ReviewResult: + response = _parse_json(self.llm.generate(prompt=self._context(prompt.review_graph_prompt, graph, chunk))) + if not isinstance(response, dict) or response.get("decision") not in ("pass", "fix", "block"): + raise ValueError("Review decision must be pass, fix, or block") + if not isinstance(response.get("issues"), list): + raise ValueError("Review issues must be an array") + issues = [] + for item in response["issues"]: + if ( + not isinstance(item, dict) + or not isinstance(item.get("path"), str) + or not item["path"].strip() + or not isinstance(item.get("message"), str) + or not item["message"].strip() + or (item.get("evidence") is not None and not isinstance(item["evidence"], str)) + ): + raise ValueError("Review issues require path/message strings and optional evidence text") + issues.append(Issue(path=item["path"], message=item["message"], evidence=item.get("evidence"))) + if (response["decision"] == "pass") != (not issues): + raise ValueError("A pass review must have no issues; fix/block must explain at least one issue") + return ReviewResult(decision=response["decision"], issues=issues) + + def repair(self, graph: dict, chunk: ChunkInput, issues: list[Issue]) -> dict: + request = self._context(prompt.repair_graph_prompt, graph, chunk, issues=[asdict(issue) for issue in issues]) + return _parse_graph(self.llm.generate(prompt=request)) + + def _context(self, instructions: str, graph: dict, chunk: ChunkInput, **extra) -> str: + context = {"text": chunk.text, "schema": self.schema, "graph": graph, **extra} + return instructions + "\n## Input\n" + json.dumps(context, ensure_ascii=False, allow_nan=False) + + def _primary_identity(self, key: str, value) -> tuple: + data_type = self.property_keys.get(key, {}).get("data_type") + if data_type and _valid_scalar(value, data_type): + if data_type == "UUID": + return data_type, _uuid_identity(value) + if data_type == "FLOAT": + return data_type, _float_identity(value) + if data_type == "DOUBLE": + return data_type, float(value).hex() + if data_type == "DATE" and isinstance(value, str): + # DateUtil uses Joda's fraction-of-second SSS, so .1 means 100ms. + date_format = "%Y-%m-%d" + (" %H:%M:%S" if re.search(r"\s", value) else "") + date_format += ".%f" if "." in value else "" + parsed = datetime.strptime(value, date_format) + # SafeDateFormat uses the server's default zone, which this schema + # does not expose. Compare wall times; never guess an epoch/text offset. + return data_type, "wall", parsed + return data_type, json.dumps(value, sort_keys=True) Review Comment: ⚠️ Important — mixed DATE encodings can bypass duplicate primary-key detection. DATE strings become wall-clock identities above, but epoch-millisecond integers fall through to a separate raw-JSON identity here. With a DATE primary key, I reproduced two vertices containing `0` and `"1970-01-01 00:00:00"` reaching `final` with no issues; on a UTC HugeGraph server both convert to the same `Date(0)` and therefore the same primary-key ID. The new distinct-identity control test currently accepts that pair. Please either compare both forms using an explicit target timezone, or report mixed epoch/text DATE primary keys as ambiguous when that timezone is unavailable, and add a UTC-equivalence regression. Avoid guessing a timezone, but do not certify potentially colliding identities as distinct. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
