src/cli_agent/code_explainer.py - master - Watashicuvu/agentic-tools

src/cli_agent/code_explainer.py
374 строки13 KB
Your Name
added come roles
28 май 2026, 15:16
28 май 2026, 15:16745363b
"""Code Explainer — объяснение кода на естественном языке.

Использует MCP инструменты:
- repository.chat: RAG-поиск контекста
- search.code_implementations: семантический поиск похожих реализаций
- search.code_pattern: поиск паттернов
- analyze_impact_radius: blast radius анализа

Workflow:
1. Найти целевой код (search/repository.chat)
2. Прочитать код (read_file/read_many_files)
3. Найти похожие реализации (search.code_pattern)
4. Проанализировать зависимости (analyze_impact_radius)
5. Сгенерировать объяснение через LLM

Usage:
    explainer = CodeExplainer(
        smart_client=client,
        project_root="/path/to/project",
        mcp_url="http://localhost:8000",
    )
    explanation = await explainer.explain("validate_payment function")
"""

import json
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional, Any, Dict, List

from rich.console import Console
from rich.panel import Panel

from src.services.async_smart_client import AsyncSmartOpenAI
from src.cli_agent.tools.mcp_bridge import StreamableHTTPMCPBridge
from src.cli_agent.tools.search_ops import grep_search

logger = logging.getLogger(__name__)
console = Console()


@dataclass
class ExplanationResult:
    """Результат объяснения кода."""

    query: str
    explanation: str
    success: bool
    target_file: Optional[str] = None
    target_function: Optional[str] = None
    code_snippet: Optional[str] = None
    similar_implementations: int = 0
    impact_files: int = 0
    errors: List[str] = field(default_factory=list)

    def to_summary(self) -> str:
        """Краткое резюме."""
        emoji = "✅" if self.success else "⚠️"
        lines = [
            f"{emoji} **Code Explanation**",
            f"",
            f"  **Query:** {self.query}",
        ]
        if self.target_file:
            lines.append(f"  **File:** `{self.target_file}`")
        if self.target_function:
            lines.append(f"  **Function:** `{self.target_function}`")
        lines.append(f"  **Similar implementations:** {self.similar_implementations}")
        lines.append(f"  **Impact radius:** {self.impact_files} files")

        if self.errors:
            lines.append(f"")
            lines.append(f"  **Errors:**")
            for err in self.errors:
                lines.append(f"    - {err}")
        return "\n".join(lines)


