diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a20cf2a..e323bec2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,23 @@ # Changelog +## [2.0.2] + +### Added +- **The Migration Assistant reports graph data-integrity health.** Alongside the query check, it now shows how many vertices are missing embeddings (by type) and how many communities have a placeholder or empty summary. +- **Targeted regeneration from the Migration Assistant.** Missing embeddings can be re-embedded, and communities with placeholder/empty summaries can be re-summarized, for just the affected items — without running a full rebuild. Items whose source content is unusable are reported as needing a rebuild. + +### Changed +- **Agentic tool-calling support is detected at runtime.** Whether the chat model can drive Agentic mode is now confirmed by a lightweight runtime probe (checked once and cached in memory, re-checked after a restart) instead of a fixed model list, so current and future tool-calling models — including new providers — enable Agentic mode automatically, and a model that can't tool-call falls back to the classic engine until its configuration changes. +- **The planned agent falls back to document search when a structured query returns nothing.** If the planned agent's structured query returns no rows, it now runs a hybrid document search before answering, so an empty structured result no longer produces a non-answer; it follows the same fallback setting as the classic engine. +- **Clearer message when an answer can't be generated.** When the assistant is unable to produce an answer, it now suggests rephrasing the question to be more specific and contacting an administrator if the problem continues, instead of a bare failure notice. +- **Token cost rates can be set per service.** LLM Config accepts optional input/output USD-per-1M-token rates for the completion, chatbot, and multimodal services; when set, Est. Cost uses them so cost is accurate for models the pricing library doesn't recognize (otherwise it keeps the library's pricing). + +### Fixed +- **Knowledge-graph rebuild no longer fails on document names containing parentheses.** Document names with `(` or `)` are normalized without dropping the rest of the name, so rebuilds complete instead of erroring while creating chunks. +- **PDF chart labels and table numbers survive extraction.** Picture-text blocks are kept intact, stacked and comma-grouped chart values are no longer glued together, and pages with unreadable embedded fonts are recovered from a page image — so figures and tables reach the graph with their numbers intact. +- **Newer Gemini models are available in agentic chat.** Gemini 3.x and later families are recognized as tool-calling models, so agentic mode works with them instead of silently falling back to classic chat. +- **Chat Stop button behaves correctly.** Stopping a response just as it finishes no longer sends a stray empty message, and the Stop button is the right size in the light theme. + ## [2.0.1] ### Changed diff --git a/README.md b/README.md index 70aa1ca5..2b7f6fa8 100644 --- a/README.md +++ b/README.md @@ -515,7 +515,7 @@ Copy the below code into `configs/server_config.json`. You shouldn’t need to c | `retrieval_include_entity` | bool \| null | `null` (auto) | Whether retriever queries include the generic `Entity` vertex alongside domain types. When unset, the server uses `false` if a domain schema exists and `true` otherwise. Set explicitly to override. | | `schema_max_sample_files` | int | `5` | Maximum number of sample documents accepted by the *Generate from sample documents* path on the *Initialize Knowledge Graph* dialog. | | `schema_max_total_mb` | int | `50` | Combined upload cap (MB) across all sample files for schema extraction. Bounds the content sent to the LLM. A single file may use the full budget; no separate per-file cap. | -| `enable_router_fallback` | bool | `true` | When the function-call or Cypher path fails after 3 retries, fall back to vector search instead of failing the query. | +| `enable_router_fallback` | bool | `true` | When a structured query returns no result, fall back to vector search instead of failing the query. Applies to the classic engine (after the function-call or Cypher path fails its retries) and to the planned agent (when a structural query returns no rows). | | `chunker_config` | object | `{}` | Chunker-specific settings (see sub-parameters below). All settings are saved regardless of which chunker is selected as default. | | ↳ `chunk_size` | int | `2048` | Maximum number of characters per chunk. Used by `character`, `markdown`, `html`, and `recursive` chunkers. Larger values produce fewer, bigger chunks; smaller values produce more, finer-grained chunks. | | ↳ `overlap_size` | int | 1/8 of `chunk_size` | Number of overlapping characters between consecutive chunks. Used by `character`, `markdown`, `html`, and `recursive` chunkers. More overlap preserves cross-chunk context but increases total chunk count. Set to `0` for no overlap. | diff --git a/VERSION b/VERSION index 10bf840e..e9307ca5 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -2.0.1 \ No newline at end of file +2.0.2 diff --git a/common/chunkers/structured.py b/common/chunkers/structured.py index aa3ec26e..0980140d 100644 --- a/common/chunkers/structured.py +++ b/common/chunkers/structured.py @@ -140,15 +140,28 @@ class Element: # pymupdf4llm artifacts: # • "==> picture [WxH] intentionally omitted <==" — image dropped (skip line) # • "----- Start of picture text -----" / "----- End of picture text -----" -# bracket OCR'd content inside an image; we fold the body into the figure -# so chart-internal labels stay with the image chunk. +# (markdown) or ```` (HTML-comment form) +# bracket pymupdf4llm picture-text inside a figure; we fold the body into +# the figure so chart-internal labels stay with the image chunk. _MD_PICTURE_OMITTED = re.compile(r"^\s*\*+\s*==>\s*picture\b.*intentionally omitted\s*<==\s*\*+.*$", re.IGNORECASE) -_MD_PICTURE_TEXT_START = re.compile(r"^\s*\*+\s*-+\s*Start of picture text\s*-+\s*\*+\s*()?\s*$", re.IGNORECASE) -_MD_PICTURE_TEXT_END = re.compile(r"^\s*\*+\s*-+\s*End of picture text\s*-+\s*\*+\s*()?\s*$", re.IGNORECASE) +_MD_PICTURE_TEXT_START = re.compile( + r"^\s*(?:\*+\s*-+\s*Start of picture text\s*-+\s*\*+|" + r")\s*(?:)?\s*$", + re.IGNORECASE, +) +_MD_PICTURE_TEXT_END = re.compile( + r"^\s*(?:\*+\s*-+\s*End of picture text\s*-+\s*\*+|" + r")\s*(?:)?\s*$", + re.IGNORECASE, +) # Inline variant of the End marker: the picture-text body can arrive as a # single
-joined line with the marker on its tail, so it is not always # line-anchored. Searched anywhere in a line to terminate the block. -_MD_PICTURE_TEXT_END_INLINE = re.compile(r"\*+\s*-+\s*End of picture text\s*-+\s*\*+\s*(?:)?", re.IGNORECASE) +_MD_PICTURE_TEXT_END_INLINE = re.compile( + r"(?:\*+\s*-+\s*End of picture text\s*-+\s*\*+|" + r")\s*(?:)?", + re.IGNORECASE, +) def _flush_prose(buf: List[str], heading: Optional[str], page: Optional[int], out: List[Element]) -> None: @@ -312,18 +325,14 @@ def markdown_to_elements(md: str, page: Optional[int] = None) -> List[Element]: i += 1 continue - # 5b. Other HTML comments (chunk markers etc.) — skip. - if _MD_HTML_COMMENT.match(line): - i += 1 - continue - # 5c. pymupdf4llm "==> picture ... intentionally omitted <==" — drop. if _MD_PICTURE_OMITTED.match(line): i += 1 continue - # 5d. pymupdf4llm picture-text block: ----- Start ... End of picture - # text ----- wraps OCR'd content (chart axis labels, legends). + # 5d. pymupdf4llm picture-text block (markdown dash markers OR HTML + # comments). Must run before the generic HTML-comment skip so + # ```` is not dropped. # Fold the body into the immediately preceding figure when # present so chart-internal text travels with the image. if _MD_PICTURE_TEXT_START.match(line): @@ -361,6 +370,12 @@ def markdown_to_elements(md: str, page: Optional[int] = None) -> List[Element]: out.append(Element(kind="figure", text=body, heading=heading, page=page)) continue + # 5b. Other HTML comments (chunk markers etc.) — skip. + # Picture-text comments are handled above. + if _MD_HTML_COMMENT.match(line): + i += 1 + continue + # 6. Blank line — flush current prose paragraph. if not stripped: _flush_prose(prose_buf, heading, page, out) diff --git a/common/config.py b/common/config.py index d229d694..d6144ca5 100644 --- a/common/config.py +++ b/common/config.py @@ -187,8 +187,13 @@ def resolve_llm_services(llm_cfg: dict) -> dict: embedding_provider = embedding.get("embedding_model_service", "").lower() completion_provider = completion.get("llm_service", "").lower() if embedding_provider and embedding_provider == completion_provider: - # Identity/schema keys that belong to the embedding service itself - embedding_own_keys = {"embedding_model_service", "model_name", "authentication_configuration", "token_limit"} + # Identity/schema keys that belong to the embedding service itself. + # Token-cost rates stay per-service too — completion's per-1M rates + # must not be applied to embedding usage, which is priced separately. + embedding_own_keys = { + "embedding_model_service", "model_name", "authentication_configuration", + "token_limit", "input_cost_per_1m", "output_cost_per_1m", + } for k, v in completion.items(): if k not in embedding_own_keys and k not in embedding: embedding[k] = v diff --git a/common/db/health.py b/common/db/health.py new file mode 100644 index 00000000..f1c72882 --- /dev/null +++ b/common/db/health.py @@ -0,0 +1,88 @@ +"""Data-integrity health checks for the Migration Assistant: embedding coverage +and community-summary completeness. + +Deterministic, read-only — NO LLM calls and NO embedding-service init. They only +run count queries + a schema check over an existing connection, so they are safe +on a polled/triggered status check. The connection is the synchronous one the +status endpoint already holds. +""" + +import logging + +from common.utils.summary_placeholders import PLACEHOLDER_MARKERS + +logger = logging.getLogger(__name__) + +_VECTOR_ATTR = "embedding" + + +def _has_vector_attr(conn, v_type: str) -> bool: + """True if *v_type* carries the embedding vector attribute, per the live + schema. Uses the connection's schema API — no embedding-service init.""" + # Native vector attributes are exposed under ``EmbeddingAttributes`` in the + # schema, not by getVertexAttrs (which omits them). + try: + for v in conn.getSchema().get("VertexTypes", []): + if v.get("Name") == v_type: + return any( + e.get("Name") == _VECTOR_ATTR + for e in (v.get("EmbeddingAttributes") or []) + ) + return False + except Exception: + return False + + +def embeddable_types(conn) -> list[str]: + """Vertex types that carry the embedding vector attribute. Schema-detected + (not hardcoded) so it stays correct as the embedded set changes.""" + # Native vector attributes are NOT returned by getVertexAttrs/getVertexTypes; + # they live under each vertex type's ``EmbeddingAttributes`` in the schema. + try: + schema = conn.getSchema() + except Exception as e: + logger.warning(f"embeddable_types: getSchema failed: {e}") + return [] + out = [] + for v in schema.get("VertexTypes", []): + if any(e.get("Name") == _VECTOR_ATTR for e in (v.get("EmbeddingAttributes") or [])): + out.append(v.get("Name")) + return out + + +def embedding_coverage(conn, v_type: str) -> dict | None: + """``{"total": M, "missing": N}`` for *v_type*, or ``None`` when the type is + not embeddable or the ``vertices_have_embedding`` query is unavailable.""" + try: + if not _has_vector_attr(conn, v_type): + return None + res = conn.runInstalledQuery( + "vertices_have_embedding", params={"vertex_type": v_type} + ) + # PRINT order: [0] all_have_embedding, [1] size (missing), [2] total. + missing = int(res[1]["size"]) + total = int(res[2]["total"]) + return {"total": total, "missing": missing} + except Exception as e: + logger.warning(f"embedding_coverage({v_type}) failed: {e}") + return None + + +def community_summary_health(conn, markers=None) -> dict | None: + """``{"total": M, "needs_resummarize": N}`` for Community vertices, or + ``None`` when the ``communities_need_resummarize`` query is unavailable. + ``needs_resummarize`` counts communities whose description is empty or a + known placeholder (old or new).""" + markers = markers if markers is not None else PLACEHOLDER_MARKERS + try: + res = conn.runInstalledQuery( + "communities_need_resummarize", params={"markers": markers} + ) + row = res[0] + return { + "total": int(row["total"]), + "needs_resummarize": int(row["needs_resummarize"]), + } + except Exception as e: + logger.warning(f"community_summary_health failed: {e}") + return None diff --git a/common/db/query_sets.py b/common/db/query_sets.py index 1852bd52..18f1ce57 100644 --- a/common/db/query_sets.py +++ b/common/db/query_sets.py @@ -37,6 +37,7 @@ "common/gsql/graphrag/louvain/stream_community", "common/gsql/graphrag/get_community_children", "common/gsql/graphrag/communities_have_desc", + "common/gsql/graphrag/communities_need_resummarize", "common/gsql/graphrag/graphrag_delete_all_communities", "common/gsql/graphrag/graphrag_stream_entity_community_pairs", "common/gsql/graphrag/graphrag_stream_all_ids", @@ -69,6 +70,13 @@ "common/gsql/supportai/Check_Nonexistent_Vertices", ] +# Data-integrity health-check queries used by the Migration Assistant panel. +# vertices_have_embedding is also installed by the embedding store; listing it +# here lets the Migration Assistant verify/repair it too. +HEALTH_QUERIES = [ + "common/gsql/vector/vertices_have_embedding", +] + # What the Migration Assistant verifies for a GraphRAG graph: everything a # GraphRAG graph actually installs. Excludes the opt-in ECC-checker queries. MIGRATION_QUERIES = ( @@ -76,6 +84,7 @@ + GRAPHRAG_COMMUNITY_QUERIES + SUPPORTAI_INIT_QUERIES + SUPPORTAI_RETRIEVER_QUERIES + + HEALTH_QUERIES ) diff --git a/common/embeddings/tigergraph_embedding_store.py b/common/embeddings/tigergraph_embedding_store.py index bfd1978a..47dbed8e 100644 --- a/common/embeddings/tigergraph_embedding_store.py +++ b/common/embeddings/tigergraph_embedding_store.py @@ -566,13 +566,14 @@ def retrieve_similar_with_score(self, query_embedding, top_k=10, similarity_thre logger.info(f"Fetch {top_k} similar entries from {vertex_types} with filter {filter_expr}") start_time = time() + # GSQL declares STRING expr=""; pyTigerGraph rejects null. verts = self.conn.runInstalledQuery( "get_topk_similar", params={ "vertex_types": vertex_types, "query_vector": query_embedding, "top_k": top_k*2, - "expr": filter_expr, + "expr": filter_expr or "", } ) end_time = time() diff --git a/common/gsql/graphrag/communities_need_resummarize.gsql b/common/gsql/graphrag/communities_need_resummarize.gsql new file mode 100644 index 00000000..9b24cf92 --- /dev/null +++ b/common/gsql/graphrag/communities_need_resummarize.gsql @@ -0,0 +1,24 @@ +CREATE OR REPLACE DISTRIBUTED QUERY communities_need_resummarize(SET markers, BOOL p=False) SYNTAX V2{ + // A community needs re-summarization when its description is empty or is a + // known placeholder marker (old or new). ``markers`` is passed in so the + // set of placeholders stays defined in one place in Python. + SumAccum @@needs; + SumAccum @@total; + Comms = {Community.*}; + Comms = SELECT c FROM Comms:c + POST-ACCUM + @@total += 1, + IF length(c.description) == 0 OR (c.description IN markers) THEN + @@needs += 1 + END; + + PRINT @@needs AS needs_resummarize, @@total AS total; + + // p=true streams the affected community ids + their Louvain layer so the + // regenerate action can re-summarize exactly those. + IF p THEN + bad = SELECT c FROM Comms:c + WHERE length(c.description) == 0 OR (c.description IN markers); + PRINT bad[bad.iteration AS iteration] AS results; + END; +} diff --git a/common/gsql/vector/vertices_have_embedding.gsql b/common/gsql/vector/vertices_have_embedding.gsql index ed6e39f1..5b8f99cd 100644 --- a/common/gsql/vector/vertices_have_embedding.gsql +++ b/common/gsql/vector/vertices_have_embedding.gsql @@ -5,7 +5,8 @@ CREATE OR REPLACE DISTRIBUTED QUERY vertices_have_embedding(String vertex_type, PRINT (results.size() == 0) as all_have_embedding; PRINT results.size() as size; - + PRINT vset.size() as total; + IF verbose THEN PRINT results[results.id as id] as results; END; diff --git a/common/llm_services/base_llm.py b/common/llm_services/base_llm.py index e2f2ad3d..fe77ac1f 100644 --- a/common/llm_services/base_llm.py +++ b/common/llm_services/base_llm.py @@ -15,6 +15,7 @@ import os import re import logging +from typing import Optional from langchain_core.output_parsers import BaseOutputParser, PydanticOutputParser from langchain_core.exceptions import OutputParserException from langchain_core.prompts import BasePromptTemplate @@ -126,7 +127,53 @@ def reset_usage_collection(): _usage_collector.set(None) -def _record_usage(caller_name: str, usage_data: dict): +def _parse_cost_rate(value) -> Optional[float]: + """Return a non-negative float rate, or None if unset/invalid.""" + if value is None or value == "": + return None + try: + rate = float(value) + except (TypeError, ValueError): + return None + if rate < 0: + return None + return rate + + +def estimate_cost_from_config( + config: Optional[dict], + input_tokens: int, + output_tokens: int, +) -> Optional[float]: + """USD cost from user-configured per-1M rates, or None if not configured. + + Both ``input_cost_per_1m`` and ``output_cost_per_1m`` must be set on + ``config`` (USD per 1M tokens). When present they always override + LangChain's built-in cost. + """ + if not config: + return None + inp_rate = _parse_cost_rate(config.get("input_cost_per_1m")) + out_rate = _parse_cost_rate(config.get("output_cost_per_1m")) + if inp_rate is None or out_rate is None: + return None + return ( + max(0, int(input_tokens or 0)) * inp_rate / 1_000_000.0 + + max(0, int(output_tokens or 0)) * out_rate / 1_000_000.0 + ) + + +def _record_usage(caller_name: str, usage_data: dict, config: Optional[dict] = None): + # User-configured rates replace LangChain cost entirely; otherwise keep + # whatever LangChain reported (may be 0 for unknown models). + user_cost = estimate_cost_from_config( + config, + usage_data.get("input_tokens", 0), + usage_data.get("output_tokens", 0), + ) + if user_cost is not None: + usage_data["cost"] = user_cost + logger.info(f"{caller_name} usage: {usage_data}") bucket = _usage_collector.get() if bucket is not None: bucket.append({"caller_name": caller_name, **usage_data}) @@ -500,8 +547,7 @@ def invoke_with_parser( usage_data["output_tokens"] = cb.completion_tokens usage_data["total_tokens"] = cb.total_tokens usage_data["cost"] = cb.total_cost - logger.info(f"{caller_name} usage: {usage_data}") - _record_usage(caller_name, usage_data) + _record_usage(caller_name, usage_data, self.config) raw_text = self._message_text(raw_output) @@ -545,8 +591,7 @@ def invoke_with_tools( usage_data["output_tokens"] = cb.completion_tokens usage_data["total_tokens"] = cb.total_tokens usage_data["cost"] = cb.total_cost - logger.info(f"{caller_name} usage: {usage_data}") - _record_usage(caller_name, usage_data) + _record_usage(caller_name, usage_data, self.config) return resp def invoke_structured( @@ -579,8 +624,7 @@ def invoke_structured( usage_data["output_tokens"] = cb.completion_tokens usage_data["total_tokens"] = cb.total_tokens usage_data["cost"] = cb.total_cost - logger.info(f"{caller_name} usage: {usage_data}") - _record_usage(caller_name, usage_data) + _record_usage(caller_name, usage_data, self.config) return result async def ainvoke_with_parser( @@ -608,8 +652,7 @@ async def ainvoke_with_parser( usage_data["output_tokens"] = cb.completion_tokens usage_data["total_tokens"] = cb.total_tokens usage_data["cost"] = cb.total_cost - logger.info(f"{caller_name} usage: {usage_data}") - _record_usage(caller_name, usage_data) + _record_usage(caller_name, usage_data, self.config) raw_text = self._message_text(raw_output) diff --git a/common/llm_services/capabilities.py b/common/llm_services/capabilities.py index 653c3fd9..0d5f3266 100644 --- a/common/llm_services/capabilities.py +++ b/common/llm_services/capabilities.py @@ -26,6 +26,7 @@ """ import logging +import threading logger = logging.getLogger(__name__) @@ -109,7 +110,17 @@ def openai_rejects_temperature(model: str) -> bool: def _gemini_tool_calling(model: str) -> bool: # Gemini 1.5+ and 2.x support function calling. - return "gemini-1.5" in model or "gemini-2" in model or "gemini-exp" in model + # Every Gemini family from 1.5 onward supports function/tool calling, and + # future families (4.x, 5.x, ...) will too. Use a denylist instead of an + # allowlist so new models work without a code change: any Gemini is capable + # except the legacy 1.0-era models that predate function calling. + if "gemini" not in model: + return False + if "gemini-1.0" in model or "gemini-pro-vision" in model: + return False + if model.strip() == "gemini-pro": # bare 1.0 alias (versioned ids are fine) + return False + return True def _gemini_thinking(model: str) -> bool: @@ -162,3 +173,164 @@ def model_supports_agentic(config: dict) -> bool: (config or {}).get("llm_model"), ) return caps["supports_tool_calling"] + + +# --------------------------------------------------------------------------- +# Runtime tool-calling probe (GML-2169) +# +# Policy: Agentic mode is ON by default. We only DISABLE it when we are sure +# the model can't tool-call — a known-legacy model, an explicit "tools not +# supported" error from the probe, or a real tool-calling failure at query +# time (mark_tool_calling_unsupported). Anything uncertain (transient error, +# ambiguous failure, timeout, no provider yet) stays ENABLED and is not cached, +# so a flaky probe never disables a capable model. Results are cached in memory +# (per process), keyed by service:model — never persisted to config, re-probed +# after a restart, and sticky until the model config changes (new key). +# --------------------------------------------------------------------------- + +_probe_cache: dict = {} +_probe_lock = threading.Lock() + +_PROBE_TIMEOUT_S = 20 + +# Error text that positively indicates the model rejects tool/function calling. +_NO_TOOL_SUPPORT_MARKERS = ( + "does not support tool", + "not support tools", + "tools are not supported", + "tool use is not supported", + "tool calling is not supported", + "function calling is not supported", + "does not support function", + "tool_choice is not supported", + "tools is not supported", +) + + +def _probe_key(config: dict) -> str: + service = (config.get("llm_service") or "").strip().lower() + model = (config.get("llm_model") or "").strip().lower() + return f"{service}:{model}" + + +def _known_no_tool_calling(config: dict) -> bool: + """True only for models we are *sure* predate tool/function calling.""" + service = (config.get("llm_service") or "").strip().lower() + model = _strip_region((config.get("llm_model") or "").strip().lower()) + if "gemini-1.0" in model or "gemini-pro-vision" in model or model == "gemini-pro": + return True + if service in ("openai", "azure", "azure_openai", "azureopenai"): + if model.startswith(("text-davinci", "davinci", "curie", "babbage", "ada", + "text-ada", "text-babbage", "text-curie")): + return True + if service in ("bedrock", "aws_bedrock", "awsbedrock"): + if ("amazon.titan" in model + or "meta.llama2" in model + or "ai21.j2" in model + or "ai21.jurassic" in model + or "anthropic.claude-instant" in model + or "anthropic.claude-v2" in model + or model.startswith("anthropic.claude-2") + or ("cohere.command" in model and "command-r" not in model)): + return True + return False + + +def _looks_like_no_tool_support(exc) -> bool: + msg = str(exc).lower() + return any(m in msg for m in _NO_TOOL_SUPPORT_MARKERS) + + +def _invoke_probe(llm): + from pydantic import BaseModel, Field + + class _ProbePing(BaseModel): + """Acknowledge readiness by calling this tool.""" + ok: bool = Field(default=True, description="always true") + + bound = llm.bind_tools([_ProbePing]) + bound.invoke([ + ("system", "You can call tools."), + ("user", "Call the ProbePing tool with ok set to true."), + ]) + + +def _run_tool_calling_probe(llm_provider): + """Bind a trivial tool and make a minimal, time-bounded call. + + Returns ``True`` (confirmed tool-calling), ``False`` (the model explicitly + rejects tools — a *confident* no-support signal), or ``None`` (unknown: + transient, ambiguous, timeout, or no usable provider). ``None`` must not + disable Agentic mode. + """ + import concurrent.futures + + llm = getattr(llm_provider, "llm", None) + if llm is None or not hasattr(llm, "bind_tools"): + return None + try: + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex: + ex.submit(_invoke_probe, llm).result(timeout=_PROBE_TIMEOUT_S) + return True + except concurrent.futures.TimeoutError: + logger.warning("tool-calling probe timed out; leaving Agentic enabled") + return None + except Exception as exc: # noqa: BLE001 + if _looks_like_no_tool_support(exc): + logger.info("tool-calling probe: model rejects tools (%s)", str(exc)[:200]) + return False + logger.warning( + "tool-calling probe inconclusive (%s); leaving Agentic enabled", + str(exc)[:200], + ) + return None + + +def supports_tool_calling(config: dict, llm_provider=None) -> bool: + """Whether the resolved chat model can drive Agentic mode. + + Optimistic: enabled unless we are sure it can't tool-call. Uses a cached + in-memory runtime probe; a cached result wins, a known-legacy model is + disabled, a confident probe/runtime failure disables, and everything else + stays enabled. Never writes to config. + """ + if not isinstance(config, dict): + return False + key = _probe_key(config) + with _probe_lock: + if key in _probe_cache: + return _probe_cache[key] + + if _known_no_tool_calling(config): + with _probe_lock: + _probe_cache[key] = False + return False + + if llm_provider is None: + return True # optimistic; can't probe yet, don't cache + + result = _run_tool_calling_probe(llm_provider) + if result is True: + with _probe_lock: + _probe_cache[key] = True + return True + if result is False: # confident no-support + with _probe_lock: + _probe_cache[key] = False + return False + return True # unknown -> stay enabled, don't cache + + +def mark_tool_calling_unsupported(config: dict) -> None: + """Record that the chat model failed tool-calling at runtime, so later + requests downgrade to the classic engine until the model config changes or + the container restarts.""" + if isinstance(config, dict): + with _probe_lock: + _probe_cache[_probe_key(config)] = False + + +def reset_tool_calling_cache() -> None: + """Clear the in-memory probe cache (test hook / manual reset).""" + with _probe_lock: + _probe_cache.clear() diff --git a/common/utils/image_data_extractor.py b/common/utils/image_data_extractor.py index 6be929cc..51721f79 100644 --- a/common/utils/image_data_extractor.py +++ b/common/utils/image_data_extractor.py @@ -119,7 +119,14 @@ def describe_image_with_llm(file_path): "stacked bar with a time-period axis), TRANSCRIBE every " "(period, value) pair you can read in the format " "`period: value; period: value; …` — do not summarize " - "the trend in place of the values; (3) the entities, " + "the trend in place of the values. For categorical " + "bar/rank charts (categories such as regions, " + "companies, or age groups on one axis and a numeric " + "unit on the other), TRANSCRIBE every visible " + "category with its value as " + "`category: value; category: value; …` including the " + "unit — never list categories without their numbers; " + "(3) the entities, " "relationships, or process steps in any diagram or " "flowchart; (4) any logo or branding mark, identified by " "name. Do NOT describe layout, background color, " diff --git a/common/utils/summary_placeholders.py b/common/utils/summary_placeholders.py new file mode 100644 index 00000000..74cd0036 --- /dev/null +++ b/common/utils/summary_placeholders.py @@ -0,0 +1,25 @@ +"""Community-summary placeholder markers, shared by ECC (writes them) and the +graphrag app / Migration Assistant (detects them). Single source of truth so the +health check and the rebuild pipeline agree on what "needs re-summarization" +means. +""" + +# Written as a community's description when summarization can't produce a real +# one. Non-empty so the layer-completion check passes and the rebuild finishes, +# and stable so it can be found and regenerated later. +COMMUNITY_SUMMARY_PLACEHOLDER = "[summary unavailable - regenerate]" + +# Placeholder written by pre-2.0.1 builds; kept so re-summarization and progress +# checks recognize communities left behind by older graphs too. +LEGACY_SUMMARY_PLACEHOLDER = "Should ignore due to summary error." + +# Non-empty markers a description may carry. An empty description is also treated +# as needing re-summarization, but "" is handled separately (GSQL length check). +PLACEHOLDER_MARKERS = [COMMUNITY_SUMMARY_PLACEHOLDER, LEGACY_SUMMARY_PLACEHOLDER] + + +def is_placeholder_summary(text: str) -> bool: + """True if a community description is a placeholder needing regeneration: + the current or legacy sentinel, or empty.""" + t = (text or "").strip() + return t in ("", COMMUNITY_SUMMARY_PLACEHOLDER, LEGACY_SUMMARY_PLACEHOLDER) diff --git a/common/utils/text_extractors.py b/common/utils/text_extractors.py index 09399a92..41a441e8 100644 --- a/common/utils/text_extractors.py +++ b/common/utils/text_extractors.py @@ -41,8 +41,11 @@ # which bloat tokens 3-5x and confuse retrieval embeddings. The CJK # Unicode ranges below cover CJK Unified Ideographs (U+4E00-U+9FFF), # Hiragana / Katakana / CJK Symbols (U+3000-U+30FF), and full-width -# / half-width forms (U+FF00-U+FFEF). -_CJK_CHAR_CLASS = r"[　-鿿＀-￯]" +# / half-width forms (U+FF00-U+FFEF) excluding fullwidth digits +# (U+FF10-U+FF19). Collapsing digit runs would glue distinct chart +# values such as 767 and 808 into ``767808``. +# One CJK char excluding fullwidth digits (U+FF10-U+FF19). +_CJK_CHAR_CLASS = r"(?:[　-鿿]|[＀-／]|[：-￯])" _VERTICAL_BOLD_CJK = re.compile( rf"(?:\*\*{_CJK_CHAR_CLASS}\*\*(?:)){{2,}}\*\*{_CJK_CHAR_CLASS}\*\*" ) @@ -60,6 +63,20 @@ _TABLE_LINE_RE = re.compile(r"^\s*\|") _BR_TAG_RE = re.compile(r"", re.IGNORECASE) +# pymupdf4llm picture-text blocks (HTML-comment or markdown-dash markers). +# These are figure-associated text from pymupdf4llm, not a separate OCR engine. +_PICTURE_TEXT_BLOCK_RE = re.compile( + r"(?:|\*{0,3}\s*-+\s*Start of picture text\s*-+\s*\*{0,3})" + r"(.*?)" + r"(?:|\*{0,3}\s*-+\s*End of picture text\s*-+\s*\*{0,3})", + re.IGNORECASE | re.DOTALL, +) + +# Adjacent comma-grouped numbers glued with no separator, e.g. ``1,5461,518``. +_GLUED_COMMA_NUMBERS_RE = re.compile( + r"(? list[dict]: def _strip_br_in_table_rows(text: str) -> str: - """Remove ``
`` tags inside markdown table rows. + """Replace ``
`` tags inside markdown table rows with spaces. - Rationale documented at _TABLE_LINE_RE. + Using a space (not empty string) keeps stacked chart values distinct + — ``|767
808|`` becomes ``|767 808|``, never ``|767808|``. """ out: list[str] = [] for line in text.split("\n"): if _TABLE_LINE_RE.match(line): line = _BR_TAG_RE.sub(" ", line) + # Collapse runs of whitespace left by consecutive
tags. + line = re.sub(r"[ \t]{2,}", " ", line) out.append(line) return "\n".join(out) +def _split_glued_comma_numbers(text: str) -> str: + """Insert a space between adjacent comma-grouped numbers. + + pymupdf4llm / chart extraction sometimes emits ``1,5461,518`` instead of + ``1,546 1,518``. Repeat until stable for longer glued runs. + """ + prev = None + while prev != text: + prev = text + text = _GLUED_COMMA_NUMBERS_RE.sub(r"\1 \2", text) + return text + + +def _normalize_picture_text_blocks(text: str) -> str: + """Normalize pymupdf4llm picture-text blocks for chunking + retrieval. + + - Rewrite HTML-comment markers to the markdown form StructuredChunker + already recognizes. + - Turn in-block ``
`` into newlines (not empty joins) so values like + ``767`` and ``808`` stay separable. + - Split glued comma-numbers inside the block. + """ + + def _rewrite(match: re.Match) -> str: + body = match.group(1) or "" + body = _BR_TAG_RE.sub("\n", body) + body = _split_glued_comma_numbers(body) + # Trim excess blank lines inside the block. + body = re.sub(r"\n{3,}", "\n\n", body).strip("\n") + return ( + "***----- Start of picture text -----***\n" + f"{body}\n" + "***----- End of picture text -----***" + ) + + return _PICTURE_TEXT_BLOCK_RE.sub(_rewrite, text) + + +_PAGE_MARKER_RE = re.compile(r"") + + +def _recover_mojibake_pages( + file_path, + markdown: str, + graphname=None, + max_pages: int = 3, +) -> str: + """Recover table/chart text from PDF pages with broken ToUnicode CMaps. + + When glyph mapping fails, embedded text often keeps numbers but corrupts + labels. Drawn (non-embedded) figures also skip the normal image-describe + pass. For page sections that look both corrupted and table-like, render + the page and multimodal-transcribe it, then append the result next to the + original page body. Capped by ``max_pages`` to bound cost. + """ + if not markdown or max_pages <= 0: + return markdown + + # Collect page numbers whose section text looks corrupted. + parts = _PAGE_MARKER_RE.split(markdown) + # parts: [pre, pageNo, body, pageNo, body, ...] + bad_pages: list[int] = [] + if len(parts) >= 3: + for i in range(1, len(parts), 2): + try: + page_no = int(parts[i]) + except (TypeError, ValueError): + continue + body = parts[i + 1] if i + 1 < len(parts) else "" + findings = _detect_mojibake(body, source_hint=f"{file_path}:p{page_no}") + # Prefer pages that look like broken tables (pipe rows + mojibake). + pipe_rows = sum(1 for ln in body.splitlines() if ln.strip().startswith("|")) + if len(findings) >= 3 and pipe_rows >= 3: + bad_pages.append(page_no) + if not bad_pages: + return markdown + + try: + import pymupdf + from common.utils.image_data_extractor import ( + describe_image_with_llm, + should_extract_images, + ) + except Exception as e: # noqa: BLE001 + logger.warning("mojibake page recovery unavailable: %s", e) + return markdown + + if not should_extract_images(graphname): + return markdown + + recovered: dict[int, str] = {} + try: + doc = pymupdf.open(str(file_path)) + except Exception as e: # noqa: BLE001 + logger.warning("mojibake recovery: cannot open %s: %s", file_path, e) + return markdown + + try: + for page_no in bad_pages[:max_pages]: + idx = page_no - 1 + if idx < 0 or idx >= doc.page_count: + continue + try: + page = doc[idx] + # ~150 dpi — enough to read table cells without huge payloads. + pix = page.get_pixmap(matrix=pymupdf.Matrix(2.0, 2.0), alpha=False) + tmp = Path(tempfile.mkdtemp(prefix="mojibake_page_")) / f"p{page_no}.png" + pix.save(str(tmp)) + desc = describe_image_with_llm(str(tmp)) + try: + shutil.rmtree(tmp.parent, ignore_errors=True) + except Exception: + pass + if not desc or "decorative image" in desc.lower(): + continue + # Skip if the transcription itself looks glyph-broken. + if len(_detect_mojibake(desc)) >= 3: + continue + recovered[page_no] = desc.strip() + logger.info( + "mojibake page recovery: %s page %s recovered %s chars", + file_path, + page_no, + len(desc), + ) + except Exception as e: # noqa: BLE001 + logger.warning( + "mojibake page recovery failed for %s p%s: %s", + file_path, + page_no, + e, + ) + finally: + doc.close() + + if not recovered: + return markdown + + # Append recovered transcription under each page marker body. + out_parts: list[str] = [parts[0]] + for i in range(1, len(parts), 2): + page_no_s = parts[i] + body = parts[i + 1] if i + 1 < len(parts) else "" + out_parts.append(f"") + out_parts.append(body) + try: + page_no = int(page_no_s) + except (TypeError, ValueError): + continue + if page_no in recovered: + out_parts.append( + "\n\n\n" + + recovered[page_no] + + "\n" + ) + return "\n".join(out_parts) + + def _collapse_vertical_cjk(text: str) -> str: """Collapse pymupdf4llm's per-character vertical-CJK runs back into a single token. Bold runs ``**X**
**Y**
**Z**`` become ``**XYZ**``; @@ -158,11 +336,21 @@ def _clean_pdf_markdown(markdown: str, source_hint: str = "") -> str: with ``
`` separators and per-character bold markers. The run is collapsed back into a single token so embedding and retrieval see the intended word (e.g. ``**個別信用購入あっせん**``) rather than ten - fragments. + fragments. Fullwidth digits are excluded so chart values are not glued. + + 4. **Picture-text blocks** — figure text wrapped in + ```` (or the dash-marker form) by + pymupdf4llm is rewritten so StructuredChunker keeps the block atomic, + ``
`` becomes newlines, and glued comma-numbers like ``1,5461,518`` + are split. """ # --- Pass 1: remove ColN placeholders --- markdown = _coln_pattern.sub('', markdown) + # --- Pass 1b: normalize pymupdf4llm picture-text blocks before CJK/table + # passes so chart
stacks become newlines rather than empty joins. + markdown = _normalize_picture_text_blocks(markdown) + # --- Pass 2: collapse vertical-CJK runs (do this BEFORE row dedup so # rows that differ only by the collapsed form aren't treated as # distinct rows). @@ -171,7 +359,10 @@ def _clean_pdf_markdown(markdown: str, source_hint: str = "") -> str: # --- Pass 2b: strip
inside markdown table rows --- markdown = _strip_br_in_table_rows(markdown) - # --- Pass 2c: log lines that look like mojibake (failed glyph decode). + # --- Pass 2c: split glued comma-grouped numbers globally --- + markdown = _split_glued_comma_numbers(markdown) + + # --- Pass 2d: log lines that look like mojibake (failed glyph decode). # We don't repair these — the underlying glyphs aren't recoverable # from the markdown — but logging gives operators a grep target. findings = _detect_mojibake(markdown, source_hint) @@ -697,6 +888,12 @@ def _to_markdown_paged(strategy: str | None = None): # Clean up artefacts common in form PDFs (duplicate rows, ColN headers) markdown_content = _clean_pdf_markdown(markdown_content, source_hint=str(file_path)) + # Pages with broken CMaps (mojibake row labels, intact numbers) need a + # page-screenshot multimodal pass — embedded images are often absent. + markdown_content = _recover_mojibake_pages( + file_path, markdown_content, graphname=graphname + ) + # Rename image files that contain spaces to avoid path-parsing issues markdown_content = _sanitize_image_filenames(image_output_folder, markdown_content) diff --git a/ecc/app/graphrag/regenerate.py b/ecc/app/graphrag/regenerate.py new file mode 100644 index 00000000..45f760e5 --- /dev/null +++ b/ecc/app/graphrag/regenerate.py @@ -0,0 +1,205 @@ +"""Targeted regeneration actions for the Migration Assistant data-integrity +panel: + +- ``regenerate_embeddings`` — re-embed vertices missing an embedding (GML-2175). +- ``regenerate_summaries`` — re-summarize communities whose description is a + placeholder or empty, then re-embed the new summary (GML-2176). + +Both reuse the rebuild pipeline's building blocks (``get_commuinty_children``, +``CommunitySummarizer``, the embedding store) — no new embedding/summarization +logic. Vertices whose source text is empty/placeholder are skipped and counted; +they need a rebuild/re-ingest, not a re-embed. +""" + +import logging + +from common.config import ( + get_llm_service, + get_completion_config, + get_embedding_service, +) +from common.embeddings.tigergraph_embedding_store import TigerGraphEmbeddingStore +from common.utils.summary_placeholders import PLACEHOLDER_MARKERS, is_placeholder_summary +from graphrag import util, community_summarizer + +logger = logging.getLogger(__name__) + +_VECTOR_ATTR = "embedding" + + +def _make_store(conn, graphname): + store = TigerGraphEmbeddingStore( + conn, get_embedding_service(), support_ai_instance=True + ) + store.set_graphname(graphname) + return store + + +async def _embeddable_types(conn) -> list[str]: + """Vertex types carrying the embedding attribute (async connection).""" + # Native vector attributes are absent from getVertexAttrs; they live under + # each vertex type's ``EmbeddingAttributes`` in the schema. + try: + schema = await conn.getSchema() + except Exception as e: + logger.warning(f"regen: getSchema failed: {e}") + return [] + out = [] + for v in schema.get("VertexTypes", []): + if any(e.get("Name") == _VECTOR_ATTR for e in (v.get("EmbeddingAttributes") or [])): + out.append(v.get("Name")) + return out + + +def _row_id(r): + """Vertex id from a printed vertex row, tolerant of the two shapes + pyTigerGraph returns (``v_id`` vs an aliased ``id`` attribute).""" + return r.get("v_id") or r.get("attributes", {}).get("id") or r.get("id") + + +async def _chunk_text(conn, chunk_id): + try: + res = await conn.runInstalledQuery( + "StreamChunkContent", params={"chunk": (chunk_id,)} + ) + rows = res[0].get("ChunkContent") if res else None + if rows: + return rows[0].get("attributes", {}).get("text", "") or "" + except Exception as e: + logger.warning(f"regen: chunk text fetch failed for {chunk_id}: {e}") + return "" + + +async def _description(conn, vtype, vid): + try: + v = await conn.getVerticesById(vtype, vid) + if v: + desc = v[0].get("attributes", {}).get("description", "") + # Entity descriptions can be a list; join for embedding. + if isinstance(desc, list): + desc = " ".join(str(x) for x in desc if x) + return desc or "" + except Exception as e: + logger.warning(f"regen: description fetch failed for {vtype} {vid}: {e}") + return "" + + +async def regenerate_embeddings(graphname, conn): + """Re-embed vertices missing an embedding, per embeddable type. Returns + ``{"regenerated": n, "skipped": m}``; skipped = empty/placeholder source + (needs a rebuild/re-summarize) or an embed error.""" + util.loading_event.set() + store = _make_store(conn, graphname) + regenerated = 0 + skipped = 0 + for vt in await _embeddable_types(conn): + try: + res = await conn.runInstalledQuery( + "vertices_have_embedding", + params={"vertex_type": vt, "verbose": True}, + ) + results = next( + (r["results"] for r in res if isinstance(r, dict) and "results" in r), + [], + ) + ids = [i for i in (_row_id(r) for r in results) if i] + except Exception as e: + logger.warning(f"regen_embeddings: list missing for {vt} failed: {e}") + continue + for vid in ids: + text = ( + await _chunk_text(conn, vid) + if vt == "DocumentChunk" + else await _description(conn, vt, vid) + ) + if not text or is_placeholder_summary(text): + skipped += 1 + continue + try: + await store.aadd_embeddings([(text, [])], [{"vertex_id": (vid, vt)}]) + regenerated += 1 + except Exception as e: + logger.warning(f"regen_embeddings: re-embed failed {vt} {vid}: {e}") + skipped += 1 + logger.info( + f"regenerate_embeddings({graphname}): " + f"regenerated={regenerated} skipped={skipped}" + ) + return {"regenerated": regenerated, "skipped": skipped} + + +async def regenerate_summaries(graphname, conn): + """Re-summarize communities with placeholder/empty descriptions, then + re-embed. Returns ``{"resummarized": n, "skipped": m}``; skipped = no usable + child text or a summarization failure (needs a rebuild/re-ingest).""" + util.loading_event.set() + store = _make_store(conn, graphname) + llm = get_llm_service(get_completion_config(graphname)) + summarizer = community_summarizer.CommunitySummarizer(llm) + resummarized = 0 + skipped = 0 + try: + res = await conn.runInstalledQuery( + "communities_need_resummarize", + params={"markers": PLACEHOLDER_MARKERS, "p": True}, + ) + targets = next( + (r["results"] for r in res if isinstance(r, dict) and "results" in r), [] + ) + except Exception as e: + logger.error(f"regen_summaries: target list failed: {e}") + return {"resummarized": 0, "skipped": 0, "error": str(e)} + + for t in targets: + cid = _row_id(t) + try: + i = int(t.get("attributes", {}).get("iteration", 0)) + except (TypeError, ValueError): + i = 0 + if not cid: + skipped += 1 + continue + try: + children = await util.get_commuinty_children(conn, i, cid) + except Exception as e: + logger.warning(f"regen_summaries: children fetch failed {cid}: {e}") + skipped += 1 + continue + if not children: + skipped += 1 + continue + if len(children) == 1: + summary = children[0] + else: + r = await summarizer.summarize(cid, children) + if r.get("error"): + logger.warning( + f"regen_summaries: summarize failed {cid}: {r.get('message')}" + ) + skipped += 1 + continue + summary = r["summary"] + if not summary or is_placeholder_summary(summary): + skipped += 1 + continue + pid = util.process_id(cid) + try: + # Direct upsert (not util.upsert_vertex, which only enqueues to the + # rebuild loader queue that isn't running in a standalone regenerate + # action — the description write would be silently dropped). + await conn.upsertVertex( + "Community", pid, + attributes={"description": summary, "iteration": i}, + ) + await store.aadd_embeddings( + [(summary, [])], [{"vertex_id": (pid, "Community")}] + ) + resummarized += 1 + except Exception as e: + logger.warning(f"regen_summaries: upsert/embed failed {cid}: {e}") + skipped += 1 + logger.info( + f"regenerate_summaries({graphname}): " + f"resummarized={resummarized} skipped={skipped}" + ) + return {"resummarized": resummarized, "skipped": skipped} diff --git a/ecc/app/graphrag/util.py b/ecc/app/graphrag/util.py index 60129e8e..37313b75 100644 --- a/ecc/app/graphrag/util.py +++ b/ecc/app/graphrag/util.py @@ -65,19 +65,15 @@ loading_event = asyncio.Event() loading_event.set() # set the event to true to allow the workers to run -# Written as a community's description when summarization can't produce a real -# one. Non-empty so the layer-completion check passes and the rebuild finishes, -# and stable so it can be found and regenerated later. -COMMUNITY_SUMMARY_PLACEHOLDER = "[summary unavailable - regenerate]" -# Placeholder written by pre-2.0.1 builds; kept so re-summarization and progress -# checks recognize communities left behind by older graphs too. -LEGACY_SUMMARY_PLACEHOLDER = "Should ignore due to summary error." - -def is_placeholder_summary(text: str) -> bool: - """True if a community description is a placeholder needing regeneration: - the current or legacy sentinel, or empty.""" - t = (text or "").strip() - return t in ("", COMMUNITY_SUMMARY_PLACEHOLDER, LEGACY_SUMMARY_PLACEHOLDER) +# Community-summary placeholder markers now live in a shared module so the +# graphrag app / Migration Assistant detect exactly what ECC writes. Re-exported +# here for the existing ECC callers. +from common.utils.summary_placeholders import ( # noqa: E402,F401 + COMMUNITY_SUMMARY_PLACEHOLDER, + LEGACY_SUMMARY_PLACEHOLDER, + PLACEHOLDER_MARKERS, + is_placeholder_summary, +) async def install_queries( requried_queries: list[str], @@ -220,9 +216,12 @@ def map_attrs(attributes: dict): def process_id(v_id: str): - has_func = re.compile(r"(.*)\(").findall(v_id) - if len(has_func) > 0: - v_id = has_func[0] + # Strip parentheses in place — do NOT truncate at "(". The old `(.*)\(` + # truncation dropped everything from the first "(" onward, which mangled + # ids for document names containing parentheses (e.g. "Q3 (2026) report"): + # the "_chunk_{i}" suffix was lost, so chunk ids no longer ended in an + # integer and the rebuild crashed on int(chunk_id.split("_")[-1]). The + # replace() below removes the parens without dropping the rest. (GML-2173) v_id = v_id.replace(" ", "_").lower().replace("/", "_").replace("(", "").replace(")", "") if v_id == "''" or v_id == '""': return "" diff --git a/ecc/app/graphrag/workers.py b/ecc/app/graphrag/workers.py index 1dc7d1b9..518a2d82 100644 --- a/ecc/app/graphrag/workers.py +++ b/ecc/app/graphrag/workers.py @@ -173,7 +173,7 @@ async def chunk_doc( # send chunks to be upserted (func, args) logger.debug("chunk writes to upsert_chan") - await upsert_chan.put((upsert_chunk, (conn, v_id, chunk_id, chunk))) + await upsert_chan.put((upsert_chunk, (conn, v_id, chunk_id, chunk, i))) # send chunks to have entities extracted logger.debug("chunk writes to extract_chan") @@ -208,7 +208,7 @@ async def upsert_doc(conn: AsyncTigerGraphConnection, doc_id, ctype, content_tex conn, "Document", doc_id, "HAS_CONTENT", "Content", doc_id ) -async def upsert_chunk(conn: AsyncTigerGraphConnection, doc_id, chunk_id, chunk): +async def upsert_chunk(conn: AsyncTigerGraphConnection, doc_id, chunk_id, chunk, idx): logger.debug(f"Upserting chunk {chunk_id}") date_added = int(time.time()) # Build the chunk's full vertex + edge bundle and enqueue atomically. @@ -220,7 +220,7 @@ async def upsert_chunk(conn: AsyncTigerGraphConnection, doc_id, chunk_id, chunk) ("DocumentChunk", chunk_id, { "epoch_added": date_added, "epoch_processed": date_added, - "idx": int(chunk_id.split("_")[-1]), + "idx": idx, }), ("Content", chunk_id, {"text": chunk, "epoch_added": date_added}), ] @@ -228,7 +228,6 @@ async def upsert_chunk(conn: AsyncTigerGraphConnection, doc_id, chunk_id, chunk) ("DocumentChunk", chunk_id, "HAS_CONTENT", "Content", chunk_id, None), ("Document", doc_id, "HAS_CHILD", "DocumentChunk", chunk_id, None), ] - idx = int(chunk_id.split("_")[-1]) if idx > 0: edges.append(( "DocumentChunk", chunk_id, "IS_AFTER", diff --git a/ecc/app/main.py b/ecc/app/main.py index 5bcb8364..9624cb36 100644 --- a/ecc/app/main.py +++ b/ecc/app/main.py @@ -435,3 +435,63 @@ def consistency_update( return f"Method unsupported, must be {SupportAIMethod.SUPPORTAI}, {SupportAIMethod.GRAPHRAG}" return {"status": "submitted", "message": ecc_status} + + +def _regen_build_conn(graphname, credentials): + """Async DB connection for a regenerate action (mirrors consistency_update + auth handling).""" + reload_db_config() + if isinstance(credentials, HTTPBasicCredentials): + conn = elevate_db_connection_to_token( + db_config.get("hostname"), credentials.username, credentials.password, + graphname, async_conn=True, + ) + elif isinstance(credentials, HTTPAuthorizationCredentials): + conn = get_db_connection_id_token( + graphname, credentials.credentials, async_conn=True + ) + else: + raise HTTPException(status_code=401, detail="Invalid authentication credentials") + asyncio.run(conn.customizeHeader( + timeout=db_config.get("default_timeout", 300) * 1000, responseSize=5000000 + )) + return conn + + +def _run_regen(graphname, credentials, task_suffix, run_func): + """Run a targeted regenerate action synchronously, returning its counts. + Refuses while a rebuild or the same action is already in flight (they both + write embeddings).""" + rebuild_key = f"{graphname}:graphrag" + if rebuild_key in running_tasks and running_tasks[rebuild_key].get("status") == "running": + raise HTTPException( + status_code=409, + detail=f"A rebuild is in progress for {graphname}; retry after it completes.", + ) + task_key = f"{graphname}:{task_suffix}" + if task_key in running_tasks and running_tasks[task_key].get("status") == "running": + raise HTTPException(status_code=409, detail=f"{task_suffix} already running for {graphname}") + conn = _regen_build_conn(graphname, credentials) + running_tasks[task_key] = {"status": "running", "started_at": time.time()} + try: + result = asyncio.run(run_func(graphname, conn)) + LogWriter.info(f"Completed ECC task: {task_key} -> {result}") + return {"status": "completed", **result} + finally: + running_tasks.pop(task_key, None) + + +@app.get("/{graphname}/graphrag/regenerate_embeddings") +def regenerate_embeddings_endpoint(graphname: str, credentials=Depends(auth_credentials)): + """Re-embed vertices missing an embedding (GML-2175). Targeted, not a full + rebuild. Runs synchronously and returns {regenerated, skipped}.""" + from graphrag.regenerate import regenerate_embeddings + return _run_regen(graphname, credentials, "regenerate_embeddings", regenerate_embeddings) + + +@app.get("/{graphname}/graphrag/regenerate_summaries") +def regenerate_summaries_endpoint(graphname: str, credentials=Depends(auth_credentials)): + """Re-summarize communities with placeholder/empty descriptions and re-embed + (GML-2176). Targeted, not a full rebuild. Returns {resummarized, skipped}.""" + from graphrag.regenerate import regenerate_summaries + return _run_regen(graphname, credentials, "regenerate_summaries", regenerate_summaries) diff --git a/ecc/app/supportai/util.py b/ecc/app/supportai/util.py index 0d62c669..630bab93 100644 --- a/ecc/app/supportai/util.py +++ b/ecc/app/supportai/util.py @@ -2,7 +2,6 @@ import base64 import json import logging -import re import traceback from glob import glob from typing import Callable @@ -182,9 +181,10 @@ def map_attrs(attributes: dict): def process_id(v_id: str): - has_func = re.compile(r"(.*)\(").findall(v_id) - if len(has_func) > 0: - v_id = has_func[0] + # Strip parentheses in place — do NOT truncate at "(". Truncating dropped + # the "_chunk_{i}" suffix for ids containing parentheses, corrupting chunk + # ids. The replace() below removes the parens without dropping the rest. + # (GML-2173) v_id = v_id.replace(" ", "_").lower().replace("/", "_").replace("(", "").replace(")", "") if v_id == "''" or v_id == '""': return "" diff --git a/graphrag-ui/package-lock.json b/graphrag-ui/package-lock.json index 7def4e45..267fbb24 100644 --- a/graphrag-ui/package-lock.json +++ b/graphrag-ui/package-lock.json @@ -1,12 +1,12 @@ { "name": "tg-cbot-v5", - "version": "0.0.5", + "version": "2.0.2", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "tg-cbot-v5", - "version": "0.0.5", + "version": "2.0.2", "dependencies": { "@hookform/resolvers": "^3.6.0", "@radix-ui/react-dialog": "^1.1.1", diff --git a/graphrag-ui/package.json b/graphrag-ui/package.json index c20d3dd8..936dba70 100755 --- a/graphrag-ui/package.json +++ b/graphrag-ui/package.json @@ -1,7 +1,7 @@ { "name": "tg-cbot-v5", "private": true, - "version": "0.0.5", + "version": "2.0.2", "type": "module", "packageManager": "pnpm@9.15.0", "scripts": { diff --git a/graphrag-ui/src/actions/MessageParser.tsx b/graphrag-ui/src/actions/MessageParser.tsx index 9645b11d..a09f212e 100644 --- a/graphrag-ui/src/actions/MessageParser.tsx +++ b/graphrag-ui/src/actions/MessageParser.tsx @@ -7,6 +7,10 @@ interface MessageParserProps { const MessageParser: React.FC = ({ children, actions }) => { const parse = (message: string) => { + // Ignore empty / whitespace-only submits. Clicking the Stop button just as + // an answer finishes can leak a send after the input was cleared, which + // otherwise fires an empty query at the backend. + if (!message || !message.trim()) return; actions.queryGraphragWs(message); }; diff --git a/graphrag-ui/src/index.css b/graphrag-ui/src/index.css index 22dc36e5..560ce602 100755 --- a/graphrag-ui/src/index.css +++ b/graphrag-ui/src/index.css @@ -78,6 +78,32 @@ } } +/* Streaming Stop control — must apply in BOTH light and dark themes. While a + response streams, ActionProvider toggles `chat-streaming` on document.body; + Bot.tsx hides the native paper-plane and portals a red Stop icon into the + send button. These previously lived inside `.dark` only, which left the + stop icon unsized in light mode (it fell back to the SVG's default size and + broke the layout). The `.react-chatbot-kit-chat-btn-send` prefix keeps the + fill from being overridden by the base `... svg { fill: #999 }` rule. */ +body.chat-streaming .react-chatbot-kit-chat-input { + pointer-events: none; + opacity: 0.5; +} +body.chat-streaming .react-chatbot-kit-chat-btn-send-icon { + display: none; +} +.react-chatbot-kit-chat-btn-send .graphrag-stop-icon { + display: block; + width: 15px; + height: 15px; + margin: 0 auto; + fill: #dc2626; + cursor: pointer; +} +.react-chatbot-kit-chat-btn-send:hover .graphrag-stop-icon { + fill: #b91c1c; +} + .dark { a { @apply text-white; @@ -88,32 +114,6 @@ .react-chatbot-kit-chat-input-container { @apply !bg-background !border-[#3D3D3D]; } - /* While a response streams, lock the text input. The Send button keeps its - rounded cap (background) but its paper-plane icon is hidden and its click - disabled; a red Stop icon is overlaid in the exact same spot (Bot.tsx - portals it into the input-container). ActionProvider toggles - ``chat-streaming`` on ``document.body`` at stream start / end. */ - body.chat-streaming .react-chatbot-kit-chat-input { - pointer-events: none; - opacity: 0.5; - } - /* Replace the send icon with a red Stop icon IN PLACE — same button, same - position. Bot.tsx hides the native paper-plane and portals the stop icon - into the send button; the button's click is intercepted to stop. */ - body.chat-streaming .react-chatbot-kit-chat-btn-send-icon { - display: none; - } - .graphrag-stop-icon { - display: block; - width: 15px; - height: 15px; - margin: 0 auto; - fill: #dc2626; - cursor: pointer; - } - .react-chatbot-kit-chat-btn-send:hover .graphrag-stop-icon { - fill: #b91c1c; - } .open-dg { @apply bg-background; } diff --git a/graphrag-ui/src/pages/TraceLogs.tsx b/graphrag-ui/src/pages/TraceLogs.tsx index a038655c..185b86fe 100644 --- a/graphrag-ui/src/pages/TraceLogs.tsx +++ b/graphrag-ui/src/pages/TraceLogs.tsx @@ -669,7 +669,7 @@ const TokenOverviewPanel: FC<{ trace: TraceData }> = ({ trace }) => { - Cost is estimated based on the model's published per-token pricing. Actual billing may differ. + Cost uses rates from LLM Config when set; otherwise LangChain pricing. Actual billing may differ. @@ -705,7 +705,7 @@ const TokenOverviewPanel: FC<{ trace: TraceData }> = ({ trace }) => { - Cost is estimated based on the model's published per-token pricing. Actual billing may differ. + Cost uses rates from LLM Config when set; otherwise LangChain pricing. Actual billing may differ. @@ -750,7 +750,7 @@ const TokenOverviewPanel: FC<{ trace: TraceData }> = ({ trace }) => { {formatNumber(usage.total_tokens)} - + {formatCost(usage.cost)} diff --git a/graphrag-ui/src/pages/setup/KGAdmin.tsx b/graphrag-ui/src/pages/setup/KGAdmin.tsx index cd43337e..2cf23b5a 100644 --- a/graphrag-ui/src/pages/setup/KGAdmin.tsx +++ b/graphrag-ui/src/pages/setup/KGAdmin.tsx @@ -79,9 +79,18 @@ const KGAdmin = () => { missing_files: string[]; }; needs_repair?: boolean; + embeddings?: { + by_type: Record; + total_missing: number; + }; + embeddings_incomplete?: boolean; + community_summaries?: { total: number; needs_resummarize: number }; + community_summaries_incomplete?: boolean; } | null>(null); const [migrationChecking, setMigrationChecking] = useState(false); const [migrationApplying, setMigrationApplying] = useState(false); + // "" | "regenerate_embeddings" | "regenerate_summaries" — which regen is running + const [migrationRegenerating, setMigrationRegenerating] = useState(""); const [migrationMessage, setMigrationMessage] = useState(""); // Reset states when dialogs close const handleInitializeDialogChange = (open: boolean) => { @@ -145,14 +154,27 @@ const KGAdmin = () => { return; } setMigrationStatus(data); - if (!data.needs_repair) { + if ( + !data.needs_repair && + !data.embeddings_incomplete && + !data.community_summaries_incomplete + ) { setMigrationMessage("✅ Graph is up to date — no repairs needed."); } else { + const parts: string[] = []; const out = data.queries?.outdated?.length || 0; const miss = data.queries?.not_installed?.length || 0; - setMigrationMessage( - `Found ${out} outdated query(s) and ${miss} not installed.` - ); + if (out || miss) + parts.push(`${out} outdated query(s), ${miss} not installed`); + if (data.embeddings_incomplete) + parts.push( + `${data.embeddings?.total_missing ?? 0} vertices missing embeddings` + ); + if (data.community_summaries_incomplete) + parts.push( + `${data.community_summaries?.needs_resummarize ?? 0} communities need re-summarization` + ); + setMigrationMessage(`Found: ${parts.join("; ")}.`); } } catch (err: any) { setMigrationMessage(`Check failed: ${err.message || err}`); @@ -226,6 +248,49 @@ const KGAdmin = () => { } }; + // Targeted data-integrity regeneration (not a full rebuild). action is + // "regenerate_embeddings" or "regenerate_summaries". + const runRegenerate = async ( + action: "regenerate_embeddings" | "regenerate_summaries" + ) => { + const auth = sessionStorage.getItem("auth"); + if (!auth) { + setMigrationMessage("Not authenticated."); + return; + } + const isEmb = action === "regenerate_embeddings"; + setMigrationRegenerating(action); + setMigrationMessage( + isEmb ? "Regenerating embeddings…" : "Regenerating community summaries…" + ); + try { + const resp = await fetch(`/ui/${migrationGraph}/migration/${action}`, { + method: "POST", + headers: { Authorization: auth }, + }); + const data = await resp.json().catch(() => ({})); + if (!resp.ok) { + setMigrationMessage( + `Regenerate failed: ${data.detail || resp.statusText}` + ); + return; + } + const done = isEmb ? data.regenerated ?? 0 : data.resummarized ?? 0; + const skipped = data.skipped ?? 0; + const verb = isEmb ? "Re-embedded" : "Re-summarized"; + setMigrationMessage( + `✅ ${verb} ${done}` + + (skipped ? `; ${skipped} skipped (need a rebuild).` : ".") + ); + // Refresh so the counts reflect the regenerated state. + await runMigrationCheck(migrationGraph); + } catch (err: any) { + setMigrationMessage(`Regenerate failed: ${err.message || err}`); + } finally { + setMigrationRegenerating(""); + } + }; + const handleRefreshDialogChange = (open: boolean) => { if (!open && isConfirmDialogOpen) { return; @@ -2888,6 +2953,97 @@ const KGAdmin = () => { )} + + {/* Data-integrity health: embedding coverage */} + {migrationStatus.embeddings && + Object.keys(migrationStatus.embeddings.by_type).length > 0 && ( +
+
+
+ Embedding health + {migrationStatus.embeddings.total_missing > 0 + ? ` — ${migrationStatus.embeddings.total_missing} missing` + : " — all embedded"} +
+ {migrationStatus.embeddings_incomplete && ( + + )} +
+
+ {Object.entries(migrationStatus.embeddings.by_type) + .map( + ([t, c]) => `${t}: ${c.missing}/${c.total} missing` + ) + .join(" · ")} +
+
+ )} + + {/* Data-integrity health: community summaries */} + {migrationStatus.community_summaries && + (migrationStatus.community_summaries.total ?? 0) > 0 && ( +
+
+
+ Community summaries —{" "} + {migrationStatus.community_summaries.needs_resummarize}/ + {migrationStatus.community_summaries.total} need + re-summarization +
+ {migrationStatus.community_summaries_incomplete && ( + + )} +
+
+ )} )} diff --git a/graphrag-ui/src/pages/setup/LLMConfig.tsx b/graphrag-ui/src/pages/setup/LLMConfig.tsx index 382bf060..77376236 100644 --- a/graphrag-ui/src/pages/setup/LLMConfig.tsx +++ b/graphrag-ui/src/pages/setup/LLMConfig.tsx @@ -140,6 +140,9 @@ const LLMConfig = () => { const [completionProvider, setCompletionProvider] = useState("openai"); const [completionConfig, setCompletionConfig] = useState>({}); const [completionDefaultModel, setCompletionDefaultModel] = useState(""); + // Optional USD per 1M tokens — when both set, override LangChain cost + const [completionInputCostPer1m, setCompletionInputCostPer1m] = useState(""); + const [completionOutputCostPer1m, setCompletionOutputCostPer1m] = useState(""); const [embeddingProvider, setEmbeddingProvider] = useState("openai"); const [embeddingConfig, setEmbeddingConfig] = useState>({}); @@ -148,6 +151,8 @@ const LLMConfig = () => { const [multimodalProvider, setMultimodalProvider] = useState("openai"); const [multimodalConfig, setMultimodalConfig] = useState>({}); const [multimodalModelName, setMultimodalModelName] = useState(""); + const [multimodalInputCostPer1m, setMultimodalInputCostPer1m] = useState(""); + const [multimodalOutputCostPer1m, setMultimodalOutputCostPer1m] = useState(""); const isChatbotOnlyMode = llmConfigAccess === "chatbot_only"; // Per-graph chatbot config state (chatbot_only mode) @@ -156,6 +161,8 @@ const LLMConfig = () => { const [chatbotProviderConfig, setChatbotProviderConfig] = useState>({}); const [chatbotModelName, setChatbotModelName] = useState(""); const [chatbotTemperature, setChatbotTemperature] = useState("0"); + const [chatbotInputCostPer1m, setChatbotInputCostPer1m] = useState(""); + const [chatbotOutputCostPer1m, setChatbotOutputCostPer1m] = useState(""); const [globalChatInfo, setGlobalChatInfo] = useState({ llm_service: "", llm_model: "" }); // Superadmin scope: "global" edits global config, "graph" edits per-graph overrides @@ -274,6 +281,16 @@ const LLMConfig = () => { setChatbotProvider(data.chatbot_config.llm_service?.toLowerCase() || defaultProv); setChatbotModelName(data.chatbot_config.llm_model || ""); setChatbotTemperature(String(data.chatbot_config.model_kwargs?.temperature ?? "0")); + setChatbotInputCostPer1m( + data.chatbot_config.input_cost_per_1m != null + ? String(data.chatbot_config.input_cost_per_1m) + : "" + ); + setChatbotOutputCostPer1m( + data.chatbot_config.output_cost_per_1m != null + ? String(data.chatbot_config.output_cost_per_1m) + : "" + ); // Resolve chatbot config: base config + chatbot overrides setChatbotProviderConfig(loadServiceConfigResolved(data.chatbot_config)); } else { @@ -297,18 +314,40 @@ const LLMConfig = () => { setChatbotProvider(chatProv || defaultProv); setChatbotModelName(llmConfig.chat_service.llm_model || ""); setChatbotTemperature(String(llmConfig.chat_service.model_kwargs?.temperature ?? "0")); + setChatbotInputCostPer1m( + llmConfig.chat_service.input_cost_per_1m != null + ? String(llmConfig.chat_service.input_cost_per_1m) + : "" + ); + setChatbotOutputCostPer1m( + llmConfig.chat_service.output_cost_per_1m != null + ? String(llmConfig.chat_service.output_cost_per_1m) + : "" + ); setChatbotProviderConfig(loadServiceConfigResolved(llmConfig.chat_service)); } else { setUseCustomChatbot(false); setChatbotProvider(defaultProv); setChatbotModelName(""); setChatbotTemperature("0"); + setChatbotInputCostPer1m(""); + setChatbotOutputCostPer1m(""); setChatbotProviderConfig({ ...baseConfig }); } // Canonical per-service state — both single and multi-provider UIs read these setCompletionProvider(completionProv || "openai"); setCompletionDefaultModel(llmConfig.completion_service?.llm_model || ""); + setCompletionInputCostPer1m( + llmConfig.completion_service?.input_cost_per_1m != null + ? String(llmConfig.completion_service.input_cost_per_1m) + : "" + ); + setCompletionOutputCostPer1m( + llmConfig.completion_service?.output_cost_per_1m != null + ? String(llmConfig.completion_service.output_cost_per_1m) + : "" + ); setCompletionConfig(loadServiceConfigResolved(llmConfig.completion_service)); setEmbeddingProvider(embeddingProv || completionProv || "openai"); @@ -318,6 +357,16 @@ const LLMConfig = () => { setMultimodalProvider(multimodalProv || completionProv || "openai"); const mmModel = llmConfig.multimodal_service?.llm_model || ""; setMultimodalModelName(mmModel); + setMultimodalInputCostPer1m( + llmConfig.multimodal_service?.input_cost_per_1m != null + ? String(llmConfig.multimodal_service.input_cost_per_1m) + : "" + ); + setMultimodalOutputCostPer1m( + llmConfig.multimodal_service?.output_cost_per_1m != null + ? String(llmConfig.multimodal_service.output_cost_per_1m) + : "" + ); setMultimodalConfig(loadServiceConfigResolved(llmConfig.multimodal_service)); setUseCustomMultimodal(!!mmModel || !!multimodalProv); } catch (error: any) { @@ -393,6 +442,70 @@ const LLMConfig = () => { return serviceConfig; }; + /** Attach optional USD-per-1M rates when both fields are filled. */ + const attachTokenCosts = (svc: Record, input: string, output: string) => { + const inp = input.trim(); + const out = output.trim(); + if (inp === "" || out === "") return; + const inpN = Number(inp); + const outN = Number(out); + if (Number.isNaN(inpN) || Number.isNaN(outN) || inpN < 0 || outN < 0) return; + svc.input_cost_per_1m = inpN; + svc.output_cost_per_1m = outN; + }; + + const renderTokenCostFields = ( + inputValue: string, + outputValue: string, + setInput: (v: string) => void, + setOutput: (v: string) => void, + ) => ( +
+ +
+
+ + { + setInput(e.target.value); + clearTestResults(); + }} + /> +
+
+ + { + setOutput(e.target.value); + clearTestResults(); + }} + /> +
+
+

