imbajin commented on code in PR #361:
URL: https://github.com/apache/hugegraph-ai/pull/361#discussion_r3497517737
##########
hugegraph-llm/src/hugegraph_llm/operators/hugegraph_op/commit_to_hugegraph.py:
##########
@@ -162,13 +247,18 @@ def load_into_graph(self, vertices, edges, schema): #
pylint: disable=too-many-
else:
result =
self._handle_graph_creation(self.client.graph().addVertex, input_label,
input_properties)
if result is None:
- raise ValueError(f"Failed to create vertex '{input_label}'
with properties {input_properties}")
+ import_result["vertices_skipped"] += 1
+ import_result["errors"].append(
+ self._import_error("vertex", vertex_index,
"create_failed", label=input_label)
+ )
+ continue
vid = result.id
+ import_result["vertices_created"] += 1
vertex["id"] = vid
if mapping_id:
vid_mapping[mapping_id] = vid
- for edge in edges:
+ for edge_index, edge in enumerate(edges):
start = vid_mapping.get(edge.get("outV"), edge.get("outV"))
Review Comment:
‼️ This fallback can still write edges that reference vertices skipped
earlier in the same import. If a vertex fails validation or creation,
`vid_mapping` has no entry, so the edge falls back to the raw `outV`/`inV` and
may connect to an existing vertex with the same id in the graph. Please track
failed/missing endpoints for the current batch and skip those edges with an
explicit `missing_endpoint`/`endpoint_vertex_failed` error instead of falling
back to a raw id.
##########
.github/workflows/hugegraph-llm.yml:
##########
@@ -131,89 +67,16 @@ jobs:
uv sync --extra llm --extra dev
uv run python -c "import nltk; nltk.download('stopwords');
nltk.download('punkt')"
- - name: Run HugeGraph boundary tests
+ - name: Run unit tests
+ working-directory: hugegraph-llm
env:
- HUGEGRAPH_REQUIRED: true
- HUGEGRAPH_URL: http://127.0.0.1:8080
- HUGEGRAPH_GRAPH: hugegraph
- HUGEGRAPH_USER: admin
- HUGEGRAPH_PASSWORD: admin
- SKIP_EXTERNAL_SERVICES: false
- run: |
- uv run pytest hugegraph-llm/src/tests -m "integration and hugegraph"
-v --tb=short
-
- - name: Dump HugeGraph diagnostics
- if: failure()
- run: |
- docker ps -a
- container_id=$(docker ps -aq --filter
ancestor=hugegraph/hugegraph:1.7.0 | head -n 1)
- if [ -n "$container_id" ]; then
- docker logs "$container_id"
- fi
-
- llm-core-smoke:
- runs-on: ubuntu-latest
- strategy:
- fail-fast: false
- matrix:
- python-version: ["3.10", "3.11"]
-
- services:
- hugegraph:
- image: hugegraph/hugegraph:1.7.0
- env:
- PASSWORD: admin
- options: --health-cmd="curl -f http://localhost:8080/versions || exit
1" --health-interval=10s --health-timeout=5s --health-retries=8
- ports:
- - 8080:8080
-
- steps:
- - uses: actions/checkout@v6
- with:
- persist-credentials: false
-
- - name: Set up Python ${{ matrix.python-version }}
- uses: actions/setup-python@v6
- with:
- python-version: ${{ matrix.python-version }}
-
- - name: Install uv
- run: |
- curl -LsSf "https://astral.sh/uv/${UV_VERSION}/install.sh"; | sh
- echo "$HOME/.local/bin" >> "$GITHUB_PATH"
- "$HOME/.local/bin/uv" --version
-
- - name: Cache dependencies
- uses: actions/cache@v5
- with:
- path: |
- ~/.cache/uv
- ~/nltk_data
- key: ${{ runner.os }}-uv-${{ matrix.python-version }}-${{
hashFiles('**/pyproject.toml', 'uv.lock') }}
- restore-keys: |
- ${{ runner.os }}-uv-${{ matrix.python-version }}-
-
- - name: Install dependencies
+ SKIP_EXTERNAL_SERVICES: true
run: |
- uv sync --extra llm --extra dev
- uv run python -c "import nltk; nltk.download('stopwords');
nltk.download('punkt')"
+ uv run pytest src/tests/config/ src/tests/document/
src/tests/middleware/ src/tests/operators/ src/tests/models/ src/tests/indices/
src/tests/test_utils.py -v --tb=short
Review Comment:
‼️ The new CI command no longer runs the API/flow/node tests that cover this
PR's main behavior. The PR adds or rewrites `/graph/extract`, jobs,
`/graph/import`, and extract-and-import tests under `src/tests/api`, plus
related flow/node/utils coverage, but this job only runs
config/document/middleware/operators/models/indices/test_utils. Please restore
an equivalent full unit/contract lane, or at least include `src/tests/api
src/tests/flows src/tests/nodes src/tests/utils`, so these API regressions
cannot pass with green checks.
##########
hugegraph-llm/src/hugegraph_llm/api/models/graph_extract_requests.py:
##########
@@ -16,96 +16,183 @@
# under the License.
import json
+from copy import deepcopy
from typing import Any, Dict, List, Literal, Optional, Union
-from fastapi import Query
from pydantic import BaseModel, ConfigDict, Field, field_validator,
model_validator
+from hugegraph_llm.config import llm_settings
+from hugegraph_llm.operators.common_op.check_schema import CheckSchema
+from hugegraph_llm.utils.schema_property import is_schema_property_value
+
+SchemaInput = Union[str, Dict[str, Any]]
+ContentInput = Union[str, List[str]]
+REQUIRED_VERTEX_KEYS = {"label", "properties"}
+REQUIRED_EDGE_KEYS = {"label", "outV", "outVLabel", "inV", "inVLabel",
"properties"}
+
+
+def _validate_schema_value(schema: SchemaInput) -> SchemaInput:
+ if isinstance(schema, dict):
+ if not schema:
+ raise ValueError("schema must not be an empty object")
+ CheckSchema(deepcopy(schema)).run()
+ return schema
+
+ schema_text = str(schema).strip()
+ if not schema_text:
+ raise ValueError("schema must not be empty")
+ if schema_text.startswith("{"):
+ try:
+ parsed_schema = json.loads(schema_text)
+ except json.JSONDecodeError as exc:
+ raise ValueError(f"schema must be valid JSON: {exc.msg}") from exc
+ if not isinstance(parsed_schema, dict) or not parsed_schema:
+ raise ValueError("schema JSON must be a non-empty object")
+ CheckSchema(deepcopy(parsed_schema)).run()
+ return schema_text
+
+
+def _schema_object(schema: SchemaInput) -> Optional[Dict[str, Any]]:
+ if isinstance(schema, dict):
+ return schema
+ schema_text = str(schema).strip()
+ if not schema_text.startswith("{"):
+ return None
+ try:
+ parsed_schema = json.loads(schema_text)
+ except json.JSONDecodeError:
+ return None
+ return parsed_schema if isinstance(parsed_schema, dict) else None
+
+
+class GraphExtractOptions(BaseModel):
+ include_meta: bool = Field(default=False, description="Whether to include
response metadata.")
+ include_warnings: bool = Field(default=True, description="Whether to
include extraction warnings.")
+
+
+class GraphImportOptions(BaseModel):
+ update_vid_embeddings: bool = Field(default=False, description="Whether to
rebuild vid embeddings after import.")
+
class GraphExtractClientConfig(BaseModel):
model_config = ConfigDict(extra="forbid")
- graph: Optional[str] = None
- user: Optional[str] = None
- pwd: Optional[str] = None
- gs: Optional[str] = None
+ graph: Optional[str] = Field(default=None, description="HugeGraph graph
name.")
+ user: Optional[str] = Field(default=None, description="HugeGraph user.")
+ pwd: Optional[str] = Field(default=None, description="HugeGraph password.")
+ gs: Optional[str] = Field(default=None, description="HugeGraph
graphspace.")
+
+ @field_validator("graph", "user", "pwd", "gs", mode="before")
+ @classmethod
+ def blank_strings_to_none(cls, value):
+ if isinstance(value, str) and not value.strip():
+ return None
+ return value
class GraphExtractRequest(BaseModel):
model_config = ConfigDict(populate_by_name=True)
- texts: Union[str, List[str]] = Field(..., description="Text or list of
texts to extract a graph from.")
- graph_schema: Union[str, Dict[str, Any]] = Field(
- ...,
- alias="schema",
- description="Graph schema as a JSON string/object, or an existing
graph name.",
+ content_type: Literal["text", "chunks"] = Field(
+ default="text", description="Whether content is raw text or chunks."
)
- example_prompt: Optional[str] = Query(None, description="Optional graph
extraction prompt header.")
- extract_type: Literal["property_graph"] = Query("property_graph",
description="Extraction type.")
- language: Literal["zh", "en"] = Query("zh", description="Language for
chunk splitting.")
- split_type: Literal["document", "paragraph", "sentence"] =
Query("document", description="Chunk split granularity.")
- include_meta: bool = Query(False, description="Include vertex/edge/text
counts in the response.")
- client_config: Optional[GraphExtractClientConfig] = Field(None,
description="Request-scoped HugeGraph connection.")
-
- @field_validator("texts")
- @classmethod
- def normalize_texts(cls, v):
- items = [v] if isinstance(v, str) else list(v)
- items = [t for t in items if t and t.strip()]
+ content: Optional[ContentInput] = Field(default=None, description="Raw
document text or pre-split chunks.")
+ texts: Optional[ContentInput] = Field(default=None,
description="Deprecated alias for text or chunk content.")
+ schema_data: SchemaInput = Field(..., alias="schema", description="Graph
schema JSON object/string, or graph name.")
Review Comment:
⚠️ This renames the model input field from `graph_schema` to `schema_data`
while keeping only `schema` as the alias. In the previous model,
`populate_by_name=True` made `GraphExtractRequest(graph_schema=...)` a valid
Python-side contract; now `graph_schema` is only a read-only property and old
callers will fail with a missing `schema`. Please keep `graph_schema` as an
accepted validation alias, or add an explicit compatibility test and migration
note.
##########
hugegraph-llm/src/hugegraph_llm/services/graph_extract_service.py:
##########
@@ -0,0 +1,393 @@
+# 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.
+
+import json
+import time
+from typing import Any, Dict, List, Optional
+
+from hugegraph_llm.api.models.graph_extract_requests import (
+ GraphExtractAndImportRequest,
+ GraphExtractRequest,
+ GraphImportRequest,
+ SchemaInput,
+ _validate_schema_value,
+)
+from hugegraph_llm.api.models.graph_extract_responses import (
+ GraphExtractResponse,
+ GraphImportResponse,
+)
+from hugegraph_llm.config import prompt
+from hugegraph_llm.flows import FlowName
+from hugegraph_llm.flows.scheduler import SchedulerSingleton
+from hugegraph_llm.utils.log import log
+
+SENSITIVE_CLIENT_CONFIG_KEYS = {"pwd", "password", "token", "api_key",
"secret"}
+SAFE_IMPORT_ERROR_KEYS = {"kind", "index", "label", "key", "reason"}
+
+
+class FlowOutputValidationError(ValueError):
+ """Raised when a workflow returns malformed output."""
+
+
+def normalize_schema(schema: SchemaInput) -> str:
+ schema = _validate_schema_value(schema)
+ if isinstance(schema, dict):
+ return json.dumps(schema, ensure_ascii=False)
+
+ schema_text = str(schema).strip()
+ if schema_text.startswith("{"):
+ try:
+ parsed_schema = json.loads(schema_text)
+ except json.JSONDecodeError as exc:
+ raise ValueError(f"schema must be valid JSON: {exc.msg}") from exc
+ return json.dumps(parsed_schema, ensure_ascii=False)
+ return schema_text
+
+
+def _redact_client_config(client_config) -> Dict[str, Any]:
+ if client_config is None:
+ return {}
+ config = client_config if isinstance(client_config, dict) else
client_config.model_dump(exclude_none=True)
+ return {key: ("***" if key in SENSITIVE_CLIENT_CONFIG_KEYS and value else
value) for key, value in config.items()}
+
+
+def _schema_graph_name(schema: str) -> Optional[str]:
+ schema_text = str(schema).strip()
+ return None if schema_text.startswith("{") else schema_text
+
+
+def apply_client_config(
+ client_config,
+ schema: Optional[str] = None,
+ align_graph_with_schema: bool = False,
+) -> Optional[Dict[str, Any]]:
+ if client_config is None:
+ config = {}
+ elif isinstance(client_config, dict):
+ config = {key: value for key, value in client_config.items() if value
is not None}
+ else:
+ config = client_config.model_dump(exclude_none=True)
+ schema_graph = _schema_graph_name(schema) if schema else None
+ if schema_graph:
+ target_graph = config.get("graph")
+ if target_graph and target_graph != schema_graph:
+ raise ValueError("schema graph name must match
client_config.graph")
+ if align_graph_with_schema:
+ config["graph"] = schema_graph
+ return config or None
+
+
+def _parse_flow_json(raw_result: Any, error_message: str) -> Dict[str, Any]:
+ if isinstance(raw_result, dict):
+ return raw_result
+ if not isinstance(raw_result, str):
+ raise FlowOutputValidationError(error_message)
+ try:
+ parsed = json.loads(raw_result)
+ except json.JSONDecodeError as exc:
+ raise FlowOutputValidationError(error_message) from exc
+ if not isinstance(parsed, dict):
+ raise FlowOutputValidationError(error_message)
+ return parsed
+
+
+def _pop_warnings(result: Dict[str, Any]) -> List[str]:
+ warnings = []
+ warning = result.pop("warning", None)
+ if warning:
+ warnings.append(str(warning))
+ extra_warnings = result.pop("warnings", None)
+ if isinstance(extra_warnings, list):
+ warnings.extend(str(item) for item in extra_warnings)
+ elif extra_warnings:
+ warnings.append(str(extra_warnings))
+ return warnings
+
+
+def _count_items(result: Dict[str, Any], key: str) -> int:
+ value = result.get(key)
+ return len(value) if isinstance(value, list) else 0
+
+
+def _validate_property_graph_result(result: Dict[str, Any]) -> None:
+ vertices = result.get("vertices", [])
+ edges = result.get("edges", [])
+ if not isinstance(vertices, list) or not isinstance(edges, list):
+ raise FlowOutputValidationError("property graph result must contain
list vertices and edges")
+ for index, vertex in enumerate(vertices):
+ if not isinstance(vertex, dict) or "label" not in vertex or
"properties" not in vertex:
+ raise FlowOutputValidationError("canonical property graph vertex
must include label and properties")
+ if not _is_non_empty_string(vertex["label"]):
+ raise FlowOutputValidationError(
+ f"canonical property graph vertex[{index}].label must be a
non-empty string"
+ )
+ if not isinstance(vertex["properties"], dict):
+ raise FlowOutputValidationError(f"canonical property graph
vertex[{index}].properties must be an object")
+ required_edge_keys = {"label", "outV", "outVLabel", "inV", "inVLabel",
"properties"}
+ for index, edge in enumerate(edges):
+ if not isinstance(edge, dict) or not required_edge_keys.issubset(edge):
+ raise FlowOutputValidationError(
+ "canonical property graph edge must include label, outV,
outVLabel, inV, inVLabel, and properties"
+ )
+ for key in ("label", "outV", "outVLabel", "inV", "inVLabel"):
+ if not _is_non_empty_string(edge[key]):
+ raise FlowOutputValidationError(
+ f"canonical property graph edge[{index}].{key} must be a
non-empty string"
+ )
+ if not isinstance(edge["properties"], dict):
+ raise FlowOutputValidationError(f"canonical property graph
edge[{index}].properties must be an object")
+
+
+def _is_non_empty_string(value: Any) -> bool:
+ return isinstance(value, str) and bool(value.strip())
+
+
+def _build_import_status(import_result: Dict[str, Any]) -> str:
+ skipped = (
+ import_result.get("vertices_skipped", 0)
+ + import_result.get("edges_skipped", 0)
+ + import_result.get("triples_skipped", 0)
+ )
+ created = (
+ import_result.get("vertices_created", 0)
+ + import_result.get("edges_created", 0)
+ + import_result.get("triples_created", 0)
+ )
+ if skipped and created:
+ return "partial"
+ if skipped and not created:
+ return "failed"
+ return "succeeded"
+
+
+def _sanitize_import_error(error: Any) -> Dict[str, Any]:
+ if not isinstance(error, dict):
+ return {"kind": "import", "reason": "import_error"}
+ sanitized = {
+ key: value for key, value in error.items() if key in
SAFE_IMPORT_ERROR_KEYS and isinstance(value, (str, int))
+ }
+ return sanitized or {"kind": "import", "reason": "import_error"}
+
+
+def _format_import_warning(error: Dict[str, Any]) -> str:
+ kind = error.get("kind", "import")
+ reason = error.get("reason", "import_error")
+ parts = ["import error" if kind == "import" else f"{kind} import error"]
+ if "index" in error:
+ parts.append(f"index={error['index']}")
+ if "label" in error:
+ parts.append(f"label={error['label']}")
+ if "key" in error:
+ parts.append(f"key={error['key']}")
+ parts.append(f"reason={reason}")
+ return " ".join(parts)
+
+
+class GraphExtractService:
+ def __init__(self, scheduler=None):
+ self._scheduler = scheduler
+
+ @property
+ def scheduler(self):
+ return self._scheduler or SchedulerSingleton.get_instance()
+
+ def extract_sync(self, request: GraphExtractRequest) ->
GraphExtractResponse:
+ started = time.perf_counter()
+ schema = normalize_schema(request.schema)
+ extract_client_config = request.client_config if
_schema_graph_name(schema) else None
+ client_config_meta =
_redact_client_config(apply_client_config(extract_client_config, schema=schema))
+ example_prompt = request.example_prompt or prompt.extract_graph_prompt
+ try:
+ raw_result = self.scheduler.schedule_flow(
+ FlowName.GRAPH_EXTRACT,
+ schema,
+ request.texts,
+ example_prompt,
+ request.extract_type,
+ language=request.language,
+ split_type=request.split_type,
+ client_config=extract_client_config,
+ content_type=request.content_type,
+ max_parallel_chunks=request.max_parallel_chunks,
+ )
+ except Exception:
+ log.exception("Graph extraction failed during scheduler execution")
+ raise
+
+ parsed_result = _parse_flow_json(raw_result, "Invalid graph extraction
flow JSON")
+ warnings = _pop_warnings(parsed_result)
+ result = self._build_result(parsed_result, request.extract_type)
+ if not request.options.include_warnings:
+ warnings = []
+ meta = (
+ self._build_extract_meta(request, parsed_result, result, started,
client_config_meta)
+ if request.options.include_meta
+ else {}
+ )
+ return GraphExtractResponse(status="succeeded", result=result,
warnings=warnings, meta=meta)
+
+ def _build_result(self, parsed_result: Dict[str, Any], extract_type: str)
-> Dict[str, Any]:
+ if extract_type == "triples":
+ triples = parsed_result.get("triples")
+ if not triples:
+ triples =
self._legacy_edges_to_triples(parsed_result.get("edges", []))
+ return {"triples": triples}
+ result = {
+ "vertices": parsed_result.get("vertices", []),
+ "edges": parsed_result.get("edges", []),
+ }
+ _validate_property_graph_result(result)
+ return result
+
+ def _legacy_edges_to_triples(self, edges: Any) -> List[Dict[str, Any]]:
+ if not isinstance(edges, list):
+ return []
+ triples = []
+ for edge in edges:
+ if not isinstance(edge, dict):
+ continue
+ start = edge.get("start", edge.get("outV"))
+ end = edge.get("end", edge.get("inV"))
+ edge_type = edge.get("type", edge.get("label"))
+ if start is not None and end is not None and edge_type is not None:
+ triples.append({"start": start, "type": edge_type, "end": end})
+ return triples
+
+ def _build_extract_meta(
+ self,
+ request: GraphExtractRequest,
+ parsed_result: Dict[str, Any],
+ result: Dict[str, Any],
+ started: float,
+ client_config_meta: Dict[str, Any],
+ ) -> Dict[str, Any]:
+ chunk_count = parsed_result.get("chunk_count")
+ if chunk_count is None:
+ chunk_count = (
+ len(request.texts)
+ if request.content_type == "chunks" and
isinstance(request.texts, (list, tuple))
+ else parsed_result.get("call_count")
+ )
+ max_parallel_chunks = parsed_result.get("max_parallel_chunks")
+ if max_parallel_chunks is None:
+ max_parallel_chunks = (
+ min(request.max_parallel_chunks, chunk_count)
+ if isinstance(chunk_count, int) and chunk_count >= 0
+ else request.max_parallel_chunks
+ )
+
+ meta = {
+ "extract_type": request.extract_type,
+ "content_type": request.content_type,
+ "language": request.language,
+ "split_type": request.split_type,
+ "text_count": 1 if request.content_type == "text" else 0,
+ "chunk_count": chunk_count,
+ "max_parallel_chunks": max_parallel_chunks,
+ "vertex_count": _count_items(result, "vertices"),
+ "edge_count": _count_items(result, "edges"),
+ "triple_count": _count_items(result, "triples"),
+ "call_count": parsed_result.get("call_count"),
+ "duration_ms": int((time.perf_counter() - started) * 1000),
+ }
+ if client_config_meta:
+ meta["client_config"] = client_config_meta
+ return meta
+
+
+class GraphImportService:
+ def __init__(self, scheduler=None):
+ self._scheduler = scheduler
+
+ @property
+ def scheduler(self):
+ return self._scheduler or SchedulerSingleton.get_instance()
+
+ def import_graph(self, request: GraphImportRequest) -> GraphImportResponse:
+ if not request.write_to_graph:
+ raise ValueError("write_to_graph must be True to confirm graph
import")
+
+ started = time.perf_counter()
+ schema = normalize_schema(request.schema)
+ graph_config = apply_client_config(request.client_config,
schema=schema, align_graph_with_schema=True)
+ client_config_meta = _redact_client_config(graph_config)
+ try:
+ raw_result = self.scheduler.schedule_flow(
+ FlowName.IMPORT_GRAPH_DATA,
+ request.data,
+ schema,
+ graph_config=graph_config,
+ )
+ except Exception:
+ log.exception("Graph import failed during scheduler execution")
+ raise
+
+ parsed_result = _parse_flow_json(raw_result, "Invalid graph import
flow JSON")
+ warnings = _pop_warnings(parsed_result)
+ import_result = parsed_result.get("import_result")
Review Comment:
‼️ Missing or malformed `import_result` is treated as a successful import
below. If the import flow returns `{}` or the commit node fails to populate
`import_result`, this branch reports `status="succeeded"` and counts the
requested input items, even though there is no evidence that anything was
written. Please validate that `import_result` is present and well-formed,
otherwise raise `FlowOutputValidationError`, and add a regression test for
scheduler output without `import_result`.
--
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]
