Spaces:
Running
Running
File size: 7,288 Bytes
8755993 | 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 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 | """
Crew workflows for multi-agent collaboration.
"""
from crewai import Crew, Task, Process
from typing import Dict, Any, Optional
from code_chatbot.agents import (
create_analyst_agent,
create_refactor_agent,
create_reviewer_agent,
create_documentation_agent
)
import logging
logger = logging.getLogger(__name__)
class RefactoringCrew:
"""
Crew for automated refactoring tasks.
Workflow:
1. Analyst examines code and identifies refactoring opportunities
2. Refactor agent implements the top refactorings
3. Reviewer checks the refactored code for correctness
"""
def __init__(self, llm=None, mcp_tools: Optional[list] = None):
"""
Initialize refactoring crew.
Args:
llm: Language model to use for agents
mcp_tools: MCP tools to provide to agents
"""
self.llm = llm
self.mcp_tools = mcp_tools or []
# Create agents
self.analyst = create_analyst_agent(llm=llm, tools=self.mcp_tools)
self.refactor = create_refactor_agent(llm=llm, tools=self.mcp_tools)
self.reviewer = create_reviewer_agent(llm=llm, tools=self.mcp_tools)
def create_crew(self, file_path: str) -> Crew:
"""
Create a crew for refactoring a specific file.
Args:
file_path: Path to file to refactor
Returns:
Configured Crew instance
"""
# Define tasks
analysis_task = Task(
description=f"""Analyze the file {file_path} and identify refactoring opportunities.
Look for:
- Long functions that should be split
- Duplicate code
- Complex conditionals
- Code smells
- Opportunities for better naming
Provide a prioritized list of the top 3-5 refactoring suggestions with rationale.""",
agent=self.analyst,
expected_output="A prioritized list of refactoring suggestions with detailed rationale"
)
refactor_task = Task(
description=f"""Based on the analysis, implement the top 3 refactorings for {file_path}.
For each refactoring:
1. Explain what you're changing and why
2. Show the before and after code
3. Ensure the refactoring is safe and doesn't break functionality
Focus on high-impact, low-risk refactorings first.""",
agent=self.refactor,
expected_output="Detailed refactoring plan with before/after code examples",
context=[analysis_task]
)
review_task = Task(
description=f"""Review the proposed refactorings for {file_path}.
Check for:
- Correctness: Do the refactorings preserve functionality?
- Quality: Do they actually improve the code?
- Safety: Are there any risks or edge cases?
- Completeness: Is anything missing?
Provide a review report with approval or requested changes.""",
agent=self.reviewer,
expected_output="Review report with approval status and any concerns",
context=[refactor_task]
)
# Create crew
crew = Crew(
agents=[self.analyst, self.refactor, self.reviewer],
tasks=[analysis_task, refactor_task, review_task],
process=Process.sequential,
verbose=True
)
return crew
def run(self, file_path: str) -> Dict[str, Any]:
"""
Run the refactoring crew on a file.
Args:
file_path: Path to file to refactor
Returns:
Crew execution result
"""
crew = self.create_crew(file_path)
result = crew.kickoff()
return {
'file_path': file_path,
'result': result,
'tasks_completed': len(crew.tasks)
}
class CodeReviewCrew:
"""
Crew for comprehensive code review.
Workflow:
1. Analyst examines code structure and patterns
2. Reviewer performs detailed code review
3. Documentation agent suggests documentation improvements
"""
def __init__(self, llm=None, mcp_tools: Optional[list] = None):
"""Initialize code review crew."""
self.llm = llm
self.mcp_tools = mcp_tools or []
self.analyst = create_analyst_agent(llm=llm, tools=self.mcp_tools)
self.reviewer = create_reviewer_agent(llm=llm, tools=self.mcp_tools)
self.documentation = create_documentation_agent(llm=llm, tools=self.mcp_tools)
def create_crew(self, file_path: str) -> Crew:
"""Create a crew for reviewing a specific file."""
analysis_task = Task(
description=f"""Analyze the structure and design of {file_path}.
Examine:
- Overall architecture and design patterns
- Code organization and modularity
- Complexity and maintainability
- Dependencies and coupling
Provide insights about the code's design quality.""",
agent=self.analyst,
expected_output="Architectural analysis with insights about design quality"
)
review_task = Task(
description=f"""Perform a detailed code review of {file_path}.
Check for:
- Bugs and potential issues
- Security vulnerabilities
- Performance problems
- Code style and best practices
- Error handling
Provide specific, actionable feedback.""",
agent=self.reviewer,
expected_output="Detailed code review with specific issues and recommendations",
context=[analysis_task]
)
documentation_task = Task(
description=f"""Review and suggest improvements for documentation in {file_path}.
Evaluate:
- Docstrings and comments
- Function/class documentation
- Code clarity and readability
- Missing documentation
Suggest specific documentation improvements.""",
agent=self.documentation,
expected_output="Documentation review with improvement suggestions",
context=[analysis_task, review_task]
)
crew = Crew(
agents=[self.analyst, self.reviewer, self.documentation],
tasks=[analysis_task, review_task, documentation_task],
process=Process.sequential,
verbose=True
)
return crew
def run(self, file_path: str) -> Dict[str, Any]:
"""Run the code review crew on a file."""
crew = self.create_crew(file_path)
result = crew.kickoff()
return {
'file_path': file_path,
'result': result,
'tasks_completed': len(crew.tasks)
}
# Export crews
__all__ = ['RefactoringCrew', 'CodeReviewCrew']
|