smokephil/paperless-ngx
1
0
Fork 0
mirror of https://github.com/paperless-ngx/paperless-ngx.git synced 2025-12-22 14:36:48 +01:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #43 from dawnsystem/copilot/add-ai-suggestions-endpoints

Browse source 
Add API endpoints for AI suggestions with tracking and statistics
This commit is contained in:
dawnsystem 2025-11-13 19:08:36 +01:00 • committed by GitHub
commit f76a546bea
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 1801 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

441
docs/API_AI_SUGGESTIONS.md Normal file
View file

@ -0,0 +1,441 @@
# AI Suggestions API Documentation
This document describes the AI Suggestions API endpoints for the IntelliDocs-ngx project.
## Overview
The AI Suggestions API allows frontend applications to:
1. Retrieve AI-generated suggestions for document metadata
2. Apply suggestions to documents
3. Reject suggestions (for user feedback)
4. View accuracy statistics for AI model improvement
## Authentication
All endpoints require authentication. Include the authentication token in the request headers:
```http
Authorization: Token <your-auth-token>
```
## Endpoints
### 1. Get AI Suggestions
Retrieve AI-generated suggestions for a specific document.
**Endpoint:** `GET /api/documents/{id}/ai-suggestions/`
**Parameters:**
- `id` (path parameter): Document ID
**Response:**
```json
{
"tags": [
{
"id": 1,
"name": "Invoice",
"color": "#FF5733",
"confidence": 0.85
},
{
"id": 2,
"name": "Important",
"color": "#33FF57",
"confidence": 0.75
}
],
"correspondent": {
"id": 5,
"name": "Acme Corporation",
"confidence": 0.90
},
"document_type": {
"id": 3,
"name": "Invoice",
"confidence": 0.88
},
"storage_path": {
"id": 2,
"name": "Financial Documents",
"path": "/documents/financial/",
"confidence": 0.80
},
"custom_fields": [
{
"field_id": 1,
"field_name": "Invoice Number",
"value": "INV-2024-001",
"confidence": 0.92
}
],
"workflows": [
{
"id": 4,
"name": "Invoice Processing",
"confidence": 0.78
}
],
"title_suggestion": {
"title": "Invoice - Acme Corporation - 2024-01-15"
}
}
```
**Error Responses:**
- `400 Bad Request`: Document has no content to analyze
- `404 Not Found`: Document not found
- `500 Internal Server Error`: Error generating suggestions
---
### 2. Apply Suggestion
Apply an AI suggestion to a document and record user feedback.
**Endpoint:** `POST /api/documents/{id}/apply-suggestion/`
**Parameters:**
- `id` (path parameter): Document ID
**Request Body:**
```json
{
"suggestion_type": "tag",
"value_id": 1,
"confidence": 0.85
}
```
**Supported Suggestion Types:**
- `tag` - Tag assignment
- `correspondent` - Correspondent assignment
- `document_type` - Document type classification
- `storage_path` - Storage path assignment
- `title` - Document title
**Note:** Custom field and workflow suggestions are supported in the API response but not yet implemented in the apply endpoint.
**For ID-based suggestions (tag, correspondent, document_type, storage_path):**
```json
{
"suggestion_type": "correspondent",
"value_id": 5,
"confidence": 0.90
}
```
**For text-based suggestions (title):**
```json
{
"suggestion_type": "title",
"value_text": "New Document Title",
"confidence": 0.80
}
```
**Response:**
```json
{
"status": "success",
"message": "Tag 'Invoice' applied"
}
```
**Error Responses:**
- `400 Bad Request`: Invalid suggestion type or missing value
- `404 Not Found`: Referenced object not found
- `500 Internal Server Error`: Error applying suggestion
---
### 3. Reject Suggestion
Reject an AI suggestion and record user feedback for model improvement.
**Endpoint:** `POST /api/documents/{id}/reject-suggestion/`
**Parameters:**
- `id` (path parameter): Document ID
**Request Body:**
```json
{
"suggestion_type": "tag",
"value_id": 2,
"confidence": 0.65
}
```
Same format as apply-suggestion endpoint.
**Response:**
```json
{
"status": "success",
"message": "Suggestion rejected and feedback recorded"
}
```
**Error Responses:**
- `400 Bad Request`: Invalid request data
- `500 Internal Server Error`: Error recording feedback
---
### 4. AI Suggestion Statistics
Get accuracy statistics and metrics for AI suggestions.
**Endpoint:** `GET /api/documents/ai-suggestion-stats/`
**Response:**
```json
{
"total_suggestions": 150,
"total_applied": 120,
"total_rejected": 30,
"accuracy_rate": 80.0,
"by_type": {
"tag": {
"total": 50,
"applied": 45,
"rejected": 5,
"accuracy_rate": 90.0
},
"correspondent": {
"total": 40,
"applied": 35,
"rejected": 5,
"accuracy_rate": 87.5
},
"document_type": {
"total": 30,
"applied": 20,
"rejected": 10,
"accuracy_rate": 66.67
},
"storage_path": {
"total": 20,
"applied": 15,
"rejected": 5,
"accuracy_rate": 75.0
},
"title": {
"total": 10,
"applied": 5,
"rejected": 5,
"accuracy_rate": 50.0
}
},
"average_confidence_applied": 0.82,
"average_confidence_rejected": 0.58,
"recent_suggestions": [
{
"id": 150,
"document": 42,
"suggestion_type": "tag",
"suggested_value_id": 5,
"suggested_value_text": "",
"confidence": 0.85,
"status": "applied",
"user": 1,
"created_at": "2024-01-15T10:30:00Z",
"applied_at": "2024-01-15T10:30:05Z",
"metadata": {}
}
]
}
```
**Error Responses:**
- `500 Internal Server Error`: Error calculating statistics
---
## Frontend Integration Example
### React/TypeScript Example
```typescript
import axios from 'axios';
const API_BASE = '/api/documents';
interface AISuggestions {
tags?: Array<{id: number; name: string; confidence: number}>;
correspondent?: {id: number; name: string; confidence: number};
document_type?: {id: number; name: string; confidence: number};
// ... other fields
}
// Get AI suggestions
async function getAISuggestions(documentId: number): Promise<AISuggestions> {
const response = await axios.get(`${API_BASE}/${documentId}/ai-suggestions/`);
return response.data;
}
// Apply a suggestion
async function applySuggestion(
documentId: number,
type: string,
valueId: number,
confidence: number
): Promise<void> {
await axios.post(`${API_BASE}/${documentId}/apply-suggestion/`, {
suggestion_type: type,
value_id: valueId,
confidence: confidence
});
}
// Reject a suggestion
async function rejectSuggestion(
documentId: number,
type: string,
valueId: number,
confidence: number
): Promise<void> {
await axios.post(`${API_BASE}/${documentId}/reject-suggestion/`, {
suggestion_type: type,
value_id: valueId,
confidence: confidence
});
}
// Get statistics
async function getStatistics() {
const response = await axios.get(`${API_BASE}/ai-suggestion-stats/`);
return response.data;
}
// Usage example
async function handleDocument(documentId: number) {
try {
// Get suggestions
const suggestions = await getAISuggestions(documentId);
// Show suggestions to user
if (suggestions.tags) {
suggestions.tags.forEach(tag => {
console.log(`Suggested tag: ${tag.name} (${tag.confidence * 100}%)`);
});
}
// User accepts a tag suggestion
if (suggestions.tags && suggestions.tags.length > 0) {
const tag = suggestions.tags[0];
await applySuggestion(documentId, 'tag', tag.id, tag.confidence);
console.log('Tag applied successfully');
}
} catch (error) {
console.error('Error handling AI suggestions:', error);
}
}
```
---
## Database Schema
### AISuggestionFeedback Model
Stores user feedback on AI suggestions for accuracy tracking and model improvement.
**Fields:**
- `id` (BigAutoField): Primary key
- `document` (ForeignKey): Reference to Document
- `suggestion_type` (CharField): Type of suggestion (tag, correspondent, etc.)
- `suggested_value_id` (IntegerField, nullable): ID of suggested object
- `suggested_value_text` (TextField): Text representation of suggestion
- `confidence` (FloatField): AI confidence score (0.0 to 1.0)
- `status` (CharField): 'applied' or 'rejected'
- `user` (ForeignKey, nullable): User who provided feedback
- `created_at` (DateTimeField): When suggestion was created
- `applied_at` (DateTimeField): When feedback was recorded
- `metadata` (JSONField): Additional metadata
**Indexes:**
- `(document, suggestion_type)`
- `(status, created_at)`
- `(suggestion_type, status)`
---
## Best Practices
1. **Confidence Thresholds:**
- High confidence (≥ 0.80): Can be auto-applied
- Medium confidence (0.60-0.79): Show to user for review
- Low confidence (< 0.60): Log but don't suggest
2. **Error Handling:**
- Always handle 400, 404, and 500 errors gracefully
- Show user-friendly error messages
- Log errors for debugging
3. **Performance:**
- Cache suggestions when possible
- Use pagination for statistics endpoint if needed
- Batch apply/reject operations when possible
4. **User Experience:**
- Show confidence scores to users
- Allow users to modify suggestions before applying
- Provide feedback on applied/rejected actions
- Show statistics to demonstrate AI improvement over time
5. **Privacy:**
- Only authenticated users can access suggestions
- Users can only see suggestions for documents they have access to
- Feedback is tied to user accounts for accountability
---
## Troubleshooting
### No suggestions returned
- Verify document has content (document.content is not empty)
- Check if AI scanner is enabled in settings
- Verify ML models are loaded correctly
### Suggestions not being applied
- Check user permissions on the document
- Verify the suggested object (tag, correspondent, etc.) still exists
- Check application logs for detailed error messages
### Statistics showing 0 accuracy
- Ensure users are applying or rejecting suggestions
- Check database for AISuggestionFeedback entries
- Verify feedback is being recorded with correct status
---
## Future Enhancements
Potential improvements for future versions:
1. Bulk operations (apply/reject multiple suggestions at once)
2. Suggestion confidence threshold configuration per user
3. A/B testing different AI models
4. Machine learning model retraining based on feedback
5. Suggestion explanations (why AI made this suggestion)
6. Custom suggestion rules per user or organization
7. Integration with external AI services
8. Real-time suggestions via WebSocket
---
## Support
For issues or questions:
- GitHub Issues: https://github.com/dawnsystem/IntelliDocs-ngx/issues
- Documentation: https://docs.paperless-ngx.com
- Community: Matrix chat or forum
---
*Last updated: 2024-11-13*
*API Version: 1.0*

