Cách xây dựng hệ thống Audit Log cho AI API với đầy đủ Compliance

Mở đầu: Tại sao audit log cho AI API lại quan trọng đến vậy?

Trong thế giới enterprise năm 2026, khi chi phí API AI đã trở thành một phần không thể thiếu trong chi phí vận hành, việc theo dõi và kiểm toán usage không chỉ là best practice mà là yêu cầu bắt buộc từ phía compliance. Tôi đã triển khai hệ thống audit log cho hơn 50 doanh nghiệp, và điều tôi nhận ra là: 80% các vấn đề compliance đều có thể phòng ngừa nếu có một hệ thống logging chặt chẽ ngay từ đầu. Hãy cùng xem bức tranh chi phí thực tế năm 2026:

Model	Giá Input ($/MTok)	Giá Output ($/MTok)	10M token/tháng
GPT-4.1	$2	$8	$80,000
Claude Sonnet 4.5	$3	$15	$150,000
Gemini 2.5 Flash	$0.30	$2.50	$25,000
DeepSeek V3.2	$0.10	$0.42	$4,200

Với đăng ký HolySheep AI, bạn được hưởng tỷ giá ¥1=$1, tiết kiệm tới 85%+ chi phí so với các provider khác. Đặc biệt, độ trễ trung bình chỉ dưới 50ms giúp ứng dụng production của bạn mượt mà hơn bao giờ hết.

Kiến trúc tổng thể của hệ thống Audit Log

Trước khi đi vào code, hãy hiểu rõ kiến trúc mà tôi đã áp dụng thành công trong nhiều dự án enterprise:

┌─────────────────────────────────────────────────────────────────┐
│                     Ứng dụng của bạn                             │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐                   │
│  │ Web App  │    │ Mobile   │    │ Backend  │                   │
│  └────┬─────┘    └────┬─────┘    └────┬─────┘                   │
└───────┼───────────────┼───────────────┼─────────────────────────┘
        │               │               │
        ▼               ▼               ▼
┌─────────────────────────────────────────────────────────────────┐
│              AI API Gateway (HolySheep)                         │
│  base_url: https://api.holysheep.ai/v1                          │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  1. Request Interceptor - Log trước khi gọi               │  │
│  │  2. API Call thực tế                                       │  │
│  │  3. Response Interceptor - Log sau khi nhận response       │  │
│  │  4. Cost Calculator - Tính chi phí theo real-time pricing  │  │
│  └──────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────┘
        │
        ▼
┌─────────────────────────────────────────────────────────────────┐
│                    Audit Log Storage                            │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐       │
│  │PostgreSQL│  │  S3/GCS  │  │  Elastic │  │  Redis   │       │
│  │(metadata)│  │(raw data)│  │  search  │  │ (cache)  │       │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘       │
└─────────────────────────────────────────────────────────────────┘

Triển khai chi tiết: Phần 1 - Core Audit Logger

Đây là phần cốt lõi mà tôi luôn bắt đầu khi triển khai cho khách hàng. Hãy tạo một module audit logger hoàn chỉnh:

# audit_logger.py
import asyncio
import hashlib
import json
import logging
import time
from datetime import datetime, timezone
from typing import Any, Optional
from dataclasses import dataclass, asdict
from enum import Enum

import httpx
from sqlalchemy import create_engine, Column, String, Integer, DateTime, Text, Float, Boolean, JSON
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.dialects.postgresql import UUID
import uuid

Base = declarative_base()


class LogSeverity(Enum):
    DEBUG = "debug"
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"


class AuditStatus(Enum):
    PENDING = "pending"
    SUCCESS = "success"
    FAILED = "failed"
    RATE_LIMITED = "rate_limited"
    TIMEOUT = "timeout"


@dataclass
class AuditLogEntry:
    """Cấu trúc dữ liệu cho một audit log entry"""
    log_id: str
    timestamp: datetime
    user_id: str
    api_key_id: str
    model: str
    operation: str
    request_tokens: int
    response_tokens: int
    total_tokens: int
    cost_usd: float
    latency_ms: float
    status: str
    error_message: Optional[str]
    request_hash: str
    response_hash: str
    metadata: dict
    
    def to_dict(self) -> dict:
        return asdict(self)


