Spaces:
Paused
Paused
File size: 7,690 Bytes
0d3f7cc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 | """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"}
|