164
src/documents/migrations/1076_aisuggestionfeedback.py Normal file
View file

@ -0,0 +1,164 @@
# Generated manually for AI Suggestions API
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
import django.core.validators
class Migration(migrations.Migration):
"""
Add AISuggestionFeedback model for tracking user feedback on AI suggestions.
This model enables:
- Tracking of applied vs rejected AI suggestions
- Accuracy statistics and improvement of AI models
- User feedback analysis
"""
dependencies = [
("documents", "1075_add_performance_indexes"),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name="AISuggestionFeedback",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
(
"suggestion_type",
models.CharField(
choices=[
("tag", "Tag"),
("correspondent", "Correspondent"),
("document_type", "Document Type"),
("storage_path", "Storage Path"),
("custom_field", "Custom Field"),
("workflow", "Workflow"),
("title", "Title"),
],
max_length=50,
verbose_name="suggestion type",
),
),
(
"suggested_value_id",
models.IntegerField(
blank=True,
help_text="ID of the suggested object (tag, correspondent, etc.)",
null=True,
verbose_name="suggested value ID",
),
),
(
"suggested_value_text",
models.TextField(
blank=True,
help_text="Text representation of the suggested value",
verbose_name="suggested value text",
),
),
(
"confidence",
models.FloatField(
help_text="AI confidence score (0.0 to 1.0)",
validators=[
django.core.validators.MinValueValidator(0.0),
django.core.validators.MaxValueValidator(1.0),
],
verbose_name="confidence",
),
),
(
"status",
models.CharField(
choices=[
("applied", "Applied"),
("rejected", "Rejected"),
],
max_length=20,
verbose_name="status",
),
),
(
"created_at",
models.DateTimeField(
auto_now_add=True,
verbose_name="created at",
),
),
(
"applied_at",
models.DateTimeField(
auto_now=True,
verbose_name="applied/rejected at",
),
),
(
"metadata",
models.JSONField(
blank=True,
default=dict,
help_text="Additional metadata about the suggestion",
verbose_name="metadata",
),
),
(
"document",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="ai_suggestion_feedbacks",
to="documents.document",
verbose_name="document",
),
),
(
"user",
models.ForeignKey(
blank=True,
help_text="User who applied or rejected the suggestion",
null=True,
on_delete=django.db.models.deletion.SET_NULL,
related_name="ai_suggestion_feedbacks",
to=settings.AUTH_USER_MODEL,
verbose_name="user",
),
),
],
options={
"verbose_name": "AI suggestion feedback",
"verbose_name_plural": "AI suggestion feedbacks",
"ordering": ["-created_at"],
},
),
migrations.AddIndex(
model_name="aisuggestionfeedback",
index=models.Index(
fields=["document", "suggestion_type"],
name="documents_a_documen_idx",
),
),
migrations.AddIndex(
model_name="aisuggestionfeedback",
index=models.Index(
fields=["status", "created_at"],
name="documents_a_status_idx",
),
),
migrations.AddIndex(
model_name="aisuggestionfeedback",
index=models.Index(
fields=["suggestion_type", "status"],
name="documents_a_suggest_idx",
),
),
]

