View module source on GitHub

get_log_level

def get_log_level(debug: bool) -> int

Set logging level based on debug flag. debug=False: Only WARNING and ERROR logs debug=True: INFO and DEBUG logs

MaximUsageCallback

class MaximUsageCallback()

Maxim usage callback.

This class represents a usage callback.

__init__

def __init__(generation_id: str)

Initialize a usage callback.

log_success_event

def log_success_event(kwargs, response_obj, start_time, end_time)

Log a success event.

Arguments:

NameDescription
kwargsThe kwargs.
response_objThe response object.
start_timeThe start time.
end_timeThe end time.

make_handle_non_streaming_wrapper

def make_handle_non_streaming_wrapper(original_method)

Make a handle non streaming wrapper.

This function wraps the original method to capture usage.

instrument_crewai

def instrument_crewai(maxim_logger: Logger, debug: bool = False)

Patches CrewAI’s core components (Crew, Agent, Task, Flow, LLM) to add comprehensive logging and tracing.

This wrapper enhances CrewAI with:

  • Detailed operation tracing for Crew, Flow, and Task executions
  • Token usage tracking for LLM calls
  • Tool execution monitoring
  • Span-based operation tracking
  • Error handling and reporting

The patching is done by wrapping key methods like:

  • Crew.kickoff
  • Agent.execute_task
  • Task.execute_sync
  • LLM.call and _handle_non_streaming_response
  • Tool._run methods

Arguments:

NameTypeDescription
maxim_loggerLoggerA Maxim Logger instance for handling the tracing and logging operations.
debugboolIf True, show INFO and DEBUG logs. If False, show only WARNING and ERROR logs.