AI Legal Workflows - Contract Review Automation 2026

import anthropic
import os
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum
import PyPDF2
import re

class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class ContractClause:
    clause_type: str
    content: str
    page_number: int
    risk_level: RiskLevel
    deviation_notes: str

@dataclass
class ContractAnalysis:
    contract_id: str
    party_names: List[str]
    key_dates: Dict[str, str]
    clauses: List[ContractClause]
    overall_risk: RiskLevel
    recommendations: List[str]
    estimated_review_time: float

class ContractReviewAgent:
    def __init__(self, api_key: str, playbook_path: str):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.playbook = self._load_playbook(playbook_path)

    def _load_playbook(self, path: str) -> Dict:
        """Load company-specific contract playbook with standard positions"""
        # In production, load from database or config management system
        return {
            "limitation_of_liability": {
                "standard": "Cap at 12 months of fees paid",
                "acceptable_deviation": "6-18 months range",
                "red_flags": ["unlimited liability", "consequential damages"]
            },
            "termination": {
                "standard": "30 days notice for convenience",
                "minimum_acceptable": "15 days notice",
                "red_flags": ["no termination right", "automatic renewal"]
            },
            "indemnification": {
                "standard": "Mutual indemnification for third-party claims",
                "red_flags": ["one-sided indemnification", "IP indemnification"]
            },
            "payment_terms": {
                "standard": "Net 30",
                "acceptable_range": "Net 15-45",
                "red_flags": ["payment in advance", "non-refundable"]
            }
        }

    def extract_text_from_pdf(self, pdf_path: str) -> Dict[int, str]:
        """Extract text from PDF preserving page numbers"""
        pages = {}
        with open(pdf_path, 'rb') as file:
            pdf_reader = PyPDF2.PdfReader(file)
            for page_num, page in enumerate(pdf_reader.pages, 1):
                pages[page_num] = page.extract_text()
        return pages

    def identify_clauses(self, contract_text: str) -> List[Dict]:
        """Use Claude to identify and extract key contract clauses"""
        prompt = f"""You are a legal contract analyst. Analyze this contract and identify all key clauses.

For each clause you identify, provide:
1. Clause type (e.g., "limitation_of_liability", "termination", "payment_terms", etc.)
2. The exact text of the clause
3. Page reference if available
4. A brief summary

Contract text:
{contract_text[:15000]}  # Truncate for context window

Return your analysis as a structured list."""

        message = self.client.messages.create(
            model="claude-sonnet-4-5-20250929",
            max_tokens=4096,
            temperature=0,
            messages=[{"role": "user", "content": prompt}]
        )

        # Parse Claude's response into structured format
        return self._parse_clause_response(message.content[0].text)

    def assess_clause_risk(self, clause_type: str, clause_text: str) -> tuple[RiskLevel, str]:
        """Compare clause against playbook and assess risk"""
        if clause_type not in self.playbook:
            return RiskLevel.LOW, "Standard clause type not in playbook"

        standards = self.playbook[clause_type]

        prompt = f"""You are a legal risk analyst. Compare this contract clause against company standards.

Clause Type: {clause_type}
Company Standard Position: {standards['standard']}
Acceptable Deviations: {standards.get('acceptable_deviation', 'None specified')}
Red Flags to Watch For: {standards.get('red_flags', [])}

Actual Clause Text:
{clause_text}

Assess:
1. Does this clause deviate from our standard position?
2. If yes, is the deviation within acceptable range?
3. Does it contain any red flag terms?
4. Risk level: LOW, MEDIUM, HIGH, or CRITICAL
5. Specific concerns and recommended actions

Provide concise, actionable analysis."""

        message = self.client.messages.create(
            model="claude-sonnet-4-5-20250929",
            max_tokens=1024,
            temperature=0,
            messages=[{"role": "user", "content": prompt}]
        )

        response = message.content[0].text

        # Extract risk level and notes
        risk_level = self._extract_risk_level(response)
        return risk_level, response

    def _extract_risk_level(self, response: str) -> RiskLevel:
        """Extract risk level from Claude's response"""
        response_lower = response.lower()
        if "critical" in response_lower:
            return RiskLevel.CRITICAL
        elif "high" in response_lower:
            return RiskLevel.HIGH
        elif "medium" in response_lower:
            return RiskLevel.MEDIUM
        return RiskLevel.LOW

    def _parse_clause_response(self, response: str) -> List[Dict]:
        """Parse structured clause data from Claude's response"""
        # Simplified parser - production version would be more robust
        clauses = []
        # Implementation would parse Claude's structured output
        return clauses

    def generate_review_summary(self, analysis: ContractAnalysis) -> str:
        """Generate executive summary of contract review"""
        high_risk_clauses = [c for c in analysis.clauses if c.risk_level in [RiskLevel.HIGH, RiskLevel.CRITICAL]]

        prompt = f"""Generate an executive summary for this contract review.

Contract ID: {analysis.contract_id}
Parties: {', '.join(analysis.party_names)}
Overall Risk Level: {analysis.overall_risk.value}
High-Risk Clauses Found: {len(high_risk_clauses)}

Key Issues:
{chr(10).join([f"- {c.clause_type}: {c.deviation_notes}" for c in high_risk_clauses])}

Create a concise executive summary (2-3 paragraphs) suitable for a General Counsel, highlighting:
1. Overall assessment and recommendation (approve/negotiate/reject)
2. Top 3 risks that require attention
3. Suggested negotiation points
4. Estimated business impact

Use clear, business-focused language."""

        message = self.client.messages.create(
            model="claude-sonnet-4-5-20250929",
            max_tokens=1024,
            temperature=0.3,
            messages=[{"role": "user", "content": prompt}]
        )

        return message.content[0].text

    def review_contract(self, pdf_path: str, contract_id: str) -> ContractAnalysis:
        """Main entry point: complete contract review workflow"""
        # Step 1: Extract text
        pages = self.extract_text_from_pdf(pdf_path)
        full_text = "\n\n".join([f"[Page {num}]\n{text}" for num, text in pages.items()])

        # Step 2: Identify clauses
        identified_clauses = self.identify_clauses(full_text)

        # Step 3: Assess each clause
        analyzed_clauses = []
        for clause_data in identified_clauses:
            risk_level, notes = self.assess_clause_risk(
                clause_data['type'],
                clause_data['text']
            )
            analyzed_clauses.append(ContractClause(
                clause_type=clause_data['type'],
                content=clause_data['text'],
                page_number=clause_data.get('page', 0),
                risk_level=risk_level,
                deviation_notes=notes
            ))

        # Step 4: Determine overall risk
        risk_levels = [c.risk_level for c in analyzed_clauses]
        if RiskLevel.CRITICAL in risk_levels:
            overall_risk = RiskLevel.CRITICAL
        elif risk_levels.count(RiskLevel.HIGH) >= 2:
            overall_risk = RiskLevel.HIGH
        elif RiskLevel.HIGH in risk_levels or risk_levels.count(RiskLevel.MEDIUM) >= 3:
            overall_risk = RiskLevel.MEDIUM
        else:
            overall_risk = RiskLevel.LOW

        # Step 5: Create analysis object
        analysis = ContractAnalysis(
            contract_id=contract_id,
            party_names=self._extract_parties(full_text),
            key_dates=self._extract_dates(full_text),
            clauses=analyzed_clauses,
            overall_risk=overall_risk,
            recommendations=[],
            estimated_review_time=1.5  # AI-assisted review time in hours
        )

        return analysis

