| Crates.io | kowalski-code-agent |
| lib.rs | kowalski-code-agent |
| version | 0.5.0 |
| created_at | 2025-06-28 21:55:52.166954+00 |
| updated_at | 2025-06-28 21:55:52.166954+00 |
| description | Kowalski Code Agent: A Rust-based agent for interacting with Ollama models |
| homepage | https://github.com/yarenty/kowalski |
| repository | https://github.com/yarenty/kowalski |
| max_upload_size | |
| id | 1730156 |
| size | 192,452 |
A specialized AI agent for code analysis, refactoring, and documentation, built on the Kowalski framework. The Code Agent provides intelligent, language-aware code analysis and improvement suggestions for multiple programming languages.
The Code Agent is an AI-powered assistant that combines large language models with language-specific static analysis tools. It helps developers analyze, refactor, and document code in languages like Java, Python, and Rust, providing actionable insights and recommendations.
use kowalski_code_agent::CodeAgent;
use kowalski_core::config::Config;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let config = Config::default();
let mut code_agent = CodeAgent::new(config).await?;
// Analyze Python code
let python_code = "...";
let analysis = code_agent.analyze_python(python_code).await?;
println!("Suggestions: {:?}", analysis.suggestions);
// Analyze Rust code
let rust_code = "...";
let analysis = code_agent.analyze_rust(rust_code).await?;
println!("Rust Issues: {:?}", analysis.issues);
Ok(())
}
Each tool provides:
Metrics Analysis: Lines, characters, functions, classes, complexity
Language-Specific Checks: Syntax, style, best practices
Quality Suggestions: Specific recommendations for improvement
Error Detection: Syntax errors, potential issues, anti-patterns
Proper Tool Trait Implementation: Full integration with the Kowalski framework
Analyzes a DataProcessor struct with caching functionality
Demonstrates Rust-specific safety and error handling analysis
Shows ownership, borrowing, and memory safety considerations
Running `/opt/ml/kowalski/target/debug/examples/java_analysis`
☕ Starting Java Code Analysis... Code Agent Conversation ID: 67677c04-2d1a-49e8-8b6d-8c18f7f180d9
📝 Java Code to Analyze:
import java.util.*;
public class Calculator { private int result;
public Calculator() {
this.result = 0;
}
public int add(int a, int b) {
result = a + b;
return result;
}
public int subtract(int a, int b) {
result = a - b;
return result;
}
public int multiply(int a, int b) {
result = a * b;
return result;
}
public double divide(int a, int b) {
if (b == 0) {
System.out.println("Error: Division by zero");
return 0;
}
result = a / b;
return (double) result;
}
public static void main(String[] args) {
Calculator calc = new Calculator();
System.out.println("Addition: " + calc.add(10, 5));
System.out.println("Subtraction: " + calc.subtract(10, 5));
System.out.println("Multiplication: " + calc.multiply(10, 5));
System.out.println("Division: " + calc.divide(10, 5));
}
}
📊 Java Analysis Results: Language: java Metrics: { "characters": 1008, "classes": 1, "comments": 0, "complexity": { "cyclomatic_complexity": 2, "for_loops": 0, "if_statements": 1, "level": "Low", "switch_statements": 0, "while_loops": 0 }, "imports": 1, "lines": 42, "methods": 8, "words": 122 } Suggestions: ["Consider using a proper logging framework instead of System.out.println", "Main method found - ensure proper exception handling"] Issues: []
🤖 AI Analysis: Code Analysis and Recommendations
The provided Java code implements a basic calculator class with methods for addition, subtraction, multiplication, and division. The analysis highlights several areas for improvement.
result and a, b are concise, they do not follow standard Java naming conventions (e.g., using camelCase instead of underscore notation). Consider renaming them to calculateResult and num1, respectively.double or int). To avoid ambiguity, use a more explicit return type, such as double in the case of division by zero handling.Here's an updated version of the code incorporating these recommendations:
import java.util.InputMismatchException;
import java.util.Scanner;
public class Calculator {
private double result;
public Calculator() {}
public int add(int a, int b) {
return (int) calculateResult(a, b);
}
public int subtract(int a, int b) {
return (int) calculateResult(a, -b);
}
public int multiply(int a, int b) {
return (int) calculateResult(a, 1 * b); // Use implicit multiplication
}
private double calculateResult(int num1, int num2) {
if (num2 == 0) {
throw new ArithmeticException("Division by zero");
}
return (double) num1 / num2;
}
public static void main(String[] args) {
Scanner scanner = new Scanner(System.in);
System.out.println("Enter numbers:");
int num1 = getValidNumber(scanner, "First number: ");
int num2 = getValidNumber(scanner, "Second number: ");
Calculator calc = new Calculator();
System.out.println("Addition: " + calc.add(num1, num2));
System.out.println("Subtraction: " + calc.subtract(num1, num2));
System.out.println("Multiplication: " + calc.multiply(num1, num2));
}
private static int getValidNumber(Scanner scanner, String prompt) {
while (true) {
try {
System.out.print(prompt);
return scanner.nextInt();
} catch (InputMismatchException e) {
System.out.println("Invalid input. Please enter a number.");
scanner.next(); // Consume invalid input
}
}
}
}
This refactored code includes:
Scanner to ensure numbers are entered correctlyThese changes enhance the overall quality of the code, making it more robust, maintainable, and efficient. ✅ Analysis complete!
🔍 Follow-up Analysis: Based on the provided Java code, here are some specific improvements that can be made:
Calculator class has multiple methods (addition, subtraction, multiplication, division) that perform similar operations. Consider extracting each operation into its own separate method or even better, create a new class Operation with different implementations for each operation.public enum Operation {
ADDITION,
SUBTRACTION,
Multiplication,
DIVISION
}
// In the Calculator class
private Operation operation;
public void setOperation(Operation operation) {
this.operation = operation;
}
// Then in the main method
Calculator calc = new Calculator();
calc.setOperation(Operation.ADDITION);
System.out.println("Addition: " + calc.calculate(10, 5));
switch statement for handling different operations: The current code uses if-else statements to handle each operation separately. This can be improved by using a switch statement which is more efficient and concise.public int calculate(int num1, int num2) {
switch (operation) {
case ADDITION:
return num1 + num2;
case SUBTRACTION:
return num1 - num2;
case Multiplication:
return num1 * num2;
case DIVISION:
if (num2 == 0) {
throw new ArithmeticException("Division by zero");
}
return (int) num1 / num2;
}
}
Double.NaN) to indicate division by zero.public int calculate(int num1, int num2) {
if (num2 == 0) {
return Double.NaN;
}
return (int) num1 / num2;
}
Use a consistent naming convention: The code uses both camelCase and underscore notation for variable names. Choose one convention throughout the codebase.
Add input validation: Currently, the code does not validate user input. Consider adding checks to ensure that users enter valid numbers.
public static void main(String[] args) {
Scanner scanner = new Scanner(System.in);
System.out.println("Enter two numbers:");
int num1 = getValidNumber(scanner, "First number: ");
int num2 = getValidNumber(scanner, "Second number: ");
Calculator calc = new Calculator();
System.out.println("Addition: " + calc.calculate(num1, num2));
}
InputMismatchException, consider using a more general exception handler (e.g., try-catch block with multiple catches).public static void main(String[] args) {
try {
Scanner scanner = new Scanner(System.in);
System.out.println("Enter two numbers:");
int num1 = scanner.nextInt();
int num2 = scanner.nextInt();
Calculator calc = new Calculator();
System.out.println("Addition: " + calc.calculate(num1, num2));
} catch (Exception e) {
System.err.println("Error: " + e.getMessage());
}
}
These improvements can enhance the code's maintainability, readability, and robustness.
Running `/opt/ml/kowalski/target/debug/examples/python_analysis`
🐍 Starting Python Code Analysis... Code Agent Conversation ID: 073f9369-9381-4eff-8305-5e7b28451b48
📝 Python Code to Analyze:
import os
import sys
from typing import List, Optional
class DataProcessor:
def __init__(self, data: List[int]):
self.data = data
self.result = 0
def calculate_sum(self) -> int:
"""Calculate the sum of all data points."""
total = 0
for item in self.data:
total += item
return total
def calculate_average(self) -> float:
"""Calculate the average of all data points."""
if len(self.data) == 0:
print("Error: No data to calculate average")
return 0.0
return self.calculate_sum() / len(self.data)
def find_max(self) -> Optional[int]:
"""Find the maximum value in the data."""
if not self.data:
return None
max_val = self.data[0]
for item in self.data:
if item > max_val:
max_val = item
return max_val
def main():
# Sample data
numbers = [10, 20, 30, 40, 50]
# Create processor
processor = DataProcessor(numbers)
# Calculate statistics
print(f"Sum: {processor.calculate_sum()}")
print(f"Average: {processor.calculate_average()}")
print(f"Maximum: {processor.find_max()}")
# Process empty data
empty_processor = DataProcessor([])
print(f"Empty average: {empty_processor.calculate_average()}")
if __name__ == "__main__":
main()
📊 Python Analysis Results:
Language: python
Metrics: {
"characters": 1407,
"classes": 1,
"comments": 4,
"complexity": {
"cyclomatic_complexity": 7,
"for_loops": 2,
"if_statements": 4,
"level": "Medium",
"try_blocks": 0,
"while_loops": 0
},
"functions": 5,
"imports": 4,
"lines": 52,
"words": 145
}
Suggestions: ["Consider using logging instead of print statements"]
PEP 8 Issues: ["Line 10: Trailing whitespace", "Line 17: Trailing whitespace", "Line 24: Trailing whitespace", "Line 38: Trailing whitespace", "Line 41: Trailing whitespace", "Line 46: Trailing whitespace"]
🤖 AI Analysis: Code Analysis and Recommendations
The provided Python code defines a DataProcessor class with methods to calculate the sum, average, and maximum of a list of integers. The code also includes a main function for testing.
The code is generally well-structured and readable. However, there are some areas that can be improved:
find_max method prints an error message if the data is empty. Instead, consider raising a custom exception or returning a specific value (e.g., None) to indicate an error condition.logging) instead, which provides more flexibility and control over log output.The code generally adheres to PEP 8 guidelines. However, there are a few instances of trailing whitespace that need attention:
total += itemmax_val = self.data[0]if len(self.data) == 0: (trailing whitespace on the same line as the condition)return self.calculate_sum() / len(self.data)print(f"Maximum: {processor.find_max()}") (missing closing parenthesis)return max_valTo fix these issues, remove the trailing whitespace and ensure that each statement or block of code is on a new line.
Based on the analysis, here are some suggestions:
print statements with logging calls to improve log output control and flexibility.DataProcessor class to reduce code duplication and make it easier to add new functionality.Refactored Code
Here's an updated version of the code incorporating some of these suggestions:
import logging
logging.basicConfig(level=logging.INFO)
class DataProcessor:
def __init__(self, data: list[int]):
"""
Initialize the data processor with sample input data.
Args:
input_data (list): List of numbers to process.
"""
self.input_data = data
self.result = 0
def calculate_sum(self) -> int:
"""
Calculate the sum of all data points.
Returns:
int: The calculated sum.
"""
total = 0
for item in self.input_data:
total += item
return total
def calculate_average(self) -> float:
"""
Calculate the average of all data points.
Raises:
ValueError: If input data is empty.
Returns:
float: The calculated average.
"""
if not self.input_data:
raise ValueError("Cannot calculate average with empty data")
return self.calculate_sum() / len(self.input_data)
def find_max(self) -> Optional[int]:
"""
Find the maximum value in all data points.
Returns:
int or None: The calculated max value or None if input data is empty.
"""
if not self.input_data:
return None
max_val = self.input_data[0]
for item in self.input_data:
if item > max_val:
max_val = item
return max_val
def main():
sample_numbers = [10, 20, 30, 40, 50]
processor = DataProcessor(sample_numbers)
logging.info(f"Sum: {processor.calculate_sum()}")
logging.info(f"Average: {processor.calculate_average()}")
logging.info(f"Maximum: {processor.find_max()}")
empty_processor = DataProcessor([])
try:
logging.info(f"Empty average: {empty_processor.calculate_average()}")
except ValueError as e:
logging.error(e)
if __name__ == "__main__":
main()
Changes and Improvements
print statements with logging calls for more flexible log output control.find_max method.ValueError) instead of printing an error message in the calculate_average method.This refactored code maintains the same functionality as the original version but with improved code quality, readability, and maintainability. ✅ Analysis complete!
🔍 Follow-up Analysis: Here are some suggestions on how to improve the provided Python code to better follow PEP 8 guidelines:
1. Use Meaningful Variable Names
Variable names like data, result, and numbers should be more descriptive. Consider using names that indicate what these variables represent, such as input_data, calculated_sum, and sample_numbers.
input_data = [10, 20, 30, 40, 50]
2. Follow PEP 8 Line Length
PEP 8 recommends keeping lines under 79 characters. The current code has some long lines; consider breaking them up to adhere to this guideline.
# Current line: 88 characters
long_line = (
"for item in self.data:"
" total += item"
)
# Refactored line (under 79 characters)
for item in self.data:
total += item
3. Indentation and Spacing
PEP 8 requires consistent indentation (4 spaces) and spacing between statements.
# Current code has inconsistent indentation and spacing
if len(self.data):
print("Error: Data is empty")
else:
# Code here
# Refactored code with consistent indentation and spacing
if len(self.data):
logging.error("Data is empty")
else:
# Code here
4. Use Comments and Docstrings
Comments should explain why the code is doing something, not what it's doing. Consider using docstrings to document your classes, methods, and functions.
# Current comment: "This is a sample data"
# Refactored comment: "Sample input data for testing purposes"
class DataProcessor:
"""
A class to calculate statistics from input data.
Attributes:
input_data (list): List of numbers to process.
Methods:
calculate_sum(): Calculate the sum of input data.
calculate_average(): Calculate the average of input data.
find_max(): Find the maximum value in input data.
"""
5. Remove Redundant if __name__ == "__main__":
PEP 8 advises against using this construct when running tests or other scripts.
# Current code has redundant if statement
if __name__ == "__main__":
main()
6. Consider Using Type Hints
Type hints can make your code more readable and self-documenting.
def calculate_sum(self) -> int:
"""Calculate the sum of all data points."""
total = 0
for item in self.data:
total += item
return total
# Current function definition: no type hint
7. Fix PEP 8 Violations
The code has several PEP 8 violations, including:
# Original line with trailing whitespace
print("Error: Data is empty")
# Refactored line without trailing whitespace
if len(self.data):
logging.error("Data is empty")
else:
# Code here
Improved Code
Here's the refactored code incorporating these improvements:
import logging
logging.basicConfig(level=logging.INFO)
class DataProcessor:
def __init__(self, input_data: list[int]):
"""
Initialize the data processor with sample input data.
Args:
input_data (list): List of numbers to process.
"""
self.input_data = input_data
self.result = 0
def calculate_sum(self) -> int:
"""
Calculate the sum of all data points.
Returns:
int: The calculated sum.
"""
total = 0
for item in self.input_data:
total += item
return total
def calculate_average(self) -> float:
"""
Calculate the average of all data points.
Raises:
ValueError: If input data is empty.
Returns:
float: The calculated average.
"""
if not self.input_data:
raise ValueError("Data is empty")
return self.calculate_sum() / len(self.input_data)
def find_max(self) -> Optional[int]:
"""
Find the maximum value in all data points.
Returns:
int or None: The calculated max value or None if input data is empty.
"""
if not self.input_data:
return None
max_val = self.input_data[0]
for item in self.input_data:
if item > max_val:
max_val = item
return max_val
def main():
sample_numbers = [10, 20, 30, 40, 50]
processor = DataProcessor(sample_numbers)
logging.info(f"Sum: {processor.calculate_sum()}")
logging.info(f"Average: {processor.calculate_average()}")
logging.info(f"Maximum: {processor.find_max()}")
empty_processor = DataProcessor([])
try:
logging.info(f"Empty average: {empty_processor.calculate_average()}")
except ValueError as e:
logging.error(e)
if __name__ == "__main__":
main()
This refactored code adheres to most PEP 8 guidelines, making it more readable and maintainable.
Running `/opt/ml/kowalski/target/debug/examples/rust_analysis`
🦀 Starting Rust Code Analysis... Code Agent Conversation ID: 33142869-16b7-4cad-bcc2-8b981abf82a3
📝 Rust Code to Analyze:
use std::collections::HashMap;
use std::error::Error;
#[derive(Debug)]
struct DataProcessor {
data: Vec<i32>,
cache: HashMap<String, i32>,
}
impl DataProcessor {
fn new(data: Vec<i32>) -> Self {
Self {
data,
cache: HashMap::new(),
}
}
fn calculate_sum(&self) -> i32 {
self.data.iter().sum()
}
fn calculate_average(&self) -> Option<f64> {
if self.data.is_empty() {
None
} else {
Some(self.calculate_sum() as f64 / self.data.len() as f64)
}
}
fn find_max(&self) -> Option<&i32> {
self.data.iter().max()
}
fn process_with_cache(&mut self, key: String) -> Result<i32, Box<dyn Error>> {
if let Some(&cached_value) = self.cache.get(&key) {
return Ok(cached_value);
}
let result = self.calculate_sum();
self.cache.insert(key, result);
Ok(result)
}
}
fn main() {
let numbers = vec![10, 20, 30, 40, 50];
let mut processor = DataProcessor::new(numbers);
println!("Sum: {}", processor.calculate_sum());
match processor.calculate_average() {
Some(avg) => println!("Average: {}", avg),
None => println!("No data to calculate average"),
}
match processor.find_max() {
Some(max) => println!("Maximum: {}", max),
None => println!("No data to find maximum"),
}
match processor.process_with_cache("sum".to_string()) {
Ok(result) => println!("Cached result: {}", result),
Err(e) => println!("Error: {}", e),
}
}
📊 Rust Analysis Results: Language: rust Metrics: { "characters": 1634, "comments": 0, "complexity": { "cyclomatic_complexity": 6, "for_loops": 0, "if_statements": 2, "let_statements": 4, "level": "Medium", "match_statements": 3, "while_loops": 0 }, "enums": 0, "functions": 6, "lines": 66, "modules": 0, "structs": 1, "traits": 0, "words": 160 } Suggestions: ["Consider using a proper logging framework instead of println!"] Rust Issues: ["Line 5: Possible missing semicolon", "Line 7: Possible missing semicolon", "Line 8: Possible missing semicolon", "Line 14: Possible missing semicolon", "Line 15: Possible missing semicolon", "Line 20: Possible missing semicolon", "Line 25: Possible missing semicolon", "Line 27: Possible missing semicolon", "Line 32: Possible missing semicolon", "Line 42: Possible missing semicolon", "Line 53: Possible missing semicolon", "Line 54: Possible missing semicolon", "Line 58: Possible missing semicolon", "Line 59: Possible missing semicolon", "Line 63: Possible missing semicolon", "Line 64: Possible missing semicolon"]
🤖 AI Analysis: Code Analysis Report
Overall Assessment
The provided Rust code is well-structured and easy to read. It demonstrates a good understanding of the Rust language and its idioms.
Insights and Suggestions
Logging: Instead of using println! for logging, consider using a proper logging framework like log or env_logger. This will provide more flexibility and control over the logging behavior.
Semicolons: There are several lines where semicolons are missing. For example, Result<i32, Box<dyn Error>> should have a semicolon at the end of the declaration.
Functionality: The code is well-organized and easy to follow. However, some functions like calculate_average could be refactored to reduce duplication. Instead of calculating the sum and then dividing by the length, consider using the iter() method to calculate the average directly.
Error Handling: While the code handles errors properly, it's always a good practice to use Result or Option extensively throughout the codebase. This will help catch errors earlier and make the code more robust.
Type Inference: Rust is known for its type inference capabilities. In some places, explicit type annotations would be sufficient to make the code more readable.
Naming Conventions: The function names are clear and concise. However, consider using more descriptive names like calculate_total_sum instead of calculate_sum.
Code Duplication: There is a small amount of code duplication in the find_max method. Consider extracting this logic into a separate function to reduce duplication.
Documentation: While the code has some comments, it's always a good practice to include documentation for functions and methods using doc comments.
Recommendations
calculate_average to use the iter() method.calculate_total_sum.find_max into a separate function.println!.Updated Code
use std::collections::{HashMap, HashSet};
use std::error::Error;
use std::fmt;
// Define a struct to represent a cache entry.
struct CacheEntry {
key: String,
value: i32,
}
impl CacheEntry {
fn new(key: &str) -> Self {
Self {
key: key.to_string(),
value: 0,
}
}
fn get(&self, cache: &mut HashMap<String, CacheEntry>) -> Option<i32> {
if let Some(entry) = cache.get(&self.key) {
return Some(entry.value);
}
None
}
fn update(&mut self, value: i32) {
self.value = value;
}
}
impl fmt::Display for CacheEntry {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "({} : {})", self.key, self.value)
}
}
#[derive(Debug)]
struct DataProcessor {
data: Vec<i32>,
cache: HashMap<String, i32>,
}
impl DataProcessor {
fn new(data: Vec<i32>) -> Self {
Self {
data,
cache: HashMap::new(),
}
}
/// Calculate the total sum of all elements in the data.
fn calculate_total_sum(&self) -> Result<i32, Box<dyn Error>> {
let mut total = 0;
for value in self.data.into_iter() {
total += value;
}
Ok(total)
}
/// Calculate the average of all elements in the data if it's not empty.
fn calculate_average(&self) -> Result<f64, Box<dyn Error>> {
let sum = self.calculate_total_sum()?;
if sum.is_null() || self.data.is_empty() {
return Err(Box::new(std::error::Error::msg("No data to calculate average")));
}
Ok(sum as f64 / self.data.len() as f64)
}
/// Find the maximum element in the data.
fn find_max(&self) -> Result<Option<&i32>, Box<dyn Error>> {
let max = self
.data
.into_iter()
.max()
.ok_or(Box::new(std::error::Error::msg("No data to find maximum")));
match max {
Some(value) => Ok(Some(value)),
None => Err(Box::new(std::error::Error::msg("No data to find maximum"))),
}
}
/// Process a key with caching. Returns an error if the result is not found in the cache.
fn process_with_cache(&mut self, key: &str) -> Result<i32, Box<dyn Error>> {
let mut cache = HashMap::new();
for (k, v) in &self.cache {
cache.insert(k.to_string(), v.clone());
}
if let Some(entry) = cache.get_mut(key) {
entry.update(self.calculate_total_sum()?);
} else {
let total = self.calculate_total_sum()?;
cache.insert(key.to_string(), total);
}
Ok(cache.get(key)?.clone())
}
}
fn main() {
let numbers = vec![10, 20, 30, 40, 50];
let mut processor = DataProcessor::new(numbers);
println!("Sum: {}", processor.calculate_total_sum()?);
match processor.calculate_average() {
Ok(avg) => println!("Average: {:?}", avg),
Err(e) => eprintln!("Error: {}", e),
}
match processor.find_max() {
Ok(Some(max)) => println!("Maximum: {:?}", max),
Err(e) => eprintln!("Error: {}", e),
}
match processor.process_with_cache("sum") {
Ok(result) => println!("Cached result: {}", result),
Err(e) => eprintln!("Error: {}", e),
}
}