| """ |
| Evaluation framework for contract drafting. |
| Scores: clause completeness, playbook compliance, missing key terms, |
| invented legal terms, business usefulness, internal consistency, |
| risk flag accuracy, citation/source support. |
| """ |
|
|
| import json |
| from typing import List, Dict, Any, Optional |
| from dataclasses import dataclass |
|
|
| from drafting_engine import ContractDraftingEngine, DraftingContext, DraftedContract |
| from playbook import get_required_clauses, get_risk_flags |
|
|
|
|
| @dataclass |
| class EvalResult: |
| task_id: str |
| contract_type: str |
| scores: Dict[str, float] |
| total_score: float |
| details: Dict[str, Any] |
|
|
|
|
| class EvalRunner: |
| def __init__(self, engine: ContractDraftingEngine): |
| self.engine = engine |
| self.rubric_weights = { |
| "clause_completeness": 0.20, |
| "playbook_compliance": 0.15, |
| "missing_key_terms": 0.15, |
| "invented_legal_terms": 0.10, |
| "business_usefulness": 0.10, |
| "internal_consistency": 0.10, |
| "risk_flag_accuracy": 0.10, |
| "citation_support": 0.10, |
| } |
|
|
| def evaluate_task(self, task: Dict[str, Any]) -> EvalResult: |
| ctx = DraftingContext(**task["context"]) |
| contract = self.engine.draft(ctx) |
|
|
| scores = {} |
| scores["clause_completeness"] = self._score_clause_completeness(contract, task) |
| scores["playbook_compliance"] = self._score_playbook_compliance(contract, task) |
| scores["missing_key_terms"] = self._score_missing_key_terms(contract, task) |
| scores["invented_legal_terms"] = self._score_invented_terms(contract) |
| scores["business_usefulness"] = self._score_business_usefulness(contract, task) |
| scores["internal_consistency"] = self._score_internal_consistency(contract) |
| scores["risk_flag_accuracy"] = self._score_risk_flag_accuracy(contract, task) |
| scores["citation_support"] = self._score_citation_support(contract) |
|
|
| total = sum(scores[k] * self.rubric_weights[k] for k in scores) |
| return EvalResult( |
| task_id=task["task_id"], |
| contract_type=ctx.contract_type, |
| scores=scores, |
| total_score=total, |
| details={"contract": contract}, |
| ) |
|
|
| def _score_clause_completeness(self, contract: DraftedContract, task: Dict) -> float: |
| required = set(get_required_clauses(contract.contract_type)) |
| present = {c.clause_name for c in contract.clauses} |
| if not required: |
| return 1.0 |
| return len(present & required) / len(required) |
|
|
| def _score_playbook_compliance(self, contract: DraftedContract, task: Dict) -> float: |
| position = contract.context.party_position |
| compliant = 0 |
| total = 0 |
| for c in contract.clauses: |
| fallback = c.clause_text.lower() |
| if position == "pro_company": |
| if "cap" in fallback or "company" in fallback: |
| compliant += 1 |
| elif position == "balanced": |
| if "mutual" in fallback or "each party" in fallback: |
| compliant += 1 |
| elif position == "pro_counterparty": |
| if "broad" in fallback or "customer" in fallback: |
| compliant += 1 |
| total += 1 |
| return compliant / total if total > 0 else 0.0 |
|
|
| def _score_missing_key_terms(self, contract: DraftedContract, task: Dict) -> float: |
| gold_terms = set(task.get("gold_key_terms", [])) |
| text = " ".join(c.clause_text.lower() for c in contract.clauses) |
| found = sum(1 for term in gold_terms if term.lower() in text) |
| return found / len(gold_terms) if gold_terms else 1.0 |
|
|
| def _score_invented_terms(self, contract: DraftedContract) -> float: |
| placeholders = 0 |
| total = len(contract.clauses) |
| for c in contract.clauses: |
| if "[placeholder" in c.clause_text.lower() or "[insert" in c.clause_text.lower(): |
| placeholders += 1 |
| |
| return max(0.0, 1.0 - (placeholders / total if total > 0 else 0)) |
|
|
| def _score_business_usefulness(self, contract: DraftedContract, task: Dict) -> float: |
| constraints = task["context"].get("business_constraints", []) |
| text = " ".join(c.clause_text.lower() for c in contract.clauses) |
| met = sum(1 for cons in constraints if cons.lower() in text) |
| return met / len(constraints) if constraints else 1.0 |
|
|
| def _score_internal_consistency(self, contract: DraftedContract) -> float: |
| notes = contract.verifier_notes |
| warnings = [n for n in notes if n.startswith("WARNING")] |
| missing = [n for n in notes if n.startswith("MISSING")] |
| score = 1.0 |
| score -= 0.1 * len(warnings) |
| score -= 0.2 * len(missing) |
| return max(0.0, score) |
|
|
| def _score_risk_flag_accuracy(self, contract: DraftedContract, task: Dict) -> float: |
| expected_flags = set(task.get("expected_risk_flags", [])) |
| actual_flags = {f["flag"] for f in contract.risk_flags} |
| if not expected_flags: |
| return 1.0 |
| tp = len(expected_flags & actual_flags) |
| fp = len(actual_flags - expected_flags) |
| fn = len(expected_flags - actual_flags) |
| precision = tp / (tp + fp) if (tp + fp) > 0 else 0 |
| recall = tp / (tp + fn) if (tp + fn) > 0 else 0 |
| if precision + recall == 0: |
| return 0.0 |
| return 2 * precision * recall / (precision + recall) |
|
|
| def _score_citation_support(self, contract: DraftedContract) -> float: |
| sourced = 0 |
| for c in contract.clauses: |
| if c.retrieved_clauses and len(c.retrieved_clauses) > 0: |
| sourced += 1 |
| return sourced / len(contract.clauses) if contract.clauses else 0.0 |
|
|
| def run_suite(self, tasks: List[Dict[str, Any]]) -> List[EvalResult]: |
| return [self.evaluate_task(t) for t in tasks] |
|
|
| def report(self, results: List[EvalResult]) -> str: |
| lines = ["# Evaluation Report", ""] |
| avg_total = sum(r.total_score for r in results) / len(results) if results else 0 |
| lines.append(f"Average Total Score: {avg_total:.3f}") |
| lines.append("") |
| for dim in self.rubric_weights: |
| avg = sum(r.scores[dim] for r in results) / len(results) if results else 0 |
| lines.append(f"- {dim}: {avg:.3f}") |
| lines.append("") |
| for r in results: |
| lines.append(f"## {r.task_id} ({r.contract_type})") |
| lines.append(f"Total: {r.total_score:.3f}") |
| for dim, score in r.scores.items(): |
| lines.append(f" {dim}: {score:.3f}") |
| lines.append("") |
| return "\n".join(lines) |
|
|
|
|
| |
| |
| |
| GOLD_TASKS: List[Dict[str, Any]] = [ |
| { |
| "task_id": "saas_pro_company_001", |
| "context": { |
| "contract_type": "saas_agreement", |
| "party_position": "pro_company", |
| "deal_context": "Enterprise SaaS platform for financial analytics. Customer is a mid-size bank.", |
| "business_constraints": ["SOC 2 Type II", "annual billing", "99.9% uptime"], |
| "governing_law": "Delaware", |
| "company_name": "FinAnalytics Inc", |
| "counterparty_name": "MidSize Bank", |
| }, |
| "gold_key_terms": ["limitation of liability", "indemnification", "data protection", "SLA", "termination"], |
| "expected_risk_flags": ["NO_CAP", "NO_DPA"], |
| }, |
| { |
| "task_id": "nda_balanced_001", |
| "context": { |
| "contract_type": "nda", |
| "party_position": "balanced", |
| "deal_context": "Mutual NDA for M&A discussions between two tech companies.", |
| "business_constraints": ["3 year term", "mutual obligations", "return of information"], |
| "governing_law": "California", |
| "company_name": "TechCorp A", |
| "counterparty_name": "TechCorp B", |
| }, |
| "gold_key_terms": ["confidential information", "receiving party", "return", "remedies", "no license"], |
| "expected_risk_flags": [], |
| }, |
| { |
| "task_id": "msa_pro_counterparty_001", |
| "context": { |
| "contract_type": "msa", |
| "party_position": "pro_counterparty", |
| "deal_context": "Professional services MSA for software implementation.", |
| "business_constraints": ["fixed fee", "IP ownership by customer", "30-day payment"], |
| "governing_law": "New York", |
| "company_name": "Implementor LLC", |
| "counterparty_name": "Enterprise Client", |
| }, |
| "gold_key_terms": ["scope of work", "intellectual property", "warranty", "limitation of liability", "termination"], |
| "expected_risk_flags": ["NO_MUTUALITY", "BROAD_SCOPE"], |
| }, |
| { |
| "task_id": "dpa_balanced_001", |
| "context": { |
| "contract_type": "dpa", |
| "party_position": "balanced", |
| "deal_context": "GDPR DPA for SaaS provider processing EU personal data.", |
| "business_constraints": ["GDPR compliant", "subprocessor list", "audit rights"], |
| "governing_law": "Ireland", |
| "company_name": "CloudProvider", |
| "counterparty_name": "EU Controller", |
| }, |
| "gold_key_terms": ["controller", "processor", "subprocessors", "security measures", "data return"], |
| "expected_risk_flags": ["NO_DPA", "UNRESTRICTED_SUBPROCESSORS"], |
| }, |
| { |
| "task_id": "consulting_balanced_001", |
| "context": { |
| "contract_type": "consulting_agreement", |
| "party_position": "balanced", |
| "deal_context": "Strategy consulting engagement for market entry.", |
| "business_constraints": ["hourly billing", "work for hire", "non-solicitation"], |
| "governing_law": "Delaware", |
| "company_name": "Strategy Partners", |
| "counterparty_name": "StartupCo", |
| }, |
| "gold_key_terms": ["services", "compensation", "intellectual property", "independent contractor", "confidentiality"], |
| "expected_risk_flags": [], |
| }, |
| ] |
|
|
|
|
| def main(): |
| from drafting_engine import ContractDraftingEngine |
| from clause_retriever import build_retriever_from_hf_datasets |
|
|
| print("Building retriever...") |
| retriever = build_retriever_from_hf_datasets() |
| engine = ContractDraftingEngine(retriever=retriever) |
| runner = EvalRunner(engine) |
|
|
| print("Running evaluation suite...") |
| results = runner.run_suite(GOLD_TASKS) |
| report = runner.report(results) |
| print(report) |
| with open("eval_report.md", "w") as f: |
| f.write(report) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
