feat(llm): schema-aware enhanced strategy for /graph/extract (#74)

docs/quality/schema-based-graph-extract-usage.md

@@ -0,0 +1,208 @@
+# Schema-based Graph Extraction — Usage Guide
+
+The enhanced graph extraction strategy is an **opt-in** upgrade to the
+`/graph/extract` API and the `GraphExtractFlow` pipeline. This document
+covers how to enable it, when it helps, and its known limitations.
+
+## Quickstart
+
+### HTTP API
+
+Existing callers keep working unchanged (baseline is still the default):
+
+```bash
+curl -X POST http://localhost:8001/graph/extract \
+  -H "Content-Type: application/json" \
+  -d '{
+    "texts": ["Tom Hanks starred in Forrest Gump."],
+    "graph_schema": "movie_graph"
+  }'
+```
+
+To enable the enhanced strategy, add `extract_strategy: "enhanced"`:
+
+```bash
+curl -X POST http://localhost:8001/graph/extract \
+  -H "Content-Type: application/json" \
+  -d '{
+    "texts": ["Tom Hanks starred in Forrest Gump."],
+    "graph_schema": "movie_graph",
+    "extract_strategy": "enhanced",
+    "include_meta": true
+  }'
+```
+
+Enhanced-only fields in the response `meta` block:
+
+| Field | Type | Description |
+| --- | --- | --- |
+| `extract_strategy` | `"enhanced"` | Echoes the strategy in effect. Absent under baseline. |
+| `chunk_count` | int | Number of chunks the input was split into. |
+| `call_count` | int | Number of LLM invocations (== chunk_count). |
+| `token_usage` | str | Placeholder `"unavailable"` until the LLM adapter surfaces token metadata. |
+| `structured_warnings` | list | Machine-readable warning records — see below. |
+| `quality_metrics` | dict | Per-run quality aggregate — see below. |
+
+Passing `include_debug: true` additionally surfaces `meta.debug_info`
+with per-chunk raw LLM output (truncated to 2 KB) plus normalized item
+counts. Use this in staging for prompt tuning; keep it off in production
+to avoid response bloat.
+
+### Python SDK
+
+```python
+from hugegraph_llm.flows import FlowName
+from hugegraph_llm.flows.scheduler import SchedulerSingleton
+
+result = SchedulerSingleton.get_instance().schedule_flow(
+    FlowName.GRAPH_EXTRACT,
+    schema=my_schema,
+    texts=["…"],
+    example_prompt="…",
+    extract_type="…",
+    extract_strategy="enhanced",   # opt-in
+    include_debug=False,           # optional, default False
+)
+```
+
+The `result` string is a JSON payload compatible with the baseline
+schema; enhanced-only fields appear as additional top-level keys.
+
+## Structured Warnings
+
+Each warning has this shape:
+
+```json
+{
+  "code": "MISSING_PRIMARY_KEY",
+  "item_type": "vertex",
+  "reason": "vertex 'Person' has no value for primary key 'name'",
+  "chunk_id": 0
+}
+```
+
+Warning codes are grouped by their producer stage (parser, normalizer,
+assembler, quality gate). Full list in
+[`property_graph_extract_enhanced/warnings.py`](../../hugegraph-llm/src/hugegraph_llm/operators/llm_op/property_graph_extract_enhanced/warnings.py).
+
+Common codes:
+
+* `MISSING_PRIMARY_KEY` — vertex is dropped because a PK value is missing.
+* `PROPERTY_COERCED` — a property value was converted to its schema type
+  (e.g. `"62"` → `62`). Non-fatal.
+* `UNKNOWN_VERTEX_LABEL` / `UNKNOWN_EDGE_LABEL` — item dropped for using a
+  label not in the schema.
+* `ENDPOINT_INCOMPATIBLE` — edge direction violates the schema
+  `source_label` / `target_label`.
+* `PENDING_ENDPOINT_UNRESOLVED` — edge references a vertex that no chunk
+  ever defined.
+* `JSON_DECODE_FAILED` — the LLM's chunk output could not be parsed.
+
+## Quality Metrics
+
+Each enhanced run reports a `quality_metrics` block with 11 fields
+including `parse_success_rate`, `endpoint_repair_rate`,
+`property_coerce_rate`, `dropped_item_count`, and
+`warning_code_distribution`. Full field list in
+[`property_graph_extract_enhanced/quality_gate.py`](../../hugegraph-llm/src/hugegraph_llm/operators/llm_op/property_graph_extract_enhanced/quality_gate.py).
+
+Use these as live signals in staging: a sudden spike in
+`property_coerce_rate` usually means the prompt drifted and started
+emitting stringified numerics; a spike in `endpoint_repair_rate` means
+chunking may be too aggressive.
+
+## Applicable Scenarios
+
+The enhanced strategy helps most when **any** of these hold:
+
+* **Multi-chunk documents.** The strategy's `DocumentGraphAssembler`
+  merges vertices across chunks and repairs cross-chunk edge endpoints —
+  the biggest F1 win in the benchmark comes from here.
+* **Strict schema with canonical id requirements.** If HugeGraph's
+  `PRIMARY_KEY` id strategy is in use, enhanced ensures every emitted
+  vertex has a valid canonical id and drops PK-missing candidates before
+  they reach the commit stage.
+* **Typed properties beyond TEXT.** INT/LONG/FLOAT/BOOLEAN/DATE columns
+  benefit from enhanced's best-effort coercion. Baseline's `filter_item`
+  is key-only and does not repair types.
+* **Prompts under drift.** Structured warnings + quality metrics give
+  operators a signal to detect when a previously-good prompt starts
+  regressing.
+
+Baseline remains a fine default when:
+
+* Every schema property is TEXT.
+* Documents fit in a single chunk.
+* No canonical id is needed downstream (e.g., `id_strategy = "AUTOMATIC"`).
+* You want to minimize LLM prompt tokens (see limitation below).
+
+## Known Limitations
+
+* **Prompt-token overhead.** The enhanced strategy appends a
+  `constraint_block` listing every schema label to the LLM prompt on
+  every chunk (~500-800 extra tokens for a small schema, more for large
+  ones). This is real cost on paid APIs. For a 100-chunk document this
+  is 50k-80k extra tokens; at DeepSeek Chat pricing (2026-07 rates) that
+  is single-digit cents but non-zero.
+
+* **Coercion is best-effort, not lossless.** `"1994"` → `1994` (INT) is
+  safe. `"1.5"` → `1` (INT) drops information — enhanced emits
+  `PROPERTY_COERCED` with the reason string. Review warnings before
+  concluding the graph is clean.
+
+* **No cross-request state.** The alias table is scoped to a single
+  request. Two entities with the same schema-canonical id extracted from
+  different `/graph/extract` calls still land as two records — this is
+  the storage layer's job, not the extractor's.
+
+* **Set-based F1.** The offline evaluator scores against deduplicated
+  predictions. Baseline's raw-count duplication cost does not surface in
+  F1; if you care about commit load, read the raw counts from the
+  benchmark's per-scenario `predicted_count_raw`.
+
+* **No streaming.** Enhanced processes all chunks synchronously and
+  produces the assembled graph at the end. For very large documents,
+  consider chunking upstream and calling `/graph/extract` per batch.
+
+* **No knowledge-base entity resolution.** Enhanced can resolve aliases
+  *within* a single request (e.g. `"He"` → `"Tom Hanks"` when both
+  appear in different chunks) via its `DocumentGraphAssembler`. It
+  cannot resolve *out-of-corpus* aliases: if the LLM extracts
+  `"Thomas Jeffrey Hanks"` from an article's opening line and the
+  downstream canonical id is `"Tom Hanks"`, both records land as
+  separate vertices. Solving this needs either LLM-side alignment
+  (system prompt telling the model to emit shortest canonical names) or
+  an external entity-resolution step (string similarity + gazetteer /
+  Wikidata / customer-supplied alias table). Both are out of scope for
+  Issue #74. See the effect report's live-benchmark failure analysis
+  for concrete numbers.
+
+## When enhanced does **not** help
+
+The live benchmark on 8 public Wikipedia corpora showed enhanced
+regressing F1 on 3 of the 8 corpora (baseline already strong; enhanced's
+stricter deduplication merged near-duplicate film titles that were
+actually different films). Consider running baseline first and switching
+to enhanced only if `warnings` count is high or if quality is
+consistently under target. A production deployment can gate this on
+first-pass baseline warning counts.
+
+## Migration Notes
+
+None required. Existing baseline callers keep byte-compatible responses
+as long as they do not send `extract_strategy` or `include_debug`.
+
+If you are the first team to opt-in on a graph, run one enhanced request
+with `include_meta=true` and inspect `meta.structured_warnings` /
+`meta.quality_metrics` — the warning codes indicate any places where the
+prompt is drifting from the schema, and you should tighten those before
+switching all traffic.
+
+## References
+
+* [Effect Report](./schema-based-graph-extract-report.md) — quality and
+  latency benchmark numbers.
+* [Test Taxonomy](./test-taxonomy.md) — where the benchmark fits in the
+  overall test pyramid.
+* Source:
+  [`hugegraph_llm/operators/llm_op/property_graph_extract_enhanced/`](../../hugegraph-llm/src/hugegraph_llm/operators/llm_op/property_graph_extract_enhanced/)

hugegraph-llm/src/hugegraph_llm/api/graph_extract_api.py

@@ -41,17 +41,50 @@ def extract_sync(req: GraphExtractRequest) -> GraphExtractResponse:
                 language=req.language,
                 split_type=req.split_type,
                 client_config=req.client_config,
+                extract_strategy=req.extract_strategy,
+                include_debug=req.include_debug,
             )
             raw = json.loads(result_str)
             warnings = [raw.pop("warning")] if "warning" in raw else []