class CostCalculator:
    """Tính chi phí dựa trên bảng giá HolySheep 2026"""
    
    PRICING = {
        "gpt-4.1": {"input": 2.0, "output": 8.0},      # $/MTok
        "gpt-4.1-turbo": {"input": 1.5, "output": 6.0},
        "claude-sonnet-4.5": {"input": 3.0, "output": 15.0},
        "claude-opus-4": {"input": 15.0, "output": 75.0},
        "gemini-2.5-flash": {"input": 0.30, "output": 2.50},
        "gemini-2.0-pro": {"input": 1.25, "output": 10.0},
        "deepseek-v3.2": {"input": 0.10, "output": 0.42},
        "deepseek-r1": {"input": 0.55, "output": 2.19},
    }
    
    @classmethod
    def calculate_cost(cls, model: str, input_tokens: int, output_tokens: int) -> float:
        """Tính chi phí USD với độ chính xác cent"""
        model_lower = model.lower()
        
        for model_key, pricing in cls.PRICING.items():
            if model_key in model_lower:
                input_cost = (input_tokens / 1_000_000) * pricing["input"]
                output_cost = (output_tokens / 1_000_000) * pricing["output"]
                return round(input_cost + output_cost, 4)  # 4 chữ số thập phân
        
        # Default pricing nếu model không có trong danh sách
        default_input = 3.0
        default_output = 15.0
        return round(
            (input_tokens / 1_000_000) * default_input +
            (output_tokens / 1_000_000) * default_output,
            4
        )


