bitflicker64 commented on code in PR #372: URL: https://github.com/apache/hugegraph-ai/pull/372#discussion_r4321622517
########## hugegraph-llm/src/hugegraph_llm/extraction_runtime/property_graph.py: ########## @@ -0,0 +1,410 @@ +# 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 uuid import UUID + +from hugegraph_llm.config import prompt +from hugegraph_llm.extraction_runtime.review 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, +) + +# 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: + """Accept standard ASCII UUID spellings; unusual Java forms require repair.""" + return ( + isinstance(value, str) + and re.fullmatch(r"[0-9a-fA-F]{32}|[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}", value) is not None + ) + + +def _float_identity(value) -> tuple[bool, int]: + """Round the decimal Number.toString value directly to IEEE float32, ties to even.""" + # struct's binary64 -> binary32 rounds 1.0000000596046448 down, whereas + # Java Float.valueOf(Number.toString()) rounds its decimal spelling up. + 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(value).int + 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 + if data_type == "DATE": + return data_type, "epoch", value + return data_type, json.dumps(value, sort_keys=True) + + @staticmethod + def _vertex_identity(vertex_id, definition: dict | None) -> tuple | None: + if not _valid_id(vertex_id): + return None + strategy = definition.get("id_strategy") if definition is not None else None + if strategy == "CUSTOMIZE_UUID": + return ("UUID", UUID(vertex_id).int) if _valid_uuid(vertex_id) else None + # PRIMARY_KEY normalization and CUSTOMIZE_STRING both produce StringId; + # numeric ids have a separate HugeGraph identity even across labels. + if strategy in ("CUSTOMIZE_STRING", "PRIMARY_KEY"): + return "STRING", vertex_id + if strategy == "CUSTOMIZE_NUMBER": + return "NUMBER", vertex_id + return ("STRING" if isinstance(vertex_id, str) else "NUMBER"), vertex_id + + 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 = {} + primary_keys_seen = set() + date_identities_seen = {} + 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) + date_keys = [key for key in keys if self.property_keys.get(key, {}).get("data_type") == "DATE"] + if date_keys and all(_valid_scalar(properties[key], "DATE") for key in date_keys): + group = ( + label, + tuple( + self._primary_identity(key, properties[key]) for key in keys if key not in date_keys + ), + ) + dates = tuple(self._primary_identity(key, properties[key]) for key in date_keys) + previous_dates = date_identities_seen.setdefault(group, []) + for previous in previous_dates: + mixed = [ + key for key, current, old in zip(date_keys, dates, previous) if current[1] != old[1] + ] + # A same-domain inequality proves distinct composite keys. + # Cross-domain dates stay unknown without the server zone, + # including when multiple DATE columns change form together. + if mixed and all( Review Comment: Minor: this marks every epoch/text pair as ambiguous, including pairs that no server timezone can make equal. With a DATE primary key, `0` and `"2024-01-01"` get `Ambiguous DATE primary key`, and with `max_fixes=0` the chunk is `blocked`. `test_mixed_date_primary_keys_can_be_repaired_and_must_be_revalidated` asserts this for the `"2024-01-01"` case. HugeGraph stores these as two different `Date` values: `DateUtil` parses text with Joda in the JVM default zone, and a Joda offset is always under 24 hours, and the values are 54 years apart. A valid chunk then costs a repair call or drops out of `result`. Please flag a mixed column only when the epoch value is less than 86,400,000 ms from the text value read as UTC. The `"2024-01-01"` repair case can use `"1970-01-01 05:00:00"` to stay blocked, and `(0, "2024-01-01")` can move to `test_distinct_identity_controls_still_reach_final`. ########## hugegraph-llm/src/hugegraph_llm/extraction_runtime/property_graph.py: ########## @@ -0,0 +1,410 @@ +# 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 uuid import UUID + +from hugegraph_llm.config import prompt +from hugegraph_llm.extraction_runtime.review 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, +) + +# 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: + """Accept standard ASCII UUID spellings; unusual Java forms require repair.""" + return ( + isinstance(value, str) + and re.fullmatch(r"[0-9a-fA-F]{32}|[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}", value) is not None + ) + + +def _float_identity(value) -> tuple[bool, int]: + """Round the decimal Number.toString value directly to IEEE float32, ties to even.""" + # struct's binary64 -> binary32 rounds 1.0000000596046448 down, whereas + # Java Float.valueOf(Number.toString()) rounds its decimal spelling up. + 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) Review Comment: Minor: the DATE check accepts a tab or newline between the date and the time, which HugeGraph rejects. `\s` in this regex matches any ASCII whitespace, and `strptime` treats the space in `" %H:%M:%S"` as any run of whitespace, so `"2024-01-01\t00:00:00"` and `"2024-01-01\n00:00:00"` validate. HugeGraph's `DateUtil` also matches them with `\s` and picks `yyyy-MM-dd HH:mm:ss`, but Joda then fails with `Invalid format ... malformed at "\t00:00:00"` (checked with Joda 2.10.8). A chunk can reach `final` with a value the server will not import. Please match a literal space here (`(?: (\d{2}:\d{2}:\d{2})(\.\d{1,3})?)?`) and add tab and newline cases to the DATE tests. -- 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]