+            # Enhanced-strategy fields — the flow layer only emits these when
+            # extract_strategy == "enhanced", so their absence signals baseline.
+            extract_strategy = raw.pop("extract_strategy", "baseline")
+            chunk_count = raw.pop("chunk_count", None)
+            call_count = raw.pop("call_count", None)
+            structured_warnings = raw.pop("structured_warnings", None)
+            quality_metrics = raw.pop("quality_metrics", None)
+            debug_info = raw.pop("debug_info", None)
+
             result = {"vertices": raw.get("vertices", []), "edges": raw.get("edges", [])}
+
+            # Surface a short top-level summary of the structured warning count
+            # so callers reading the legacy warnings[] can still notice enhanced
+            # activity without parsing meta.
+            if structured_warnings:
+                warnings.append(f"enhanced graph extraction generated {len(structured_warnings)} structured warning(s)")
+
             meta = {}
             if req.include_meta:
                 meta = {
                     "vertex_count": len(result["vertices"]),
                     "edge_count": len(result["edges"]),
                     "text_count": len(req.texts),
                 }
+                if extract_strategy == "enhanced":
+                    meta["extract_strategy"] = extract_strategy
+                    if chunk_count is not None:
+                        meta["chunk_count"] = chunk_count
+                    if call_count is not None:
+                        meta["call_count"] = call_count
+                    meta["token_usage"] = "unavailable"
+                    if structured_warnings is not None:
+                        meta["structured_warnings"] = structured_warnings
+                    if quality_metrics is not None:
+                        meta["quality_metrics"] = quality_metrics
+            # include_debug adds debug_info regardless of include_meta so callers
+            # can grab the debug payload without opting in to counts.
+            if req.include_debug and debug_info is not None:
+                meta["debug_info"] = debug_info
             return GraphExtractResponse(result=result, warnings=warnings, meta=meta)
         except HTTPException:
             raise

