VGalaxies commented on code in PR #361:
URL: https://github.com/apache/hugegraph-ai/pull/361#discussion_r3564717983
##########
hugegraph-llm/src/hugegraph_llm/api/models/graph_extract_requests.py:
##########
@@ -16,96 +16,189 @@
# 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 pydantic import AliasChoices, 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()
Review Comment:
**⚠️
`hugegraph-llm/src/hugegraph_llm/api/models/graph_extract_requests.py:38` -
Fully validate property-key schema entries**
**Evidence**
- `CheckSchema.run()` indexes every `propertykeys` entry’s `"name"`
directly. Model construction with `propertykeys=[{}]` raises an uncaught
`KeyError`, while entries missing `data_type` or `cardinality` are accepted and
fail later in `Commit2Graph`.
**Impact**
- Malformed requests can produce unstructured 500 responses or fail after
graph-schema mutation has begun.
**Requested fix**
- Validate each property key’s shape, supported data type, and cardinality
at the request boundary, converting all structural errors into normal 422
validation responses.
##########
hugegraph-llm/src/hugegraph_llm/nodes/llm_node/extract_info.py:
##########
@@ -32,7 +32,7 @@ class ExtractNode(BaseNode):
extract_type: str = None
def node_init(self):
- llm = get_chat_llm(llm_settings)
+ llm = get_extract_llm(llm_settings)
Review Comment:
**‼️ `hugegraph-llm/src/hugegraph_llm/nodes/llm_node/extract_info.py:35` -
Preserve legacy cross-provider extraction configuration**
**Evidence**
- `origin/main` always used `get_chat_llm()`. The new `get_extract_llm()`
fallback only activates when `chat_llm_type` matches the extract provider. A
deployment configured only with Ollama chat settings therefore selects the
default, unconfigured OpenAI extract client.
**Impact**
- Existing graph extraction deployments can fail with missing credentials or
contact the wrong provider after upgrading.
**Requested fix**
- When extract settings were not explicitly configured, fall back to the
configured chat client regardless of provider. Add cross-provider regression
tests.
##########
hugegraph-llm/src/hugegraph_llm/api/graph_extract_api.py:
##########
@@ -15,55 +15,355 @@
# specific language governing permissions and limitations
# under the License.
-import json
+from typing import Optional
-from fastapi import APIRouter, HTTPException, status
+from fastapi import APIRouter, HTTPException, Request, status
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+from fastapi.routing import APIRoute
-from hugegraph_llm.api.models.graph_extract_requests import GraphExtractRequest
-from hugegraph_llm.api.models.graph_extract_responses import
GraphExtractResponse
-from hugegraph_llm.config import prompt
-from hugegraph_llm.flows import FlowName
-from hugegraph_llm.flows.scheduler import SchedulerSingleton
+from hugegraph_llm.api.models.graph_extract_requests import (
+ GraphExtractAndImportRequest,
+ GraphExtractRequest,
+ GraphImportRequest,
+)
+from hugegraph_llm.api.models.graph_extract_responses import (
+ GraphExtractAndImportResponse,
+ GraphExtractError,
+ GraphExtractJobCreateResponse,
+ GraphExtractJobStatusResponse,
+ GraphExtractResponse,
+ GraphImportResponse,
+)
+from hugegraph_llm.services.graph_extract_jobs import (
+ GraphExtractJob,
+ GraphExtractJobStatus,
+ InMemoryGraphExtractJobStore,
+)
+from hugegraph_llm.services.graph_extract_service import (
+ FlowOutputValidationError,
+ GraphExtractService,
+ GraphImportService,
+)
from hugegraph_llm.utils.log import log
+GRAPH_EXTRACT_FLOW_OUTPUT_ERROR = "Graph extraction flow output is invalid"
+GRAPH_EXTRACT_RUNTIME_ERROR = "Graph extraction failed during execution"
+GRAPH_IMPORT_FLOW_OUTPUT_ERROR = "Graph import flow output is invalid"
+GRAPH_IMPORT_RUNTIME_ERROR = "Graph import failed during execution"
-class GraphExtractService:
- @staticmethod
- def extract_sync(req: GraphExtractRequest) -> GraphExtractResponse:
+
+def _error(code: str, message: str, phase: str, job_id: Optional[str] = None)
-> dict:
+ return GraphExtractError(code=code, message=message, phase=phase,
job_id=job_id).model_dump(exclude_none=True)
+
+
+def _job_ts(value) -> Optional[str]:
+ return value.isoformat() if value else None
+
+
+def _job_status_response(job: GraphExtractJob) ->
GraphExtractJobStatusResponse:
+ return GraphExtractJobStatusResponse(
+ job_id=job.job_id,
+ status=job.status,
+ created_at=_job_ts(job.created_at),
+ updated_at=_job_ts(job.updated_at),
+ started_at=_job_ts(job.started_at),
+ finished_at=_job_ts(job.finished_at),
+ expires_at=_job_ts(job.expires_at),
+ error=job.error,
+ )
+
+
+def _validation_message(errors) -> str:
+ details = []
+ for error in errors:
+ loc = ".".join(str(part) for part in error.get("loc", []) if part not
in {"body"})
+ msg = error.get("msg", "invalid input")
+ err_type = error.get("type", "validation_error")
+ details.append(f"{loc or 'request'}: {msg} ({err_type})")
+ return "; ".join(details) or "request validation failed"
+
+
+def _validation_error_for_path(path: str, message: str) -> dict:
+ if path.endswith("/graph/import"):
+ return _error("GRAPH_IMPORT_VALIDATION_ERROR", message, "import")
+ if path.endswith("/graph/extract-and-import"):
+ return _error("GRAPH_EXTRACT_AND_IMPORT_VALIDATION_ERROR", message,
"request")
+ return _error("GRAPH_EXTRACT_VALIDATION_ERROR", message, "request")
+
+
+class GraphExtractAPIRoute(APIRoute):
+ def get_route_handler(self):
+ original_route_handler = super().get_route_handler()
+
+ async def custom_route_handler(request: Request):
+ try:
+ return await original_route_handler(request)
+ except RequestValidationError as exc:
+ message = _validation_message(exc.errors())
+ return JSONResponse(
+ status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+ content={"detail":
_validation_error_for_path(request.url.path, message)},
+ )
+
+ return custom_route_handler
+
+
+def graph_extract_http_api(
+ router: APIRouter,
+ service=None,
+ job_store=None,
+ import_service=None,
+ run_jobs_inline: Optional[bool] = None,
+):
+ extract_service = service or GraphExtractService()
+ graph_import_service = import_service or GraphImportService()
+ jobs = job_store or InMemoryGraphExtractJobStore()
+ original_route_class = router.route_class
+ router.route_class = GraphExtractAPIRoute
+
+ @router.post("/graph/extract", status_code=status.HTTP_200_OK,
response_model=GraphExtractResponse)
+ def graph_extract_api(req: GraphExtractRequest) -> GraphExtractResponse:
+ try:
+ return extract_service.extract_sync(req)
+ except FlowOutputValidationError as exc:
+ log.error("Graph extraction flow output is invalid: %s", exc)
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ detail=_error("GRAPH_EXTRACT_INVALID_FLOW_OUTPUT",
GRAPH_EXTRACT_FLOW_OUTPUT_ERROR, "extract"),
+ ) from exc
+ except ValueError as exc:
+ log.error("Graph extraction input is invalid: %s", exc)
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=_error("GRAPH_EXTRACT_INVALID_INPUT", str(exc),
"request"),
+ ) from exc
+ except Exception as exc:
+ log.error("Unexpected graph extraction error: %s", exc,
exc_info=True)
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ detail=_error("GRAPH_EXTRACT_FAILED",
GRAPH_EXTRACT_RUNTIME_ERROR, "extract"),
+ ) from exc
+
+ @router.post(
+ "/graph/extract/jobs",
+ status_code=status.HTTP_202_ACCEPTED,
+ response_model=GraphExtractJobCreateResponse,
+ )
+ def create_graph_extract_job(
+ req: GraphExtractRequest,
+ ) -> GraphExtractJobCreateResponse:
+ """Create a process-local graph extraction job.
+
+ Jobs are stored in memory and are not shared across API worker
processes.
+ Job status/results are lost on service restart, and cancellation only
applies before
+ a queued job starts running; it cannot interrupt an active LLM call.
+ """
+ try:
+ job = jobs.create(req)
+ except ValueError as exc:
+ raise HTTPException(
+ status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+ detail=_error("GRAPH_EXTRACT_JOB_LIMIT_EXCEEDED", str(exc),
"job"),
+ ) from exc
+ if run_jobs_inline is True:
+ jobs.run_job(job.job_id, extract_service)
+ elif run_jobs_inline is None:
+ try:
+ jobs.submit_job(job.job_id, extract_service)
+ except ValueError as exc:
+ raise HTTPException(
+ status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+ detail=_error("GRAPH_EXTRACT_JOB_QUEUE_FULL", str(exc),
"job", job.job_id),
+ ) from exc
+ return GraphExtractJobCreateResponse(
+ job_id=job.job_id,
+ status=job.status,
+ result_url=f"/graph/extract/jobs/{job.job_id}/result",
+ created_at=_job_ts(job.created_at),
+ updated_at=_job_ts(job.updated_at),
+ )
+
+ @router.get(
+ "/graph/extract/jobs/{job_id}",
+ status_code=status.HTTP_200_OK,
+ response_model=GraphExtractJobStatusResponse,
+ )
+ def get_graph_extract_job(job_id: str) -> GraphExtractJobStatusResponse:
+ jobs.expire_jobs()
+ job = jobs.get(job_id)
+ if job is None:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=_error("GRAPH_EXTRACT_JOB_NOT_FOUND", f"Job {job_id}
was not found", "job", job_id),
+ )
+ return _job_status_response(job)
+
+ @router.get(
+ "/graph/extract/jobs/{job_id}/result",
+ status_code=status.HTTP_200_OK,
+ response_model=GraphExtractResponse,
+ )
+ def get_graph_extract_job_result(job_id: str) -> GraphExtractResponse:
+ jobs.expire_jobs()
+ job = jobs.get(job_id)
+ if job is None:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=_error("GRAPH_EXTRACT_JOB_NOT_FOUND", f"Job {job_id}
was not found", "job", job_id),
+ )
+ if job.status == GraphExtractJobStatus.EXPIRED:
+ raise HTTPException(
+ status_code=status.HTTP_410_GONE,
+ detail=_error("GRAPH_EXTRACT_JOB_EXPIRED", f"Job {job_id}
result has expired", "job", job_id),
+ )
+ if job.status in {GraphExtractJobStatus.PENDING,
GraphExtractJobStatus.RUNNING}:
+ raise HTTPException(
+ status_code=status.HTTP_202_ACCEPTED,
+ detail=_error(
+ "GRAPH_EXTRACT_JOB_NOT_COMPLETE",
+ f"Job {job_id} is not complete",
+ "job",
+ job_id,
+ ),
+ )
+ if job.status == GraphExtractJobStatus.CANCELLED:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail=_error("GRAPH_EXTRACT_JOB_CANCELLED", f"Job {job_id}
was cancelled", "job", job_id),
+ )
+ if job.status == GraphExtractJobStatus.FAILED:
+ raise
HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=job.error.model_dump())
+ if job.result is None:
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ detail=_error(
+ "GRAPH_EXTRACT_JOB_RESULT_MISSING", f"Job {job_id}
finished without a result", "job", job_id
+ ),
+ )
+ return job.result
+
+ @router.delete(
+ "/graph/extract/jobs/{job_id}",
+ status_code=status.HTTP_200_OK,
+ response_model=GraphExtractJobStatusResponse,
+ )
+ def cancel_graph_extract_job(job_id: str) -> GraphExtractJobStatusResponse:
+ job = jobs.cancel(job_id)
+ if job is None:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=_error("GRAPH_EXTRACT_JOB_NOT_FOUND", f"Job {job_id}
was not found", "job", job_id),
+ )
+ if job.status == GraphExtractJobStatus.RUNNING:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail=_error(
+ "GRAPH_EXTRACT_JOB_NOT_CANCELLABLE",
+ f"Job {job_id} is already running and cannot be
interrupted",
+ "job",
+ job_id,
+ ),
+ )
+ return _job_status_response(job)
+
+ @router.post("/graph/import", status_code=status.HTTP_200_OK,
response_model=GraphImportResponse)
+ def graph_import_api(req: GraphImportRequest) -> GraphImportResponse:
+ if not req.write_to_graph:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=_error(
+ "GRAPH_IMPORT_CONFIRMATION_REQUIRED",
+ "write_to_graph=true is required before writing graph data
to HugeGraph",
+ "import",
+ ),
+ )
try:
- scheduler = SchedulerSingleton.get_instance()
- result_str = scheduler.schedule_flow(
- FlowName.GRAPH_EXTRACT,
- req.graph_schema,
- req.texts,
- req.example_prompt or prompt.extract_graph_prompt,
- req.extract_type,
- language=req.language,
- split_type=req.split_type,
- client_config=req.client_config,
+ return graph_import_service.import_graph(req)
+ except FlowOutputValidationError as exc:
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ detail=_error("GRAPH_IMPORT_INVALID_FLOW_OUTPUT",
GRAPH_IMPORT_FLOW_OUTPUT_ERROR, "import"),
+ ) from exc
+ except ValueError as exc:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=_error("GRAPH_IMPORT_INVALID_INPUT", str(exc),
"import"),
+ ) from exc
+ except Exception as exc:
+ log.error("Unexpected graph import error: %s", exc, exc_info=True)
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ detail=_error("GRAPH_IMPORT_FAILED",
GRAPH_IMPORT_RUNTIME_ERROR, "import"),
+ ) from exc
+
+ @router.post(
+ "/graph/extract-and-import",
+ status_code=status.HTTP_200_OK,
+ response_model=GraphExtractAndImportResponse,
+ )
+ def graph_extract_and_import_api(req: GraphExtractAndImportRequest) ->
GraphExtractAndImportResponse:
+ if not req.write_to_graph:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=_error(
+ "GRAPH_IMPORT_CONFIRMATION_REQUIRED",
+ "write_to_graph=true is required before writing extraction
results to HugeGraph",
+ "import",
+ ),
)
- raw = json.loads(result_str)
- warnings = [raw.pop("warning")] if "warning" in raw else []
- result = {"vertices": raw.get("vertices", []), "edges":
raw.get("edges", [])}
- meta = {}
- if req.include_meta:
- meta = {
- "vertex_count": len(result["vertices"]),
- "edge_count": len(result["edges"]),
- "text_count": len(req.texts),
- }
- return GraphExtractResponse(result=result, warnings=warnings,
meta=meta)
- except HTTPException:
- raise
- except Exception as e:
- log.error("Error in graph_extract_api: %s", e)
+ try:
+ extract_response = extract_service.extract_sync(req)
+ except FlowOutputValidationError as exc:
+ log.error("Extract-and-import extraction flow output is invalid:
%s", exc)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
- detail="An unexpected error occurred during graph extraction.",
- ) from e
+ detail=_error("GRAPH_EXTRACT_INVALID_FLOW_OUTPUT",
GRAPH_EXTRACT_FLOW_OUTPUT_ERROR, "extract"),
+ ) from exc
+ except ValueError as exc:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=_error("GRAPH_EXTRACT_INVALID_INPUT", str(exc),
"request"),
+ ) from exc
+ except Exception as exc:
+ log.error("Unexpected extract-and-import extraction error: %s",
exc, exc_info=True)
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ detail=_error("GRAPH_EXTRACT_FAILED",
GRAPH_EXTRACT_RUNTIME_ERROR, "extract"),
+ ) from exc
+ try:
+ import_response = graph_import_service.import_graph(
+ GraphImportRequest(
+ schema=req.schema,
+ data=extract_response.result,
+ write_to_graph=True,
+ client_config=req.client_config,
+ options=req.import_options,
+ )
+ )
+ except FlowOutputValidationError as exc:
+ log.error("Extract-and-import import flow output is invalid: %s",
exc)
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ detail=_error("GRAPH_IMPORT_INVALID_FLOW_OUTPUT",
GRAPH_IMPORT_FLOW_OUTPUT_ERROR, "import"),
+ ) from exc
+ except ValueError as exc:
Review Comment:
**‼️ `hugegraph-llm/src/hugegraph_llm/api/graph_extract_api.py:351` - Do not
expose generated import validation errors**
**Evidence**
- `/graph/extract-and-import` constructs `GraphImportRequest` from
LLM-generated data. A schema-type mismatch raises a Pydantic `ValidationError`,
which is caught as `ValueError` and returned through `str(exc)`. Pydantic error
strings include `input_value`, potentially containing extracted content and
`client_config.pwd`.
**Impact**
- Invalid LLM output can disclose request data or graph credentials and is
incorrectly classified as a client-side 400.
**Requested fix**
- Treat this as sanitized invalid flow output, never return `str(exc)`, and
add a regression test containing a password and invalid extracted property.
##########
hugegraph-llm/src/hugegraph_llm/api/models/graph_extract_requests.py:
##########
@@ -125,3 +218,177 @@ def validate_schema_and_client_config(self):
f"(got schema='{schema}',
client_config.graph='{self.client_config.graph}')."
)
return self
+
+
+class GraphImportRequest(BaseModel):
+ model_config = ConfigDict(populate_by_name=True)
+
+ schema_data: SchemaInput = Field(..., alias="schema", description="Graph
schema JSON object/string, or graph name.")
+ data: Dict[str, Any] = Field(..., description="Property graph data with
vertices and edges.")
+ write_to_graph: bool = Field(default=False, description="Required
confirmation for graph writes.")
+ client_config: Optional[GraphExtractClientConfig] = Field(default=None)
+ options: GraphImportOptions = Field(default_factory=GraphImportOptions)
+
+ @property
+ def schema(self) -> SchemaInput:
+ return self.schema_data
+
+ @field_validator("schema_data")
+ @classmethod
+ def validate_schema(cls, schema: SchemaInput) -> SchemaInput:
+ return _validate_schema_value(schema)
+
+ @field_validator("data")
+ @classmethod
+ def validate_data(cls, data: Dict[str, Any]) -> Dict[str, Any]:
+ vertices = cls._optional_list(data, "vertices")
+ edges = cls._optional_list(data, "edges")
+ triples = cls._optional_list(data, "triples")
+ if triples:
+ raise ValueError("triples import is not supported; submit property
graph vertices or edges")
+ if not vertices and not edges and not triples:
+ raise ValueError("data must contain at least one vertex or edge")
+ for index, vertex in enumerate(vertices):
+ if not isinstance(vertex, dict) or not
REQUIRED_VERTEX_KEYS.issubset(vertex):
+ raise ValueError(f"vertices[{index}] must include label and
properties")
+ if not isinstance(vertex["label"], str) or not
vertex["label"].strip():
+ raise ValueError(f"vertices[{index}].label must be a non-empty
string")
+ if not isinstance(vertex["properties"], dict):
+ raise ValueError(f"vertices[{index}].properties must be an
object")
+ for index, edge in enumerate(edges):
+ if not isinstance(edge, dict) or not
REQUIRED_EDGE_KEYS.issubset(edge):
+ raise ValueError(f"edges[{index}] must include label, outV,
outVLabel, inV, inVLabel, and properties")
+ for key in ("label", "outV", "outVLabel", "inV", "inVLabel"):
Review Comment:
**⚠️
`hugegraph-llm/src/hugegraph_llm/api/models/graph_extract_requests.py:261` -
Allow numeric HugeGraph endpoint IDs**
**Evidence**
- The validator requires `outV` and `inV` to be strings, although HugeGraph
supports `AUTOMATIC` and `CUSTOMIZE_NUMBER` vertex IDs and
`PyHugeClient.addEdge()` preserves integer endpoints.
**Impact**
- Edge-only imports targeting existing numeric-ID vertices are rejected
before reaching HugeGraph.
**Requested fix**
- Accept non-boolean integers or non-empty strings for endpoint IDs and add
an API regression test using numeric endpoints.
##########
hugegraph-llm/src/hugegraph_llm/api/graph_extract_api.py:
##########
@@ -15,55 +15,355 @@
# specific language governing permissions and limitations
# under the License.
-import json
+from typing import Optional
-from fastapi import APIRouter, HTTPException, status
+from fastapi import APIRouter, HTTPException, Request, status
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+from fastapi.routing import APIRoute
-from hugegraph_llm.api.models.graph_extract_requests import GraphExtractRequest
-from hugegraph_llm.api.models.graph_extract_responses import
GraphExtractResponse
-from hugegraph_llm.config import prompt
-from hugegraph_llm.flows import FlowName
-from hugegraph_llm.flows.scheduler import SchedulerSingleton
+from hugegraph_llm.api.models.graph_extract_requests import (
+ GraphExtractAndImportRequest,
+ GraphExtractRequest,
+ GraphImportRequest,
+)
+from hugegraph_llm.api.models.graph_extract_responses import (
+ GraphExtractAndImportResponse,
+ GraphExtractError,
+ GraphExtractJobCreateResponse,
+ GraphExtractJobStatusResponse,
+ GraphExtractResponse,
+ GraphImportResponse,
+)
+from hugegraph_llm.services.graph_extract_jobs import (
+ GraphExtractJob,
+ GraphExtractJobStatus,
+ InMemoryGraphExtractJobStore,
+)
+from hugegraph_llm.services.graph_extract_service import (
+ FlowOutputValidationError,
+ GraphExtractService,
+ GraphImportService,
+)
from hugegraph_llm.utils.log import log
+GRAPH_EXTRACT_FLOW_OUTPUT_ERROR = "Graph extraction flow output is invalid"
+GRAPH_EXTRACT_RUNTIME_ERROR = "Graph extraction failed during execution"
+GRAPH_IMPORT_FLOW_OUTPUT_ERROR = "Graph import flow output is invalid"
+GRAPH_IMPORT_RUNTIME_ERROR = "Graph import failed during execution"
-class GraphExtractService:
- @staticmethod
- def extract_sync(req: GraphExtractRequest) -> GraphExtractResponse:
+
+def _error(code: str, message: str, phase: str, job_id: Optional[str] = None)
-> dict:
+ return GraphExtractError(code=code, message=message, phase=phase,
job_id=job_id).model_dump(exclude_none=True)
+
+
+def _job_ts(value) -> Optional[str]:
+ return value.isoformat() if value else None
+
+
+def _job_status_response(job: GraphExtractJob) ->
GraphExtractJobStatusResponse:
+ return GraphExtractJobStatusResponse(
+ job_id=job.job_id,
+ status=job.status,
+ created_at=_job_ts(job.created_at),
+ updated_at=_job_ts(job.updated_at),
+ started_at=_job_ts(job.started_at),
+ finished_at=_job_ts(job.finished_at),
+ expires_at=_job_ts(job.expires_at),
+ error=job.error,
+ )
+
+
+def _validation_message(errors) -> str:
+ details = []
+ for error in errors:
+ loc = ".".join(str(part) for part in error.get("loc", []) if part not
in {"body"})
+ msg = error.get("msg", "invalid input")
+ err_type = error.get("type", "validation_error")
+ details.append(f"{loc or 'request'}: {msg} ({err_type})")
+ return "; ".join(details) or "request validation failed"
+
+
+def _validation_error_for_path(path: str, message: str) -> dict:
+ if path.endswith("/graph/import"):
+ return _error("GRAPH_IMPORT_VALIDATION_ERROR", message, "import")
+ if path.endswith("/graph/extract-and-import"):
+ return _error("GRAPH_EXTRACT_AND_IMPORT_VALIDATION_ERROR", message,
"request")
+ return _error("GRAPH_EXTRACT_VALIDATION_ERROR", message, "request")
+
+
+class GraphExtractAPIRoute(APIRoute):
+ def get_route_handler(self):
+ original_route_handler = super().get_route_handler()
+
+ async def custom_route_handler(request: Request):
+ try:
+ return await original_route_handler(request)
+ except RequestValidationError as exc:
+ message = _validation_message(exc.errors())
+ return JSONResponse(
Review Comment:
**⚠️ `hugegraph-llm/src/hugegraph_llm/api/graph_extract_api.py:104` - Match
OpenAPI’s 422 schema to runtime responses**
**Evidence**
- The custom route returns an object-shaped structured error under `detail`,
but FastAPI still documents the default `HTTPValidationError` array schema
because no custom 422 response model is declared.
**Impact**
- Generated clients cannot reliably deserialize validation failures from the
new endpoints.
**Requested fix**
- Declare the structured error envelope under each endpoint’s 422 response
and test that the OpenAPI schema matches the runtime response.
##########
hugegraph-llm/src/hugegraph_llm/services/graph_extract_jobs.py:
##########
@@ -0,0 +1,224 @@
+# 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 queue
+import threading
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from enum import Enum
+from typing import Any, Dict, List, Optional
+
+from hugegraph_llm.api.models.graph_extract_requests import GraphExtractRequest
+from hugegraph_llm.api.models.graph_extract_responses import
GraphExtractError, GraphExtractResponse
+from hugegraph_llm.utils.log import log
+
+JOB_RUNTIME_ERROR = "Graph extraction job failed during execution"
+
+
+class GraphExtractJobStatus(str, Enum):
+ PENDING = "pending"
+ RUNNING = "running"
+ SUCCEEDED = "succeeded"
+ FAILED = "failed"
+ CANCELLED = "cancelled"
+ EXPIRED = "expired"
+
+
+@dataclass
+class GraphExtractJob:
+ job_id: str
+ request: Any
+ status: GraphExtractJobStatus
+ created_at: datetime
+ updated_at: datetime
+ started_at: Optional[datetime] = None
+ finished_at: Optional[datetime] = None
+ expires_at: Optional[datetime] = None
+ result: Optional[GraphExtractResponse] = None
+ error: Optional[GraphExtractError] = None
+
+
+class InMemoryGraphExtractJobStore:
+ def __init__(self, max_jobs: int = 100, result_ttl_seconds: int = 3600,
max_running_jobs: int = 2):
+ self.max_jobs = max_jobs
+ self.result_ttl_seconds = result_ttl_seconds
+ self.max_running_jobs = max(1, max_running_jobs)
+ self._jobs: Dict[str, GraphExtractJob] = {}
+ self._lock = threading.RLock()
+ self._queue = queue.Queue(maxsize=max_jobs)
+ self._workers_started = False
+
+ def create(self, request: Any) -> GraphExtractJob:
+ with self._lock:
+ self.cleanup()
+ if len(self._jobs) >= self.max_jobs:
+ raise ValueError("graph extraction job store is full")
+ now = self._now()
+ job = GraphExtractJob(
+ job_id=f"gex_{uuid.uuid4().hex}",
+ request=request,
+ status=GraphExtractJobStatus.PENDING,
+ created_at=now,
+ updated_at=now,
+ )
+ self._jobs[job.job_id] = job
+ return job
+
+ def get(self, job_id: str) -> Optional[GraphExtractJob]:
+ with self._lock:
+ return self._jobs.get(job_id)
Review Comment:
**⚠️ `hugegraph-llm/src/hugegraph_llm/services/graph_extract_jobs.py:84` -
Return atomic job snapshots**
**Evidence**
- `get()` returns the live mutable job after releasing the lock. Completion
updates `status` before `result` or `error`, so polling can observe `succeeded`
without a result or `failed` without an error.
**Impact**
- Concurrent polling can return `GRAPH_EXTRACT_JOB_RESULT_MISSING` or
trigger an unstructured exception despite successful job completion.
**Requested fix**
- Copy an immutable snapshot while holding the lock, or resolve status and
result atomically inside the store. Add a completion-versus-polling concurrency
test.
--
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]
