feat: v0.19.0 — compress_tool_result() tool result 지능형 압축 (실제 XGEN 76K→116 tok, 97.6% 절감)

- compressor 모듈 신규: JSON list/dict, HTML, Error, Text 타입별 압축
- HTTP 응답 자동 감지 — headers 제거, status+body만 보존
- JSON list: 첫 샘플 full 구조 + 나머지 identity keys only
- 4개 integration point: ToolGraph.execute, create_gateway_tools, MCPProxy, MCP server
- 실제 XGEN API 12개 fixture 기준 301K→7.2K chars (97.6%)
- zero-dependency 유지 (stdlib html.parser + re만 사용)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: SonAIengine

graph_tool_call/__init__.py

@@ -6,9 +6,11 @@
 
 __all__ = [
     "CategorySummary",
+    "CompressConfig",
     "DuplicatePair",
     "GraphAnalysisReport",
     "GraphToolkit",
+    "compress_tool_result",
     "create_gateway_tools",
     "MCPAnnotations",
     "MergeStrategy",
@@ -42,6 +44,8 @@
     "create_gateway_tools": ("graph_tool_call.langchain.gateway", "create_gateway_tools"),
     "filter_tools": ("graph_tool_call.toolkit", "filter_tools"),
     "GraphToolkit": ("graph_tool_call.toolkit", "GraphToolkit"),
+    "compress_tool_result": ("graph_tool_call.compressor", "compress_tool_result"),
+    "CompressConfig": ("graph_tool_call.compressor", "CompressConfig"),
 }
 
 

graph_tool_call/compressor/__init__.py

@@ -0,0 +1,18 @@
+"""Tool result compressor: intelligently compress large tool outputs for LLM context.
+
+Usage::
+
+    from graph_tool_call.compressor import compress_tool_result, CompressConfig
+
+    # Simple — auto-detect type, default 4000 chars
+    compressed = compress_tool_result(huge_json)
+
+    # Custom config
+    cfg = CompressConfig(max_chars=2000, max_list_items=5)
+    compressed = compress_tool_result(data, config=cfg)
+"""
+
+from graph_tool_call.compressor.base import CompressConfig
+from graph_tool_call.compressor.detector import compress_tool_result
+
+__all__ = ["CompressConfig", "compress_tool_result"]

graph_tool_call/compressor/base.py

@@ -0,0 +1,24 @@
+"""Base types for the compressor module."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+class CompressConfig:
+    """Compression configuration.
+
+    Attributes:
+        max_chars: Maximum output characters (~4 chars per token).
+        max_list_items: Number of sample items to keep from JSON arrays.
+        max_value_len: Maximum character length for individual JSON values.
+        max_depth: Maximum nesting depth before summarising nested structures.
+        preserve_keys: JSON keys whose values are always kept in full.
+    """
+
+    max_chars: int = 4000
+    max_list_items: int = 3
+    max_value_len: int = 80
+    max_depth: int = 2
+    preserve_keys: list[str] = field(default_factory=list)

graph_tool_call/compressor/detector.py

@@ -0,0 +1,124 @@
+"""Auto-detect content type and route to the appropriate compressor."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from graph_tool_call.compressor.base import CompressConfig
+from graph_tool_call.compressor.error_comp import (
+    compress_error_dict,
+    compress_error_text,
+    is_error_dict,
+    is_error_text,
+)
+from graph_tool_call.compressor.html_comp import compress_html, is_html
+from graph_tool_call.compressor.json_comp import compress_json_dict, compress_json_list
+from graph_tool_call.compressor.text_comp import compress_text
+
+
+def _detect_and_compress(content: Any, config: CompressConfig) -> str:
+    """Detect content type and compress accordingly."""
+    # -- Already structured data --
+    if isinstance(content, list):
+        return compress_json_list(content, config)
+
+    if isinstance(content, dict):
+        if is_error_dict(content):
+            return compress_error_dict(content, config)
+        return compress_json_dict(content, config)
+
+    # -- String content: try to parse / classify --
+    if not isinstance(content, str):
+        content = str(content)
+
+    # Short enough — no compression needed.
+    if len(content) <= config.max_chars:
+        return content
+
+    # Try JSON parse.
+    try:
+        parsed = json.loads(content)
+    except (json.JSONDecodeError, ValueError):
+        parsed = None
+
+    if parsed is not None:
+        if isinstance(parsed, list):
+            return compress_json_list(parsed, config)
+        if isinstance(parsed, dict):
+            if is_error_dict(parsed):
+                return compress_error_dict(parsed, config)
+            return compress_json_dict(parsed, config)
+
+    # HTML detection.
+    if is_html(content):
+        return compress_html(content, config)
+
+    # Error text detection.
+    if is_error_text(content):
+        return compress_error_text(content, config)
+
+    # Fallback: plain text.
+    return compress_text(content, config)
+
+
+def compress_tool_result(
+    content: str | dict | list | Any,
+    *,
+    config: CompressConfig | None = None,
+    max_chars: int = 4000,
+    content_type: str | None = None,
+) -> str:
+    """Intelligently compress a tool result for LLM context.
+
+    Parameters:
+        content: The tool result — str, dict, list, or anything with ``__str__``.
+        config: Compression configuration.  When *None*, a default
+            ``CompressConfig(max_chars=max_chars)`` is created.
+        max_chars: Shorthand for ``CompressConfig(max_chars=...)``.
+            Ignored when *config* is provided.
+        content_type: Force a specific compressor instead of auto-detecting.
+            One of ``"json"``, ``"html"``, ``"error"``, ``"text"``.
+
+    Returns:
+        The compressed string.  If *content* is already short enough it is
+        returned as-is (for strings) or serialised (for dicts/lists).
+    """
+    if config is None:
+        config = CompressConfig(max_chars=max_chars)
+
+    # Forced content type — skip auto-detection.
+    if content_type is not None:
+        return _compress_by_type(content, content_type, config)
+
+    return _detect_and_compress(content, config)
+
+
+def _compress_by_type(content: Any, content_type: str, config: CompressConfig) -> str:
+    """Route to a specific compressor by name."""
+    if isinstance(content, str):
+        text = content
+    else:
+        text = json.dumps(content, ensure_ascii=False, default=str)
+
+    if content_type == "json":
+        try:
+            parsed = json.loads(text) if isinstance(content, str) else content
+        except (json.JSONDecodeError, ValueError):
+            return compress_text(text, config)
+        if isinstance(parsed, list):
+            return compress_json_list(parsed, config)
+        if isinstance(parsed, dict):
+            return compress_json_dict(parsed, config)
+        return compress_text(text, config)
+
+    if content_type == "html":
+        return compress_html(text, config)
+
+    if content_type == "error":
+        if isinstance(content, dict):
+            return compress_error_dict(content, config)
+        return compress_error_text(text, config)
+
+    # "text" or unknown
+    return compress_text(text, config)

