LRriver commented on code in PR #361:
URL: https://github.com/apache/hugegraph-ai/pull/361#discussion_r3379133067
##########
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:
Fixed in 7850be7. Empty chunks now return without LLM calls and keep
max_parallel_chunks metadata positive, with regression coverage.
##########
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:
Not changed in this pass. The current LLM wrappers used here do not maintain
per-request mutable response buffers in PropertyGraphExtract; serializing
extract_property_graph_by_llm behind one lock would effectively disable the
chunk-level parallelism this API is adding. The API also bounds per-request
parallelism by config and request. If a future provider wrapper proves
non-thread-safe, the better fix would be provider-local isolation rather than a
global lock in the extraction operator.
##########
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:
Fixed in 7850be7. Added response_model declarations for job, import, and
extract-and-import endpoints, with route registration coverage.
##########
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:
Fixed in 7850be7. Added response_model declarations for job, import, and
extract-and-import endpoints, with route registration coverage.
##########
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:
Fixed in 7850be7. Empty input messages now mention vertices, edges, and
triples. Schema-free mode rejects vertices or edges, and schema mode rejects
triples so mixed inputs are not silently dropped. Added regression coverage.
--
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]
