Copilot commented on code in PR #361:
URL: https://github.com/apache/hugegraph-ai/pull/361#discussion_r3378892619
##########
hugegraph-llm/src/hugegraph_llm/operators/llm_op/property_graph_extract.py:
##########
@@ -91,15 +98,28 @@ def run(self, context: Dict[str, Any]) -> Dict[str,
List[Any]]:
if "edges" not in context:
context["edges"] = []
items = []
- for chunk in chunks:
- proceeded_chunk = self.extract_property_graph_by_llm(schema, chunk)
+ try:
+ max_parallel_chunks = max(1,
int(context.get("max_parallel_chunks") or self.max_parallel_chunks))
+ except (TypeError, ValueError):
+ max_parallel_chunks = max(1, self.max_parallel_chunks)
+ chunk_count = len(chunks)
+ worker_count = min(max_parallel_chunks, chunk_count)
+ context["max_parallel_chunks"] = worker_count
+ if worker_count <= 1:
+ proceeded_chunks = [self.extract_property_graph_by_llm(schema,
chunk) for chunk in chunks]
+ else:
+ with ThreadPoolExecutor(max_workers=worker_count) as executor:
+ proceeded_chunks = list(
+ executor.map(lambda chunk:
self.extract_property_graph_by_llm(schema, chunk), chunks)
+ )
Review Comment:
`worker_count` becomes `0` when `chunks` is empty, and the code stores
`context["max_parallel_chunks"] = 0`. This conflicts with the rest of the
codebase/API contract where parallelism is treated as `>= 1`, and can produce
confusing metadata downstream. Consider either early-returning when
`chunk_count == 0` (and keeping `max_parallel_chunks` unset/None), or setting
`context["max_parallel_chunks"]` to `max(1, worker_count)` when there is any
work to do.
##########
hugegraph-llm/src/hugegraph_llm/api/graph_extract_api.py:
##########
@@ -15,55 +15,270 @@
# 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 GraphExtractService,
GraphImportService
from hugegraph_llm.utils.log import log
-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,
+ )
+
+
+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:
+ return JSONResponse(
+ status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+ content={
+ "detail": _error(
+ "GRAPH_EXTRACT_VALIDATION_ERROR",
+ str(exc),
+ "request",
+ )
+ },
+ )
+
+ 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 ValueError as exc:
+ log.error("Graph extraction request failed validation: %s", exc)
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ detail=_error("GRAPH_EXTRACT_INVALID_FLOW_OUTPUT", str(exc),
"extract"),
+ ) 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", str(exc), "extract"),
+ ) from exc
+
+ @router.post("/graph/extract/jobs", status_code=status.HTTP_202_ACCEPTED)
+ def create_graph_extract_job(
+ req: GraphExtractRequest,
+ ) -> GraphExtractJobCreateResponse:
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,
+ 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)
+ 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)
+ 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),
)
- 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)
+ 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="An unexpected error occurred during graph extraction.",
- ) from e
+ 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)
+ 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)
-def graph_extract_http_api(router: APIRouter):
- @router.post("/graph/extract", status_code=status.HTTP_200_OK,
response_model=GraphExtractResponse)
- def graph_extract_api(req: GraphExtractRequest):
- return GraphExtractService.extract_sync(req)
+ @router.post("/graph/import", status_code=status.HTTP_200_OK)
+ def graph_import_api(req: GraphImportRequest) -> GraphImportResponse:
Review Comment:
These endpoints don’t declare `response_model=...` in the decorator, which
means FastAPI won’t apply response-model validation/serialization guarantees or
generate as precise OpenAPI output as it could. Consider adding
`response_model=GraphImportResponse` and
`response_model=GraphExtractAndImportResponse` (and similarly for job
endpoints) to keep the API contract explicit and consistent with
`/graph/extract`.
##########
hugegraph-llm/src/hugegraph_llm/operators/hugegraph_op/commit_to_hugegraph.py:
##########
@@ -27,31 +27,51 @@
class Commit2Graph:
- def __init__(self):
+ def __init__(self, graph_config=None):
+ graph_config = graph_config or {}
self.client = PyHugeClient(
- url=huge_settings.graph_url,
- graph=huge_settings.graph_name,
- user=huge_settings.graph_user,
- pwd=huge_settings.graph_pwd,
- graphspace=huge_settings.graph_space,
+ url=graph_config.get("url") or huge_settings.graph_url,
+ graph=graph_config.get("graph") or huge_settings.graph_name,
+ user=graph_config.get("user") or huge_settings.graph_user,
+ pwd=graph_config.get("pwd") or huge_settings.graph_pwd,
+ graphspace=graph_config.get("gs") or huge_settings.graph_space,
)
self.schema = self.client.schema()
+ def _empty_import_result(self, vertices=None, edges=None, triples=None) ->
Dict[str, Any]:
+ return {
+ "vertices_attempted": len(vertices or []),
+ "vertices_created": 0,
+ "vertices_skipped": 0,
+ "edges_attempted": len(edges or []),
+ "edges_created": 0,
+ "edges_skipped": 0,
+ "triples_attempted": len(triples or []),
+ "triples_created": 0,
+ "triples_skipped": 0,
+ "errors": [],
+ }
+
def run(self, data: dict) -> Dict[str, Any]:
schema = data.get("schema")
- vertices = data.get("vertices", [])
- edges = data.get("edges", [])
- if not vertices and not edges:
+ vertices = data.get("vertices", []) or []
+ edges = data.get("edges", []) or []
+ triples = data.get("triples", []) or []
+ if not vertices and not edges and not triples:
log.critical("(Loading) Both vertices and edges are empty. Please
check the input data again.")
raise ValueError("Both vertices and edges input are empty.")
Review Comment:
The raised/logged message says “Both vertices and edges are empty” but the
preceding condition also checks `triples`, and the schema-present branch
performs a second (slightly different) emptiness check that ignores `triples`.
To reduce confusion for callers (and avoid duplicated logic), consider
consolidating these checks and updating the message to reflect the actual
accepted inputs (e.g., “vertices/edges/triples are empty” or “property-graph
vertices/edges required when schema is provided”).
##########
hugegraph-llm/src/hugegraph_llm/operators/hugegraph_op/commit_to_hugegraph.py:
##########
@@ -27,31 +27,51 @@
class Commit2Graph:
- def __init__(self):
+ def __init__(self, graph_config=None):
+ graph_config = graph_config or {}
self.client = PyHugeClient(
- url=huge_settings.graph_url,
- graph=huge_settings.graph_name,
- user=huge_settings.graph_user,
- pwd=huge_settings.graph_pwd,
- graphspace=huge_settings.graph_space,
+ url=graph_config.get("url") or huge_settings.graph_url,
+ graph=graph_config.get("graph") or huge_settings.graph_name,
+ user=graph_config.get("user") or huge_settings.graph_user,
+ pwd=graph_config.get("pwd") or huge_settings.graph_pwd,
+ graphspace=graph_config.get("gs") or huge_settings.graph_space,
)
self.schema = self.client.schema()
+ def _empty_import_result(self, vertices=None, edges=None, triples=None) ->
Dict[str, Any]:
+ return {
+ "vertices_attempted": len(vertices or []),
+ "vertices_created": 0,
+ "vertices_skipped": 0,
+ "edges_attempted": len(edges or []),
+ "edges_created": 0,
+ "edges_skipped": 0,
+ "triples_attempted": len(triples or []),
+ "triples_created": 0,
+ "triples_skipped": 0,
+ "errors": [],
+ }
+
def run(self, data: dict) -> Dict[str, Any]:
schema = data.get("schema")
- vertices = data.get("vertices", [])
- edges = data.get("edges", [])
- if not vertices and not edges:
+ vertices = data.get("vertices", []) or []
+ edges = data.get("edges", []) or []
+ triples = data.get("triples", []) or []
+ if not vertices and not edges and not triples:
log.critical("(Loading) Both vertices and edges are empty. Please
check the input data again.")
raise ValueError("Both vertices and edges input are empty.")
if not schema:
# TODO: ensure the function works correctly (update the logic
later)
- self.schema_free_mode(data.get("triples", []))
+ import_result = self.schema_free_mode(triples)
log.warning("Using schema_free mode, could try schema_define mode
for better effect!")
else:
+ if not vertices and not edges:
+ log.critical("(Loading) Both vertices and edges are empty.
Please check the input data again.")
+ raise ValueError("Both vertices and edges input are empty.")
Review Comment:
The raised/logged message says “Both vertices and edges are empty” but the
preceding condition also checks `triples`, and the schema-present branch
performs a second (slightly different) emptiness check that ignores `triples`.
To reduce confusion for callers (and avoid duplicated logic), consider
consolidating these checks and updating the message to reflect the actual
accepted inputs (e.g., “vertices/edges/triples are empty” or “property-graph
vertices/edges required when schema is provided”).
##########
hugegraph-llm/src/hugegraph_llm/api/graph_extract_api.py:
##########
@@ -15,55 +15,270 @@
# 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 GraphExtractService,
GraphImportService
from hugegraph_llm.utils.log import log
-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,
+ )
+
+
+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:
+ return JSONResponse(
+ status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+ content={
+ "detail": _error(
+ "GRAPH_EXTRACT_VALIDATION_ERROR",
+ str(exc),
+ "request",
+ )
+ },
+ )
Review Comment:
Using `str(exc)` for request validation errors tends to be verbose and
unstable (it can vary across FastAPI/Pydantic versions) and may not reliably
highlight which field failed in a user-friendly way. A more actionable error
payload would derive the message from `exc.errors()` (e.g., include `loc`,
`msg`, and `type`) and format it into a predictable structure.
##########
hugegraph-llm/src/hugegraph_llm/operators/llm_op/property_graph_extract.py:
##########
@@ -91,15 +98,28 @@ def run(self, context: Dict[str, Any]) -> Dict[str,
List[Any]]:
if "edges" not in context:
context["edges"] = []
items = []
- for chunk in chunks:
- proceeded_chunk = self.extract_property_graph_by_llm(schema, chunk)
+ try:
+ max_parallel_chunks = max(1,
int(context.get("max_parallel_chunks") or self.max_parallel_chunks))
+ except (TypeError, ValueError):
+ max_parallel_chunks = max(1, self.max_parallel_chunks)
+ chunk_count = len(chunks)
+ worker_count = min(max_parallel_chunks, chunk_count)
+ context["max_parallel_chunks"] = worker_count
+ if worker_count <= 1:
+ proceeded_chunks = [self.extract_property_graph_by_llm(schema,
chunk) for chunk in chunks]
+ else:
+ with ThreadPoolExecutor(max_workers=worker_count) as executor:
+ proceeded_chunks = list(
+ executor.map(lambda chunk:
self.extract_property_graph_by_llm(schema, chunk), chunks)
+ )
Review Comment:
Parallelizing `extract_property_graph_by_llm()` via threads can be unsafe if
the underlying `llm` client maintains mutable shared state (e.g., shared
session objects, retry/backoff counters, streaming buffers). If the LLM client
is not explicitly thread-safe, this can lead to intermittent failures or
cross-talk between requests. A tangible mitigation is to serialize calls behind
a lock (while still parallelizing JSON parsing/normalization), or ensure each
worker uses an isolated client instance (e.g., thread-local client
construction).
##########
hugegraph-llm/src/hugegraph_llm/api/graph_extract_api.py:
##########
@@ -15,55 +15,270 @@
# 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 GraphExtractService,
GraphImportService
from hugegraph_llm.utils.log import log
-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,
+ )
+
+
+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:
+ return JSONResponse(
+ status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+ content={
+ "detail": _error(
+ "GRAPH_EXTRACT_VALIDATION_ERROR",
+ str(exc),
+ "request",
+ )
+ },
+ )
+
+ 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 ValueError as exc:
+ log.error("Graph extraction request failed validation: %s", exc)
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ detail=_error("GRAPH_EXTRACT_INVALID_FLOW_OUTPUT", str(exc),
"extract"),
+ ) 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", str(exc), "extract"),
+ ) from exc
+
+ @router.post("/graph/extract/jobs", status_code=status.HTTP_202_ACCEPTED)
+ def create_graph_extract_job(
+ req: GraphExtractRequest,
+ ) -> GraphExtractJobCreateResponse:
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,
+ 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)
+ 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)
+ 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),
)
- 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)
+ 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="An unexpected error occurred during graph extraction.",
- ) from e
+ 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)
+ 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)
-def graph_extract_http_api(router: APIRouter):
- @router.post("/graph/extract", status_code=status.HTTP_200_OK,
response_model=GraphExtractResponse)
- def graph_extract_api(req: GraphExtractRequest):
- return GraphExtractService.extract_sync(req)
+ @router.post("/graph/import", status_code=status.HTTP_200_OK)
+ 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:
+ return graph_import_service.import_graph(req)
+ 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", str(exc), "import"),
+ ) from exc
+
+ @router.post("/graph/extract-and-import", status_code=status.HTTP_200_OK)
+ def graph_extract_and_import_api(req: GraphExtractAndImportRequest) ->
GraphExtractAndImportResponse:
Review Comment:
These endpoints don’t declare `response_model=...` in the decorator, which
means FastAPI won’t apply response-model validation/serialization guarantees or
generate as precise OpenAPI output as it could. Consider adding
`response_model=GraphImportResponse` and
`response_model=GraphExtractAndImportResponse` (and similarly for job
endpoints) to keep the API contract explicit and consistent with
`/graph/extract`.
--
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]
