Files
DEMO-AGENT/apps/chat/views.py

498 lines
20 KiB
Python

from django.contrib import messages
from django.utils import timezone
from django.shortcuts import get_object_or_404, redirect, render
from django.urls import reverse
from django.views.decorators.http import require_POST
from agent_core.orchestrator import run_agent
from agent_core.results import AgentResult
from apps.audit.services import create_audit_log, create_notification_record
from apps.documents.models import SubmissionBatch, UploadedDocument
from apps.documents.services import append_documents_to_batch
from apps.scenarios.services import get_scenario
from .export_service import generate_registration_export, update_conversation_with_export_report
from .forms import ChatForm, ConversationUploadForm
from .models import Conversation
RISK_LEVEL_DISPLAY = {
"high": "",
"medium": "",
"low": "",
}
PASS_STATUS_DISPLAY = {
"blocked": "已阻断",
"failed": "失败",
"review_required": "待复核",
"manual_review": "待复核",
"completed": "已完成",
"passed": "已完成",
}
EXPORT_STATUS_DISPLAY = {
"completed": "已完成",
"draft_only": "待复核",
"review_required": "待复核",
"manual_review": "待复核",
"blocked": "已阻断",
"failed": "失败",
"processing": "处理中",
"pending": "处理中",
}
NOTIFY_MESSAGE_STATUS_DISPLAY = {
"sent": "已发送",
"failed": "失败",
"pending": "处理中",
}
def index(request):
conversations = Conversation.objects.all()
if conversations.exists():
return redirect("chat:detail", conversation_id=conversations.first().conversation_id)
return render(
request,
"chat/index.html",
{
"conversation": None,
"conversations": [],
"conversation_history": [],
"form": ChatForm(),
"documents": [],
"result": None,
"audit_log": None,
"node_results": [],
"active_node": None,
},
)
def detail(request, conversation_id: str):
conversation = get_object_or_404(Conversation, conversation_id=conversation_id)
batch = SubmissionBatch.objects.filter(batch_id=conversation.batch_id).first()
documents = UploadedDocument.objects.filter(batch=batch)
form = ChatForm(request.POST or None, documents=documents)
upload_form = ConversationUploadForm()
result = None
audit_log = None
active_node = None
task_modes = [
{"name": "目录汇总", "description": "汇总文件、页数、章节点和目录型文档。"},
{"name": "完整性检查", "description": "对照法规模板检查齐套性、缺失项和错放项。"},
{"name": "字段抽取", "description": "抽取产品名称、规格、适用范围、储存条件等核心字段。"},
{"name": "一致性核查", "description": "比较申请表、说明书和产品列表的字段一致性。"},
{"name": "综合风险报告", "description": "形成高优先级问题、建议动作和责任人通知。"},
]
if request.method == "POST" and form.is_valid():
scenario = get_scenario("document_review")
message = form.cleaned_data["message"]
try:
result = run_agent(
scenario,
message,
options={
"conversation_id": conversation.conversation_id,
"batch_id": conversation.batch_id,
"product_name": conversation.product_name,
"document_ids": form.cleaned_data["document_ids"],
},
)
except Exception as exc:
result = AgentResult(status="failed", error=str(exc), answer="")
audit_log = create_audit_log(
"document_review",
"注册审核智能体",
message,
result,
batch_id=conversation.batch_id,
conversation_id=conversation.conversation_id,
product_name=conversation.product_name,
)
_apply_agent_result_to_conversation(conversation, result)
_persist_notification_records(
result,
web_detail_url=reverse("audit:detail", args=[audit_log.id]),
)
active_node = "risk"
conversation.refresh_from_db()
workspace_summary = _build_workspace_summary(conversation, batch)
conversation_context = _build_conversation_context(conversation, batch, workspace_summary)
prompt_templates = _build_prompt_templates()
analysis_card = _build_analysis_card(result, conversation)
export_card = _build_export_card(result, conversation)
risk_card = _build_risk_card(result, conversation)
notify_card = _build_notify_card(result, conversation)
conversation_history = _build_conversation_history(Conversation.objects.all())
return render(
request,
"chat/index.html",
{
"conversation": conversation,
"conversations": Conversation.objects.all(),
"conversation_history": conversation_history,
"batch": batch,
"form": form,
"documents": documents,
"document_count": documents.count(),
"result": result,
"audit_log": audit_log,
"task_modes": task_modes,
"node_results": conversation.node_results,
"active_node": active_node,
"workspace_summary": workspace_summary,
"conversation_context": conversation_context,
"prompt_templates": prompt_templates,
"analysis_card": analysis_card,
"upload_form": upload_form,
"export_card": export_card,
"risk_card": risk_card,
"notify_card": notify_card,
},
)
@require_POST
def upload_documents(request, conversation_id: str):
conversation = get_object_or_404(Conversation, conversation_id=conversation_id)
batch = get_object_or_404(SubmissionBatch, batch_id=conversation.batch_id)
upload_form = ConversationUploadForm(request.POST, request.FILES)
if upload_form.is_valid():
result = append_documents_to_batch(
"document_review",
batch,
upload_form.cleaned_data["uploaded_files"],
)
warning_count = len(result["registration_overview_report"]["warnings"])
message = "资料已补充到当前资料包。"
if warning_count:
message += f" 当前有 {warning_count} 条待复核提示。"
messages.success(request, message)
else:
messages.error(
request,
"补充资料失败:" + " ".join(upload_form.non_field_errors()) if upload_form.non_field_errors() else "补充资料失败。",
)
return redirect("chat:detail", conversation_id=conversation.conversation_id)
@require_POST
def export_word(request, conversation_id: str):
conversation = get_object_or_404(Conversation, conversation_id=conversation_id)
batch = get_object_or_404(SubmissionBatch, batch_id=conversation.batch_id)
upstream_summary = (
(conversation.latest_summary or {}).get("upstream_structured_output")
or (conversation.latest_summary or {}).get("structured_output")
or {}
)
try:
export_report = generate_registration_export(
batch=batch,
conversation=conversation,
upstream_summary=upstream_summary,
)
update_conversation_with_export_report(conversation, export_report)
create_audit_log(
"document_review",
"Word 回填导出",
"生成 Word 导出文件",
AgentResult(
answer=export_report.get("summary", ""),
structured_output=export_report,
status="success",
conversation_id=conversation.conversation_id,
batch_id=conversation.batch_id,
product_name=conversation.product_name,
node_results=conversation.node_results,
),
batch_id=conversation.batch_id,
conversation_id=conversation.conversation_id,
product_name=conversation.product_name,
)
messages.success(request, "已生成新的 Word 导出文件。")
except Exception as exc:
create_audit_log(
"document_review",
"Word 回填导出",
"生成 Word 导出文件",
AgentResult(
answer="",
status="failed",
error=str(exc),
conversation_id=conversation.conversation_id,
batch_id=conversation.batch_id,
product_name=conversation.product_name,
),
batch_id=conversation.batch_id,
conversation_id=conversation.conversation_id,
product_name=conversation.product_name,
)
messages.error(request, f"Word 导出失败:{exc}")
return redirect("chat:detail", conversation_id=conversation.conversation_id)
def _persist_notification_records(result: AgentResult, *, web_detail_url: str = "") -> None:
payload = result.notification_payload or {}
owners = payload.get("owners") or []
if not owners:
return
resolved_detail_url = payload.get("web_detail_url") or web_detail_url
resolved_message_status = payload.get("message_status") or (
"sent" if result.status == "success" else "failed"
)
resolved_receipt = payload.get("receipt") or {"status": result.status}
for owner in owners:
create_notification_record(
batch_id=payload.get("batch_id", ""),
conversation_id=payload.get("conversation_id", ""),
product_name=payload.get("product_name", ""),
trigger_source="agent_execution",
notify_reason=payload.get("notify_reason", "task_completed"),
owner_role=owner.get("owner_role", ""),
feishu_user_id=owner.get("feishu_user_id", ""),
message_status=resolved_message_status,
web_detail_url=resolved_detail_url,
receipt=resolved_receipt,
)
def _build_workspace_summary(conversation: Conversation, batch: SubmissionBatch | None) -> dict:
node_status_map = {node.get("label"): node.get("status", "") for node in conversation.node_results}
risk_status = node_status_map.get("风险预警", "待处理")
notify_status = node_status_map.get("飞书通知", "待处理")
export_status = node_status_map.get("Word 回填导出", "待处理")
highest_risk_level = "" if risk_status in {"已阻断", "待复核"} else ""
latest_summary = conversation.latest_summary or {}
structured_output = latest_summary.get("structured_output") or {}
explicit_export_flag = structured_output.get("can_export_formally")
export_allowed = (
""
if explicit_export_flag is True
else ""
if explicit_export_flag is False
else ""
if risk_status in {"已阻断", "待复核"} or export_status in {"已阻断", "待复核", "失败"}
else ""
)
return {
"highest_risk_level": highest_risk_level,
"export_allowed": export_allowed,
"notify_status": notify_status,
"export_status": export_status,
"download_url": structured_output.get("download_url", ""),
"file_count": batch.file_count if batch else 0,
"page_count": batch.page_count if batch else 0,
}
def _build_conversation_context(
conversation: Conversation,
batch: SubmissionBatch | None,
workspace_summary: dict,
) -> dict:
return {
"batch_id": conversation.batch_id,
"product_name": conversation.product_name,
"workflow_type": batch.workflow_type if batch else "registration",
"task_status": conversation.get_task_status_display_text(),
"highest_risk_level": workspace_summary.get("highest_risk_level", "-"),
"export_allowed": workspace_summary.get("export_allowed", "-"),
}
def _build_prompt_templates() -> list[str]:
return [
"请汇总当前资料包的章节点、页数和目录覆盖情况",
"请检查当前资料包缺失了哪些必交项和错放项",
"请抽取当前资料包的核心字段并标记低置信度项",
"请给出当前资料包的高风险项、责任人和整改建议",
]
def _build_conversation_history(conversations) -> list[dict]:
"""
组装左栏会话历史摘要。
左栏只展示稳定摘要字段,不在模板里拼风险判断逻辑。
"""
history = []
for item in conversations:
node_status_map = {node.get("label"): node.get("status", "") for node in item.node_results}
risk_status = node_status_map.get("风险预警", "待处理")
history.append(
{
"conversation_id": item.conversation_id,
"title": item.title,
"product_name": item.product_name,
"batch_id": item.batch_id,
"risk_level": "" if risk_status in {"已阻断", "待复核"} else "",
"updated_at": item.updated_at,
"batch_binding_label": "已绑定资料包" if item.batch_id else "未绑定资料包",
}
)
return history
def _build_analysis_card(result: AgentResult | None, conversation: Conversation) -> dict:
structured_output = {}
if result and result.structured_output:
structured_output = result.structured_output
else:
structured_output = (conversation.latest_summary or {}).get("structured_output") or {}
output_type = structured_output.get("output_type")
if output_type == "registration_overview_report":
return {
"kind": "overview",
"title": "目录汇总能力卡",
"summary": structured_output.get("product_name", ""),
"stats": [
{"label": "资料文件数", "value": structured_output.get("file_count", 0)},
{"label": "总页数", "value": structured_output.get("total_page_count", 0)},
],
"items": structured_output.get("chapter_summary") or [],
"warnings": structured_output.get("warnings") or [],
}
if output_type == "registration_completeness_report":
return {
"kind": "completeness",
"title": "完整性检查能力卡",
"summary": structured_output.get("summary", ""),
"stats": [{"label": "风险等级", "value": _get_risk_level_display_text(structured_output.get("risk_level", "-"))}],
"items": structured_output.get("missing_items") or [],
"warnings": structured_output.get("misplaced_items") or [],
}
if output_type == "registration_field_extraction_report":
return {
"kind": "field_extraction",
"title": "字段抽取能力卡",
"summary": structured_output.get("summary", ""),
"stats": [{"label": "字段数", "value": len(structured_output.get("field_items") or [])}],
"items": structured_output.get("field_items") or [],
"warnings": structured_output.get("low_confidence_items") or [],
}
if output_type == "registration_consistency_report":
return {
"kind": "consistency",
"title": "一致性核查能力卡",
"summary": structured_output.get("summary", ""),
"stats": [{"label": "风险等级", "value": _get_risk_level_display_text(structured_output.get("risk_level", "-"))}],
"items": structured_output.get("conflict_items") or [],
"warnings": structured_output.get("mixed_document_risks") or [],
}
return {}
def _build_export_card(result: AgentResult | None, conversation: Conversation) -> dict:
"""
统一组装 Word 导出能力卡上下文。
优先使用本次执行结果;若本次未执行,则回退到会话最新摘要。
"""
structured_output = {}
if result and result.structured_output:
structured_output = result.structured_output
else:
structured_output = (conversation.latest_summary or {}).get("structured_output") or {}
if structured_output.get("output_type") != "registration_word_export_report":
return {}
return {
"template_name": structured_output.get("template_name", ""),
"template_version": structured_output.get("template_version", ""),
"export_status": _get_export_status_display_text(structured_output.get("export_status", "")),
"filled_fields": structured_output.get("filled_fields") or [],
"blocked_fields": structured_output.get("blocked_fields") or [],
"download_url": structured_output.get("download_url", ""),
}
def _build_risk_card(result: AgentResult | None, conversation: Conversation) -> dict:
structured_output = {}
if result and result.structured_output:
structured_output = result.structured_output
else:
structured_output = (conversation.latest_summary or {}).get("structured_output") or {}
if structured_output.get("output_type") != "registration_risk_report":
return {}
return {
"summary": structured_output.get("summary", ""),
"highest_risk_level": _get_risk_level_display_text(
structured_output.get("highest_risk_level", "")
),
"pass_status": _get_pass_status_display_text(structured_output.get("pass_status", "")),
"manual_review_items": structured_output.get("manual_review_items") or [],
"risk_items": structured_output.get("risk_items") or [],
"owner_roles": structured_output.get("owner_roles") or [],
}
def _build_notify_card(result: AgentResult | None, conversation: Conversation) -> dict:
latest_summary = conversation.latest_summary or {}
structured_output = latest_summary.get("structured_output") or {}
notification_payload = latest_summary.get("notification_payload") or {}
if result and result.structured_output:
structured_output = result.structured_output
if result and result.notification_payload:
notification_payload = result.notification_payload
notify_reason = (
structured_output.get("notify_reason")
or notification_payload.get("notify_reason")
or ""
)
mentioned_users = structured_output.get("mentioned_users") or notification_payload.get("mentioned_users") or []
message_status = structured_output.get("message_status") or notification_payload.get("message_status") or ""
web_detail_url = structured_output.get("web_detail_url") or notification_payload.get("web_detail_url") or ""
owners = structured_output.get("owner_roles") or notification_payload.get("owners") or []
if not any([notify_reason, mentioned_users, message_status, web_detail_url, owners]):
return {}
return {
"notify_reason": notify_reason,
"mentioned_users": mentioned_users,
"message_status": _get_notify_message_status_display_text(message_status),
"web_detail_url": web_detail_url,
"owners": owners,
}
def _get_risk_level_display_text(level: str) -> str:
return RISK_LEVEL_DISPLAY.get(level, level)
def _get_pass_status_display_text(status: str) -> str:
return PASS_STATUS_DISPLAY.get(status, status)
def _get_export_status_display_text(status: str) -> str:
return EXPORT_STATUS_DISPLAY.get(status, status)
def _get_notify_message_status_display_text(status: str) -> str:
return NOTIFY_MESSAGE_STATUS_DISPLAY.get(status, status)
def _apply_agent_result_to_conversation(conversation: Conversation, result: AgentResult) -> None:
conversation.task_status = result.status
if result.node_results:
conversation.node_results = result.node_results
conversation.latest_summary = {
"answer": result.answer,
"status": result.status,
"error": result.error,
"structured_output": result.structured_output,
"notification_payload": result.notification_payload,
}
conversation.last_run_at = timezone.now()
conversation.save(
update_fields=[
"task_status",
"node_results",
"latest_summary",
"last_run_at",
"updated_at",
]
)