/
Watashicuvu
/
agentic-tools
Обзор
Документация
Войти
/
Watashicuvu
/
agentic-tools
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/cli_agent/explorer_agent.py
440 строк
16 KB
Your Name
added come roles
28 май 2026, 15:16
28 май 2026, 15:16
745363b
Код
Авторство
О чём код?
"""Explorer Agent — обзор кодовой базы и поиск существующих реализаций. Использует: - MCP tools (search.code_implementations, search.code_pattern, search.cross_project, heuristics.*) - Fallback на grep_search когда MCP недоступен Выдаёт: Markdown report с architectural insights, existing implementations, duplication risks. Usage: explorer = ExplorerAgent( smart_client=client, project_root="/path/to/project", mcp_url="http://localhost:8000", # Optional ) report = await explorer.explore("Payment validation logic") """ import json import logging from dataclasses import dataclass, field from datetime import datetime, timezone 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.search_ops import grep_search from src.cli_agent.llm_streaming_client import LLMStreamingClient, StreamEvent logger = logging.getLogger(__name__) console = Console() # ============================================================================ # Models # ============================================================================ @dataclass class ExplorerResult: """Результат работы Explorer.""" query: str report: str success: bool files_explored: int = 0 patterns_searched: int = 0 implementations_found: int = 0 errors: List[str] = field(default_factory=list) search_tools_used: List[str] = field(default_factory=list) def to_summary(self) -> str: """Краткое резюме.""" emoji = "✅" if self.success else "⚠️" lines = [ f"{emoji} **Explorer Report**", f"", f" **Query:** {self.query}", f" **Files explored:** {self.files_explored}", f" **Patterns searched:** {self.patterns_searched}", f" **Implementations found:** {self.implementations_found}", f" **Tools used:** {', '.join(self.search_tools_used)}", ] if self.errors: lines.append(f"") lines.append(f" **Errors:**") for err in self.errors: lines.append(f" - {err}") return "\n".join(lines) # ============================================================================ # ExplorerAgent # ============================================================================ class ExplorerAgent: """Агент для обзора кодовой базы. Workflow: 1. MCP search (если доступен) → semantic + pattern search 2. Fallback на grep_search → если MCP недоступен 3. Cross-project search → если включено 4. Heuristics analysis → анализ структуры проекта 5. LLM synthesis → генерация отчёта Args: smart_client: Клиент для LLM project_root: Корень проекта mcp_url: URL MCP сервера (None если недоступен) model: Модель LLM enable_cross_project: Включить кросс-проектный поиск """ def __init__( self, smart_client: AsyncSmartOpenAI, project_root: str, mcp_url: Optional[str] = None, model: Optional[str] = None, enable_cross_project: bool = False, ): self.smart_client = smart_client self.project_root = Path(project_root) self.mcp_url = mcp_url self.model = model or "gpt-4" self.enable_cross_project = enable_cross_project # MCP bridge (ленивая инициализация) self._mcp_bridge = None self._mcp_available = False async def _init_mcp_bridge(self) -> bool: """Инициализировать MCP bridge. Returns: True если MCP доступен """ if self._mcp_bridge is not None: return self._mcp_available if not self.mcp_url: self._mcp_available = False return False try: from src.cli_agent.tools.mcp_bridge import StreamableHTTPMCPBridge 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"[Explorer] MCP connected: {result.get('count', 0)} tools") else: self._mcp_available = False logger.warning(f"[Explorer] MCP connect failed: {result.get('error')}") except Exception as e: self._mcp_available = False logger.warning(f"[Explorer] MCP initialization error: {e}") return self._mcp_available async def explore(self, query: str) -> ExplorerResult: """Выполнить обзор кодовой базы. Args: query: Запрос пользователя (что искать) Returns: ExplorerResult с отчётом """ console.print(Panel( f"🔍 **Explorer: поиск по запросу**\n\n`{query}`", title="Explorer Agent", border_style="blue" )) # Инициализация MCP mcp_available = await self._init_mcp_bridge() # Сбор данных search_results: Dict[str, Any] = {} tools_used: List[str] = [] errors: List[str] = [] if mcp_available: console.print("[cyan]📡 MCP доступен — использую search.* инструменты[/cyan]") # 1. Semantic search semantic_result = await self._mcp_semantic_search(query) if semantic_result: search_results["semantic"] = semantic_result tools_used.append("search.code_implementations") # 2. Pattern search pattern_result = await self._mcp_pattern_search(query) if pattern_result: search_results["pattern"] = pattern_result tools_used.append("search.code_pattern") # 3. Cross-project search (если включено) if self.enable_cross_project: cross_result = await self._mcp_cross_project_search(query) if cross_result: search_results["cross_project"] = cross_result tools_used.append("search.cross_project") # 4. Heuristics analysis heuristics_result = await self._mcp_get_heuristics() if heuristics_result: search_results["heuristics"] = heuristics_result tools_used.append("heuristics.list") else: console.print("[yellow]⚠️ MCP недоступен — использую grep_search fallback[/yellow]") # Fallback: grep search grep_result = await self._grep_search(query) if grep_result: search_results["grep"] = grep_result tools_used.append("grep_search") # Генерация отчёта через LLM console.print("[cyan]📝 Генерация отчёта...[/cyan]") report = await self._generate_report( query=query, search_results=search_results, tools_used=tools_used, ) # Подсчёт статистики files_explored = 0 implementations_found = 0 if "semantic" in search_results: implementations_found = search_results["semantic"].get("count", 0) if "pattern" in search_results: files_explored = search_results["pattern"].get("file_count", 0) if "grep" in search_results: files_explored = search_results["grep"].get("count", 0) success = len(tools_used) > 0 and report is not None return ExplorerResult( query=query, report=report or "❌ Failed to generate report", success=success, files_explored=files_explored, patterns_searched=len(tools_used), implementations_found=implementations_found, errors=errors, search_tools_used=tools_used, ) # ============================================================================ # MCP Tools # ============================================================================ async def _mcp_semantic_search(self, query: str) -> Optional[Dict[str, Any]]: """search.code_implementations через MCP.""" try: result = await self._mcp_bridge.call_tool( "search.code_implementations", arguments={ "query": query, "scope": "all", "limit": 5, }, ) return {"success": True, "result": result} except Exception as e: logger.warning(f"[Explorer] Semantic search failed: {e}") return None async def _mcp_pattern_search(self, query: str) -> Optional[Dict[str, Any]]: """search.code_pattern через MCP.""" try: # Извлекаем паттерны из запроса (простая эвристика) pattern = query.replace(" ", ".*")[:50] result = await self._mcp_bridge.call_tool( "search.code_pattern", arguments={ "pattern": pattern, "file_types": [".py"], "scope": "all", "max_chunks": 3, }, ) return {"success": True, "result": result} except Exception as e: logger.warning(f"[Explorer] Pattern search failed: {e}") return None async def _mcp_cross_project_search(self, query: str) -> Optional[Dict[str, Any]]: """search.cross_project через MCP.""" try: result = await self._mcp_bridge.call_tool( "search.cross_project", arguments={ "query": query, "scope": "all", "limit": 5, }, ) return {"success": True, "result": result} except Exception as e: logger.warning(f"[Explorer] Cross-project search failed: {e}") return None async def _mcp_get_heuristics(self) -> Optional[Dict[str, Any]]: """heuristics.list через MCP.""" try: result = await self._mcp_bridge.call_tool( "heuristics.list", arguments={"heuristic_type": "all"}, ) return {"success": True, "result": result} except Exception as e: logger.warning(f"[Explorer] Heuristics list failed: {e}") return None # ============================================================================ # Fallback Tools (без MCP) # ============================================================================ async def _grep_search(self, query: str) -> Optional[Dict[str, Any]]: """grep_search — локальный инструмент (без MCP).""" try: # Извлекаем паттерн из запроса pattern = query.replace(" ", ".*")[:30] result = await grep_search( pattern=pattern, path=str(self.project_root), glob="*.py", limit=20, ) if result.get("success"): return { "success": True, "matches": result.get("matches", []), "count": result.get("count", 0), } else: logger.warning(f"[Explorer] Grep search failed: {result.get('error')}") return None except Exception as e: logger.warning(f"[Explorer] Grep search exception: {e}") return None # ============================================================================ # Report Generation # ============================================================================ async def _generate_report( self, query: str, search_results: Dict[str, Any], tools_used: List[str], ) -> str: """Сгенерировать отчёт через LLM.""" system_prompt = """Ты — Explorer Agent, агент для анализа кодовой базы. Твоя задача: 1. Проанализировать результаты поиска 2. Найти существующие реализации и потенциальные дубликаты 3. Определить архитектурную зону для новой реализации 4. Выдать рекомендации по reuse vs созданию нового кода Формат отчёта: ## 📋 Explorer Report **Query:** <запрос> **Tools:** <использованные инструменты> ### 🔍 Existing Implementations <Список найденных реализаций с путями и кратким описанием> ### 📊 Architectural Analysis <Рекомендуемая архитектурная зона> <Зависимости от других модулей> ### ⚠️ Duplication Risks <Потенциальные дубликаты и риски> ### 💡 Recommendations <reuse или создать новое> <Конкретные файлы для изменения> Важно: - Будь конкретен (пути файлов, имена функций) - Отмечай дубликаты явно - Давай actionable рекомендации""" # Формируем контекст поиска search_context = self._format_search_results(search_results) user_prompt = f"""Запрос: {query} Результаты поиска: {search_context} Сгенерируй подробный отчёт по анализу.""" 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: return response.choices[0].message.content else: return str(response) except Exception as e: logger.error(f"[Explorer] Report generation failed: {e}") return f"❌ Failed to generate report: {e}" def _format_search_results(self, search_results: Dict[str, Any]) -> str: """Форматировать результаты поиска для промпта.""" parts = [] if "semantic" in search_results: result = search_results["semantic"] parts.append(f"### Semantic Search Results\n{result.get('result', 'No results')}") if "pattern" in search_results: result = search_results["pattern"] parts.append(f"### Pattern Search Results\n{result.get('result', 'No results')}") if "cross_project" in search_results: result = search_results["cross_project"] parts.append(f"### Cross-Project Search Results\n{result.get('result', 'No results')}") if "heuristics" in search_results: result = search_results["heuristics"] parts.append(f"### Project Heuristics\n{result.get('result', 'No results')}") if "grep" in search_results: result = search_results["grep"] parts.append(f"### Grep Search Results\nFound: {result.get('count', 0)} matches") for match in result.get("matches", [])[:10]: parts.append(f"- {match.get('file')}:{match.get('line')} {match.get('content', '')[:100]}") return "\n\n".join(parts) if parts else "No search results available"