What is CrewAI?
CrewAI is an open-source framework that orchestrates multiple AI agents to work together on complex tasks. Think of it as a project manager for LLMs ā you define agents with specific roles, goals, and tools, then assign them tasks that they execute collaboratively. Each agent can use different language models, search tools, or custom functions, and CrewAI handles the handoffs, context sharing, and sequential or parallel execution automatically.
At its core, CrewAI gives you:
- Role-based agents ā define who does what with clear role descriptions and backstories
- Task orchestration ā chain tasks together with dependencies and context passing
- Tool integration ā equip agents with web search, code execution, file reading, and custom tools
- Process management ā run tasks sequentially or in parallel with built-in validation
Why Build a Code Review Agent?
Manual code review is essential but time-consuming. Senior developers spend hours scanning pull requests for bugs, style violations, security issues, and logical errors. A Code Review Agent built with CrewAI can automate the first pass ā catching common issues, enforcing style guides, flagging potential vulnerabilities, and even suggesting improvements ā all before a human ever looks at the code.
Key benefits of an automated code review agent:
- Speed ā review hundreds of files in seconds, not hours
- Consistency ā apply the same rigorous standards to every commit
- Scalability ā handle large monorepos and frequent PRs without fatigue
- Augmented feedback ā combine static analysis, style checking, and LLM-powered semantic review in one pipeline
- Developer experience ā give immediate feedback to contributors, reducing back-and-forth
By using CrewAI, you can split the review workload across multiple specialized agents ā one for security analysis, one for style enforcement, one for logic review ā and have them collaborate to produce a single, comprehensive review report.
Setting Up Your Environment
First, install CrewAI and its dependencies. You'll need Python 3.10+ and an API key for an LLM provider (OpenAI, Anthropic, or a local model via Ollama).
# Create a virtual environment
python -m venv crewai-env
source crewai-env/bin/activate # On Windows: crewai-env\Scripts\activate
# Install CrewAI and dependencies
pip install crewai crewai-tools
# Optional: install for local model support
pip install litellm
Set your API key as an environment variable:
export OPENAI_API_KEY="sk-your-key-here"
# Or for Anthropic:
export ANTHROPIC_API_KEY="your-key-here"
Now create a project structure:
code_review_agent/
āāā main.py # Entry point
āāā agents.py # Agent definitions
āāā tasks.py # Task definitions
āāā tools.py # Custom tools (file reader, git diff parser)
āāā config/
āāā review_rules.yaml # Configurable review rules
Building the Code Review Agent ā Step by Step
Step 1: Define Your Agents
Agents are the heart of CrewAI. Each agent has a role, goal, backstory, and optional tools. For a code review crew, you might create three specialized agents: a Style Checker, a Security Auditor, and a Logic Reviewer. Each one focuses on a different aspect of the code.
Create agents.py:
from crewai import Agent
from crewai_tools import tool
from tools import read_file_tool, analyze_ast_tool
# Agent 1: Style Checker
style_checker = Agent(
role="Senior Style Enforcement Specialist",
goal="Enforce coding style guidelines, naming conventions, and formatting consistency across all code files",
backstory=(
"You are a meticulous code stylist with 15 years of experience maintaining "
"large codebases. You know PEP 8, Airbnb style guide, and Google Java style "
"guide by heart. You notice inconsistent indentation, poor variable names, "
"and formatting issues instantly. You provide clear, actionable feedback."
),
tools=[read_file_tool],
allow_delegation=False,
verbose=True,
llm="gpt-4o"
)
# Agent 2: Security Auditor
security_auditor = Agent(
role="Application Security Auditor",
goal="Identify security vulnerabilities including injection risks, exposed secrets, unsafe dependencies, and authentication flaws",
backstory=(
"You are a white-hat security researcher with deep expertise in OWASP Top 10, "
"CWE classifications, and real-world exploit patterns. You've audited hundreds "
"of production applications and can spot subtle security issues that automated "
"scanners miss. You explain each vulnerability with severity ratings and "
"remediation steps."
),
tools=[read_file_tool, analyze_ast_tool],
allow_delegation=True,
verbose=True,
llm="gpt-4o"
)
# Agent 3: Logic & Architecture Reviewer
logic_reviewer = Agent(
role="Principal Software Architect",
goal="Review code logic, architecture patterns, error handling, performance implications, and test coverage",
backstory=(
"You are a principal architect who has designed systems serving millions of users. "
"You review code for logical correctness, edge case handling, race conditions, "
"memory leaks, and architectural fit. You think about how the code interacts with "
"the broader system and whether the design patterns are appropriate."
),
tools=[read_file_tool],
allow_delegation=True,
verbose=True,
llm="gpt-4o"
)
Step 2: Create Custom Tools
Agents need tools to read files and analyze code. CrewAI supports @tool decorators for custom functions. Create tools.py:
from crewai_tools import tool
import os
import ast
import re
@tool("Read Code File")
def read_file_tool(file_path: str) -> str:
"""
Read the contents of a code file given its absolute or relative path.
Returns the full file contents as a string with line numbers prepended.
"""
# Resolve path
resolved_path = os.path.abspath(file_path)
if not os.path.exists(resolved_path):
return f"ERROR: File not found at {resolved_path}"
with open(resolved_path, 'r', encoding='utf-8', errors='replace') as f:
lines = f.readlines()
# Add line numbers for easy reference
numbered_lines = [f"{i+1:4d}: {line}" for i, line in enumerate(lines)]
return "".join(numbered_lines)
@tool("Analyze Python AST")
def analyze_ast_tool(file_path: str) -> str:
"""
Parse a Python file into its Abstract Syntax Tree and return structural information.
Identifies function definitions, class definitions, imports, decorators,
and potential structural issues.
"""
resolved_path = os.path.abspath(file_path)
if not os.path.exists(resolved_path):
return f"ERROR: File not found at {resolved_path}"
with open(resolved_path, 'r', encoding='utf-8') as f:
source = f.read()
try:
tree = ast.parse(source)
except SyntaxError as e:
return f"SYNTAX ERROR: {str(e)}"
report_lines = [f"AST Analysis for: {file_path}", "=" * 50]
functions = [node for node in ast.walk(tree) if isinstance(node, ast.FunctionDef)]
classes = [node for node in ast.walk(tree) if isinstance(node, ast.ClassDef)]
imports = [node for node in ast.walk(tree) if isinstance(node, (ast.Import, ast.ImportFrom))]
report_lines.append(f"Total functions: {len(functions)}")
for func in functions:
args = [a.arg for a in func.args.args]
report_lines.append(f" - def {func.name}({', '.join(args)}) at line {func.lineno}")
report_lines.append(f"\nTotal classes: {len(classes)}")
for cls in classes:
methods = [n for n in ast.walk(cls) if isinstance(n, ast.FunctionDef)]
report_lines.append(f" - class {cls.name} at line {cls.lineno} ({len(methods)} methods)")
report_lines.append(f"\nTotal imports: {len(imports)}")
# Check for potentially dangerous imports
dangerous_patterns = ['pickle', 'eval', 'exec', 'subprocess', 'os.system']
for imp in imports:
names = []
if isinstance(imp, ast.Import):
names = [alias.name for alias in imp.names]
else:
names = [imp.module] if imp.module else []
for name in names:
if any(d in name for d in dangerous_patterns):
report_lines.append(f" ā ļø POTENTIALLY DANGEROUS IMPORT: {name}")
return "\n".join(report_lines)
@tool("Find Pattern in Code")
def grep_pattern_tool(params: str) -> str:
"""
Search for a regex pattern across code files.
Input format: 'file_path::pattern'
Example: 'src/main.py::TODO|FIXME|HACK'
Returns matching lines with context.
"""
parts = params.split("::")
if len(parts) != 2:
return "ERROR: Use format 'file_path::pattern'"
file_path, pattern = parts
resolved_path = os.path.abspath(file_path)
if not os.path.exists(resolved_path):
return f"ERROR: File not found at {resolved_path}"
with open(resolved_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
matches = []
for i, line in enumerate(lines):
if re.search(pattern, line):
# Add context: 2 lines before and after
start = max(0, i - 2)
end = min(len(lines), i + 3)
context_block = "".join([
f"{j+1:4d}: {lines[j]}" for j in range(start, end)
])
matches.append(f"Match at line {i+1}:\n{context_block}")
if not matches:
return f"No matches found for pattern '{pattern}' in {file_path}"
return "\n---\n".join(matches[:20]) # Limit to 20 matches
Step 3: Define Review Tasks
Tasks specify what each agent should do, in what order, and how they should produce output. Tasks can depend on each other and pass context forward. Create tasks.py:
from crewai import Task
from agents import style_checker, security_auditor, logic_reviewer
# Task 1: Style Review (runs first, independently)
style_review_task = Task(
description=(
"Review the code file at {file_path} for style violations.\n"
"Check for:\n"
"- Consistent indentation (4 spaces for Python)\n"
"- Naming conventions (snake_case for variables/functions, PascalCase for classes)\n"
"- Line length (max 120 characters)\n"
"- Trailing whitespace\n"
"- Missing or excessive blank lines\n"
"- Comment quality and placement\n"
"- Import ordering (standard library ā third-party ā local)\n"
"Provide a structured report with: file path, line numbers, severity (ERROR/WARN/INFO), "
"and specific suggestions for each issue found."
),
agent=style_checker,
expected_output="A detailed style review report with line-numbered issues and fix recommendations",
output_file="review_output/style_report.md"
)
# Task 2: Security Audit (runs in parallel with style review)
security_audit_task = Task(
description=(
"Perform a thorough security audit of the code file at {file_path}.\n"
"Look for:\n"
"- Hardcoded secrets (API keys, passwords, tokens)\n"
"- SQL injection vulnerabilities (string formatting in queries)\n"
"- Command injection risks (os.system, subprocess with shell=True)\n"
"- Unsafe deserialization (pickle, yaml.unsafe_load)\n"
"- Path traversal vulnerabilities\n"
"- Missing input validation\n"
"- Use of dangerous functions (eval, exec)\n"
"- Insecure cryptographic practices (MD5, SHA1 for passwords)\n"
"Rate each finding as CRITICAL, HIGH, MEDIUM, or LOW severity. "
"Include OWASP references and concrete fix examples."
),
agent=security_auditor,
expected_output="A security audit report with severity ratings, CWE references, and remediation code examples",
output_file="review_output/security_report.md"
)
# Task 3: Logic & Architecture Review (runs after style and security tasks complete)
logic_review_task = Task(
description=(
"Review the code file at {file_path} for logical correctness and architectural quality.\n"
"Consider:\n"
"- Correctness of algorithms and data structures\n"
"- Edge case handling (null/empty inputs, boundary values)\n"
"- Error handling and exception management\n"
"- Race conditions in concurrent code\n"
"- Resource management (file handles, connections, memory)\n"
"- Performance implications (O(n²) loops, unnecessary allocations)\n"
"- Test coverage and testability\n"
"- Adherence to SOLID principles\n"
"- Appropriate use of design patterns\n"
"Incorporate findings from the style and security reviews already completed. "
"Provide a comprehensive assessment with an overall score (1-10) and actionable recommendations."
),
agent=logic_reviewer,
context=[style_review_task, security_audit_task], # Waits for these to complete
expected_output="A comprehensive architecture and logic review report with overall score and prioritized recommendations",
output_file="review_output/logic_report.md"
)
# Task 4: Aggregated Final Report (runs after all reviews)
final_report_task = Task(
description=(
"Compile all previous review reports into a single, consolidated code review summary.\n"
"The final report should include:\n"
"- Executive summary with overall assessment\n"
"- Categorized issue list (Style, Security, Logic/Architecture)\n"
"- Top 5 most critical issues that MUST be fixed before merge\n"
"- A checklist for the developer to verify fixes\n"
"- Merge readiness verdict: APPROVED / CHANGES REQUESTED / BLOCKED\n\n"
"Style report: {style_report}\n"
"Security report: {security_report}\n"
"Logic report: {logic_report}"
),
agent=logic_reviewer, # Reuse the architect for final synthesis
context=[style_review_task, security_audit_task, logic_review_task],
expected_output="A single consolidated code review report ready for developer consumption",
output_file="review_output/FINAL_REVIEW.md"
)
Step 4: Assemble and Run the Crew
Now wire everything together in main.py. The Crew object orchestrates agents and tasks with a defined process flow.
from crewai import Crew, Process
from agents import style_checker, security_auditor, logic_reviewer
from tasks import style_review_task, security_audit_task, logic_review_task, final_report_task
import os
import sys
def run_code_review(file_path: str):
"""
Run the full code review crew on a single file.
Args:
file_path: Path to the code file to review
"""
# Validate file exists
if not os.path.exists(file_path):
print(f"ā Error: File '{file_path}' not found.")
sys.exit(1)
# Ensure output directory exists
os.makedirs("review_output", exist_ok=True)
print(f"š Starting Code Review Crew for: {file_path}")
print("=" * 60)
# Assemble the crew
review_crew = Crew(
agents=[style_checker, security_auditor, logic_reviewer],
tasks=[style_review_task, security_audit_task, logic_review_task, final_report_task],
process=Process.sequential, # Tasks execute in defined order
verbose=True,
memory=True, # Agents remember context across tasks
planning=True, # Enable planning step before execution
manager_llm="gpt-4o" # LLM used for orchestration decisions
)
# Kick off the review
result = review_crew.kickoff(inputs={"file_path": file_path})
print("\n" + "=" * 60)
print("ā
Code Review Complete!")
print(f"š Reports saved in: review_output/")
print(f" - style_report.md")
print(f" - security_report.md")
print(f" - logic_report.md")
print(f" - FINAL_REVIEW.md (consolidated)")
return result
if __name__ == "__main__":
# Accept file path from command line or use default
target_file = sys.argv[1] if len(sys.argv) > 1 else "src/app.py"
run_code_review(target_file)
Complete Working Example: Review Configuration
For production use, you'll want configurable review rules. Create config/review_rules.yaml:
# Code Review Rules Configuration
# Adjust these to match your team's standards
style:
indent_size: 4
max_line_length: 120
naming_convention: "snake_case"
require_docstrings: true
max_function_lines: 80
max_file_lines: 500
import_order:
- "standard_library"
- "third_party"
- "local"
security:
forbidden_imports:
- "pickle"
- "eval"
- "exec"
require_input_validation: true
check_hardcoded_secrets: true
secret_patterns:
- "api[_-]?key\\s*="
- "password\\s*="
- "secret\\s*="
- "token\\s*="
max_complexity_warning: 15 # Cyclomatic complexity threshold
logic:
require_error_handling: true
check_null_safety: true
check_race_conditions: true
performance:
warn_nested_loops: true
warn_large_allocations: true
solid_principles_check: true
Load these rules in your agents to make reviews adaptable:
# Add to agents.py
import yaml
def load_review_rules():
with open("config/review_rules.yaml", "r") as f:
return yaml.safe_load(f)
rules = load_review_rules()
# Update agent backstories with rule specifics
style_checker.backstory += (
f"\n\nYou enforce these specific rules: {rules['style']}"
)
security_auditor.backstory += (
f"\n\nYou specifically check for: {rules['security']}"
)
Running Multiple Files in Parallel
For reviewing entire PRs with multiple changed files, you can scale the crew to handle batches:
# batch_review.py
from crewai import Crew, Process
from agents import style_checker, security_auditor, logic_reviewer
from tasks import style_review_task, security_audit_task, logic_review_task
import os
import concurrent.futures
def review_single_file(file_path: str):
"""Review one file with its own crew instance."""
crew = Crew(
agents=[style_checker, security_auditor, logic_reviewer],
tasks=[style_review_task, security_audit_task, logic_review_task],
process=Process.sequential,
verbose=False
)
return crew.kickoff(inputs={"file_path": file_path})
def review_pull_request(pr_files: list[str]):
"""
Review multiple files from a pull request concurrently.
Each file gets its own independent review crew.
"""
results = {}
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
future_to_file = {
executor.submit(review_single_file, f): f for f in pr_files
}
for future in concurrent.futures.as_completed(future_to_file):
file_path = future_to_file[future]
try:
results[file_path] = future.result()
print(f"ā
Completed review: {file_path}")
except Exception as e:
print(f"ā Failed to review {file_path}: {e}")
# Generate aggregate PR report
print("\nš Pull Request Review Summary")
print(f"Files reviewed: {len(results)}/{len(pr_files)}")
return results
# Example usage
changed_files = [
"src/auth.py",
"src/database.py",
"src/api/handlers.py",
"tests/test_auth.py"
]
review_pull_request(changed_files)
Best Practices for Code Review Agents
1. Keep Agent Roles Focused and Narrow
Each agent should have one clear responsibility. A "Style Checker" should not also audit security ā that creates confusion in the output. Narrow agents produce more consistent, reliable results. If you need more coverage, add more specialized agents rather than expanding existing ones.
2. Use Context Wisely
CrewAI's context parameter on tasks is powerful. When a logic reviewer receives the output of style and security reviews, it can avoid duplicate findings and build on previous work. Always chain tasks that benefit from previous results. But don't over-chain ā some tasks (like style and security reviews) can run in parallel for faster execution.
3. Provide Detailed Task Descriptions
LLMs perform better with specific, checklist-style task descriptions. Instead of "Review the code for issues," write out exactly what to look for with bullet points. Include examples of good vs. bad patterns in the description. The more specific you are, the more consistent the agent's output will be.
4. Structure Output with Clear Schemas
Use expected_output and output_file to enforce structured results. For even more control, define a JSON schema or Markdown template in the task description:
expected_output=(
"A JSON object with keys: 'file_path', 'issues' (array of objects with "
"'line', 'severity', 'category', 'description', 'suggestion'), "
"'summary' (string), and 'score' (integer 1-10)"
)
5. Implement a Human-in-the-Loop Step
For critical code paths, add a verification task that requires human approval before the final report is generated. CrewAI supports human_input=True on tasks:
verification_task = Task(
description="Review the auto-generated findings and confirm or override each one",
agent=logic_reviewer,
human_input=True, # Pauses execution for human input
expected_output="Verified and adjusted review findings"
)
6. Cache and Reuse Results
If you're reviewing the same files repeatedly (e.g., during iterative PR updates), cache agent outputs keyed by file hash. This saves API costs and speeds up re-reviews:
import hashlib, json, os
def get_file_hash(file_path: str) -> str:
with open(file_path, 'rb') as f:
return hashlib.sha256(f.read()).hexdigest()
def cached_review(file_path: str):
cache_key = get_file_hash(file_path)
cache_file = f".cache/{cache_key}.json"
if os.path.exists(cache_file):
with open(cache_file, 'r') as f:
return json.load(f)
result = review_single_file(file_path)
os.makedirs(".cache", exist_ok=True)
with open(cache_file, 'w') as f:
json.dump(result, f)
return result
7. Monitor and Calibrate Agent Performance
Track false positives and false negatives from your agents. Keep a log of overridden findings and use that data to refine task descriptions and agent backstories. Over time, you can tune the agents to match your team's specific standards and reduce noise.
8. Handle Large Files Efficiently
For files exceeding ~2000 lines, chunk the content and review in sections. LLMs have context windows ā feeding a 5000-line file may cause truncation or degraded analysis:
def chunk_file(file_path: str, chunk_size: int = 500) -> list[dict]:
"""Split a large file into overlapping chunks for review."""
with open(file_path, 'r') as f:
lines = f.readlines()
chunks = []
for i in range(0, len(lines), chunk_size - 50): # 50-line overlap
chunk_lines = lines[i:i + chunk_size]
chunks.append({
"start_line": i + 1,
"end_line": i + len(chunk_lines),
"content": "".join(chunk_lines)
})
return chunks
Integrating with CI/CD Pipelines
A Code Review Agent becomes truly powerful when integrated into your CI/CD workflow. Here's a GitHub Actions example that runs the review on every pull request:
# .github/workflows/code_review.yml
name: AI Code Review
on:
pull_request:
types: [opened, synchronize, reopened]
jobs:
ai-review:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0 # Fetch all history for diff
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install crewai crewai-tools pyyaml
- name: Get changed files
id: changed-files
run: |
CHANGED=$(git diff --name-only ${{ github.event.pull_request.base.sha }} ${{ github.sha }} | grep -E '\.(py|js|ts|java|go|rb)$' || true)
echo "files=$CHANGED" >> $GITHUB_OUTPUT
- name: Run AI Code Review
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
python batch_review.py ${{ steps.changed-files.outputs.files }}
- name: Post review as PR comment
uses: actions/github-script@v7
with:
script: |
const fs = require('fs');
const report = fs.readFileSync('review_output/FINAL_REVIEW.md', 'utf8');
await github.rest.issues.createComment({
issue_number: context.issue.number,
owner: context.repo.owner,
repo: context.repo.repo,
body: report
});
Conclusion
Building a Code Review Agent with CrewAI transforms code review from a purely human bottleneck into an augmented, AI-assisted workflow. By decomposing the review process into specialized agents ā style, security, logic ā you get thorough, consistent, and fast first-pass reviews that catch issues before they reach production. The framework's task orchestration, context sharing, and tool integration make it straightforward to build a sophisticated review pipeline that mirrors how experienced engineering teams actually work.
Start with the three-agent setup described here, calibrate the rules to your team's standards, integrate it into your CI pipeline, and iterate based on the feedback from your developers. The result is a code review process that's faster, more thorough, and ultimately more enjoyable for everyone involved.