法律 AI 行业全景：从合同审查到自主法律 Agent 的完整技术栈

import spacy
from spacy.matcher import Matcher
import networkx as nx
from typing import List, Dict, Tuple

class LegalKnowledgeGraph:
    """法律知识图谱构建器
    
    功能：从法律文本中提取条款、法条引用，
    构建法律概念之间的引用关系图。
    """
    
    def __init__(self, model_name: str = "zh_core_web_sm"):
        self.nlp = spacy.load(model_name)
        self.graph = nx.DiGraph()
        self._setup_patterns()
    
    def _setup_patterns(self):
        """设置法律引用匹配模式"""
        self.matcher = Matcher(self.nlp.vocab)
        
        # 匹配 "第X条" 模式
        self.matcher.add("ARTICLE_REF", [
            [{"LOWER": "第"}, {"POS": "NUM"}, {"LOWER": "条"}]
        ])
        
        # 匹配 "根据XXX法第X条" 模式
        self.matcher.add("LEGAL_CITATION", [
            [{"LOWER": "根据"}],
            [{"POS": "PROPN"}, {"LOWER": "法"}],
            [{"LOWER": "第"}, {"POS": "NUM"}, {"LOWER": "条"}]
        ])
    
    def parse_legal_document(self, text: str) -> Dict:
        """解析法律文档，提取结构信息"""
        doc = self.nlp(text)
        
        # 提取条款
        articles = self._extract_articles(doc)
        
        # 提取引用关系
        citations = self._extract_citations(doc)
        
        # 构建知识图谱
        self._build_graph(articles, citations)
        
        return {
            "articles": articles,
            "citations": citations,
            "graph_nodes": self.graph.number_of_nodes(),
            "graph_edges": self.graph.number_of_edges()
        }
    
    def _extract_articles(self, doc) -> List[Dict]:
        """提取法律条款"""
        articles = []
        for match_id, start, end in self.matcher(doc, "ARTICLE_REF"):
            articles.append({
                "text": doc[start:end].text,
                "start": start,
                "end": end
            })
        return articles
    
    def _extract_citations(self, doc) -> List[Tuple[str, str]]:
        """提取法条引用关系"""
        citations = []
        for match_id, start, end in self.matcher(doc, "LEGAL_CITATION"):
            cited_law = doc[start+1:start+2].text + "法"
            cited_article = doc[start+2:end].text
            citations.append((cited_law, cited_article))
        return citations
    
    def _build_graph(self, articles, citations):
        """构建法律知识图谱"""
        for article in articles:
            self.graph.add_node(article["text"], type="article")
        
        for source, target in citations:
            self.graph.add_edge(source, target, relation="cites")

# 使用示例
kg = LegalKnowledgeGraph()
legal_text = """
根据民法典第一千零二十四条，民事主体享有名誉权。
任何组织或者个人不得以侮辱、诽谤等方式侵害他人名誉权。
根据刑法第二百四十六条，以暴力或者其他方法公然侮辱他人
或者捏造事实诽谤他人，情节严重的，处三年以下有期徒刑。
"""
result = kg.parse_legal_document(legal_text)
print(f"提取 {len(result['articles'])} 个条款，"
      f"{len(result['citations'])} 个引用关系")
print(f"知识图谱：{result['graph_nodes']} 个节点，"
      f"{result['graph_edges']} 条边")

from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
import json

class RiskLevel(Enum):
    LOW = "低"
    MEDIUM = "中"
    HIGH = "高"
    CRITICAL = "严重"

@dataclass
class ContractRisk:
    """合同风险项"""
    clause_text: str
    risk_type: str
    risk_level: RiskLevel
    explanation: str
    legal_basis: str  # 法律依据
    suggestion: str   # 修改建议

