92444e7eae
- 新增 25+ 条风险规则（预算/报销/申请/通用类），完善风险规则模拟与反馈发布机制 - 引入费用审批动态路由、平台风险分级、预审与风险阶段管理 - 预算中心列表化改造，优化票据夹仪表盘与数字员工工作看板 - 新增 Hermes 风险线索收集器、Agent 链路追踪中心 - 扩展数字员工能力库（18 个领域 Skill）与交通费用自动预估 - 完善报销申请快速预览、权限控制与前端测试覆盖
130 lines
5.2 KiB
Python
130 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.core.agent_enums import AgentAssetDomain
|
|
from app.db.base import Base
|
|
from app.models.agent_asset import AgentAsset
|
|
from app.schemas.agent_asset import (
|
|
AgentAssetRiskRuleGenerateRequest,
|
|
AgentAssetRiskRuleSimulationRequest,
|
|
)
|
|
from app.services.agent_asset_rule_library import AgentAssetRuleLibraryManager
|
|
from app.services.agent_asset_spreadsheet import RISK_RULES_LIBRARY
|
|
from app.services.agent_assets import AgentAssetService
|
|
from app.services.risk_rule_generation import RiskRuleGenerationService
|
|
|
|
|
|
class NullRuntimeChatService:
|
|
def complete(self, *args, **kwargs) -> None:
|
|
return None
|
|
|
|
|
|
def build_session() -> Session:
|
|
engine = create_engine(
|
|
"sqlite+pysqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
Base.metadata.create_all(bind=engine)
|
|
session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
|
|
return session_factory()
|
|
|
|
|
|
def test_generated_risk_rule_contains_semantic_plan_and_flow_model(tmp_path) -> None:
|
|
with build_session() as db:
|
|
manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules")
|
|
asset_id = RiskRuleGenerationService(
|
|
db,
|
|
rule_library_manager=manager,
|
|
runtime_chat_service=NullRuntimeChatService(),
|
|
).generate_rule_asset(
|
|
AgentAssetRiskRuleGenerateRequest(
|
|
business_domain=AgentAssetDomain.EXPENSE,
|
|
expense_category="travel",
|
|
rule_title="差旅票据城市一致性校验",
|
|
natural_language=(
|
|
"差旅报销时，交通票或住宿票据中的城市必须与申报目的地一致；"
|
|
"未说明绕行、跨城或改签原因时标记高风险。"
|
|
),
|
|
requires_attachment=True,
|
|
),
|
|
actor="pytest",
|
|
)
|
|
asset = db.get(AgentAsset, asset_id)
|
|
assert asset is not None
|
|
payload = manager.read_rule_library_json(
|
|
library=RISK_RULES_LIBRARY,
|
|
file_name=asset.config_json["rule_document"]["file_name"],
|
|
)
|
|
|
|
assert payload["semantic_plan"]["required_fields"]
|
|
assert payload["semantic_plan"]["risk_action"]["risk_level"] == payload["outcomes"]["fail"]["severity"]
|
|
assert payload["flow_model"]["source"] == "json_dsl"
|
|
assert payload["flow_model"]["nodes"][0]["id"] == "start"
|
|
assert any(node["type"] == "risk" for node in payload["flow_model"]["nodes"])
|
|
assert payload["metadata"]["flow_model"]["nodes"] == payload["flow_model"]["nodes"]
|
|
assert payload["flow_diagram_svg"].startswith("<svg")
|
|
assert "风险关键词" not in payload["semantic_plan"]["judgment_steps"][0]["description"]
|
|
|
|
|
|
def test_simulation_returns_execution_trace_for_ticket_city_mismatch(tmp_path) -> None:
|
|
with build_session() as db:
|
|
manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules")
|
|
asset_id = RiskRuleGenerationService(
|
|
db,
|
|
rule_library_manager=manager,
|
|
runtime_chat_service=NullRuntimeChatService(),
|
|
).generate_rule_asset(
|
|
AgentAssetRiskRuleGenerateRequest(
|
|
business_domain=AgentAssetDomain.EXPENSE,
|
|
expense_category="travel",
|
|
rule_title="当前差旅票据城市一致性规则",
|
|
natural_language=(
|
|
"差旅报销时，交通票或住宿票据中的城市必须与申报目的地一致；"
|
|
"未说明绕行、跨城或改签原因时标记高风险。"
|
|
),
|
|
requires_attachment=True,
|
|
),
|
|
actor="pytest",
|
|
)
|
|
service = AgentAssetService(db)
|
|
service.rule_library_manager = manager
|
|
|
|
simulation = service.simulate_risk_rule_message(
|
|
asset_id,
|
|
AgentAssetRiskRuleSimulationRequest(
|
|
message="去北京出差3天",
|
|
attachments=[
|
|
{
|
|
"name": "train-ticket.pdf",
|
|
"content_type": "application/pdf",
|
|
"ocr_text": "武汉 到 上海",
|
|
"summary": "高铁票 武汉-上海",
|
|
"document_fields": [
|
|
{"key": "route", "label": "行程路线", "value": "武汉-上海"}
|
|
],
|
|
}
|
|
],
|
|
),
|
|
)
|
|
|
|
assert simulation.ready is True
|
|
assert simulation.hit is True
|
|
assert simulation.normalized_fields["claim.location"] == "北京"
|
|
assert simulation.ocr_raw_fields[0]["attachment_name"] == "train-ticket.pdf"
|
|
assert simulation.ocr_raw_fields[0]["label"] == "行程路线"
|
|
assert any(
|
|
field["key"] == "attachment.route_cities"
|
|
for field in simulation.hermes_normalized_fields
|
|
)
|
|
assert any(
|
|
field["key"] == "attachment.route_cities" and field["required"] is True
|
|
for field in simulation.executor_input_fields
|
|
)
|
|
assert simulation.trace["matched"] is True
|
|
assert "hit" in simulation.trace["path_node_ids"]
|
|
assert simulation.trace["steps"]
|