graph_tool_call/compressor/error_comp.py

@@ -0,0 +1,86 @@
+"""Error response compressor: extract status + message only."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from graph_tool_call.compressor.base import CompressConfig
+
+# Keys that typically carry the error message, checked in priority order.
+_MESSAGE_KEYS = ("message", "detail", "error", "reason", "error_description", "msg")
+
+# Keys that carry nested error detail dicts.
+_DETAIL_CONTAINER_KEYS = ("body", "response", "data", "error")
+
+
+def _extract_message(data: dict[str, Any]) -> str | None:
+    """Recursively look for an error message string."""
+    for key in _MESSAGE_KEYS:
+        val = data.get(key)
+        if isinstance(val, str) and val:
+            return val
+
+    # Check one level deeper in container keys.
+    for key in _DETAIL_CONTAINER_KEYS:
+        nested = data.get(key)
+        if isinstance(nested, dict):
+            msg = _extract_message(nested)
+            if msg:
+                return msg
+    return None
+
+
+def is_error_dict(data: dict[str, Any]) -> bool:
+    """Heuristic: does *data* look like an error response?"""
+    status = data.get("status") or data.get("status_code") or data.get("statusCode")
+    if isinstance(status, int) and 400 <= status < 600:
+        return True
+    if "error" in data or "traceback" in data or "stack_trace" in data or "exception" in data:
+        return True
+    return False
+
+
+def compress_error_dict(data: dict[str, Any], config: CompressConfig) -> str:
+    """Compress an error-shaped dict to ``HTTP {status}: {message}``."""
+    status = data.get("status") or data.get("status_code") or data.get("statusCode") or "?"
+
+    # Prefer the most specific nested message over generic top-level "error".
+    message = None
+    for key in _DETAIL_CONTAINER_KEYS:
+        nested = data.get(key)
+        if isinstance(nested, dict):
+            message = _extract_message(nested)
+            if message:
+                break
+    if not message:
+        message = _extract_message(data) or "Unknown error"
+    if isinstance(message, dict):
+        message = str(message)
+
+    result = f"HTTP {status}: {message}"
+    return result[: config.max_chars]
+
+
+def compress_error_text(text: str, config: CompressConfig) -> str:
+    """Compress an error-like text string (e.g. tracebacks)."""
+    lines = text.strip().splitlines()
+    if not lines:
+        return text
+
+    # For Python tracebacks keep the last exception line.
+    for line in reversed(lines):
+        stripped = line.strip()
+        if stripped and not stripped.startswith("File ") and not stripped.startswith("at "):
+            return stripped[: config.max_chars]
+
+    return lines[-1][: config.max_chars]
+
+
+def is_error_text(text: str) -> bool:
+    """Heuristic: does *text* look like an error/traceback?"""
+    if re.search(r"Traceback \(most recent call", text):
+        return True
+    if re.search(r"^[A-Z]\w*(Error|Exception):", text, re.MULTILINE):
+        return True
+    return False