diff --git a/backend/.dockerignore b/backend/.dockerignore deleted file mode 100644 index 951bc993f..000000000 --- a/backend/.dockerignore +++ /dev/null @@ -1,13 +0,0 @@ -.venv/ -.ruff_cache/ -.git/ -__pycache__/ -*.pyc -*.pyo -.env -.mypy_cache/ -.pytest_cache/ -.DS_Store -*.egg-info/ -*.log -.python-version \ No newline at end of file diff --git a/backend/.gitignore b/backend/.gitignore deleted file mode 100644 index 779cc96c4..000000000 --- a/backend/.gitignore +++ /dev/null @@ -1,177 +0,0 @@ -# Byte-compiled / optimized / DLL files -__pycache__/ -*.py[cod] -*$py.class - -# C extensions -*.so - -# Distribution / packaging -.Python -build/ -develop-eggs/ -dist/ -downloads/ -eggs/ -.eggs/ -lib/ -lib64/ -parts/ -sdist/ -var/ -wheels/ -share/python-wheels/ -*.egg-info/ -.installed.cfg -*.egg -MANIFEST - -# PyInstaller -# Usually these files are written by a python script from a template -# before PyInstaller builds the exe, so as to inject date/other infos into it. -*.manifest -*.spec - -# Installer logs -pip-log.txt -pip-delete-this-directory.txt - -# Unit test / coverage reports -htmlcov/ -.tox/ -.nox/ -.coverage -.coverage.* -.cache -nosetests.xml -coverage.xml -*.cover -*.py,cover -.hypothesis/ -.pytest_cache/ -cover/ - -# Translations -*.mo -*.pot - -# Django stuff: -*.log -local_settings.py -db.sqlite3 -db.sqlite3-journal - -# Flask stuff: -instance/ -.webassets-cache - -# Scrapy stuff: -.scrapy - -# Sphinx documentation -docs/_build/ - -# PyBuilder -.pybuilder/ -target/ - -# Jupyter Notebook -.ipynb_checkpoints - -# IPython -profile_default/ -ipython_config.py - -# pyenv -# For a library or package, you might want to ignore these files since the code is -# intended to run in multiple environments; otherwise, check them in: -# .python-version - -# pipenv -# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. -# However, in case of collaboration, if having platform-specific dependencies or dependencies -# having no cross-platform support, pipenv may install dependencies that don't work, or not -# install all needed dependencies. -#Pipfile.lock - -# UV -# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -#uv.lock - -# poetry -# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control -#poetry.lock - -# pdm -# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. -#pdm.lock -# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it -# in version control. -# https://pdm.fming.dev/latest/usage/project/#working-with-version-control -.pdm.toml -.pdm-python -.pdm-build/ - -# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm -__pypackages__/ - -# Celery stuff -celerybeat-schedule -celerybeat.pid - -# SageMath parsed files -*.sage.py - -# Environments -.env -.venv -env/ -venv/ -ENV/ -env.bak/ -venv.bak/ - -# Spyder project settings -.spyderproject -.spyproject - -# Rope project settings -.ropeproject - -# mkdocs documentation -/site - -# mypy -.mypy_cache/ -.dmypy.json -dmypy.json - -# Pyre type checker -.pyre/ - -# pytype static type analyzer -.pytype/ - -# Cython debug symbols -cython_debug/ - -# PyCharm -# JetBrains specific template is maintained in a separate JetBrains.gitignore that can -# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore -# and can be added to the global gitignore or merged into this file. For a more nuclear -# option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ - -# Ruff stuff: -.ruff_cache/ - -# PyPI configuration file -.pypirc - -# VSCode -.vscode/ \ No newline at end of file diff --git a/backend/.python-version b/backend/.python-version deleted file mode 100644 index e4fba2183..000000000 --- a/backend/.python-version +++ /dev/null @@ -1 +0,0 @@ -3.12 diff --git a/backend/migrations/versions/a1b2c3d4e5f6_add_analytics_tables.py b/backend/migrations/versions/a1b2c3d4e5f6_add_analytics_tables.py new file mode 100644 index 000000000..7f19a579b --- /dev/null +++ b/backend/migrations/versions/a1b2c3d4e5f6_add_analytics_tables.py @@ -0,0 +1,224 @@ +"""Add analytics tables + +Revision ID: a1b2c3d4e5f6 +Revises: 8403bb364491 +Create Date: 2024-12-24 00:00:00.000000 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision: str = "a1b2c3d4e5f6" +down_revision: Union[str, None] = "8403bb364491" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # Create execution status enum + execution_status_enum = postgresql.ENUM( + 'pending', 'running', 'success', 'failure', 'timeout', 'cancelled', + name='executionstatus', + create_type=False + ) + execution_status_enum.create(op.get_bind(), checkfirst=True) + + # Create execution trace step type enum + trace_step_type_enum = postgresql.ENUM( + 'thought', 'action', 'observation', 'agent_invoke', 'tool_call', 'final_answer', + name='executiontracesteptype', + create_type=False + ) + trace_step_type_enum.create(op.get_bind(), checkfirst=True) + + # Create budget alert type enum + budget_alert_type_enum = postgresql.ENUM( + 'warning', 'hard_stop', + name='budgetalerttype', + create_type=False + ) + budget_alert_type_enum.create(op.get_bind(), checkfirst=True) + + # Create budget scope enum + budget_scope_enum = postgresql.ENUM( + 'user', 'agent', 'flow', 'global', + name='budgetscope', + create_type=False + ) + budget_scope_enum.create(op.get_bind(), checkfirst=True) + + # Create agentexecutions table + op.create_table( + 'agentexecutions', + sa.Column('id', sa.UUID(), nullable=False), + sa.Column('request_id', sa.UUID(), nullable=False), + sa.Column('session_id', sa.UUID(), nullable=False), + sa.Column('agent_id', sa.UUID(), nullable=True), + sa.Column('agent_type', sa.String(), nullable=True), + sa.Column('agent_name', sa.String(), nullable=True), + sa.Column('user_id', sa.UUID(), nullable=True), + sa.Column('model_config_id', sa.UUID(), nullable=True), + sa.Column('model_name', sa.String(), nullable=True), + sa.Column('started_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False), + sa.Column('completed_at', sa.DateTime(), nullable=True), + sa.Column('status', postgresql.ENUM('pending', 'running', 'success', 'failure', 'timeout', 'cancelled', name='executionstatus', create_type=False), nullable=False), + sa.Column('error_message', sa.String(), nullable=True), + sa.Column('error_type', sa.String(), nullable=True), + sa.Column('input_tokens', sa.Integer(), nullable=False, server_default='0'), + sa.Column('output_tokens', sa.Integer(), nullable=False, server_default='0'), + sa.Column('total_tokens', sa.Integer(), nullable=False, server_default='0'), + sa.Column('execution_time_ms', sa.Integer(), nullable=False, server_default='0'), + sa.Column('llm_time_ms', sa.Integer(), nullable=True, server_default='0'), + sa.Column('cost_usd', sa.Numeric(precision=10, scale=6), nullable=False, server_default='0.000000'), + sa.Column('parent_execution_id', sa.UUID(), nullable=True), + sa.Column('query_preview', sa.String(), nullable=True), + sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False), + sa.Column('updated_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False), + sa.ForeignKeyConstraint(['user_id'], ['users.id'], ondelete='SET NULL'), + sa.ForeignKeyConstraint(['model_config_id'], ['modelconfigs.id'], ondelete='SET NULL'), + sa.ForeignKeyConstraint(['parent_execution_id'], ['agentexecutions.id'], ondelete='SET NULL'), + sa.PrimaryKeyConstraint('id') + ) + op.create_index(op.f('ix_agentexecutions_id'), 'agentexecutions', ['id'], unique=False) + op.create_index(op.f('ix_agentexecutions_request_id'), 'agentexecutions', ['request_id'], unique=False) + op.create_index(op.f('ix_agentexecutions_session_id'), 'agentexecutions', ['session_id'], unique=False) + op.create_index(op.f('ix_agentexecutions_agent_id'), 'agentexecutions', ['agent_id'], unique=False) + op.create_index(op.f('ix_agentexecutions_user_id'), 'agentexecutions', ['user_id'], unique=False) + op.create_index(op.f('ix_agentexecutions_model_config_id'), 'agentexecutions', ['model_config_id'], unique=False) + op.create_index(op.f('ix_agentexecutions_parent_execution_id'), 'agentexecutions', ['parent_execution_id'], unique=False) + + # Create tokenusages table + op.create_table( + 'tokenusages', + sa.Column('id', sa.UUID(), nullable=False), + sa.Column('execution_id', sa.UUID(), nullable=False), + sa.Column('component', sa.String(), nullable=False), + sa.Column('step_number', sa.Integer(), nullable=True), + sa.Column('input_tokens', sa.Integer(), nullable=False, server_default='0'), + sa.Column('output_tokens', sa.Integer(), nullable=False, server_default='0'), + sa.Column('model', sa.String(), nullable=True), + sa.Column('cost_usd', sa.Numeric(precision=10, scale=6), nullable=False, server_default='0.000000'), + sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False), + sa.ForeignKeyConstraint(['execution_id'], ['agentexecutions.id'], ondelete='CASCADE'), + sa.PrimaryKeyConstraint('id') + ) + op.create_index(op.f('ix_tokenusages_id'), 'tokenusages', ['id'], unique=False) + op.create_index(op.f('ix_tokenusages_execution_id'), 'tokenusages', ['execution_id'], unique=False) + + # Create executiontraces table + op.create_table( + 'executiontraces', + sa.Column('id', sa.UUID(), nullable=False), + sa.Column('execution_id', sa.UUID(), nullable=False), + sa.Column('step_number', sa.Integer(), nullable=False), + sa.Column('step_type', postgresql.ENUM('thought', 'action', 'observation', 'agent_invoke', 'tool_call', 'final_answer', name='executiontracesteptype', create_type=False), nullable=False), + sa.Column('content', sa.String(), nullable=True), + sa.Column('invoked_agent_id', sa.UUID(), nullable=True), + sa.Column('invoked_agent_name', sa.String(), nullable=True), + sa.Column('timestamp', sa.DateTime(), server_default=sa.text('now()'), nullable=False), + sa.Column('duration_ms', sa.Integer(), nullable=False, server_default='0'), + sa.Column('input_tokens', sa.Integer(), nullable=True, server_default='0'), + sa.Column('output_tokens', sa.Integer(), nullable=True, server_default='0'), + sa.ForeignKeyConstraint(['execution_id'], ['agentexecutions.id'], ondelete='CASCADE'), + sa.PrimaryKeyConstraint('id') + ) + op.create_index(op.f('ix_executiontraces_id'), 'executiontraces', ['id'], unique=False) + op.create_index(op.f('ix_executiontraces_execution_id'), 'executiontraces', ['execution_id'], unique=False) + + # Create budgetalerts table + op.create_table( + 'budgetalerts', + sa.Column('id', sa.UUID(), nullable=False), + sa.Column('user_id', sa.UUID(), nullable=False), + sa.Column('scope', postgresql.ENUM('user', 'agent', 'flow', 'global', name='budgetscope', create_type=False), nullable=False), + sa.Column('scope_id', sa.UUID(), nullable=True), + sa.Column('threshold_usd', sa.Numeric(precision=10, scale=2), nullable=False), + sa.Column('period_days', sa.Integer(), nullable=False, server_default='30'), + sa.Column('alert_type', postgresql.ENUM('warning', 'hard_stop', name='budgetalerttype', create_type=False), nullable=False), + sa.Column('alert_at_percentage', sa.Integer(), nullable=False, server_default='80'), + sa.Column('webhook_url', sa.String(), nullable=True), + sa.Column('email_notification', sa.Boolean(), nullable=False, server_default='true'), + sa.Column('is_active', sa.Boolean(), nullable=False, server_default='true'), + sa.Column('last_alert_sent_at', sa.DateTime(), nullable=True), + sa.Column('current_spend_usd', sa.Numeric(precision=10, scale=2), nullable=False, server_default='0.00'), + sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False), + sa.Column('updated_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False), + sa.ForeignKeyConstraint(['user_id'], ['users.id'], ondelete='CASCADE'), + sa.PrimaryKeyConstraint('id') + ) + op.create_index(op.f('ix_budgetalerts_id'), 'budgetalerts', ['id'], unique=False) + op.create_index(op.f('ix_budgetalerts_user_id'), 'budgetalerts', ['user_id'], unique=False) + + # Create analyticssnapshots table + op.create_table( + 'analyticssnapshots', + sa.Column('id', sa.UUID(), nullable=False), + sa.Column('snapshot_date', sa.DateTime(), nullable=False), + sa.Column('period_type', sa.String(), nullable=False), + sa.Column('user_id', sa.UUID(), nullable=True), + sa.Column('agent_id', sa.UUID(), nullable=True), + sa.Column('agent_name', sa.String(), nullable=True), + sa.Column('model_name', sa.String(), nullable=True), + sa.Column('total_executions', sa.Integer(), nullable=False, server_default='0'), + sa.Column('successful_executions', sa.Integer(), nullable=False, server_default='0'), + sa.Column('failed_executions', sa.Integer(), nullable=False, server_default='0'), + sa.Column('timeout_executions', sa.Integer(), nullable=False, server_default='0'), + sa.Column('avg_execution_time_ms', sa.Integer(), nullable=False, server_default='0'), + sa.Column('p50_execution_time_ms', sa.Integer(), nullable=False, server_default='0'), + sa.Column('p95_execution_time_ms', sa.Integer(), nullable=False, server_default='0'), + sa.Column('p99_execution_time_ms', sa.Integer(), nullable=False, server_default='0'), + sa.Column('total_input_tokens', sa.Integer(), nullable=False, server_default='0'), + sa.Column('total_output_tokens', sa.Integer(), nullable=False, server_default='0'), + sa.Column('total_tokens', sa.Integer(), nullable=False, server_default='0'), + sa.Column('total_cost_usd', sa.Numeric(precision=10, scale=2), nullable=False, server_default='0.00'), + sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False), + sa.ForeignKeyConstraint(['user_id'], ['users.id'], ondelete='CASCADE'), + sa.PrimaryKeyConstraint('id') + ) + op.create_index(op.f('ix_analyticssnapshots_id'), 'analyticssnapshots', ['id'], unique=False) + op.create_index(op.f('ix_analyticssnapshots_snapshot_date'), 'analyticssnapshots', ['snapshot_date'], unique=False) + op.create_index(op.f('ix_analyticssnapshots_user_id'), 'analyticssnapshots', ['user_id'], unique=False) + op.create_index(op.f('ix_analyticssnapshots_agent_id'), 'analyticssnapshots', ['agent_id'], unique=False) + + +def downgrade() -> None: + """Downgrade schema.""" + # Drop tables + op.drop_index(op.f('ix_analyticssnapshots_agent_id'), table_name='analyticssnapshots') + op.drop_index(op.f('ix_analyticssnapshots_user_id'), table_name='analyticssnapshots') + op.drop_index(op.f('ix_analyticssnapshots_snapshot_date'), table_name='analyticssnapshots') + op.drop_index(op.f('ix_analyticssnapshots_id'), table_name='analyticssnapshots') + op.drop_table('analyticssnapshots') + + op.drop_index(op.f('ix_budgetalerts_user_id'), table_name='budgetalerts') + op.drop_index(op.f('ix_budgetalerts_id'), table_name='budgetalerts') + op.drop_table('budgetalerts') + + op.drop_index(op.f('ix_executiontraces_execution_id'), table_name='executiontraces') + op.drop_index(op.f('ix_executiontraces_id'), table_name='executiontraces') + op.drop_table('executiontraces') + + op.drop_index(op.f('ix_tokenusages_execution_id'), table_name='tokenusages') + op.drop_index(op.f('ix_tokenusages_id'), table_name='tokenusages') + op.drop_table('tokenusages') + + op.drop_index(op.f('ix_agentexecutions_parent_execution_id'), table_name='agentexecutions') + op.drop_index(op.f('ix_agentexecutions_model_config_id'), table_name='agentexecutions') + op.drop_index(op.f('ix_agentexecutions_user_id'), table_name='agentexecutions') + op.drop_index(op.f('ix_agentexecutions_agent_id'), table_name='agentexecutions') + op.drop_index(op.f('ix_agentexecutions_session_id'), table_name='agentexecutions') + op.drop_index(op.f('ix_agentexecutions_request_id'), table_name='agentexecutions') + op.drop_index(op.f('ix_agentexecutions_id'), table_name='agentexecutions') + op.drop_table('agentexecutions') + + # Drop enums + sa.Enum(name='budgetscope').drop(op.get_bind(), checkfirst=True) + sa.Enum(name='budgetalerttype').drop(op.get_bind(), checkfirst=True) + sa.Enum(name='executiontracesteptype').drop(op.get_bind(), checkfirst=True) + sa.Enum(name='executionstatus').drop(op.get_bind(), checkfirst=True) diff --git a/backend/scripts/seed_analytics_data.py b/backend/scripts/seed_analytics_data.py new file mode 100644 index 000000000..9a8947da8 --- /dev/null +++ b/backend/scripts/seed_analytics_data.py @@ -0,0 +1,392 @@ +""" +Seed script to populate analytics tables with synthetic data for testing. + +Run with: python -m scripts.seed_analytics_data +Or via docker: docker exec -it genai-backend python -m scripts.seed_analytics_data +""" + +import asyncio +import random +import uuid +from datetime import datetime, timedelta +import os + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession +from sqlalchemy.orm import sessionmaker + +# Configuration - use localhost for running outside Docker +DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://postgres:postgres@localhost:5432/postgres") + +# Sample data +AGENT_NAMES = [ + "CustomerSupportAgent", + "DataAnalysisAgent", + "CodeReviewAgent", + "ResearchAssistant", + "ContentWriterAgent", + "TranslationAgent", + "SummarizationAgent", + "QATestingAgent", +] + +AGENT_TYPES = ["genai", "a2a", "mcp"] + +MODEL_NAMES = [ + "gpt-4o", + "gpt-4o-mini", + "claude-3-5-sonnet", + "claude-3-haiku", + "gemini-1.5-pro", + "gemini-1.5-flash", +] + +# Cost per 1K tokens (input/output) +MODEL_COSTS = { + "gpt-4o": (0.005, 0.015), + "gpt-4o-mini": (0.00015, 0.0006), + "claude-3-5-sonnet": (0.003, 0.015), + "claude-3-haiku": (0.00025, 0.00125), + "gemini-1.5-pro": (0.00125, 0.005), + "gemini-1.5-flash": (0.000075, 0.0003), +} + +SAMPLE_QUERIES = [ + "What are the quarterly sales figures?", + "Summarize this document for me", + "Help me debug this Python code", + "Translate this text to Spanish", + "Write a blog post about AI trends", + "Analyze customer feedback sentiment", + "Review this pull request", + "Generate test cases for this function", + "Explain this concept in simple terms", + "Create a marketing email template", +] + +ERROR_TYPES = ["timeout", "rate_limit", "context_length", "api_error", "validation_error"] +ERROR_MESSAGES = [ + "Request timed out after 30 seconds", + "Rate limit exceeded, please retry", + "Context length exceeded maximum tokens", + "API returned unexpected error", + "Input validation failed", +] + + +def random_datetime_in_range(start: datetime, end: datetime) -> datetime: + """Generate a random datetime between start and end.""" + delta = end - start + random_seconds = random.randint(0, int(delta.total_seconds())) + return start + timedelta(seconds=random_seconds) + + +async def get_user_id(session: AsyncSession) -> uuid.UUID: + """Get a user ID from the database, or create a test user.""" + result = await session.execute(text("SELECT id FROM users LIMIT 1")) + row = result.fetchone() + if row: + return row[0] + + # Create a test user if none exists + user_id = uuid.uuid4() + await session.execute( + text(""" + INSERT INTO users (id, username, password_hash, created_at, updated_at) + VALUES (:id, :username, :password, NOW(), NOW()) + ON CONFLICT (username) DO NOTHING + """), + {"id": user_id, "username": "test_analytics_user", "password": "hashed_password"} + ) + await session.commit() + return user_id + + +async def seed_executions(session: AsyncSession, user_id: uuid.UUID, count: int = 200): + """Seed agent execution records.""" + print(f"Seeding {count} agent executions...") + + end_date = datetime.utcnow() + start_date = end_date - timedelta(days=90) + + executions = [] + for i in range(count): + execution_id = uuid.uuid4() + agent_name = random.choice(AGENT_NAMES) + agent_type = random.choice(AGENT_TYPES) + model_name = random.choice(MODEL_NAMES) + + # Determine status (80% success, 15% failure, 5% other) + status_roll = random.random() + if status_roll < 0.80: + status = "success" + error_type = None + error_message = None + elif status_roll < 0.95: + status = "failure" + error_idx = random.randint(0, len(ERROR_TYPES) - 1) + error_type = ERROR_TYPES[error_idx] + error_message = ERROR_MESSAGES[error_idx] + else: + status = random.choice(["timeout", "cancelled"]) + error_type = status + error_message = f"Execution {status}" + + # Generate realistic token counts + input_tokens = random.randint(100, 8000) + output_tokens = random.randint(50, 4000) + total_tokens = input_tokens + output_tokens + + # Calculate cost based on model + input_cost_per_1k, output_cost_per_1k = MODEL_COSTS.get(model_name, (0.001, 0.002)) + cost = (input_tokens / 1000 * input_cost_per_1k) + (output_tokens / 1000 * output_cost_per_1k) + + # Generate timestamps + started_at = random_datetime_in_range(start_date, end_date) + execution_time_ms = random.randint(500, 30000) + llm_time_ms = int(execution_time_ms * random.uniform(0.6, 0.9)) + completed_at = started_at + timedelta(milliseconds=execution_time_ms) + + executions.append({ + "id": execution_id, + "user_id": user_id, + "request_id": uuid.uuid4(), + "session_id": uuid.uuid4(), + "agent_id": uuid.uuid4(), + "agent_name": agent_name, + "agent_type": agent_type, + "model_name": model_name, + "status": status, + "query_preview": random.choice(SAMPLE_QUERIES)[:100], + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "total_tokens": total_tokens, + "cost_usd": round(cost, 6), + "execution_time_ms": execution_time_ms, + "llm_time_ms": llm_time_ms, + "started_at": started_at, + "completed_at": completed_at if status == "success" else None, + "error_type": error_type, + "error_message": error_message, + }) + + # Insert executions + for exec_data in executions: + await session.execute( + text(""" + INSERT INTO agentexecutions ( + id, user_id, request_id, session_id, agent_id, agent_name, agent_type, + model_name, status, query_preview, input_tokens, output_tokens, total_tokens, + cost_usd, execution_time_ms, llm_time_ms, started_at, completed_at, + error_type, error_message, created_at, updated_at + ) VALUES ( + :id, :user_id, :request_id, :session_id, :agent_id, :agent_name, :agent_type, + :model_name, :status, :query_preview, :input_tokens, :output_tokens, :total_tokens, + :cost_usd, :execution_time_ms, :llm_time_ms, :started_at, :completed_at, + :error_type, :error_message, NOW(), NOW() + ) + """), + exec_data + ) + + await session.commit() + print(f" Created {count} executions") + return executions + + +async def seed_token_usages(session: AsyncSession, executions: list): + """Seed token usage records for executions.""" + print("Seeding token usage records...") + + count = 0 + for exec_data in executions: + # Create 1-5 token usage records per execution + num_usages = random.randint(1, 5) + components = ["main_agent", "sub_agent", "tool_call", "retrieval", "summarizer"] + + for i in range(num_usages): + input_tokens = random.randint(50, 2000) + output_tokens = random.randint(25, 1000) + input_cost_per_1k, output_cost_per_1k = MODEL_COSTS.get(exec_data["model_name"], (0.001, 0.002)) + cost = (input_tokens / 1000 * input_cost_per_1k) + (output_tokens / 1000 * output_cost_per_1k) + + await session.execute( + text(""" + INSERT INTO tokenusages ( + id, execution_id, component, model, input_tokens, output_tokens, + cost_usd, step_number, created_at + ) VALUES ( + :id, :execution_id, :component, :model, :input_tokens, :output_tokens, + :cost_usd, :step_number, NOW() + ) + """), + { + "id": uuid.uuid4(), + "execution_id": exec_data["id"], + "component": random.choice(components), + "model": exec_data["model_name"], + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "cost_usd": round(cost, 6), + "step_number": i + 1, + } + ) + count += 1 + + await session.commit() + print(f" Created {count} token usage records") + + +async def seed_execution_traces(session: AsyncSession, executions: list): + """Seed execution trace records for ReAct loop visualization.""" + print("Seeding execution traces...") + + step_types = ["thought", "action", "observation", "tool_call", "final_answer"] + sample_contents = { + "thought": [ + "I need to analyze the user's request and determine the best approach.", + "Let me break down this problem into smaller steps.", + "I should first gather more information before proceeding.", + "Based on the context, I'll need to use a specific tool.", + ], + "action": [ + "Calling the search API with the user's query.", + "Executing database lookup for relevant records.", + "Invoking the code analysis tool.", + "Sending request to external service.", + ], + "observation": [ + "Received 5 relevant results from the search.", + "Database returned 12 matching records.", + "Code analysis completed with no critical issues.", + "External service responded with success status.", + ], + "tool_call": [ + "Tool: web_search | Args: {'query': 'latest AI trends'}", + "Tool: code_executor | Args: {'language': 'python', 'code': '...'}", + "Tool: file_reader | Args: {'path': '/data/report.csv'}", + ], + "final_answer": [ + "Based on my analysis, here are the key findings...", + "I've completed the task. The results show that...", + "Here's the summary of what I found...", + ], + } + + count = 0 + for exec_data in executions: + if exec_data["status"] != "success": + continue + + # Create 3-8 trace steps per successful execution + num_steps = random.randint(3, 8) + base_time = exec_data["started_at"] + + for i in range(num_steps): + step_type = step_types[i % len(step_types)] + if i == num_steps - 1: + step_type = "final_answer" + + duration_ms = random.randint(100, 3000) + timestamp = base_time + timedelta(milliseconds=i * 1000) + + await session.execute( + text(""" + INSERT INTO executiontraces ( + id, execution_id, step_number, step_type, content, + input_tokens, output_tokens, duration_ms, timestamp + ) VALUES ( + :id, :execution_id, :step_number, :step_type, :content, + :input_tokens, :output_tokens, :duration_ms, :timestamp + ) + """), + { + "id": uuid.uuid4(), + "execution_id": exec_data["id"], + "step_number": i + 1, + "step_type": step_type, + "content": random.choice(sample_contents.get(step_type, ["Processing..."])), + "input_tokens": random.randint(50, 500) if step_type in ["thought", "action"] else 0, + "output_tokens": random.randint(25, 300) if step_type in ["thought", "final_answer"] else 0, + "duration_ms": duration_ms, + "timestamp": timestamp, + } + ) + count += 1 + + await session.commit() + print(f" Created {count} execution traces") + + +async def seed_budget_alerts(session: AsyncSession, user_id: uuid.UUID): + """Seed budget alert records.""" + print("Seeding budget alerts...") + + alerts = [ + {"threshold": 50.00, "period_days": 7, "alert_type": "warning", "percentage": 80}, + {"threshold": 200.00, "period_days": 30, "alert_type": "warning", "percentage": 90}, + {"threshold": 500.00, "period_days": 30, "alert_type": "hard_stop", "percentage": 100}, + ] + + for alert in alerts: + await session.execute( + text(""" + INSERT INTO budgetalerts ( + id, user_id, scope, threshold_usd, period_days, alert_type, + alert_at_percentage, is_active, email_notification, current_spend_usd, created_at, updated_at + ) VALUES ( + :id, :user_id, 'user', :threshold_usd, :period_days, :alert_type, + :alert_at_percentage, true, true, 0.00, NOW(), NOW() + ) + """), + { + "id": uuid.uuid4(), + "user_id": user_id, + "threshold_usd": alert["threshold"], + "period_days": alert["period_days"], + "alert_type": alert["alert_type"], + "alert_at_percentage": alert["percentage"], + } + ) + + await session.commit() + print(f" Created {len(alerts)} budget alerts") + + +async def main(): + print("=" * 50) + print("Analytics Data Seeder") + print("=" * 50) + + engine = create_async_engine(DATABASE_URL, echo=False) + async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + + async with async_session() as session: + # Get or create user + user_id = await get_user_id(session) + print(f"Using user ID: {user_id}") + + # Clear existing analytics data + print("\nClearing existing analytics data...") + await session.execute(text("DELETE FROM executiontraces")) + await session.execute(text("DELETE FROM tokenusages")) + await session.execute(text("DELETE FROM agentexecutions")) + await session.execute(text("DELETE FROM budgetalerts")) + await session.commit() + + # Seed data + print("\nSeeding new data...") + executions = await seed_executions(session, user_id, count=250) + await seed_token_usages(session, executions) + await seed_execution_traces(session, executions) + await seed_budget_alerts(session, user_id) + + print("\n" + "=" * 50) + print("Seeding complete!") + print("=" * 50) + + await engine.dispose() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/backend/src/models/__init__.py b/backend/src/models/__init__.py index ad5ca185f..93e6a5df7 100644 --- a/backend/src/models/__init__.py +++ b/backend/src/models/__init__.py @@ -1,9 +1,11 @@ import uuid +from datetime import datetime from typing import List -from sqlalchemy import ForeignKey, UniqueConstraint +from sqlalchemy import ForeignKey, Numeric, UniqueConstraint from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.orm import Mapped, mapped_column, relationship +from typing import Optional from src.db.annotations import ( created_at, @@ -15,8 +17,16 @@ updated_at, uuid_pk, ) +from decimal import Decimal + from src.db.base import Base -from src.utils.enums import SenderType +from src.utils.enums import ( + BudgetAlertType, + BudgetScope, + ExecutionStatus, + ExecutionTraceStepType, + SenderType, +) class UserProjectAssociation(Base): @@ -414,3 +424,235 @@ class UserProfile(Base): user: Mapped["User"] = relationship(back_populates="profile", single_parent=True) # TODO: config fields, other credentials, etc + + +# ============================================================================ +# Analytics Models +# ============================================================================ + + +class AgentExecution(Base): + """Tracks individual agent execution instances for analytics.""" + + __tablename__ = "agentexecutions" + + id: Mapped[uuid_pk] + request_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), index=True, nullable=False + ) + session_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), index=True, nullable=False + ) + + # The agent that was executed (can be genai agent, mcp tool, or a2a card) + agent_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), index=True, nullable=True + ) + agent_type: Mapped[str] = mapped_column(nullable=True) # genai, mcp, a2a, flow + agent_name: Mapped[str] = mapped_column(nullable=True) + + # User who triggered the execution + user_id: Mapped[uuid.UUID] = mapped_column( + ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True + ) + + # Model configuration used + model_config_id: Mapped[uuid.UUID] = mapped_column( + ForeignKey("modelconfigs.id", ondelete="SET NULL"), nullable=True, index=True + ) + model_name: Mapped[str] = mapped_column(nullable=True) + + # Timing + started_at: Mapped[created_at] + completed_at: Mapped[Optional[datetime]] = mapped_column(nullable=True) + + # Status + status: Mapped[ExecutionStatus] = mapped_column(default=ExecutionStatus.pending) + error_message: Mapped[str] = mapped_column(nullable=True) + error_type: Mapped[str] = mapped_column(nullable=True) + + # Token usage + input_tokens: Mapped[int] = mapped_column(default=0) + output_tokens: Mapped[int] = mapped_column(default=0) + total_tokens: Mapped[int] = mapped_column(default=0) + + # Execution metrics + execution_time_ms: Mapped[int] = mapped_column(default=0) + llm_time_ms: Mapped[int] = mapped_column(default=0, nullable=True) + + # Cost (stored in USD) + cost_usd: Mapped[Decimal] = mapped_column( + Numeric(precision=10, scale=6), default=Decimal("0.000000") + ) + + # Parent execution for nested agent calls + parent_execution_id: Mapped[uuid.UUID] = mapped_column( + ForeignKey("agentexecutions.id", ondelete="SET NULL"), nullable=True, index=True + ) + + # Query/input summary (truncated for privacy) + query_preview: Mapped[str] = mapped_column(nullable=True) + + # Relationships + traces: Mapped[List["ExecutionTrace"]] = relationship( + back_populates="execution", cascade="all, delete-orphan" + ) + token_usages: Mapped[List["TokenUsage"]] = relationship( + back_populates="execution", cascade="all, delete-orphan" + ) + + created_at: Mapped[created_at] + updated_at: Mapped[updated_at] + + def __repr__(self) -> str: + return f"" + + +class TokenUsage(Base): + """Detailed token usage tracking per component within an execution.""" + + __tablename__ = "tokenusages" + + id: Mapped[uuid_pk] + execution_id: Mapped[uuid.UUID] = mapped_column( + ForeignKey("agentexecutions.id", ondelete="CASCADE"), nullable=False, index=True + ) + execution: Mapped["AgentExecution"] = relationship(back_populates="token_usages") + + # Which component used the tokens + component: Mapped[str] = mapped_column(nullable=False) # master_agent, agent_{name}, mcp_tool, etc. + step_number: Mapped[int] = mapped_column(nullable=True) + + # Token counts + input_tokens: Mapped[int] = mapped_column(default=0) + output_tokens: Mapped[int] = mapped_column(default=0) + + # Model used + model: Mapped[str] = mapped_column(nullable=True) + + # Cost + cost_usd: Mapped[Decimal] = mapped_column( + Numeric(precision=10, scale=6), default=Decimal("0.000000") + ) + + created_at: Mapped[created_at] + + +class ExecutionTrace(Base): + """Step-by-step trace of execution for debugging and visualization.""" + + __tablename__ = "executiontraces" + + id: Mapped[uuid_pk] + execution_id: Mapped[uuid.UUID] = mapped_column( + ForeignKey("agentexecutions.id", ondelete="CASCADE"), nullable=False, index=True + ) + execution: Mapped["AgentExecution"] = relationship(back_populates="traces") + + step_number: Mapped[int] = mapped_column(nullable=False) + step_type: Mapped[ExecutionTraceStepType] + + # Content of the step (thought, action, observation, etc.) + content: Mapped[str] = mapped_column(nullable=True) + + # For agent invocations + invoked_agent_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), nullable=True + ) + invoked_agent_name: Mapped[str] = mapped_column(nullable=True) + + # Timing + timestamp: Mapped[created_at] + duration_ms: Mapped[int] = mapped_column(default=0) + + # Token usage for this step + input_tokens: Mapped[int] = mapped_column(default=0, nullable=True) + output_tokens: Mapped[int] = mapped_column(default=0, nullable=True) + + +class BudgetAlert(Base): + """Budget alerts and spending limits.""" + + __tablename__ = "budgetalerts" + + id: Mapped[uuid_pk] + + # Who owns this budget + user_id: Mapped[uuid.UUID] = mapped_column( + ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True + ) + + # What the budget applies to + scope: Mapped[BudgetScope] = mapped_column(default=BudgetScope.user) + scope_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), nullable=True + ) # agent_id, flow_id, etc. + + # Budget configuration + threshold_usd: Mapped[Decimal] = mapped_column( + Numeric(precision=10, scale=2), nullable=False + ) + period_days: Mapped[int] = mapped_column(default=30) # Rolling window + + # Alert settings + alert_type: Mapped[BudgetAlertType] = mapped_column(default=BudgetAlertType.warning) + alert_at_percentage: Mapped[int] = mapped_column(default=80) # Alert at 80% of budget + webhook_url: Mapped[str] = mapped_column(nullable=True) + email_notification: Mapped[bool] = mapped_column(default=True) + + # Status + is_active: Mapped[bool] = mapped_column(default=True) + last_alert_sent_at: Mapped[Optional[datetime]] = mapped_column(nullable=True) + current_spend_usd: Mapped[Decimal] = mapped_column( + Numeric(precision=10, scale=2), default=Decimal("0.00") + ) + + created_at: Mapped[created_at] + updated_at: Mapped[updated_at] + + +class AnalyticsSnapshot(Base): + """Pre-aggregated analytics data for fast dashboard queries.""" + + __tablename__ = "analyticssnapshots" + + id: Mapped[uuid_pk] + + # Snapshot period + snapshot_date: Mapped[datetime] = mapped_column(index=True, nullable=False) + period_type: Mapped[str] = mapped_column(nullable=False) # hourly, daily, weekly, monthly + + # Scope + user_id: Mapped[uuid.UUID] = mapped_column( + ForeignKey("users.id", ondelete="CASCADE"), nullable=True, index=True + ) + agent_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), nullable=True, index=True + ) + agent_name: Mapped[str] = mapped_column(nullable=True) + model_name: Mapped[str] = mapped_column(nullable=True) + + # Execution metrics + total_executions: Mapped[int] = mapped_column(default=0) + successful_executions: Mapped[int] = mapped_column(default=0) + failed_executions: Mapped[int] = mapped_column(default=0) + timeout_executions: Mapped[int] = mapped_column(default=0) + + # Timing metrics (in ms) + avg_execution_time_ms: Mapped[int] = mapped_column(default=0) + p50_execution_time_ms: Mapped[int] = mapped_column(default=0) + p95_execution_time_ms: Mapped[int] = mapped_column(default=0) + p99_execution_time_ms: Mapped[int] = mapped_column(default=0) + + # Token metrics + total_input_tokens: Mapped[int] = mapped_column(default=0) + total_output_tokens: Mapped[int] = mapped_column(default=0) + total_tokens: Mapped[int] = mapped_column(default=0) + + # Cost metrics + total_cost_usd: Mapped[Decimal] = mapped_column( + Numeric(precision=10, scale=2), default=Decimal("0.00") + ) + + created_at: Mapped[created_at] + diff --git a/backend/src/repositories/analytics.py b/backend/src/repositories/analytics.py new file mode 100644 index 000000000..30c4d543a --- /dev/null +++ b/backend/src/repositories/analytics.py @@ -0,0 +1,647 @@ +from datetime import datetime, timedelta +from decimal import Decimal +from typing import List, Optional, Tuple + +from sqlalchemy import and_, case, desc, func, select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload + +from src.models import ( + AgentExecution, + AnalyticsSnapshot, + BudgetAlert, + ExecutionTrace, + TokenUsage, +) +from src.repositories.base import CRUDBase +from src.schemas.api.analytics.dto import ( + AgentExecutionDTO, + AgentExecutionSummaryDTO, + AgentStatsDTO, + AnalyticsOverviewDTO, + AnalyticsTimeSeriesDataPoint, + CostBreakdownDTO, + CostForecastDTO, + PaginatedExecutionsDTO, + TopAgentDTO, +) +from src.schemas.api.analytics.schemas import ( + AgentExecutionCreate, + AgentExecutionUpdate, + BudgetAlertCreate, + BudgetAlertUpdate, + ExecutionTraceCreate, + TokenUsageCreate, +) +from src.utils.enums import ExecutionStatus + + +class AgentExecutionRepository(CRUDBase[AgentExecution, AgentExecutionCreate, AgentExecutionUpdate]): + """Repository for agent execution analytics.""" + + async def get_with_details( + self, db: AsyncSession, execution_id: str + ) -> Optional[AgentExecution]: + """Get execution with traces and token usages.""" + result = await db.execute( + select(self.model) + .options( + selectinload(AgentExecution.traces), + selectinload(AgentExecution.token_usages), + ) + .where(self.model.id == execution_id) + ) + return result.scalars().first() + + async def get_by_request_id( + self, db: AsyncSession, request_id: str + ) -> List[AgentExecution]: + """Get all executions for a request.""" + result = await db.execute( + select(self.model) + .where(self.model.request_id == request_id) + .order_by(self.model.started_at) + ) + return result.scalars().all() + + async def get_by_session_id( + self, db: AsyncSession, session_id: str, limit: int = 100 + ) -> List[AgentExecution]: + """Get all executions for a session.""" + result = await db.execute( + select(self.model) + .where(self.model.session_id == session_id) + .order_by(desc(self.model.started_at)) + .limit(limit) + ) + return result.scalars().all() + + async def list_paginated( + self, + db: AsyncSession, + user_id: Optional[str] = None, + agent_id: Optional[str] = None, + agent_type: Optional[str] = None, + model_name: Optional[str] = None, + status: Optional[ExecutionStatus] = None, + start_date: Optional[datetime] = None, + end_date: Optional[datetime] = None, + page: int = 1, + page_size: int = 20, + sort_by: str = "started_at", + sort_order: str = "desc", + ) -> PaginatedExecutionsDTO: + """Get paginated list of executions with filters.""" + # Build filters + filters = [] + if user_id: + filters.append(self.model.user_id == user_id) + if agent_id: + filters.append(self.model.agent_id == agent_id) + if agent_type: + filters.append(self.model.agent_type == agent_type) + if model_name: + filters.append(self.model.model_name == model_name) + if status: + filters.append(self.model.status == status) + if start_date: + filters.append(self.model.started_at >= start_date) + if end_date: + filters.append(self.model.started_at <= end_date) + + # Count total + count_query = select(func.count()).select_from(self.model) + if filters: + count_query = count_query.where(and_(*filters)) + total_result = await db.execute(count_query) + total = total_result.scalar() or 0 + + # Get items + query = select(self.model) + if filters: + query = query.where(and_(*filters)) + + # Sort + sort_column = getattr(self.model, sort_by, self.model.started_at) + if sort_order == "desc": + query = query.order_by(desc(sort_column)) + else: + query = query.order_by(sort_column) + + # Paginate + offset = (page - 1) * page_size + query = query.offset(offset).limit(page_size) + + result = await db.execute(query) + items = result.scalars().all() + + return PaginatedExecutionsDTO( + items=[ + AgentExecutionSummaryDTO( + id=str(item.id), + request_id=str(item.request_id), + session_id=str(item.session_id), + agent_name=item.agent_name, + agent_type=item.agent_type, + model_name=item.model_name, + status=item.status, + execution_time_ms=item.execution_time_ms, + total_tokens=item.total_tokens, + cost_usd=item.cost_usd, + started_at=item.started_at, + completed_at=item.completed_at, + query_preview=item.query_preview, + ) + for item in items + ], + total=total, + page=page, + page_size=page_size, + total_pages=(total + page_size - 1) // page_size, + ) + + async def get_overview( + self, + db: AsyncSession, + user_id: Optional[str] = None, + start_date: Optional[datetime] = None, + end_date: Optional[datetime] = None, + ) -> AnalyticsOverviewDTO: + """Get analytics overview for dashboard.""" + # Default to last 24 hours if no dates specified + if not end_date: + end_date = datetime.utcnow() + if not start_date: + start_date = end_date - timedelta(hours=24) + + # Build base filter + filters = [ + self.model.started_at >= start_date, + self.model.started_at <= end_date, + ] + if user_id: + filters.append(self.model.user_id == user_id) + + # Get aggregate metrics + metrics_query = select( + func.count().label("total"), + func.sum(case((self.model.status == ExecutionStatus.success, 1), else_=0)).label( + "successful" + ), + func.sum(case((self.model.status == ExecutionStatus.failure, 1), else_=0)).label( + "failed" + ), + func.avg(self.model.execution_time_ms).label("avg_time"), + func.sum(self.model.input_tokens).label("input_tokens"), + func.sum(self.model.output_tokens).label("output_tokens"), + func.sum(self.model.total_tokens).label("total_tokens"), + func.sum(self.model.cost_usd).label("total_cost"), + ).where(and_(*filters)) + + result = await db.execute(metrics_query) + metrics = result.first() + + total_executions = metrics.total or 0 + successful_executions = metrics.successful or 0 + failed_executions = metrics.failed or 0 + success_rate = ( + (successful_executions / total_executions * 100) + if total_executions > 0 + else 0.0 + ) + + # Get percentile execution times + time_query = select(self.model.execution_time_ms).where( + and_(*filters, self.model.execution_time_ms > 0) + ).order_by(self.model.execution_time_ms) + time_result = await db.execute(time_query) + times = [r[0] for r in time_result.fetchall()] + + p50 = times[len(times) // 2] if times else 0 + p95 = times[int(len(times) * 0.95)] if times else 0 + p99 = times[int(len(times) * 0.99)] if times else 0 + + # Get top agents + top_agents_query = ( + select( + self.model.agent_name, + self.model.agent_id, + self.model.agent_type, + func.count().label("count"), + func.sum(case((self.model.status == ExecutionStatus.success, 1), else_=0)).label( + "success_count" + ), + func.sum(case((self.model.status == ExecutionStatus.failure, 1), else_=0)).label( + "failure_count" + ), + func.avg(self.model.execution_time_ms).label("avg_time"), + func.sum(self.model.total_tokens).label("total_tokens"), + func.sum(self.model.cost_usd).label("total_cost"), + ) + .where(and_(*filters, self.model.agent_name.isnot(None))) + .group_by(self.model.agent_name, self.model.agent_id, self.model.agent_type) + .order_by(desc("count")) + .limit(10) + ) + top_agents_result = await db.execute(top_agents_query) + top_agents = [ + TopAgentDTO( + agent_name=row.agent_name or "Unknown", + agent_id=str(row.agent_id) if row.agent_id else None, + agent_type=row.agent_type, + execution_count=row.count, + success_count=row.success_count or 0, + failure_count=row.failure_count or 0, + success_rate=(row.success_count or 0) / row.count * 100 if row.count > 0 else 0, + avg_execution_time_ms=float(row.avg_time or 0), + total_tokens=row.total_tokens or 0, + total_cost_usd=Decimal(str(row.total_cost or 0)), + ) + for row in top_agents_result.fetchall() + ] + + # Get cost breakdown by model + model_cost_query = ( + select( + self.model.model_name, + func.sum(self.model.cost_usd).label("cost"), + func.count().label("count"), + func.sum(self.model.total_tokens).label("tokens"), + ) + .where(and_(*filters, self.model.model_name.isnot(None))) + .group_by(self.model.model_name) + .order_by(desc("cost")) + ) + model_cost_result = await db.execute(model_cost_query) + total_cost = Decimal(str(metrics.total_cost or 0)) + cost_by_model = [ + CostBreakdownDTO( + name=row.model_name or "Unknown", + category="model", + total_cost_usd=Decimal(str(row.cost or 0)), + percentage=float(row.cost / total_cost * 100) if total_cost > 0 else 0, + execution_count=row.count, + total_tokens=row.tokens or 0, + ) + for row in model_cost_result.fetchall() + ] + + # Calculate comparison with previous period + period_length = end_date - start_date + prev_start = start_date - period_length + prev_end = start_date + + prev_filters = [ + self.model.started_at >= prev_start, + self.model.started_at < prev_end, + ] + if user_id: + prev_filters.append(self.model.user_id == user_id) + + prev_query = select( + func.count().label("total"), + func.sum(case((self.model.status == ExecutionStatus.success, 1), else_=0)).label( + "successful" + ), + func.avg(self.model.execution_time_ms).label("avg_time"), + func.sum(self.model.cost_usd).label("total_cost"), + ).where(and_(*prev_filters)) + + prev_result = await db.execute(prev_query) + prev_metrics = prev_result.first() + + prev_total = prev_metrics.total or 0 + prev_successful = prev_metrics.successful or 0 + prev_success_rate = (prev_successful / prev_total * 100) if prev_total > 0 else 0 + prev_avg_time = float(prev_metrics.avg_time or 0) + prev_cost = Decimal(str(prev_metrics.total_cost or 0)) + + def calc_change(current: float, previous: float) -> Optional[float]: + if previous == 0: + return None + return ((current - previous) / previous) * 100 + + # Count active agents + active_agents_query = select( + func.count(func.distinct(self.model.agent_id)) + ).where(and_(*filters, self.model.agent_id.isnot(None))) + active_result = await db.execute(active_agents_query) + active_agents = active_result.scalar() or 0 + + return AnalyticsOverviewDTO( + total_executions=total_executions, + successful_executions=successful_executions, + failed_executions=failed_executions, + success_rate=success_rate, + avg_execution_time_ms=float(metrics.avg_time or 0), + p50_execution_time_ms=float(p50), + p95_execution_time_ms=float(p95), + p99_execution_time_ms=float(p99), + total_input_tokens=metrics.input_tokens or 0, + total_output_tokens=metrics.output_tokens or 0, + total_tokens=metrics.total_tokens or 0, + total_cost_usd=total_cost, + estimated_monthly_cost_usd=total_cost * 30 / max(period_length.days, 1), + executions_change_percent=calc_change(float(total_executions), float(prev_total)), + execution_time_change_percent=calc_change(float(metrics.avg_time or 0), prev_avg_time), + success_rate_change_percent=calc_change(success_rate, prev_success_rate), + cost_change_percent=calc_change(float(total_cost), float(prev_cost)), + active_agents_count=active_agents, + top_agents=top_agents, + cost_by_model=cost_by_model, + ) + + async def get_agent_stats( + self, + db: AsyncSession, + agent_id: str, + user_id: Optional[str] = None, + start_date: Optional[datetime] = None, + end_date: Optional[datetime] = None, + ) -> Optional[AgentStatsDTO]: + """Get detailed statistics for a specific agent.""" + if not end_date: + end_date = datetime.utcnow() + if not start_date: + start_date = end_date - timedelta(days=30) + + filters = [ + self.model.agent_id == agent_id, + self.model.started_at >= start_date, + self.model.started_at <= end_date, + ] + if user_id: + filters.append(self.model.user_id == user_id) + + # Get aggregate stats + stats_query = select( + self.model.agent_name, + self.model.agent_type, + func.count().label("total"), + func.sum(case((self.model.status == ExecutionStatus.success, 1), else_=0)).label( + "successful" + ), + func.sum(case((self.model.status == ExecutionStatus.failure, 1), else_=0)).label( + "failed" + ), + func.sum(case((self.model.status == ExecutionStatus.timeout, 1), else_=0)).label( + "timeout" + ), + func.avg(self.model.execution_time_ms).label("avg_time"), + func.min(self.model.execution_time_ms).label("min_time"), + func.max(self.model.execution_time_ms).label("max_time"), + func.avg(self.model.input_tokens).label("avg_input"), + func.avg(self.model.output_tokens).label("avg_output"), + func.sum(self.model.total_tokens).label("total_tokens"), + func.sum(self.model.cost_usd).label("total_cost"), + func.max(self.model.started_at).label("last_executed"), + func.min(self.model.started_at).label("first_executed"), + ).where(and_(*filters)).group_by(self.model.agent_name, self.model.agent_type) + + result = await db.execute(stats_query) + stats = result.first() + + if not stats: + return None + + total = stats.total or 0 + successful = stats.successful or 0 + + return AgentStatsDTO( + agent_id=agent_id, + agent_name=stats.agent_name or "Unknown", + agent_type=stats.agent_type, + total_executions=total, + successful_executions=successful, + failed_executions=stats.failed or 0, + timeout_executions=stats.timeout or 0, + success_rate=(successful / total * 100) if total > 0 else 0, + avg_execution_time_ms=float(stats.avg_time or 0), + min_execution_time_ms=stats.min_time or 0, + max_execution_time_ms=stats.max_time or 0, + avg_input_tokens=float(stats.avg_input or 0), + avg_output_tokens=float(stats.avg_output or 0), + total_tokens=stats.total_tokens or 0, + total_cost_usd=Decimal(str(stats.total_cost or 0)), + avg_cost_per_execution_usd=Decimal(str(stats.total_cost or 0)) / total if total > 0 else Decimal("0"), + last_executed_at=stats.last_executed, + first_executed_at=stats.first_executed, + ) + + async def get_cost_summary( + self, + db: AsyncSession, + user_id: Optional[str] = None, + start_date: Optional[datetime] = None, + end_date: Optional[datetime] = None, + ) -> dict: + """Get cost summary breakdown.""" + if not end_date: + end_date = datetime.utcnow() + if not start_date: + start_date = end_date - timedelta(days=30) + + filters = [ + self.model.started_at >= start_date, + self.model.started_at <= end_date, + ] + if user_id: + filters.append(self.model.user_id == user_id) + + # Total cost + total_query = select(func.sum(self.model.cost_usd)).where(and_(*filters)) + total_result = await db.execute(total_query) + total_cost = Decimal(str(total_result.scalar() or 0)) + + # By model + by_model_query = ( + select( + self.model.model_name, + func.sum(self.model.cost_usd).label("cost"), + func.count().label("count"), + ) + .where(and_(*filters)) + .group_by(self.model.model_name) + .order_by(desc("cost")) + ) + by_model_result = await db.execute(by_model_query) + by_model = [ + { + "model": row.model_name or "Unknown", + "cost_usd": Decimal(str(row.cost or 0)), + "execution_count": row.count, + "percentage": float(row.cost / total_cost * 100) if total_cost > 0 else 0, + } + for row in by_model_result.fetchall() + ] + + # By agent + by_agent_query = ( + select( + self.model.agent_name, + self.model.agent_id, + func.sum(self.model.cost_usd).label("cost"), + func.count().label("count"), + ) + .where(and_(*filters)) + .group_by(self.model.agent_name, self.model.agent_id) + .order_by(desc("cost")) + ) + by_agent_result = await db.execute(by_agent_query) + by_agent = [ + { + "agent_name": row.agent_name or "Unknown", + "agent_id": str(row.agent_id) if row.agent_id else None, + "cost_usd": Decimal(str(row.cost or 0)), + "execution_count": row.count, + "percentage": float(row.cost / total_cost * 100) if total_cost > 0 else 0, + } + for row in by_agent_result.fetchall() + ] + + return { + "total_cost_usd": total_cost, + "period_days": (end_date - start_date).days, + "by_model": by_model, + "by_agent": by_agent, + } + + async def get_cost_forecast( + self, + db: AsyncSession, + user_id: Optional[str] = None, + ) -> CostForecastDTO: + """Get cost forecast based on recent usage.""" + now = datetime.utcnow() + last_7_days = now - timedelta(days=7) + prev_7_days = last_7_days - timedelta(days=7) + + filters_current = [self.model.started_at >= last_7_days] + filters_prev = [ + self.model.started_at >= prev_7_days, + self.model.started_at < last_7_days, + ] + if user_id: + filters_current.append(self.model.user_id == user_id) + filters_prev.append(self.model.user_id == user_id) + + # Current period cost + current_query = select(func.sum(self.model.cost_usd)).where(and_(*filters_current)) + current_result = await db.execute(current_query) + current_cost = Decimal(str(current_result.scalar() or 0)) + + # Previous period cost + prev_query = select(func.sum(self.model.cost_usd)).where(and_(*filters_prev)) + prev_result = await db.execute(prev_query) + prev_cost = Decimal(str(prev_result.scalar() or 0)) + + daily_avg = current_cost / 7 + weekly_projected = current_cost + monthly_projected = daily_avg * 30 + + # Determine trend + if prev_cost > 0: + change = float((current_cost - prev_cost) / prev_cost * 100) + if change > 10: + trend = "increasing" + elif change < -10: + trend = "decreasing" + else: + trend = "stable" + else: + change = 0.0 + trend = "stable" + + return CostForecastDTO( + current_period_cost_usd=current_cost, + projected_monthly_cost_usd=monthly_projected, + projected_weekly_cost_usd=weekly_projected, + daily_average_cost_usd=daily_avg, + cost_trend=trend, + trend_percentage=change, + ) + + +class TokenUsageRepository(CRUDBase[TokenUsage, TokenUsageCreate, TokenUsageCreate]): + """Repository for token usage records.""" + + async def get_by_execution( + self, db: AsyncSession, execution_id: str + ) -> List[TokenUsage]: + """Get all token usage records for an execution.""" + result = await db.execute( + select(self.model) + .where(self.model.execution_id == execution_id) + .order_by(self.model.step_number) + ) + return result.scalars().all() + + +class ExecutionTraceRepository(CRUDBase[ExecutionTrace, ExecutionTraceCreate, ExecutionTraceCreate]): + """Repository for execution trace records.""" + + async def get_by_execution( + self, db: AsyncSession, execution_id: str + ) -> List[ExecutionTrace]: + """Get all trace records for an execution.""" + result = await db.execute( + select(self.model) + .where(self.model.execution_id == execution_id) + .order_by(self.model.step_number) + ) + return result.scalars().all() + + +class BudgetAlertRepository(CRUDBase[BudgetAlert, BudgetAlertCreate, BudgetAlertUpdate]): + """Repository for budget alerts.""" + + async def get_active_by_user( + self, db: AsyncSession, user_id: str + ) -> List[BudgetAlert]: + """Get all active budget alerts for a user.""" + result = await db.execute( + select(self.model).where( + and_( + self.model.user_id == user_id, + self.model.is_active.is_(True), + ) + ) + ) + return result.scalars().all() + + async def check_budget_exceeded( + self, db: AsyncSession, user_id: str + ) -> List[Tuple[BudgetAlert, Decimal]]: + """Check if any budgets are exceeded and return alerts with current spend.""" + alerts = await self.get_active_by_user(db, user_id) + exceeded = [] + + for alert in alerts: + # Calculate current spend for the period + start_date = datetime.utcnow() - timedelta(days=alert.period_days) + + filters = [ + AgentExecution.user_id == user_id, + AgentExecution.started_at >= start_date, + ] + + if alert.scope.value == "agent" and alert.scope_id: + filters.append(AgentExecution.agent_id == alert.scope_id) + + query = select(func.sum(AgentExecution.cost_usd)).where(and_(*filters)) + result = await db.execute(query) + current_spend = Decimal(str(result.scalar() or 0)) + + threshold = alert.threshold_usd * Decimal(str(alert.alert_at_percentage)) / 100 + + if current_spend >= threshold: + exceeded.append((alert, current_spend)) + + return exceeded + + +# Repository instances +agent_execution_repo = AgentExecutionRepository(AgentExecution) +token_usage_repo = TokenUsageRepository(TokenUsage) +execution_trace_repo = ExecutionTraceRepository(ExecutionTrace) +budget_alert_repo = BudgetAlertRepository(BudgetAlert) diff --git a/backend/src/routes/analytics/__init__.py b/backend/src/routes/analytics/__init__.py new file mode 100644 index 000000000..1cd41fcf4 --- /dev/null +++ b/backend/src/routes/analytics/__init__.py @@ -0,0 +1,3 @@ +from src.routes.analytics.routes import analytics_router + +__all__ = ["analytics_router"] diff --git a/backend/src/routes/analytics/routes.py b/backend/src/routes/analytics/routes.py new file mode 100644 index 000000000..7f49aa759 --- /dev/null +++ b/backend/src/routes/analytics/routes.py @@ -0,0 +1,623 @@ +from datetime import datetime, timedelta +from typing import Annotated, List, Optional +from uuid import UUID + +from fastapi import APIRouter, HTTPException, Query +from fastapi.responses import StreamingResponse +import csv +import io +import json + +from src.auth.dependencies import CurrentUserDependency +from src.db.session import AsyncDBSession +from src.repositories.analytics import ( + agent_execution_repo, + budget_alert_repo, + execution_trace_repo, + token_usage_repo, +) +from src.schemas.api.analytics.dto import ( + AgentExecutionDTO, + AgentStatsDTO, + AnalyticsOverviewDTO, + BudgetAlertDTO, + CostForecastDTO, + ExecutionTraceDTO, + PaginatedExecutionsDTO, + TokenUsageDTO, +) +from src.schemas.api.analytics.schemas import ( + BudgetAlertCreate, + BudgetAlertUpdate, +) +from src.utils.enums import ExecutionStatus + +analytics_router = APIRouter(tags=["Analytics"], prefix="/analytics") + + +def parse_period(period: str) -> tuple[datetime, datetime]: + """Parse period string into start and end dates.""" + now = datetime.utcnow() + if period == "24h": + return now - timedelta(hours=24), now + elif period == "7d": + return now - timedelta(days=7), now + elif period == "30d": + return now - timedelta(days=30), now + elif period == "90d": + return now - timedelta(days=90), now + else: + # Default to 24h + return now - timedelta(hours=24), now + + +@analytics_router.get("/overview", response_model=AnalyticsOverviewDTO) +async def get_analytics_overview( + db: AsyncDBSession, + user: CurrentUserDependency, + period: Annotated[str, Query(description="Time period: 24h, 7d, 30d, 90d")] = "24h", + start_date: Annotated[Optional[datetime], Query()] = None, + end_date: Annotated[Optional[datetime], Query()] = None, +) -> AnalyticsOverviewDTO: + """ + Get analytics overview for the dashboard. + + Returns summary statistics including: + - Total executions and success rate + - Execution time metrics (avg, p50, p95, p99) + - Token usage totals + - Cost summary + - Top agents by usage + - Cost breakdown by model + """ + if start_date and end_date: + pass # Use provided dates + else: + start_date, end_date = parse_period(period) + + return await agent_execution_repo.get_overview( + db=db, + user_id=str(user.id), + start_date=start_date, + end_date=end_date, + ) + + +@analytics_router.get("/executions", response_model=PaginatedExecutionsDTO) +async def list_executions( + db: AsyncDBSession, + user: CurrentUserDependency, + period: Annotated[str, Query()] = "24h", + start_date: Annotated[Optional[datetime], Query()] = None, + end_date: Annotated[Optional[datetime], Query()] = None, + agent_id: Annotated[Optional[UUID], Query()] = None, + agent_type: Annotated[Optional[str], Query()] = None, + model_name: Annotated[Optional[str], Query()] = None, + status: Annotated[Optional[ExecutionStatus], Query()] = None, + page: Annotated[int, Query(ge=1)] = 1, + page_size: Annotated[int, Query(ge=1, le=100)] = 20, + sort_by: Annotated[str, Query()] = "started_at", + sort_order: Annotated[str, Query()] = "desc", +) -> PaginatedExecutionsDTO: + """ + Get paginated list of agent executions. + + Supports filtering by: + - Time period + - Agent ID and type + - Model name + - Execution status + """ + if start_date and end_date: + pass + else: + start_date, end_date = parse_period(period) + + return await agent_execution_repo.list_paginated( + db=db, + user_id=str(user.id), + agent_id=str(agent_id) if agent_id else None, + agent_type=agent_type, + model_name=model_name, + status=status, + start_date=start_date, + end_date=end_date, + page=page, + page_size=page_size, + sort_by=sort_by, + sort_order=sort_order, + ) + + +@analytics_router.get("/executions/{execution_id}", response_model=AgentExecutionDTO) +async def get_execution_details( + db: AsyncDBSession, + user: CurrentUserDependency, + execution_id: UUID, +) -> AgentExecutionDTO: + """ + Get detailed information about a specific execution. + + Includes: + - Full execution metadata + - Step-by-step trace + - Token usage breakdown + """ + execution = await agent_execution_repo.get_with_details( + db=db, execution_id=str(execution_id) + ) + + if not execution: + raise HTTPException(status_code=404, detail="Execution not found") + + # Check user access + if execution.user_id and str(execution.user_id) != str(user.id): + raise HTTPException(status_code=403, detail="Access denied") + + traces = [ + ExecutionTraceDTO( + id=str(t.id), + execution_id=str(t.execution_id), + step_number=t.step_number, + step_type=t.step_type, + content=t.content, + invoked_agent_id=str(t.invoked_agent_id) if t.invoked_agent_id else None, + invoked_agent_name=t.invoked_agent_name, + timestamp=t.timestamp, + duration_ms=t.duration_ms, + input_tokens=t.input_tokens, + output_tokens=t.output_tokens, + ) + for t in execution.traces + ] + + token_usages = [ + TokenUsageDTO( + id=str(tu.id), + execution_id=str(tu.execution_id), + component=tu.component, + step_number=tu.step_number, + input_tokens=tu.input_tokens, + output_tokens=tu.output_tokens, + model=tu.model, + cost_usd=tu.cost_usd, + created_at=tu.created_at, + ) + for tu in execution.token_usages + ] + + return AgentExecutionDTO( + id=str(execution.id), + request_id=str(execution.request_id), + session_id=str(execution.session_id), + agent_id=str(execution.agent_id) if execution.agent_id else None, + agent_type=execution.agent_type, + agent_name=execution.agent_name, + user_id=str(execution.user_id) if execution.user_id else None, + model_config_id=str(execution.model_config_id) if execution.model_config_id else None, + model_name=execution.model_name, + started_at=execution.started_at, + completed_at=execution.completed_at, + status=execution.status, + error_message=execution.error_message, + error_type=execution.error_type, + input_tokens=execution.input_tokens, + output_tokens=execution.output_tokens, + total_tokens=execution.total_tokens, + execution_time_ms=execution.execution_time_ms, + llm_time_ms=execution.llm_time_ms, + cost_usd=execution.cost_usd, + parent_execution_id=str(execution.parent_execution_id) if execution.parent_execution_id else None, + query_preview=execution.query_preview, + created_at=execution.created_at, + updated_at=execution.updated_at, + traces=traces, + token_usages=token_usages, + ) + + +@analytics_router.get("/agents/{agent_id}/stats", response_model=AgentStatsDTO) +async def get_agent_stats( + db: AsyncDBSession, + user: CurrentUserDependency, + agent_id: UUID, + period: Annotated[str, Query()] = "30d", + start_date: Annotated[Optional[datetime], Query()] = None, + end_date: Annotated[Optional[datetime], Query()] = None, +) -> AgentStatsDTO: + """ + Get detailed statistics for a specific agent. + + Includes: + - Execution counts and success rate + - Timing metrics (avg, min, max, percentiles) + - Token usage averages + - Cost totals and averages + - Error analysis + """ + if start_date and end_date: + pass + else: + start_date, end_date = parse_period(period) + + stats = await agent_execution_repo.get_agent_stats( + db=db, + agent_id=str(agent_id), + user_id=str(user.id), + start_date=start_date, + end_date=end_date, + ) + + if not stats: + raise HTTPException(status_code=404, detail="No data found for this agent") + + return stats + + +@analytics_router.get("/costs/summary") +async def get_cost_summary( + db: AsyncDBSession, + user: CurrentUserDependency, + period: Annotated[str, Query()] = "30d", + start_date: Annotated[Optional[datetime], Query()] = None, + end_date: Annotated[Optional[datetime], Query()] = None, +) -> dict: + """ + Get cost breakdown summary. + + Returns: + - Total cost for the period + - Cost breakdown by model + - Cost breakdown by agent + """ + if start_date and end_date: + pass + else: + start_date, end_date = parse_period(period) + + return await agent_execution_repo.get_cost_summary( + db=db, + user_id=str(user.id), + start_date=start_date, + end_date=end_date, + ) + + +@analytics_router.get("/costs/forecast", response_model=CostForecastDTO) +async def get_cost_forecast( + db: AsyncDBSession, + user: CurrentUserDependency, +) -> CostForecastDTO: + """ + Get cost forecast based on recent usage patterns. + + Returns: + - Current period cost + - Projected weekly and monthly costs + - Daily average + - Cost trend (increasing/decreasing/stable) + """ + return await agent_execution_repo.get_cost_forecast( + db=db, + user_id=str(user.id), + ) + + +# Budget Alerts + + +@analytics_router.get("/budgets", response_model=List[BudgetAlertDTO]) +async def list_budget_alerts( + db: AsyncDBSession, + user: CurrentUserDependency, +) -> List[BudgetAlertDTO]: + """Get all budget alerts for the current user.""" + alerts = await budget_alert_repo.get_active_by_user(db, str(user.id)) + return [ + BudgetAlertDTO( + id=str(a.id), + user_id=str(a.user_id), + scope=a.scope, + scope_id=str(a.scope_id) if a.scope_id else None, + threshold_usd=a.threshold_usd, + period_days=a.period_days, + alert_type=a.alert_type, + alert_at_percentage=a.alert_at_percentage, + webhook_url=a.webhook_url, + email_notification=a.email_notification, + is_active=a.is_active, + last_alert_sent_at=a.last_alert_sent_at, + current_spend_usd=a.current_spend_usd, + created_at=a.created_at, + updated_at=a.updated_at, + ) + for a in alerts + ] + + +@analytics_router.post("/budgets", response_model=BudgetAlertDTO) +async def create_budget_alert( + db: AsyncDBSession, + user: CurrentUserDependency, + alert_data: BudgetAlertCreate, +) -> BudgetAlertDTO: + """Create a new budget alert.""" + from src.models import BudgetAlert + + alert = BudgetAlert( + user_id=user.id, + scope=alert_data.scope, + scope_id=alert_data.scope_id, + threshold_usd=alert_data.threshold_usd, + period_days=alert_data.period_days, + alert_type=alert_data.alert_type, + alert_at_percentage=alert_data.alert_at_percentage, + webhook_url=alert_data.webhook_url, + email_notification=alert_data.email_notification, + ) + + created = await budget_alert_repo.create(db, obj_in=alert) + + return BudgetAlertDTO( + id=str(created.id), + user_id=str(created.user_id), + scope=created.scope, + scope_id=str(created.scope_id) if created.scope_id else None, + threshold_usd=created.threshold_usd, + period_days=created.period_days, + alert_type=created.alert_type, + alert_at_percentage=created.alert_at_percentage, + webhook_url=created.webhook_url, + email_notification=created.email_notification, + is_active=created.is_active, + last_alert_sent_at=created.last_alert_sent_at, + current_spend_usd=created.current_spend_usd, + created_at=created.created_at, + updated_at=created.updated_at, + ) + + +@analytics_router.patch("/budgets/{alert_id}", response_model=BudgetAlertDTO) +async def update_budget_alert( + db: AsyncDBSession, + user: CurrentUserDependency, + alert_id: UUID, + alert_data: BudgetAlertUpdate, +) -> BudgetAlertDTO: + """Update a budget alert.""" + alert = await budget_alert_repo.get(db, id_=str(alert_id)) + + if not alert: + raise HTTPException(status_code=404, detail="Budget alert not found") + + if str(alert.user_id) != str(user.id): + raise HTTPException(status_code=403, detail="Access denied") + + updated = await budget_alert_repo.update(db, db_obj=alert, obj_in=alert_data) + + return BudgetAlertDTO( + id=str(updated.id), + user_id=str(updated.user_id), + scope=updated.scope, + scope_id=str(updated.scope_id) if updated.scope_id else None, + threshold_usd=updated.threshold_usd, + period_days=updated.period_days, + alert_type=updated.alert_type, + alert_at_percentage=updated.alert_at_percentage, + webhook_url=updated.webhook_url, + email_notification=updated.email_notification, + is_active=updated.is_active, + last_alert_sent_at=updated.last_alert_sent_at, + current_spend_usd=updated.current_spend_usd, + created_at=updated.created_at, + updated_at=updated.updated_at, + ) + + +@analytics_router.delete("/budgets/{alert_id}") +async def delete_budget_alert( + db: AsyncDBSession, + user: CurrentUserDependency, + alert_id: UUID, +) -> dict: + """Delete a budget alert.""" + alert = await budget_alert_repo.get(db, id_=str(alert_id)) + + if not alert: + raise HTTPException(status_code=404, detail="Budget alert not found") + + if str(alert.user_id) != str(user.id): + raise HTTPException(status_code=403, detail="Access denied") + + await budget_alert_repo.delete(db, id_=str(alert_id)) + return {"status": "deleted", "id": str(alert_id)} + + +# Export + + +@analytics_router.get("/export") +async def export_analytics_data( + db: AsyncDBSession, + user: CurrentUserDependency, + format: Annotated[str, Query(description="Export format: json or csv")] = "json", + period: Annotated[str, Query()] = "7d", + start_date: Annotated[Optional[datetime], Query()] = None, + end_date: Annotated[Optional[datetime], Query()] = None, + include_traces: Annotated[bool, Query()] = False, +): + """ + Export analytics data in JSON or CSV format. + """ + if start_date and end_date: + pass + else: + start_date, end_date = parse_period(period) + + # Get executions + paginated = await agent_execution_repo.list_paginated( + db=db, + user_id=str(user.id), + start_date=start_date, + end_date=end_date, + page=1, + page_size=10000, # Export all + ) + + if format == "csv": + # Create CSV + output = io.StringIO() + writer = csv.DictWriter( + output, + fieldnames=[ + "id", + "request_id", + "session_id", + "agent_name", + "agent_type", + "model_name", + "status", + "execution_time_ms", + "total_tokens", + "cost_usd", + "started_at", + "completed_at", + "query_preview", + ], + ) + writer.writeheader() + for item in paginated.items: + writer.writerow({ + "id": item.id, + "request_id": item.request_id, + "session_id": item.session_id, + "agent_name": item.agent_name, + "agent_type": item.agent_type, + "model_name": item.model_name, + "status": item.status.value if item.status else "", + "execution_time_ms": item.execution_time_ms, + "total_tokens": item.total_tokens, + "cost_usd": str(item.cost_usd), + "started_at": item.started_at.isoformat() if item.started_at else "", + "completed_at": item.completed_at.isoformat() if item.completed_at else "", + "query_preview": item.query_preview, + }) + + output.seek(0) + return StreamingResponse( + iter([output.getvalue()]), + media_type="text/csv", + headers={ + "Content-Disposition": f"attachment; filename=analytics_export_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.csv" + }, + ) + else: + # JSON format + data = { + "export_date": datetime.utcnow().isoformat(), + "period": {"start": start_date.isoformat(), "end": end_date.isoformat()}, + "total_records": paginated.total, + "executions": [ + { + "id": item.id, + "request_id": item.request_id, + "session_id": item.session_id, + "agent_name": item.agent_name, + "agent_type": item.agent_type, + "model_name": item.model_name, + "status": item.status.value if item.status else None, + "execution_time_ms": item.execution_time_ms, + "total_tokens": item.total_tokens, + "cost_usd": str(item.cost_usd), + "started_at": item.started_at.isoformat() if item.started_at else None, + "completed_at": item.completed_at.isoformat() if item.completed_at else None, + "query_preview": item.query_preview, + } + for item in paginated.items + ], + } + + output = io.StringIO() + json.dump(data, output, indent=2) + output.seek(0) + + return StreamingResponse( + iter([output.getvalue()]), + media_type="application/json", + headers={ + "Content-Disposition": f"attachment; filename=analytics_export_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.json" + }, + ) + + +# Model Comparison + + +@analytics_router.get("/comparison/models") +async def compare_models( + db: AsyncDBSession, + user: CurrentUserDependency, + period: Annotated[str, Query()] = "7d", + start_date: Annotated[Optional[datetime], Query()] = None, + end_date: Annotated[Optional[datetime], Query()] = None, +) -> List[dict]: + """ + Compare performance across different models. + + Returns metrics for each model: + - Execution count + - Success rate + - Average execution time + - Average tokens + - Total cost + """ + if start_date and end_date: + pass + else: + start_date, end_date = parse_period(period) + + from sqlalchemy import and_, case, desc, func, select + from src.models import AgentExecution + + filters = [ + AgentExecution.user_id == str(user.id), + AgentExecution.started_at >= start_date, + AgentExecution.started_at <= end_date, + AgentExecution.model_name.isnot(None), + ] + + query = ( + select( + AgentExecution.model_name, + func.count().label("execution_count"), + func.sum(case((AgentExecution.status == ExecutionStatus.success, 1), else_=0)).label( + "success_count" + ), + func.avg(AgentExecution.execution_time_ms).label("avg_execution_time_ms"), + func.avg(AgentExecution.total_tokens).label("avg_tokens"), + func.sum(AgentExecution.total_tokens).label("total_tokens"), + func.sum(AgentExecution.cost_usd).label("total_cost"), + ) + .where(and_(*filters)) + .group_by(AgentExecution.model_name) + .order_by(desc("execution_count")) + ) + + result = await db.execute(query) + models = result.fetchall() + + return [ + { + "model_name": row.model_name, + "execution_count": row.execution_count, + "success_count": row.success_count or 0, + "success_rate": (row.success_count or 0) / row.execution_count * 100 + if row.execution_count > 0 + else 0, + "avg_execution_time_ms": float(row.avg_execution_time_ms or 0), + "avg_tokens": float(row.avg_tokens or 0), + "total_tokens": row.total_tokens or 0, + "total_cost_usd": str(row.total_cost or 0), + } + for row in models + ] diff --git a/backend/src/routes/api.py b/backend/src/routes/api.py index 6ad1aca01..d4eb4a2ac 100644 --- a/backend/src/routes/api.py +++ b/backend/src/routes/api.py @@ -2,6 +2,7 @@ from src.routes.a2a.routes import a2a_router from src.routes.agents.routes import agent_router +from src.routes.analytics.routes import analytics_router from src.routes.chat.routes import chat_router from src.routes.flows.routes import flow_router from src.routes.llms.routes import llm_router @@ -19,3 +20,4 @@ api_router.include_router(chat_router) api_router.include_router(mcp_router) api_router.include_router(a2a_router) +api_router.include_router(analytics_router) diff --git a/backend/src/schemas/api/analytics/__init__.py b/backend/src/schemas/api/analytics/__init__.py new file mode 100644 index 000000000..ac1f78cee --- /dev/null +++ b/backend/src/schemas/api/analytics/__init__.py @@ -0,0 +1,37 @@ +from src.schemas.api.analytics.dto import ( + AgentExecutionDTO, + AnalyticsOverviewDTO, + AnalyticsTimeSeriesDataPoint, + BudgetAlertDTO, + CostBreakdownDTO, + ExecutionTraceDTO, + TokenUsageDTO, + TopAgentDTO, +) +from src.schemas.api.analytics.schemas import ( + AgentExecutionCreate, + AgentExecutionUpdate, + BudgetAlertCreate, + BudgetAlertUpdate, + ExecutionTraceCreate, + TokenUsageCreate, +) + +__all__ = [ + # DTOs + "AgentExecutionDTO", + "AnalyticsOverviewDTO", + "AnalyticsTimeSeriesDataPoint", + "BudgetAlertDTO", + "CostBreakdownDTO", + "ExecutionTraceDTO", + "TokenUsageDTO", + "TopAgentDTO", + # Schemas + "AgentExecutionCreate", + "AgentExecutionUpdate", + "BudgetAlertCreate", + "BudgetAlertUpdate", + "ExecutionTraceCreate", + "TokenUsageCreate", +] diff --git a/backend/src/schemas/api/analytics/dto.py b/backend/src/schemas/api/analytics/dto.py new file mode 100644 index 000000000..8434b421e --- /dev/null +++ b/backend/src/schemas/api/analytics/dto.py @@ -0,0 +1,276 @@ +from datetime import datetime +from decimal import Decimal +from typing import List, Optional + +from pydantic import BaseModel, Field + +from src.utils.enums import ( + BudgetAlertType, + BudgetScope, + ExecutionStatus, + ExecutionTraceStepType, +) + + +class TokenUsageDTO(BaseModel): + """Token usage details for a single component.""" + + id: str + execution_id: str + component: str + step_number: Optional[int] = None + input_tokens: int = 0 + output_tokens: int = 0 + model: Optional[str] = None + cost_usd: Decimal = Decimal("0.000000") + created_at: datetime + + class Config: + from_attributes = True + + +class ExecutionTraceDTO(BaseModel): + """Step-by-step trace entry.""" + + id: str + execution_id: str + step_number: int + step_type: ExecutionTraceStepType + content: Optional[str] = None + invoked_agent_id: Optional[str] = None + invoked_agent_name: Optional[str] = None + timestamp: datetime + duration_ms: int = 0 + input_tokens: Optional[int] = 0 + output_tokens: Optional[int] = 0 + + class Config: + from_attributes = True + + +class AgentExecutionDTO(BaseModel): + """Detailed execution information.""" + + id: str + request_id: str + session_id: str + agent_id: Optional[str] = None + agent_type: Optional[str] = None + agent_name: Optional[str] = None + user_id: Optional[str] = None + model_config_id: Optional[str] = None + model_name: Optional[str] = None + started_at: datetime + completed_at: Optional[datetime] = None + status: ExecutionStatus + error_message: Optional[str] = None + error_type: Optional[str] = None + input_tokens: int = 0 + output_tokens: int = 0 + total_tokens: int = 0 + execution_time_ms: int = 0 + llm_time_ms: Optional[int] = 0 + cost_usd: Decimal = Decimal("0.000000") + parent_execution_id: Optional[str] = None + query_preview: Optional[str] = None + created_at: datetime + updated_at: datetime + + # Related data (optional, loaded on detail view) + traces: Optional[List[ExecutionTraceDTO]] = None + token_usages: Optional[List[TokenUsageDTO]] = None + + class Config: + from_attributes = True + + +class AgentExecutionSummaryDTO(BaseModel): + """Lightweight execution summary for list views.""" + + id: str + request_id: str + session_id: str + agent_name: Optional[str] = None + agent_type: Optional[str] = None + model_name: Optional[str] = None + status: ExecutionStatus + execution_time_ms: int = 0 + total_tokens: int = 0 + cost_usd: Decimal = Decimal("0.000000") + started_at: datetime + completed_at: Optional[datetime] = None + query_preview: Optional[str] = None + + class Config: + from_attributes = True + + +class BudgetAlertDTO(BaseModel): + """Budget alert configuration.""" + + id: str + user_id: str + scope: BudgetScope + scope_id: Optional[str] = None + threshold_usd: Decimal + period_days: int = 30 + alert_type: BudgetAlertType + alert_at_percentage: int = 80 + webhook_url: Optional[str] = None + email_notification: bool = True + is_active: bool = True + last_alert_sent_at: Optional[datetime] = None + current_spend_usd: Decimal = Decimal("0.00") + created_at: datetime + updated_at: datetime + + class Config: + from_attributes = True + + +class TopAgentDTO(BaseModel): + """Top agent by usage.""" + + agent_id: Optional[str] = None + agent_name: str + agent_type: Optional[str] = None + execution_count: int = 0 + success_count: int = 0 + failure_count: int = 0 + success_rate: float = 0.0 + avg_execution_time_ms: float = 0.0 + total_tokens: int = 0 + total_cost_usd: Decimal = Decimal("0.00") + + +class CostBreakdownDTO(BaseModel): + """Cost breakdown by model or agent.""" + + name: str + category: str # 'model' or 'agent' + total_cost_usd: Decimal = Decimal("0.00") + percentage: float = 0.0 + execution_count: int = 0 + total_tokens: int = 0 + + +class AnalyticsTimeSeriesDataPoint(BaseModel): + """Single data point for time series charts.""" + + timestamp: datetime + value: float = 0.0 + label: Optional[str] = None + + +class AnalyticsOverviewDTO(BaseModel): + """Dashboard overview statistics.""" + + # Summary metrics + total_executions: int = 0 + successful_executions: int = 0 + failed_executions: int = 0 + success_rate: float = 0.0 + + # Execution time metrics + avg_execution_time_ms: float = 0.0 + p50_execution_time_ms: float = 0.0 + p95_execution_time_ms: float = 0.0 + p99_execution_time_ms: float = 0.0 + + # Token metrics + total_input_tokens: int = 0 + total_output_tokens: int = 0 + total_tokens: int = 0 + + # Cost metrics + total_cost_usd: Decimal = Decimal("0.00") + estimated_monthly_cost_usd: Decimal = Decimal("0.00") + + # Comparison with previous period + executions_change_percent: Optional[float] = None + execution_time_change_percent: Optional[float] = None + success_rate_change_percent: Optional[float] = None + cost_change_percent: Optional[float] = None + + # Active resources + active_agents_count: int = 0 + + # Top agents + top_agents: List[TopAgentDTO] = Field(default_factory=list) + + # Cost breakdown + cost_by_model: List[CostBreakdownDTO] = Field(default_factory=list) + cost_by_agent: List[CostBreakdownDTO] = Field(default_factory=list) + + # Time series data + execution_time_trend: List[AnalyticsTimeSeriesDataPoint] = Field(default_factory=list) + executions_trend: List[AnalyticsTimeSeriesDataPoint] = Field(default_factory=list) + cost_trend: List[AnalyticsTimeSeriesDataPoint] = Field(default_factory=list) + + +class CostForecastDTO(BaseModel): + """Cost forecast based on current usage patterns.""" + + current_period_cost_usd: Decimal = Decimal("0.00") + projected_monthly_cost_usd: Decimal = Decimal("0.00") + projected_weekly_cost_usd: Decimal = Decimal("0.00") + daily_average_cost_usd: Decimal = Decimal("0.00") + + # Trend + cost_trend: str = "stable" # increasing, decreasing, stable + trend_percentage: float = 0.0 + + # Breakdown projections + by_model: List[CostBreakdownDTO] = Field(default_factory=list) + by_agent: List[CostBreakdownDTO] = Field(default_factory=list) + + +class AgentStatsDTO(BaseModel): + """Statistics for a specific agent.""" + + agent_id: Optional[str] = None + agent_name: str + agent_type: Optional[str] = None + + # Execution metrics + total_executions: int = 0 + successful_executions: int = 0 + failed_executions: int = 0 + timeout_executions: int = 0 + success_rate: float = 0.0 + + # Timing + avg_execution_time_ms: float = 0.0 + p50_execution_time_ms: float = 0.0 + p95_execution_time_ms: float = 0.0 + min_execution_time_ms: int = 0 + max_execution_time_ms: int = 0 + + # Tokens + avg_input_tokens: float = 0.0 + avg_output_tokens: float = 0.0 + total_tokens: int = 0 + + # Cost + total_cost_usd: Decimal = Decimal("0.00") + avg_cost_per_execution_usd: Decimal = Decimal("0.00") + + # Usage pattern + last_executed_at: Optional[datetime] = None + first_executed_at: Optional[datetime] = None + + # Error analysis + common_error_types: List[str] = Field(default_factory=list) + + # Time series + execution_trend: List[AnalyticsTimeSeriesDataPoint] = Field(default_factory=list) + + +class PaginatedExecutionsDTO(BaseModel): + """Paginated list of executions.""" + + items: List[AgentExecutionSummaryDTO] + total: int + page: int + page_size: int + total_pages: int diff --git a/backend/src/schemas/api/analytics/schemas.py b/backend/src/schemas/api/analytics/schemas.py new file mode 100644 index 000000000..bb13567ef --- /dev/null +++ b/backend/src/schemas/api/analytics/schemas.py @@ -0,0 +1,126 @@ +from datetime import datetime +from decimal import Decimal +from typing import Optional +from uuid import UUID + +from pydantic import BaseModel, Field + +from src.utils.enums import ( + BudgetAlertType, + BudgetScope, + ExecutionStatus, + ExecutionTraceStepType, +) + + +class AgentExecutionCreate(BaseModel): + """Create a new agent execution record.""" + + request_id: UUID + session_id: UUID + agent_id: Optional[UUID] = None + agent_type: Optional[str] = None + agent_name: Optional[str] = None + user_id: Optional[UUID] = None + model_config_id: Optional[UUID] = None + model_name: Optional[str] = None + parent_execution_id: Optional[UUID] = None + query_preview: Optional[str] = None + status: ExecutionStatus = ExecutionStatus.pending + + +class AgentExecutionUpdate(BaseModel): + """Update an agent execution record.""" + + completed_at: Optional[datetime] = None + status: Optional[ExecutionStatus] = None + error_message: Optional[str] = None + error_type: Optional[str] = None + input_tokens: Optional[int] = None + output_tokens: Optional[int] = None + total_tokens: Optional[int] = None + execution_time_ms: Optional[int] = None + llm_time_ms: Optional[int] = None + cost_usd: Optional[Decimal] = None + + +class TokenUsageCreate(BaseModel): + """Create a token usage record.""" + + execution_id: UUID + component: str + step_number: Optional[int] = None + input_tokens: int = 0 + output_tokens: int = 0 + model: Optional[str] = None + cost_usd: Decimal = Decimal("0.000000") + + +class ExecutionTraceCreate(BaseModel): + """Create an execution trace step.""" + + execution_id: UUID + step_number: int + step_type: ExecutionTraceStepType + content: Optional[str] = None + invoked_agent_id: Optional[UUID] = None + invoked_agent_name: Optional[str] = None + duration_ms: int = 0 + input_tokens: Optional[int] = 0 + output_tokens: Optional[int] = 0 + + +class BudgetAlertCreate(BaseModel): + """Create a budget alert.""" + + scope: BudgetScope = BudgetScope.user + scope_id: Optional[UUID] = None + threshold_usd: Decimal + period_days: int = 30 + alert_type: BudgetAlertType = BudgetAlertType.warning + alert_at_percentage: int = Field(default=80, ge=1, le=100) + webhook_url: Optional[str] = None + email_notification: bool = True + + +class BudgetAlertUpdate(BaseModel): + """Update a budget alert.""" + + threshold_usd: Optional[Decimal] = None + period_days: Optional[int] = None + alert_type: Optional[BudgetAlertType] = None + alert_at_percentage: Optional[int] = Field(default=None, ge=1, le=100) + webhook_url: Optional[str] = None + email_notification: Optional[bool] = None + is_active: Optional[bool] = None + + +class AnalyticsQueryParams(BaseModel): + """Query parameters for analytics endpoints.""" + + # Time range + start_date: Optional[datetime] = None + end_date: Optional[datetime] = None + period: Optional[str] = "24h" # 24h, 7d, 30d, custom + + # Filters + agent_id: Optional[UUID] = None + agent_type: Optional[str] = None + model_name: Optional[str] = None + status: Optional[ExecutionStatus] = None + + # Pagination + page: int = Field(default=1, ge=1) + page_size: int = Field(default=20, ge=1, le=100) + + # Sorting + sort_by: Optional[str] = "started_at" + sort_order: Optional[str] = "desc" # asc, desc + + +class ExportFormat(BaseModel): + """Export format specification.""" + + format: str = "json" # json, csv + include_traces: bool = False + include_token_details: bool = False diff --git a/backend/src/utils/enums.py b/backend/src/utils/enums.py index b64da1236..f1f34ee7b 100644 --- a/backend/src/utils/enums.py +++ b/backend/src/utils/enums.py @@ -34,3 +34,33 @@ class AgentIdType(Enum): agent_id = "agent_id" mcp_tool_id = "mcp_tool_id" a2a_card_id = "a2a_card_id" + + +class ExecutionStatus(Enum): + pending = "pending" + running = "running" + success = "success" + failure = "failure" + timeout = "timeout" + cancelled = "cancelled" + + +class ExecutionTraceStepType(Enum): + thought = "thought" + action = "action" + observation = "observation" + agent_invoke = "agent_invoke" + tool_call = "tool_call" + final_answer = "final_answer" + + +class BudgetAlertType(Enum): + warning = "warning" + hard_stop = "hard_stop" + + +class BudgetScope(Enum): + user = "user" + agent = "agent" + flow = "flow" + global_ = "global" diff --git a/docs/ANALYTICS.md b/docs/ANALYTICS.md new file mode 100644 index 000000000..5d6bcf0f1 --- /dev/null +++ b/docs/ANALYTICS.md @@ -0,0 +1,391 @@ +# Agent Analytics & Observability + +This document describes the Analytics and Observability features for GenAI AgentOS, providing comprehensive monitoring of agent executions, cost tracking, and performance insights. + +## Overview + +The Analytics module provides: +- **Real-time execution monitoring** - Track all agent invocations +- **Performance analytics** - Response times, success rates, throughput +- **Cost tracking** - Token usage and cost breakdown by model/agent +- **Query inspection** - Detailed ReAct loop visualization +- **Budget management** - Set spending thresholds and alerts + +## Screenshots + +### Analytics Dashboard +![Analytics Dashboard](./screenshots/01_LandingScreen.png) + +The main dashboard displays: +- **Total Executions** - Number of agent invocations in the selected period +- **Success Rate** - Percentage of successful executions +- **Avg Response Time** - Mean execution time with P95 percentile +- **Total Cost** - Cumulative cost with monthly projection +- **Total Tokens** - Input/output token breakdown +- **Active Agents** - Unique agents used in the period +- **Response Time Percentiles** - P50, P95, P99 latency metrics + +### Top Agents & Cost Breakdown +![Cost Breakdown](./screenshots/02_Detail1.png) + +- **Top Agents by Usage** - Table showing most-used agents with execution counts, success rates, average times, and costs +- **Cost by Model** - Visual breakdown of spending across different LLM models (GPT-4o, Claude, Gemini, etc.) + +### Recent Executions +![Execution Details](./screenshots/03_Detail2.png) + +- **Recent Executions Table** - List of recent agent invocations with: + - Query preview + - Agent name + - Status (success/failure/timeout) + - Duration + - Token count + - Cost + - Timestamp +- Click any execution to view detailed ReAct loop traces + +## Accessing Analytics + +Navigate to **Analytics** in the sidebar or visit `/analytics` directly. + +### Time Period Selection +Use the period selector (24h, 7d, 30d, 90d) in the top-right to adjust the analytics window. + +### Export Data +Click **Export CSV** to download analytics data for external analysis. + +--- + +## Integration Guide for Developers + +### Automatic Logging + +Agent executions are **automatically logged** when using the GenAI Agent framework. No additional code is required for basic tracking. + +### Adding Execution Tracking to Custom Agents + +If you're building custom agents outside the standard framework, you need to record executions using the analytics service. + +#### Python Example + +```python +from datetime import datetime +from uuid import uuid4 +from src.repositories.analytics import AgentExecutionRepository +from src.schemas.api.analytics.schemas import AgentExecutionCreate + +async def track_agent_execution( + db_session, + user_id: str, + agent_name: str, + agent_type: str, + model_name: str, + input_tokens: int, + output_tokens: int, + execution_time_ms: int, + status: str = "success", + error_message: str = None, + query_preview: str = None, +): + """Record an agent execution for analytics.""" + + repo = AgentExecutionRepository(db_session) + + execution = AgentExecutionCreate( + request_id=uuid4(), + session_id=uuid4(), + agent_name=agent_name, + agent_type=agent_type, + model_name=model_name, + status=status, + input_tokens=input_tokens, + output_tokens=output_tokens, + total_tokens=input_tokens + output_tokens, + execution_time_ms=execution_time_ms, + cost_usd=calculate_cost(model_name, input_tokens, output_tokens), + query_preview=query_preview[:100] if query_preview else None, + error_message=error_message, + ) + + return await repo.create(execution, user_id=user_id) +``` + +#### Recording Execution Traces (ReAct Loop) + +To enable detailed step-by-step visualization: + +```python +from src.repositories.analytics import ExecutionTraceRepository +from src.schemas.api.analytics.schemas import ExecutionTraceCreate + +async def record_trace_step( + db_session, + execution_id: str, + step_number: int, + step_type: str, # "thought", "action", "observation", "tool_call", "final_answer" + content: str, + duration_ms: int, + input_tokens: int = 0, + output_tokens: int = 0, +): + """Record a single step in the ReAct loop.""" + + repo = ExecutionTraceRepository(db_session) + + trace = ExecutionTraceCreate( + execution_id=execution_id, + step_number=step_number, + step_type=step_type, + content=content, + duration_ms=duration_ms, + input_tokens=input_tokens, + output_tokens=output_tokens, + ) + + return await repo.create(trace) +``` + +#### Recording Token Usage + +For detailed cost attribution: + +```python +from src.repositories.analytics import TokenUsageRepository +from src.schemas.api.analytics.schemas import TokenUsageCreate + +async def record_token_usage( + db_session, + execution_id: str, + component: str, # "main_agent", "sub_agent", "tool_call", etc. + model: str, + input_tokens: int, + output_tokens: int, + step_number: int = None, +): + """Record token usage for a specific component.""" + + repo = TokenUsageRepository(db_session) + + usage = TokenUsageCreate( + execution_id=execution_id, + component=component, + model=model, + input_tokens=input_tokens, + output_tokens=output_tokens, + cost_usd=calculate_cost(model, input_tokens, output_tokens), + step_number=step_number, + ) + + return await repo.create(usage) +``` + +### Middleware Hook (Recommended) + +For automatic tracking across all agents, add the analytics middleware: + +```python +# In your agent initialization or middleware setup +from src.middleware.analytics import AnalyticsMiddleware + +# Add to your FastAPI app or agent router +app.add_middleware(AnalyticsMiddleware) +``` + +### WebSocket Integration + +For real-time agent executions via WebSocket, analytics are automatically captured in the message handler: + +```python +# In src/routes/websocket.py - analytics hooks are already integrated +# When processing agent responses, execution data is automatically logged +``` + +--- + +## API Reference + +### Analytics Endpoints + +| Endpoint | Method | Description | +|----------|--------|-------------| +| `/api/analytics/overview` | GET | Dashboard summary statistics | +| `/api/analytics/executions` | GET | Paginated list of executions | +| `/api/analytics/executions/{id}` | GET | Detailed execution with traces | +| `/api/analytics/agents/{id}/stats` | GET | Per-agent statistics | +| `/api/analytics/costs/summary` | GET | Cost breakdown by model/agent | +| `/api/analytics/costs/forecast` | GET | Projected monthly costs | +| `/api/analytics/budgets` | GET/POST | List/create budget alerts | +| `/api/analytics/budgets/{id}` | PATCH/DELETE | Update/delete budget alert | +| `/api/analytics/comparison/models` | GET | Model performance comparison | +| `/api/analytics/export` | GET | Export data (JSON/CSV) | + +### Query Parameters + +#### Overview Endpoint +``` +GET /api/analytics/overview?period=7d +GET /api/analytics/overview?start_date=2024-01-01&end_date=2024-01-31 +``` + +#### Executions List +``` +GET /api/analytics/executions?page=1&page_size=20 +GET /api/analytics/executions?status=success +GET /api/analytics/executions?agent_id= +GET /api/analytics/executions?sort_by=started_at&sort_order=desc +``` + +#### Export +``` +GET /api/analytics/export?format=json +GET /api/analytics/export?format=csv&period=30d +``` + +--- + +## Budget Alerts + +### Creating a Budget Alert + +```bash +curl -X POST /api/analytics/budgets \ + -H "Authorization: Bearer " \ + -H "Content-Type: application/json" \ + -d '{ + "threshold_usd": 100.00, + "period_days": 30, + "alert_type": "warning", + "alert_at_percentage": 80, + "email_notification": true + }' +``` + +### Alert Types +- **warning** - Notification only, execution continues +- **hard_stop** - Blocks further executions when threshold reached + +### Budget Scopes +- **user** - Per-user spending limit +- **agent** - Per-agent spending limit +- **flow** - Per-workflow spending limit +- **global** - Organization-wide limit + +--- + +## Database Schema + +### Core Tables + +```sql +-- Agent executions +CREATE TABLE agentexecutions ( + id UUID PRIMARY KEY, + user_id UUID REFERENCES users(id), + request_id UUID NOT NULL, + session_id UUID NOT NULL, + agent_id UUID, + agent_name VARCHAR, + agent_type VARCHAR, + model_name VARCHAR, + status executionstatus NOT NULL, + input_tokens INTEGER DEFAULT 0, + output_tokens INTEGER DEFAULT 0, + total_tokens INTEGER DEFAULT 0, + cost_usd NUMERIC(10,6) DEFAULT 0, + execution_time_ms INTEGER DEFAULT 0, + llm_time_ms INTEGER DEFAULT 0, + query_preview VARCHAR, + error_type VARCHAR, + error_message VARCHAR, + started_at TIMESTAMP DEFAULT NOW(), + completed_at TIMESTAMP +); + +-- Execution traces (ReAct loop steps) +CREATE TABLE executiontraces ( + id UUID PRIMARY KEY, + execution_id UUID REFERENCES agentexecutions(id) ON DELETE CASCADE, + step_number INTEGER NOT NULL, + step_type executiontracesteptype NOT NULL, + content VARCHAR, + input_tokens INTEGER DEFAULT 0, + output_tokens INTEGER DEFAULT 0, + duration_ms INTEGER DEFAULT 0, + timestamp TIMESTAMP DEFAULT NOW() +); + +-- Token usage breakdown +CREATE TABLE tokenusages ( + id UUID PRIMARY KEY, + execution_id UUID REFERENCES agentexecutions(id) ON DELETE CASCADE, + component VARCHAR, + model VARCHAR, + input_tokens INTEGER NOT NULL, + output_tokens INTEGER NOT NULL, + cost_usd NUMERIC(10,6) NOT NULL, + step_number INTEGER +); + +-- Budget alerts +CREATE TABLE budgetalerts ( + id UUID PRIMARY KEY, + user_id UUID REFERENCES users(id) ON DELETE CASCADE, + scope budgetscope NOT NULL, + threshold_usd NUMERIC(10,2) NOT NULL, + period_days INTEGER DEFAULT 30, + alert_type budgetalerttype NOT NULL, + alert_at_percentage INTEGER DEFAULT 80, + is_active BOOLEAN DEFAULT TRUE, + email_notification BOOLEAN DEFAULT TRUE, + current_spend_usd NUMERIC(10,2) DEFAULT 0 +); +``` + +--- + +## Seeding Test Data + +For development and testing, you can populate the analytics tables with synthetic data: + +```bash +# From the backend directory +python scripts/seed_analytics_data.py + +# Or via Docker +docker exec genai-backend python -m scripts.seed_analytics_data +``` + +This creates: +- 250 sample agent executions across 90 days +- 750+ token usage records +- 1000+ execution trace steps +- 3 sample budget alerts + +--- + +## Troubleshooting + +### No Data Showing in Dashboard + +1. **Check user association** - Analytics are user-scoped. Ensure you're logged in as the user who ran the agents. + +2. **Verify data exists**: + ```bash + docker exec genai-postgres psql -U postgres -d postgres \ + -c "SELECT COUNT(*) FROM agentexecutions WHERE user_id = '';" + ``` + +3. **Check time period** - Ensure the selected period (24h, 7d, etc.) includes your execution data. + +### Executions Not Being Logged + +1. Ensure the agent is using the standard GenAI framework +2. Check that the database connection is working +3. Verify the analytics repository is being called in your agent code + +### Cost Calculations Incorrect + +Cost is calculated based on model pricing. Update pricing in: +``` +backend/src/utils/model_pricing.py +``` \ No newline at end of file diff --git a/docs/screenshots/01_LandingScreen.png b/docs/screenshots/01_LandingScreen.png new file mode 100644 index 000000000..29867d2ff Binary files /dev/null and b/docs/screenshots/01_LandingScreen.png differ diff --git a/docs/screenshots/02_Detail1.png b/docs/screenshots/02_Detail1.png new file mode 100644 index 000000000..5c91d4946 Binary files /dev/null and b/docs/screenshots/02_Detail1.png differ diff --git a/docs/screenshots/03_Detail2.png b/docs/screenshots/03_Detail2.png new file mode 100644 index 000000000..7fbb2c3ee Binary files /dev/null and b/docs/screenshots/03_Detail2.png differ diff --git a/frontend/src/assets/icons/analytics.svg b/frontend/src/assets/icons/analytics.svg new file mode 100644 index 000000000..fe1a8dbc3 --- /dev/null +++ b/frontend/src/assets/icons/analytics.svg @@ -0,0 +1,5 @@ + + + + + diff --git a/frontend/src/components/layout/Sidebar.tsx b/frontend/src/components/layout/Sidebar.tsx index 7a49ff015..6847afc3a 100644 --- a/frontend/src/components/layout/Sidebar.tsx +++ b/frontend/src/components/layout/Sidebar.tsx @@ -10,6 +10,7 @@ import A2AAgentsIcon from '@/assets/icons/a2a.svg'; import MCPAgentsIcon from '@/assets/icons/mcp.svg'; import FlowsIcon from '@/assets/icons/tree.svg'; import NewFlowIcon from '@/assets/icons/new-flow.svg'; +import AnalyticsIcon from '@/assets/icons/analytics.svg'; interface SidebarProps { collapsed: boolean; @@ -38,6 +39,11 @@ const pages: PageLink[] = [ title: 'New Agent Flow', Icon: NewFlowIcon, }, + { + path: '/analytics', + title: 'Analytics', + Icon: AnalyticsIcon, + }, ]; const Sidebar: FC = memo(({ collapsed, setCollapsed }) => { diff --git a/frontend/src/pages/AnalyticsPage.tsx b/frontend/src/pages/AnalyticsPage.tsx new file mode 100644 index 000000000..8757a57df --- /dev/null +++ b/frontend/src/pages/AnalyticsPage.tsx @@ -0,0 +1,563 @@ +import { useState, useEffect, useCallback } from 'react'; +import type { FC } from 'react'; +import { useNavigate } from 'react-router-dom'; +import { + TrendingUp, + TrendingDown, + Minus, + Clock, + CheckCircle, + XCircle, + Coins, + Activity, + Download, + RefreshCw, +} from 'lucide-react'; + +import { MainLayout } from '@/components/layout/MainLayout'; +import { Button } from '@/components/ui/button'; +import { Tabs, TabsList, TabsTrigger } from '@/components/ui/tabs'; +import { analyticsService, AnalyticsQueryParams } from '@/services/analyticsService'; +import { + AnalyticsOverview, + PaginatedExecutions, + TopAgent, + CostBreakdown, + AgentExecutionSummary, +} from '@/types/analytics'; + +type Period = '24h' | '7d' | '30d' | '90d'; + +interface StatCardProps { + title: string; + value: string | number; + change?: number | null; + icon: React.ReactNode; + trend?: 'up' | 'down' | 'neutral'; + subtitle?: string; +} + +const StatCard: FC = ({ title, value, change, icon, trend, subtitle }) => { + const getTrendIcon = () => { + if (trend === 'up') return ; + if (trend === 'down') return ; + return ; + }; + + const getChangeColor = () => { + if (change === null || change === undefined) return 'text-gray-500'; + if (change > 0) return 'text-green-500'; + if (change < 0) return 'text-red-500'; + return 'text-gray-500'; + }; + + return ( +
+
+ {title} +
{icon}
+
+
+
+