# Example usage
if __name__ == "__main__":
    agent = ContractReviewAgent(
        api_key=os.environ.get("ANTHROPIC_API_KEY"),
        playbook_path="./legal_playbook.json"
    )

    analysis = agent.review_contract(
        pdf_path="./contracts/vendor_agreement_2026.pdf",
        contract_id="VNDR-2026-001"
    )

    print(f"Contract Review Complete: {analysis.contract_id}")
    print(f"Overall Risk: {analysis.overall_risk.value}")
    print(f"High-Risk Clauses: {len([c for c in analysis.clauses if c.risk_level == RiskLevel.HIGH])}")

import Anthropic from '@anthropic-ai/sdk';

interface AgentTask {
  taskId: string;
  agentType: 'contract_analyst' | 'risk_assessor' | 'compliance_checker' | 'negotiation_advisor';
  input: any;
  dependencies: string[];
}

interface AgentResult {
  taskId: string;
  status: 'completed' | 'failed' | 'pending';
  output: any;
  confidence: number;
}

class LegalWorkflowOrchestrator {
  private client: Anthropic;
  private agents: Map<string, LegalAgent>;

  constructor(apiKey: string) {
    this.client = new Anthropic({ apiKey });
    this.agents = new Map();
    this.initializeAgents();
  }