class CodeExplainer:
    """Агент для объяснения кода на естественном языке.

    Args:
        smart_client: Клиент для LLM
        project_root: Корень проекта
        mcp_url: URL MCP сервера (None если недоступен)
        model: Модель LLM
    """

    _mcp_bridge: Optional[StreamableHTTPMCPBridge]
    _mcp_available: bool

    def __init__(
        self,
        smart_client: AsyncSmartOpenAI,
        project_root: str,
        mcp_url: Optional[str] = None,
        model: Optional[str] = None,
    ):
        self.smart_client = smart_client
        self.project_root = Path(project_root)
        self.mcp_url = mcp_url
        self.model = model or "gpt-4"

        self._mcp_bridge = None
        self._mcp_available = False

    async def _init_mcp_bridge(self) -> bool:
        """Инициализировать MCP bridge."""
        if self._mcp_bridge is not None:
            return self._mcp_available

        if not self.mcp_url:
            self._mcp_available = False
            return False

        try:
            self._mcp_bridge = StreamableHTTPMCPBridge(
                base_url=self.mcp_url,
                timeout=10,
            )
            result = await self._mcp_bridge.connect()

            if result.get("success"):
                self._mcp_available = True
                logger.info(f"[Explainer] MCP connected: {result.get('count', 0)} tools")
            else:
                self._mcp_available = False

        except Exception as e:
            self._mcp_available = False
            logger.warning(f"[Explainer] MCP init failed: {e}")

        return self._mcp_available

    async def _call_mcp_tool(self, tool_name: str, **kwargs) -> Optional[Any]:
        """Вызвать MCP инструмент с проверкой."""
        if not self._mcp_available or self._mcp_bridge is None:
            return None
        try:
            return await self._mcp_bridge.call_tool(tool_name, **kwargs)
        except Exception as e:
            logger.warning(f"[Explainer] MCP tool {tool_name} failed: {e}")
            return None
    
    async def explain(self, query: str) -> ExplanationResult:
        """Объяснить код по запросу.
        
        Args:
            query: Запрос пользователя (что объяснить)
            
        Returns:
            ExplanationResult с объяснением
        """
        console.print(Panel(
            f"💡 **Explainer: объяснение кода**\n\n`{query}`",
            title="Code Explainer",
            border_style="green"
        ))
        
        mcp_available = await self._init_mcp_bridge()
        
        # Шаг 1: Найти целевой код
        console.print("[cyan]1/4 ⏳ Searching for target code...[/cyan]")
        target_info = await self._find_target_code(query, mcp_available)
        
        if not target_info:
            return ExplanationResult(
                query=query,
                explanation=f"❌ Не удалось найти код по запросу: {query}",
                success=False,
                errors=["Target code not found"],
            )
        
        # Шаг 2: Найти похожие реализации
        console.print("[cyan]2/4 ⏳ Searching for similar implementations...[/cyan]")
        similar = await self._find_similar_implementations(query, mcp_available)
        
        # Шаг 3: Анализ зависимостей
        console.print("[cyan]3/4 ⏳ Analyzing dependencies...[/cyan]")
        impact = await self._analyze_impact(target_info.get("file"), mcp_available)
        
        # Шаг 4: Генерация объяснения
        console.print("[cyan]4/4 ⏳ Generating explanation...[/cyan]")
        explanation = await self._generate_explanation(
            query=query,
            target_info=target_info,
            similar_implementations=similar,
            impact_analysis=impact,
        )
        
        return ExplanationResult(
            query=query,
            explanation=explanation,
            success=True,
            target_file=target_info.get("file"),
            target_function=target_info.get("function"),
            code_snippet=target_info.get("code"),
            similar_implementations=len(similar) if isinstance(similar, list) else 0,
            impact_files=len(impact) if isinstance(impact, list) else 0,
        )
    
    async def _find_target_code(
        self,
        query: str,
        mcp_available: bool,
    ) -> Optional[Dict[str, Any]]:
        """Найти целевой код для объяснения."""
        if mcp_available:
            # repository.chat для поиска
            try:
                result = await self._call_mcp_tool(
                    "repository.chat",
                    question=f"Where is {query} implemented?",
                    max_results=3,
                )
                # Парсим результат
                return {"file": "unknown", "code": result, "source": "repository.chat"}
            except Exception as e:
                logger.warning(f"[Explainer] repository.chat failed: {e}")
        
        # Fallback: grep search
        try:
            pattern = query.replace(" ", ".*")[:30]
            result = await grep_search(
                pattern=pattern,
                path=str(self.project_root),
                glob="*.py",
                limit=5,
            )
            
            if result.get("success") and result.get("matches"):
                first_match = result["matches"][0]
                return {
                    "file": first_match.get("file"),
                    "line": first_match.get("line"),
                    "code": first_match.get("content", ""),
                    "source": "grep_search",
                }
        except Exception as e:
            logger.warning(f"[Explainer] grep search failed: {e}")
        
        return None
    
    async def _find_similar_implementations(
        self,
        query: str,
        mcp_available: bool,
    ) -> List[Dict[str, Any]]:
        """Найти похожие реализации."""
        if not mcp_available:
            return []
        
        try:
            result = await self._call_mcp_tool(
                "search.code_implementations",
                query=query,
                scope="all",
                limit=3,
            )
            return [{"result": result}] if result else []
        except Exception as e:
            logger.warning(f"[Explainer] Similar search failed: {e}")
            return []
    
    async def _analyze_impact(
        self,
        file_path: Optional[str],
        mcp_available: bool,
    ) -> List[Dict[str, Any]]:
        """Анализировать зависимости (blast radius)."""
        if not mcp_available or not file_path:
            return []

        try:
            result = await self._call_mcp_tool(
                "analyze_impact_radius",
                file_path=file_path,
            )
            return [{"file": file_path, "result": result}] if result else []
        except Exception as e:
            logger.warning(f"[Explainer] Impact analysis failed: {e}")
            return []
    
    async def _generate_explanation(
        self,
        query: str,
        target_info: Dict[str, Any],
        similar_implementations: List[Dict[str, Any]],
        impact_analysis: List[Dict[str, Any]],
    ) -> str:
        """Сгенерировать объяснение через LLM."""
        
        system_prompt = """Ты — опытный разработчик-наставник. Твоя задача — объяснять код на понятном естественном языке.

Правила:
1. Объясняй простыми словами, без излишнего жаргона
2. Описывай ЧТО делает код, ЗАТЕМ КАК
3. Указывай зависимости и побочные эффекты
4. Приводи примеры использования если возможно
5. Отмечай потенциальные проблемы или улучшения

Формат:
## 💡 Explanation

**Function/Module:** <имя>
**File:** <путь>

### Что делает
<Краткое описание назначения>

### Как работает
<Пошаговое объяснение логики>

### Зависимости
<Какие модули/функции использует>

### Побочные эффекты
<Влияние на другие части системы>

### Примеры использования
<Примеры вызова если есть>

### Рекомендации
<Возможные улучшения>"""

        # Формируем контекст
        context_parts = [
            f"### Target Code ({target_info.get('file', 'unknown')}):",
            "```python",
            target_info.get("code", "Code not available"),
            "```",
            "",
        ]
        
        if similar_implementations:
            context_parts.append("### Similar Implementations Found:")
            for sim in similar_implementations[:2]:
                context_parts.append(f"- {sim.get('result', 'N/A')[:200]}...")
            context_parts.append("")
        
        if impact_analysis:
            context_parts.append("### Impact Analysis:")
            for imp in impact_analysis:
                context_parts.append(f"- File: {imp.get('file')}")
                context_parts.append(f"  Result: {str(imp.get('result', 'N/A'))[:300]}")
            context_parts.append("")
        
        user_prompt = f"""Запрос пользователя: {query}

Контекст кода:
{chr(10).join(context_parts)}

Объясни этот код подробно на естественном языке."""

        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ]
        
        try:
            response = await self.smart_client.chat_completion(
                messages=messages,
                model=self.model,
            )

            if hasattr(response, "choices") and response.choices:
                content = response.choices[0].message.content
                return content if content else "No explanation generated"
            else:
                return str(response)

        except Exception as e:
            logger.error(f"[Explainer] Explanation generation failed: {e}")
            return f"❌ Failed to generate explanation: {e}"