class ContractReviewPipeline:
    """合同审查 Pipeline
    
    结合规则引擎和 LLM 模型进行合同风险分析。
    规则引擎处理明确的法律红线，LLM 处理
    语义理解和上下文推理。
    """
    
    def __init__(self, llm_client=None):
        self.llm = llm_client
        self.risk_rules = self._load_risk_rules()
    
    def _load_risk_rules(self) -> List[dict]:
        """加载风险审查规则"""
        return [
            {
                "name": "无限责任条款",
                "keywords": ["无限责任", "承担全部", "连带责任"],
                "risk_level": RiskLevel.CRITICAL,
                "explanation": "合同中约定无限责任可能导致极端风险",
                "legal_basis": "《民法典》第五百七十七条",
                "suggestion": "建议增加责任上限条款，约定赔偿上限"
            },
            {
                "name": "单方解除权",
                "keywords": ["单方解除", "随时解除", "无需通知"],
                "risk_level": RiskLevel.HIGH,
                "explanation": "单方解除权使合同缺乏对等约束力",
                "legal_basis": "《民法典》第五百六十三条",
                "suggestion": "建议约定双方对等的解除条件和程序"
            },
            {
                "name": "管辖权不利条款",
                "keywords": ["异地管辖", "境外仲裁", "排除管辖"],
                "risk_level": RiskLevel.MEDIUM,
                "explanation": "管辖权约定可能增加维权成本",
                "legal_basis": "《民事诉讼法》第三十四条",
                "suggestion": "建议选择对己方有利的管辖地"
            }
        ]
    
    def review(self, contract_text: str) -> List[ContractRisk]:
        """执行合同审查"""
        risks = []
        
        # 第一阶段：规则引擎扫描
        risks.extend(self._rule_based_scan(contract_text))
        
        # 第二阶段：LLM 语义分析
        if self.llm:
            risks.extend(self._llm_semantic_analysis(contract_text))
        
        # 去重和排序
        risks = self._deduplicate_and_sort(risks)
        
        return risks
    
    def _rule_based_scan(self, text: str) -> List[ContractRisk]:
        """基于规则的风险扫描"""
        risks = []
        sentences = self._split_sentences(text)
        
        for rule in self.risk_rules:
            for sentence in sentences:
                if any(kw in sentence for kw in rule["keywords"]):
                    risks.append(ContractRisk(
                        clause_text=sentence,
                        risk_type=rule["name"],
                        risk_level=rule["risk_level"],
                        explanation=rule["explanation"],
                        legal_basis=rule["legal_basis"],
                        suggestion=rule["suggestion"]
                    ))
        
        return risks
    
    def _llm_semantic_analysis(self, text: str) -> List[ContractRisk]:
        """LLM 语义分析——发现规则引擎无法覆盖的风险"""
        prompt = f"""
        作为资深合同法律师，请审查以下合同文本，
        识别潜在的法律风险条款。

        合同文本：{text[:3000]}

        请按以下 JSON 格式返回结果（最多 5 条）：
        [
            {{
                "clause_text": "风险条款原文",
                "risk_type": "风险类型",
                "risk_level": "低/中/高/严重",
                "explanation": "风险说明",
                "legal_basis": "法律依据",
                "suggestion": "修改建议"
            }}
        ]
        """
        response = self.llm.generate(prompt)
        # 解析 LLM 返回的 JSON 结果
        return self._parse_llm_response(response)
    
    def _split_sentences(self, text: str) -> List[str]:
        """简单句子分割"""
        import re
        return re.split(r'[。；
]', text)
    
    def _deduplicate_and_sort(self, risks: List[ContractRisk]) -> List[ContractRisk]:
        """去重并按风险等级排序"""
        seen = set()
        unique = []
        for risk in risks:
            key = (risk.risk_type, risk.clause_text[:50])
            if key not in seen:
                seen.add(key)
                unique.append(risk)
        
        level_order = {
            RiskLevel.CRITICAL: 0,
            RiskLevel.HIGH: 1,
            RiskLevel.MEDIUM: 2,
            RiskLevel.LOW: 3
        }
        return sorted(unique, key=lambda r: level_order[r.risk_level])