+ When both are set, Est. Cost always uses these rates (overrides LangChain). Leave empty to use LangChain pricing. +

+
+ ); + /** * Build the candidate LLM config payload. * Used by both test-connection and save — same structure, single source of truth. @@ -410,6 +523,11 @@ const LLMConfig = () => { prompt_path: `./common/prompts/${getPromptPath(completionProvider)}/`, ...buildServiceConfig(completionProvider, completionConfig) }; + attachTokenCosts( + completionServiceConfig, + completionInputCostPer1m, + completionOutputCostPer1m, + ); llmConfigData = { graphname: selectedGraph || undefined, @@ -430,6 +548,11 @@ const LLMConfig = () => { model_kwargs: { temperature: 0 }, ...buildServiceConfig(multimodalProvider, multimodalConfig) }; + attachTokenCosts( + llmConfigData.multimodal_service, + multimodalInputCostPer1m, + multimodalOutputCostPer1m, + ); } else { llmConfigData.multimodal_service = null; } @@ -442,6 +565,11 @@ const LLMConfig = () => { model_kwargs: { temperature: parseFloat(chatbotTemperature) || 0 }, ...buildServiceConfig(chatbotProvider, chatbotProviderConfig), }; + attachTokenCosts( + llmConfigData.chat_service, + chatbotInputCostPer1m, + chatbotOutputCostPer1m, + ); } else { llmConfigData.chat_service = null; } @@ -453,6 +581,11 @@ const LLMConfig = () => { prompt_path: `./common/prompts/${getPromptPath(completionProvider)}/`, ...buildServiceConfig(completionProvider, completionConfig) }; + attachTokenCosts( + completionServiceConfig, + completionInputCostPer1m, + completionOutputCostPer1m, + ); llmConfigData = { graphname: selectedGraph || undefined, @@ -468,6 +601,11 @@ const LLMConfig = () => { llmConfigData.multimodal_service = { llm_model: multimodalModelName, }; + attachTokenCosts( + llmConfigData.multimodal_service, + multimodalInputCostPer1m, + multimodalOutputCostPer1m, + ); } else { llmConfigData.multimodal_service = null; } @@ -478,6 +616,11 @@ const LLMConfig = () => { ...(chatbotModelName.trim() ? { llm_model: chatbotModelName } : {}), model_kwargs: { temperature: chatTemp }, }; + attachTokenCosts( + llmConfigData.chat_service, + chatbotInputCostPer1m, + chatbotOutputCostPer1m, + ); } else { llmConfigData.chat_service = null; } @@ -509,6 +652,7 @@ const LLMConfig = () => { model_kwargs: { temperature: parseFloat(chatbotTemperature) || 0 }, ...buildServiceConfig(chatbotProvider, chatbotProviderConfig), }; + attachTokenCosts(chatService, chatbotInputCostPer1m, chatbotOutputCostPer1m); llmConfigData = { graphname: selectedGraph || undefined, chat_service: chatService }; } else { // Revert to inherit: send null chat_service @@ -916,6 +1060,13 @@ const LLMConfig = () => { /> + {renderTokenCostFields( + chatbotInputCostPer1m, + chatbotOutputCostPer1m, + setChatbotInputCostPer1m, + setChatbotOutputCostPer1m, + )} +
+ {renderTokenCostFields( + completionInputCostPer1m, + completionOutputCostPer1m, + setCompletionInputCostPer1m, + setCompletionOutputCostPer1m, + )} +
@@ -1171,16 +1329,26 @@ const LLMConfig = () => {
{useCustomChatbot && ( - { - setChatbotModelName(e.target.value); - clearTestResults(); - }} - /> + <> + { + setChatbotModelName(e.target.value); + clearTestResults(); + }} + /> +
+ {renderTokenCostFields( + chatbotInputCostPer1m, + chatbotOutputCostPer1m, + setChatbotInputCostPer1m, + setChatbotOutputCostPer1m, + )} +
+ )}

Used by the chatbot for answering user questions @@ -1233,16 +1401,26 @@ const LLMConfig = () => {

)} {useCustomMultimodal && ( - { - setMultimodalModelName(e.target.value); - clearTestResults(); - }} - /> + <> + { + setMultimodalModelName(e.target.value); + clearTestResults(); + }} + /> +
+ {renderTokenCostFields( + multimodalInputCostPer1m, + multimodalOutputCostPer1m, + setMultimodalInputCostPer1m, + setMultimodalOutputCostPer1m, + )} +
+ )}

Used for processing images and multimodal content @@ -1324,6 +1502,13 @@ const LLMConfig = () => { Model for entity extraction and community summarization.

+ + {renderTokenCostFields( + completionInputCostPer1m, + completionOutputCostPer1m, + setCompletionInputCostPer1m, + setCompletionOutputCostPer1m, + )} @@ -1394,6 +1579,13 @@ const LLMConfig = () => { /> + {renderTokenCostFields( + chatbotInputCostPer1m, + chatbotOutputCostPer1m, + setChatbotInputCostPer1m, + setChatbotOutputCostPer1m, + )} +
+ + {renderTokenCostFields( + multimodalInputCostPer1m, + multimodalOutputCostPer1m, + setMultimodalInputCostPer1m, + setMultimodalOutputCostPer1m, + )} )} diff --git a/graphrag/app/agent/agent.py b/graphrag/app/agent/agent.py index 9f82df84..7dc55fa3 100644 --- a/graphrag/app/agent/agent.py +++ b/graphrag/app/agent/agent.py @@ -292,14 +292,14 @@ def make_agent(graphname, conn, use_cypher, ws: WebSocket = None, supportai_retr caller can surface it to the user. """ from common.config import get_agent_mode - from common.llm_services.capabilities import model_supports_agentic + from common.llm_services.capabilities import supports_tool_calling llm_provider = get_llm_service(get_chat_config(graphname)) chat_config = llm_provider.config resolved_mode = (mode or get_agent_mode(graphname)).lower() want_agentic = resolved_mode == "agentic" - agentic = want_agentic and model_supports_agentic(chat_config) + agentic = want_agentic and supports_tool_calling(chat_config, llm_provider) logger.info( f"[CHATBOT] graph={graphname} model={chat_config['llm_model']} " diff --git a/graphrag/app/agent/agent_generation.py b/graphrag/app/agent/agent_generation.py index 68074f33..d6dea51a 100644 --- a/graphrag/app/agent/agent_generation.py +++ b/graphrag/app/agent/agent_generation.py @@ -83,7 +83,12 @@ def generate_answer(self, question: str, context: str | dict, query: str = "") - except Exception: logger.warning("generate_answer: generation failed") generation = GraphRAGAnswerOutput( - generated_answer="I wasn't able to generate an answer for this question.", + generated_answer=( + "I wasn't able to generate an answer for this question. " + "Try asking again, or rephrase it to be more specific or " + "focused on a single topic. If the problem continues, " + "contact your administrator for more details." + ), citation=[], ) diff --git a/graphrag/app/agent/agentic_agent.py b/graphrag/app/agent/agentic_agent.py index 77f32d14..b585c6e7 100644 --- a/graphrag/app/agent/agentic_agent.py +++ b/graphrag/app/agent/agentic_agent.py @@ -128,6 +128,7 @@ def __init__( self.cypher_gen = GenerateCypher(self.conn, self.llm) if use_cypher else None self.q = Q() if ws is not None else None + self._ws = ws logger.debug(f"request_id={req_id_cv.get()} agentic agent initialized") @@ -235,11 +236,35 @@ def question_for_agent( # config; "auto" defers to the configured default. config_style = (ctx.graphrag_cfg or {}).get("agent_style", "planned") style = _resolve_style(self.agent_style, config_style) - if style == "planned": - answer = run_agentic(ctx, self.llm, question, convo) - else: - # "reactive" (UI) / "react" (config) -> free tool-calling loop - answer = run_react(ctx, self.llm, question, convo) + try: + if style == "planned": + answer = run_agentic(ctx, self.llm, question, convo) + else: + # "reactive" (UI) / "react" (config) -> free tool-calling loop + answer = run_react(ctx, self.llm, question, convo) + except Exception as run_exc: + # Runtime backstop (GML-2169): if the model turns out not to + # support tool-calling, disable Agentic for it and answer via the + # classic engine. Only trigger on a confident tool-support signal; + # any other error propagates to the normal handler. + from common.llm_services.capabilities import ( + _looks_like_no_tool_support, + mark_tool_calling_unsupported, + ) + if not _looks_like_no_tool_support(run_exc): + raise + logger.warning( + f"request_id={req_id_cv.get()} agentic run hit a tool-calling " + f"failure ({str(run_exc)[:200]}); disabling Agentic for this " + "model and falling back to the classic engine" + ) + mark_tool_calling_unsupported(self.llm.config) + from agent.agent import make_agent + classic = make_agent( + self.conn.graphname, self.conn, self.use_cypher, + ws=self._ws, mode="classic", + ) + return classic.question_for_agent(question, conversation) # Aggregate usage across all LLM calls in this run for the UI. usage = get_collected_usage() or [] diff --git a/graphrag/app/agent/agentic_graph.py b/graphrag/app/agent/agentic_graph.py index 2644e625..7a88c02c 100644 --- a/graphrag/app/agent/agentic_graph.py +++ b/graphrag/app/agent/agentic_graph.py @@ -24,11 +24,11 @@ import logging import time -from agent.agentic_executor import cap_for_trace, execute_plan, _usage_since +from agent.agentic_executor import cap_for_trace, execute_plan, _run_step, _usage_since from agent.agentic_planner import plan_question from agent.agentic_synthesizer import _gather, has_context, synthesize from common.llm_services.base_llm import get_collected_usage -from common.py_schemas import GraphRAGResponse +from common.py_schemas import GraphRAGResponse, PlanStep logger = logging.getLogger(__name__) @@ -38,6 +38,22 @@ # ``agent_max_replans`` and ``agent_max_total_steps``. Raise them for # complex-system graphs (e.g. multi-hop what-if simulation). +# Deterministic fallback: a structural query that returns no rows leaves the +# answer with nothing to stand on. Rather than depend on the LLM planner to +# add an unstructured step on replan (it does so inconsistently), we guarantee +# a hybrid search runs before giving up. +_STRUCTURAL_TOOL = "graphrag__structural_retrieve" +_HYBRID_TOOL = "graphrag__hybrid_search" + + +def _hybrid_fallback_step() -> PlanStep: + return PlanStep( + id="fallback_hybrid", + kind="unstructured", + tool=_HYBRID_TOOL, + rationale="Structural query returned no rows; falling back to hybrid search", + ) + def run_agentic(ctx, llm, question, conversation=None) -> GraphRAGResponse: """Run the agentic workflow for one question and return a response. @@ -80,6 +96,22 @@ def run_agentic(ctx, llm, question, conversation=None) -> GraphRAGResponse: results.update(new_results) agent_steps.extend(step_traces) + # Deterministic safety net: if a structural retrieve was attempted and + # produced no context, fall back to a hybrid search directly rather + # than relying on the planner to add one on replan. Runs at most once. + # Shares the classic engine's ``enable_router_fallback`` knob (default + # True) so both engines fall back — or don't — consistently. + used = {t.get("tool") for t in agent_steps} + if (_cfg.get("enable_router_fallback", True) + and not has_context(results) + and _STRUCTURAL_TOOL in used + and _HYBRID_TOOL not in used): + emit("No structured results; falling back to hybrid search") + fb_traces: list = [] + _run_step(_hybrid_fallback_step(), {"question": question}, + ctx, results, fb_traces) + agent_steps.extend(fb_traces) + if has_context(results) or replans >= max_replans or len(results) >= max_total_steps: break diff --git a/graphrag/app/main.py b/graphrag/app/main.py index 56772e9e..cfa79aaa 100644 --- a/graphrag/app/main.py +++ b/graphrag/app/main.py @@ -47,18 +47,18 @@ async def lifespan(app: FastAPI): except Exception as e: logging.getLogger(__name__).warning(f"mcp library install failed: {e}") - # Warn if the configured chat model can't tool-call, so operators know the - # agentic engine will fall back to the classic engine. + # Agentic mode is on by default and confirmed by a runtime probe on first + # use. Only warn when the configured chat model is a known-legacy model we + # are sure can't tool-call, so Agentic will run as the classic engine. try: from common.config import get_chat_config - from common.llm_services.capabilities import model_capabilities + from common.llm_services.capabilities import _known_no_tool_calling cfg = get_chat_config() - if not model_capabilities(cfg).get("supports_tool_calling"): + if _known_no_tool_calling(cfg): logging.getLogger(__name__).warning( - "Chat model llm_service=%r llm_model=%r does not support " - "tool-calling; the agentic chat engine is unavailable and " - "requests will use the classic engine. Configure a " - "tool-calling model to enable Agentic mode.", + "Chat model llm_service=%r llm_model=%r is a legacy model without " + "tool-calling; Agentic mode is unavailable and requests will use " + "the classic engine. Configure a newer model to enable Agentic mode.", (cfg or {}).get("llm_service"), (cfg or {}).get("llm_model"), ) except Exception as e: diff --git a/graphrag/app/routers/ui.py b/graphrag/app/routers/ui.py index dc72ef12..a482e90f 100644 --- a/graphrag/app/routers/ui.py +++ b/graphrag/app/routers/ui.py @@ -1372,6 +1372,32 @@ def migration_status( except Exception as e: logger.warning(f"migration_status prompt check failed: {e}") + # Data-integrity health: embedding coverage and community-summary + # completeness. DETERMINISTIC, read-only, NO LLM calls — same contract as + # the query/prompt checks. Reported separately from ``needs_repair`` because + # neither is fixed by a query reinstall (they have their own regenerate + # actions). Best-effort, never fatal. + embeddings_by_type: dict = {} + embeddings_total_missing = 0 + community_summaries: dict = {} + try: + from common.db.health import ( + embeddable_types, + embedding_coverage, + community_summary_health, + ) + + for vt in embeddable_types(conn): + cov = embedding_coverage(conn, vt) + if cov is not None: + embeddings_by_type[vt] = cov + embeddings_total_missing += cov["missing"] + csh = community_summary_health(conn) + if csh is not None: + community_summaries = csh + except Exception as e: + logger.warning(f"migration_status health check failed: {e}") + return { "graphname": graphname, "queries": { @@ -1387,10 +1413,57 @@ def migration_status( "schema_change_required": False, }, "prompts": prompt_issues, + "embeddings": { + "by_type": embeddings_by_type, + "total_missing": embeddings_total_missing, + }, + "embeddings_incomplete": embeddings_total_missing > 0, + "community_summaries": community_summaries, + "community_summaries_incomplete": bool( + community_summaries.get("needs_resummarize", 0) + ), "needs_repair": bool(outdated) or bool(not_installed) or bool(prompt_issues), } +async def _proxy_regen(graphname, creds, action): + """Forward a targeted regenerate action to ECC (synchronous) and return its + counts. Refuses while a rebuild is in flight (both write embeddings).""" + if get_rebuilding_graph() == graphname: + raise HTTPException( + status_code=409, + detail=f"Graph '{graphname}' is being rebuilt; retry after it completes.", + ) + auth_header = _ecc_auth_header(creds[1]) + ecc_base = graphrag_config.get("ecc", "http://graphrag-ecc:8001") + url = f"{ecc_base}/{graphname}/graphrag/{action}" + async with httpx.AsyncClient(timeout=None) as client: + resp = await client.get(url, headers={"Authorization": auth_header}) + if resp.status_code != 200: + raise HTTPException(status_code=resp.status_code, detail=resp.text[:300]) + return resp.json() + + +@router.post(route_prefix + "/{graphname}/migration/regenerate_embeddings") +async def migration_regenerate_embeddings( + graphname: ValidGraphName, + creds: Annotated[tuple[list[str], HTTPBasicCredentials], Depends(ui_basic_auth)], +): + """Re-embed vertices missing an embedding (GML-2175). Targeted — not a full + rebuild. Returns {regenerated, skipped}.""" + return await _proxy_regen(graphname, creds, "regenerate_embeddings") + + +@router.post(route_prefix + "/{graphname}/migration/regenerate_summaries") +async def migration_regenerate_summaries( + graphname: ValidGraphName, + creds: Annotated[tuple[list[str], HTTPBasicCredentials], Depends(ui_basic_auth)], +): + """Re-summarize communities with placeholder/empty summaries and re-embed + (GML-2176). Targeted — not a full rebuild. Returns {resummarized, skipped}.""" + return await _proxy_regen(graphname, creds, "regenerate_summaries") + + @router.post(route_prefix + "/{graphname}/migration/apply") def migration_apply( graphname: ValidGraphName, diff --git a/graphrag/tests/regression/evaluator.py b/graphrag/tests/regression/evaluator.py index 3638fd44..528a40a3 100644 --- a/graphrag/tests/regression/evaluator.py +++ b/graphrag/tests/regression/evaluator.py @@ -434,7 +434,7 @@ def _query_graphrag( "include_fields": "query_sources", }, auth=(username, password), - timeout=120.0, + timeout=600.0, # agentic planned mode can take >120s on complex multi-hop questions ) resp.raise_for_status() data = resp.json() diff --git a/graphrag/tests/regression/recall_evaluator.py b/graphrag/tests/regression/recall_evaluator.py new file mode 100644 index 00000000..45d30b8c --- /dev/null +++ b/graphrag/tests/regression/recall_evaluator.py @@ -0,0 +1,936 @@ +"""GraphRAG Regression — Recall@K Evaluator + +Measures retrieval recall against a labelled dataset using: + + Recall@K = (ground-truth chunks found in top-K retrieved chunks) + / (total ground-truth chunks for that question) + +Averaged over all questions to produce a final Avg Recall@K score. + +Dataset layout expected: + test_questions// + ├── data/_corpus.txt — one raw text file per source (GraphRAG chunks naturally) + ├── questions.csv — single column: question + └── ground_truth_chunks.csv — columns: question_index, chunk_index, context + Recall denominator = GT chunks per question (not total dataset chunks). + +Matching strategy — two options via --match: + + embedding (default): + Each GT chunk and each retrieved chunk are embedded with the model from + server_config.json (e.g. gemini-embedding-001). A GT chunk is "found" when + cosine similarity ≥ threshold (default 0.75). Handles paraphrasing and + contextual retrievers correctly. + + llm: + An LLM (same model as completions in server_config.json) judges whether the + retrieved chunk contains the same factual information as the GT chunk. + Most accurate but slowest and costliest. + +Run via: + ./graphrag/tests/regression/run_recall.sh --dataset Multihop30 --graphname +""" + +from __future__ import annotations + +import argparse +import csv +import json +import logging +import math +import os +import re +import sys +import threading +import time +import warnings +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import asdict, dataclass, field +from datetime import datetime +from typing import Any, Dict, List, Optional, Tuple + +import httpx + +if "/code" not in sys.path: + sys.path.insert(0, "/code") + +warnings.filterwarnings("ignore") +warnings.showwarning = lambda *a, **k: None +logging.basicConfig(level=logging.WARNING, format="%(levelname)s %(name)s: %(message)s") + +# ── ANSI colours (mirrors evaluator.py) ────────────────────────────────────── +if os.environ.get("NO_COLOR"): + _USE_COLOR = False +elif os.environ.get("FORCE_COLOR"): + _USE_COLOR = True +else: + _USE_COLOR = sys.stdout.isatty() + +_G = "\033[32m" if _USE_COLOR else "" +_R = "\033[31m" if _USE_COLOR else "" +_Y = "\033[33m" if _USE_COLOR else "" +_C = "\033[36m" if _USE_COLOR else "" +_B = "\033[1m" if _USE_COLOR else "" +_X = "\033[0m" if _USE_COLOR else "" + +# ── Constants ───────────────────────────────────────────────────────────────── +_DEFAULT_K = 5 +_MAX_CHUNKS = 50 # max chunks extracted from GraphRAG response +_MAX_CHUNK_CH = 5000 # truncation per chunk before matching — keep higher than GT chunk size (4000) +_EMBED_THRESHOLD = 0.70 # cosine similarity threshold for embedding match + # Calibrated on Multihop30: genuine misses cluster at 0.54–0.64; + # true matches cluster at 0.70–0.85. Gap is clear and stable. +_EMBED_MAX_CHARS = 8000 # truncate text before embedding (API limit safety) +_EMBED_CONCURRENCY = 4 # max parallel embedding API calls +_LLM_CHUNK_CHARS = 4000 # truncate chunk text sent to LLM judge + +# Corpus passages are stored in the graph with a "===== Title =====" header prepended +# by the ingestion pipeline. GT chunks never have this header, so we strip it from +# retrieved text before embedding / LLM comparison to avoid a spurious similarity drop. +_TITLE_HEADER_RE = re.compile(r"^\s*={2,}[^=\n]+={2,}\s*\n?", re.MULTILINE) + +# ── Shared caches (embedding vectors, LLM verdicts) ─────────────────────────── +_embed_cache: Dict[str, List[float]] = {} +_llm_cache: Dict[str, bool] = {} +_embed_lock = threading.Lock() +_llm_lock = threading.Lock() +_embed_sem = threading.Semaphore(_EMBED_CONCURRENCY) + + +# ─── Data types ─────────────────────────────────────────────────────────────── + +@dataclass +class RecallQuestion: + index: int # 1-based, matches question_index in ground_truth_contexts.csv + question: str + gt_contexts: List[str] # GT chunk text (from ground_truth_chunks.csv) + gt_titles: List[Optional[str]] = field(default_factory=list) # Wikipedia title (may be None) + + +@dataclass +class RecallResult: + # ── written to CSV ──────────────────────────────────────────────────────── + question_index: int + question: str + gt_context_count: int = 0 + retrieved_chunk_count: int = 0 + matched_count: int = 0 + recall_at_k: Optional[float] = None + agent_mode: Optional[str] = None + search_type_used: Optional[str] = None + response_time_seconds: float = 0.0 + # ── CLI-only ────────────────────────────────────────────────────────────── + error: Optional[str] = field(default=None, repr=False) + answered_question: bool = field(default=False, repr=False) + matched_indices: List[int] = field(default_factory=list, repr=False) + unmatched_contexts: List[str] = field(default_factory=list, repr=False) + embed_scores: List[float] = field(default_factory=list, repr=False) # max cosine per GT chunk + + +# ─── Loaders ───────────────────────────────────────────────────────────────── + +def _detect_encoding(path: str) -> str: + with open(path, "rb") as f: + raw = f.read(4) + if raw[:3] == b"\xef\xbb\xbf": + return "utf-8-sig" + try: + import chardet + with open(path, "rb") as f: + detected = chardet.detect(f.read()) + enc = detected.get("encoding") or "utf-8" + return enc if (detected.get("confidence") or 0) >= 0.5 else "utf-8" + except ImportError: + return "utf-8" + + +def _read_single_column(path: str, column: str) -> List[str]: + encoding = _detect_encoding(path) + with open(path, newline="", encoding=encoding) as f: + reader = csv.reader(f) + rows = list(reader) + if not rows: + sys.exit(f"ERROR: {os.path.basename(path)} is empty.") + header = rows[0] + try: + col_idx = header.index(column) + except ValueError: + sys.exit( + f"ERROR: {os.path.basename(path)} must have a '{column}' column. " + f"Found: {header}" + ) + results = [] + for row in rows[1:]: + if not row: + continue + value = ",".join(row[col_idx:]).strip() + if value: + results.append(value) + return results + + +def load_recall_questions(dataset_dir: str) -> List[RecallQuestion]: + """Load questions.csv and ground_truth_chunks.csv. + + ground_truth_chunks.csv columns: + question_index — 1-based int + chunk_index — chunk number within the question (1-based) + context — chunk text (~400-500 chars) + + Denominator for Recall@K is the number of GT chunks for THAT question, + not the total chunks in the dataset. + Run build_gt_chunks.py once after setup to generate this file. + """ + q_path = os.path.join(dataset_dir, "questions.csv") + chunks_path = os.path.join(dataset_dir, "ground_truth_chunks.csv") + + if not os.path.exists(q_path): + sys.exit(f"ERROR: questions.csv not found at {q_path}") + if not os.path.exists(chunks_path): + sys.exit( + f"ERROR: ground_truth_chunks.csv not found in {dataset_dir}.\n" + f"Run build_gt_chunks.py first:\n" + f" python {dataset_dir}/build_gt_chunks.py" + ) + + questions_raw = _read_single_column(q_path, "question") + + encoding = _detect_encoding(chunks_path) + gt_map: Dict[int, List[str]] = {} + with open(chunks_path, newline="", encoding=encoding) as f: + reader = csv.DictReader(f) + for row in reader: + try: + idx = int(row["question_index"]) + except (KeyError, ValueError): + continue + ctx = row.get("context", "").strip() + if ctx: + gt_map.setdefault(idx, []).append(ctx) + + result: List[RecallQuestion] = [] + for i, question in enumerate(questions_raw, start=1): + chunks = gt_map.get(i, []) + if not chunks: + print(f" {_Y}WARNING: no GT chunks for Q{i} — " + f"recall will be 0 for this question{_X}", flush=True) + result.append(RecallQuestion( + index=i, question=question, + gt_contexts=chunks, gt_titles=[], + )) + + return result + + +# ─── Embedding cosine similarity matching ──────────────────────────────────── + +def _cosine(a: List[float], b: List[float]) -> float: + dot = sum(x * y for x, y in zip(a, b)) + norm_a = math.sqrt(sum(x * x for x in a)) + norm_b = math.sqrt(sum(x * x for x in b)) + return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0 + + +def _get_embedding_service(): + """Return the GraphRAG embedding service (already configured in the container).""" + from common.config import get_embedding_service + return get_embedding_service() + + +def _embed(text: str, **_kwargs) -> Optional[List[float]]: + """Return embedding for text using GraphRAG's configured embedding service (cached). + Extra kwargs (api_key, model) are accepted but ignored — config comes from + the container's server_config.json via common.config.get_embedding_service(). + Returns None on any error. + """ + cache_key = text[:200] + with _embed_lock: + if cache_key in _embed_cache: + return _embed_cache[cache_key] + try: + with _embed_sem: + svc = _get_embedding_service() + vecs = svc.embed_documents([text[:_EMBED_MAX_CHARS]]) + vec = vecs[0] if vecs else None + if vec: + with _embed_lock: + _embed_cache[cache_key] = vec + return vec + except Exception as e: + logging.warning("Embedding error: %s", e) + return None + + +def _check_embedding_api(**_kwargs) -> Optional[str]: + """Return None if the embedding service works, else an error string.""" + vec = _embed("ping") + if vec is None: + return "embedding service unavailable — check container logs" + return None + + +def _chunk_found_embedding(gt_chunk: str, chunks: List[str], + threshold: float = _EMBED_THRESHOLD, + **_kwargs) -> Tuple[bool, float]: + """Check whether any retrieved chunk is similar enough to the GT chunk. + + Returns: + (found, max_cosine_similarity) + found — True if max_cosine_similarity >= threshold + max_cosine_similarity — highest cosine score observed (0.0 if no embeddings) + """ + gt_vec = _embed(gt_chunk) + if gt_vec is None: + return False, 0.0 + max_sim = 0.0 + for retrieved in chunks: + retrieved_vec = _embed(retrieved) + if retrieved_vec is not None: + sim = _cosine(gt_vec, retrieved_vec) + if sim > max_sim: + max_sim = sim + return max_sim >= threshold, max_sim + + +# ─── LLM judge matching ──────────────────────────────────────────────────────── + +_LLM_PROMPT = """\ +You are evaluating information retrieval quality for a RAG system. + +Ground-truth passage (the relevant source text): +{gt} + +Retrieved chunk (returned by the retriever): +{chunk} + +Task: Does the retrieved chunk contain information that is RELEVANT to the ground-truth passage? +The retrieved chunk does NOT need to be identical or cover everything — answer YES if it overlaps +on the same topic, entity, or key facts. Answer NO only if the retrieved chunk is clearly about +a completely different topic with no meaningful overlap. + +Answer with a single word: YES or NO.""" + + +def _llm_judge(gt_chunk: str, retrieved_chunk: str, **_kwargs) -> bool: + """Ask GraphRAG's configured LLM whether the retrieved chunk contains the GT chunk's information. + Uses the same get_llm_service / get_chat_config pattern as evaluator.py. + Returns True for YES, False for NO or on any error. + """ + cache_key = f"{gt_chunk[:100]}|||{retrieved_chunk[:100]}" + with _llm_lock: + if cache_key in _llm_cache: + return _llm_cache[cache_key] + try: + from common.config import get_chat_config, get_llm_service + from langchain_core.messages import HumanMessage + + cfg = get_chat_config() + llm = get_llm_service(cfg) + lc = getattr(llm, "llm", None) or llm + prompt = _LLM_PROMPT.format( + gt=gt_chunk[:_LLM_CHUNK_CHARS], + chunk=retrieved_chunk[:_LLM_CHUNK_CHARS], + ) + resp = lc.invoke([HumanMessage(content=prompt)]) + text = (resp.content if hasattr(resp, "content") else str(resp)).strip().upper() + verdict = text.startswith("YES") + with _llm_lock: + _llm_cache[cache_key] = verdict + return verdict + except Exception as e: + logging.warning("LLM judge error: %s", e) + return False + + +def _chunk_found_llm(gt_chunk: str, retrieved_chunks: List[str]) -> bool: + """True if the LLM judges any retrieved chunk as relevant to the GT chunk.""" + for retrieved in retrieved_chunks: + if _llm_judge(gt_chunk, retrieved): + return True + return False + + +# ─── Unified recall computation ─────────────────────────────────────────────── + +def _compute_recall( + gt_contexts: List[str], + top_k_chunks: List[str], + match_cfg: Optional[Dict[str, Any]] = None, + embed_threshold: float = _EMBED_THRESHOLD, + gt_titles: Optional[List[Optional[str]]] = None, # unused, kept for compat +) -> Tuple[float, int, List[int], List[str], List[float]]: + """Compute Recall@K for one question. + + match_cfg keys: + strategy — "embedding" (default) or "llm" + + Returns: + recall, matched_count, matched_indices, unmatched_chunks, embed_scores + embed_scores — max cosine similarity per GT chunk (empty list when strategy="llm") + """ + if not gt_contexts: + return 0.0, 0, [], [], [] + + strategy = (match_cfg or {}).get("strategy", "embedding") + + matched_idxs: List[int] = [] + unmatched: List[str] = [] + embed_scores: List[float] = [] + + for i, gt in enumerate(gt_contexts): + if strategy == "llm": + found = _chunk_found_llm(gt, top_k_chunks) + embed_scores.append(-1.0) # not applicable + else: + found, score = _chunk_found_embedding(gt, top_k_chunks, threshold=embed_threshold) + embed_scores.append(score) + + if found: + matched_idxs.append(i) + else: + unmatched.append(gt) + + recall = len(matched_idxs) / len(gt_contexts) + return recall, len(matched_idxs), matched_idxs, unmatched, embed_scores + + +# ─── GraphRAG query ─────────────────────────────────────────────────────────── + +_RETRIEVER_LABELS = { + "similaritysearch": "Similarity Search", + "contextualsearch": "Contextual Search", + "hybridsearch": "Hybrid Search", + "communitysearch": "Community Search", +} + + +def _query_graphrag( + url: str, + graphname: str, + username: str, + password: str, + question: str, + mode: str, + rag_pattern: str, +) -> dict: + resp = httpx.get( + f"{url}/ui/{graphname}/query", + params={ + "q": question, + "mode": mode, + "rag_pattern": rag_pattern, + "include_fields": "query_sources", + }, + auth=(username, password), + timeout=600.0, # agentic mode can take >300s on complex multi-hop questions + ) + resp.raise_for_status() + data = resp.json() + if isinstance(data, str): + data = json.loads(data) + return data + + +def _extract_chunk_texts(final_retrieval: dict) -> List[str]: + """Extract text strings from a final_retrieval dict. + + Each value is either: + - list[str] → one text per item (common case after ingestion) + - str → single text value + - dict → extract 'content' or 'text' sub-key + """ + texts: List[str] = [] + for v in final_retrieval.values(): + if isinstance(v, list): + for item in v: + if isinstance(item, str) and item.strip(): + texts.append(item) + elif isinstance(item, dict): + t = item.get("content") or item.get("text") or "" + if t: + texts.append(str(t)) + elif isinstance(v, str) and v.strip(): + texts.append(v) + elif isinstance(v, dict): + t = v.get("content") or v.get("text") or "" + if t: + texts.append(str(t)) + return texts + + +def _extract_top_k_chunks(query_sources: dict, k: int, mode: str = "classic") -> List[str]: + """Extract retrieved text chunks from the GraphRAG response. + + Classic mode (similarity/hybrid/contextual/community search): + Chunks are ordered by relevance score — most relevant first. + We return only the first k so Recall@K is honest and cannot + benefit from chunks beyond rank k. + + Agentic mode (planned/reactive/auto agent): + The agent calls multiple tools in arbitrary sequence — there is no + single relevance ranking across tool calls. Applying a k-cutoff + here would silently drop relevant chunks the agent DID retrieve but + that happen to appear after position k in dict-iteration order. + We therefore return ALL chunks the agent retrieved (up to _MAX_CHUNKS) + so the recall score reflects what the agent actually found, not an + arbitrary ordering artefact. + + Chunk extraction strategy: + 1. Top-level result.final_retrieval — populated in classic mode. + 2. Per-step result.unstructured[*].result.final_retrieval — populated + in agentic mode for each vector/hybrid search tool call. + 3. NO general fallback walker: agentic structural-only answers produce + no text chunks; falling back to walking the whole response would + extract Cypher query strings and reasoning text as fake chunks, leading + to artificially high LLM recall (entity name matching) and artificially + low embedding recall (near-zero cosine similarity with real passages). + + Title-header stripping: + The ingestion pipeline prepends "===== Title =====" to each stored chunk. + GT chunks contain only the passage body, so we strip headers before + comparison to avoid spurious similarity penalties. + """ + if not query_sources: + return [] + + result = query_sources.get("result") or {} + raw: List[str] = [] + + # ── Path 1: top-level final_retrieval (classic mode, and some agentic configs) ── + top_fr = result.get("final_retrieval") or {} + if isinstance(top_fr, dict) and top_fr: + raw.extend(_extract_chunk_texts(top_fr)) + + # ── Path 2: per-step unstructured results (agentic mode) ──────────────────────── + if not raw: + for u_item in result.get("unstructured") or []: + u_result = u_item.get("result") if isinstance(u_item, dict) else None + if not isinstance(u_result, dict): + continue + inner_fr = u_result.get("final_retrieval") or {} + if isinstance(inner_fr, dict) and inner_fr: + raw.extend(_extract_chunk_texts(inner_fr)) + + # ── Deduplicate, strip title headers, apply length guard, cap at _MAX_CHUNKS ──── + seen: set = set() + out: List[str] = [] + for text in raw: + # Strip "===== Title =====" header added by ingestion pipeline + text = _TITLE_HEADER_RE.sub("", text, count=1).strip() + if len(text) < 5 or text in seen: + continue + seen.add(text) + out.append(text[:_MAX_CHUNK_CH]) + if len(out) >= _MAX_CHUNKS: + break + + # For classic mode honour the K limit (ranked by relevance). + # For agentic mode return everything the agent retrieved — no meaningful rank exists. + if mode == "agentic": + return out + return out[:k] + + +# ─── Per-question evaluation ────────────────────────────────────────────────── + +def _eval_one_recall( + rq: RecallQuestion, + graphname: str, + url: str, + username: str, + password: str, + mode: str, + rag_pattern: str, + k: int, + match_cfg: Optional[Dict[str, Any]] = None, + embed_threshold: float = _EMBED_THRESHOLD, +) -> RecallResult: + r = RecallResult( + question_index=rq.index, + question=rq.question, + gt_context_count=len(rq.gt_contexts), + ) + t0 = time.monotonic() + + try: + data = _query_graphrag(url, graphname, username, password, rq.question, mode, rag_pattern) + r.answered_question = bool(data.get("answered_question", False)) + query_sources = data.get("query_sources") or {} + _chosen = (query_sources.get("chosen_retriever") or "").lower().replace(" ", "") + r.search_type_used = _RETRIEVER_LABELS.get(_chosen) or data.get("response_type") or None + r.agent_mode = f"{mode}/{rag_pattern}" + + top_k = _extract_top_k_chunks(query_sources, k, mode=mode) + r.retrieved_chunk_count = len(top_k) + except httpx.HTTPStatusError as e: + r.error = f"HTTP {e.response.status_code}: {e.response.text[:200]}" + r.response_time_seconds = time.monotonic() - t0 + return r + except Exception as e: + r.error = f"{type(e).__name__}: {e}" + r.response_time_seconds = time.monotonic() - t0 + return r + finally: + r.response_time_seconds = time.monotonic() - t0 + + if not rq.gt_contexts: + r.recall_at_k = 0.0 + return r + + if not top_k: + # No text chunks retrieved — likely a structural-only (graph traversal) answer. + # Retrieval recall cannot be measured without text chunks. + r.recall_at_k = 0.0 + r.matched_count = 0 + r.unmatched_contexts = list(rq.gt_contexts) + return r + + recall, matched, matched_idxs, unmatched, embed_scores = _compute_recall( + rq.gt_contexts, top_k, + match_cfg=match_cfg, + embed_threshold=embed_threshold, + ) + r.recall_at_k = round(recall, 4) + r.matched_count = matched + r.matched_indices = matched_idxs + r.unmatched_contexts = unmatched + r.embed_scores = embed_scores + + return r + + +# ─── Compact / detailed printer ────────────────────────────────────────────── + +def _recall_cell(r: RecallResult) -> str: + if r.recall_at_k is None: + return f"{_Y} n/a{_X}" + pct = r.recall_at_k * 100 + col = _G if pct >= 60 else _Y if pct >= 30 else _R + bar_filled = int(pct / 10) + bar = "█" * bar_filled + "░" * (10 - bar_filled) + return f"{col}{pct:5.1f}% {bar}{_X}" + + +def _print_compact(r: RecallResult, total: int, k: int) -> None: + n = r.question_index + if r.error: + print(f" Q{n:>3}/{total} [{r.response_time_seconds:5.1f}s] " + f"{_R}error: {r.error[:100]}{_X}", flush=True) + return + if r.gt_context_count == 0: + print(f" Q{n:>3}/{total} [{r.response_time_seconds:5.1f}s] " + f"{_Y}no GT contexts — skipped{_X}", flush=True) + return + if r.retrieved_chunk_count == 0: + print(f" Q{n:>3}/{total} [{r.response_time_seconds:5.1f}s] " + f"{_Y}no text chunks (structural-only answer) — recall=0{_X}", flush=True) + return + scope = f"top-{k}" if r.retrieved_chunk_count <= k else f"all-{r.retrieved_chunk_count}" + frac = f"({r.matched_count}/{r.gt_context_count} GT chunks in {scope})" + print( + f" Q{n:>3}/{total} [{r.response_time_seconds:5.1f}s] " + f"Recall@{k}={_recall_cell(r)} {frac}", + flush=True, + ) + + +def _print_detailed(r: RecallResult, total: int, k: int) -> None: + n = r.question_index + q_short = (r.question[:70] + "…") if len(r.question) > 71 else r.question + print(f" {'─' * 70}", flush=True) + print(f" {_B}Q{n}/{total}{_X} [{r.response_time_seconds:.1f}s] {q_short}", flush=True) + if r.error: + print(f" {_R}error: {r.error}{_X}", flush=True) + return + if r.gt_context_count == 0: + print(f" {_Y}no ground-truth contexts — skipped{_X}", flush=True) + return + print( + f" {_C}Recall@{k}{_X} = {_recall_cell(r)} " + f"matched {r.matched_count}/{r.gt_context_count} " + f"(retrieved {r.retrieved_chunk_count} chunks)", + flush=True, + ) + if r.unmatched_contexts: + strategy = "embedding" # default; score list is non-negative only for embedding + print(f" {_Y}Unmatched GT chunks:{_X}", flush=True) + # Build a map: gt_text → embed score (for embedding runs only) + score_map: Dict[str, float] = {} + if r.embed_scores and any(s >= 0 for s in r.embed_scores): + # embed_scores is aligned with gt_contexts; reconstruct unmatched scores + for idx, (score, _matched) in enumerate( + zip(r.embed_scores, [i in r.matched_indices for i in range(len(r.embed_scores))]) + ): + pass + # Simpler: zip embed_scores with all GT chunks, show score for unmatched + all_gt = r.unmatched_contexts # only unmatched are in this list + unmatched_scores = [s for i, s in enumerate(r.embed_scores) if i not in r.matched_indices] + for uc, sc in zip(all_gt, unmatched_scores if unmatched_scores else [None]*len(all_gt)): + score_str = f" max_cos={sc:.3f}" if sc is not None and sc >= 0 else "" + print(f" - {uc[:100]}{'…' if len(uc) > 100 else ''}{score_str}", flush=True) + else: + for uc in r.unmatched_contexts: + print(f" - {uc[:120]}{'…' if len(uc) > 120 else ''}", flush=True) + + +# ─── Main eval loop ─────────────────────────────────────────────────────────── + +def _parse_mode(mode_str: str) -> Tuple[str, str]: + """Parse --mode into (api_mode, rag_pattern). + + Agentic styles (mode=agentic, rag_pattern=