Files
DEMO-AGENT/review_agent/views.py

328 lines
12 KiB
Python

from django.contrib.auth.decorators import login_required
from django.db.models import Count, Q
import json
from django.http import HttpRequest, HttpResponse, JsonResponse, StreamingHttpResponse
from django.shortcuts import redirect, render
from django.views.decorators.http import require_http_methods
from .services import (
create_conversation,
get_conversation_for_user,
list_conversations,
send_message,
stream_message,
)
from .models import ApplicationFormFillBatch, Conversation, FileAttachment, FileSummaryBatch, RegulatoryReviewBatch, WorkflowNodeRun
from .knowledge_base import build_knowledge_base_context, search_knowledge_base
from .knowledge_base import (
build_knowledge_base_context_for_user,
create_document_from_upload,
delete_document,
index_managed_document,
list_documents_for_user,
serialize_document,
update_document,
)
from .models import KnowledgeBaseDocument
from .regulatory_review.services.info_extract import ensure_regulatory_condition_candidates
@login_required
@require_http_methods(["GET", "POST"])
def workspace(request: HttpRequest) -> HttpResponse:
"""Renders the review-agent workspace and handles conversation actions."""
if request.method == "POST":
action = request.POST.get("action")
conversation = get_conversation_for_user(request.user, request.POST.get("conversation_id"))
if action == "new_conversation":
conversation = create_conversation(request.user)
return redirect(f"/?conversation={conversation.pk}")
if action == "send_message":
content = (request.POST.get("prompt") or "").strip()
if not conversation:
conversation = create_conversation(request.user)
if content:
send_message(conversation, content)
return redirect(f"/?conversation={conversation.pk}")
search = (request.GET.get("q") or "").strip()
conversations = list_conversations(request.user, search)
current = get_conversation_for_user(request.user, request.GET.get("conversation"))
if current is None and conversations.exists():
current = conversations.first()
workflow_cards = build_workflow_cards(current) if current else []
condition_confirmation = build_condition_confirmation(workflow_cards)
return render(
request,
"home.html",
{
"page_title": "审核智能体",
"search_query": search,
"conversations": conversations,
"current_conversation": current,
"messages": current.messages.all() if current else [],
"attachments": FileAttachment.objects.filter(conversation=current).order_by("original_name", "-version_no") if current else [],
"workflow_cards": workflow_cards,
"condition_confirmation": condition_confirmation,
},
)
@login_required
@require_http_methods(["GET"])
def attachment_manager(request: HttpRequest) -> HttpResponse:
conversations = (
Conversation.objects.filter(user=request.user)
.annotate(
attachment_count=Count(
"file_attachments",
filter=~Q(file_attachments__upload_status=FileAttachment.UploadStatus.DELETED),
)
)
.order_by("-updated_at", "-id")
)
selected = get_conversation_for_user(request.user, request.GET.get("conversation"))
attachments = (
FileAttachment.objects.filter(conversation=selected)
.order_by("original_name", "-version_no")
if selected
else []
)
return render(
request,
"attachment_manager.html",
{
"page_title": "附件管理",
"conversations": conversations,
"selected_conversation": selected,
"attachments": attachments,
},
)
@login_required
@require_http_methods(["GET"])
def knowledge_base_manager(request: HttpRequest) -> HttpResponse:
context = build_knowledge_base_context_for_user(request.user)
return render(
request,
"knowledge_base.html",
{
"page_title": "知识库管理",
"knowledge_base": context,
},
)
@login_required
@require_http_methods(["GET"])
def knowledge_base_status(request: HttpRequest) -> JsonResponse:
return JsonResponse(build_knowledge_base_context_for_user(request.user))
@login_required
@require_http_methods(["POST"])
def knowledge_base_search(request: HttpRequest) -> JsonResponse:
if request.content_type == "application/json":
try:
payload = json.loads(request.body.decode("utf-8") or "{}")
except json.JSONDecodeError:
payload = {}
query = payload.get("query", "")
else:
query = request.POST.get("query", "")
return JsonResponse(search_knowledge_base(str(query)))
@login_required
@require_http_methods(["GET", "POST"])
def knowledge_base_documents(request: HttpRequest) -> JsonResponse:
if request.method == "GET":
return JsonResponse({"documents": list_documents_for_user(request.user)})
uploaded_file = request.FILES.get("file")
if uploaded_file is None:
return JsonResponse({"error": "请上传知识库材料。"}, status=400)
is_active = str(request.POST.get("is_active", "true")).lower() not in {"0", "false", "off"}
document = create_document_from_upload(
user=request.user,
uploaded_file=uploaded_file,
display_name=request.POST.get("display_name", ""),
description=request.POST.get("description", ""),
is_active=is_active,
)
return JsonResponse({"document": serialize_document(document)})
@login_required
@require_http_methods(["GET", "PATCH", "DELETE"])
def knowledge_base_document_detail(request: HttpRequest, document_id: int) -> JsonResponse:
try:
document = KnowledgeBaseDocument.objects.get(
pk=document_id,
user=request.user,
)
except KnowledgeBaseDocument.DoesNotExist:
return JsonResponse({"error": "知识库材料不存在。"}, status=404)
if document.status == KnowledgeBaseDocument.Status.DELETED:
return JsonResponse({"error": "知识库材料不存在。"}, status=404)
if request.method == "GET":
return JsonResponse({"document": serialize_document(document)})
if request.method == "DELETE":
delete_document(document)
return JsonResponse({"document": serialize_document(document)})
try:
payload = json.loads(request.body.decode("utf-8") or "{}")
except json.JSONDecodeError:
payload = {}
update_document(document, payload)
return JsonResponse({"document": serialize_document(document)})
@login_required
@require_http_methods(["POST"])
def knowledge_base_document_index(request: HttpRequest, document_id: int) -> JsonResponse:
try:
document = KnowledgeBaseDocument.objects.get(
pk=document_id,
user=request.user,
)
except KnowledgeBaseDocument.DoesNotExist:
return JsonResponse({"error": "知识库材料不存在。"}, status=404)
if document.status == KnowledgeBaseDocument.Status.DELETED:
return JsonResponse({"error": "知识库材料不存在。"}, status=404)
chunk_count = index_managed_document(document)
document.refresh_from_db()
return JsonResponse({"document": serialize_document(document), "chunk_count": chunk_count})
@login_required
@require_http_methods(["POST"])
def stream_chat(request: HttpRequest) -> HttpResponse:
"""Streams one assistant reply so the UI can render incremental output."""
content = (request.POST.get("prompt") or "").strip()
if not content:
return JsonResponse({"error": "消息内容不能为空。"}, status=400)
conversation = get_conversation_for_user(request.user, request.POST.get("conversation_id"))
if not conversation:
conversation = create_conversation(request.user)
response = StreamingHttpResponse(
streaming_content=stream_message(conversation, content),
content_type="text/event-stream",
)
response["Cache-Control"] = "no-cache"
response["X-Accel-Buffering"] = "no"
return response
def build_workflow_cards(conversation: Conversation) -> list[dict[str, object]]:
cards: list[dict[str, object]] = []
for batch in FileSummaryBatch.objects.filter(conversation=conversation).prefetch_related("node_runs"):
cards.append(
{
"id": batch.pk,
"workflow_type": "file_summary",
"batch_no": batch.batch_no,
"status": batch.status,
"error_message": batch.error_message,
"risk_label": "",
"created_at": batch.created_at,
"nodes": list(batch.node_runs.order_by("id")),
}
)
regulatory_batches = RegulatoryReviewBatch.objects.filter(conversation=conversation)
for batch in regulatory_batches:
condition_candidates = ensure_regulatory_condition_candidates(batch)
cards.append(
{
"id": batch.pk,
"workflow_type": "regulatory_review",
"batch_no": batch.batch_no,
"status": batch.status,
"error_message": batch.error_message,
"risk_label": _format_risk_label(batch.risk_summary or {}),
"condition_json": batch.condition_json or {},
"condition_candidates": condition_candidates,
"notification_count": batch.notifications.count(),
"review_record_count": batch.artifacts.filter(metadata__artifact="review_record").count(),
"created_at": batch.created_at,
"nodes": list(
WorkflowNodeRun.objects.filter(
workflow_type="regulatory_review",
workflow_batch_id=batch.pk,
).order_by("id")
),
}
)
form_fill_batches = ApplicationFormFillBatch.objects.filter(conversation=conversation, is_deleted=False)
for batch in form_fill_batches:
cards.append(
{
"id": batch.pk,
"workflow_type": "application_form_fill",
"batch_no": batch.batch_no,
"status": batch.status,
"error_message": batch.error_message,
"risk_label": _format_form_fill_label(batch),
"created_at": batch.created_at,
"nodes": list(
WorkflowNodeRun.objects.filter(
workflow_type="application_form_fill",
workflow_batch_id=batch.pk,
).order_by("id")
),
}
)
return sorted(cards, key=lambda item: item["created_at"], reverse=True)[:5]
def build_condition_confirmation(workflow_cards: list[dict[str, object]]) -> dict[str, object] | None:
for card in workflow_cards:
if (
card.get("workflow_type") == "regulatory_review"
and card.get("status") == RegulatoryReviewBatch.Status.WAITING_USER
and card.get("condition_candidates")
):
return {
"id": card["id"],
"batch_no": card["batch_no"],
"candidates": card["condition_candidates"],
}
return None
def _format_risk_label(risk_summary: dict) -> str:
parts = []
labels = [
("blocking", "阻断项"),
("high", "高风险"),
("medium", "中风险"),
("low", "低风险"),
("info", "提示"),
]
for key, label in labels:
count = int(risk_summary.get(key) or 0)
if count:
parts.append(f"{label} {count}")
return " · ".join(parts)
def _format_form_fill_label(batch: ApplicationFormFillBatch) -> str:
parts = []
if batch.selected_templates:
parts.append("模板 " + "".join(batch.selected_templates))
if batch.conflict_summary:
parts.append(f"冲突字段 {len(batch.conflict_summary)}")
if batch.risk_notes:
parts.append(f"提示 {len(batch.risk_notes)}")
return " · ".join(parts)