{value}

+ {subtitle &&

{subtitle}

} +
+ {change !== null && change !== undefined && ( +
+ {getTrendIcon()} + + {change > 0 ? '+' : ''} + {change.toFixed(1)}% + +
+ )} +
+
+ ); +}; + +interface TopAgentsTableProps { + agents: TopAgent[]; +} + +const TopAgentsTable: FC = ({ agents }) => { + return ( +
+
+

Top Agents by Usage

+
+
+ + + + + + + + + + + + {agents.length === 0 ? ( + + + + ) : ( + agents.map((agent, index) => ( + + + + + + + + )) + )} + +
+ Agent + + Executions + + Success Rate + + Avg Time + + Cost +
+ No agent data available +
+
+
+ {index + 1} +
+
+

{agent.agent_name}

+

{agent.agent_type || 'unknown'}

+
+
+ 		+ {agent.execution_count.toLocaleString()} + + = 90 + ? 'bg-green-100 text-green-800' + : agent.success_rate >= 70 + ? 'bg-yellow-100 text-yellow-800' + : 'bg-red-100 text-red-800' + }`} + > + {agent.success_rate.toFixed(1)}% + + + {(agent.avg_execution_time_ms / 1000).toFixed(2)}s + + ${parseFloat(agent.total_cost_usd).toFixed(4)} +
+
+
+ ); +}; + +interface CostBreakdownChartProps { + data: CostBreakdown[]; + title: string; +} + +const CostBreakdownChart: FC = ({ data, title }) => { + const colors = [ + 'bg-blue-500', + 'bg-green-500', + 'bg-yellow-500', + 'bg-purple-500', + 'bg-pink-500', + 'bg-indigo-500', + 'bg-orange-500', + ]; + + return ( +
+

{title}

+ {data.length === 0 ? ( +

No cost data available

+ ) : ( +
+ {data.slice(0, 5).map((item, index) => ( +
+
+ {item.name} + + ${parseFloat(item.total_cost_usd).toFixed(4)} ({item.percentage.toFixed(1)}%) + +
+
+
+
+
+ ))} +
+ )} +
+ ); +}; + +interface RecentExecutionsTableProps { + executions: AgentExecutionSummary[]; + onViewDetails: (id: string) => void; +} + +const RecentExecutionsTable: FC = ({ + executions, + onViewDetails, +}) => { + const getStatusBadge = (status: string) => { + const statusConfig: Record = { + success: { bg: 'bg-green-100', text: 'text-green-800' }, + failure: { bg: 'bg-red-100', text: 'text-red-800' }, + pending: { bg: 'bg-yellow-100', text: 'text-yellow-800' }, + running: { bg: 'bg-blue-100', text: 'text-blue-800' }, + timeout: { bg: 'bg-orange-100', text: 'text-orange-800' }, + cancelled: { bg: 'bg-gray-100', text: 'text-gray-800' }, + }; + const config = statusConfig[status] || statusConfig.pending; + return ( + + {status} + + ); + }; + + return ( +
+
+

Recent Executions

+
+
+ + + + + + + + + + + + + + {executions.length === 0 ? ( + + + + ) : ( + executions.map((execution) => ( + onViewDetails(execution.id)} + > + + + + + + + + + )) + )} + +
+ Query + + Agent + + Status + + Duration + + Tokens + + Cost + + Time +
+ No executions found +
+

+ {execution.query_preview || 'No query preview'} +

+ 		+

{execution.agent_name || 'Unknown'}

+

{execution.model_name || 'No model'}

+ 		+ {getStatusBadge(execution.status)} + + {(execution.execution_time_ms / 1000).toFixed(2)}s + + {execution.total_tokens.toLocaleString()} + + ${parseFloat(execution.cost_usd).toFixed(4)} + + {new Date(execution.started_at).toLocaleTimeString()} +
+
+
+ ); +}; + +const AnalyticsPage: FC = () => { + const navigate = useNavigate(); + const [period, setPeriod] = useState('24h'); + const [overview, setOverview] = useState(null); + const [executions, setExecutions] = useState(null); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(null); + + const fetchData = useCallback(async () => { + setLoading(true); + setError(null); + try { + const params: AnalyticsQueryParams = { period }; + const [overviewData, executionsData] = await Promise.all([ + analyticsService.getOverview(params), + analyticsService.getExecutions({ ...params, page_size: 10 }), + ]); + setOverview(overviewData); + setExecutions(executionsData); + } catch (err) { + setError('Failed to load analytics data'); + console.error('Analytics error:', err); + } finally { + setLoading(false); + } + }, [period]); + + useEffect(() => { + fetchData(); + }, [fetchData]); + + const handleExport = async (format: 'json' | 'csv') => { + try { + const blob = await analyticsService.exportData(format, { period }); + const url = window.URL.createObjectURL(blob); + const a = document.createElement('a'); + a.href = url; + a.download = `analytics_export.${format}`; + document.body.appendChild(a); + a.click(); + window.URL.revokeObjectURL(url); + document.body.removeChild(a); + } catch (err) { + console.error('Export failed:', err); + } + }; + + const handleViewDetails = (executionId: string) => { + // Navigate to execution details + navigate(`/analytics/executions/${executionId}`); + }; + + const formatNumber = (num: number): string => { + if (num >= 1000000) return `${(num / 1000000).toFixed(1)}M`; + if (num >= 1000) return `${(num / 1000).toFixed(1)}K`; + return num.toString(); + }; + + const formatTime = (ms: number): string => { + if (ms >= 1000) return `${(ms / 1000).toFixed(2)}s`; + return `${ms}ms`; + }; + + const actionItems = ( +
+ + +
+ ); + + return ( + +
+ {/* Period Selector */} +
+
+

Agent Analytics

+

Monitor performance, costs, and usage patterns

+
+ setPeriod(v as Period)}> + + 24h + 7d + 30d + 90d + + +
+ + {loading && ( +
+
+
+ )} + + {error && ( +
+ {error} +
+ )} + + {!loading && !error && overview && ( + <> + {/* Summary Stats */} +
+ 0 + ? 'up' + : 'down' + : 'neutral' + } + icon={} + subtitle={`${overview.successful_executions} successful`} + /> + 0 + ? 'up' + : 'down' + : 'neutral' + } + icon={} + subtitle={`${overview.failed_executions} failures`} + /> + } + subtitle={`P95: ${formatTime(overview.p95_execution_time_ms)}`} + /> + 0 + ? 'up' + : 'down' + : 'neutral' + } + icon={} + subtitle={`Est. monthly: $${parseFloat(overview.estimated_monthly_cost_usd).toFixed(2)}`} + /> +
+ + {/* Token Usage Stats */} +
+
+

Total Tokens

+

+ {formatNumber(overview.total_tokens)} +

+
+
+ Input:{' '} + + {formatNumber(overview.total_input_tokens)} + +
+
+ Output:{' '} + + {formatNumber(overview.total_output_tokens)} + +
+
+
+
+

Active Agents

+

+ {overview.active_agents_count} +

+

Unique agents used in this period

+
+
+

Response Time Percentiles

+
+
+ P50 + + {formatTime(overview.p50_execution_time_ms)} + +
+
+ P95 + + {formatTime(overview.p95_execution_time_ms)} + +
+
+ P99 + + {formatTime(overview.p99_execution_time_ms)} + +
+
+
+
+ + {/* Top Agents and Cost Breakdown */} +
+ + +
+ + {/* Recent Executions */} + {executions && ( + + )} + + )} +
+ + ); +}; + +export default AnalyticsPage; diff --git a/frontend/src/pages/ExecutionDetailsPage.tsx b/frontend/src/pages/ExecutionDetailsPage.tsx new file mode 100644 index 000000000..5864becba --- /dev/null +++ b/frontend/src/pages/ExecutionDetailsPage.tsx @@ -0,0 +1,435 @@ +import { useState, useEffect } from 'react'; +import type { FC } from 'react'; +import { useParams, useNavigate } from 'react-router-dom'; +import { + ArrowLeft, + Clock, + Coins, + CheckCircle, + XCircle, + AlertCircle, + Brain, + Zap, + Eye, + MessageSquare, + Bot, +} from 'lucide-react'; + +import { MainLayout } from '@/components/layout/MainLayout'; +import { Button } from '@/components/ui/button'; +import { Tabs, TabsContent, TabsList, TabsTrigger } from '@/components/ui/tabs'; +import { analyticsService } from '@/services/analyticsService'; +import { + AgentExecution, + ExecutionTrace, + TokenUsage, + ExecutionTraceStepType, +} from '@/types/analytics'; + +interface TimelineStepProps { + step: ExecutionTrace; + isLast: boolean; +} + +const TimelineStep: FC = ({ step, isLast }) => { + const getStepIcon = (stepType: ExecutionTraceStepType) => { + switch (stepType) { + case 'thought': + return ; + case 'action': + return ; + case 'observation': + return ; + case 'agent_invoke': + return ; + case 'tool_call': + return ; + case 'final_answer': + return ; + default: + return ; + } + }; + + const getStepColor = (stepType: ExecutionTraceStepType) => { + switch (stepType) { + case 'thought': + return 'bg-blue-100 text-blue-700 border-blue-200'; + case 'action': + return 'bg-purple-100 text-purple-700 border-purple-200'; + case 'observation': + return 'bg-green-100 text-green-700 border-green-200'; + case 'agent_invoke': + return 'bg-orange-100 text-orange-700 border-orange-200'; + case 'tool_call': + return 'bg-yellow-100 text-yellow-700 border-yellow-200'; + case 'final_answer': + return 'bg-emerald-100 text-emerald-700 border-emerald-200'; + default: + return 'bg-gray-100 text-gray-700 border-gray-200'; + } + }; + + return ( +
+ {/* Timeline line */} +
+
+ {getStepIcon(step.step_type)} +
+ {!isLast &&
} +
+ + {/* Step content */} +
+
+ + {step.step_type.replace('_', ' ').toUpperCase()} + + Step {step.step_number} + | + {step.duration_ms}ms + {step.input_tokens !== undefined && step.output_tokens !== undefined && ( + <> + | + + {step.input_tokens} / {step.output_tokens} tokens + + + )} +
+ + {step.invoked_agent_name && ( +
+ Invoked: {step.invoked_agent_name} +
+ )} + + {step.content && ( +
+ {step.content} +
+ )} + +
+ {new Date(step.timestamp).toLocaleString()} +
+
+
+ ); +}; + +interface TokenUsageTableProps { + tokenUsages: TokenUsage[]; +} + +const TokenUsageTable: FC = ({ tokenUsages }) => { + const totalInput = tokenUsages.reduce((sum, t) => sum + t.input_tokens, 0); + const totalOutput = tokenUsages.reduce((sum, t) => sum + t.output_tokens, 0); + const totalCost = tokenUsages.reduce((sum, t) => sum + parseFloat(t.cost_usd), 0); + + return ( +
+
+

Token Usage Breakdown

+
+
+ + + + + + + + + + + + {tokenUsages.map((usage) => ( + + + + + + + + ))} + + + + + + + +
+ Component + + Model + + Input Tokens + + Output Tokens + + Cost +
+ {usage.component} + {usage.step_number != null && ( + (step {usage.step_number}) + )} + + {usage.model || 'N/A'} + + {usage.input_tokens.toLocaleString()} + + {usage.output_tokens.toLocaleString()} + + ${parseFloat(usage.cost_usd).toFixed(6)} +
+ Total + + {totalInput.toLocaleString()} + + {totalOutput.toLocaleString()} + + ${totalCost.toFixed(6)} +
+
+
+ ); +}; + +const ExecutionDetailsPage: FC = () => { + const { id } = useParams<{ id: string }>(); + const navigate = useNavigate(); + const [execution, setExecution] = useState(null); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(null); + + useEffect(() => { + const fetchExecution = async () => { + if (!id) return; + + setLoading(true); + setError(null); + try { + const data = await analyticsService.getExecutionDetails(id); + setExecution(data); + } catch (err) { + setError('Failed to load execution details'); + console.error('Error loading execution:', err); + } finally { + setLoading(false); + } + }; + + fetchExecution(); + }, [id]); + + const getStatusBadge = (status: string) => { + const statusConfig: Record< + string, + { bg: string; text: string; icon: React.ReactNode } + > = { + success: { + bg: 'bg-green-100', + text: 'text-green-800', + icon: , + }, + failure: { + bg: 'bg-red-100', + text: 'text-red-800', + icon: , + }, + pending: { + bg: 'bg-yellow-100', + text: 'text-yellow-800', + icon: , + }, + running: { + bg: 'bg-blue-100', + text: 'text-blue-800', + icon: , + }, + timeout: { + bg: 'bg-orange-100', + text: 'text-orange-800', + icon: , + }, + cancelled: { + bg: 'bg-gray-100', + text: 'text-gray-800', + icon: , + }, + }; + const config = statusConfig[status] || statusConfig.pending; + return ( + + {config.icon} + {status.charAt(0).toUpperCase() + status.slice(1)} + + ); + }; + + return ( + +
+ {/* Header */} +
+ +
+ + {loading && ( +
+
+
+ )} + + {error && ( +
+ {error} +
+ )} + + {!loading && !error && execution && ( + <> + {/* Execution Header */} +
+
+
+

+ {execution.query_preview || 'Execution Details'} +

+
+ Request: {execution.request_id.slice(0, 8)}... + Session: {execution.session_id.slice(0, 8)}... + {new Date(execution.started_at).toLocaleString()} +
+
+ {getStatusBadge(execution.status)} +
+ + {/* Summary Stats */} +
+
+
+ + Agent +
+

+ {execution.agent_name || 'Unknown'} +

+

{execution.agent_type || 'N/A'}

+
+ +
+
+ + Duration +
+

+ {(execution.execution_time_ms / 1000).toFixed(2)}s +

+ {execution.llm_time_ms && ( +

+ LLM: {(execution.llm_time_ms / 1000).toFixed(2)}s +

+ )} +
+ +
+
+ + Tokens +
+

+ {execution.total_tokens.toLocaleString()} +

+

+ In: {execution.input_tokens.toLocaleString()} / Out:{' '} + {execution.output_tokens.toLocaleString()} +

+
+ +
+
+ + Cost +
+

+ ${parseFloat(execution.cost_usd).toFixed(6)} +

+

{execution.model_name || 'N/A'}

+
+
+ + {/* Error Message */} + {execution.error_message && ( +
+
+ + Error: {execution.error_type || 'Unknown'} +
+

{execution.error_message}

+
+ )} +
+ + {/* Tabs */} + + + Execution Timeline + Token Usage + + + +
+

+ ReAct Loop Timeline +

+ {execution.traces && execution.traces.length > 0 ? ( +
+ {execution.traces + .sort((a, b) => a.step_number - b.step_number) + .map((trace, index) => ( + + ))} +
+ ) : ( +

+ No trace data available for this execution +

+ )} +
+ + + + {execution.token_usages && execution.token_usages.length > 0 ? ( + + ) : ( +
+ No token usage data available +
+ )} + + + + )} +
+ + ); +}; + +export default ExecutionDetailsPage; diff --git a/frontend/src/router.tsx b/frontend/src/router.tsx index 20bb84d29..ac3d3c409 100644 --- a/frontend/src/router.tsx +++ b/frontend/src/router.tsx @@ -14,6 +14,8 @@ const AgentsPage = lazy(() => import('./pages/AgentsPage')); const AgentFlowsPage = lazy(() => import('./pages/AgentFlowsPage')); const AgentFlowsEditPage = lazy(() => import('./pages/AgentFlowsEditPage')); const AgentsTracePage = lazy(() => import('./pages/AgentsTracePage')); +const AnalyticsPage = lazy(() => import('./pages/AnalyticsPage')); +const ExecutionDetailsPage = lazy(() => import('./pages/ExecutionDetailsPage')); const NotFoundPage = lazy(() => import('./pages/NotFoundPage')); // Loading component @@ -145,6 +147,26 @@ export const router = createBrowserRouter( ), }, + { + path: '/analytics', + element: ( + + }> + + + + ), + }, + { + path: '/analytics/executions/:id', + element: ( + + }> + + + + ), + }, { path: '*', element: ( diff --git a/frontend/src/services/analyticsService.ts b/frontend/src/services/analyticsService.ts new file mode 100644 index 000000000..5790c8437 --- /dev/null +++ b/frontend/src/services/analyticsService.ts @@ -0,0 +1,202 @@ +import { apiService } from './apiService'; +import { + AnalyticsOverview, + PaginatedExecutions, + AgentExecution, + AgentStats, + CostSummary, + CostForecast, + BudgetAlert, + BudgetAlertCreate, + BudgetAlertUpdate, + ModelComparison, +} from '../types/analytics'; + +export interface AnalyticsQueryParams { + period?: '24h' | '7d' | '30d' | '90d'; + start_date?: string; + end_date?: string; + agent_id?: string; + agent_type?: string; + model_name?: string; + status?: string; + page?: number; + page_size?: number; + sort_by?: string; + sort_order?: 'asc' | 'desc'; +} + +export const analyticsService = { + /** + * Get analytics overview for the dashboard + */ + async getOverview(params: AnalyticsQueryParams = {}): Promise { + const queryParams: Record = {}; + if (params.period) queryParams.period = params.period; + if (params.start_date) queryParams.start_date = params.start_date; + if (params.end_date) queryParams.end_date = params.end_date; + + const response = await apiService.get('/api/analytics/overview', { + params: queryParams, + }); + return response.data; + }, + + /** + * Get paginated list of executions + */ + async getExecutions(params: AnalyticsQueryParams = {}): Promise { + const queryParams: Record = {}; + if (params.period) queryParams.period = params.period; + if (params.start_date) queryParams.start_date = params.start_date; + if (params.end_date) queryParams.end_date = params.end_date; + if (params.agent_id) queryParams.agent_id = params.agent_id; + if (params.agent_type) queryParams.agent_type = params.agent_type; + if (params.model_name) queryParams.model_name = params.model_name; + if (params.status) queryParams.status = params.status; + if (params.page) queryParams.page = params.page.toString(); + if (params.page_size) queryParams.page_size = params.page_size.toString(); + if (params.sort_by) queryParams.sort_by = params.sort_by; + if (params.sort_order) queryParams.sort_order = params.sort_order; + + const response = await apiService.get('/api/analytics/executions', { + params: queryParams, + }); + return response.data; + }, + + /** + * Get detailed information about a specific execution + */ + async getExecutionDetails(executionId: string): Promise { + const response = await apiService.get( + `/api/analytics/executions/${executionId}`, + ); + return response.data; + }, + + /** + * Get statistics for a specific agent + */ + async getAgentStats( + agentId: string, + params: AnalyticsQueryParams = {}, + ): Promise { + const queryParams: Record = {}; + if (params.period) queryParams.period = params.period; + if (params.start_date) queryParams.start_date = params.start_date; + if (params.end_date) queryParams.end_date = params.end_date; + + const response = await apiService.get( + `/api/analytics/agents/${agentId}/stats`, + { params: queryParams }, + ); + return response.data; + }, + + /** + * Get cost breakdown summary + */ + async getCostSummary(params: AnalyticsQueryParams = {}): Promise { + const queryParams: Record = {}; + if (params.period) queryParams.period = params.period; + if (params.start_date) queryParams.start_date = params.start_date; + if (params.end_date) queryParams.end_date = params.end_date; + + const response = await apiService.get('/api/analytics/costs/summary', { + params: queryParams, + }); + return response.data; + }, + + /** + * Get cost forecast + */ + async getCostForecast(): Promise { + const response = await apiService.get('/api/analytics/costs/forecast'); + return response.data; + }, + + /** + * Get all budget alerts + */ + async getBudgetAlerts(): Promise { + const response = await apiService.get('/api/analytics/budgets'); + return response.data; + }, + + /** + * Create a new budget alert + */ + async createBudgetAlert(alert: BudgetAlertCreate): Promise { + const response = await apiService.post('/api/analytics/budgets', alert); + return response.data; + }, + + /** + * Update a budget alert + */ + async updateBudgetAlert(alertId: string, update: BudgetAlertUpdate): Promise { + const response = await apiService.patch( + `/api/analytics/budgets/${alertId}`, + update, + ); + return response.data; + }, + + /** + * Delete a budget alert + */ + async deleteBudgetAlert(alertId: string): Promise { + await apiService.delete(`/api/analytics/budgets/${alertId}`); + }, + + /** + * Compare model performance + */ + async compareModels(params: AnalyticsQueryParams = {}): Promise { + const queryParams: Record = {}; + if (params.period) queryParams.period = params.period; + if (params.start_date) queryParams.start_date = params.start_date; + if (params.end_date) queryParams.end_date = params.end_date; + + const response = await apiService.get( + '/api/analytics/comparison/models', + { params: queryParams }, + ); + return response.data; + }, + + /** + * Export analytics data + */ + async exportData( + format: 'json' | 'csv' = 'json', + params: AnalyticsQueryParams = {}, + ): Promise { + const queryParams: Record = { format }; + if (params.period) queryParams.period = params.period; + if (params.start_date) queryParams.start_date = params.start_date; + if (params.end_date) queryParams.end_date = params.end_date; + + const token = localStorage.getItem('token'); + if (!token) { + throw new Error('Authentication token is missing. Please log in again.'); + } + + const response = await fetch( + `/api/analytics/export?${new URLSearchParams(queryParams).toString()}`, + { + headers: { + Authorization: `Bearer ${token}`, + }, + }, + ); + + if (!response.ok) { + throw new Error('Export failed'); + } + + return response.blob(); + }, +}; diff --git a/frontend/src/types/analytics.ts b/frontend/src/types/analytics.ts new file mode 100644 index 000000000..c55f18cd9 --- /dev/null +++ b/frontend/src/types/analytics.ts @@ -0,0 +1,275 @@ +// Execution status types +export type ExecutionStatus = + | 'pending' + | 'running' + | 'success' + | 'failure' + | 'timeout' + | 'cancelled'; + +// Execution trace step types +export type ExecutionTraceStepType = + | 'thought' + | 'action' + | 'observation' + | 'agent_invoke' + | 'tool_call' + | 'final_answer'; + +// Budget types +export type BudgetAlertType = 'warning' | 'hard_stop'; +export type BudgetScope = 'user' | 'agent' | 'flow' | 'global'; + +// Token usage for a single component +export interface TokenUsage { + id: string; + execution_id: string; + component: string; + step_number?: number; + input_tokens: number; + output_tokens: number; + model?: string; + cost_usd: string; + created_at: string; +} + +// Execution trace step +export interface ExecutionTrace { + id: string; + execution_id: string; + step_number: number; + step_type: ExecutionTraceStepType; + content?: string; + invoked_agent_id?: string; + invoked_agent_name?: string; + timestamp: string; + duration_ms: number; + input_tokens?: number; + output_tokens?: number; +} + +// Full execution details +export interface AgentExecution { + id: string; + request_id: string; + session_id: string; + agent_id?: string; + agent_type?: string; + agent_name?: string; + user_id?: string; + model_config_id?: string; + model_name?: string; + started_at: string; + completed_at?: string; + status: ExecutionStatus; + error_message?: string; + error_type?: string; + input_tokens: number; + output_tokens: number; + total_tokens: number; + execution_time_ms: number; + llm_time_ms?: number; + cost_usd: string; + parent_execution_id?: string; + query_preview?: string; + created_at: string; + updated_at: string; + traces?: ExecutionTrace[]; + token_usages?: TokenUsage[]; +} + +// Lightweight execution summary for list views +export interface AgentExecutionSummary { + id: string; + request_id: string; + session_id: string; + agent_name?: string; + agent_type?: string; + model_name?: string; + status: ExecutionStatus; + execution_time_ms: number; + total_tokens: number; + cost_usd: string; + started_at: string; + completed_at?: string; + query_preview?: string; +} + +// Paginated executions response +export interface PaginatedExecutions { + items: AgentExecutionSummary[]; + total: number; + page: number; + page_size: number; + total_pages: number; +} + +// Top agent by usage +export interface TopAgent { + agent_id?: string; + agent_name: string; + agent_type?: string; + execution_count: number; + success_count: number; + failure_count: number; + success_rate: number; + avg_execution_time_ms: number; + total_tokens: number; + total_cost_usd: string; +} + +// Cost breakdown +export interface CostBreakdown { + name: string; + category: string; + total_cost_usd: string; + percentage: number; + execution_count: number; + total_tokens: number; +} + +// Time series data point +export interface TimeSeriesDataPoint { + timestamp: string; + value: number; + label?: string; +} + +// Analytics overview for dashboard +export interface AnalyticsOverview { + total_executions: number; + successful_executions: number; + failed_executions: number; + success_rate: number; + avg_execution_time_ms: number; + p50_execution_time_ms: number; + p95_execution_time_ms: number; + p99_execution_time_ms: number; + total_input_tokens: number; + total_output_tokens: number; + total_tokens: number; + total_cost_usd: string; + estimated_monthly_cost_usd: string; + executions_change_percent?: number; + execution_time_change_percent?: number; + success_rate_change_percent?: number; + cost_change_percent?: number; + active_agents_count: number; + top_agents: TopAgent[]; + cost_by_model: CostBreakdown[]; + cost_by_agent: CostBreakdown[]; + execution_time_trend: TimeSeriesDataPoint[]; + executions_trend: TimeSeriesDataPoint[]; + cost_trend: TimeSeriesDataPoint[]; +} + +// Agent statistics +export interface AgentStats { + agent_id?: string; + agent_name: string; + agent_type?: string; + total_executions: number; + successful_executions: number; + failed_executions: number; + timeout_executions: number; + success_rate: number; + avg_execution_time_ms: number; + p50_execution_time_ms: number; + p95_execution_time_ms: number; + min_execution_time_ms: number; + max_execution_time_ms: number; + avg_input_tokens: number; + avg_output_tokens: number; + total_tokens: number; + total_cost_usd: string; + avg_cost_per_execution_usd: string; + last_executed_at?: string; + first_executed_at?: string; + common_error_types: string[]; + execution_trend: TimeSeriesDataPoint[]; +} + +// Cost summary +export interface CostSummary { + total_cost_usd: string; + period_days: number; + by_model: { + model: string; + cost_usd: string; + execution_count: number; + percentage: number; + }[]; + by_agent: { + agent_name: string; + agent_id?: string; + cost_usd: string; + execution_count: number; + percentage: number; + }[]; +} + +// Cost forecast +export interface CostForecast { + current_period_cost_usd: string; + projected_monthly_cost_usd: string; + projected_weekly_cost_usd: string; + daily_average_cost_usd: string; + cost_trend: 'increasing' | 'decreasing' | 'stable'; + trend_percentage: number; + by_model: CostBreakdown[]; + by_agent: CostBreakdown[]; +} + +// Budget alert +export interface BudgetAlert { + id: string; + user_id: string; + scope: BudgetScope; + scope_id?: string; + threshold_usd: string; + period_days: number; + alert_type: BudgetAlertType; + alert_at_percentage: number; + webhook_url?: string; + email_notification: boolean; + is_active: boolean; + last_alert_sent_at?: string; + current_spend_usd: string; + created_at: string; + updated_at: string; +} + +// Budget alert creation +export interface BudgetAlertCreate { + scope?: BudgetScope; + scope_id?: string; + threshold_usd: number; + period_days?: number; + alert_type?: BudgetAlertType; + alert_at_percentage?: number; + webhook_url?: string; + email_notification?: boolean; +} + +// Budget alert update +export interface BudgetAlertUpdate { + threshold_usd?: number; + period_days?: number; + alert_type?: BudgetAlertType; + alert_at_percentage?: number; + webhook_url?: string; + email_notification?: boolean; + is_active?: boolean; +} + +// Model comparison +export interface ModelComparison { + model_name: string; + execution_count: number; + success_count: number; + success_rate: number; + avg_execution_time_ms: number; + avg_tokens: number; + total_tokens: number; + total_cost_usd: string; +} diff --git a/tests/TestAPI/test_analytics.py b/tests/TestAPI/test_analytics.py new file mode 100644 index 000000000..7da3fe9a4 --- /dev/null +++ b/tests/TestAPI/test_analytics.py @@ -0,0 +1,486 @@ +""" +Tests for the Analytics API endpoints. + +This module contains tests for: +- Analytics overview endpoint +- Executions list endpoint +- Execution details endpoint +- Agent stats endpoint +- Cost summary endpoint +- Cost forecast endpoint +- Budget alerts CRUD +- Model comparison endpoint +- Export endpoint +""" + +import pytest +from datetime import datetime, timedelta +from uuid import uuid4 + +import aiohttp + +ANALYTICS_BASE = "/api/analytics" +OVERVIEW_ENDPOINT = f"{ANALYTICS_BASE}/overview" +EXECUTIONS_ENDPOINT = f"{ANALYTICS_BASE}/executions" +EXECUTION_DETAIL_ENDPOINT = f"{ANALYTICS_BASE}/executions/{{execution_id}}" +AGENT_STATS_ENDPOINT = f"{ANALYTICS_BASE}/agents/{{agent_id}}/stats" +COST_SUMMARY_ENDPOINT = f"{ANALYTICS_BASE}/costs/summary" +COST_FORECAST_ENDPOINT = f"{ANALYTICS_BASE}/costs/forecast" +BUDGETS_ENDPOINT = f"{ANALYTICS_BASE}/budgets" +BUDGET_DETAIL_ENDPOINT = f"{ANALYTICS_BASE}/budgets/{{alert_id}}" +MODEL_COMPARISON_ENDPOINT = f"{ANALYTICS_BASE}/comparison/models" +EXPORT_ENDPOINT = f"{ANALYTICS_BASE}/export" + +BASE_URL = "http://localhost:8000" + + +class HttpClient: + """Simple HTTP client that returns JSON directly.""" + + def __init__(self, base_url: str = BASE_URL): + self.base_url = base_url.rstrip("/") + + async def _request(self, method: str, path: str, **kwargs): + url = f"{self.base_url}/{path.lstrip('/')}" + async with aiohttp.ClientSession() as session: + async with session.request(method, url, **kwargs) as response: + if response.status >= 400: + raise aiohttp.ClientResponseError( + response.request_info, + response.history, + status=response.status, + message=await response.text(), + ) + content_type = response.headers.get("Content-Type", "") + if "application/json" in content_type: + return await response.json() + return await response.text() + + async def get(self, path: str, **kwargs): + return await self._request("GET", path, **kwargs) + + async def post(self, path: str, **kwargs): + return await self._request("POST", path, **kwargs) + + async def patch(self, path: str, **kwargs): + return await self._request("PATCH", path, **kwargs) + + async def delete(self, path: str, **kwargs): + return await self._request("DELETE", path, **kwargs) + + +http_client = HttpClient() + + +class TestAnalyticsOverview: + """Tests for the analytics overview endpoint.""" + + @pytest.mark.asyncio + async def test_get_overview_default_period(self, user_jwt_token: str): + """Test getting analytics overview with default 24h period.""" + response = await http_client.get( + path=OVERVIEW_ENDPOINT, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + # Verify response structure + assert "total_executions" in response + assert "successful_executions" in response + assert "failed_executions" in response + assert "success_rate" in response + assert "avg_execution_time_ms" in response + assert "total_cost_usd" in response + assert "top_agents" in response + assert "cost_by_model" in response + assert isinstance(response["total_executions"], int) + assert isinstance(response["top_agents"], list) + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "period", + ["24h", "7d", "30d", "90d"], + ids=["24h period", "7d period", "30d period", "90d period"], + ) + async def test_get_overview_different_periods(self, user_jwt_token: str, period: str): + """Test getting analytics overview with different time periods.""" + response = await http_client.get( + path=OVERVIEW_ENDPOINT, + params={"period": period}, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + assert "total_executions" in response + assert response["total_executions"] >= 0 + + @pytest.mark.asyncio + async def test_get_overview_custom_date_range(self, user_jwt_token: str): + """Test getting analytics overview with custom date range.""" + end_date = datetime.utcnow() + start_date = end_date - timedelta(days=7) + + response = await http_client.get( + path=OVERVIEW_ENDPOINT, + params={ + "start_date": start_date.isoformat(), + "end_date": end_date.isoformat(), + }, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + assert "total_executions" in response + + @pytest.mark.asyncio + async def test_get_overview_unauthorized(self): + """Test that unauthorized requests are rejected.""" + try: + await http_client.get(path=OVERVIEW_ENDPOINT) + assert False, "Should have raised an error" + except Exception: + pass # Expected to fail without auth + + +class TestExecutionsList: + """Tests for the executions list endpoint.""" + + @pytest.mark.asyncio + async def test_list_executions_default(self, user_jwt_token: str): + """Test listing executions with default parameters.""" + response = await http_client.get( + path=EXECUTIONS_ENDPOINT, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + assert "items" in response + assert "total" in response + assert "page" in response + assert "page_size" in response + assert "total_pages" in response + assert isinstance(response["items"], list) + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "page,page_size", + [(1, 10), (1, 20), (2, 10)], + ids=["page 1, size 10", "page 1, size 20", "page 2, size 10"], + ) + async def test_list_executions_pagination( + self, user_jwt_token: str, page: int, page_size: int + ): + """Test executions list pagination.""" + response = await http_client.get( + path=EXECUTIONS_ENDPOINT, + params={"page": page, "page_size": page_size}, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + assert response["page"] == page + assert response["page_size"] == page_size + assert len(response["items"]) <= page_size + + @pytest.mark.asyncio + async def test_list_executions_filter_by_status(self, user_jwt_token: str): + """Test filtering executions by status.""" + response = await http_client.get( + path=EXECUTIONS_ENDPOINT, + params={"status": "success"}, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + assert "items" in response + for item in response["items"]: + assert item["status"] == "success" + + @pytest.mark.asyncio + async def test_list_executions_sort_by_time(self, user_jwt_token: str): + """Test sorting executions by time.""" + response = await http_client.get( + path=EXECUTIONS_ENDPOINT, + params={"sort_by": "started_at", "sort_order": "desc"}, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + assert "items" in response + + +class TestExecutionDetails: + """Tests for the execution details endpoint.""" + + @pytest.mark.asyncio + async def test_get_execution_not_found(self, user_jwt_token: str): + """Test getting a non-existent execution.""" + fake_id = str(uuid4()) + try: + await http_client.get( + path=EXECUTION_DETAIL_ENDPOINT.format(execution_id=fake_id), + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + assert False, "Should have raised 404" + except Exception: + pass # Expected + + @pytest.mark.asyncio + async def test_get_execution_invalid_uuid(self, user_jwt_token: str): + """Test getting execution with invalid UUID.""" + try: + await http_client.get( + path=EXECUTION_DETAIL_ENDPOINT.format(execution_id="invalid-uuid"), + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + assert False, "Should have raised error" + except Exception: + pass # Expected + + +class TestAgentStats: + """Tests for the agent stats endpoint.""" + + @pytest.mark.asyncio + async def test_get_agent_stats_not_found(self, user_jwt_token: str): + """Test getting stats for non-existent agent.""" + fake_id = str(uuid4()) + try: + await http_client.get( + path=AGENT_STATS_ENDPOINT.format(agent_id=fake_id), + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + assert False, "Should have raised 404" + except Exception: + pass # Expected + + +class TestCostEndpoints: + """Tests for cost-related endpoints.""" + + @pytest.mark.asyncio + async def test_get_cost_summary(self, user_jwt_token: str): + """Test getting cost summary.""" + response = await http_client.get( + path=COST_SUMMARY_ENDPOINT, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + assert "total_cost_usd" in response + assert "period_days" in response + assert "by_model" in response + assert "by_agent" in response + + @pytest.mark.asyncio + async def test_get_cost_forecast(self, user_jwt_token: str): + """Test getting cost forecast.""" + response = await http_client.get( + path=COST_FORECAST_ENDPOINT, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + assert "current_period_cost_usd" in response + assert "projected_monthly_cost_usd" in response + assert "daily_average_cost_usd" in response + assert "cost_trend" in response + + +class TestBudgetAlerts: + """Tests for budget alert CRUD operations.""" + + @pytest.mark.asyncio + async def test_list_budget_alerts_empty(self, user_jwt_token: str): + """Test listing budget alerts when none exist.""" + response = await http_client.get( + path=BUDGETS_ENDPOINT, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + assert isinstance(response, list) + + @pytest.mark.asyncio + async def test_create_budget_alert(self, user_jwt_token: str): + """Test creating a budget alert.""" + alert_data = { + "threshold_usd": 100.00, + "period_days": 30, + "alert_type": "warning", + "alert_at_percentage": 80, + "email_notification": True, + } + + response = await http_client.post( + path=BUDGETS_ENDPOINT, + json=alert_data, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + assert "id" in response + assert response["threshold_usd"] == "100.00" or float(response["threshold_usd"]) == 100.00 + assert response["period_days"] == 30 + assert response["alert_at_percentage"] == 80 + + @pytest.mark.asyncio + async def test_create_budget_alert_invalid_percentage(self, user_jwt_token: str): + """Test creating a budget alert with invalid percentage.""" + alert_data = { + "threshold_usd": 100.00, + "alert_at_percentage": 150, # Invalid: > 100 + } + + try: + await http_client.post( + path=BUDGETS_ENDPOINT, + json=alert_data, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + assert False, "Should have raised validation error" + except Exception: + pass # Expected + + @pytest.mark.asyncio + async def test_update_budget_alert(self, user_jwt_token: str): + """Test updating a budget alert.""" + # First create an alert + alert_data = {"threshold_usd": 50.00} + created = await http_client.post( + path=BUDGETS_ENDPOINT, + json=alert_data, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + # Update it + update_data = {"threshold_usd": 75.00, "is_active": False} + updated = await http_client.patch( + path=BUDGET_DETAIL_ENDPOINT.format(alert_id=created["id"]), + json=update_data, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + assert float(updated["threshold_usd"]) == 75.00 or updated["threshold_usd"] == "75.00" + assert updated["is_active"] is False + + @pytest.mark.asyncio + async def test_delete_budget_alert(self, user_jwt_token: str): + """Test deleting a budget alert.""" + # First create an alert + alert_data = {"threshold_usd": 25.00} + created = await http_client.post( + path=BUDGETS_ENDPOINT, + json=alert_data, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + # Delete it + delete_data = await http_client.delete( + path=BUDGET_DETAIL_ENDPOINT.format(alert_id=created["id"]), + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + assert delete_data["status"] == "deleted" + + @pytest.mark.asyncio + async def test_delete_budget_alert_not_found(self, user_jwt_token: str): + """Test deleting a non-existent budget alert.""" + fake_id = str(uuid4()) + try: + await http_client.delete( + path=BUDGET_DETAIL_ENDPOINT.format(alert_id=fake_id), + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + assert False, "Should have raised 404" + except Exception: + pass # Expected + + +class TestModelComparison: + """Tests for model comparison endpoint.""" + + @pytest.mark.asyncio + async def test_compare_models(self, user_jwt_token: str): + """Test getting model comparison data.""" + response = await http_client.get( + path=MODEL_COMPARISON_ENDPOINT, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + assert isinstance(response, list) + # If there's data, verify structure + if response: + item = response[0] + assert "model_name" in item + assert "execution_count" in item + assert "success_rate" in item + + +class TestExport: + """Tests for the export endpoint.""" + + @pytest.mark.asyncio + async def test_export_json(self, user_jwt_token: str): + """Test exporting analytics data as JSON.""" + response = await http_client.get( + path=EXPORT_ENDPOINT, + params={"format": "json"}, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + + # Response should be JSON with export data + assert "export_date" in response or isinstance(response, (dict, list)) + + @pytest.mark.asyncio + async def test_export_csv(self, user_jwt_token: str): + """Test exporting analytics data as CSV.""" + # For CSV, we can't easily verify the structure in the same way + # Just ensure the endpoint responds + try: + await http_client.get( + path=EXPORT_ENDPOINT, + params={"format": "csv"}, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + except Exception: + # CSV response might not be JSON-parseable + pass + + +class TestAnalyticsIntegration: + """Integration tests for analytics functionality.""" + + @pytest.mark.asyncio + async def test_analytics_workflow(self, user_jwt_token: str): + """Test a complete analytics workflow.""" + # 1. Get overview + overview = await http_client.get( + path=OVERVIEW_ENDPOINT, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + assert overview is not None + + # 2. Get executions list + executions = await http_client.get( + path=EXECUTIONS_ENDPOINT, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + assert "items" in executions + + # 3. Get cost summary + cost_summary = await http_client.get( + path=COST_SUMMARY_ENDPOINT, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + assert "total_cost_usd" in cost_summary + + # 4. Create a budget alert + alert = await http_client.post( + path=BUDGETS_ENDPOINT, + json={"threshold_usd": 100.00}, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + assert "id" in alert + + # 5. List budget alerts + alerts = await http_client.get( + path=BUDGETS_ENDPOINT, + headers={"Authorization": f"Bearer {user_jwt_token}"}, + ) + assert len(alerts) > 0 + + # 6. Delete the budget alert + await http_client.delete( + path=BUDGET_DETAIL_ENDPOINT.format(alert_id=alert["id"]), + headers={"Authorization": f"Bearer {user_jwt_token}"}, + )