# 生成审查报告
def generate_review_report(risks: List[ContractRisk]) -> str:
    """生成结构化审查报告"""
    report = []
    report.append(f"# 合同审查报告")
    report.append(f"共发现 {len(risks)} 项风险
")
    
    for i, risk in enumerate(risks, 1):
        report.append(f"## {i}. {risk.risk_type} [{risk.risk_level.value}]")
        report.append(f"条款: {risk.clause_text}")
        report.append(f"说明: {risk.explanation}")
        report.append(f"法律依据: {risk.legal_basis}")
        report.append(f"建议: {risk.suggestion}
")
    
    return "
".join(report)

import re
from dataclasses import dataclass
from typing import List, Optional, Dict

@dataclass
class LegalCitation:
    """法条引用"""
    law_name: str      # 法律名称（如"民法典"）
    article_number: str # 条款号（如"第五百七十七条"）
    raw_text: str      # 原始引用文本

@dataclass
class CitationVerification:
    """引用验证结果"""
    citation: LegalCitation
    exists: bool        # 法条是否存在
    current: bool       # 法条是否当前有效
    content: str        # 法条内容（如果存在）
    last_revised: str   # 最近修订日期

class LegalCitationVerifier:
    """法条引用验证系统
    
    功能：验证 AI 生成的法条引用是否真实存在，
    是否当前有效，并返回法条内容。
    
    这是防止"法律幻觉"的关键基础设施。
    """
    
    def __init__(self, legal_database: Dict):
        """
        Args:
            legal_database: 法律知识数据库
            格式: {
                "民法典": {
                    "第五百七十七条": {
                        "content": "当事人一方不履行合同义务...",
                        "effective": True,
                        "last_revised": "2020-05-28"
                    }
                }
            }
        """
        self.db = legal_database
    
    def extract_citations(self, text: str) -> List[LegalCitation]:
        """从文本中提取法条引用"""
        citations = []
        
        # 匹配模式："XXX法第X条"
        pattern = r'([一-龥]+?)法[s第]*?([一9a-zA-Zd百千万]+?)条'
        matches = re.finditer(pattern, text)
        
        for match in matches:
            law_name = match.group(1) + "法"
            article_num = match.group(2)
            citations.append(LegalCitation(
                law_name=law_name,
                article_number=article_num,
                raw_text=match.group(0)
            ))
        
        return citations
    
    def verify(self, citation: LegalCitation) -> CitationVerification:
        """验证单个法条引用"""
        law_db = self.db.get(citation.law_name, {})
        article_db = law_db.get(citation.article_number)
        
        if article_db is None:
            return CitationVerification(
                citation=citation,
                exists=False,
                current=False,
                content="",
                last_revised=""
            )
        
        return CitationVerification(
            citation=citation,
            exists=True,
            current=article_db.get("effective", False),
            content=article_db.get("content", ""),
            last_revised=article_db.get("last_revised", "")
        )
    
    def verify_text(self, text: str) -> List[CitationVerification]:
        """验证文本中的所有法条引用"""
        citations = self.extract_citations(text)
        return [self.verify(c) for c in citations]
    
    def get_verification_report(self, text: str) -> str:
        """生成验证报告"""
        results = self.verify_text(text)
        
        report = []
        report.append("## 法条引用验证报告
")
        
        valid = sum(1 for r in results if r.exists)
        total = len(results)
        report.append(f"共检测到 {total} 处法条引用，"
                     f"其中 {valid} 处验证通过，"
                     f"{total - valid} 处未找到对应法条
")
        
        for r in results:
            status = "✅" if r.exists else "❌"
            report.append(f"{status} {r.citation.raw_text}")
            if r.exists:
                if not r.current:
                    report.append(f"  ⚠️ 该法条已修订，最后修订日期: {r.last_revised}")
                report.append(f"  内容: {r.content[:100]}...")
            else:
                report.append(f"  🚨 警告：该法条引用可能为虚构")
            report.append("")
        
        return "
".join(report)

# 使用示例
db = {
    "民法典": {
        "第五百七十七条": {
            "content": "当事人一方不履行合同义务或者履行合同义务不符合约定的，应当承担继续履行、采取补救措施或者赔偿损失等违约责任。",
            "effective": True,
            "last_revised": "2020-05-28"
        }
    }
}
verifier = LegalCitationVerifier(db)
text = "根据民法典第五百七十七条，当事人应当承担违约责任。"
print(verifier.get_verification_report(text))

from datasets import Dataset, DatasetDict
from transformers import (
    AutoTokenizer, 
    AutoModelForCausalLM,
    TrainingArguments,
    Trainer,
    DataCollatorForSeq2Seq
)
import json
from typing import List, Dict

class LegalInstructionDataset:
    """法律指令数据集构建器
    
    构建用于法律大模型指令微调的数据集，
    支持多种任务类型和质量控制。
    """
    
    def __init__(self, task_types: List[str] = None):
        self.task_types = task_types or [
            "contract_review",    # 合同审查
            "legal_research",     # 法律检索
            "document_drafting",  # 文书起草
            "compliance_check",   # 合规检查
            "legal_reasoning"     # 法律推理
        ]
    
    def load_from_json(self, filepath: str) -> List[Dict]:
        """从 JSON 文件加载指令数据"""
        with open(filepath, 'r', encoding='utf-8') as f:
            data = json.load(f)
        
        # 验证数据格式
        validated = []
        for item in data:
            if self._validate_item(item):
                validated.append(item)
            else:
                print(f"⚠️ 跳过无效数据项: {item.get('id', 'unknown')}")
        
        return validated
    
    def _validate_item(self, item: Dict) -> bool:
        """验证指令数据项的格式"""
        required_fields = ["instruction", "input", "output"]
        if not all(field in item for field in required_fields):
            return False
        
        # 检查输出质量
        if len(item["output"].strip()) < 20:
            return False  # 输出太短，可能是低质量数据
        
        if item.get("task_type") not in self.task_types:
            return False  # 任务类型不在预期范围内
        
        return True
    
    def to_huggingface_format(self, data: List[Dict]) -> Dataset:
        """转换为 HuggingFace Dataset 格式"""
        formatted = []
        for item in data:
            # 构建指令模板
            prompt = self._build_prompt(item)
            formatted.append({
                "prompt": prompt,
                "response": item["output"],
                "task_type": item.get("task_type", "unknown"),
                "difficulty": item.get("difficulty", "medium")
            })
        
        return Dataset.from_list(formatted)
    
    def _build_prompt(self, item: Dict) -> str:
        """构建指令微调的 prompt"""
        task_type = item.get("task_type", "通用")
        instruction = item["instruction"]
        input_text = item.get("input", "")
        
        prompt = f"""[法律任务类型]: {task_type}

[任务指令]: {instruction}
"""
        if input_text:
            prompt += f"""
[输入内容]:
{input_text}
"""
        prompt += """
[输出]:
"""
        return prompt
    
    def split_train_eval(self, dataset: Dataset, 
                        eval_ratio: float = 0.1) -> DatasetDict:
        """划分训练集和验证集"""
        return dataset.train_test_split(test_size=eval_ratio)


def train_legal_model(
    model_name: str,
    train_dataset: Dataset,
    eval_dataset: Dataset,
    output_dir: str = "./legal-model-output"
):
    """执行法律模型指令微调训练
    
    Args:
        model_name: 基础模型名称（如 "Qwen/Qwen2-7B"）
        train_dataset: 训练数据集
        eval_dataset: 验证数据集
        output_dir: 输出目录
    """
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModelForCausalLM.from_pretrained(
        model_name,
        torch_dtype="auto",
        device_map="auto"
    )
    
    # 数据处理
    def tokenize_fn(examples):
        return tokenizer(
            examples["prompt"],
            text_target=examples["response"],
            truncation=True,
            max_length=4096,
            padding="max_length"
        )
    
    tokenized_train = train_dataset.map(
        tokenize_fn, batched=True, remove_columns=train_dataset.column_names
    )
    tokenized_eval = eval_dataset.map(
        tokenize_fn, batched=True, remove_columns=eval_dataset.column_names
    )
    
    # 训练配置
    training_args = TrainingArguments(
        output_dir=output_dir,
        num_train_epochs=3,
        per_device_train_batch_size=4,
        gradient_accumulation_steps=8,
        learning_rate=2e-5,
        fp16=True,
        logging_steps=50,
        save_steps=200,
        eval_steps=200,
        eval_strategy="steps",
        save_total_limit=3,
        load_best_model_at_end=True,
        metric_for_best_model="eval_loss"
    )
    
    data_collator = DataCollatorForSeq2Seq(
        tokenizer=tokenizer, model=model
    )
    
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=tokenized_train,
        eval_dataset=tokenized_eval,
        data_collator=data_collator,
        tokenizer=tokenizer
    )
    
    # 执行训练
    trainer.train()
    trainer.save_model(output_dir)
    print(f"✅ 法律模型训练完成，保存到 {output_dir}")