src/agent_server/local_tools/diff_tool.py - master - man4j/agent-server

src/agent_server/local_tools/diff_tool.py
161 строка6 KB
Vladimir
initial
18 апр 2026, 19:36
18 апр 2026, 19:36a95f341
from __future__ import annotations

from typing import Any

from agent_server.chat.types import MarkdownDiffToolResult
from .base import LocalToolModule
import re
from difflib import SequenceMatcher


_FENCE_RE = re.compile(
    r"^\s*```(?P<lang>[^\n`]*)\n(?P<body>.*?)\n```(?:\s*)$",
    re.DOTALL,
)


LOCAL_DIFF_TOOL = {
    "type": "function",
    "function": {
        "name": "render_diff",
        "description": (
            "UI-инструмент для отображения diff только между двумя версиями запроса к источнику данных. "
            "Используй только для SQL-запросов, HTTP/API-запросов, аргументов вызова data-access tool "
            "и других текстов, которые являются именно запросом к источнику данных. "
            "Не используй для пояснений, описаний шагов, выводов, markdown-обёрток, "
            "Plotly spec и любого иного сопроводительного текста. "
            "После вызова инструмента не дублируй и не пересказывай его результат в обычном тексте ответа: "
            "интерфейс покажет diff автоматически."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "old_markdown": {
                    "type": "string",
                    "description": (
                        "Старая версия только самого запроса к источнику данных, без пояснений и без сопроводительного текста"
                    ),
                },
                "new_markdown": {
                    "type": "string",
                    "description": (
                        "Новая версия только самого запроса к источнику данных, без пояснений и без сопроводительного текста"
                    ),
                },
                "title": {
                    "type": "string",
                    "description": "Короткий заголовок diff, например: Изменение SQL-запроса",
                },
            },
            "required": ["old_markdown", "new_markdown"],
        },
    },
}



def _unwrap_fenced_block_once(text: str) -> tuple[str, str | None, bool]:
    value = (text or "").strip()
    match = _FENCE_RE.match(value)
    if not match:
        return text or "", None, False

    lang = (match.group("lang") or "").strip() or None
    body = match.group("body")
    return body, lang, True


def _unwrap_all_fenced_blocks(text: str) -> tuple[str, str | None]:
    value = text or ""
    detected_lang: str | None = None

    while True:
        unwrapped, lang, changed = _unwrap_fenced_block_once(value)
        if not changed:
            break
        if lang and detected_lang is None:
            detected_lang = lang
        value = unwrapped

    return value.replace("\r\n", "\n"), detected_lang


def _split_lines(text: str) -> list[str]:
    prepared, _ = _unwrap_all_fenced_blocks(text)
    return prepared.split("\n")


def build_markdown_diff(
    old_text: str,
    new_text: str,
    *,
    title: str = "Изменения",
) -> str:
    old_lines = _split_lines(old_text)
    new_lines = _split_lines(new_text)

    matcher = SequenceMatcher(a=old_lines, b=new_lines)
    diff_lines: list[str] = []

    for tag, i1, i2, j1, j2 in matcher.get_opcodes():
        if tag == "equal":
            for line in old_lines[i1:i2]:
                diff_lines.append(f"  {line}")
        elif tag == "delete":
            for line in old_lines[i1:i2]:
                diff_lines.append(f"- {line}")
        elif tag == "insert":
            for line in new_lines[j1:j2]:
                diff_lines.append(f"+ {line}")
        elif tag == "replace":
            for line in old_lines[i1:i2]:
                diff_lines.append(f"- {line}")
            for line in new_lines[j1:j2]:
                diff_lines.append(f"+ {line}")

    if not diff_lines:
        diff_part = f"### {title}\n\n_Изменений нет_"
    else:
        diff_part = (
            f"### {title}\n\n"
            f"```diff\n"
            + "\n".join(diff_lines)
            + "\n```"
        )

    return diff_part


async def dispatch_diff_tool(
    tool_name: str,
    tool_input: dict[str, Any],
) -> MarkdownDiffToolResult | dict[str, Any] | None:
    old_markdown = tool_input.get("old_markdown")
    new_markdown = tool_input.get("new_markdown")
    title = (tool_input.get("title") or "Изменения запроса").strip()

    if not isinstance(old_markdown, str):
        return {"error": "field 'old_markdown' must be a string"}
    if not isinstance(new_markdown, str):
        return {"error": "field 'new_markdown' must be a string"}

    markdown = build_markdown_diff(
        old_text=old_markdown,
        new_text=new_markdown,
        title=title,
    )

    return {
        "ok": True,
        "ui_type": "markdown_diff",
        "markdown": markdown.rstrip() + "\n\n\u200B\n\n",
        "model_visible": False,
        "model_message": "Diff запроса к источнику данных отображён пользователю интерфейсом.",
    }


DIFF_TOOL_MODULE = LocalToolModule(
    tools=[LOCAL_DIFF_TOOL],
    dispatcher=dispatch_diff_tool,
    ui_builder=None,
)