  private initializeAgents() {
    // Each agent has specialized prompts and validation rules
    this.agents.set('contract_analyst', new ContractAnalystAgent(this.client));
    this.agents.set('risk_assessor', new RiskAssessorAgent(this.client));
    this.agents.set('compliance_checker', new ComplianceCheckerAgent(this.client));
    this.agents.set('negotiation_advisor', new NegotiationAdvisorAgent(this.client));
  }

  async executeWorkflow(tasks: AgentTask[]): Promise<Map<string, AgentResult>> {
    const results = new Map<string, AgentResult>();
    const taskGraph = this.buildDependencyGraph(tasks);

    // Execute tasks in dependency order
    for (const level of taskGraph) {
      // Execute tasks at same level in parallel
      const levelResults = await Promise.all(
        level.map(task => this.executeTask(task, results))
      );

      levelResults.forEach(result => {
        results.set(result.taskId, result);
      });
    }

    return results;
  }

  private buildDependencyGraph(tasks: AgentTask[]): AgentTask[][] {
    // Topological sort to determine execution order
    const graph: AgentTask[][] = [];
    const taskMap = new Map(tasks.map(t => [t.taskId, t]));
    const completed = new Set<string>();

    while (completed.size < tasks.length) {
      const currentLevel = tasks.filter(task =>
        !completed.has(task.taskId) &&
        task.dependencies.every(dep => completed.has(dep))
      );

      if (currentLevel.length === 0) {
        throw new Error('Circular dependency detected in task graph');
      }

      graph.push(currentLevel);
      currentLevel.forEach(task => completed.add(task.taskId));
    }

    return graph;
  }

  private async executeTask(
    task: AgentTask,
    previousResults: Map<string, AgentResult>
  ): Promise<AgentResult> {
    const agent = this.agents.get(task.agentType);
    if (!agent) {
      throw new Error(`Unknown agent type: ${task.agentType}`);
    }

    // Gather dependency outputs
    const dependencyOutputs = task.dependencies.map(depId => {
      const result = previousResults.get(depId);
      if (!result || result.status !== 'completed') {
        throw new Error(`Dependency ${depId} not completed`);
      }
      return result.output;
    });

    try {
      const output = await agent.execute(task.input, dependencyOutputs);

      return {
        taskId: task.taskId,
        status: 'completed',
        output,
        confidence: output.confidence || 0.85
      };
    } catch (error) {
      console.error(`Task ${task.taskId} failed:`, error);
      return {
        taskId: task.taskId,
        status: 'failed',
        output: { error: error.message },
        confidence: 0
      };
    }
  }
}

abstract class LegalAgent {
  protected client: Anthropic;

  constructor(client: Anthropic) {
    this.client = client;
  }

  abstract execute(input: any, dependencyOutputs: any[]): Promise<any>;
}

class ContractAnalystAgent extends LegalAgent {
  async execute(input: { contractText: string }, deps: any[]): Promise<any> {
    const message = await this.client.messages.create({
      model: 'claude-sonnet-4-5-20250929',
      max_tokens: 4096,
      temperature: 0,
      messages: [{
        role: 'user',
        content: `Analyze this contract and extract structured data:

Contract:
${input.contractText}

Extract:
1. Party names and roles
2. Key dates (effective date, term, renewal dates)
3. Financial terms (pricing, payment terms, penalties)
4. Deliverables and obligations
5. Special conditions or unusual clauses

Return as structured JSON.`
      }]
    });

    const analysisText = message.content[0].type === 'text'
      ? message.content[0].text
      : '';

    return {
      analysis: this.parseContractAnalysis(analysisText),
      confidence: 0.9
    };
  }

  private parseContractAnalysis(text: string): any {
    // Parse structured output from Claude
    try {
      // Look for JSON in the response
      const jsonMatch = text.match(/\{[\s\S]*\}/);
      if (jsonMatch) {
        return JSON.parse(jsonMatch[0]);
      }
    } catch (e) {
      // Fallback to structured parsing
    }

    return { rawAnalysis: text };
  }
}

