diff --git a/agent_core/governance.py b/agent_core/governance.py new file mode 100644 index 0000000..fcdfec5 --- /dev/null +++ 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 diff --git a/agent_core/orchestrator.py b/agent_core/orchestrator.py index c37b684..5214e5d 100644 --- 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", "")), diff --git a/apps/platform_ui/services.py b/apps/platform_ui/services.py index d5daf01..48e3728 100644 --- 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"] diff --git a/tests/test_agent_core.py b/tests/test_agent_core.py index 1d54b1e..4a4f9fd 100644 --- 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"