113
src/documents/models.py
View file

@ -1734,3 +1734,116 @@ class DeletionRequest(models.Model):
self.save() self.save()
return True return True
class AISuggestionFeedback(models.Model):
"""
Model to track user feedback on AI suggestions (applied/rejected).
Used for improving AI accuracy and providing statistics.
"""
# Suggestion types
TYPE_TAG = 'tag'
TYPE_CORRESPONDENT = 'correspondent'
TYPE_DOCUMENT_TYPE = 'document_type'
TYPE_STORAGE_PATH = 'storage_path'
TYPE_CUSTOM_FIELD = 'custom_field'
TYPE_WORKFLOW = 'workflow'
TYPE_TITLE = 'title'
SUGGESTION_TYPES = (
(TYPE_TAG, _('Tag')),
(TYPE_CORRESPONDENT, _('Correspondent')),
(TYPE_DOCUMENT_TYPE, _('Document Type')),
(TYPE_STORAGE_PATH, _('Storage Path')),
(TYPE_CUSTOM_FIELD, _('Custom Field')),
(TYPE_WORKFLOW, _('Workflow')),
(TYPE_TITLE, _('Title')),
)
# Feedback status
STATUS_APPLIED = 'applied'
STATUS_REJECTED = 'rejected'
FEEDBACK_STATUS = (
(STATUS_APPLIED, _('Applied')),
(STATUS_REJECTED, _('Rejected')),
)
document = models.ForeignKey(
Document,
on_delete=models.CASCADE,
related_name='ai_suggestion_feedbacks',
verbose_name=_('document'),
)
suggestion_type = models.CharField(
_('suggestion type'),
max_length=50,
choices=SUGGESTION_TYPES,
)
suggested_value_id = models.IntegerField(
_('suggested value ID'),
null=True,
blank=True,
help_text=_('ID of the suggested object (tag, correspondent, etc.)'),
)
suggested_value_text = models.TextField(
_('suggested value text'),
blank=True,
help_text=_('Text representation of the suggested value'),
)
confidence = models.FloatField(
_('confidence'),
help_text=_('AI confidence score (0.0 to 1.0)'),
validators=[MinValueValidator(0.0), MaxValueValidator(1.0)],
)
status = models.CharField(
_('status'),
max_length=20,
choices=FEEDBACK_STATUS,
)
user = models.ForeignKey(
User,
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name='ai_suggestion_feedbacks',
verbose_name=_('user'),
help_text=_('User who applied or rejected the suggestion'),
)
created_at = models.DateTimeField(
_('created at'),
auto_now_add=True,
)
applied_at = models.DateTimeField(
_('applied/rejected at'),
auto_now=True,
)
metadata = models.JSONField(
_('metadata'),
default=dict,
blank=True,
help_text=_('Additional metadata about the suggestion'),
)
class Meta:
verbose_name = _('AI suggestion feedback')
verbose_name_plural = _('AI suggestion feedbacks')
ordering = ['-created_at']
indexes = [
models.Index(fields=['document', 'suggestion_type']),
models.Index(fields=['status', 'created_at']),
models.Index(fields=['suggestion_type', 'status']),
]
def __str__(self):
return f"{self.suggestion_type} suggestion for document {self.document_id} - {self.status}"

17
src/documents/serializers/__init__.py Normal file
View file

@ -0,0 +1,17 @@
"""Serializers package for documents app."""
from .ai_suggestions import (
AISuggestionFeedbackSerializer,
AISuggestionsSerializer,
AISuggestionStatsSerializer,
ApplySuggestionSerializer,
RejectSuggestionSerializer,
)
__all__ = [
'AISuggestionFeedbackSerializer',
'AISuggestionsSerializer',
'AISuggestionStatsSerializer',
'ApplySuggestionSerializer',
'RejectSuggestionSerializer',
]

331
src/documents/serializers/ai_suggestions.py Normal file
View file