hugegraph-llm/src/hugegraph_llm/api/models/graph_extract_requests.py

@@ -45,6 +45,20 @@ class GraphExtractRequest(BaseModel):
     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.")
+    extract_strategy: Literal["baseline", "enhanced"] = Query(
+        "baseline",
+        description=(
+            "Which extraction strategy to use. 'baseline' preserves the existing behavior; "
+            "'enhanced' opts into the schema-aware graph quality layer."
+        ),
+    )
+    include_debug: bool = Query(
+        False,
+        description=(
+            "When true, the enhanced strategy attaches per-chunk debug details under 'meta.debug_info'. "
+            "Ignored by the baseline strategy."
+        ),
+    )
     client_config: Optional[GraphExtractClientConfig] = Field(None, description="Request-scoped HugeGraph connection.")
 
     @field_validator("texts")

hugegraph-llm/src/hugegraph_llm/flows/graph_extract.py

@@ -14,6 +14,7 @@
 #  limitations under the License.
 
 import json
+from typing import Any, Dict
 
 from pycgraph import GPipeline
 
@@ -56,6 +57,9 @@ def prepare(
         prepared_input.example_prompt = example_prompt
         prepared_input.schema = schema
         prepared_input.extract_type = extract_type
+        # Enhanced graph extraction opt-in flags (baseline behavior stays unchanged when defaults hold).
+        prepared_input.extract_strategy = kwargs.get("extract_strategy", "baseline")
+        prepared_input.include_debug = bool(kwargs.get("include_debug", False))
         client_config = kwargs.get("client_config")
         if client_config:
             # URL stays server-controlled; only identity/graphspace are request-scoped.
@@ -109,20 +113,30 @@ def post_deal(self, pipeline=None, **kwargs):
         vertices = res.get("vertices", [])
         edges = res.get("edges", [])
         chunk_count = len(res.get("chunks", []))
+        extract_strategy = res.get("extract_strategy", "baseline")
         log.info("Graph extraction chunk_count: %s", chunk_count)
+        if extract_strategy == "enhanced":
+            log.info("Enhanced graph extraction strategy in use.")
+        payload: Dict[str, Any] = {"vertices": vertices, "edges": edges}
+        # Enhanced-strategy fields flow to the API layer, which decides which
+        # of them to surface under `meta` based on include_meta / include_debug.
+        # Baseline stays byte-compatible: none of these keys appear.
+        if extract_strategy == "enhanced":
+            payload["extract_strategy"] = extract_strategy
+            payload["chunk_count"] = chunk_count
+            call_count = res.get("call_count")
+            if call_count is not None:
+                payload["call_count"] = call_count
+            structured_warnings = res.get("structured_warnings")
+            if structured_warnings is not None:
+                payload["structured_warnings"] = structured_warnings
+            quality_metrics = res.get("quality_metrics")
+            if quality_metrics is not None:
+                payload["quality_metrics"] = quality_metrics
+            debug_info = res.get("debug_info")
+            if debug_info is not None:
+                payload["debug_info"] = debug_info
         if not vertices and not edges:
             log.info("Please check the schema.(The schema may not match the Doc)")
-            return json.dumps(
-                {
-                    "vertices": vertices,
-                    "edges": edges,
-                    "warning": "The schema may not match the Doc",
-                },
-                ensure_ascii=False,
-                indent=2,
-            )
-        return json.dumps(
-            {"vertices": vertices, "edges": edges},
-            ensure_ascii=False,
-            indent=2,
-        )
+            payload["warning"] = "The schema may not match the Doc"
+        return json.dumps(payload, ensure_ascii=False, indent=2)

hugegraph-llm/src/hugegraph_llm/nodes/llm_node/extract_info.py

@@ -38,10 +38,17 @@ def node_init(self):
         example_prompt = self.wk_input.example_prompt
         extract_type = self.wk_input.extract_type
         self.extract_type = extract_type
+        extract_strategy = self.wk_input.extract_strategy or "baseline"
+        include_debug = bool(self.wk_input.include_debug)
         if extract_type == "triples":
             self.info_extract = InfoExtract(llm, example_prompt)
         elif extract_type == "property_graph":
-            self.property_graph_extract = PropertyGraphExtract(llm, example_prompt)
+            self.property_graph_extract = PropertyGraphExtract(
+                llm,
+                example_prompt,
+                extract_strategy=extract_strategy,
+                include_debug=include_debug,
+            )
         else:
             return CStatus(-1, f"Unsupported extract_type: {extract_type}")
         return super().node_init()