feat: 统一治理配置与通知责任人映射

2026-06-04 01:45:27 +08:00
parent 4914ee3a75
commit 1e18fd2be9
4 changed files with 201 additions and 89 deletions
--- a/agent_core/governance.py
+++ b/agent_core/governance.py
@@ -0,0 +1,88 @@
+from pathlib import Path
+
+import yaml
+from django.conf import settings
+
+
+def governance_defaults() -> dict:
+    return {
+        "owner_mappings": [
+            {
+                "owner_role": "注册资料负责人",
+                "owner_name": "张三",
+                "department": "注册事务部",
+                "chapter_scope": "CH1",
+                "risk_scope": "字段冲突 / 缺失项",
+                "feishu_user_id": "ou_demo_1",
+                "feishu_open_id": "on_demo_1",
+                "feishu_name": "张三",
+                "notify_enabled": "是",
+            },
+            {
+                "owner_role": "注册申报负责人",
+                "owner_name": "李四",
+                "department": "临床注册组",
+                "chapter_scope": "CH2-CH6",
+                "risk_scope": "完整性风险 / 导出阻断",
+                "feishu_user_id": "ou_demo_2",
+                "feishu_open_id": "on_demo_2",
+                "feishu_name": "李四",
+                "notify_enabled": "是",
+            },
+        ],
+        "feishu_configs": [
+            {
+                "config_name": "注册审核完成通知",
+                "notify_reason": "task_completed",
+                "channel": "群机器人",
+                "message_template": "审核完成摘要 + @处理人",
+                "status": "启用",
+            },
+            {
+                "config_name": "注册审核异常通知",
+                "notify_reason": "task_failed",
+                "channel": "群机器人",
+                "message_template": "异常摘要 + @处理人",
+                "status": "启用",
+            },
+        ],
+        "template_mappings": [
+            {
+                "template_name": "注册证导出模板",
+                "output_type": "registration_word_export_report",
+                "version": "V1.0",
+                "placeholder_count": 18,
+                "status": "启用",
+                "field_mapping_summary": "产品名称 / 注册人 / 适用机型 / 储存条件",
+            },
+            {
+                "template_name": "风险摘要导出模板",
+                "output_type": "registration_word_export_report",
+                "version": "V0.9",
+                "placeholder_count": 10,
+                "status": "待校验",
+                "field_mapping_summary": "风险等级 / 批次号 / 责任人 / 证据摘要",
+            },
+        ],
+    }
+
+
+def read_governance_yaml() -> dict:
+    raw_path = getattr(settings, "GOVERNANCE_CONFIG_PATH", "")
+    if not raw_path:
+        return {}
+    config_path = Path(raw_path)
+    if not config_path.exists() or not config_path.is_file():
+        return {}
+    with config_path.open("r", encoding="utf-8") as file:
+        return yaml.safe_load(file) or {}
+
+
+def load_governance_config() -> dict:
+    defaults = governance_defaults()
+    config = read_governance_yaml()
+    for key, default_value in defaults.items():
+        configured_value = config.get(key)
+        if isinstance(default_value, list) and configured_value:
+            defaults[key] = configured_value
+    return defaults
--- a/agent_core/orchestrator.py
+++ b/agent_core/orchestrator.py
@@ -1,6 +1,7 @@
 import json
 import time