@ -0,0 +1,331 @@
"""
Serializers for AI Suggestions API.
This module provides serializers for exposing AI scanner results
and handling user feedback on AI suggestions.
"""
from __future__ import annotations
from typing import Any, Dict
from rest_framework import serializers
from documents.models import (
AISuggestionFeedback,
Correspondent,
CustomField,
DocumentType,
StoragePath,
Tag,
Workflow,
)
# Suggestion type choices - used across multiple serializers
SUGGESTION_TYPE_CHOICES = [
'tag',
'correspondent',
'document_type',
'storage_path',
'custom_field',
'workflow',
'title',
]
# Types that require value_id
ID_REQUIRED_TYPES = ['tag', 'correspondent', 'document_type', 'storage_path', 'workflow']
# Types that require value_text
TEXT_REQUIRED_TYPES = ['title']
# Types that can use either (custom_field can be ID or text)
class TagSuggestionSerializer(serializers.Serializer):
"""Serializer for tag suggestions."""
id = serializers.IntegerField()
name = serializers.CharField()
color = serializers.CharField()
confidence = serializers.FloatField()
class CorrespondentSuggestionSerializer(serializers.Serializer):
"""Serializer for correspondent suggestions."""
id = serializers.IntegerField()
name = serializers.CharField()
confidence = serializers.FloatField()
class DocumentTypeSuggestionSerializer(serializers.Serializer):
"""Serializer for document type suggestions."""
id = serializers.IntegerField()
name = serializers.CharField()
confidence = serializers.FloatField()
class StoragePathSuggestionSerializer(serializers.Serializer):
"""Serializer for storage path suggestions."""
id = serializers.IntegerField()
name = serializers.CharField()
path = serializers.CharField()
confidence = serializers.FloatField()
class CustomFieldSuggestionSerializer(serializers.Serializer):
"""Serializer for custom field suggestions."""
field_id = serializers.IntegerField()
field_name = serializers.CharField()
value = serializers.CharField()
confidence = serializers.FloatField()
class WorkflowSuggestionSerializer(serializers.Serializer):
"""Serializer for workflow suggestions."""
id = serializers.IntegerField()
name = serializers.CharField()
confidence = serializers.FloatField()
class TitleSuggestionSerializer(serializers.Serializer):
"""Serializer for title suggestions."""
title = serializers.CharField()
class AISuggestionsSerializer(serializers.Serializer):
"""
Main serializer for AI scan results.
Converts AIScanResult objects to JSON format for API responses.
"""
tags = TagSuggestionSerializer(many=True, required=False)
correspondent = CorrespondentSuggestionSerializer(required=False, allow_null=True)
document_type = DocumentTypeSuggestionSerializer(required=False, allow_null=True)
storage_path = StoragePathSuggestionSerializer(required=False, allow_null=True)
custom_fields = CustomFieldSuggestionSerializer(many=True, required=False)
workflows = WorkflowSuggestionSerializer(many=True, required=False)
title_suggestion = TitleSuggestionSerializer(required=False, allow_null=True)
@staticmethod
def from_scan_result(scan_result, document_id: int) -> Dict[str, Any]:
"""
Convert an AIScanResult object to serializer data.
Args:
scan_result: AIScanResult instance from ai_scanner
document_id: Document ID for reference
Returns:
Dictionary ready for serialization
"""
data = {}
# Tags
if scan_result.tags:
tag_suggestions = []
for tag_id, confidence in scan_result.tags:
try:
tag = Tag.objects.get(pk=tag_id)
tag_suggestions.append({
'id': tag.id,
'name': tag.name,
'color': getattr(tag, 'color', '#000000'),
'confidence': confidence,
})
except Tag.DoesNotExist:
# Tag no longer exists in database; skip this suggestion
pass
data['tags'] = tag_suggestions
# Correspondent
if scan_result.correspondent:
corr_id, confidence = scan_result.correspondent
try:
correspondent = Correspondent.objects.get(pk=corr_id)
data['correspondent'] = {
'id': correspondent.id,
'name': correspondent.name,
'confidence': confidence,
}
except Correspondent.DoesNotExist:
# Correspondent no longer exists in database; omit from suggestions
pass
# Document Type
if scan_result.document_type:
type_id, confidence = scan_result.document_type
try:
doc_type = DocumentType.objects.get(pk=type_id)
data['document_type'] = {
'id': doc_type.id,
'name': doc_type.name,
'confidence': confidence,
}
except DocumentType.DoesNotExist:
# Document type no longer exists in database; omit from suggestions
pass
# Storage Path
if scan_result.storage_path:
path_id, confidence = scan_result.storage_path
try:
storage_path = StoragePath.objects.get(pk=path_id)
data['storage_path'] = {
'id': storage_path.id,
'name': storage_path.name,
'path': storage_path.path,
'confidence': confidence,
}
except StoragePath.DoesNotExist:
# Storage path no longer exists in database; omit from suggestions
pass
# Custom Fields
if scan_result.custom_fields:
field_suggestions = []
for field_id, (value, confidence) in scan_result.custom_fields.items():
try:
field = CustomField.objects.get(pk=field_id)
field_suggestions.append({
'field_id': field.id,
'field_name': field.name,
'value': str(value),
'confidence': confidence,
})
except CustomField.DoesNotExist:
# Custom field no longer exists in database; skip this suggestion
pass
data['custom_fields'] = field_suggestions
# Workflows
if scan_result.workflows:
workflow_suggestions = []
for workflow_id, confidence in scan_result.workflows:
try:
workflow = Workflow.objects.get(pk=workflow_id)
workflow_suggestions.append({
'id': workflow.id,
'name': workflow.name,
'confidence': confidence,
})
except Workflow.DoesNotExist:
# Workflow no longer exists in database; skip this suggestion
pass
data['workflows'] = workflow_suggestions
# Title suggestion
if scan_result.title_suggestion:
data['title_suggestion'] = {
'title': scan_result.title_suggestion,
}
return data
class SuggestionSerializerMixin:
"""
Mixin to provide validation logic for suggestion serializers.
"""
def validate(self, attrs):
"""Validate that the correct value field is provided for the suggestion type."""
suggestion_type = attrs.get('suggestion_type')
value_id = attrs.get('value_id')
value_text = attrs.get('value_text')
# Types that require value_id
if suggestion_type in ID_REQUIRED_TYPES and not value_id:
raise serializers.ValidationError(
f"value_id is required for suggestion_type '{suggestion_type}'"
)
# Types that require value_text
if suggestion_type in TEXT_REQUIRED_TYPES and not value_text:
raise serializers.ValidationError(
f"value_text is required for suggestion_type '{suggestion_type}'"
)
# For custom_field, either is acceptable
if suggestion_type == 'custom_field' and not value_id and not value_text:
raise serializers.ValidationError(
"Either value_id or value_text must be provided for custom_field"
)
return attrs
class ApplySuggestionSerializer(SuggestionSerializerMixin, serializers.Serializer):
"""
Serializer for applying AI suggestions.
"""
suggestion_type = serializers.ChoiceField(
choices=SUGGESTION_TYPE_CHOICES,
required=True,
)
value_id = serializers.IntegerField(required=False, allow_null=True)
value_text = serializers.CharField(required=False, allow_blank=True)
confidence = serializers.FloatField(required=True)
class RejectSuggestionSerializer(SuggestionSerializerMixin, serializers.Serializer):
"""
Serializer for rejecting AI suggestions.
"""
suggestion_type = serializers.ChoiceField(
choices=SUGGESTION_TYPE_CHOICES,
required=True,
)
value_id = serializers.IntegerField(required=False, allow_null=True)
value_text = serializers.CharField(required=False, allow_blank=True)
confidence = serializers.FloatField(required=True)
class AISuggestionFeedbackSerializer(serializers.ModelSerializer):
"""Serializer for AI suggestion feedback model."""
class Meta:
model = AISuggestionFeedback
fields = [
'id',
'document',
'suggestion_type',
'suggested_value_id',
'suggested_value_text',
'confidence',
'status',
'user',
'created_at',
'applied_at',
'metadata',
]
read_only_fields = ['id', 'created_at', 'applied_at']
class AISuggestionStatsSerializer(serializers.Serializer):
"""
Serializer for AI suggestion accuracy statistics.
"""
total_suggestions = serializers.IntegerField()
total_applied = serializers.IntegerField()
total_rejected = serializers.IntegerField()
accuracy_rate = serializers.FloatField()
by_type = serializers.DictField(
child=serializers.DictField(),
help_text="Statistics broken down by suggestion type",
)
average_confidence_applied = serializers.FloatField()
average_confidence_rejected = serializers.FloatField()
recent_suggestions = AISuggestionFeedbackSerializer(many=True, required=False)

