From 1b6c54fe789da3e5714816cba0a587a402b649e8 Mon Sep 17 00:00:00 2001 From: bruce Date: Thu, 4 Jun 2026 04:31:55 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E4=B8=8B=E6=B2=89=E4=BC=9A?= =?UTF-8?q?=E8=AF=9D=E6=89=A7=E8=A1=8C=E7=BC=96=E6=8E=92=E5=88=B0=20chat?= =?UTF-8?q?=20=E6=9C=8D=E5=8A=A1=E5=B1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/chat/services.py | 99 +++++++++++++++++++++++++++++++++++++++++++ apps/chat/views.py | 87 +++---------------------------------- tests/test_chat.py | 60 +++++++++++++++++++++++--- 3 files changed, 159 insertions(+), 87 deletions(-) diff --git a/apps/chat/services.py b/apps/chat/services.py index 2cd6e15..4ed86a7 100644 --- a/apps/chat/services.py +++ b/apps/chat/services.py @@ -1,3 +1,12 @@ +from collections.abc import Callable + +from django.utils import timezone + +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.scenarios.services import get_scenario + from .models import Conversation @@ -19,6 +28,48 @@ def create_conversation_for_batch(batch_id: str, product_name: str) -> Conversat return conversation +def execute_conversation_agent( + *, + conversation: Conversation, + message: str, + document_ids: list[int], + detail_url_builder: Callable[[int], str] | None = None, +) -> tuple[AgentResult, object]: + """ + 在服务层串起会话执行、审计留痕与通知落库。 + + View 只负责收集请求参数和渲染结果，不直接承载 Agent Core 编排。 + """ + scenario = get_scenario("document_review") + try: + result = run_agent( + scenario, + message, + options={ + "conversation_id": conversation.conversation_id, + "batch_id": conversation.batch_id, + "product_name": conversation.product_name, + "document_ids": 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) + detail_url = detail_url_builder(audit_log.id) if detail_url_builder else "" + _persist_notification_records(result, web_detail_url=detail_url) + return result, audit_log + + def _generate_conversation_id() -> str: return f"conv-{Conversation.objects.count() + 1:03d}" @@ -34,3 +85,51 @@ def _build_initial_node_results() -> list[dict]: {"code": "word_export", "label": "Word 回填导出", "status": "待处理"}, {"code": "feishu_notify", "label": "飞书通知", "status": "待处理"}, ] + + +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 _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", + ] + ) diff --git a/apps/chat/views.py b/apps/chat/views.py index f08c60f..62499db 100644 --- a/apps/chat/views.py +++ b/apps/chat/views.py @@ -1,19 +1,17 @@ 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.audit.services import create_audit_log 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 +from .services import execute_conversation_agent RISK_LEVEL_DISPLAY = { "high": "高", @@ -86,34 +84,12 @@ def detail(request, conversation_id: str): {"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]), + result, audit_log = execute_conversation_agent( + conversation=conversation, + message=message, + document_ids=form.cleaned_data["document_ids"], + detail_url_builder=lambda log_id: reverse("audit:detail", args=[log_id]), ) active_node = "risk" conversation.refresh_from_db() @@ -233,32 +209,6 @@ def export_word(request, conversation_id: str): 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, @@ -495,26 +445,3 @@ def _get_export_status_display_text(status: str) -> str: 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", - ] - ) diff --git a/tests/test_chat.py b/tests/test_chat.py index 50c5cca..f1bd023 100644 --- a/tests/test_chat.py +++ b/tests/test_chat.py @@ -8,7 +8,7 @@ from agent_core.results import AgentResult from apps.audit.models import AgentAuditLog from apps.audit.models import NotificationRecord from apps.chat.models import Conversation -from apps.chat.services import create_conversation_for_batch +from apps.chat.services import create_conversation_for_batch, execute_conversation_agent from apps.documents.models import ExportedDocument, SubmissionBatch, UploadedDocument @@ -48,7 +48,7 @@ def test_chat_post_returns_agent_result_and_audit_log(client, db, monkeypatch): ) monkeypatch.setattr( - "apps.chat.views.run_agent", + "apps.chat.services.run_agent", lambda *args, **kwargs: AgentResult(answer="模拟回答", status="success"), ) @@ -99,7 +99,7 @@ def test_chat_passes_selected_document_ids_to_agent_core(client, db, monkeypatch captured["options"] = options or {} return AgentResult(answer="ok", status="success") - monkeypatch.setattr("apps.chat.views.run_agent", fake_run_agent) + monkeypatch.setattr("apps.chat.services.run_agent", fake_run_agent) response = client.post( reverse("chat:detail", args=[conversation.conversation_id]), @@ -169,7 +169,7 @@ def test_chat_execution_creates_notification_record_from_agent_result(client, db ) monkeypatch.setattr( - "apps.chat.views.run_agent", + "apps.chat.services.run_agent", lambda *args, **kwargs: AgentResult( answer="执行完成", status="success", @@ -212,7 +212,7 @@ def test_chat_execution_uses_notification_payload_message_status_and_receipt(cli ) monkeypatch.setattr( - "apps.chat.views.run_agent", + "apps.chat.services.run_agent", lambda *args, **kwargs: AgentResult( answer="通知已发送", status="success", @@ -257,7 +257,7 @@ def test_chat_execution_creates_failed_notification_record_and_updates_conversat ) monkeypatch.setattr( - "apps.chat.views.run_agent", + "apps.chat.services.run_agent", lambda *args, **kwargs: AgentResult( answer="执行失败", status="failed", @@ -310,7 +310,7 @@ def test_chat_execution_persists_agent_node_results_to_conversation(client, db, ) monkeypatch.setattr( - "apps.chat.views.run_agent", + "apps.chat.services.run_agent", lambda *args, **kwargs: AgentResult( answer="已生成风险结论", status="success", @@ -366,6 +366,52 @@ def test_create_conversation_for_batch_initializes_eight_workflow_nodes(db): ] +def test_execute_conversation_agent_runs_in_service_layer_and_persists_audit_and_notifications(db, monkeypatch): + batch, conversation = _create_conversation_with_batch() + captured = {} + + def fake_run_agent(scenario_config, user_input, options=None): + captured["scenario_id"] = scenario_config["id"] + captured["user_input"] = user_input + captured["options"] = options or {} + return AgentResult( + answer="服务层执行完成", + status="success", + notification_payload={ + "batch_id": batch.batch_id, + "conversation_id": conversation.conversation_id, + "product_name": batch.product_name, + "notify_reason": "task_completed", + "owners": [ + { + "owner_role": "注册资料负责人", + "feishu_user_id": "ou_service_1", + } + ], + }, + ) + + monkeypatch.setattr("apps.chat.services.run_agent", fake_run_agent) + + result, audit_log = execute_conversation_agent( + conversation=conversation, + message="服务层执行审核", + document_ids=[101, 102], + detail_url_builder=lambda log_id: f"/audit/{log_id}/", + ) + + conversation.refresh_from_db() + assert result.answer == "服务层执行完成" + assert audit_log.batch_id == batch.batch_id + assert captured["scenario_id"] == "document_review" + assert captured["user_input"] == "服务层执行审核" + assert captured["options"]["document_ids"] == [101, 102] + assert conversation.latest_summary["answer"] == "服务层执行完成" + record = NotificationRecord.objects.get() + assert record.notify_reason == "task_completed" + assert record.web_detail_url == f"/audit/{audit_log.id}/" + + def test_chat_page_shows_upload_entry_and_dynamic_context_cards(client, db): batch, conversation = _create_conversation_with_batch() conversation.node_results = [