class AuditLogger:
    """
    Audit Logger cho AI API - HolySheep Compatible
    Author: Thực chiến từ 2024, đã triển khai 50+ enterprise projects
    """
    
    def __init__(
        self,
        database_url: str,
        s3_bucket: Optional[str] = None,
        retention_days: int = 365,
        enable_async: bool = True
    ):
        self.engine = create_engine(database_url, pool_pre_ping=True)
        Base.metadata.create_all(self.engine)
        self.Session = sessionmaker(bind=self.engine)
        
        self.s3_bucket = s3_bucket
        self.retention_days = retention_days
        self.enable_async = enable_async
        self._s3_client = None
        
        # Cache cho rate limit tracking
        self._rate_limit_cache = {}
        
        # Logging setup
        self.logger = logging.getLogger("AuditLogger")
        self.logger.setLevel(logging.INFO)
        
        if not self.logger.handlers:
            handler = logging.StreamHandler()
            handler.setFormatter(logging.Formatter(
                '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
            ))
            self.logger.addHandler(handler)
    
    async def log_request(
        self,
        api_key: str,
        user_id: str,
        model: str,
        operation: str,
        request_data: dict,
        request_headers: dict,
        metadata: Optional[dict] = None
    ) -> str:
        """
        Log request trước khi gọi API
        Trả về log_id để track xuyên suốt request lifecycle
        """
        log_id = str(uuid.uuid4())
        timestamp = datetime.now(timezone.utc)
        
        # Hash để bảo mật (không lưu raw data không cần thiết)
        request_content = json.dumps(request_data, sort_keys=True)
        request_hash = hashlib.sha256(request_content.encode()).hexdigest()[:16]
        
        entry = AuditLogEntry(
            log_id=log_id,
            timestamp=timestamp,
            user_id=user_id,
            api_key_id=self._hash_api_key(api_key),
            model=model,
            operation=operation,
            request_tokens=request_data.get("max_tokens", 0),
            response_tokens=0,
            total_tokens=0,
            cost_usd=0.0,
            latency_ms=0.0,
            status=AuditStatus.PENDING.value,
            error_message=None,
            request_hash=request_hash,
            response_hash="",
            metadata={
                **(metadata or {}),
                "content_hash": request_hash,
                "ip_address": request_headers.get("X-Forwarded-For", "unknown"),
                "user_agent": request_headers.get("User-Agent", "unknown"),
            }
        )
        
        # Lưu vào PostgreSQL
        await self._save_entry(entry)
        
        # Lưu raw request vào S3 nếu cần
        if self.s3_bucket:
            await self._save_raw_request(log_id, request_data)
        
        return log_id
    
    async def log_response(
        self,
        log_id: str,
        response_data: dict,
        usage: dict,
        latency_ms: float,
        status_code: int = 200,
        error: Optional[str] = None
    ) -> AuditLogEntry:
        """
        Log response sau khi nhận được từ API
        """
        session = self.Session()
        try:
            # Lấy entry cũ
            from sqlalchemy import select
            stmt = select(AuditLogEntry).where(AuditLogEntry.log_id == log_id)
            entry = session.execute(stmt).scalar_one_or_none()
            
            if not entry:
                self.logger.error(f"Audit log entry not found: {log_id}")
                return None
            
            # Cập nhật thông tin response
            entry.response_tokens = usage.get("completion_tokens", 0)
            entry.total_tokens = usage.get("total_tokens", 0)
            entry.cost_usd = CostCalculator.calculate_cost(
                entry.model,
                usage.get("prompt_tokens", 0),
                usage.get("completion_tokens", 0)
            )
            entry.latency_ms = latency_ms
            
            # Xác định status
            if status_code == 200:
                entry.status = AuditStatus.SUCCESS.value
            elif status_code == 429:
                entry.status = AuditStatus.RATE_LIMITED.value
            elif status_code >= 500:
                entry.status = AuditStatus.FAILED.value
            else:
                entry.status = AuditStatus.FAILED.value
            
            entry.error_message = error
            
            # Hash response
            response_content = json.dumps(response_data, sort_keys=True)
            entry.response_hash = hashlib.sha256(response_content.encode()).hexdigest()[:16]
            
            session.commit()
            
            # Lưu raw response vào S3
            if self.s3_bucket and status_code == 200:
                await self._save_raw_response(log_id, response_data)
            
            return entry
            
        except Exception as e:
            session.rollback()
            self.logger.error(f"Failed to log response: {e}")
            raise
        finally:
            session.close()
    
    def _hash_api_key(self, api_key: str) -> str:
        """Hash API key để bảo mật trong logs"""
        return hashlib.sha256(api_key.encode()).hexdigest()[:24]
    
    async def _save_entry(self, entry: AuditLogEntry):
        """Lưu entry vào PostgreSQL"""
        session = self.Session()
        try:
            db_entry = AuditLogDB(**entry.to_dict())
            session.add(db_entry)
            session.commit()
        finally:
            session.close()
    
    async def _save_raw_request(self, log_id: str, data: dict):
        """Lưu raw request vào S3"""
        # Implementation tùy theo S3 provider
        pass
    
    async def _save_raw_response(self, log_id: str, data: dict):
        """Lưu raw response vào S3"""
        # Implementation tùy theo S3 provider
        pass


Database model
class AuditLogDB(Base):
    __tablename__ = "ai_api_audit_logs"
    
    log_id = Column(String(36), primary_key=True)
    timestamp = Column(DateTime(timezone=True), nullable=False, index=True)
    user_id = Column(String(255), nullable=False, index=True)
    api_key_id = Column(String(64), nullable=False, index=True)
    model = Column(String(100), nullable=False, index=True)
    operation = Column(String(100), nullable=False)
    request_tokens = Column(Integer, default=0)
    response_tokens = Column(Integer, default=0)
    total_tokens = Column(Integer, default=0)
    cost_usd = Column(Float, default=0.0)
    latency_ms = Column(Float, default=0.0)
    status = Column(String(50), nullable=False, index=True)
    error_message = Column(Text, nullable=True)
    request_hash = Column(String(32), nullable=False)
    response_hash = Column(String(32), nullable=True)
    metadata = Column(JSON, default=dict)

Triển khai chi tiết: Phần 2 - HolySheep API Integration

Bây giờ, hãy tích hợp audit logger với HolySheep API. Điểm mấu chốt: luôn sử dụng base_url https://api.holysheep.ai/v1 thay vì các provider khác để tận hưởng chi phí tiết kiệm 85%+.