462
src/documents/tests/test_api_ai_suggestions.py Normal file
View file

@ -0,0 +1,462 @@
"""
Tests for AI Suggestions API endpoints.
"""
from unittest import mock
from django.contrib.auth.models import User
from rest_framework import status
from rest_framework.test import APITestCase
from documents.ai_scanner import AIScanResult
from documents.models import (
AISuggestionFeedback,
Correspondent,
Document,
DocumentType,
StoragePath,
Tag,
)
from documents.tests.utils import DirectoriesMixin
class TestAISuggestionsAPI(DirectoriesMixin, APITestCase):
"""Test cases for AI suggestions API endpoints."""
def setUp(self):
super().setUp()
# Create test user
self.user = User.objects.create_superuser(username="test_admin")
self.client.force_authenticate(user=self.user)
# Create test data
self.correspondent = Correspondent.objects.create(
name="Test Corp",
pk=1,
)
self.doc_type = DocumentType.objects.create(
name="Invoice",
pk=1,
)
self.tag1 = Tag.objects.create(
name="Important",
pk=1,
)
self.tag2 = Tag.objects.create(
name="Urgent",
pk=2,
)
self.storage_path = StoragePath.objects.create(
name="Archive",
path="/archive/",
pk=1,
)
# Create test document
self.document = Document.objects.create(
title="Test Document",
content="This is a test document with some content for AI analysis.",
checksum="abc123",
mime_type="application/pdf",
)
def test_ai_suggestions_endpoint_exists(self):
"""Test that the ai-suggestions endpoint is accessible."""
response = self.client.get(
f"/api/documents/{self.document.pk}/ai-suggestions/"
)
# Should not be 404
self.assertNotEqual(response.status_code, status.HTTP_404_NOT_FOUND)
@mock.patch('documents.ai_scanner.get_ai_scanner')
def test_get_ai_suggestions_success(self, mock_get_scanner):
"""Test successfully getting AI suggestions for a document."""
# Create mock scan result
scan_result = AIScanResult()
scan_result.tags = [(self.tag1.id, 0.85), (self.tag2.id, 0.75)]
scan_result.correspondent = (self.correspondent.id, 0.90)
scan_result.document_type = (self.doc_type.id, 0.88)
scan_result.storage_path = (self.storage_path.id, 0.80)
scan_result.title_suggestion = "Suggested Title"
# Mock scanner
mock_scanner = mock.Mock()
mock_scanner.scan_document.return_value = scan_result
mock_get_scanner.return_value = mock_scanner
# Make request
response = self.client.get(
f"/api/documents/{self.document.pk}/ai-suggestions/"
)
# Verify response
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
# Check tags
self.assertIn('tags', data)
self.assertEqual(len(data['tags']), 2)
self.assertEqual(data['tags'][0]['id'], self.tag1.id)
self.assertEqual(data['tags'][0]['confidence'], 0.85)
# Check correspondent
self.assertIn('correspondent', data)
self.assertEqual(data['correspondent']['id'], self.correspondent.id)
self.assertEqual(data['correspondent']['confidence'], 0.90)
# Check document type
self.assertIn('document_type', data)
self.assertEqual(data['document_type']['id'], self.doc_type.id)
# Check title suggestion
self.assertIn('title_suggestion', data)
self.assertEqual(data['title_suggestion']['title'], "Suggested Title")
def test_get_ai_suggestions_no_content(self):
"""Test getting AI suggestions for document without content."""
# Create document without content
doc = Document.objects.create(
title="Empty Document",
content="",
checksum="empty123",
mime_type="application/pdf",
)
response = self.client.get(f"/api/documents/{doc.pk}/ai-suggestions/")
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("no content", response.json()['detail'].lower())
def test_get_ai_suggestions_document_not_found(self):
"""Test getting AI suggestions for non-existent document."""
response = self.client.get("/api/documents/99999/ai-suggestions/")
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
def test_apply_suggestion_tag(self):
"""Test applying a tag suggestion."""
request_data = {
'suggestion_type': 'tag',
'value_id': self.tag1.id,
'confidence': 0.85,
}
response = self.client.post(
f"/api/documents/{self.document.pk}/apply-suggestion/",
data=request_data,
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.json()['status'], 'success')
# Verify tag was applied
self.document.refresh_from_db()
self.assertIn(self.tag1, self.document.tags.all())
# Verify feedback was recorded
feedback = AISuggestionFeedback.objects.filter(
document=self.document,
suggestion_type='tag',
).first()
self.assertIsNotNone(feedback)
self.assertEqual(feedback.status, AISuggestionFeedback.STATUS_APPLIED)
self.assertEqual(feedback.suggested_value_id, self.tag1.id)
self.assertEqual(feedback.confidence, 0.85)
self.assertEqual(feedback.user, self.user)
def test_apply_suggestion_correspondent(self):
"""Test applying a correspondent suggestion."""
request_data = {
'suggestion_type': 'correspondent',
'value_id': self.correspondent.id,
'confidence': 0.90,
}
response = self.client.post(
f"/api/documents/{self.document.pk}/apply-suggestion/",
data=request_data,
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
# Verify correspondent was applied
self.document.refresh_from_db()
self.assertEqual(self.document.correspondent, self.correspondent)
# Verify feedback was recorded
feedback = AISuggestionFeedback.objects.filter(
document=self.document,
suggestion_type='correspondent',
).first()
self.assertIsNotNone(feedback)
self.assertEqual(feedback.status, AISuggestionFeedback.STATUS_APPLIED)
def test_apply_suggestion_document_type(self):
"""Test applying a document type suggestion."""
request_data = {
'suggestion_type': 'document_type',
'value_id': self.doc_type.id,
'confidence': 0.88,
}
response = self.client.post(
f"/api/documents/{self.document.pk}/apply-suggestion/",
data=request_data,
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
# Verify document type was applied
self.document.refresh_from_db()
self.assertEqual(self.document.document_type, self.doc_type)
def test_apply_suggestion_title(self):
"""Test applying a title suggestion."""
request_data = {
'suggestion_type': 'title',
'value_text': 'New Suggested Title',
'confidence': 0.80,
}
response = self.client.post(
f"/api/documents/{self.document.pk}/apply-suggestion/",
data=request_data,
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
# Verify title was applied
self.document.refresh_from_db()
self.assertEqual(self.document.title, 'New Suggested Title')
def test_apply_suggestion_invalid_type(self):
"""Test applying suggestion with invalid type."""
request_data = {
'suggestion_type': 'invalid_type',
'value_id': 1,
'confidence': 0.85,
}
response = self.client.post(
f"/api/documents/{self.document.pk}/apply-suggestion/",
data=request_data,
format='json',
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
def test_apply_suggestion_missing_value(self):
"""Test applying suggestion without value_id or value_text."""
request_data = {
'suggestion_type': 'tag',
'confidence': 0.85,
}
response = self.client.post(
f"/api/documents/{self.document.pk}/apply-suggestion/",
data=request_data,
format='json',
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
def test_apply_suggestion_nonexistent_object(self):
"""Test applying suggestion with non-existent object ID."""
request_data = {
'suggestion_type': 'tag',
'value_id': 99999,
'confidence': 0.85,
}
response = self.client.post(
f"/api/documents/{self.document.pk}/apply-suggestion/",
data=request_data,
format='json',
)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
def test_reject_suggestion(self):
"""Test rejecting an AI suggestion."""
request_data = {
'suggestion_type': 'tag',
'value_id': self.tag1.id,
'confidence': 0.65,
}
response = self.client.post(
f"/api/documents/{self.document.pk}/reject-suggestion/",
data=request_data,
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.json()['status'], 'success')
# Verify feedback was recorded
feedback = AISuggestionFeedback.objects.filter(
document=self.document,
suggestion_type='tag',
).first()
self.assertIsNotNone(feedback)
self.assertEqual(feedback.status, AISuggestionFeedback.STATUS_REJECTED)
self.assertEqual(feedback.suggested_value_id, self.tag1.id)
self.assertEqual(feedback.confidence, 0.65)
self.assertEqual(feedback.user, self.user)
def test_reject_suggestion_with_text(self):
"""Test rejecting a suggestion with text value."""
request_data = {
'suggestion_type': 'title',
'value_text': 'Bad Title Suggestion',
'confidence': 0.50,
}
response = self.client.post(
f"/api/documents/{self.document.pk}/reject-suggestion/",
data=request_data,
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
# Verify feedback was recorded
feedback = AISuggestionFeedback.objects.filter(
document=self.document,
suggestion_type='title',
).first()
self.assertIsNotNone(feedback)
self.assertEqual(feedback.status, AISuggestionFeedback.STATUS_REJECTED)
self.assertEqual(feedback.suggested_value_text, 'Bad Title Suggestion')
def test_ai_suggestion_stats_empty(self):
"""Test getting statistics when no feedback exists."""
response = self.client.get("/api/documents/ai-suggestion-stats/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data['total_suggestions'], 0)
self.assertEqual(data['total_applied'], 0)
self.assertEqual(data['total_rejected'], 0)
self.assertEqual(data['accuracy_rate'], 0)
def test_ai_suggestion_stats_with_data(self):
"""Test getting statistics with feedback data."""
# Create some feedback entries
AISuggestionFeedback.objects.create(
document=self.document,
suggestion_type='tag',
suggested_value_id=self.tag1.id,
confidence=0.85,
status=AISuggestionFeedback.STATUS_APPLIED,
user=self.user,
)
AISuggestionFeedback.objects.create(
document=self.document,
suggestion_type='tag',
suggested_value_id=self.tag2.id,
confidence=0.70,
status=AISuggestionFeedback.STATUS_APPLIED,
user=self.user,
)
AISuggestionFeedback.objects.create(
document=self.document,
suggestion_type='correspondent',
suggested_value_id=self.correspondent.id,
confidence=0.60,
status=AISuggestionFeedback.STATUS_REJECTED,
user=self.user,
)
response = self.client.get("/api/documents/ai-suggestion-stats/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
# Check overall stats
self.assertEqual(data['total_suggestions'], 3)
self.assertEqual(data['total_applied'], 2)
self.assertEqual(data['total_rejected'], 1)
self.assertAlmostEqual(data['accuracy_rate'], 66.67, places=1)
# Check by_type stats
self.assertIn('by_type', data)
self.assertIn('tag', data['by_type'])
self.assertEqual(data['by_type']['tag']['total'], 2)
self.assertEqual(data['by_type']['tag']['applied'], 2)
self.assertEqual(data['by_type']['tag']['rejected'], 0)
# Check confidence averages
self.assertGreater(data['average_confidence_applied'], 0)
self.assertGreater(data['average_confidence_rejected'], 0)
# Check recent suggestions
self.assertIn('recent_suggestions', data)
self.assertEqual(len(data['recent_suggestions']), 3)
def test_ai_suggestion_stats_accuracy_calculation(self):
"""Test that accuracy rate is calculated correctly."""
# Create 7 applied and 3 rejected = 70% accuracy
for i in range(7):
AISuggestionFeedback.objects.create(
document=self.document,
suggestion_type='tag',
suggested_value_id=self.tag1.id,
confidence=0.80,
status=AISuggestionFeedback.STATUS_APPLIED,
user=self.user,
)
for i in range(3):
AISuggestionFeedback.objects.create(
document=self.document,
suggestion_type='tag',
suggested_value_id=self.tag2.id,
confidence=0.60,
status=AISuggestionFeedback.STATUS_REJECTED,
user=self.user,
)
response = self.client.get("/api/documents/ai-suggestion-stats/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data['total_suggestions'], 10)
self.assertEqual(data['total_applied'], 7)
self.assertEqual(data['total_rejected'], 3)
self.assertEqual(data['accuracy_rate'], 70.0)
def test_authentication_required(self):
"""Test that authentication is required for all endpoints."""
self.client.force_authenticate(user=None)
# Test ai-suggestions endpoint
response = self.client.get(
f"/api/documents/{self.document.pk}/ai-suggestions/"
)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
# Test apply-suggestion endpoint
response = self.client.post(
f"/api/documents/{self.document.pk}/apply-suggestion/",
data={},
)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
# Test reject-suggestion endpoint
response = self.client.post(
f"/api/documents/{self.document.pk}/reject-suggestion/",
data={},
)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
# Test stats endpoint
response = self.client.get("/api/documents/ai-suggestion-stats/")
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

273
src/documents/views.py
View file

@ -1358,6 +1358,279 @@ class UnifiedSearchViewSet(DocumentViewSet):
) )
return Response(max_asn + 1) return Response(max_asn + 1)
@action(detail=True, methods=["GET"], name="Get AI Suggestions")
def ai_suggestions(self, request, pk=None):
"""
Get AI suggestions for a document.
Returns AI-generated suggestions for tags, correspondent, document type,
storage path, custom fields, workflows, and title.
"""
from documents.ai_scanner import get_ai_scanner
from documents.serializers.ai_suggestions import AISuggestionsSerializer
try:
document = self.get_object()
# Check if document has content to scan
if not document.content:
return Response(
{"detail": "Document has no content to analyze"},
status=400,
)
# Get AI scanner instance
scanner = get_ai_scanner()
# Perform AI scan
scan_result = scanner.scan_document(
document=document,
document_text=document.content,
original_file_path=document.source_path if hasattr(document, 'source_path') else None,
)
# Convert scan result to serializable format
data = AISuggestionsSerializer.from_scan_result(scan_result, document.id)
# Serialize and return
serializer = AISuggestionsSerializer(data=data)
serializer.is_valid(raise_exception=True)
return Response(serializer.validated_data)
except Exception as e:
logger.error(f"Error getting AI suggestions for document {pk}: {e}", exc_info=True)
return Response(
{"detail": "Error generating AI suggestions. Please check the logs for details."},
status=500,
)
@action(detail=True, methods=["POST"], name="Apply AI Suggestion")
def apply_suggestion(self, request, pk=None):
"""
Apply an AI suggestion to a document.
Records user feedback and applies the suggested change.
"""
from documents.models import AISuggestionFeedback
from documents.serializers.ai_suggestions import ApplySuggestionSerializer
try:
document = self.get_object()
# Validate input
serializer = ApplySuggestionSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
suggestion_type = serializer.validated_data['suggestion_type']
value_id = serializer.validated_data.get('value_id')
value_text = serializer.validated_data.get('value_text')
confidence = serializer.validated_data['confidence']
# Apply the suggestion based on type
applied = False
result_message = ""
if suggestion_type == 'tag' and value_id:
tag = Tag.objects.get(pk=value_id)
document.tags.add(tag)
applied = True
result_message = f"Tag '{tag.name}' applied"
elif suggestion_type == 'correspondent' and value_id:
correspondent = Correspondent.objects.get(pk=value_id)
document.correspondent = correspondent
document.save()
applied = True
result_message = f"Correspondent '{correspondent.name}' applied"
elif suggestion_type == 'document_type' and value_id:
doc_type = DocumentType.objects.get(pk=value_id)
document.document_type = doc_type
document.save()
applied = True
result_message = f"Document type '{doc_type.name}' applied"
elif suggestion_type == 'storage_path' and value_id:
storage_path = StoragePath.objects.get(pk=value_id)
document.storage_path = storage_path
document.save()
applied = True
result_message = f"Storage path '{storage_path.name}' applied"
elif suggestion_type == 'title' and value_text:
document.title = value_text
document.save()
applied = True
result_message = f"Title updated to '{value_text}'"
if applied:
# Record feedback
AISuggestionFeedback.objects.create(
document=document,
suggestion_type=suggestion_type,
suggested_value_id=value_id,
suggested_value_text=value_text or "",
confidence=confidence,
status=AISuggestionFeedback.STATUS_APPLIED,
user=request.user,
)
return Response({
"status": "success",
"message": result_message,
})
else:
return Response(
{"detail": "Invalid suggestion type or missing value"},
status=400,
)
except (Tag.DoesNotExist, Correspondent.DoesNotExist,
DocumentType.DoesNotExist, StoragePath.DoesNotExist):
return Response(
{"detail": "Referenced object not found"},
status=404,
)
except Exception as e:
logger.error(f"Error applying suggestion for document {pk}: {e}", exc_info=True)
return Response(
{"detail": "Error applying suggestion. Please check the logs for details."},
status=500,
)
@action(detail=True, methods=["POST"], name="Reject AI Suggestion")
def reject_suggestion(self, request, pk=None):
"""
Reject an AI suggestion for a document.
Records user feedback for improving AI accuracy.
"""
from documents.models import AISuggestionFeedback
from documents.serializers.ai_suggestions import RejectSuggestionSerializer
try:
document = self.get_object()
# Validate input
serializer = RejectSuggestionSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
suggestion_type = serializer.validated_data['suggestion_type']
value_id = serializer.validated_data.get('value_id')
value_text = serializer.validated_data.get('value_text')
confidence = serializer.validated_data['confidence']
# Record feedback
AISuggestionFeedback.objects.create(
document=document,
suggestion_type=suggestion_type,
suggested_value_id=value_id,
suggested_value_text=value_text or "",
confidence=confidence,
status=AISuggestionFeedback.STATUS_REJECTED,
user=request.user,
)
return Response({
"status": "success",
"message": "Suggestion rejected and feedback recorded",
})
except Exception as e:
logger.error(f"Error rejecting suggestion for document {pk}: {e}", exc_info=True)
return Response(
{"detail": "Error rejecting suggestion. Please check the logs for details."},
status=500,
)
@action(detail=False, methods=["GET"], name="AI Suggestion Statistics")
def ai_suggestion_stats(self, request):
"""
Get statistics about AI suggestion accuracy.
Returns aggregated data about applied vs rejected suggestions,
accuracy rates, and confidence scores.
"""
from django.db.models import Avg, Count, Q
from documents.models import AISuggestionFeedback
from documents.serializers.ai_suggestions import AISuggestionStatsSerializer
try:
# Get overall counts
total_feedbacks = AISuggestionFeedback.objects.count()
total_applied = AISuggestionFeedback.objects.filter(
status=AISuggestionFeedback.STATUS_APPLIED
).count()
total_rejected = AISuggestionFeedback.objects.filter(
status=AISuggestionFeedback.STATUS_REJECTED
).count()
# Calculate accuracy rate
accuracy_rate = (total_applied / total_feedbacks * 100) if total_feedbacks > 0 else 0
# Get statistics by suggestion type using a single aggregated query
stats_by_type = AISuggestionFeedback.objects.values('suggestion_type').annotate(
total=Count('id'),
applied=Count('id', filter=Q(status=AISuggestionFeedback.STATUS_APPLIED)),
rejected=Count('id', filter=Q(status=AISuggestionFeedback.STATUS_REJECTED))
)
# Build the by_type dictionary using the aggregated results
by_type = {}
for stat in stats_by_type:
suggestion_type = stat['suggestion_type']
type_total = stat['total']
type_applied = stat['applied']
type_rejected = stat['rejected']
by_type[suggestion_type] = {
'total': type_total,
'applied': type_applied,
'rejected': type_rejected,
'accuracy_rate': (type_applied / type_total * 100) if type_total > 0 else 0,
}
# Get average confidence scores
avg_confidence_applied = AISuggestionFeedback.objects.filter(
status=AISuggestionFeedback.STATUS_APPLIED
).aggregate(Avg('confidence'))['confidence__avg'] or 0.0
avg_confidence_rejected = AISuggestionFeedback.objects.filter(
status=AISuggestionFeedback.STATUS_REJECTED
).aggregate(Avg('confidence'))['confidence__avg'] or 0.0
# Get recent suggestions (last 10)
recent_suggestions = AISuggestionFeedback.objects.order_by('-created_at')[:10]
# Build response data
from documents.serializers.ai_suggestions import AISuggestionFeedbackSerializer
data = {
'total_suggestions': total_feedbacks,
'total_applied': total_applied,
'total_rejected': total_rejected,
'accuracy_rate': accuracy_rate,
'by_type': by_type,
'average_confidence_applied': avg_confidence_applied,
'average_confidence_rejected': avg_confidence_rejected,
'recent_suggestions': AISuggestionFeedbackSerializer(
recent_suggestions, many=True
).data,
}
# Serialize and return
serializer = AISuggestionStatsSerializer(data=data)
serializer.is_valid(raise_exception=True)
return Response(serializer.validated_data)
except Exception as e:
logger.error(f"Error getting AI suggestion statistics: {e}", exc_info=True)
return Response(
{"detail": "Error getting statistics. Please check the logs for details."},
status=500,
)
@extend_schema_view( @extend_schema_view(
list=extend_schema( list=extend_schema(