ai_essay_evaluator.evaluator package

class ai_essay_evaluator.evaluator.AsyncLogger(enabled=False, log_directory='logs')

Bases: object

Async wrapper for logging that allows concurrent log operations

close()

Close the log handler

get_log_file() → str | None

Return the log file path if logging is enabled

async log(level, message, module=None, exc_info=False)

Asynchronously log a message

ai_essay_evaluator.evaluator.analyze_cost(usage_list, model: str = 'gpt-4o-mini')

Analyze and compute the total cost based on cumulative usage details.

Each usage object is expected to have:

• “prompt_tokens”: total prompt tokens used,

• “completion_tokens”: total output tokens,

• “prompt_tokens_details”: an object containing “cached_tokens”.

Args:

usage_list: List of usage objects from OpenAI API responses model: The OpenAI model name for pricing lookup (default: “gpt-4o-mini”)

Returns:

Dictionary with token counts and computed costs including: - total_cached_tokens: Total cached prompt tokens - total_prompt_tokens: Total prompt tokens (cached + uncached) - total_output_tokens: Total completion/output tokens - total_uncached_tokens: Total uncached prompt tokens - cost_uncached: Cost of uncached tokens - cost_cached: Cost of cached tokens - cost_output: Cost of output tokens - total_cost: Total estimated cost - model: Model name used for pricing

async ai_essay_evaluator.evaluator.call_openai_parse(messages: list[dict[str, str]], model: str, client: AsyncOpenAI, scoring_format: str, async_logger=None, student_id: str = 'Unknown')

ai_essay_evaluator.evaluator.merge_csv_files(file_paths, output_path, scoring_format, calculate_totals=True)

Merge multiple CSV files from different passes while preserving pass information. Uses the total_score from each pass and calculates a new merged total.

Args:

file_paths: List of Paths to CSV files to merge output_path: Path to save the merged output scoring_format: The scoring format (extended, item-specific, short) calculate_totals: Whether to calculate total scores

async ai_essay_evaluator.evaluator.process_csv(input_file, export_folder, file_name, scoring_format, openai_project, api_key, ai_model, log, cost_analysis, passes, merge_results, story_folder, rubric_folder, question_file, start_time, show_progress=True, calculate_totals=True)

ai_essay_evaluator.evaluator.read_text_files(folder: Path) → dict[str, str]

ai_essay_evaluator.evaluator.save_results(df, output_path, calculate_totals=True)

ai_essay_evaluator.evaluator.validate_csv(df: DataFrame) → None

Submodules

ai_essay_evaluator.evaluator.async_logger module

class ai_essay_evaluator.evaluator.async_logger.AsyncLogger(enabled=False, log_directory='logs')

Bases: object

Async wrapper for logging that allows concurrent log operations

close()

Close the log handler

get_log_file() → str | None

Return the log file path if logging is enabled

async log(level, message, module=None, exc_info=False)

Asynchronously log a message

ai_essay_evaluator.evaluator.cli module

ai_essay_evaluator.evaluator.cli.grader(project_folder: Path = <typer.models.OptionInfo object>, input_file: Path = <typer.models.OptionInfo object>, export_folder: Path = <typer.models.OptionInfo object>, export_file_name: str = <typer.models.OptionInfo object>, scoring_format: str = <typer.models.OptionInfo object>, story_folder: Path = <typer.models.OptionInfo object>, rubric_folder: Path = <typer.models.OptionInfo object>, question_file: Path = <typer.models.OptionInfo object>, api_key: str = <typer.models.OptionInfo object>, openai_project: str = <typer.models.OptionInfo object>, ai_model: str = <typer.models.OptionInfo object>, log: bool = <typer.models.OptionInfo object>, cost_analysis: bool = <typer.models.OptionInfo object>, passes: int = <typer.models.OptionInfo object>, merge_results: bool = <typer.models.OptionInfo object>, show_progress: bool = <typer.models.OptionInfo object>, calculate_totals: bool = <typer.models.OptionInfo object>)

ai_essay_evaluator.evaluator.cost_analysis module

class ai_essay_evaluator.evaluator.cost_analysis.ModelPricing(input_price: float, cached_input_price: float, output_price: float)

Bases: object

Pricing structure for OpenAI models per 1M tokens.

cached_input_price: float

input_price: float

output_price: float

ai_essay_evaluator.evaluator.cost_analysis.analyze_cost(usage_list, model: str = 'gpt-4o-mini')

Analyze and compute the total cost based on cumulative usage details.

