yuqei/ai-contract-platform
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

backend: harden legal article references

Browse Source
This commit is contained in:
yuqei
2026-05-25 18:25:12 +08:00
parent b8432dce30
commit 8a126a3808
3 changed files with 356 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/repositories/legal_article_repository.py
+39 -1
View File
@@ -1,6 +1,6 @@
from typing import Optional
from sqlalchemy import func, or_
from sqlalchemy import case, func, or_
from sqlalchemy.orm import Session
from app.models.legal_article import LegalArticle
@@ -50,6 +50,44 @@ class LegalArticleRepository:
.first()
)
def get_published_by_source_article(self, tenant_id: str, source_id: str, article_no: str) -> Optional[LegalArticle]:
return (
self.db.query(LegalArticle)
.filter(LegalArticle.tenant_id == tenant_id)
.filter(LegalArticle.source_id == source_id)
.filter(LegalArticle.article_no == article_no)
.filter(LegalArticle.published_status == "published")
.filter(LegalArticle.index_status == "indexed")
.first()
)
def get_published_by_source_title_article(
self,
tenant_id: str,
source_titles: list[str],
article_no: str,
) -> Optional[LegalArticle]:
normalized_titles = self._normalize_search_terms(source_titles)
if not normalized_titles or not article_no:
return None
title_priority = case(
*[
(LegalArticle.source_title == title, index)
for index, title in enumerate(normalized_titles)
],
else_=len(normalized_titles),
)
return (
self.db.query(LegalArticle)
.filter(LegalArticle.tenant_id == tenant_id)
.filter(LegalArticle.article_no == article_no)
.filter(LegalArticle.source_title.in_(normalized_titles))
.filter(LegalArticle.published_status == "published")
.filter(LegalArticle.index_status == "indexed")
.order_by(title_priority.asc(), func.length(LegalArticle.source_title).asc(), LegalArticle.source_id.asc())
.first()
)
def get_vectorization_summary(
self,
tenant_id: str,
backend/app/services/review_service.py
+116 -34
View File
@@ -1737,37 +1737,62 @@ class ReviewService:
)
return enriched
def _attach_legal_article_refs(self, risk_items: List[Dict[str, Any]], tenant_id: str) -> List[Dict[str, Any]]:
fallback_candidates = self._load_legal_article_reference_candidates(tenant_id)
def _attach_legal_article_refs(
self,
risk_items: List[Dict[str, Any]],
tenant_id: str,
source_text: Optional[str] = None,
) -> List[Dict[str, Any]]:
enriched: List[Dict[str, Any]] = []
source_rule_candidates = self._search_rule_target_legal_article_candidates((source_text or "").lower(), tenant_id) if source_text else []
for item in risk_items:
if item.get("legal_article_refs"):
enriched.append(item)
continue
candidates = self._build_legal_article_candidates_for_risk(item, tenant_id, fallback_candidates)
candidates = self._build_legal_article_candidates_for_risk(item, tenant_id, source_rule_candidates)
if not candidates:
enriched.append(item)
continue
by_source_article = {(article.source_id, article.article_no): article for article in candidates}
refs = self._resolve_legal_article_refs_for_risk(item, candidates, by_source_article)
if (
not refs
and source_rule_candidates
and self._risk_item_should_inherit_source_rule_refs(item)
):
refs = self._format_inherited_source_rule_legal_article_refs(source_rule_candidates)
enriched.append({**item, "legal_article_refs": refs})
return enriched
def _load_legal_article_reference_candidates(self, tenant_id: str):
try:
return self.legal_article_repository.list_reference_candidates(tenant_id)
except Exception:
return []
def _build_legal_article_candidates_for_risk(
self,
item: Dict[str, Any],
tenant_id: str,
fallback_candidates: List[Any],
source_rule_candidates: Optional[List[Any]] = None,
) -> List[Any]:
targeted_candidates = self._search_legal_article_candidates_for_risk(item, tenant_id)
return self._merge_legal_article_candidates(targeted_candidates, fallback_candidates)
if not targeted_candidates and self._risk_item_should_inherit_source_rule_refs(item):
targeted_candidates = source_rule_candidates or []
return self._merge_legal_article_candidates(targeted_candidates)
@staticmethod
def _risk_item_should_inherit_source_rule_refs(item: Dict[str, Any]) -> bool:
level = str(item.get("level") or item.get("severity") or "").strip().lower()
return level in {"high", "medium"}
def _format_inherited_source_rule_legal_article_refs(self, candidates: List[Any]) -> List[Dict[str, Any]]:
refs: List[Dict[str, Any]] = []
seen: set[str] = set()
for article in candidates:
article_id = str(getattr(article, "id", "") or "")
if not article_id or article_id in seen:
continue
seen.add(article_id)
refs.append(self._format_legal_article_ref(article, "原合同文本命中高风险规则关键词"))
if len(refs) >= 3:
break
return refs
def _search_legal_article_candidates_for_risk(self, item: Dict[str, Any], tenant_id: str) -> List[Any]:
if not hasattr(self.legal_article_repository, "search_published_for_runtime"):
@@ -1777,6 +1802,8 @@ class ReviewService:
candidates: List[Any] = []
try:
candidates.extend(self._search_rule_target_legal_article_candidates(risk_text, tenant_id))
if candidates:
return self._merge_legal_article_candidates(candidates)
if source_titles or article_nos:
candidates.extend(
self.legal_article_repository.search_published_for_runtime(
@@ -1786,7 +1813,11 @@ class ReviewService:
limit=30,
)
)
if len(candidates) < 6 and keywords:
if len(candidates) < 6 and keywords and self._has_high_signal_legal_reference_terms(
source_titles,
article_nos,
keywords,
):
candidates.extend(
self.legal_article_repository.search_published_for_runtime(
tenant_id,
@@ -1794,19 +1825,20 @@ class ReviewService:
limit=30,
)
)
if len(candidates) < 6 and keywords:
candidates.extend(
self.legal_article_repository.search_published_for_runtime(
tenant_id,
keywords=keywords[:4],
include_article_text=True,
limit=20,
)
)
except Exception:
return []
return self._merge_legal_article_candidates(candidates)
@staticmethod
def _has_high_signal_legal_reference_terms(
source_titles: List[str],
article_nos: List[str],
keywords: List[str],
) -> bool:
if source_titles or article_nos:
return True
return any(keyword and ("法" in keyword or "law" in keyword.lower()) for keyword in keywords[:6])
def _search_rule_target_legal_article_candidates(self, risk_text: str, tenant_id: str) -> List[Any]:
candidates: List[Any] = []
for rule in LEGAL_ARTICLE_REFERENCE_RULES:
@@ -1814,19 +1846,31 @@ class ReviewService:
if not matched_trigger:
continue
for source_id, article_no in rule["targets"]:
source_titles = LEGAL_SOURCE_ALIAS_MAP.get(source_id, [])
if not source_titles or not article_no:
if not source_id or not article_no:
continue
candidates.extend(
self.legal_article_repository.search_published_for_runtime(
tenant_id,
source_titles=source_titles,
article_nos=[article_no],
limit=10,
)
)
article = self._get_published_legal_article_by_source_article(tenant_id, source_id, article_no)
if article is None:
article = self._get_published_legal_article_by_source_title_article(tenant_id, source_id, article_no)
if article is not None:
candidates.append(article)
return self._merge_legal_article_candidates(candidates)
def _get_published_legal_article_by_source_article(self, tenant_id: str, source_id: str, article_no: str) -> Optional[Any]:
if hasattr(self.legal_article_repository, "get_published_by_source_article"):
return self.legal_article_repository.get_published_by_source_article(tenant_id, source_id, article_no)
article = self.legal_article_repository.get_by_source_article(tenant_id, source_id, article_no)
if article is None:
return None
if getattr(article, "published_status", None) != "published" or getattr(article, "index_status", None) != "indexed":
return None
return article
def _get_published_legal_article_by_source_title_article(self, tenant_id: str, source_id: str, article_no: str) -> Optional[Any]:
source_titles = LEGAL_SOURCE_ALIAS_MAP.get(source_id, [])
if not source_titles or not hasattr(self.legal_article_repository, "get_published_by_source_title_article"):
return None
return self.legal_article_repository.get_published_by_source_title_article(tenant_id, source_titles, article_no)
@staticmethod
def _merge_legal_article_candidates(*groups: List[Any]) -> List[Any]:
merged: List[Any] = []
@@ -1978,7 +2022,7 @@ class ReviewService:
def _build_report_payload(self, review: ReviewTask, result: Dict[str, Any]) -> Dict[str, Any]:
enriched = self._enrich_review_result(review, result)
risk_items = self._attach_legal_article_refs(enriched["risk_items"], review.tenant_id)
risk_items = self._attach_legal_article_refs(enriched["risk_items"], review.tenant_id, review.input_text)
legal_article_refs = self._collect_unique_legal_article_refs(risk_items)
report_download_metadata = self._build_report_download_metadata(legal_article_refs)
rendered_report = enriched.get("rendered_report")
@@ -2078,9 +2122,9 @@ class ReviewService:
fallback_prompt_version="legal-review.v1",
)
setattr(review, "_prompt_runtime", prompt_runtime)
result = self._get_legal_review_execution_gateway().run_review(review)
result = self._run_review_with_optional_local_fallback(review)
result = self._enrich_review_result(review, result, getattr(review, "_extraction_metadata", None))
risk_items = self._attach_legal_article_refs(result["risk_items"], review.tenant_id)
risk_items = self._attach_legal_article_refs(result["risk_items"], review.tenant_id, review.input_text)
result["risk_items"] = risk_items
risk_summary = result["risk_summary"]
confidence_summary = result.get("confidence_summary")
@@ -2130,6 +2174,44 @@ class ReviewService:
return self.repository.update(review)
def _run_review_with_optional_local_fallback(self, review: ReviewTask) -> Dict[str, Any]:
try:
return self._get_legal_review_execution_gateway().run_review(review)
except Exception as exc:
if not settings.shared_ai_platform_legal_review_fallback_to_local:
raise
return self._build_local_review_fallback_result(review, exc)
def _build_local_review_fallback_result(self, review: ReviewTask, exc: Exception) -> Dict[str, Any]:
risk_items = self._build_risk_items(review)
risk_summary = self._build_risk_summary(risk_items)
return {
"basic_info": {
"contract_type": review.contract_type,
"contract_title": review.title,
"party_a": DEFAULT_PARTY_A,
"party_b": DEFAULT_PARTY_B,
"amount": DEFAULT_AMOUNT,
"term": DEFAULT_TERM,
},
"risk_summary": risk_summary,
"clauses": self._build_clauses(review),
"risk_items": risk_items,
"review_opinion": self._build_review_opinion(risk_summary),
"recommendation": DEFAULT_RECOMMENDATION,
"confidence_summary": {
"provider": "local-rule-fallback",
"model": None,
"prompt_version": "legal-review.local-fallback",
"knowledge_source": "local-review-rules",
"analysis_mode": "local_rule_review",
"provider_mode": "local_fallback",
"router_status": "shared_ai_failed",
"fallback_reason": f"{type(exc).__name__}: {str(exc)[:240]}",
"template_source": "local_default",
},
}
def _build_review_prompt_preferred_keys(self, contract_type: Optional[str]) -> List[str]:
normalized_contract_type = self._normalize_review_prompt_contract_type(contract_type)
preferred_keys: List[str] = []
backend/tests/test_review_service.py
+201 -1
View File
@@ -8,7 +8,8 @@ from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.cidfonts import UnicodeCIDFont
from reportlab.pdfgen import canvas
from app.services.review_service import LEGAL_ARTICLE_REFERENCE_RULES, LEGAL_SOURCE_ALIAS_MAP, ReviewService
from app.core.config import settings
from app.services.review_service import DEFAULT_RISK_RULES, LEGAL_ARTICLE_REFERENCE_RULES, LEGAL_SOURCE_ALIAS_MAP, ReviewService
class StubRepository:
@@ -125,6 +126,10 @@ class StubLegalArticle:
class StubLegalArticleRepository:
def __init__(self):
self.reference_candidate_count = None
self.list_reference_candidates_calls = []
self.search_calls = []
self.get_published_by_source_article_calls = []
self.get_published_by_source_title_article_calls = []
self.items = [
StubLegalArticle("lart-civil-470", "LAW-CIVIL-CODE-2020", "中华人民共和国民法典", "第四百七十条", ["合同条款", "合同内容"], "合同的内容由当事人约定。"),
StubLegalArticle("lart-civil-496", "LAW-CIVIL-CODE-2020", "中华人民共和国民法典", "第四百九十六条", ["格式条款", "提示说明"], "格式条款是当事人为了重复使用而预先拟定的条款。"),
@@ -136,6 +141,7 @@ class StubLegalArticleRepository:
]
def list_reference_candidates(self, tenant_id, limit=500):
self.list_reference_candidates_calls.append({"tenant_id": tenant_id, "limit": limit})
if self.reference_candidate_count is not None:
return self.items[: self.reference_candidate_count]
return self.items[:limit]
@@ -150,6 +156,15 @@ class StubLegalArticleRepository:
include_article_text=False,
limit=20,
):
self.search_calls.append(
{
"keywords": list(keywords or []),
"source_titles": list(source_titles or []),
"article_nos": list(article_nos or []),
"include_article_text": include_article_text,
"limit": limit,
}
)
results = []
for item in self.items:
if source_titles and not any(term in item.source_title for term in source_titles):
@@ -173,6 +188,26 @@ class StubLegalArticleRepository:
results.append(item)
return results[:limit]
def get_published_by_source_article(self, tenant_id, source_id, article_no):
self.get_published_by_source_article_calls.append(
{"tenant_id": tenant_id, "source_id": source_id, "article_no": article_no}
)
for item in self.items:
if item.source_id == source_id and item.article_no == article_no:
return item
return None
def get_published_by_source_title_article(self, tenant_id, source_titles, article_no):
self.get_published_by_source_title_article_calls.append(
{"tenant_id": tenant_id, "source_titles": list(source_titles or []), "article_no": article_no}
)
source_titles = list(source_titles or [])
for source_title in source_titles:
for item in self.items:
if item.source_title == source_title and item.article_no == article_no:
return item
return None
class StubLegalReviewGateway:
def run_review(self, review):
@@ -272,8 +307,14 @@ class StubLegalReviewGateway:
}
class FailingLegalReviewGateway:
def run_review(self, review):
raise TimeoutError("shared ai legal review timed out")
class ReviewServiceTestCase(unittest.TestCase):
def setUp(self):
self.original_legal_review_fallback_to_local = settings.shared_ai_platform_legal_review_fallback_to_local
self.service = ReviewService(db=None)
self.service.repository = StubRepository()
self.service.review_rule_repository = StubRuleRepository()
@@ -285,6 +326,9 @@ class ReviewServiceTestCase(unittest.TestCase):
self.same_tenant_other_user = {"id": "user-2", "tenant_id": "tenant-demo-001", "role": "admin"}
self.other_user = {"id": "user-3", "tenant_id": "tenant-other", "role": "admin"}
def tearDown(self):
settings.shared_ai_platform_legal_review_fallback_to_local = self.original_legal_review_fallback_to_local
def test_review_prompt_preferred_keys_route_to_specialized_domains(self):
expectations = {
"investment_agreement": "contract_review_investment_finance",
@@ -413,6 +457,125 @@ class ReviewServiceTestCase(unittest.TestCase):
self.assertEqual(enriched[0]["legal_article_refs"][0]["source_id"], "FLK-CIVIL-CODE-FULL")
self.assertEqual(enriched[0]["legal_article_refs"][0]["article_no"], article_no)
def test_legal_article_refs_skip_keyword_fallback_when_rule_targets_hit(self):
trigger = LEGAL_ARTICLE_REFERENCE_RULES[0]["triggers"][0]
risk_item = {
"title": f"{trigger} rule target",
"level": "medium",
"description": f"{trigger} with additional business keywords",
"suggestion": f"review {trigger} cap and compensation range",
"legal_basis": "",
"evidence": {
"quote": trigger,
"clause_title": "liability",
"matched_keywords": [trigger, "compensation", "contract"],
},
}
enriched = self.service._attach_legal_article_refs([risk_item], self.current_user["tenant_id"])
self.assertEqual(enriched[0]["legal_article_refs"][0]["id"], "lart-civil-585")
self.assertTrue(self.service.legal_article_repository.get_published_by_source_article_calls)
keyword_calls = [
call
for call in self.service.legal_article_repository.search_calls
if call["keywords"] or call["include_article_text"]
]
self.assertEqual(keyword_calls, [])
self.assertEqual(self.service.legal_article_repository.search_calls, [])
self.assertEqual(self.service.legal_article_repository.list_reference_candidates_calls, [])
def test_legal_article_refs_skip_low_signal_keyword_and_fulltext_scan(self):
self.service.legal_article_repository.reference_candidate_count = 0
risk_item = {
"title": "commercial term needs review",
"level": "medium",
"description": "renewal payment liability notice evidence commercial risk",
"suggestion": "add clearer business terms",
"legal_basis": "",
"evidence": {
"quote": "renewal payment liability",
"clause_title": "term",
"matched_keywords": ["renewal", "payment", "liability", "notice"],
},
}
enriched = self.service._attach_legal_article_refs([risk_item], self.current_user["tenant_id"])
self.assertEqual(enriched[0].get("legal_article_refs", []), [])
self.assertEqual(self.service.legal_article_repository.search_calls, [])
self.assertEqual(self.service.legal_article_repository.list_reference_candidates_calls, [])
def test_legal_article_refs_do_not_load_global_reference_candidates(self):
trigger = LEGAL_ARTICLE_REFERENCE_RULES[0]["triggers"][0]
risk_item = {
"title": f"{trigger} rule target",
"level": "high",
"description": trigger,
"suggestion": "review target law articles",
"legal_basis": "",
"evidence": {"quote": trigger, "clause_title": "liability"},
}
enriched = self.service._attach_legal_article_refs([risk_item], self.current_user["tenant_id"])
self.assertTrue(enriched[0]["legal_article_refs"])
self.assertTrue(self.service.legal_article_repository.get_published_by_source_article_calls)
self.assertEqual(self.service.legal_article_repository.search_calls, [])
self.assertEqual(self.service.legal_article_repository.list_reference_candidates_calls, [])
def test_high_risk_items_inherit_source_rule_refs_when_remote_item_lacks_trigger(self):
trigger = LEGAL_ARTICLE_REFERENCE_RULES[0]["triggers"][0]
risk_item = {
"title": "liability cap requires review",
"level": "high",
"description": "Remote risk item is normalized in English and lacks the original trigger.",
"suggestion": "Revise the liability clause.",
"legal_basis": "",
"evidence": {"quote": "50% of total contract price", "clause_title": "liability"},
}
enriched = self.service._attach_legal_article_refs(
[risk_item],
self.current_user["tenant_id"],
source_text=f"合同包含{trigger}等高风险条款。",
)
self.assertEqual(enriched[0]["legal_article_refs"][0]["id"], "lart-civil-585")
self.assertTrue(self.service.legal_article_repository.get_published_by_source_article_calls)
self.assertEqual(self.service.legal_article_repository.search_calls, [])
self.assertEqual(self.service.legal_article_repository.list_reference_candidates_calls, [])
def test_rule_targets_fallback_to_source_title_when_canonical_source_id_is_missing(self):
source_id, article_no = LEGAL_ARTICLE_REFERENCE_RULES[0]["targets"][0]
trigger = LEGAL_ARTICLE_REFERENCE_RULES[0]["triggers"][0]
source_title = LEGAL_SOURCE_ALIAS_MAP[source_id][-1]
self.service.legal_article_repository.items = [
StubLegalArticle(
"lart-civil-585-flk",
"FLK-civil-code",
source_title,
article_no,
["违约金"],
"当事人可以约定违约金。",
)
]
risk_item = {
"title": trigger,
"level": "high",
"description": trigger,
"suggestion": "review damages",
"legal_basis": "",
"evidence": {"quote": trigger, "clause_title": "liability"},
}
enriched = self.service._attach_legal_article_refs([risk_item], self.current_user["tenant_id"])
self.assertEqual(enriched[0]["legal_article_refs"][0]["id"], "lart-civil-585-flk")
self.assertTrue(self.service.legal_article_repository.get_published_by_source_article_calls)
self.assertTrue(self.service.legal_article_repository.get_published_by_source_title_article_calls)
self.assertEqual(self.service.legal_article_repository.search_calls, [])
def test_report_metadata_and_content_match_review(self):
created = self.service.create_review_task(
tenant_id=self.current_user["tenant_id"],
@@ -846,6 +1009,43 @@ class ReviewServiceTestCase(unittest.TestCase):
self.assertEqual(result["workflow_state"]["source_document"]["source_kind"], "scanned_pdf")
self.assertEqual(result["workflow_state"]["source_text_length"], result["source_document"]["text_length"])
def test_finalize_review_task_uses_local_fallback_when_shared_ai_fails(self):
settings.shared_ai_platform_legal_review_fallback_to_local = True
self.service._get_legal_review_execution_gateway = lambda: FailingLegalReviewGateway()
sample_text = "\n".join(rule["keyword"] for rule in DEFAULT_RISK_RULES[:3])
created = self.service.create_review_task(
tenant_id=self.current_user["tenant_id"],
user_id=self.current_user["id"],
review_as="party_b",
contract_type="service",
text=sample_text,
file_name="high-risk.txt",
)
result = self.service.get_review_result(self.current_user, created["id"])
assert result is not None
self.assertEqual(created["status"], "completed")
self.assertEqual(result["risk_summary"]["overall"], "high")
self.assertEqual(result["confidence_summary"]["provider"], "local-rule-fallback")
self.assertEqual(result["confidence_summary"]["provider_mode"], "local_fallback")
self.assertGreater(result["legal_article_ref_count"], 0)
self.assertTrue(any(item.get("legal_article_refs") for item in result["risk_items"]))
def test_finalize_review_task_reraises_shared_ai_failure_without_fallback(self):
settings.shared_ai_platform_legal_review_fallback_to_local = False
self.service._get_legal_review_execution_gateway = lambda: FailingLegalReviewGateway()
with self.assertRaises(TimeoutError):
self.service.create_review_task(
tenant_id=self.current_user["tenant_id"],
user_id=self.current_user["id"],
review_as="party_b",
contract_type="service",
text=DEFAULT_RISK_RULES[1]["keyword"],
file_name="high-risk.txt",
)
def test_merge_review_text_prefers_both_sources(self):
merged = self.service.merge_review_text("手工补充", "自动续约")