Spaces:
Paused
Paused
| """Code analysis agent implementation.""" | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| from typing import Any | |
| from hermes.agents.base.agent import BaseAgent | |
| from hermes.core.types import AgentStrategy, ToolCall, ToolResult | |
| from hermes.tools.base.registry import tool_registry | |
| logger = logging.getLogger(__name__) | |
| class CodeAnalysisAgent(BaseAgent): | |
| """Agent specialized in code analysis and quality assessment.""" | |
| def __init__(self, llm_provider: Any | None = None) -> None: | |
| super().__init__( | |
| agent_type="code_analysis", | |
| strategy=AgentStrategy.CHAIN_OF_THOUGHT, | |
| tools=["code_analyzer", "github_repo_reader", "file_reader"], | |
| llm_provider=llm_provider, | |
| ) | |
| async def plan(self, task: str) -> list[str]: | |
| """Create code analysis plan.""" | |
| return [ | |
| f"Analyze code structure for: {task}", | |
| "Review code quality and complexity metrics", | |
| "Identify patterns and potential issues", | |
| "Check for code smells and anti-patterns", | |
| "Compile code analysis report", | |
| ] | |
| async def think(self, task: str, observations: list[str]) -> dict[str, Any]: | |
| """Reason about code analysis approach using LLM.""" | |
| observations_text = "\n".join(f"- {obs[:500]}" for obs in observations) if observations else "None yet." | |
| prompt = f"""You are a code analysis agent. Your task: {task} | |
| Available tools (CHOOSE ONE): | |
| 1. code_analyzer — Analyze code structure and quality | |
| Required: {{"action": "analyze_project", "path": "."}} | |
| Actions: analyze_file (single file), analyze_project (entire project), analyze_pr (pull request) | |
| 2. github_repo_reader — Read GitHub repository files | |
| Required: {{"action": "get_readme", "owner": "owner_name", "repo": "repo_name"}} | |
| Actions: get_readme, list_files, read_file, get_repo | |
| 3. file_reader — Read a local file | |
| Required: {{"action": "read", "path": "file/path.txt"}} | |
| Previous observations: | |
| {observations_text} | |
| What should be your NEXT action? Choose the most appropriate tool. | |
| Respond in JSON format ONLY: | |
| {{"reasoning": "why this tool", "tool": "tool_name", "arguments": {{"key": "value"}}, "done": false}} | |
| Rules: | |
| - Include ALL required arguments for the tool you choose | |
| - If you have enough information, set "done": true and "tool": "none" | |
| - Do NOT make up tool names — use ONLY the 3 tools listed above""" | |
| response = await self._call_llm([{"role": "user", "content": prompt}]) | |
| parsed = self._parse_json_response(response) | |
| if parsed and "tool" in parsed: | |
| parsed.setdefault("reasoning", "") | |
| parsed.setdefault("arguments", {}) | |
| parsed.setdefault("done", False) | |
| return parsed | |
| # Fallback: default to project analysis | |
| return { | |
| "reasoning": f"Analyzing project structure for: {task}", | |
| "tool": "code_analyzer", | |
| "arguments": {"action": "analyze_project", "path": "."}, | |
| "done": False, | |
| } | |
| async def act(self, thought: dict[str, Any]) -> ToolCall: | |
| """Execute code analysis action based on LLM decision.""" | |
| tool_name = thought.get("tool", "code_analyzer") | |
| arguments = thought.get("arguments", {}) | |
| valid_tools = ["code_analyzer", "github_repo_reader", "file_reader"] | |
| if tool_name not in valid_tools: | |
| tool_name = "code_analyzer" | |
| # Ensure required arguments for each tool | |
| if tool_name == "code_analyzer": | |
| if "action" not in arguments: | |
| arguments["action"] = "analyze_project" | |
| if "path" not in arguments: | |
| arguments["path"] = "." | |
| elif tool_name == "github_repo_reader": | |
| if "action" not in arguments: | |
| arguments["action"] = "get_readme" | |
| if "owner" not in arguments: | |
| arguments["owner"] = "" | |
| if "repo" not in arguments: | |
| arguments["repo"] = "" | |
| elif tool_name == "file_reader": | |
| if "action" not in arguments: | |
| arguments["action"] = "read" | |
| if "path" not in arguments: | |
| arguments["path"] = "README.md" | |
| return ToolCall(tool_name=tool_name, arguments=arguments) | |
| async def observe(self, result: ToolResult) -> str: | |
| """Observe code analysis results using LLM to extract key findings.""" | |
| if not hasattr(result, "success") or not result.success: | |
| return f"Tool execution failed: {result}" | |
| output = result.output if hasattr(result, "output") else str(result) | |
| output_text = json.dumps(output, default=str)[:3000] if not isinstance(output, str) else output[:3000] | |
| prompt = f"""Extract key findings from this code analysis result. | |
| Focus on: code quality, patterns, potential issues, complexity. | |
| Provide a concise summary (2-3 sentences max). | |
| Tool result: | |
| {output_text} | |
| Key findings:""" | |
| response = await self._call_llm([{"role": "user", "content": prompt}]) | |
| if response and not response.startswith("[LLM unavailable"): | |
| return response.strip() | |
| # Fallback: extract from structure | |
| if isinstance(output, dict): | |
| summary = output.get("summary", {}) | |
| if isinstance(summary, dict): | |
| files = summary.get("total_files", 0) | |
| lines = summary.get("total_lines", 0) | |
| issues = summary.get("issues", 0) | |
| return f"Code analysis: {files} files, {lines} lines, {issues} issues found" | |
| return f"Analysis result: {str(output)[:800]}" | |
| return f"Got result: {str(output)[:800]}" | |
| async def synthesize(self, task: str) -> str: | |
| """Synthesize code analysis findings using LLM.""" | |
| observations = self.state.observations | |
| if not observations: | |
| return f"Code analysis completed for: {task}. No issues found." | |
| observations_text = "\n\n".join(f"Finding {i+1}: {obs}" for i, obs in enumerate(observations[:10])) | |
| prompt = f"""You are a code analysis agent synthesizing findings for: | |
| Task: {task} | |
| Analysis findings: | |
| {observations_text} | |
| Please synthesize these into a comprehensive code analysis report: | |
| 1. Code quality assessment | |
| 2. Patterns and anti-patterns identified | |
| 3. Potential issues and technical debt | |
| 4. Recommendations for improvement | |
| 5. Complexity analysis | |
| Report:""" | |
| response = await self._call_llm([{"role": "user", "content": prompt}]) | |
| if response and not response.startswith("[LLM unavailable"): | |
| return response.strip() | |
| summary = f"Code Analysis Report for: {task}\n\n" | |
| summary += f"Analyzed {len(observations)} code aspects.\n\n" | |
| for i, obs in enumerate(observations[:5], 1): | |
| summary += f"Finding {i}: {obs[:300]}\n\n" | |
| return summary | |
| async def analyze_repo(self, owner: str, repo: str) -> dict[str, Any]: | |
| """Analyze a GitHub repository.""" | |
| tool = tool_registry.get("github_repo_reader") | |
| if tool: | |
| result = await tool.execute(action="list_files", owner=owner, repo=repo) | |
| return result if isinstance(result, dict) else {"result": str(result)} | |
| return {"error": "GitHub tool not available"} | |
| async def analyze_file(self, path: str) -> dict[str, Any]: | |
| """Analyze a local file.""" | |
| tool = tool_registry.get("code_analyzer") | |
| if tool: | |
| result = await tool.execute(action="analyze_file", path=path) | |
| return result if isinstance(result, dict) else {"result": str(result)} | |
| return {"error": "Code analyzer not available"} | |