Each usage object is expected to have:

• “prompt_tokens”: total prompt tokens used,

• “completion_tokens”: total output tokens,

• “prompt_tokens_details”: an object containing “cached_tokens”.

Args:

usage_list: List of usage objects from OpenAI API responses model: The OpenAI model name for pricing lookup (default: “gpt-4o-mini”)

Returns:

Dictionary with token counts and computed costs including: - total_cached_tokens: Total cached prompt tokens - total_prompt_tokens: Total prompt tokens (cached + uncached) - total_output_tokens: Total completion/output tokens - total_uncached_tokens: Total uncached prompt tokens - cost_uncached: Cost of uncached tokens - cost_cached: Cost of cached tokens - cost_output: Cost of output tokens - total_cost: Total estimated cost - model: Model name used for pricing

ai_essay_evaluator.evaluator.cost_analysis.get_model_pricing(model: str) → ModelPricing

Get pricing for a specific model.

Args:

model: The model name (e.g., “gpt-4o”, “gpt-4o-mini”)

Returns:

ModelPricing object with pricing information

Raises:

ValueError: If model is not found in the pricing registry

ai_essay_evaluator.evaluator.file_handler module

ai_essay_evaluator.evaluator.file_handler.merge_csv_files(file_paths, output_path, scoring_format, calculate_totals=True)

Merge multiple CSV files from different passes while preserving pass information. Uses the total_score from each pass and calculates a new merged total.

Args:

file_paths: List of Paths to CSV files to merge output_path: Path to save the merged output scoring_format: The scoring format (extended, item-specific, short) calculate_totals: Whether to calculate total scores

ai_essay_evaluator.evaluator.file_handler.save_results(df, output_path, calculate_totals=True)

ai_essay_evaluator.evaluator.openai_client module

class ai_essay_evaluator.evaluator.openai_client.ExtendedScoringResponse(*, idea_development_score: int, idea_development_feedback: str, language_conventions_score: int, language_conventions_feedback: str)

Bases: BaseModel

idea_development_feedback: str

idea_development_score: int

language_conventions_feedback: str

language_conventions_score: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ai_essay_evaluator.evaluator.openai_client.RateLimitTracker

Bases: object

class ai_essay_evaluator.evaluator.openai_client.StandardScoringResponse(*, score: int, feedback: str)

Bases: BaseModel

feedback: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

score: int

async ai_essay_evaluator.evaluator.openai_client.adaptive_rate_limit(async_logger=None)

Implement adaptive rate limiting based on time window tracking

async ai_essay_evaluator.evaluator.openai_client.call_openai_parse(messages: list[dict[str, str]], model: str, client: AsyncOpenAI, scoring_format: str, async_logger=None, student_id: str = 'Unknown')

async ai_essay_evaluator.evaluator.openai_client.extract_structured_response(response, scoring_format, async_logger=None, student_id='Unknown')

ai_essay_evaluator.evaluator.openai_client.generate_prompt(row, scoring_format, story_dict, rubric_text, question_text)

ai_essay_evaluator.evaluator.openai_client.get_default_response(scoring_format)

ai_essay_evaluator.evaluator.openai_client.parse_reset_time(reset_str: str) → int

Parses a reset time string (e.g. “1s” or “6m0s”) and returns the number of seconds.

async ai_essay_evaluator.evaluator.openai_client.process_with_openai(df, ai_model, api_key, stories, rubrics, question, scoring_format, openai_project, progress_callback=None, async_logger=None)

ai_essay_evaluator.evaluator.openai_client.simplify_prompt(messages)

Create a simpler version of the prompt to increase chances of successful processing

ai_essay_evaluator.evaluator.processor module

async ai_essay_evaluator.evaluator.processor.process_csv(input_file, export_folder, file_name, scoring_format, openai_project, api_key, ai_model, log, cost_analysis, passes, merge_results, story_folder, rubric_folder, question_file, start_time, show_progress=True, calculate_totals=True)

ai_essay_evaluator.evaluator.utils module

ai_essay_evaluator.evaluator.utils.normalize_response_text(df: DataFrame) → DataFrame

Normalize text columns in a DataFrame.

ai_essay_evaluator.evaluator.utils.normalize_text(text)

Normalize text to handle encoding issues with special characters.

ai_essay_evaluator.evaluator.utils.read_text_files(folder: Path) → dict[str, str]

ai_essay_evaluator.evaluator.utils.validate_csv(df: DataFrame) → None