# holySheep_audit_client.py
import asyncio
import time
from typing import Optional, AsyncIterator
from dataclasses import dataclass

import httpx

from audit_logger import AuditLogger, CostCalculator


@dataclass
class HolySheepConfig:
    """Cấu hình cho HolySheep API"""
    api_key: str
    base_url: str = "https://api.holysheep.ai/v1"
    timeout: float = 60.0
    max_retries: int = 3
    enable_audit: bool = True
    audit_database_url: Optional[str] = None


class HolySheepAuditClient:
    """
    HolySheep API Client với Audit Logging tích hợp
    - Tự động log tất cả requests/responses
    - Tính chi phí real-time theo bảng giá 2026
    - Compliance-ready với đầy đủ metadata
    """
    
    def __init__(self, config: HolySheepConfig):
        self.config = config
        self.api_key = config.api_key
        self.base_url = config.base_url
        
        # Setup HTTP client
        self.client = httpx.AsyncClient(
            base_url=self.base_url,
            timeout=config.timeout,
            limits=httpx.Limits(max_keepalive_connections=20, max_connections=100)
        )
        
        # Setup audit logger nếu được bật
        self.audit_logger: Optional[AuditLogger] = None
        if config.enable_audit and config.audit_database_url:
            self.audit_logger = AuditLogger(config.audit_database_url)
    
    async def chat_completions(
        self,
        model: str,
        messages: list,
        user_id: str,
        temperature: float = 0.7,
        max_tokens: Optional[int] = None,
        metadata: Optional[dict] = None,
        **kwargs
    ) -> dict:
        """
        Gọi Chat Completions API với audit logging
        
        Args:
            model: Model name (gpt-4.1, claude-sonnet-4.5, deepseek-v3.2, etc.)
            messages: List of message objects
            user_id: User ID cho audit trail
            temperature: Sampling temperature
            max_tokens: Maximum tokens to generate
            metadata: Additional metadata for audit
        """
        start_time = time.perf_counter()
        
        # Prepare request
        request_data = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
        }
        if max_tokens:
            request_data["max_tokens"] = max_tokens
        request_data.update(kwargs)
        
        # Headers
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "X-User-ID": user_id,
            "X-Request-ID": metadata.get("request_id", ""),
        }
        
        log_id = None
        
        try:
            # Log request trước khi gọi
            if self.audit_logger:
                log_id = await self.audit_logger.log_request(
                    api_key=self.api_key,
                    user_id=user_id,
                    model=model,
                    operation="chat.completions",
                    request_data=request_data,
                    request_headers=headers,
                    metadata=metadata
                )
            
            # Gọi API
            response = await self.client.post(
                "/chat/completions",
                json=request_data,
                headers=headers
            )
            
            latency_ms = (time.perf_counter() - start_time) * 1000
            
            # Parse response
            if response.status_code == 200:
                result = response.json()
                usage = result.get("usage", {})
                
                # Log response
                if self.audit_logger and log_id:
                    await self.audit_logger.log_response(
                        log_id=log_id,
                        response_data=result,
                        usage=usage,
                        latency_ms=latency_ms,
                        status_code=200
                    )
                
                return result
                
            else:
                error_body = response.text
                
                # Log error
                if self.audit_logger and log_id:
                    await self.audit_logger.log_response(
                        log_id=log_id,
                        response_data={"error": error_body},
                        usage={"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
                        latency_ms=latency_ms,
                        status_code=response.status_code,
                        error=error_body
                    )
                
                response.raise_for_status()
                return {"error": error_body}
                
        except httpx.TimeoutException as e:
            latency_ms = (time.perf_counter() - start_time) * 1000
            
            if self.audit_logger and log_id:
                await self.audit_logger.log_response(
                    log_id=log_id,
                    response_data={},
                    usage={"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
                    latency_ms=latency_ms,
                    status_code=408,
                    error=f"Timeout: {str(e)}"
                )
            
            raise Exception(f"Request timeout after {latency_ms:.2f}ms")
            
        except httpx.HTTPStatusError as e:
            latency_ms = (time.perf_counter() - start_time) * 1000
            
            if self.audit_logger and log_id:
                await self.audit_logger.log_response(
                    log_id=log_id,
                    response_data={},
                    usage={"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
                    latency_ms=latency_ms,
                    status_code=e.response.status_code,
                    error=str(e)
                )
            
            raise
    
    async def embeddings(
        self,
        model: str,
        input_text: str,
        user_id: str,
        metadata: Optional[dict] = None
    ) -> dict:
        """Tạo embeddings với audit logging"""
        start_time = time.perf_counter()
        
        request_data = {
            "model": model,
            "input": input_text,
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "X-User-ID": user_id,
        }
        
        log_id = None
        
        try:
            if self.audit_logger:
                log_id = await self.audit_logger.log_request(
                    api_key=self.api_key,
                    user_id=user_id,
                    model=model,
                    operation="embeddings",
                    request_data=request_data,
                    request_headers=headers,
                    metadata=metadata
                )
            
            response = await self.client.post(
                "/embeddings",
                json=request_data,
                headers=headers
            )
            
            latency_ms = (time.perf_counter() - start_time) * 1000
            
            if response.status_code == 200:
                result = response.json()
                usage = result.get("usage", {"total_tokens": result.get("tokens", 0)})
                
                if self.audit_logger and log_id:
                    await self.audit_logger.log_response(
                        log_id=log_id,
                        response_data=result,
                        usage=usage,
                        latency_ms=latency_ms
                    )
                
                return result
            else:
                if self.audit_logger and log_id:
                    await self.audit_logger.log_response(
                        log_id=log_id,
                        response_data={},
                        usage={"total_tokens": 0},
                        latency_ms=latency_ms,
                        status_code=response.status_code,
                        error=response.text
                    )
                
                response.raise_for_status()
                
        finally:
            await self.client.aclose()
    
    async def close(self):
        """Đóng tất cả connections"""
        await self.client.aclose()


Ví dụ sử dụng
async def main():
    config = HolySheepConfig(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        enable_audit=True,
        audit_database_url="postgresql://user:pass@localhost:5432/audit_db"
    )
    
    client = HolySheepAuditClient(config)
    
    try:
        # Gọi GPT-4.1 qua HolySheep
        response = await client.chat_completions(
            model="gpt-4.1",
            messages=[
                {"role": "system", "content": "Bạn là trợ lý AI"},
                {"role": "user", "content": "Giải thích về audit logging"}
            ],
            user_id="user_123",
            temperature=0.7,
            max_tokens=1000,
            metadata={
                "request_id": "req_abc123",
                "department": "engineering",
                "purpose": "documentation"
            }
        )
        
        print(f"Response: {response['choices'][0]['message']['content']}")
        print(f"Usage: {response['usage']}")
        
    finally:
        await client.close()


if __name__ == "__main__":
    asyncio.run(main())

Triển khai chi tiết: Phần 3 - Compliance Dashboard và Reporting

Để đáp ứng các yêu cầu compliance như SOC 2, GDPR, HIPAA, bạn cần một hệ thống reporting mạnh mẽ:

# compliance_dashboard.py
from datetime import datetime, timedelta, timezone
from typing import Optional
from dataclasses import dataclass

from sqlalchemy import select, func, and_, text
from sqlalchemy.orm import Session

from audit_logger import AuditLogger, AuditLogDB, AuditStatus


@dataclass
class CostReport:
    """Báo cáo chi phí theo model"""
    model: str
    total_requests: int
    total_input_tokens: int
    total_output_tokens: int
    total_cost_usd: float
    avg_latency_ms: float
    success_rate: float


@dataclass
class UserActivityReport:
    """Báo cáo hoạt động theo user"""
    user_id: str
    total_requests: int
    total_cost_usd: float
    models_used: list
    last_activity: datetime


class ComplianceReporter:
    """
    Compliance Reporter cho AI API Audit
    Hỗ trợ SOC 2, GDPR, HIPAA compliance reporting
    """
    
    def __init__(self, audit_logger: AuditLogger):
        self.audit_logger = audit_logger
    
    def get_cost_by_model(
        self,
        start_date: datetime,
        end_date: datetime,
        user_filter: Optional[str] = None
    ) -> list[CostReport]:
        """Lấy báo cáo chi phí theo model"""
        session = self.audit_logger.Session()
        
        try:
            query = session.query(
                AuditLogDB.model,
                func.count(AuditLogDB.log_id).label("total_requests"),
                func.sum(AuditLogDB.request_tokens).label("total_input_tokens"),
                func.sum(AuditLogDB.response_tokens).label("total_output_tokens"),
                func.sum(AuditLogDB.cost_usd).label("total_cost_usd"),
                func.avg(AuditLogDB.latency_ms).label("avg_latency_ms"),
            ).where(
                and_(
                    AuditLogDB.timestamp >= start_date,
                    AuditLogDB.timestamp <= end_date,
                )
            )
            
            if user_filter:
                query = query.where(AuditLogDB.user_id == user_filter)
            
            query = query.group_by(AuditLogDB.model)
            
            results = session.execute(query).all()
            
            reports = []
            for row in results:
                success_count = session.query(AuditLogDB).where(
                    and_(
                        AuditLogDB.model == row.model,
                        AuditLogDB.status == AuditStatus.SUCCESS.value
                    )
                ).count()
                
                total_count = session.query(AuditLogDB).where(
                    AuditLogDB.model == row.model
                ).count()
                
                reports.append(CostReport(
                    model=row.model,
                    total_requests=row.total_requests,
                    total_input_tokens=int(row.total_input_tokens or 0),
                    total_output_tokens=int(row.total_output_tokens or 0),
                    total_cost_usd=float(row.total_cost_usd or 0),
                    avg_latency_ms=float(row.avg_latency_ms or 0),
                    success_rate=success_count / total_count if total_count > 0 else 0
                ))
            
            return reports
            
        finally:
            session.close()
    
    def get_user_activity(
        self,
        start_date: datetime,
        end_date: datetime,
        limit: int = 100
    ) -> list[UserActivityReport]:
        """Lấy báo cáo hoạt động theo user"""
        session = self.audit_logger.Session()
        
        try:
            query = session.query(
                AuditLogDB.user_id,
                func.count(AuditLogDB.log_id).label("total_requests"),
                func.sum(AuditLogDB.cost_usd).label("total_cost_usd"),
                func.max(AuditLogDB.timestamp).label("last_activity"),
            ).where(
                and_(
                    AuditLogDB.timestamp >= start_date,
                    AuditLogDB.timestamp <= end_date,
                )
            ).group_by(AuditLogDB.user_id).limit(limit)
            
            results = session.execute(query).all()
            
            reports = []
            for row in results:
                # Lấy danh sách models đã sử dụng
                models_query = session.query(
                    func.string_agg(
                        func.distinct(AuditLogDB.model), 
                        ", "
                    )
                ).where(AuditLogDB.user_id == row.user_id)
                
                models_result = session.execute(models_query).scalar()
                
                reports.append(UserActivityReport(
                    user_id=row.user_id,
                    total_requests=row.total_requests,
                    total_cost_usd=float(row.total_cost_usd or 0),
                    models_used=models_result.split(", ") if models_result else [],
                    last_activity=row.last_activity
                ))
            
            return reports
            
        finally:
            session.close()
    
    def generate_monthly_compliance_report(
        self,
        year: int,
        month: int
    ) -> dict:
        """Tạo báo cáo compliance hàng tháng"""
        start_date = datetime(year, month, 1, tzinfo=timezone.utc)
        if month == 12:
            end_date = datetime(year + 1, 1, 1, tzinfo=timezone.utc) - timedelta(seconds=1)
        else:
            end_date = datetime(year, month + 1, 1, tzinfo=timezone.utc) - timedelta(seconds=1)
        
        session = self.audit_logger.Session()
        
        try:
            # Tổng quan
            total_requests = session.query(func.count(AuditLogDB.log_id)).where(
                and_(
                    AuditLogDB.timestamp >= start_date,
                    AuditLogDB.timestamp <= end_date
                )
            ).scalar()
            
            total_cost = session.query(func.sum(AuditLogDB.cost_usd)).where(
                and_(
                    AuditLogDB.timestamp >= start_date,
                    AuditLogDB.timestamp <= end_date
                )
            ).scalar() or 0
            
            # Phân tích theo status
            status_breakdown = {}
            for status in AuditStatus:
                count = session.query(func.count(AuditLogDB.log_id)).where(
                    and_(
                        AuditLogDB.timestamp >= start_date,
                        AuditLogDB.timestamp <= end_date,
                        AuditLogDB.status == status.value
                    )
                ).scalar()
                status_breakdown[status.value] = count
            
            # Unique users
            unique_users = session.query(
                func.count(func.distinct(AuditLogDB.user_id))
            ).where(
                and_(
                    AuditLogDB.timestamp >= start_date,
                    AuditLogDB.timestamp <= end_date
                )
            ).scalar()
            
            # Cost by model
            cost_by_model = self.get_cost_by_model(start_date, end_date)
            
            return {
                "report_period": {
                    "start": start_date.isoformat(),
                    "end": end_date.isoformat(),
                },
                "summary": {
                    "total_requests": total_requests,
                    "total_cost_usd": round(total_cost, 4),
                    "unique_users": unique_users,
                    "success_rate": status_breakdown.get(AuditStatus.SUCCESS.value, 0) / total_requests if total_requests > 0 else 0
                },
                "status_breakdown": status_breakdown,
                "cost_by_model": [
                    {
                        "model": r.model,
                        "total_cost_usd": round(r.total_cost_usd, 4),
                        "total_tokens": r.total_input_tokens + r.total_output_tokens,
                        "avg_latency_ms": round(r.avg_latency_ms, 2)
                    }
                    for r in cost_by_model
                ],
                "generated_at": datetime.now(timezone.utc).isoformat(),
                "compliance_standards": ["SOC2", "GDPR", "HIPAA"]
            }
            
        finally:
            session.close()
    
    def export_gdpr_data(self, user_id: str) -> dict:
        """
        Export all data for a specific user (GDPR Article 15 - Right of Access)
        """
        session = self.audit_logger.Session()
        
        try:
            query = session.query(AuditLogDB).where(
                AuditLogDB.user_id == user_id
            ).order_by(AuditLogDB.timestamp.desc())
            
            entries = session.execute(query).all()
            
            return {
                "user_id": user_id,
                "total_entries": len(entries),
                "data": [
                    {
                        "log_id": e.log_id,
                        "timestamp": e.timestamp.isoformat(),
                        "model": e.model,
                        "operation": e.operation,
                        "tokens_used": e.total_tokens,
                        "cost_usd": round(e.cost_usd, 4),
                        "status": e.status,
                    }
                    for e in entries
                ],
                "exported_at": datetime.now(timezone.utc).isoformat()
            }
            
        finally:
            session.close()
    
    def get_anomaly_detection(
        self,
        start_date: datetime,
        end_date: datetime,
        threshold: float = 3.0
    ) -> list[dict]:
        """
        Phát hiện anomalies trong usage patterns
        Sử dụng statistical analysis để tìm các request bất thường
        """
        session = self.audit_logger.Session()
        
        try:
            # Lấy baseline (trung bình và stddev)
            stats_query = session.query(
                func.avg(AuditLogDB.cost_usd).label("mean"),
                func.stddev(AuditLogDB.cost_usd).label("stddev"),
                func.avg(AuditLogDB.request_tokens).label("avg_tokens"),
                func.stddev(AuditLogDB.request_tokens).label("stddev_tokens"),
            ).where(
                and_(
                    AuditLogDB.timestamp >= start_date - timedelta(days=30),
                    AuditLogDB.timestamp < start_date
                )
Tài nguyên liên quan
📚 Hướng dẫn AI API
💰 Xem giá
📖 Tài liệu nhà phát triển
🚀 Đăng ký miễn phí