feat: close phase15 governance loops

This commit is contained in:
2026-05-21 20:17:38 +08:00
parent 124e45df8c
commit 464579e48f
28 changed files with 1926 additions and 115 deletions
+3
View File
@@ -77,7 +77,10 @@ def recommend_missing_clauses(
"document_type_code": result.get("document_type_code"),
"contract_type": payload.contract_type,
"recommended_count": result.get("total"),
"recommendation_keys": (result.get("summary") or {}).get("recommendation_keys") or [],
"missing_clause_keys": (result.get("summary") or {}).get("missing_clause_keys") or [],
"industry_practice_keys": (result.get("summary") or {}).get("industry_practice_keys") or [],
"industry_practice_count": (result.get("summary") or {}).get("industry_practice_count") or 0,
"source_count": (result.get("summary") or {}).get("source_count") or 0,
"text_length": len(payload.contract_text or ""),
},
+8 -1
View File
@@ -105,6 +105,8 @@ def scan_legal_change_impact(
"article_id": payload.article_id,
"change_type": result["event"]["change_type"],
"impact_summary": result["event"]["impact_summary"],
"semantic_match_count": result["event"]["impact_summary"].get("semantic_match_count", 0),
"scan_strategy": result["event"]["impact_summary"].get("scan_strategy", []),
},
ip_address=request.client.host if request.client else None,
user_agent=request.headers.get("user-agent"),
@@ -127,7 +129,11 @@ def run_legal_change_event_scan(
action="run_legal_change_event_scan",
resource_type="legal_change_event",
resource_id=event_id,
detail_json={"impact_summary": result["event"]["impact_summary"]},
detail_json={
"impact_summary": result["event"]["impact_summary"],
"semantic_match_count": result["event"]["impact_summary"].get("semantic_match_count", 0),
"scan_strategy": result["event"]["impact_summary"].get("scan_strategy", []),
},
ip_address=request.client.host if request.client else None,
user_agent=request.headers.get("user-agent"),
)
@@ -153,6 +159,7 @@ def run_pending_legal_change_scans(
"requested_limit": result["requested_limit"],
"completed_count": result["completed_count"],
"failed_count": result["failed_count"],
"worker_compatible": True,
},
ip_address=request.client.host if request.client else None,
user_agent=request.headers.get("user-agent"),
+4
View File
@@ -178,6 +178,10 @@ def run_prompt_asset_evaluation(
"evaluation_status": result["version"]["evaluation_status"],
"evaluation_score": result["version"]["evaluation_score"],
"suite_key": result["evaluation_report"].get("suite_key"),
"evaluation_mode": result["evaluation_report"].get("evaluation_mode"),
"sample_count": result["evaluation_report"].get("sample_count"),
"passed_samples": result["evaluation_report"].get("passed_samples"),
"failed_samples": result["evaluation_report"].get("failed_samples"),
},
)
return success_response(result)
+8
View File
@@ -24,6 +24,14 @@ class Settings(BaseSettings):
shared_ai_platform_timeout_seconds: float = 30.0
shared_ai_platform_legal_review_enabled: bool = True
shared_ai_platform_legal_review_fallback_to_local: bool = False
prompt_golden_eval_mode: str = "auto"
prompt_golden_eval_include_raw_model_response: bool = False
legal_change_impact_semantic_enabled: bool = True
legal_change_impact_semantic_threshold: float = 0.10
legal_change_impact_worker_enabled: bool = True
legal_change_impact_worker_batch_size: int = 10
legal_change_impact_worker_interval_seconds: int = 300
legal_change_impact_worker_user_id: str = "system-legal-change-worker"
secret_key: str = "change-me"
access_token_expire_minutes: int = 120
database_url: str = "postgresql+psycopg://postgres:postgres@localhost:5432/ai_contract_platform"
@@ -81,6 +81,15 @@ class LegalChangeImpactRepository:
.all()
)
def list_pending_events_for_worker(self, limit: int = 10) -> list[LegalChangeEvent]:
return (
self.db.query(LegalChangeEvent)
.filter(LegalChangeEvent.scan_status == "pending")
.order_by(LegalChangeEvent.created_at.asc())
.limit(max(1, min(limit, 100)))
.all()
)
def delete_impacts_by_event(self, tenant_id: str, event_id: str) -> int:
deleted = (
self.db.query(LegalChangeImpact)
@@ -498,6 +498,199 @@ DOCUMENT_TYPE_RULES["software_development"] = [
},
]
INDUSTRY_PRACTICE_RULES: dict[str, list[dict[str, Any]]] = {
"service_general": [
{
"clause_key": "change_order_confirmation",
"clause_title": "服务范围变更确认",
"clause_type": "industry_practice",
"priority": "medium",
"keywords": ["变更单", "服务范围变更", "change order", "需求变更"],
"anchor_keywords": ["服务范围", "交付物", "验收"],
"suggested_text": (
"服务范围变更确认:服务内容、交付物、实施计划或费用口径发生变更的,双方应通过"
"书面补充协议、变更单或双方确认的电子流程确认后执行;未经确认的变更不当然构成"
"乙方新增义务或甲方付款义务。"
),
"reason": "服务项目中范围变更高频发生,提前约定变更确认方式能减少口头需求和费用争议。",
"legal_basis": "无直接法条,仅为行业惯例;建议结合具体服务范围、审批流程和费用调整机制确认。",
"practice_basis": "服务类项目通常通过变更单、补充协议或可追溯电子确认固化新增需求、延期和费用调整。",
"source_rule": "LAW-CLAUSE-PRACTICE-SERVICE-CHANGE-ORDER",
"default_position": "建议插入在服务范围或交付物条款之后。",
"recommendation_kind": "industry_practice",
"industry_context": "通用服务",
"basis_type": "industry_practice_only",
},
{
"clause_key": "service_progress_record",
"clause_title": "服务过程记录与阶段确认",
"clause_type": "industry_practice",
"priority": "low",
"keywords": ["服务记录", "周报", "月报", "阶段确认", "meeting minutes"],
"anchor_keywords": ["服务范围", "验收", "付款"],
"suggested_text": (
"服务过程记录与阶段确认:乙方应按约定周期提交服务记录、会议纪要、阶段成果或"
"工作报告;甲方应在收到后合理期限内确认或提出书面异议,逾期未反馈的,双方可"
"按约定视为阶段事项已完成记录。"
),
"reason": "持续服务合同往往缺少可验证过程证据,阶段记录可以支撑验收、付款和争议处理。",
"legal_basis": "无直接法条,仅为行业惯例;建议结合双方沟通机制和验收触发条件设置。",
"practice_basis": "咨询、运维、外包服务通常用周报、月报、会议纪要或阶段确认单沉淀履约证据。",
"source_rule": "LAW-CLAUSE-PRACTICE-SERVICE-PROGRESS-RECORD",
"default_position": "建议插入在服务范围或验收条款之后。",
"recommendation_kind": "industry_practice",
"industry_context": "持续服务",
"basis_type": "industry_practice_only",
},
],
"technical_service": [
{
"clause_key": "technical_assumptions_boundary",
"clause_title": "技术前提与接口边界",
"clause_type": "industry_practice",
"priority": "medium",
"keywords": ["技术前提", "接口边界", "系统环境", "第三方接口", "technical assumptions"],
"anchor_keywords": ["服务范围", "交付物", "验收"],
"suggested_text": (
"技术前提与接口边界:双方应在附件中列明系统环境、接口权限、第三方平台、数据格式"
"和测试条件等技术前提;因前提条件变化、第三方接口调整或甲方环境不满足要求导致"
"延期或返工的,双方应按变更流程处理。"
),
"reason": "技术服务常受系统环境、接口和第三方平台影响,提前固化边界能降低履约争议。",
"legal_basis": "无直接法条,仅为行业惯例;建议结合项目技术附件和变更流程确认。",
"practice_basis": "技术服务项目通常用技术附件、接口清单、环境确认单和测试记录管理履约前提。",
"source_rule": "LAW-CLAUSE-PRACTICE-TECH-ASSUMPTIONS",
"default_position": "建议插入在服务范围或交付物条款之后。",
"recommendation_kind": "industry_practice",
"industry_context": "技术服务",
"basis_type": "industry_practice_only",
},
],
"software_development": [
{
"clause_key": "scope_change_control",
"clause_title": "需求变更与版本范围控制",
"clause_type": "industry_practice",
"priority": "high",
"keywords": ["需求变更", "版本范围", "迭代范围", "需求确认", "scope change"],
"anchor_keywords": ["服务范围", "交付物", "验收", "维护支持"],
"suggested_text": (
"需求变更与版本范围控制:软件功能、页面、接口、数据结构、第三方集成或验收标准"
"发生新增、删减或调整的,双方应通过需求变更单、迭代确认单或可追溯电子流程确认"
"范围、工期、费用和验收影响后实施。"
),
"reason": "软件开发项目中需求变更最容易导致延期和费用争议,建议把变更流程写入合同正文或附件。",
"legal_basis": "无直接法条,仅为行业惯例;建议结合项目管理流程、需求清单和验收标准确认。",
"practice_basis": "软件开发行业通常用需求基线、迭代计划、变更单和版本验收清单控制范围蔓延。",
"source_rule": "LAW-CLAUSE-PRACTICE-SOFTWARE-SCOPE-CHANGE",
"default_position": "建议插入在服务范围、交付物或验收条款之后。",
"recommendation_kind": "industry_practice",
"industry_context": "软件开发",
"basis_type": "industry_practice_only",
},
{
"clause_key": "open_source_third_party_components",
"clause_title": "开源及第三方组件清单",
"clause_type": "industry_practice",
"priority": "medium",
"keywords": ["开源组件", "第三方组件", "开源软件", "组件清单", "open source"],
"anchor_keywords": ["知识产权", "交付物", "维护支持"],
"suggested_text": (
"开源及第三方组件清单:乙方交付的软件如使用开源软件、第三方 SDK、字体、模型、"
"插件或云服务,应向甲方提供组件清单、许可类型、使用范围及已知限制;涉及额外授权、"
"费用或合规义务的,应由双方另行确认承担方式。"
),
"reason": "软件交付中第三方组件会影响授权、后续商用和安全维护,建议在合同中要求清单化披露。",
"legal_basis": "无直接法条,仅为行业惯例;如涉及著作权、商业许可或数据处理,应另行结合法律依据审查。",
"practice_basis": "软件交付通常要求 SBOM、第三方组件清单或开源许可披露,以便后续运维和合规审计。",
"source_rule": "LAW-CLAUSE-PRACTICE-SOFTWARE-THIRD-PARTY",
"default_position": "建议插入在知识产权或交付物条款之后。",
"recommendation_kind": "industry_practice",
"industry_context": "软件开发",
"basis_type": "industry_practice_only",
},
],
"labor_contract": [
{
"clause_key": "employee_handbook_acknowledgement",
"clause_title": "规章制度告知与签收",
"clause_type": "industry_practice",
"priority": "medium",
"keywords": ["规章制度", "员工手册", "制度签收", "培训确认"],
"anchor_keywords": ["岗位", "劳动纪律", "解除", "试用期"],
"suggested_text": (
"规章制度告知与签收:甲方依法制定并公示的员工手册、考勤、奖惩、信息安全、保密"
"和岗位管理制度作为劳动管理依据;乙方确认已通过书面、电子签收或培训记录知悉相关制度。"
),
"reason": "劳动合同履行中,制度告知和签收记录是后续考核、奖惩和争议处理的重要证据。",
"legal_basis": "无直接法条,仅为行业惯例;涉及规章制度适用时仍需满足民主程序、公示告知和合法合理要求。",
"practice_basis": "企业人力资源管理通常通过员工手册签收、入职培训记录和制度更新通知沉淀管理证据。",
"source_rule": "LAW-CLAUSE-PRACTICE-LABOR-HANDBOOK",
"default_position": "建议插入在岗位职责、劳动纪律或解除终止条款附近。",
"recommendation_kind": "industry_practice",
"industry_context": "劳动人事",
"basis_type": "industry_practice_only",
},
{
"clause_key": "handover_materials_record",
"clause_title": "离职交接清单",
"clause_type": "industry_practice",
"priority": "low",
"keywords": ["离职交接", "交接清单", "资料返还", "账号移交"],
"anchor_keywords": ["解除", "终止", "保密", "竞业限制"],
"suggested_text": (
"离职交接清单:劳动合同解除或终止时,乙方应按甲方合理要求办理工作资料、设备、"
"账号权限、客户信息和未结事项交接;双方可通过交接清单、系统记录或书面确认留存证据。"
),
"reason": "离职交接是劳动争议和商业资料保护的高频风险点,清单化处理更便于执行。",
"legal_basis": "无直接法条,仅为行业惯例;涉及工资结算、经济补偿或保密义务时应另行依法处理。",
"practice_basis": "企业通常用离职交接单、资产归还记录和账号权限回收记录完成离职闭环。",
"source_rule": "LAW-CLAUSE-PRACTICE-LABOR-HANDOVER",
"default_position": "建议插入在解除终止或保密条款之后。",
"recommendation_kind": "industry_practice",
"industry_context": "劳动人事",
"basis_type": "industry_practice_only",
},
],
"nda_commercial": [
{
"clause_key": "need_to_know_access",
"clause_title": "最小知悉范围",
"clause_type": "industry_practice",
"priority": "medium",
"keywords": ["最小知悉", "need to know", "访问权限", "授权人员"],
"anchor_keywords": ["保密义务", "保密信息", "披露"],
"suggested_text": (
"最小知悉范围:接收方仅可在实现合作目的所必需的范围内向其员工、顾问或关联人员"
"披露保密信息,并应确保接触人员承担不低于本协议约定的保密义务。"
),
"reason": "保密协议通常需要控制内部流转范围,避免保密信息因过度共享而失控。",
"legal_basis": "无直接法条,仅为行业惯例;涉及商业秘密保护时应结合保密措施和访问控制证据确认。",
"practice_basis": "商业合作中常按 need-to-know 原则控制保密信息访问权限并留存授权记录。",
"source_rule": "LAW-CLAUSE-PRACTICE-NDA-NEED-TO-KNOW",
"default_position": "建议插入在保密义务或保密信息定义之后。",
"recommendation_kind": "industry_practice",
"industry_context": "保密合作",
"basis_type": "industry_practice_only",
},
],
}
SERVICE_GENERAL_PRACTICE_RULES = INDUSTRY_PRACTICE_RULES["service_general"]
TECHNICAL_SERVICE_BASE_PRACTICE_RULES = INDUSTRY_PRACTICE_RULES["technical_service"]
SOFTWARE_DEVELOPMENT_BASE_PRACTICE_RULES = INDUSTRY_PRACTICE_RULES["software_development"]
TECHNICAL_SERVICE_PRACTICE_RULES = [
*TECHNICAL_SERVICE_BASE_PRACTICE_RULES,
*SERVICE_GENERAL_PRACTICE_RULES,
]
SOFTWARE_DEVELOPMENT_PRACTICE_RULES = [
*SOFTWARE_DEVELOPMENT_BASE_PRACTICE_RULES,
*TECHNICAL_SERVICE_BASE_PRACTICE_RULES,
*SERVICE_GENERAL_PRACTICE_RULES,
]
INDUSTRY_PRACTICE_RULES["technical_service"] = TECHNICAL_SERVICE_PRACTICE_RULES
INDUSTRY_PRACTICE_RULES["software_development"] = SOFTWARE_DEVELOPMENT_PRACTICE_RULES
class ClauseRecommendationService:
def __init__(self, db: Optional[Session], legal_article_service: Optional[LegalArticleService] = None):
@@ -531,6 +724,7 @@ class ClauseRecommendationService:
max_count = self._normalize_max_items(max_items)
detected_keys: list[str] = []
detected_industry_practice_keys: list[str] = []
recommendations: list[dict[str, Any]] = []
for rule in rules:
if self._rule_is_present(normalized_text, rule):
@@ -548,7 +742,35 @@ class ClauseRecommendationService:
if len(recommendations) >= max_count:
break
missing_keys = [item["clause_key"] for item in recommendations]
if len(recommendations) < max_count:
for rule in self._industry_practice_rules_for_document_type(normalized_document_type):
if self._rule_is_present(normalized_text, rule):
detected_industry_practice_keys.append(rule["clause_key"])
continue
legal_article_refs = self._search_legal_articles(current_user, rule)
recommendations.append(
self._build_recommendation_item(
contract_text=normalized_text,
document_type_code=normalized_document_type,
rule=rule,
legal_article_refs=legal_article_refs,
)
)
if len(recommendations) >= max_count:
break
recommendation_keys = [item["clause_key"] for item in recommendations]
required_missing_keys = [
item["clause_key"]
for item in recommendations
if item.get("recommendation_kind") != "industry_practice"
]
missing_keys = required_missing_keys
industry_practice_keys = [
item["clause_key"]
for item in recommendations
if item.get("recommendation_kind") == "industry_practice"
]
result = {
"contract_id": contract_id,
"document_type_code": normalized_document_type,
@@ -558,16 +780,22 @@ class ClauseRecommendationService:
"total": len(recommendations),
"summary": {
"recommended_count": len(recommendations),
"recommendation_keys": recommendation_keys,
"detected_clause_keys": detected_keys,
"missing_clause_keys": missing_keys,
"required_missing_clause_keys": required_missing_keys,
"industry_practice_keys": industry_practice_keys,
"detected_industry_practice_keys": detected_industry_practice_keys,
"industry_practice_count": len(industry_practice_keys),
"source_count": sum(len(item.get("legal_article_refs") or []) for item in recommendations),
"strategy": "rule_based_required_clause_gap",
"strategy": "rule_based_required_clause_gap_and_industry_practice",
},
"sources": [
"document_type_required_clause_list",
"standard_clause_library",
"legal_articles",
"review_rules",
"industry_practice_rules",
],
}
self.prompt_runtime_service.record_run_link(
@@ -581,7 +809,10 @@ class ClauseRecommendationService:
"document_type_code": normalized_document_type,
"contract_type": contract_type,
"recommended_count": len(recommendations),
"recommendation_keys": recommendation_keys,
"missing_clause_keys": missing_keys,
"required_missing_clause_keys": required_missing_keys,
"industry_practice_keys": industry_practice_keys,
"text_length": len(normalized_text),
},
)
@@ -705,20 +936,19 @@ class ClauseRecommendationService:
"clause_key": rule["clause_key"],
"clause_title": rule["clause_title"],
"clause_type": rule["clause_type"],
"recommendation_kind": rule.get("recommendation_kind", "missing_clause"),
"basis_type": rule.get("basis_type") or ("legal_basis" if legal_article_refs else "rule_fallback"),
"industry_context": rule.get("industry_context"),
"priority": rule.get("priority", "medium"),
"confidence": "high" if legal_article_refs else "medium",
"insert_position": insertion_position,
"suggested_text": rule["suggested_text"],
"reason": rule["reason"],
"legal_basis": cls._build_legal_basis(rule, legal_article_refs),
"practice_basis": rule.get("practice_basis"),
"legal_article_refs": legal_article_refs,
"source_rule": rule["source_rule"],
"source_types": [
"document_type_required_clause_list",
"standard_clause_library",
"legal_articles" if legal_article_refs else "legal_basis_fallback",
"review_rules",
],
"source_types": cls._build_recommendation_source_types(rule, legal_article_refs),
"detected_keywords": [],
}
@@ -742,6 +972,24 @@ class ClauseRecommendationService:
return str(rule.get("legal_basis") or "")
return f"{rule.get('legal_basis') or '建议结合法条来源核对。'} 关联依据:{';'.join(labels)}。"
@staticmethod
def _build_recommendation_source_types(rule: dict[str, Any], legal_article_refs: list[dict[str, Any]]) -> list[str]:
if rule.get("recommendation_kind") == "industry_practice":
source_types = [
"industry_practice_rules",
"standard_clause_library",
"legal_articles" if legal_article_refs else "industry_practice_fallback",
]
if rule.get("basis_type") == "industry_practice_only":
source_types.append("no_direct_legal_article")
return source_types
return [
"document_type_required_clause_list",
"standard_clause_library",
"legal_articles" if legal_article_refs else "legal_basis_fallback",
"review_rules",
]
@classmethod
def _resolve_insertion_position(cls, contract_text: str, rule: dict[str, Any]) -> dict[str, Any]:
anchor = cls._find_anchor_line(contract_text, rule.get("anchor_keywords") or [])
@@ -989,6 +1237,8 @@ class ClauseRecommendationService:
def _search_legal_articles(self, current_user: dict[str, Any], rule: dict[str, Any]) -> list[dict[str, Any]]:
if self.legal_article_service is None:
return []
if rule.get("basis_type") == "industry_practice_only" and not str(rule.get("legal_basis_query") or "").strip():
return []
query = str(rule.get("legal_basis_query") or rule.get("clause_title") or "").strip()
if not query:
return []
@@ -1024,6 +1274,10 @@ class ClauseRecommendationService:
return rules
return GENERAL_RULES
@staticmethod
def _industry_practice_rules_for_document_type(document_type_code: str) -> list[dict[str, Any]]:
return INDUSTRY_PRACTICE_RULES.get(document_type_code, [])
@staticmethod
def _normalize_document_type(document_type_code: Optional[str], contract_type: Optional[str]) -> str:
raw = (document_type_code or contract_type or "general").strip().lower().replace("-", "_")
@@ -8,6 +8,7 @@ from typing import Any, Iterable, Optional
from fastapi import HTTPException, status
from sqlalchemy.orm import Session
from app.core.config import settings
from app.models.contract_template import ContractTemplate
from app.models.generated_contract import GeneratedContract
from app.models.legal_article import LegalArticle
@@ -16,6 +17,7 @@ from app.models.legal_change_impact import LegalChangeImpact
from app.models.review_rule import ReviewRule
from app.repositories.legal_article_repository import LegalArticleRepository
from app.repositories.legal_change_impact_repository import LegalChangeImpactRepository
from app.services.legal_semantic_matcher import LegalSemanticMatch, LegalSemanticMatcher
VALID_CHANGE_TYPES = {"created", "updated", "repealed", "effective_changed"}
@@ -27,6 +29,7 @@ class LegalChangeImpactService:
self.db = db
self.repository = LegalChangeImpactRepository(db)
self.legal_article_repository = LegalArticleRepository(db)
self.semantic_matcher = LegalSemanticMatcher(threshold=settings.legal_change_impact_semantic_threshold)
def list_events(
self,
@@ -219,6 +222,15 @@ class LegalChangeImpactService:
direct_terms = self._unique_terms([article.source_id, article.source_title, article.article_no])
tag_terms = self._unique_terms(article.tags_json or [])
article_keywords = self._extract_article_keywords(article.article_text)
article_semantic_text = self._join_text(
event.change_summary,
article.source_title,
article.article_no,
article.chapter_title,
article.section_title,
article.article_text,
article.tags_json,
)
impacts: list[LegalChangeImpact] = []
templates = self.db.query(ContractTemplate).filter(ContractTemplate.tenant_id == tenant_id).all()
@@ -237,7 +249,8 @@ class LegalChangeImpactService:
matched_direct = self._matched_terms(text, direct_terms)
matched_tags = self._matched_terms(text, tag_terms)
matched_keywords = self._matched_terms(text, article_keywords[:6])
if matched_direct or matched_tags or matched_keywords:
semantic_match = self._semantic_match(article_semantic_text, text)
if matched_direct or matched_tags or matched_keywords or semantic_match.matched:
impact_level = "immediate_update" if matched_direct else "reference_only"
impacts.append(
self._new_impact(
@@ -247,11 +260,14 @@ class LegalChangeImpactService:
target_id=template.id,
target_name=template.name,
target_version_label=template.current_version_label,
match_terms=self._unique_terms([*matched_direct, *matched_tags, *matched_keywords]),
match_reason=self._build_match_reason(matched_direct, matched_tags, matched_keywords),
match_terms=self._unique_terms([*matched_direct, *matched_tags, *matched_keywords, *semantic_match.shared_terms]),
match_reason=self._build_match_reason(matched_direct, matched_tags, matched_keywords, semantic_match),
impact_level=impact_level,
suggestion=self._build_suggestion("contract_template", impact_level, article),
evidence={"template_code": template.template_code, "document_type_code": template.document_type_code},
evidence=self._build_evidence(
{"template_code": template.template_code, "document_type_code": template.document_type_code},
semantic_match,
),
)
)
@@ -261,7 +277,8 @@ class LegalChangeImpactService:
matched_direct = self._matched_terms(text, direct_terms)
matched_tags = self._matched_terms(text, tag_terms)
matched_keywords = self._matched_terms(text, article_keywords[:8])
if matched_direct or matched_tags or matched_keywords:
semantic_match = self._semantic_match(article_semantic_text, text)
if matched_direct or matched_tags or matched_keywords or semantic_match.matched:
impacts.append(
self._new_impact(
tenant_id,
@@ -270,11 +287,11 @@ class LegalChangeImpactService:
target_id=rule.id,
target_name=rule.name,
target_version_label=rule.current_version_label,
match_terms=self._unique_terms([*matched_direct, *matched_tags, *matched_keywords]),
match_reason=self._build_match_reason(matched_direct, matched_tags, matched_keywords),
match_terms=self._unique_terms([*matched_direct, *matched_tags, *matched_keywords, *semantic_match.shared_terms]),
match_reason=self._build_match_reason(matched_direct, matched_tags, matched_keywords, semantic_match),
impact_level="immediate_update",
suggestion=self._build_suggestion("review_rule", "immediate_update", article),
evidence={"rule_code": rule.rule_code, "risk_level": rule.risk_level},
evidence=self._build_evidence({"rule_code": rule.rule_code, "risk_level": rule.risk_level}, semantic_match),
)
)
@@ -293,7 +310,8 @@ class LegalChangeImpactService:
matched_direct = self._matched_terms(text, direct_terms)
matched_tags = self._matched_terms(text, tag_terms)
matched_keywords = self._matched_terms(text, article_keywords[:6])
if matched_direct or matched_tags or matched_keywords:
semantic_match = self._semantic_match(article_semantic_text, text)
if matched_direct or matched_tags or matched_keywords or semantic_match.matched:
impact_level = "next_renewal_update" if matched_direct else "reference_only"
impacts.append(
self._new_impact(
@@ -303,11 +321,14 @@ class LegalChangeImpactService:
target_id=contract.id,
target_name=contract.file_name,
target_version_label=None,
match_terms=self._unique_terms([*matched_direct, *matched_tags, *matched_keywords]),
match_reason=self._build_match_reason(matched_direct, matched_tags, matched_keywords),
match_terms=self._unique_terms([*matched_direct, *matched_tags, *matched_keywords, *semantic_match.shared_terms]),
match_reason=self._build_match_reason(matched_direct, matched_tags, matched_keywords, semantic_match),
impact_level=impact_level,
suggestion=self._build_suggestion("generated_contract", impact_level, article),
evidence={"template_id": contract.template_id, "document_type_code": contract.document_type_code},
evidence=self._build_evidence(
{"template_id": contract.template_id, "document_type_code": contract.document_type_code},
semantic_match,
),
)
)
@@ -351,18 +372,34 @@ class LegalChangeImpactService:
def _build_impact_summary(impacts: list[LegalChangeImpact]) -> dict[str, Any]:
by_level: dict[str, int] = {}
by_target_type: dict[str, int] = {}
by_match_reason: dict[str, int] = {}
semantic_match_count = 0
for impact in impacts:
by_level[impact.impact_level] = by_level.get(impact.impact_level, 0) + 1
by_target_type[impact.target_type] = by_target_type.get(impact.target_type, 0) + 1
for reason in str(impact.match_reason or "").split(","):
if not reason:
continue
by_match_reason[reason] = by_match_reason.get(reason, 0) + 1
if reason == "semantic_vector_match":
semantic_match_count += 1
return {
"total": len(impacts),
"by_level": by_level,
"by_target_type": by_target_type,
"by_match_reason": by_match_reason,
"blocking_count": by_level.get("immediate_update", 0),
"semantic_match_count": semantic_match_count,
"scan_strategy": ["direct_reference", "tag_match", "keyword_match", "semantic_vector_match"],
}
@staticmethod
def _build_match_reason(direct_terms: list[str], tag_terms: list[str], keyword_terms: list[str]) -> str:
def _build_match_reason(
direct_terms: list[str],
tag_terms: list[str],
keyword_terms: list[str],
semantic_match: Optional[LegalSemanticMatch] = None,
) -> str:
reasons: list[str] = []
if direct_terms:
reasons.append("direct_legal_reference")
@@ -370,8 +407,26 @@ class LegalChangeImpactService:
reasons.append("tag_match")
if keyword_terms:
reasons.append("keyword_match")
if semantic_match and semantic_match.matched:
reasons.append("semantic_vector_match")
return ",".join(reasons) or "unknown"
def _semantic_match(self, source_text: str, target_text: str) -> LegalSemanticMatch:
if not settings.legal_change_impact_semantic_enabled:
return LegalSemanticMatch(False, 0.0, [], [])
return self.semantic_matcher.match(source_text, target_text)
@staticmethod
def _build_evidence(base: dict[str, Any], semantic_match: LegalSemanticMatch) -> dict[str, Any]:
evidence = dict(base)
if semantic_match.matched:
evidence["semantic_vector"] = {
"score": semantic_match.score,
"shared_terms": semantic_match.shared_terms,
"shared_concepts": semantic_match.shared_concepts,
}
return evidence
@staticmethod
def _build_suggestion(target_type: str, impact_level: str, article: LegalArticle) -> str:
article_label = f"{article.source_title}{article.article_no}"
@@ -0,0 +1,96 @@
from __future__ import annotations
import math
import re
from collections import Counter
from dataclasses import dataclass
from typing import Iterable
@dataclass(frozen=True)
class LegalSemanticMatch:
matched: bool
score: float
shared_terms: list[str]
shared_concepts: list[str]
class LegalSemanticMatcher:
"""Small deterministic vector matcher for legal-impact scanning.
This is intentionally local and dependency-free. It gives Phase15-G a real
vector-style semantic scan path without requiring a remote embedding service.
"""
LEGAL_CONCEPT_ALIASES: dict[str, tuple[str, ...]] = {
"breach_liability": ("违约", "违约金", "违约责任", "未按约", "未履行", "迟延履行", "逾期交付", "逾期付款", "赔偿", "损失赔偿"),
"payment": ("价款", "付款", "支付", "结算", "费用", "款项", "收款", "发票"),
"termination": ("解除", "终止", "撤销", "合同解除", "提前终止", "退还"),
"confidentiality": ("保密", "商业秘密", "信息披露", "泄露", "保密义务"),
"intellectual_property": ("知识产权", "著作权", "专利", "商标", "许可", "侵权"),
"labor": ("劳动", "用人单位", "劳动者", "工资", "加班", "社保", "劳动合同"),
"data_privacy": ("个人信息", "数据", "隐私", "处理", "委托处理", "信息安全"),
"dispute_resolution": ("争议", "仲裁", "诉讼", "管辖", "法院", "调解"),
"delivery_acceptance": ("交付", "验收", "质量", "瑕疵", "整改", "交付物"),
}
def __init__(self, *, threshold: float = 0.16):
self.threshold = max(0.0, min(1.0, threshold))
def match(self, source_text: str, target_text: str) -> LegalSemanticMatch:
source_vector = self.vectorize(source_text)
target_vector = self.vectorize(target_text)
if not source_vector or not target_vector:
return LegalSemanticMatch(False, 0.0, [], [])
score = self._cosine(source_vector, target_vector)
shared = [term for term in source_vector if term in target_vector]
shared_terms = [term for term in shared if not term.startswith("concept:")][:12]
shared_concepts = [term.replace("concept:", "", 1) for term in shared if term.startswith("concept:")]
return LegalSemanticMatch(score >= self.threshold, round(score, 4), shared_terms, shared_concepts)
def vectorize(self, text: str) -> Counter[str]:
normalized = self._normalize(text)
vector: Counter[str] = Counter()
if not normalized:
return vector
for token in self._word_tokens(normalized):
vector[token] += 2.0
for token in self._han_ngrams(normalized, n=2):
vector[token] += 0.35
for token in self._han_ngrams(normalized, n=3):
vector[token] += 0.5
for concept, aliases in self.LEGAL_CONCEPT_ALIASES.items():
matched_aliases = [alias for alias in aliases if alias in normalized]
if matched_aliases:
vector[f"concept:{concept}"] += 3.0
for alias in matched_aliases:
vector[alias] += 1.5
return vector
@staticmethod
def _normalize(text: str) -> str:
return re.sub(r"\s+", " ", str(text or "").lower()).strip()
@staticmethod
def _word_tokens(text: str) -> Iterable[str]:
for token in re.findall(r"[a-z0-9_]{2,}|[\u4e00-\u9fff]{2,}", text):
if len(token) <= 64:
yield token
@staticmethod
def _han_ngrams(text: str, *, n: int) -> Iterable[str]:
for chunk in re.findall(r"[\u4e00-\u9fff]{%d,}" % n, text):
for index in range(0, len(chunk) - n + 1):
yield chunk[index : index + n]
@staticmethod
def _cosine(left: Counter[str], right: Counter[str]) -> float:
shared = set(left).intersection(right)
numerator = sum(left[key] * right[key] for key in shared)
left_norm = math.sqrt(sum(value * value for value in left.values()))
right_norm = math.sqrt(sum(value * value for value in right.values()))
if left_norm == 0 or right_norm == 0:
return 0.0
return numerator / (left_norm * right_norm)
+11 -32
View File
@@ -14,6 +14,7 @@ from app.repositories.prompt_asset_evaluation_run_repository import PromptAssetE
from app.repositories.prompt_asset_repository import PromptAssetRepository
from app.repositories.prompt_asset_run_link_repository import PromptAssetRunLinkRepository
from app.repositories.prompt_asset_version_repository import PromptAssetVersionRepository
from app.services.prompt_golden_sample_evaluator import PromptGoldenSampleEvaluator
class PromptAssetService:
@@ -42,6 +43,7 @@ class PromptAssetService:
self.version_repository = PromptAssetVersionRepository(db)
self.run_link_repository = PromptAssetRunLinkRepository(db)
self.evaluation_run_repository = PromptAssetEvaluationRunRepository(db)
self.golden_sample_evaluator = PromptGoldenSampleEvaluator()
def list_prompt_assets(
self,
@@ -214,7 +216,7 @@ class PromptAssetService:
created_at=now,
)
)
report = self._build_prompt_evaluation_report(asset, version)
report = self._build_prompt_evaluation_report(current_user, asset, version)
version.evaluation_report_json = report
version.evaluation_score = float(report["quality_score"])
@@ -222,6 +224,7 @@ class PromptAssetService:
version = self.version_repository.update(version)
evaluation_run.status = version.evaluation_status
evaluation_run.suite_key = str(report.get("suite_key") or evaluation_run.suite_key)
evaluation_run.sample_count = int(report.get("sample_count") or 0)
evaluation_run.passed_samples = int(report.get("passed_samples") or 0)
evaluation_run.failed_samples = int(report.get("failed_samples") or 0)
@@ -474,38 +477,14 @@ class PromptAssetService:
asset.latest_evaluation_score = version.evaluation_score
asset.latest_evaluation_report_json = version.evaluation_report_json
def _build_prompt_evaluation_report(self, asset: PromptAsset, version: PromptAssetVersion) -> dict[str, Any]:
suite = self.PROMPT_GOLDEN_SUITES.get(asset.prompt_type, {"suite_key": "custom_prompt_suite", "sample_count": 1})
sample_count = int(suite["sample_count"])
def _build_prompt_evaluation_report(
self,
current_user: Dict[str, Any],
asset: PromptAsset,
version: PromptAssetVersion,
) -> dict[str, Any]:
checks = self._build_prompt_quality_checks(asset, version)
blocking_failures = [item for item in checks if item["required"] and item["status"] != "pass"]
warning_failures = [item for item in checks if not item["required"] and item["status"] != "pass"]
passed = not blocking_failures
quality_score = max(0.0, 1.0 - (0.25 * len(blocking_failures)) - (0.05 * len(warning_failures)))
if not passed:
quality_score = min(quality_score, 0.79)
passed_samples = sample_count if passed else max(0, sample_count - len(blocking_failures))
pass_rate = 1.0 if sample_count == 0 else round(passed_samples / sample_count, 4)
return {
"schema_version": "phase15-prompt-evaluation-report.v1",
"evaluator": "prompt_asset_api_static_gate",
"prompt_asset_id": asset.id,
"prompt_asset_version_id": version.id,
"prompt_key": asset.prompt_key,
"prompt_type": asset.prompt_type,
"version_label": version.version_label,
"suite_key": suite["suite_key"],
"sample_count": sample_count,
"passed_samples": passed_samples,
"failed_samples": sample_count - passed_samples,
"pass_rate": pass_rate,
"quality_score": round(quality_score, 4),
"passed": passed,
"failed_items": [item for item in checks if item["status"] != "pass"],
"checks": checks,
"evaluated_at": self._now().isoformat(),
"source": "prompt_assets.evaluate_current_version",
}
return self.golden_sample_evaluator.evaluate_prompt_asset(current_user, asset, version, static_checks=checks)
def _suite_key_for_prompt_type(self, prompt_type: str) -> str:
return str(self.PROMPT_GOLDEN_SUITES.get(prompt_type, {"suite_key": "custom_prompt_suite"})["suite_key"])
@@ -0,0 +1,539 @@
from __future__ import annotations
import csv
import json
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable, Mapping, Optional
from app.clients.shared_ai_platform import RemoteSharedAIPlatformClient
from app.core.config import settings
from app.models.prompt_asset import PromptAsset
from app.models.prompt_asset_version import PromptAssetVersion
class PromptGoldenSampleEvaluator:
"""Run Prompt assets against Phase13/Phase15 golden samples."""
SUITE_CONFIG: dict[str, dict[str, Any]] = {
"contract_generation": {
"suite_key": "phase13_generation",
"sample_family": "phase13_generation",
"min_samples": 8,
},
"contract_review": {
"suite_key": "phase13_review",
"sample_family": "phase13_review",
"min_samples": 8,
},
"legal_chat": {
"suite_key": "phase13_kb_qa",
"sample_family": "phase13_kb_qa",
"min_samples": 10,
},
"clause_recommendation": {
"suite_key": "phase13_clause_generation",
"sample_family": "phase13_clause_generation",
"min_samples": 6,
},
"wording_rewrite": {
"suite_key": "phase15_wording_rewrite",
"sample_family": "phase15_wording_rewrite",
"min_samples": 3,
},
"regulation_impact": {
"suite_key": "future_regulation_impact",
"sample_family": "future_regulation_impact",
"min_samples": 0,
},
}
def __init__(
self,
*,
repo_root: Optional[Path | str] = None,
llm_client: Optional[Any] = None,
evaluation_mode: Optional[str] = None,
include_raw_model_response: Optional[bool] = None,
):
self.repo_root = Path(repo_root) if repo_root is not None else Path(__file__).resolve().parents[3]
self.llm_client = llm_client
self.evaluation_mode = self._normalize_mode(evaluation_mode or settings.prompt_golden_eval_mode)
self.include_raw_model_response = (
settings.prompt_golden_eval_include_raw_model_response
if include_raw_model_response is None
else include_raw_model_response
)
def evaluate_prompt_asset(
self,
current_user: dict[str, Any],
asset: PromptAsset,
version: PromptAssetVersion,
*,
static_checks: list[dict[str, Any]],
) -> dict[str, Any]:
suite = self.SUITE_CONFIG.get(asset.prompt_type, self.SUITE_CONFIG["regulation_impact"])
samples = self.load_samples(asset.prompt_type)
llm_client = self._resolve_llm_client()
use_llm = llm_client is not None and self.evaluation_mode in {"auto", "llm"}
strict_llm = self.evaluation_mode == "llm"
sample_results: list[dict[str, Any]] = []
for sample in samples:
if use_llm:
sample_results.append(
self._evaluate_sample_with_llm(
current_user,
asset,
version,
sample,
llm_client=llm_client,
fallback_to_local=not strict_llm,
)
)
elif strict_llm:
sample_results.append(
self._failed_sample_result(
sample,
issue="llm_not_configured",
detail="Shared AI platform is not configured; real per-sample LLM evaluation cannot run.",
)
)
else:
sample_results.append(self._evaluate_sample_locally(asset, version, sample, static_checks))
required_static_failures = [item for item in static_checks if item.get("required") and item.get("status") != "pass"]
warning_static_failures = [item for item in static_checks if not item.get("required") and item.get("status") != "pass"]
passed_samples = sum(1 for item in sample_results if item.get("passed") is True)
sample_count = len(sample_results)
min_samples = int(suite.get("min_samples") or 0)
suite_issues = []
if sample_count < min_samples:
suite_issues.append(
{
"key": "sample_count_below_minimum",
"label": "Golden sample count is below the required minimum",
"required": True,
"status": "fail",
"detail": f"sample_count={sample_count}; min_samples={min_samples}",
}
)
sample_failures = [item for item in sample_results if item.get("passed") is not True]
passed = not required_static_failures and not suite_issues and not sample_failures
pass_rate = 1.0 if sample_count == 0 and min_samples == 0 else round(passed_samples / sample_count, 4) if sample_count else 0.0
sample_score = (
sum(float(item.get("score") or 0) for item in sample_results) / sample_count
if sample_count
else (1.0 if min_samples == 0 else 0.0)
)
quality_score = sample_score - (0.1 * len(required_static_failures)) - (0.03 * len(warning_static_failures)) - (0.15 * len(suite_issues))
if not passed:
quality_score = min(quality_score, 0.79)
quality_score = max(0.0, min(1.0, quality_score))
failed_items = [
*suite_issues,
*[item for item in static_checks if item.get("status") != "pass"],
*[
{
"key": f"sample_failed:{item.get('sample_id')}",
"label": f"Golden sample failed: {item.get('sample_id')}",
"required": True,
"status": "fail",
"detail": "; ".join(str(issue) for issue in item.get("issues") or [])[:500],
}
for item in sample_failures
],
]
return {
"schema_version": "phase15-prompt-evaluation-report.v2",
"phase": "Phase15-F",
"evaluator": "prompt_asset_api_llm_golden_sample_gate",
"evaluation_mode": "llm" if use_llm else "local_fallback" if not strict_llm else "llm_required_not_configured",
"prompt_asset_id": asset.id,
"prompt_asset_version_id": version.id,
"prompt_key": asset.prompt_key,
"prompt_type": asset.prompt_type,
"version_label": version.version_label,
"suite_key": suite["suite_key"],
"sample_family": suite["sample_family"],
"min_samples": min_samples,
"sample_count": sample_count,
"passed_samples": passed_samples,
"failed_samples": sample_count - passed_samples,
"pass_rate": pass_rate,
"quality_score": round(quality_score, 4),
"passed": passed,
"model_invocation": {
"configured": bool(use_llm),
"strict_llm": strict_llm,
"provider": "shared-ai-platform" if use_llm else None,
"mode": "llm" if use_llm else self.evaluation_mode,
"sample_invocation_count": sum(1 for item in sample_results if item.get("evaluator_mode") == "llm"),
},
"sample_results": sample_results,
"failed_items": failed_items,
"checks": static_checks,
"evaluated_at": datetime.now(timezone.utc).isoformat(),
"source": "prompt_assets.evaluate_current_version.golden_samples",
}
def load_samples(self, prompt_type: str) -> list[dict[str, Any]]:
if prompt_type == "contract_generation":
return [
self._generation_sample(row, sample_family="phase13_generation")
for row in self._read_phase13_csv("generation-golden-sample-inventory.csv")
if str(row.get("sample_id") or "").startswith("GEN-SAMPLE-TPL-")
]
if prompt_type == "clause_recommendation":
return [
self._generation_sample(row, sample_family="phase13_clause_generation")
for row in self._read_phase13_csv("generation-golden-sample-inventory.csv")
if str(row.get("sample_id") or "").startswith("GEN-SAMPLE-CLAUSE-")
]
if prompt_type == "contract_review":
return [self._review_sample(row) for row in self._read_phase13_csv("review-golden-sample-inventory.csv")]
if prompt_type == "legal_chat":
return [self._kb_qa_sample(row) for row in self._read_phase13_csv("kb-qa-golden-sample-inventory.csv")]
if prompt_type == "wording_rewrite":
return self._wording_rewrite_samples()
return []
def _evaluate_sample_with_llm(
self,
current_user: dict[str, Any],
asset: PromptAsset,
version: PromptAssetVersion,
sample: dict[str, Any],
*,
llm_client: Any,
fallback_to_local: bool,
) -> dict[str, Any]:
message = self._build_llm_evaluation_message(asset, version, sample)
try:
raw_text = self._collect_llm_text(
llm_client.stream_answer(
current_user,
None,
message,
review_context={
"task": "phase15_prompt_golden_sample_evaluation",
"prompt_type": asset.prompt_type,
"sample_id": sample.get("sample_id"),
},
template_key=None,
)
)
verdict = self._parse_llm_verdict(raw_text)
result = {
"sample_id": sample["sample_id"],
"prompt_type": sample["prompt_type"],
"sample_family": sample["sample_family"],
"status": "pass" if verdict["passed"] else "fail",
"passed": verdict["passed"],
"score": verdict["score"],
"evaluator_mode": "llm",
"input_summary": sample.get("input_summary"),
"expected_summary": sample.get("expected_summary"),
"observed_output_summary": verdict.get("observed_output_summary"),
"issues": verdict.get("issues") or [],
"matched_expectations": verdict.get("matched_expectations") or [],
"model_trace": verdict.get("model_trace") or {},
}
if self.include_raw_model_response:
result["raw_model_response"] = raw_text[:4000]
return result
except Exception as exc:
if fallback_to_local:
result = self._evaluate_sample_locally(asset, version, sample, [])
result["evaluator_mode"] = "local_fallback_after_llm_error"
result["issues"] = [*result.get("issues", []), f"llm_error:{str(exc)[:160]}"]
return result
return self._failed_sample_result(sample, issue="llm_call_failed", detail=str(exc)[:300])
def _evaluate_sample_locally(
self,
asset: PromptAsset,
version: PromptAssetVersion,
sample: dict[str, Any],
static_checks: list[dict[str, Any]],
) -> dict[str, Any]:
content = f"{version.system_prompt or ''}\n{version.prompt_content or ''}".lower()
required_failures = [item for item in static_checks if item.get("required") and item.get("status") != "pass"]
issues: list[str] = []
checks: list[dict[str, Any]] = []
if required_failures:
issues.append("required_static_prompt_checks_failed")
if not sample.get("expected"):
issues.append("sample_expected_payload_missing")
if sample.get("expected_citation_required") and not self._contains_any(content, ["law", "legal", "basis", "citation", "source", "article"]):
checks.append({"key": "citation_instruction", "status": "warn", "detail": "Prompt does not clearly require legal source citations."})
if asset.prompt_type == "wording_rewrite" and not self._contains_any(content, ["rewrite", "wording", "optimize", "revise"]):
issues.append("wording_rewrite_instruction_missing")
if asset.prompt_type == "contract_review" and not self._contains_any(content, ["review", "risk", "issue", "finding"]):
issues.append("contract_review_instruction_missing")
if asset.prompt_type == "legal_chat" and not self._contains_any(content, ["chat", "answer", "question", "reply"]):
checks.append({"key": "chat_answer_instruction", "status": "warn", "detail": "Prompt does not clearly ask the model to answer the user's question."})
passed = not issues
score = 0.94 if passed else 0.55
if checks:
score = min(score, 0.88)
return {
"sample_id": sample["sample_id"],
"prompt_type": sample["prompt_type"],
"sample_family": sample["sample_family"],
"status": "pass" if passed else "fail",
"passed": passed,
"score": round(score, 4),
"evaluator_mode": "local_fallback",
"input_summary": sample.get("input_summary"),
"expected_summary": sample.get("expected_summary"),
"observed_output_summary": "Local fallback evaluation: checked prompt/sample/expected/source-boundary alignment.",
"issues": issues,
"matched_expectations": [item for item in ["sample_loaded", "expected_loaded", "prompt_static_gate"] if item],
"checks": checks,
}
def _failed_sample_result(self, sample: dict[str, Any], *, issue: str, detail: str) -> dict[str, Any]:
return {
"sample_id": sample["sample_id"],
"prompt_type": sample["prompt_type"],
"sample_family": sample["sample_family"],
"status": "fail",
"passed": False,
"score": 0.0,
"evaluator_mode": "llm",
"input_summary": sample.get("input_summary"),
"expected_summary": sample.get("expected_summary"),
"observed_output_summary": detail,
"issues": [issue],
"matched_expectations": [],
}
def _build_llm_evaluation_message(self, asset: PromptAsset, version: PromptAssetVersion, sample: dict[str, Any]) -> str:
payload = {
"prompt_asset": {
"prompt_key": asset.prompt_key,
"prompt_type": asset.prompt_type,
"version_label": version.version_label,
"system_prompt": version.system_prompt,
"prompt_content": version.prompt_content,
"variables": version.variables_json or [],
"output_schema": version.output_schema_json,
"model_config": version.model_config_json,
},
"golden_sample": sample,
}
return (
"You are the Phase15-F prompt golden-sample evaluator for the AI contract platform. "
"Treat prompt_asset as the prompt under test, reason through the golden_sample input, "
"and compare the expected boundaries. Return JSON only, no Markdown.\n"
"JSON schema: {\"passed\":true|false,\"score\":0..1,\"observed_output_summary\":\"...\","
"\"issues\":[\"...\"],\"matched_expectations\":[\"...\"],\"model_trace\":{\"reason\":\"...\"}}.\n"
"Check sample expectations, legal-source grounding, and avoid over-confident legal conclusions.\n"
f"Evaluation input: {json.dumps(payload, ensure_ascii=False, default=str)}"
)
def _parse_llm_verdict(self, raw_text: str) -> dict[str, Any]:
payload = self._extract_json_object(raw_text)
passed = bool(payload.get("passed"))
score = payload.get("score")
try:
numeric_score = float(score)
except (TypeError, ValueError):
numeric_score = 1.0 if passed else 0.0
issues = payload.get("issues") if isinstance(payload.get("issues"), list) else []
matched = payload.get("matched_expectations") if isinstance(payload.get("matched_expectations"), list) else []
return {
"passed": passed,
"score": max(0.0, min(1.0, numeric_score)),
"observed_output_summary": str(payload.get("observed_output_summary") or "")[:1000],
"issues": [str(item)[:300] for item in issues],
"matched_expectations": [str(item)[:200] for item in matched],
"model_trace": payload.get("model_trace") if isinstance(payload.get("model_trace"), dict) else {},
}
def _extract_json_object(self, raw_text: str) -> dict[str, Any]:
text = str(raw_text or "").strip()
try:
payload = json.loads(text)
return payload if isinstance(payload, dict) else {}
except json.JSONDecodeError:
pass
match = re.search(r"\{.*\}", text, flags=re.S)
if not match:
return {"passed": False, "score": 0, "issues": ["model_response_not_json"], "observed_output_summary": text[:500]}
try:
payload = json.loads(match.group(0))
except json.JSONDecodeError:
return {"passed": False, "score": 0, "issues": ["model_response_json_parse_failed"], "observed_output_summary": text[:500]}
return payload if isinstance(payload, dict) else {}
def _collect_llm_text(self, chunks: Iterable[str]) -> str:
output: list[str] = []
for chunk in chunks:
text = str(chunk or "")
parsed = self._extract_sse_text(text)
output.append(parsed if parsed is not None else text)
return "".join(output).strip()
def _extract_sse_text(self, value: str) -> Optional[str]:
data_parts: list[str] = []
for line in value.splitlines():
if not line.startswith("data:"):
continue
raw = line[5:].strip()
if not raw:
continue
try:
payload = json.loads(raw)
except json.JSONDecodeError:
data_parts.append(raw)
continue
if isinstance(payload, Mapping):
if payload.get("type") in {"text", "delta"}:
data_parts.append(str(payload.get("content") or ""))
elif payload.get("content"):
data_parts.append(str(payload.get("content")))
if not data_parts:
return None
return "".join(data_parts)
def _resolve_llm_client(self) -> Optional[Any]:
if self.evaluation_mode == "local":
return None
if self.llm_client is not None:
return self.llm_client
if not settings.shared_ai_platform_base_url.strip() or not settings.shared_ai_platform_api_key.strip():
return None
return RemoteSharedAIPlatformClient(
base_url=settings.shared_ai_platform_base_url,
app_id=settings.shared_ai_platform_app_id,
api_key=settings.shared_ai_platform_api_key,
timeout_seconds=settings.shared_ai_platform_timeout_seconds,
)
def _read_phase13_csv(self, file_name: str) -> list[dict[str, str]]:
path = self.repo_root / "project-docs" / "phase13-legal-knowledge-template-library" / "inventories" / file_name
return self._read_csv(path)
def _read_csv(self, path: Path) -> list[dict[str, str]]:
if not path.exists():
return []
with path.open("r", encoding="utf-8-sig", newline="") as handle:
return [dict(row) for row in csv.DictReader(handle)]
def _generation_sample(self, row: Mapping[str, str], *, sample_family: str) -> dict[str, Any]:
base = self.repo_root / "project-docs" / "phase13-legal-knowledge-template-library"
input_payload = self._read_json_file(base / str(row.get("variable_payload_path") or ""))
expected_text = self._read_text_file(base / str(row.get("expected_document_path") or ""))
return {
"sample_id": row.get("sample_id"),
"prompt_type": "clause_recommendation" if sample_family == "phase13_clause_generation" else "contract_generation",
"sample_family": sample_family,
"input": input_payload,
"expected": {
"expected_document_excerpt": expected_text[:2000],
"expected_review_rule_ids": self._split_semicolon(row.get("expected_review_rule_ids")),
"expected_source_citations": self._split_semicolon(row.get("expected_source_citations")),
"contract_type": row.get("contract_type"),
"asset_under_test": row.get("template_id"),
},
"expected_citation_required": bool(row.get("expected_source_citations")),
"input_summary": f"{row.get('sample_id')} variables={len(input_payload) if isinstance(input_payload, Mapping) else 'n/a'}",
"expected_summary": f"rules={row.get('expected_review_rule_ids')}; sources={row.get('expected_source_citations')}",
}
def _review_sample(self, row: Mapping[str, str]) -> dict[str, Any]:
base = self.repo_root / "project-docs" / "phase13-legal-knowledge-template-library"
input_text = self._read_text_file(base / str(row.get("input_sample_path") or ""))
expected_text = self._read_text_file(base / str(row.get("expected_findings_path") or ""))
return {
"sample_id": row.get("sample_id"),
"prompt_type": "contract_review",
"sample_family": "phase13_review",
"input": {"contract_text": input_text, "contract_type": row.get("contract_type")},
"expected": {
"expected_findings_excerpt": expected_text[:2000],
"risk_categories": self._split_semicolon(row.get("risk_categories")),
"expected_severity": row.get("expected_severity"),
"expected_source_citations": self._split_semicolon(row.get("expected_source_citations")),
},
"expected_citation_required": bool(row.get("expected_source_citations")),
"input_summary": f"{row.get('sample_id')} contract_chars={len(input_text)}",
"expected_summary": f"severity={row.get('expected_severity')}; categories={row.get('risk_categories')}",
}
def _kb_qa_sample(self, row: Mapping[str, str]) -> dict[str, Any]:
return {
"sample_id": row.get("sample_id"),
"prompt_type": "legal_chat",
"sample_family": "phase13_kb_qa",
"input": {"question": row.get("question")},
"expected": {
"expected_answer_boundary": row.get("expected_answer_boundary"),
"expected_source_ids": self._split_semicolon(row.get("expected_source_ids")),
"expected_citation_required": str(row.get("expected_citation_required")).lower() == "true",
"expected_refusal_required": str(row.get("expected_refusal_required")).lower() == "true",
},
"expected_citation_required": str(row.get("expected_citation_required")).lower() == "true",
"input_summary": str(row.get("question") or "")[:200],
"expected_summary": str(row.get("expected_answer_boundary") or "")[:300],
}
def _wording_rewrite_samples(self) -> list[dict[str, Any]]:
path = self.repo_root / "project-docs" / "phase15-prompt-golden-samples" / "wording-rewrite-golden-samples.json"
payload = self._read_json_file(path)
samples = payload.get("samples") if isinstance(payload, Mapping) else []
results: list[dict[str, Any]] = []
for item in samples if isinstance(samples, list) else []:
if not isinstance(item, Mapping):
continue
input_payload = item.get("input") if isinstance(item.get("input"), Mapping) else {}
expected_payload = item.get("expected") if isinstance(item.get("expected"), Mapping) else {}
results.append(
{
"sample_id": item.get("sample_id"),
"prompt_type": "wording_rewrite",
"sample_family": "phase15_wording_rewrite",
"rewrite_type": item.get("rewrite_type"),
"document_type_code": item.get("document_type_code"),
"input": input_payload,
"expected": expected_payload,
"expected_citation_required": False,
"input_summary": str(input_payload.get("clause_text") or "")[:200],
"expected_summary": json.dumps(expected_payload, ensure_ascii=False)[:300],
}
)
return results
def _read_json_file(self, path: Path) -> Any:
if not path.exists() or not path.is_file():
return {}
try:
return json.loads(path.read_text(encoding="utf-8-sig"))
except json.JSONDecodeError:
return {}
def _read_text_file(self, path: Path) -> str:
if not path.exists() or not path.is_file():
return ""
return path.read_text(encoding="utf-8-sig")
@staticmethod
def _split_semicolon(value: Optional[str]) -> list[str]:
return [item.strip() for item in str(value or "").split(";") if item.strip()]
@staticmethod
def _normalize_mode(value: Optional[str]) -> str:
normalized = str(value or "auto").strip().lower().replace("-", "_")
if normalized not in {"auto", "llm", "local"}:
return "auto"
return normalized
@staticmethod
def _contains_any(value: str, candidates: list[str]) -> bool:
return any(candidate.lower() in value for candidate in candidates)
@@ -0,0 +1,116 @@
from __future__ import annotations
import argparse
import json
import time
from datetime import datetime, timezone
from typing import Any, Callable, Optional
from fastapi import HTTPException
from sqlalchemy.orm import Session
from app.core.config import settings
from app.db import SessionLocal
from app.repositories.legal_change_impact_repository import LegalChangeImpactRepository
from app.services.legal_change_impact_service import LegalChangeImpactService
class LegalChangeImpactWorker:
"""Consume pending legal-change events in a standalone worker process."""
def __init__(self, session_factory: Callable[[], Session] = SessionLocal):
self.session_factory = session_factory
def run_once(self, *, limit: Optional[int] = None) -> dict[str, Any]:
if not settings.legal_change_impact_worker_enabled:
return {
"worker": "legal_change_impact_worker",
"status": "disabled",
"completed_count": 0,
"failed_count": 0,
"completed": [],
"failed": [],
}
batch_size = max(1, min(limit or settings.legal_change_impact_worker_batch_size, 100))
db = self.session_factory()
try:
repository = LegalChangeImpactRepository(db)
events = repository.list_pending_events_for_worker(limit=batch_size)
completed: list[dict[str, Any]] = []
failed: list[dict[str, Any]] = []
for event in events:
current_user = {
"id": settings.legal_change_impact_worker_user_id,
"tenant_id": event.tenant_id,
"role": "admin",
}
try:
result = LegalChangeImpactService(db).run_event_scan(current_user, event.id)
completed.append(
{
"event_id": event.id,
"tenant_id": event.tenant_id,
"impact_summary": result["event"].get("impact_summary", {}),
}
)
except HTTPException as exc:
failed.append({"event_id": event.id, "tenant_id": event.tenant_id, "status_code": exc.status_code, "message": exc.detail})
except Exception as exc: # pragma: no cover - defensive worker safety net
self._mark_event_failed(db, event.tenant_id, event.id, str(exc))
failed.append({"event_id": event.id, "tenant_id": event.tenant_id, "status_code": 500, "message": str(exc)[:300]})
return {
"worker": "legal_change_impact_worker",
"status": "completed",
"requested_limit": batch_size,
"completed_count": len(completed),
"failed_count": len(failed),
"completed": completed,
"failed": failed,
"ran_at": datetime.now(timezone.utc).isoformat(),
}
finally:
db.close()
def run_forever(self, *, limit: Optional[int] = None, interval_seconds: Optional[int] = None, max_loops: Optional[int] = None) -> None:
interval = max(5, interval_seconds or settings.legal_change_impact_worker_interval_seconds)
loops = 0
while True:
result = self.run_once(limit=limit)
print(json.dumps(result, ensure_ascii=False), flush=True)
loops += 1
if max_loops is not None and loops >= max_loops:
return
time.sleep(interval)
@staticmethod
def _mark_event_failed(db: Session, tenant_id: str, event_id: str, message: str) -> None:
repository = LegalChangeImpactRepository(db)
event = repository.get_event_by_id(tenant_id, event_id)
if event is None:
return
event.scan_status = "failed"
event.impact_summary_json = {"error": message[:500], "total": 0, "blocking_count": 0}
event.updated_at = datetime.now(timezone.utc)
repository.update_event(event)
def main(argv: Optional[list[str]] = None) -> int:
parser = argparse.ArgumentParser(description="Run the legal-change impact background worker.")
parser.add_argument("--once", action="store_true", help="Run one batch and exit.")
parser.add_argument("--limit", type=int, default=None, help="Maximum pending events to consume per batch.")
parser.add_argument("--interval-seconds", type=int, default=None, help="Polling interval for daemon mode.")
parser.add_argument("--max-loops", type=int, default=None, help="Optional loop cap for smoke tests.")
args = parser.parse_args(argv)
worker = LegalChangeImpactWorker()
if args.once:
print(json.dumps(worker.run_once(limit=args.limit), ensure_ascii=False), flush=True)
return 0
worker.run_forever(limit=args.limit, interval_seconds=args.interval_seconds, max_loops=args.max_loops)
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -96,6 +96,36 @@ class ClauseRecommendationServiceTestCase(unittest.TestCase):
self.assertEqual(result["document_type_code"], "service_general")
self.assertEqual([item["clause_key"] for item in result["items"]], ["service_scope", "deliverables"])
def test_software_contract_returns_industry_practice_recommendations(self):
result = self.service.recommend_missing_clauses(
self.current_user,
document_type_code="software_development",
contract_text=(
"软件开发合同\n"
"一、服务范围:乙方为甲方开发客户管理系统。\n"
"二、交付物:乙方提交源代码、部署包和操作文档。\n"
"三、验收:甲方在收到交付物后五个工作日内完成验收。\n"
"四、知识产权:定制成果权利归甲方所有。\n"
"五、保密:双方对合作资料承担保密义务。\n"
"六、甲方配合义务:甲方提供接口账号和测试环境。\n"
"七、维护支持:乙方在验收后提供必要缺陷修复。"
),
max_items=3,
)
self.assertEqual(result["document_type_code"], "software_development")
self.assertGreaterEqual(result["summary"]["industry_practice_count"], 1)
self.assertEqual(result["summary"]["missing_clause_keys"], [])
first = result["items"][0]
self.assertEqual(first["clause_key"], "scope_change_control")
self.assertEqual(first["recommendation_kind"], "industry_practice")
self.assertEqual(first["basis_type"], "industry_practice_only")
self.assertIn("无直接法条", first["legal_basis"])
self.assertIn("industry_practice_rules", first["source_types"])
self.assertIn("no_direct_legal_article", first["source_types"])
self.assertEqual(first["legal_article_refs"], [])
self.assertIn(first["clause_key"], result["summary"]["industry_practice_keys"])
def test_professional_rewrite_expands_ambiguous_delivery_deadline(self):
result = self.service.rewrite_clause_wording(
self.current_user,
+11 -1
View File
@@ -32,6 +32,9 @@ class StubClauseRecommendationService:
"clause_key": "acceptance",
"clause_title": "验收标准与流程",
"clause_type": "acceptance",
"recommendation_kind": "missing_clause",
"basis_type": "legal_basis",
"industry_context": None,
"priority": "high",
"confidence": "high",
"insert_position": {
@@ -42,6 +45,7 @@ class StubClauseRecommendationService:
"suggested_text": "验收标准与流程:甲方应在收到货物后完成验收。",
"reason": "买卖合同需要把交付和验收衔接起来。",
"legal_basis": "《民法典》合同履行规则。",
"practice_basis": None,
"legal_article_refs": [],
"source_rule": "LAW-CLAUSE-PROCUREMENT-ACCEPTANCE",
"source_types": ["document_type_required_clause_list", "standard_clause_library"],
@@ -51,10 +55,15 @@ class StubClauseRecommendationService:
"total": 1,
"summary": {
"recommended_count": 1,
"recommendation_keys": ["acceptance"],
"detected_clause_keys": [],
"missing_clause_keys": ["acceptance"],
"required_missing_clause_keys": ["acceptance"],
"industry_practice_keys": [],
"detected_industry_practice_keys": [],
"industry_practice_count": 0,
"source_count": 0,
"strategy": "rule_based_required_clause_gap",
"strategy": "rule_based_required_clause_gap_and_industry_practice",
},
"sources": ["document_type_required_clause_list"],
}
@@ -128,6 +137,7 @@ class ClausesApiTestCase(unittest.TestCase):
self.assertEqual(self.clause_service.calls[0][1]["contract_id"], "contract-001")
self.assertEqual(self.audit_log_service.events[0]["action"], "recommend_missing_clauses")
self.assertEqual(self.audit_log_service.events[0]["detail_json"]["missing_clause_keys"], ["acceptance"])
self.assertEqual(self.audit_log_service.events[0]["detail_json"]["industry_practice_count"], 0)
def test_rewrite_clause_wording_returns_result_and_records_audit(self):
response = self.client.post(
@@ -11,6 +11,7 @@ from app.models.contract_template import ContractTemplate
from app.models.generated_contract import GeneratedContract
from app.models.legal_article import LegalArticle
from app.models.review_rule import ReviewRule
from app.tasks.legal_change_impact_worker import LegalChangeImpactWorker
from app.services.legal_change_impact_service import LegalChangeImpactService
@@ -19,6 +20,7 @@ class LegalChangeImpactServiceTestCase(unittest.TestCase):
self.engine = create_engine("sqlite:///:memory:", future=True)
Base.metadata.create_all(bind=self.engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=self.engine, future=True)
self.SessionLocal = SessionLocal
self.db = SessionLocal()
self.current_user = {"id": "user-admin", "tenant_id": "tenant-001", "role": "admin"}
self.service = LegalChangeImpactService(self.db)
@@ -134,6 +136,54 @@ class LegalChangeImpactServiceTestCase(unittest.TestCase):
self.assertEqual(result["completed"][0]["event"]["scan_status"], "completed")
self.assertEqual(len(result["completed"][0]["impacts"]), 3)
def test_semantic_vector_scan_finds_related_contract_without_direct_reference(self):
semantic_contract = GeneratedContract(
id="contract-semantic-001",
tenant_id="tenant-001",
template_id="tpl-semantic",
template_name="通用服务协议",
contract_type="service",
document_type_code="service_general",
generated_by="user-admin",
generated_by_name="Admin",
export_format="docx",
file_name="服务协议.docx",
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
file_size=456,
generated_content="若任一方未按约履行交付义务,应承担迟延履行责任并补偿对方实际损失。",
file_content_base64="SGVsbG8=",
status="generated",
)
self.db.add(semantic_contract)
self.db.commit()
result = self.service.scan_article_change(self.current_user, article_id="lart-civil-585")
impact = next(item for item in result["impacts"] if item["target_id"] == "contract-semantic-001")
self.assertIn("semantic_vector_match", impact["match_reason"])
self.assertEqual(impact["impact_level"], "reference_only")
self.assertGreaterEqual(impact["evidence"]["semantic_vector"]["score"], 0.10)
self.assertIn("breach_liability", impact["evidence"]["semantic_vector"]["shared_concepts"])
self.assertGreaterEqual(result["event"]["impact_summary"]["semantic_match_count"], 1)
def test_background_worker_consumes_pending_events(self):
queued = self.service.enqueue_article_change(
self.current_user,
article_id="lart-civil-585",
change_type="updated",
change_summary="后台 worker 消费",
)
worker = LegalChangeImpactWorker(session_factory=self.SessionLocal)
worker_result = worker.run_once(limit=1)
self.db.expire_all()
detail = self.service.get_event_detail(self.current_user, queued["event"]["id"])
self.assertEqual(worker_result["completed_count"], 1)
self.assertEqual(worker_result["failed_count"], 0)
self.assertEqual(detail["event"]["scan_status"], "completed")
self.assertEqual(detail["event"]["impact_summary"]["total"], 3)
def test_update_impact_status_records_handling_note(self):
result = self.service.scan_article_change(self.current_user, article_id="lart-civil-585")
impact_id = result["impacts"][0]["id"]
+11 -2
View File
@@ -38,7 +38,14 @@ class StubLegalChangeImpactService:
"effective_date": "2021-01-01",
"tags": ["违约金"],
"scan_status": "completed",
"impact_summary": {"total": 1, "blocking_count": 1, "by_level": {"immediate_update": 1}},
"impact_summary": {
"total": 1,
"blocking_count": 1,
"by_level": {"immediate_update": 1},
"by_match_reason": {"semantic_vector_match": 1},
"semantic_match_count": 1,
"scan_strategy": ["direct_reference", "tag_match", "keyword_match", "semantic_vector_match"],
},
"created_by": "user-admin",
"created_at": "2026-05-21T00:00:00+00:00",
"scanned_at": "2026-05-21T00:01:00+00:00",
@@ -61,7 +68,7 @@ class StubLegalChangeImpactService:
"impact_level": "immediate_update",
"suggestion": "请立即检查模板。",
"impact_status": "open",
"evidence": {},
"evidence": {"semantic_vector": {"score": 0.42, "shared_concepts": ["breach_liability"]}},
"handled_by": None,
"handled_at": None,
"created_at": "2026-05-21T00:01:00+00:00",
@@ -151,6 +158,8 @@ class LegalChangeImpactsApiTestCase(unittest.TestCase):
"run_pending_legal_change_scans",
"update_legal_change_impact_status",
])
self.assertEqual(self.audit_log_service.events[0]["detail_json"]["semantic_match_count"], 1)
self.assertTrue(self.audit_log_service.events[2]["detail_json"]["worker_compatible"])
def test_member_cannot_access_legal_change_impacts(self):
app.dependency_overrides[require_current_user] = lambda: {
+74 -6
View File
@@ -1,6 +1,7 @@
import json
import unittest
from datetime import datetime, timezone
from pathlib import Path
from fastapi import HTTPException
@@ -8,6 +9,7 @@ from app.models.prompt_asset import PromptAsset
from app.models.prompt_asset_evaluation_run import PromptAssetEvaluationRun
from app.models.prompt_asset_run_link import PromptAssetRunLink
from app.models.prompt_asset_version import PromptAssetVersion
from app.services.prompt_golden_sample_evaluator import PromptGoldenSampleEvaluator
from app.services.prompt_asset_service import PromptAssetService
@@ -117,6 +119,32 @@ class StubPromptAssetEvaluationRunRepository:
return [item for item in self.items.values() if item.tenant_id == tenant_id and item.prompt_asset_id == asset_id][:limit]
class FakeGoldenSampleLLMClient:
def __init__(self):
self.calls = []
def stream_answer(self, current_user, conversation_id, message, review_context=None, template_key=None):
self.calls.append(
{
"current_user": current_user,
"conversation_id": conversation_id,
"message": message,
"review_context": review_context,
"template_key": template_key,
}
)
yield json.dumps(
{
"passed": True,
"score": 0.97,
"observed_output_summary": "Sample expectations are satisfied.",
"issues": [],
"matched_expectations": ["expected_boundary", "legal_safety"],
"model_trace": {"reason": "fake evaluator accepted this sample"},
}
)
class PromptAssetServiceTestCase(unittest.TestCase):
def setUp(self):
self.service = PromptAssetService(db=None)
@@ -127,13 +155,21 @@ class PromptAssetServiceTestCase(unittest.TestCase):
self.admin_user = {"id": "user-admin", "tenant_id": "tenant-demo-001", "role": "admin", "display_name": "Admin"}
self.member_user = {"id": "user-member", "tenant_id": "tenant-demo-001", "role": "member"}
def _create_asset(self, evaluation=True):
def _create_asset(
self,
evaluation=True,
*,
prompt_key="contract_generation_default",
prompt_type="contract_generation",
name="Contract generation default",
prompt_content="Generate a contract from structured elements.",
):
return self.service.create_prompt_asset(
self.admin_user,
prompt_key="contract_generation_default",
prompt_type="contract_generation",
name="Contract generation default",
prompt_content="Generate a contract from structured elements.",
prompt_key=prompt_key,
prompt_type=prompt_type,
name=name,
prompt_content=prompt_content,
variables_json=json.dumps([{"name": "party_a", "required": True}]),
output_schema_json=json.dumps({"type": "object"}),
model_config_json=json.dumps({"model": "gpt-5.4"}),
@@ -168,10 +204,42 @@ class PromptAssetServiceTestCase(unittest.TestCase):
self.assertEqual(evaluation["asset"]["latest_evaluation_status"], "passed")
self.assertEqual(evaluation["version"]["evaluation_status"], "passed")
self.assertEqual(evaluation["evaluation_report"]["suite_key"], "phase13_generation")
self.assertEqual(evaluation["evaluation_report"]["schema_version"], "phase15-prompt-evaluation-report.v2")
self.assertEqual(evaluation["evaluation_report"]["evaluation_mode"], "local_fallback")
self.assertGreaterEqual(evaluation["evaluation_run"]["sample_count"], 8)
self.assertEqual(evaluation["evaluation_run"]["sample_count"], len(evaluation["evaluation_report"]["sample_results"]))
self.assertEqual(evaluation["evaluation_run"]["status"], "passed")
self.assertEqual(evaluation["evaluation_run"]["sample_count"], 8)
self.assertEqual(published["status"], "published")
def test_run_evaluation_invokes_llm_once_per_golden_sample(self):
fake_llm_client = FakeGoldenSampleLLMClient()
self.service.golden_sample_evaluator = PromptGoldenSampleEvaluator(
repo_root=Path.cwd(),
llm_client=fake_llm_client,
evaluation_mode="llm",
)
created = self._create_asset(
evaluation=False,
prompt_key="wording_rewrite_default",
prompt_type="wording_rewrite",
name="Wording rewrite default",
prompt_content="Rewrite and optimize clause wording, explain risk boundaries, and preserve legal safety.",
)
evaluation = self.service.evaluate_current_prompt_asset(self.admin_user, created["id"])
self.assertEqual(evaluation["version"]["evaluation_status"], "passed")
self.assertEqual(evaluation["evaluation_report"]["suite_key"], "phase15_wording_rewrite")
self.assertEqual(evaluation["evaluation_report"]["evaluation_mode"], "llm")
self.assertGreaterEqual(evaluation["evaluation_run"]["sample_count"], 3)
self.assertEqual(len(fake_llm_client.calls), evaluation["evaluation_run"]["sample_count"])
self.assertEqual(
evaluation["evaluation_report"]["model_invocation"]["sample_invocation_count"],
evaluation["evaluation_run"]["sample_count"],
)
self.assertTrue(all(item["evaluator_mode"] == "llm" for item in evaluation["evaluation_report"]["sample_results"]))
self.assertTrue(all(call["current_user"]["id"] == self.admin_user["id"] for call in fake_llm_client.calls))
def test_publish_with_report_and_create_next_version(self):
created = self._create_asset(evaluation=False)
published = self.service.publish_prompt_asset(
+44 -4
View File
@@ -107,7 +107,16 @@ class StubPromptAssetService:
"failed_samples": 0,
"pass_rate": 1.0,
"quality_score": 0.95,
"report": {"suite_key": "phase13_generation", "passed": True},
"report": {
"schema_version": "phase15-prompt-evaluation-report.v2",
"suite_key": "phase13_generation",
"evaluation_mode": "llm",
"sample_count": 8,
"passed_samples": 8,
"failed_samples": 0,
"sample_results": [{"sample_id": "GEN-SAMPLE-TPL-001", "passed": True}],
"passed": True,
},
"failure_reason": None,
"triggered_by": "user-admin",
"started_at": "2026-05-21T00:00:00+00:00",
@@ -143,7 +152,16 @@ class StubPromptAssetService:
"model_config": None,
"evaluation_status": "passed",
"evaluation_score": 0.95,
"evaluation_report": {"suite_key": "phase13_generation", "passed": True},
"evaluation_report": {
"schema_version": "phase15-prompt-evaluation-report.v2",
"suite_key": "phase13_generation",
"evaluation_mode": "llm",
"sample_count": 8,
"passed_samples": 8,
"failed_samples": 0,
"sample_results": [{"sample_id": "GEN-SAMPLE-TPL-001", "passed": True}],
"passed": True,
},
"change_summary": "Initial prompt version",
"rollback_from_version_id": None,
"is_current": True,
@@ -154,7 +172,17 @@ class StubPromptAssetService:
"published_at": None,
"published_by": None,
},
"evaluation_report": {"suite_key": "phase13_generation", "passed": True, "quality_score": 0.95},
"evaluation_report": {
"schema_version": "phase15-prompt-evaluation-report.v2",
"suite_key": "phase13_generation",
"evaluation_mode": "llm",
"sample_count": 8,
"passed_samples": 8,
"failed_samples": 0,
"sample_results": [{"sample_id": "GEN-SAMPLE-TPL-001", "passed": True}],
"passed": True,
"quality_score": 0.95,
},
"evaluation_run": {
"id": "prompteval-123",
"tenant_id": "tenant-demo-001",
@@ -169,7 +197,16 @@ class StubPromptAssetService:
"failed_samples": 0,
"pass_rate": 1.0,
"quality_score": 0.95,
"report": {"suite_key": "phase13_generation", "passed": True},
"report": {
"schema_version": "phase15-prompt-evaluation-report.v2",
"suite_key": "phase13_generation",
"evaluation_mode": "llm",
"sample_count": 8,
"passed_samples": 8,
"failed_samples": 0,
"sample_results": [{"sample_id": "GEN-SAMPLE-TPL-001", "passed": True}],
"passed": True,
},
"failure_reason": None,
"triggered_by": current_user["id"],
"started_at": "2026-05-21T00:00:00+00:00",
@@ -264,6 +301,7 @@ class PromptAssetsApiTestCase(unittest.TestCase):
self.assertEqual(version_response.json()["data"]["current_version_label"], "v1.1")
self.assertEqual(evaluation_response.json()["data"]["version"]["evaluation_status"], "passed")
self.assertEqual(evaluation_response.json()["data"]["evaluation_run"]["status"], "passed")
self.assertEqual(evaluation_response.json()["data"]["evaluation_report"]["evaluation_mode"], "llm")
self.assertEqual(publish_response.json()["data"]["status"], "published")
self.assertEqual(rollback_response.json()["data"]["current_version_label"], "v1.2")
self.assertEqual(run_link_response.json()["data"]["prompt_asset_version_id"], "promptver-123")
@@ -275,6 +313,8 @@ class PromptAssetsApiTestCase(unittest.TestCase):
"rollback_prompt_asset",
"record_prompt_asset_run_link",
])
self.assertEqual(self.audit_log_service.events[2]["detail_json"]["evaluation_mode"], "llm")
self.assertEqual(self.audit_log_service.events[2]["detail_json"]["sample_count"], 8)
def test_member_cannot_list_prompt_assets(self):
app.dependency_overrides[require_current_user] = lambda: {
+37
View File
@@ -55,6 +55,43 @@ services:
ports:
- "8001:8000"
legal-change-impact-worker:
build:
context: ./backend
dockerfile: Dockerfile
restart: unless-stopped
depends_on:
postgres:
condition: service_healthy
backend:
condition: service_started
environment:
APP_NAME: AI Contract Platform
APP_ENV: production
SECRET_KEY: ${SECRET_KEY:?Set SECRET_KEY in .env before starting production}
DATABASE_URL: postgresql+psycopg://postgres:postgres@postgres:5432/ai_contract_platform
REDIS_URL: redis://redis:6379/0
MINIO_ENDPOINT: 192.168.3.90:9000
MINIO_ACCESS_KEY: minioadmin
MINIO_SECRET_KEY: minioadmin
MINIO_BUCKET: contracts
SHARED_AI_PLATFORM_BASE_URL: http://192.168.3.90:18011
SHARED_AI_PLATFORM_API_PREFIX: /api/v0.1
SHARED_AI_PLATFORM_APP_ID: ai-contract-platform
SHARED_AI_PLATFORM_API_KEY: "shared-ai-platform-admin-dev-token"
SHARED_AI_PLATFORM_TIMEOUT_SECONDS: 180
SHARED_AI_PLATFORM_LEGAL_REVIEW_ENABLED: "true"
SHARED_AI_PLATFORM_LEGAL_REVIEW_FALLBACK_TO_LOCAL: "false"
LEGAL_CHANGE_IMPACT_WORKER_BATCH_SIZE: 10
LEGAL_CHANGE_IMPACT_WORKER_INTERVAL_SECONDS: 300
LEGAL_CHANGE_IMPACT_SEMANTIC_ENABLED: "true"
command: >
python -m app.tasks.legal_change_impact_worker
--limit 10
--interval-seconds 300
volumes:
- backend_storage:/app/storage
frontend:
build:
context: ./frontend
+28 -5
View File
@@ -57,6 +57,16 @@ const clausePriorityColorMap: Record<string, string> = {
low: "blue",
};
const recommendationKindLabelMap: Record<string, string> = {
missing_clause: "缺失条款",
industry_practice: "行业惯例",
};
const recommendationKindColorMap: Record<string, string> = {
missing_clause: "orange",
industry_practice: "purple",
};
function formatDateTime(value?: string | null) {
if (!value) {
return "-";
@@ -111,6 +121,10 @@ function getErrorMessage(error: unknown, fallback: string) {
return error instanceof Error && error.message ? error.message : fallback;
}
function readableRecommendationKind(kind?: string | null) {
return recommendationKindLabelMap[kind ?? ""] ?? "条款建议";
}
export default function GeneratedContractEditorPage({ params }: { params: Promise<{ id: string }> }) {
const [messageApi, contextHolder] = message.useMessage();
const router = useRouter();
@@ -474,7 +488,7 @@ export default function GeneratedContractEditorPage({ params }: { params: Promis
showIcon
type="info"
message="编辑器说明"
description="左侧正文可直接编辑;右侧可选择段落、检查缺失条款、改写条款并把结果插入正文。保存后会生成可追溯的新版本。"
description="左侧正文可直接编辑;右侧可选择段落、检查缺失条款和行业惯例、改写条款并把结果插入正文。保存后会生成可追溯的新版本。"
/>
</section>
@@ -486,7 +500,7 @@ export default function GeneratedContractEditorPage({ params }: { params: Promis
</div>
<Space direction="vertical" size="small" style={{ width: "100%" }}>
<Button block loading={recommendationLoading} onClick={() => void handleRecommend()}>
检查缺失条款
检查条款与惯例
</Button>
<Button block icon={<MessageOutlined />} onClick={() => router.push(buildEditorChatHref(contract, editorContent))}>
带正文追问 Chat
@@ -551,7 +565,10 @@ export default function GeneratedContractEditorPage({ params }: { params: Promis
<section className="document-side-panel">
<div className="document-side-panel-header">
<BulbOutlined />
<strong>缺失条款推荐</strong>
<strong>条款与行业惯例推荐</strong>
{recommendations?.summary.industry_practice_count ? (
<Tag color="purple">{recommendations.summary.industry_practice_count} 条惯例</Tag>
) : null}
</div>
{recommendations?.items.length ? (
<div className="document-editor-recommend-list">
@@ -559,10 +576,16 @@ export default function GeneratedContractEditorPage({ params }: { params: Promis
<article className="document-editor-recommend-card" key={item.id}>
<div>
<strong>{item.clause_title}</strong>
<Tag color={recommendationKindColorMap[item.recommendation_kind ?? ""] ?? "blue"}>
{readableRecommendationKind(item.recommendation_kind)}
</Tag>
<Tag color={clausePriorityColorMap[item.priority] ?? "default"}>{item.priority}</Tag>
</div>
<Text type="secondary">{item.insert_position.description}</Text>
<Paragraph>{item.suggested_text}</Paragraph>
<Text type="secondary">
{item.practice_basis ?? item.legal_basis}
</Text>
<Button size="small" type="primary" onClick={() => void handleApplyRecommendation(item)}>
插入正文并生成版本
</Button>
@@ -570,9 +593,9 @@ export default function GeneratedContractEditorPage({ params }: { params: Promis
))}
</div>
) : recommendations ? (
<Text type="secondary">当前未发现首批必备条款缺失。</Text>
<Text type="secondary">当前未发现必备条款缺失或行业惯例建议。</Text>
) : (
<Text type="secondary">点击“检查缺失条款”后展示推荐。</Text>
<Text type="secondary">点击“检查条款与惯例”后展示推荐。</Text>
)}
</section>
+35 -7
View File
@@ -134,6 +134,16 @@ const clausePriorityLabelMap: Record<string, string> = {
low: "低优先级",
};
const recommendationKindLabelMap: Record<string, string> = {
missing_clause: "缺失条款",
industry_practice: "行业惯例",
};
const recommendationKindColorMap: Record<string, string> = {
missing_clause: "orange",
industry_practice: "purple",
};
const clauseRewriteModeOptions = [
{ label: "专业化", value: "professional" },
{ label: "简化解释", value: "simplify" },
@@ -224,6 +234,10 @@ function getErrorMessage(error: unknown, fallback: string) {
return error instanceof Error && error.message ? error.message : fallback;
}
function readableRecommendationKind(kind?: string | null) {
return recommendationKindLabelMap[kind ?? ""] ?? "条款建议";
}
function renderField(field: LegalDocumentElementField) {
if (field.type === "select") {
return <Select options={normalizeOptions(field)} placeholder={`请选择${readableFieldLabel(field)}`} />;
@@ -767,17 +781,22 @@ export default function LegalDocumentBuilderPage({ params }: { params: Promise<{
<div className="document-clause-recommendations">
<div className="document-clause-recommendations-header">
<strong><BulbOutlined /> 智能条款推荐</strong>
<strong><BulbOutlined /> 智能条款与行业惯例推荐</strong>
{clauseRecommendations ? (
<Tag color={clauseRecommendations.total ? "orange" : "green"}>
{clauseRecommendations.total ? `${clauseRecommendations.total} 条建议` : "暂无缺失条款"}
</Tag>
<Space size={6} wrap>
<Tag color={clauseRecommendations.total ? "orange" : "green"}>
{clauseRecommendations.total ? `${clauseRecommendations.total} 条建议` : "暂无建议"}
</Tag>
{clauseRecommendations.summary.industry_practice_count ? (
<Tag color="purple">{clauseRecommendations.summary.industry_practice_count} 条行业惯例</Tag>
) : null}
</Space>
) : null}
</div>
{clauseRecommendationLoading ? (
<div className="document-clause-loading">
<Spin size="small" />
<Text type="secondary">正在检查缺失条款...</Text>
<Text type="secondary">正在检查缺失条款和行业惯例...</Text>
</div>
) : clauseRecommendations?.items.length ? (
<div className="document-clause-list">
@@ -786,10 +805,17 @@ export default function LegalDocumentBuilderPage({ params }: { params: Promise<{
<div className="document-clause-card-head">
<strong>{item.clause_title}</strong>
<Space size={6} wrap>
<Tag color={recommendationKindColorMap[item.recommendation_kind ?? ""] ?? "blue"}>
{readableRecommendationKind(item.recommendation_kind)}
</Tag>
<Tag color={clausePriorityColorMap[item.priority] ?? "default"}>
{clausePriorityLabelMap[item.priority] ?? item.priority}
</Tag>
<Tag>{item.confidence === "high" ? "依据充分" : "规则建议"}</Tag>
<Tag>
{item.basis_type === "industry_practice_only"
? "无直接法条"
: item.confidence === "high" ? "依据充分" : "规则建议"}
</Tag>
<Button
size="small"
icon={<EditOutlined />}
@@ -810,6 +836,8 @@ export default function LegalDocumentBuilderPage({ params }: { params: Promise<{
<div className="document-clause-meta">
<span>推荐理由:{item.reason}</span>
<span>依据:{item.legal_basis}</span>
{item.practice_basis ? <span>惯例说明:{item.practice_basis}</span> : null}
{item.industry_context ? <span>适用场景:{item.industry_context}</span> : null}
</div>
{item.legal_article_refs.length ? (
<div className="document-clause-refs">
@@ -824,7 +852,7 @@ export default function LegalDocumentBuilderPage({ params }: { params: Promise<{
))}
</div>
) : (
<Text type="secondary">当前生成正文暂未发现需要补充的首批必备条款。</Text>
<Text type="secondary">当前生成正文暂未发现需要补充的必备条款或行业惯例建议。</Text>
)}
</div>
+46 -10
View File
@@ -86,6 +86,29 @@ function impactStatusTag(value: string) {
return <Tag color={meta.color}>{meta.label}</Tag>;
}
function matchReasonTags(item: LegalChangeImpactItem) {
const reasons = item.match_reason.split(",").filter(Boolean);
return reasons.map((reason) => {
if (reason === "semantic_vector_match") return <Tag color="purple" key={reason}>语义向量</Tag>;
if (reason === "direct_legal_reference") return <Tag color="red" key={reason}>直接引用</Tag>;
if (reason === "tag_match") return <Tag color="blue" key={reason}>标签</Tag>;
if (reason === "keyword_match") return <Tag color="gold" key={reason}>关键词</Tag>;
return <Tag key={reason}>{reason}</Tag>;
});
}
function getSemanticEvidence(item: LegalChangeImpactItem) {
const semantic = item.evidence?.semantic_vector;
if (!semantic || typeof semantic !== "object") {
return null;
}
const payload = semantic as { score?: unknown; shared_concepts?: unknown };
return {
score: typeof payload.score === "number" ? payload.score : null,
sharedConcepts: Array.isArray(payload.shared_concepts) ? payload.shared_concepts.map(String) : [],
};
}
export default function LegalChangeImpactsPage() {
const [messageApi, contextHolder] = message.useMessage();
const currentUser = useAuthStore((state) => state.currentUser);
@@ -294,6 +317,7 @@ export default function LegalChangeImpactsPage() {
<Text type={(item.impact_summary?.blocking_count ?? 0) > 0 ? "danger" : "secondary"}>
立即修改:{item.impact_summary?.blocking_count ?? 0}
</Text>
<Text type="secondary">语义命中:{item.impact_summary?.semantic_match_count ?? 0}</Text>
</Space>
),
},
@@ -361,14 +385,25 @@ export default function LegalChangeImpactsPage() {
title: "命中原因",
key: "match",
width: 260,
render: (_, item) => (
<Space wrap size={4}>
{item.match_terms.slice(0, 5).map((term) => (
<Tag key={term}>{term}</Tag>
))}
{item.match_terms.length > 5 ? <Tag>+{item.match_terms.length - 5}</Tag> : null}
</Space>
),
render: (_, item) => {
const semantic = getSemanticEvidence(item);
return (
<Space direction="vertical" size={4}>
<Space wrap size={4}>{matchReasonTags(item)}</Space>
<Space wrap size={4}>
{item.match_terms.slice(0, 5).map((term) => (
<Tag key={term}>{term}</Tag>
))}
{item.match_terms.length > 5 ? <Tag>+{item.match_terms.length - 5}</Tag> : null}
</Space>
{semantic ? (
<Text type="secondary">
相似度 {semantic.score ?? "-"} {semantic.sharedConcepts.length ? ` / ${semantic.sharedConcepts.join(", ")}` : ""}
</Text>
) : null}
</Space>
);
},
},
{
title: "处理建议",
@@ -419,7 +454,7 @@ export default function LegalChangeImpactsPage() {
<div className="review-page-hero-copy">
<span className="landing-badge">Regulation Impact</span>
<Title level={2}>法规变更影响分析</Title>
<Text type="secondary">当法条新增、修改、生效状态变化时,主动扫描模板、审查规则和历史合同,给出处理等级和建议。</Text>
<Text type="secondary">当法条新增、修改、生效状态变化时,主动扫描模板、审查规则和历史合同,结合直接引用、标签关键词和语义向量命中给出处理等级和建议。</Text>
</div>
<Card className="review-side-card" variant="borderless">
<span className="landing-panel-eyebrow">Phase15-G</span>
@@ -429,7 +464,8 @@ export default function LegalChangeImpactsPage() {
<ul className="review-page-tip-list">
<li>立即修改:模板或规则直接引用法条。</li>
<li>续约时修改:历史合同直接引用法条。</li>
<li>仅供参考:标签或关键词相关。</li>
<li>仅供参考:标签、关键词或语义向量相关。</li>
<li>后台 worker 可定时消费待扫描队列。</li>
</ul>
</Card>
</section>
+42 -26
View File
@@ -36,7 +36,7 @@ import type {
PromptAssetVersionItem,
} from "@/types/api";
const { Title, Text, Paragraph } = Typography;
const { Title, Text } = Typography;
type PromptAssetFormValues = {
prompt_key: string;
@@ -108,6 +108,19 @@ function getEvaluationTag(value?: string | null) {
return <Tag color="default">未评测</Tag>;
}
function getEvaluationModeTag(report?: Record<string, unknown> | null) {
const mode = typeof report?.evaluation_mode === "string" ? report.evaluation_mode : null;
if (mode === "llm") return <Tag color="purple">逐条大模型</Tag>;
if (mode === "local_fallback") return <Tag color="blue">本地兜底</Tag>;
if (mode === "llm_required_not_configured") return <Tag color="red">模型未配置</Tag>;
return <Tag>未知</Tag>;
}
function getEvaluationSampleResultCount(report?: Record<string, unknown> | null) {
const results = report?.sample_results;
return Array.isArray(results) ? results.length : 0;
}
function getPreviousVersionLabel(currentVersionLabel: string) {
const [, minorValue] = currentVersionLabel.replace(/^v/, "").split(".");
const minor = Number(minorValue ?? "0");
@@ -303,7 +316,8 @@ export default function PromptAssetsPage() {
return;
}
const statusText = result.data.version.evaluation_status === "passed" ? "通过" : "未通过";
messageApi.success(`Prompt 评测已完成:${statusText}`);
const modeText = result.data.evaluation_report.evaluation_mode === "llm" ? "逐条大模型" : "本地兜底";
messageApi.success(`Prompt 评测已完成:${statusText},${modeText}`);
await refreshAfterMutation(item.id);
} catch (error) {
messageApi.error(error instanceof Error ? error.message : "Prompt 评测失败");
@@ -368,30 +382,21 @@ export default function PromptAssetsPage() {
const handleRecordRunLink = async (values: RunLinkFormValues) => {
if (!selectedDetail) {
messageApi.warning("请先选择一个 Prompt 资产");
return;
}
setMutatingId(selectedDetail.asset.id);
try {
assertJson(values.trace_json);
const result = await recordPromptRunLink(selectedDetail.asset.id, {
run_type: values.run_type,
run_id: values.run_id?.trim(),
resource_type: values.resource_type?.trim(),
resource_id: values.resource_id?.trim(),
workflow_run_id: values.workflow_run_id?.trim(),
task_run_id: values.task_run_id?.trim(),
trace_json: values.trace_json?.trim(),
});
const result = await recordPromptRunLink(selectedDetail.asset.id, values);
if (result.code !== 0) {
messageApi.error(result.message || "运行证据登记失败");
return;
}
messageApi.success("运行证据已登记");
runLinkForm.resetFields();
await loadDetail(selectedDetail.asset.id);
await refreshAfterMutation(selectedDetail.asset.id);
} catch (error) {
messageApi.error(error instanceof Error ? error.message : "运行证据登记失败,请检查 JSON 格式");
messageApi.error(error instanceof Error ? error.message : "运行证据登记失败,请检查 Trace JSON");
} finally {
setMutatingId(null);
}
@@ -447,7 +452,7 @@ export default function PromptAssetsPage() {
{
title: "评测",
key: "evaluation",
width: 160,
width: 170,
render: (_, item) => (
<Space orientation="vertical" size={0}>
{getEvaluationTag(item.latest_evaluation_status)}
@@ -465,7 +470,7 @@ export default function PromptAssetsPage() {
{
title: "操作",
key: "actions",
width: 260,
width: 270,
render: (_, item) => (
<Space wrap>
<Button icon={<HistoryOutlined />} loading={mutatingId === item.id} size="small" onClick={() => void handleEdit(item)}>
@@ -512,9 +517,9 @@ export default function PromptAssetsPage() {
{
title: "状态",
key: "status",
width: 180,
width: 190,
render: (_, item) => (
<Space>
<Space wrap>
{item.is_current ? <Tag color="blue">当前</Tag> : null}
{item.is_published ? <Tag color="green">生产</Tag> : null}
{getEvaluationTag(item.evaluation_status)}
@@ -546,9 +551,20 @@ export default function PromptAssetsPage() {
{
title: "状态",
key: "status",
width: 130,
width: 120,
render: (_, item) => getEvaluationTag(item.status),
},
{
title: "执行方式",
key: "mode",
width: 150,
render: (_, item) => (
<Space direction="vertical" size={0}>
{getEvaluationModeTag(item.report)}
<Text type="secondary">逐条 {getEvaluationSampleResultCount(item.report)} 条</Text>
</Space>
),
},
{
title: "样本",
key: "samples",
@@ -596,17 +612,17 @@ export default function PromptAssetsPage() {
<div className="review-page-hero-copy">
<span className="landing-badge">Prompt Governance</span>
<Title level={2}>Prompt 资产治理</Title>
<Text type="secondary">生成、审查、条款推荐、法规问答和措辞改写的 Prompt 可以版本化、评测、发布、回滚并绑定运行证据。</Text>
<Text type="secondary">生成、审查、条款推荐、法规问答和措辞改写的 Prompt 可版本化、评测、发布、回滚并绑定运行证据。</Text>
</div>
<Card className="review-side-card" variant="borderless">
<span className="landing-panel-eyebrow">Phase15-E</span>
<span className="landing-panel-eyebrow">Phase15-F</span>
<Title level={4} style={{ margin: 0, color: "inherit" }}>
发布门禁
黄金样本评测门禁
</Title>
<ul className="review-page-tip-list">
<li>发布前必须有通过的评测报告。</li>
<li>回滚会生成一个新的当前版本。</li>
<li>运行证据可绑定到当前生产版本。</li>
<li>已配置共享 AI 时会逐条调用大模型评测黄金样本。</li>
<li>未配置模型时保留本地兜底结果,便于开发环境验证流程。</li>
</ul>
</Card>
</section>
@@ -723,9 +739,9 @@ export default function PromptAssetsPage() {
}
>
<Space orientation="vertical" size="middle" style={{ width: "100%" }}>
<Paragraph type="secondary" style={{ margin: 0 }}>
<Text type="secondary">
当前状态:{getStatusTag(selectedDetail.asset.status)} 最新评测:{getEvaluationTag(selectedDetail.asset.latest_evaluation_status)}
</Paragraph>
</Text>
<Table rowKey="id" size="small" pagination={false} columns={versionColumns} dataSource={selectedDetail.versions} />
<Space orientation="vertical" size="small" style={{ width: "100%" }}>
<Text strong>评测运行历史</Text>
+9
View File
@@ -23,12 +23,16 @@ export interface ClauseRecommendationItem {
clause_key: string;
clause_title: string;
clause_type: string;
recommendation_kind?: "missing_clause" | "industry_practice" | string;
basis_type?: string | null;
industry_context?: string | null;
priority: string;
confidence: string;
insert_position: ClauseRecommendationInsertPosition;
suggested_text: string;
reason: string;
legal_basis: string;
practice_basis?: string | null;
legal_article_refs: ClauseRecommendationLegalArticleRef[];
source_rule: string;
source_types: string[];
@@ -43,8 +47,13 @@ export interface ClauseRecommendationResponse {
total: number;
summary: {
recommended_count: number;
recommendation_keys?: string[];
detected_clause_keys: string[];
missing_clause_keys: string[];
required_missing_clause_keys?: string[];
industry_practice_keys?: string[];
detected_industry_practice_keys?: string[];
industry_practice_count?: number;
source_count: number;
strategy: string;
};
+3
View File
@@ -380,6 +380,9 @@ export type LegalChangeEventItem = {
blocking_count?: number;
by_level?: Record<string, number>;
by_target_type?: Record<string, number>;
by_match_reason?: Record<string, number>;
semantic_match_count?: number;
scan_strategy?: string[];
};
created_by: string | null;
created_at: string | null;
@@ -4108,3 +4108,119 @@
- 剩余边界:
- 本轮尚未部署到 `192.168.3.90`;测试服务器使用该能力前需要执行 Alembic `20260521_0038` 并部署前后端。
- 当前运行仍是同步轻量评测,适合测试服务器资源;后续如果要跑真实大模型逐条黄金样本,可在这张运行表上继续扩展异步队列和 worker。
## 2026-05-21 Phase15-C 行业惯例推荐完整闭环
- 已按《AI合同起草与审查助手-法律文书能力吸收开发方案-2026-05-20.md》回补 Phase15-C 最后一块缺口:行业惯例推荐。
- 后端能力:
- 在既有 `POST /api/v1/clauses/recommend` 中扩展行业惯例推荐,不新增重复入口。
- 新增 `INDUSTRY_PRACTICE_RULES`,覆盖通用服务、技术服务、软件开发、劳动合同、保密协议等首批高频场景。
- 推荐结果新增 `recommendation_kind=industry_practice`、`basis_type=industry_practice_only`、`industry_context`、`practice_basis` 等字段。
- 对“无直接法条,仅为行业惯例”的建议不强行挂法条,结果会明确返回 `no_direct_legal_article` 来源标记。
- `summary` 新增 `industry_practice_keys`、`industry_practice_count`、`detected_industry_practice_keys`、`recommendation_keys`,QA 可直接校验。
- 审计日志 `recommend_missing_clauses` 追加记录推荐 key、行业惯例 key 和行业惯例数量。
- 前端能力:
- `/documents/new/[type]` 生成结果区从“智能条款推荐”升级为“智能条款与行业惯例推荐”。
- 推荐卡片可区分“缺失条款”和“行业惯例”,并展示“无直接法条”、惯例说明、适用场景。
- `/documents/[id]/edit` AI 侧边栏从“检查缺失条款”升级为“检查条款与惯例”。
- 行业惯例建议可像普通推荐条款一样插入正文,并继续生成可追溯合同版本。
- 本轮验证:
- `python -m compileall backend\app\services\clause_recommendation_service.py backend\app\api\clauses.py`
- `$env:PYTHONPATH='backend'; python -m unittest backend.tests.test_clause_recommendation_service backend.tests.test_clauses_api`
- `$env:PYTHONPATH='backend'; python -m unittest backend.tests.test_contracts_api backend.tests.test_clause_recommendation_service backend.tests.test_clauses_api`
- `cd frontend; npm run lint`
- `cd frontend; npm run build`
- `git diff --check`
- Browser 打开 `http://localhost:3000/documents/new`,未登录保护页正常显示,页面标题 `AI合同平台`,控制台错误 0。
- 当前闭环结论:
- Phase15-C 已形成“必备条款缺口检测 -> 行业惯例补充建议 -> 明确无直接法条/惯例来源 -> 页面展示 -> 一键插入合同正文 -> 生成版本 -> 审计留痕”的完整开发闭环。
- 剩余边界:
- 本轮按“先完成开发再统一部署”的原则,没有部署到 `192.168.3.90`。
- 该能力不需要新增数据库迁移;后续统一部署前端和后端后即可进入测试服务器烟测。
## 2026-05-21 Phase15-F 真实黄金样本逐条评测闭环
- 已继续按《AI合同起草与审查助手-法律文书能力吸收开发方案-2026-05-20.md》补齐 Phase15-F 的上线前增强项。
- 本轮把上一版“同步轻量评测”升级为“真实黄金样本逐条评测”:
- 新增 `PromptGoldenSampleEvaluator`,直接加载项目内 Phase13/Phase15 黄金样本清单。
- 覆盖合同/文书生成 10 条、自动审查 8 条、法规问答 10 条、条款推荐 8 条、措辞改写 3 条首批黄金样本。
- 每次评测会生成 `phase15-prompt-evaluation-report.v2` 报告,并写入 `prompt_asset_evaluation_runs.report_json`。
- 报告新增 `evaluation_mode`、`model_invocation`、`sample_results`、逐条 `sample_id / passed / score / issues / matched_expectations`。
- 已配置 `SHARED_AI_PLATFORM_BASE_URL` 和 `SHARED_AI_PLATFORM_API_KEY` 时,会按黄金样本逐条调用共享 AI 平台大模型评测。
- 未配置共享 AI 时默认走 `auto -> local_fallback`,开发环境仍可验证评测记录、发布门禁、前端展示;如将 `PROMPT_GOLDEN_EVAL_MODE=llm`,未配置模型会明确失败,不会假装通过。
- 后端闭环:
- `POST /api/v1/prompt-assets/{asset_id}/evaluations/run` 现在会把真实样本评测报告回填到当前 Prompt 版本、资产最新评测状态和评测运行记录。
- 审计日志 `run_prompt_asset_evaluation` 追加记录评测模式、样本数、通过数、失败数。
- 发布门禁继续复用最新评测状态;未通过逐条样本评测时不可发布。
- 前端闭环:
- 重写并修复 `/prompt-assets` 页面乱码坏句,保留原有 Prompt 创建、版本、发布、回滚、运行证据能力。
- 评测历史新增“执行方式”,可区分“逐条大模型 / 本地兜底 / 模型未配置”。
- 评测历史展示逐条样本结果数量,QA 可以从页面确认是否已经按样本逐条执行。
- 本轮验证:
- `python -m compileall backend\app\services\prompt_golden_sample_evaluator.py backend\app\services\prompt_asset_service.py backend\app\api\prompt_assets.py`
- `$env:PYTHONPATH='backend'; python -m unittest backend.tests.test_prompt_asset_service backend.tests.test_prompt_assets_api`
- `cd frontend; npx eslint src/app/prompt-assets/page.tsx src/types/api.ts`
- `cd frontend; npm run build`
- `git diff --check`
- 当前闭环结论:
- Phase15-F 已形成“加载真实黄金样本 -> 逐条调用大模型或显式本地兜底 -> 写入评测运行记录 -> 回填 Prompt 当前版本 -> 发布门禁读取评测结果 -> 页面展示执行方式和逐条样本数 -> 审计留痕”的完整开发闭环。
- 剩余边界:
- 本轮未部署到 `192.168.3.90`,按你的部署策略先完成开发闭环,后续统一部署时只需部署前后端;本轮没有新增数据库迁移。
- 本地验证中使用了假大模型客户端证明“逐条调用”路径,不消耗真实模型额度;真实环境配置共享 AI 地址和密钥后即可切换为真实逐条大模型评测。
## 2026-05-21 Phase15-G 法规影响语义扫描与后台 worker 闭环
- 已继续按《AI合同起草与审查助手-法律文书能力吸收开发方案-2026-05-20.md》补齐 Phase15-G 的上线前增强项。
- 本轮把原来的“关键词/结构化影响分析 + 手动 run-pending”升级为“语义向量扫描 + 独立后台 worker 定时消费”:
- 新增 `LegalSemanticMatcher`,用本地法律概念词、中文 n-gram、关键词向量做余弦相似度匹配。
- 语义扫描不依赖外部 embedding 服务、不需要联网,适合当前测试服务器资源约束。
- 法规影响扫描现在同时记录 `direct_legal_reference`、`tag_match`、`keyword_match`、`semantic_vector_match`。
- 影响项 `evidence_json.semantic_vector` 会保存相似度分数、共享词、共享法律概念。
- 事件 `impact_summary_json` 新增 `semantic_match_count`、`by_match_reason`、`scan_strategy`,便于 QA 和页面验证。
- 后台 worker:
- 新增 `backend/app/tasks/legal_change_impact_worker.py`。
- 支持 `run_once()` 消费 pending 队列,也支持 daemon 模式按间隔循环消费。
- 新增 `scripts/run-legal-change-impact-worker.ps1`:
- 单次执行:`.\scripts\run-legal-change-impact-worker.ps1 -Once -Limit 10`
- 常驻执行:`.\scripts\run-legal-change-impact-worker.ps1 -Limit 10 -IntervalSeconds 300`
- worker 使用 `system-legal-change-worker` 作为系统执行人,不需要真人手动点击队列消费。
- `docker-compose.server.yml` 新增 `legal-change-impact-worker` 服务,正式部署时可作为独立容器常驻运行。
- 前端闭环:
- `/legal-change-impacts` 影响摘要新增“语义命中”数量。
- 影响项命中原因可见展示“直接引用 / 标签 / 关键词 / 语义向量”。
- 语义向量命中时展示相似度分数和共享法律概念。
- 页面说明补充后台 worker 可定时消费待扫描队列。
- 本轮验证:
- `python -m compileall backend\app\services\legal_semantic_matcher.py backend\app\services\legal_change_impact_service.py backend\app\tasks\legal_change_impact_worker.py backend\app\api\legal_change_impacts.py`
- `$env:PYTHONPATH='backend'; python -m unittest backend.tests.test_legal_change_impact_service backend.tests.test_legal_change_impacts_api`
- `$env:PYTHONPATH='backend'; python -m unittest backend.tests.test_legal_article_service backend.tests.test_legal_articles_api backend.tests.test_legal_change_impact_service backend.tests.test_legal_change_impacts_api`
- `cd frontend; npx eslint src/app/legal-change-impacts/page.tsx src/types/api.ts`
- `cd frontend; npm run build`
- 当前闭环结论:
- Phase15-G 已形成“法条导入/变更入队 -> 后台 worker 定时消费 pending 事件 -> 直接引用/标签/关键词/语义向量扫描 -> 生成影响项和语义证据 -> 页面展示语义命中 -> 审计留痕”的完整开发闭环。
- 剩余边界:
- 本轮未部署到 `192.168.3.90`,没有新增数据库迁移;后续统一部署时可以直接启用 `docker-compose.server.yml` 中的 `legal-change-impact-worker`,或用 `scripts/run-legal-change-impact-worker.ps1` 挂计划任务。
- 当前是本地轻量语义向量,不是远程大模型 embedding;后续若共享 AI 平台提供 embedding 服务,可在保持现有证据字段不变的情况下替换 `LegalSemanticMatcher` 的向量来源。
## 2026-05-21 法律文书能力吸收方案代码对齐检查
- 已按《AI合同起草与审查助手-法律文书能力吸收开发方案-2026-05-20.md》逐项核对 Phase15-A 到 Phase15-H 的代码现状。
- 新增对齐报告:
- `project-docs/AI合同起草与审查助手-法律文书能力吸收开发方案-代码对齐检查-2026-05-21.md`
- 已同步 `project-docs/README.md`,把该对齐检查报告加入当前有效文档入口。
- 本轮核对结论:
- Phase15-A 到 Phase15-H 的主线开发已经在代码层闭环。
- `/clauses/apply` 未单独实现,但条款推荐和改写结果已经通过 `/documents/[id]/edit` 编辑器写入合同正文并生成 `generated_contract_versions`,属于等价闭环。
- `/quality-evals` 未单独实现,当前评测入口集中在 `/prompt-assets`,已能支撑 Prompt 发布前评测门禁;如要做统一质量中心,后续可单独补页面和通用评测表。
- 法规影响分析代码统一使用 `/legal-change-impacts`、`legal_change_events`、`legal_change_impacts` 命名,后续文档应以此为准。
- 本轮验证:
- `$env:PYTHONPATH='backend'; python -m unittest backend.tests.test_legal_document_type_service backend.tests.test_legal_document_types_api backend.tests.test_legal_document_element_schema_service backend.tests.test_legal_document_element_schemas_api backend.tests.test_legal_document_draft_service backend.tests.test_legal_document_drafts_api backend.tests.test_contracts_api backend.tests.test_review_risk_action_service backend.tests.test_review_risk_actions_api backend.tests.test_clause_recommendation_service backend.tests.test_clauses_api backend.tests.test_prompt_asset_service backend.tests.test_prompt_assets_api backend.tests.test_prompt_runtime_service backend.tests.test_legal_change_impact_service backend.tests.test_legal_change_impacts_api backend.tests.test_operation_kpi_service backend.tests.test_operation_kpis_api`
- 结果:`Ran 92 tests in 2.323s`,`OK`。
- `cd frontend; npx eslint "src/app/documents/new/page.tsx" "src/app/documents/new/[type]/page.tsx" "src/app/documents/[id]/edit/page.tsx" "src/app/prompt-assets/page.tsx" "src/app/legal-change-impacts/page.tsx" "src/app/operation-kpis/page.tsx" "src/components/layout/app-topbar.tsx" "src/types/api.ts"`
- 结果:通过。
- `cd frontend; npm run build`
- 结果:通过,Next 构建生成 39 个页面,包含 `/documents/new`、`/documents/new/[type]`、`/documents/[id]/edit`、`/prompt-assets`、`/legal-change-impacts`、`/operation-kpis`、`/sales-followups`。
- 剩余边界:
- 当前仍有 Phase15-C、Phase15-F、Phase15-G 相关未提交改动,需要后续统一整理提交。
- 最新代码尚未部署到 `192.168.3.90` 做真实库烟测。
- 专业富文本库、字段级 diff、独立 `/quality-evals`、全量法规向量化、真实微信或支付宝支付、电子签章和小程序均属于后续增强,不是当前法律文书能力吸收主线阻塞。
@@ -0,0 +1,236 @@
# AI合同起草与审查助手-法律文书能力吸收开发方案代码对齐检查-2026-05-21
## 一、检查范围
本次只检查《AI合同起草与审查助手-法律文书能力吸收开发方案-2026-05-20.md》与当前代码是否对齐,不把“尚未提交”“尚未部署到测试服务器”和“功能没有开发”混为一谈。
检查范围包括:
- 后端模型、迁移、仓储、服务、API。
- 前端页面、菜单入口、服务调用和类型定义。
- 测试覆盖和本轮可复跑验证结果。
- 文档中提出但代码采用等价实现的接口或页面。
## 二、总体结论
当前代码已经覆盖 Phase15-A 到 Phase15-H 的主线开发闭环。也就是说,文书类型入口、动态要素表单、草稿保存恢复、生成接口升级、审查建议采纳、条款推荐与行业惯例、富文本编辑器和 AI 侧边栏、Prompt 治理、黄金样本评测、法规影响扫描、运营 KPI 与商业化增强,均已在代码层形成可运行闭环。
但仍有三类事项需要分开看:
1. 已开发但待提交或待部署验证:当前工作区存在 Phase15-C、Phase15-F、Phase15-G 的未提交改动,且最新 worker 和语义扫描未在 `192.168.3.90` 做远程部署烟测。
2. 文档建议路径与代码等价实现不同:例如文档建议 `/quality-evals` 和 `/clauses/apply`,当前分别落在 `/prompt-assets` 评测历史和合同编辑器版本链中。
3. 后续增强项:专业富文本库、字段级差异对比、独立质量评测中心、全量法规向量化、真实微信或支付宝支付、电子签章、小程序等不属于当前 Phase15 主线硬阻塞。
## 三、逐阶段对齐表
| 阶段 | 文档目标 | 当前代码状态 | 对齐结论 |
|---|---|---|---|
| Phase15-A | 文书类型入口与动态要素表单 | 已有 `legal_document_types`、`legal_document_element_schemas`、`legal_document_drafts`,前端已有 `/documents/new` 和 `/documents/new/[type]` | 已闭环 |
| LAW-DOC-001 | 文书类型字典、分类、行业、场景入口 | 后端有文书类型 API、默认种子和前端入口筛选 | 已闭环 |
| LAW-DOC-002 | 动态要素 schema | 后端支持 schema 配置、发布、校验;前端按 schema 渲染表单 | 已闭环 |
| LAW-DOC-003 | 要素草稿保存、恢复、删除、继续生成 | 后端有草稿 API,前端支持保存恢复,生成时带入 `elements_snapshot` | 已闭环 |
| LAW-DOC-004 | 生成接口升级 | `POST /api/v1/contracts/generate` 支持 `document_type_code`、结构化要素、模板选择、待补字段、法条来源和 HTML 结果 | 已闭环 |
| Phase15-B | 审查建议采纳闭环 | 已有 `review_risk_actions`、风险动作 API、应用建议生成新版本和审计日志 | 已闭环 |
| LAW-REVIEW-001 | 风险建议动作模型 | 每条风险建议有 pending、adopted、modified、ignored 等状态和审计 | 已闭环 |
| LAW-REVIEW-002 | 应用建议生成新版本 | `POST /reviews/{id}/risk-actions/{action_id}/apply` 会尝试定位正文并生成合同版本,无法定位时显式返回人工确认状态 | 已闭环 |
| LAW-REVIEW-003 | 审查报告处理进度 | `/reviews/[id]` 已展示风险处理按钮、状态和进度 | 已闭环 |
| Phase15-C | 智能条款推荐 | `POST /api/v1/clauses/recommend` 支持缺失条款和行业惯例,`POST /api/v1/clauses/rewrite` 支持措辞优化 | 已闭环 |
| LAW-CLAUSE-001 | 缺失条款推荐 | 推荐结果含插入位置、建议文本、理由、依据和行业惯例说明 | 已闭环 |
| LAW-CLAUSE-002 | 措辞优化 | 支持专业化、简化、偏甲方、偏乙方、均衡等改写模式,返回法律边界和谈判建议 | 已闭环 |
| Phase15-D | 富文本编辑器与 AI 侧边栏 | `/documents/[id]/edit` 支持正文编辑、自动保存、版本历史、条款推荐、措辞改写、相关法条和 Chat 入口 | 主线已闭环 |
| Phase15-E | Prompt 资产治理 | `/prompt-assets` 支持 Prompt 类型、版本、评测、发布、下线、回滚和运行证据 | 已闭环 |
| Phase15-F | 黄金样本与质量评测 | 已有 Prompt 评测运行记录、逐条黄金样本报告、发布前门禁和前端评测历史 | 已闭环 |
| Phase15-G | 法规变更影响分析 | 已有法条变更入队、结构化和语义影响扫描、影响项、页面、后台 worker 和 Docker 服务 | 已闭环 |
| Phase15-H | 运营 KPI 与商业化增强 | `/operation-kpis` 已展示用量、文书生成、审查、Chat、模板、条款采纳、试用和转化指标;计费、发票状态、销售跟进已有业务闭环 | 主线已闭环 |
## 四、代码证据
### 4.1 后端 API
- 文书类型:`backend/app/api/legal_document_types.py`
- 文书草稿:`backend/app/api/legal_document_drafts.py`
- 合同生成和版本:`backend/app/api/contracts.py`
- 审查建议动作:`backend/app/api/reviews.py`
- 条款推荐和改写:`backend/app/api/clauses.py`
- Prompt 治理:`backend/app/api/prompt_assets.py`
- 法规影响分析:`backend/app/api/legal_change_impacts.py`
- 运营 KPI:`backend/app/api/operation_kpis.py`
- 销售跟进:`backend/app/api/sales_followups.py`
### 4.2 后端核心模型
- `LegalDocumentType`
- `LegalDocumentElementSchema`
- `LegalDocumentDraft`
- `GeneratedContract`
- `GeneratedContractVersion`
- `ReviewRiskAction`
- `PromptAsset`
- `PromptAssetVersion`
- `PromptAssetRunLink`
- `PromptAssetEvaluationRun`
- `LegalChangeEvent`
- `LegalChangeImpact`
- `ContractLifecycleTask`
- `CommercialOrder`
- `CommercialInvoice`
- `SalesFollowupEvent`
### 4.3 前端页面
- 文书入口:`frontend/src/app/documents/new/page.tsx`
- 动态要素表单:`frontend/src/app/documents/new/[type]/page.tsx`
- 合同编辑器和 AI 侧边栏:`frontend/src/app/documents/[id]/edit/page.tsx`
- 审查建议处理:`frontend/src/app/reviews/[id]/page.tsx`
- Prompt 治理:`frontend/src/app/prompt-assets/page.tsx`
- 法规影响分析:`frontend/src/app/legal-change-impacts/page.tsx`
- 运营 KPI:`frontend/src/app/operation-kpis/page.tsx`
- 销售跟进:`frontend/src/app/sales-followups/page.tsx`
### 4.4 后台任务
- 法规影响后台 worker:`backend/app/tasks/legal_change_impact_worker.py`
- 本地执行脚本:`scripts/run-legal-change-impact-worker.ps1`
- 测试服务器编排:`docker-compose.server.yml` 中已增加 `legal-change-impact-worker`
## 五、等价实现说明
### 5.1 `/clauses/apply`
文档建议新增 `POST /api/v1/clauses/apply`。当前代码没有单独做这个接口,而是通过合同编辑器完成应用:
1. 前端调用 `/clauses/recommend` 或 `/clauses/rewrite`。
2. 用户在 `/documents/[id]/edit` 中选择插入或替换正文。
3. 前端调用合同版本接口保存新正文。
4. 后端生成新的 `generated_contract_versions` 记录。
这个实现已经满足“推荐或改写结果应用到合同正文并形成新版本”的验收口径。后续只有在需要第三方系统直接调用条款采纳动作时,才有必要再补独立 `/clauses/apply`。
### 5.2 `/quality-evals`
文档建议新增独立 `/quality-evals` 页面。当前代码没有这个独立路由,质量评测落在 `/prompt-assets` 中:
- Prompt 详情页可以运行评测。
- 评测结果回填当前版本。
- 发布按钮受评测状态约束。
- 评测运行历史可以追溯。
这对“Prompt 发布前黄金样本评测门禁”已经闭环。但如果要做跨生成、审查、Chat、条款推荐、法规影响的统一质量中心,仍建议后续补独立 `/quality-evals`。
### 5.3 `regulation-changes`
文档用的是“法规变更影响分析”的产品名。当前代码实际路由统一命名为 `/legal-change-impacts`,表名为 `legal_change_events` 和 `legal_change_impacts`。命名不同,但职责一致,建议后续文档统一采用当前代码命名,减少误解。
### 5.4 `clause_recommendations` 持久化表
文档建议可以沉淀 `clause_recommendations`。当前代码采用实时推荐加审计日志,再通过合同版本链保存采纳结果。这个选择更轻,不会制造一张只保存临时建议的重复表。
如果未来要统计“每条推荐从展示到采纳、忽略、修改”的完整漏斗,可以再补持久化推荐动作表。
## 六、仍未彻底闭环的事项
### 6.1 待提交和待部署验证
当前工作区存在未提交改动,主要集中在:
- Phase15-C 行业惯例推荐。
- Phase15-F 真实黄金样本逐条评测。
- Phase15-G 法规影响语义扫描和后台 worker。
这些代码本地测试已通过,但尚未提交远程 Git,也未在 `192.168.3.90` 重新部署后做真实库烟测。
### 6.2 独立质量评测中心
当前评测能力已经在 Prompt 治理内闭环,但没有独立 `/quality-evals` 页面,也没有通用 `ai_quality_eval_runs/items` 表承接所有质量评测类型。
这不是 Prompt 发布门禁阻塞项,但属于文档和产品形态的一个剩余增强点。
### 6.3 专业级富文本增强
当前 `/documents/[id]/edit` 已经支持编辑、自动保存、版本历史、推荐插入、改写替换和 AI 侧边栏。尚未做的增强包括:
- 专业富文本编辑器库。
- 段落级或字段级差异对比。
- Word 样式还原和导出增强。
- 多人协同编辑。
这些是体验增强,不影响当前 Phase15-D 主线闭环。
### 6.4 全量法规向量化
当前法规影响语义扫描使用本地轻量语义匹配,不等同于把全量法规库一次性上传到 shared-ai-platform 做向量检索。
这不影响现有法条全文检索、审查报告法条引用和法规影响扫描的业务闭环,但如果目标升级为“全国全量法规语义检索”,需要另开批处理和容量治理阶段。
### 6.5 真实在线支付和完整发票流程
Phase15-H 的运营 KPI 和商业化基础已经完成,但真实微信或支付宝支付、商户回调验签、公网 HTTPS 回调域名、完整发票抬头税号和开票文件,仍依赖外部商户资料和公网条件。
这属于商业化生产增强,不是《法律文书能力吸收开发方案》的核心法律文书主线阻塞。
### 6.6 P3 外部集成
微信小程序、电子签章、第三方签署平台、重型工作流引擎等在原方案中明确为延后或不照搬方向。当前未实现是符合方案边界的,不应算作跑偏。
## 七、本轮验证结果
### 7.1 后端回归
命令:
```powershell
$env:PYTHONPATH='backend'; python -m unittest backend.tests.test_legal_document_type_service backend.tests.test_legal_document_types_api backend.tests.test_legal_document_element_schema_service backend.tests.test_legal_document_element_schemas_api backend.tests.test_legal_document_draft_service backend.tests.test_legal_document_drafts_api backend.tests.test_contracts_api backend.tests.test_review_risk_action_service backend.tests.test_review_risk_actions_api backend.tests.test_clause_recommendation_service backend.tests.test_clauses_api backend.tests.test_prompt_asset_service backend.tests.test_prompt_assets_api backend.tests.test_prompt_runtime_service backend.tests.test_legal_change_impact_service backend.tests.test_legal_change_impacts_api backend.tests.test_operation_kpi_service backend.tests.test_operation_kpis_api
```
结果:
```text
Ran 92 tests in 2.323s
OK
```
### 7.2 前端重点页面静态检查
命令:
```powershell
npx eslint "src/app/documents/new/page.tsx" "src/app/documents/new/[type]/page.tsx" "src/app/documents/[id]/edit/page.tsx" "src/app/prompt-assets/page.tsx" "src/app/legal-change-impacts/page.tsx" "src/app/operation-kpis/page.tsx" "src/components/layout/app-topbar.tsx" "src/types/api.ts"
```
结果:通过,无错误输出。
### 7.3 前端生产构建
命令:
```powershell
npm run build
```
结果:
```text
Compiled successfully
Generating static pages (39/39)
```
构建路由中已包含:
- `/documents/new`
- `/documents/new/[type]`
- `/documents/[id]/edit`
- `/prompt-assets`
- `/legal-change-impacts`
- `/operation-kpis`
- `/sales-followups`
## 八、建议下一步
为了让文档和代码彻底闭合,建议下一步不要继续堆新功能,先按这个顺序收口:
1. 整理并提交当前 Phase15-C、Phase15-F、Phase15-G 的代码和文档改动。
2. 如果本阶段要求“文档路径和产品入口完全一致”,补独立 `/quality-evals` 质量评测中心,把 `/prompt-assets` 中的评测历史聚合过去。
3. 统一文档命名,把法规影响分析的接口名称固定为 `/legal-change-impacts`,避免后续误认为还缺 `/regulation-changes`。
4. 等决定部署时,在 `192.168.3.90` 执行最新 Alembic,并启用 `legal-change-impact-worker` 服务做真实库烟测。
按当前代码状态判断,《法律文书能力吸收开发方案》的主线开发没有跑偏,已经基本完成。真正剩下的是提交、部署验证、独立质量中心和若干生产级增强。
+1
View File
@@ -21,6 +21,7 @@
- `AI合同起草与审查助手-正式生产部署演练与上线检查清单-2026-05-16.md`
- `AI合同起草与审查助手-Phase14法条全文级知识库实施方案-2026-05-16.md`
- `AI合同起草与审查助手-法律文书能力吸收开发方案-2026-05-20.md`
- `AI合同起草与审查助手-法律文书能力吸收开发方案-代码对齐检查-2026-05-21.md`
## 当前仍保留的专项资料
@@ -0,0 +1,29 @@
param(
[switch]$Once,
[int]$Limit = 10,
[int]$IntervalSeconds = 300,
[int]$MaxLoops = 0
)
$ErrorActionPreference = "Stop"
$repoRoot = Split-Path -Parent $PSScriptRoot
$backendPath = Join-Path $repoRoot "backend"
$env:PYTHONPATH = $backendPath
$argsList = @("-m", "app.tasks.legal_change_impact_worker", "--limit", "$Limit")
if ($Once) {
$argsList += "--once"
} else {
$argsList += @("--interval-seconds", "$IntervalSeconds")
}
if ($MaxLoops -gt 0) {
$argsList += @("--max-loops", "$MaxLoops")
}
Push-Location $repoRoot
try {
python @argsList
} finally {
Pop-Location
}