""" 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 # Score = 1 - fraction of placeholders 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 for evaluation # --------------------------------------------------------------------------- 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()