+from .governance import load_governance_config
 from .llm_provider import create_llm_provider, get_runtime_llm_config
 from .results import AgentResult
 from .structured_output import (
@@ -184,6 +185,8 @@ def _build_notification_payload(structured_output: dict, options: dict, status:
        "task_completed" if status == "success" else "task_failed"
    )
    owners = structured_output.get("owner_roles") or []
+    if not owners:
+        owners = load_governance_config()["owner_mappings"]
    return {
        "batch_id": str(options.get("batch_id", "")),
        "conversation_id": str(options.get("conversation_id", "")),
--- a/apps/platform_ui/services.py
+++ b/apps/platform_ui/services.py
@@ -1,91 +1,4 @@
-from pathlib import Path
-
-import yaml
-from django.conf import settings
-
-
-def _governance_defaults() -> dict:
-    return {
-        "owner_mappings": [
-            {
-                "owner_role": "注册资料负责人",
-                "owner_name": "张三",
-                "department": "注册事务部",
-                "chapter_scope": "CH1",
-                "risk_scope": "字段冲突 / 缺失项",
-                "feishu_user_id": "ou_demo_1",
-                "feishu_open_id": "on_demo_1",
-                "feishu_name": "张三",
-                "notify_enabled": "是",
-            },
-            {
-                "owner_role": "注册申报负责人",
-                "owner_name": "李四",
-                "department": "临床注册组",
-                "chapter_scope": "CH2-CH6",
-                "risk_scope": "完整性风险 / 导出阻断",
-                "feishu_user_id": "ou_demo_2",
-                "feishu_open_id": "on_demo_2",
-                "feishu_name": "李四",
-                "notify_enabled": "是",
-            },
-        ],
-        "feishu_configs": [
-            {
-                "config_name": "注册审核完成通知",
-                "notify_reason": "task_completed",
-                "channel": "群机器人",
-                "message_template": "审核完成摘要 + @处理人",
-                "status": "启用",
-            },
-            {
-                "config_name": "注册审核异常通知",
-                "notify_reason": "task_failed",
-                "channel": "群机器人",
-                "message_template": "异常摘要 + @处理人",
-                "status": "启用",
-            },
-        ],
-        "template_mappings": [
-            {
-                "template_name": "注册证导出模板",
-                "output_type": "registration_word_export_report",
-                "version": "V1.0",
-                "placeholder_count": 18,
-                "status": "启用",
-                "field_mapping_summary": "产品名称 / 注册人 / 适用机型 / 储存条件",
-            },
-            {
-                "template_name": "风险摘要导出模板",
-                "output_type": "registration_word_export_report",
-                "version": "V0.9",
-                "placeholder_count": 10,
-                "status": "待校验",
-                "field_mapping_summary": "风险等级 / 批次号 / 责任人 / 证据摘要",
-            },
-        ],
-    }
-
-
-def _read_governance_yaml() -> dict:
-    raw_path = getattr(settings, "GOVERNANCE_CONFIG_PATH", "")
-    if not raw_path:
-        return {}
-    config_path = Path(raw_path)
-    if not config_path.exists() or not config_path.is_file():
-        return {}
-    with config_path.open("r", encoding="utf-8") as file:
-        return yaml.safe_load(file) or {}
-
-
-def _load_governance_config() -> dict:
-    defaults = _governance_defaults()
-    config = _read_governance_yaml()
-    for key, default_value in defaults.items():
-        configured_value = config.get(key)
-        if isinstance(default_value, list) and configured_value:
-            defaults[key] = configured_value
-    return defaults
+from agent_core.governance import load_governance_config


 def get_platform_demo_context():
@@ -241,7 +154,7 @@ def get_platform_demo_context():
            "status": "待校订",
        },
    ]
-    governance_config = _load_governance_config()
+    governance_config = load_governance_config()
    owner_mappings = governance_config["owner_mappings"]
    feishu_configs = governance_config["feishu_configs"]
    template_mappings = governance_config["template_mappings"]
--- a/tests/test_agent_core.py
+++ b/tests/test_agent_core.py
@@ -2,6 +2,7 @@ from agent_core.orchestrator import build_messages, run_agent
 from agent_core.rag.ingest import _split_text, ingest_document
 from agent_core.rag.retriever import retrieve
 from agent_core.schemas.outputs import SUPPORTED_OUTPUT_TYPES
+from django.test import override_settings


 def test_run_agent_returns_structured_result_from_llm_output():