class RiskAssessorAgent extends LegalAgent {
  async execute(input: any, deps: any[]): Promise<any> {
    const contractAnalysis = deps[0]?.analysis;

    const message = await this.client.messages.create({
      model: 'claude-sonnet-4-5-20250929',
      max_tokens: 2048,
      temperature: 0,
      messages: [{
        role: 'user',
        content: `You are a legal risk analyst. Assess the risks in this contract.

Contract Analysis:
${JSON.stringify(contractAnalysis, null, 2)}

Evaluate:
1. Financial risks (payment terms, penalties, liability caps)
2. Operational risks (SLAs, termination rights, IP ownership)
3. Compliance risks (regulatory requirements, data privacy)
4. Reputational risks (exclusivity, confidentiality breaches)

For each risk, provide:
- Risk category
- Severity (LOW, MEDIUM, HIGH, CRITICAL)
- Specific concern
- Recommended mitigation

Return as structured JSON array.`
      }]
    });

    const riskText = message.content[0].type === 'text'
      ? message.content[0].text
      : '';

    return {
      risks: this.parseRiskAssessment(riskText),
      confidence: 0.87
    };
  }

  private parseRiskAssessment(text: string): any[] {
    // Parse risk assessment from Claude's response
    return [{ rawRisks: text }];
  }
}

class ComplianceCheckerAgent extends LegalAgent {
  async execute(input: { requirements: string[] }, deps: any[]): Promise<any> {
    const contractAnalysis = deps[0]?.analysis;

    // Check contract against specific compliance requirements
    const checks = await Promise.all(
      input.requirements.map(req => this.checkRequirement(req, contractAnalysis))
    );

    return {
      complianceChecks: checks,
      overallCompliance: checks.every(c => c.compliant),
      confidence: 0.92
    };
  }

  private async checkRequirement(requirement: string, analysis: any): Promise<any> {
    const message = await this.client.messages.create({
      model: 'claude-sonnet-4-5-20250929',
      max_tokens: 512,
      temperature: 0,
      messages: [{
        role: 'user',
        content: `Check if this contract meets the following requirement:

Requirement: ${requirement}

Contract Analysis:
${JSON.stringify(analysis, null, 2)}

Respond with:
1. COMPLIANT or NON_COMPLIANT
2. Explanation
3. Supporting evidence from contract`
      }]
    });

    const responseText = message.content[0].type === 'text'
      ? message.content[0].text
      : '';

    return {
      requirement,
      compliant: responseText.includes('COMPLIANT') && !responseText.includes('NON_COMPLIANT'),
      explanation: responseText
    };
  }
}

class NegotiationAdvisorAgent extends LegalAgent {
  async execute(input: any, deps: any[]): Promise<any> {
    const risks = deps[0]?.risks || [];
    const complianceIssues = deps[1]?.complianceChecks?.filter(c => !c.compliant) || [];

    const message = await this.client.messages.create({
      model: 'claude-sonnet-4-5-20250929',
      max_tokens: 2048,
      temperature: 0.3, // Slightly higher for creative negotiation strategies
      messages: [{
        role: 'user',
        content: `You are a legal negotiation advisor. Based on identified risks and compliance issues, provide negotiation recommendations.

Identified Risks:
${JSON.stringify(risks, null, 2)}

Compliance Issues:
${JSON.stringify(complianceIssues, null, 2)}

For each significant issue, provide:
1. Negotiation priority (MUST-HAVE, SHOULD-HAVE, NICE-TO-HAVE)
2. Proposed contract language change
3. Fallback position if counterparty resists
4. Business justification
5. Estimated impact on deal timeline

Format as actionable negotiation playbook.`
      }]
    });

    const adviceText = message.content[0].type === 'text'
      ? message.content[0].text
      : '';

    return {
      negotiationStrategy: adviceText,
      confidence: 0.85
    };
  }
}

// Example usage
async function main() {
  const orchestrator = new LegalWorkflowOrchestrator(process.env.ANTHROPIC_API_KEY!);

  const workflow: AgentTask[] = [
    {
      taskId: 'analyze',
      agentType: 'contract_analyst',
      input: { contractText: '...' },
      dependencies: []
    },
    {
      taskId: 'assess_risk',
      agentType: 'risk_assessor',
      input: {},
      dependencies: ['analyze']
    },
    {
      taskId: 'check_compliance',
      agentType: 'compliance_checker',
      input: { requirements: ['GDPR compliance', 'SOC 2 requirements'] },
      dependencies: ['analyze']
    },
    {
      taskId: 'negotiation_advice',
      agentType: 'negotiation_advisor',
      input: {},
      dependencies: ['assess_risk', 'check_compliance']
    }
  ];

  const results = await orchestrator.executeWorkflow(workflow);

  console.log('Workflow completed:');
  results.forEach((result, taskId) => {
    console.log(`${taskId}: ${result.status} (confidence: ${result.confidence})`);
  });
}