@@ -352,3 +353,110 @@ def test_registration_risk_report_builds_eight_business_nodes():
    assert result.node_results[5]["status"] == "已阻断"
    assert result.node_results[6]["status"] == "待处理"
    assert result.node_results[7]["status"] == "待处理"
+
+
+@override_settings(GOVERNANCE_CONFIG_PATH="")
+def test_registration_risk_payload_falls_back_to_governance_owner_mappings(tmp_path):
+    config_path = tmp_path / "governance.yaml"
+    config_path.write_text(
+        """
+owner_mappings:
+  - owner_role: 法规负责人
+    owner_name: 赵六
+    department: 法规事务部
+    chapter_scope: CH1
+    risk_scope: 高风险缺失
+    feishu_user_id: ou_governance_1
+    feishu_open_id: on_governance_1
+    feishu_name: 赵六
+    notify_enabled: 是
+""".strip(),
+        encoding="utf-8",
+    )
+    scenario = {
+        "id": "document_review",
+        "name": "注册审核智能体",
+        "agent": {
+            "role": "注册审核助手",
+            "goal": "输出风险结果",
+            "instructions": ["输出结构化风险结果"],
+        },
+        "rag": {"enabled": False},
+        "tools": [],
+        "output": {"type": "registration_risk_report"},
+    }
+    provider_response = """
+    {
+      "summary": "存在高风险项，需人工复核。",
+      "highest_risk_level": "high",
+      "pass_status": "blocked",
+      "owner_roles": [],
+      "notify_reason": "task_completed"
+    }
+    """
+
+    class FakeProvider:
+        def generate(self, messages, response_format=None):
+            from agent_core.llm_provider import LLMResponse
+
+            return LLMResponse(content=provider_response, model_name="demo-model", success=True)
+
+    with override_settings(GOVERNANCE_CONFIG_PATH=config_path):
+        result = run_agent(scenario, "请输出风险结果", options={"llm_provider": FakeProvider()})
+
+    owner = result.notification_payload["owners"][0]
+    assert owner["owner_role"] == "法规负责人"
+    assert owner["feishu_user_id"] == "ou_governance_1"
+    assert result.notification_payload["notify_reason"] == "task_completed"
+
+
+@override_settings(GOVERNANCE_CONFIG_PATH="")
+def test_failed_agent_result_uses_governance_owner_mappings_for_failed_notification(tmp_path):
+    config_path = tmp_path / "governance.yaml"
+    config_path.write_text(
+        """
+owner_mappings:
+  - owner_role: 注册资料负责人
+    owner_name: 孙七
+    department: 注册事务部
+    chapter_scope: CH2
+    risk_scope: 执行异常
+    feishu_user_id: ou_failed_1
+    feishu_open_id: on_failed_1
+    feishu_name: 孙七
+    notify_enabled: 是
+""".strip(),
+        encoding="utf-8",
+    )
+    scenario = {
+        "id": "document_review",
+        "name": "注册审核智能体",
+        "agent": {
+            "role": "注册审核助手",
+            "goal": "输出风险结果",
+            "instructions": ["输出结构化风险结果"],
+        },
+        "rag": {"enabled": False},
+        "tools": [],
+        "output": {"type": "registration_risk_report"},
+    }
+
+    class FailedProvider:
+        def generate(self, messages, response_format=None):
+            from agent_core.llm_provider import LLMResponse
+
+            return LLMResponse(
+                content="",
+                model_name="demo-model",
+                success=False,
+                error="provider down",
+            )
+
+    with override_settings(GOVERNANCE_CONFIG_PATH=config_path):
+        result = run_agent(scenario, "请输出风险结果", options={"llm_provider": FailedProvider()})
+
+    owner = result.notification_payload["owners"][0]
+    assert result.status == "failed"
+    assert result.notification_payload["notify_reason"] == "task_failed"
+    assert owner["owner_name"] == "孙七"
+    assert owner["feishu_open_id"] == "on_failed_1"