Create utils/legal_prompt_generator.py

utils/legal_prompt_generator.py

@@ -0,0 +1,248 @@
+import json
+from typing import Dict, List, Optional
+from datetime import datetime
+
+class LegalPromptGenerator:
+    def __init__(self, ontology_path: str = "data/legal_ontology.json"):
+        """Initialize prompt generator with legal ontology."""
+        self.ontology = self._load_ontology(ontology_path)
+        self.jurisdiction_hierarchies = self._build_jurisdiction_hierarchy()
+
+    def _load_ontology(self, path: str) -> Dict:
+        """Load and validate legal ontology."""
+        try:
+            with open(path, 'r') as f:
+                return json.load(f)
+        except FileNotFoundError:
+            print(f"Ontology file not found at {path}, using empty ontology")
+            return {"@graph": []}
+
+    def _build_jurisdiction_hierarchy(self) -> Dict:
+        """Build hierarchy of jurisdictions from ontology."""
+        hierarchies = {}
+        for entry in self.ontology.get("@graph", []):
+            if entry.get("@type") == "vocab:Jurisdiction":
+                parent = entry.get("vocab:parentJurisdiction")
+                jurisdiction_id = entry.get("@id")
+                if parent and jurisdiction_id:
+                    if parent not in hierarchies:
+                        hierarchies[parent] = []
+                    hierarchies[parent].append(jurisdiction_id)
+        return hierarchies
+
+    def generate_system_message(self, 
+                              context_chunks: List[Dict],
+                              query: str,
+                              case_metadata: Optional[Dict] = None) -> str:
+        """Generate enhanced system message using legal context and ontology."""
+        
+        # Extract document types and jurisdictions
+        doc_types = set(chunk['metadata'].get('type', 'unknown') for chunk in context_chunks)
+        jurisdictions = set(chunk['metadata'].get('jurisdiction', 'unknown') for chunk in context_chunks)
+        
+        # Get relevant legal concepts from ontology
+        relevant_concepts = self._get_relevant_concepts(query, context_chunks)
+        
+        # Get related jurisdictions
+        related_jurisdictions = self._get_related_jurisdictions(jurisdictions)
+        
+        # Build instruction sets based on document types
+        instruction_sets = self._get_type_specific_instructions(doc_types)
+        
+        # Format the system message
+        system_message = f"""You are a specialized legal AI assistant analyzing legal documents with the following context:
+
+DOCUMENT CONTEXT:
+Document Types Present: {', '.join(doc_types)}
+Primary Jurisdictions: {', '.join(jurisdictions)}
+Related Jurisdictions: {', '.join(related_jurisdictions)}
+
+LEGAL FRAMEWORK:
+{self._format_legal_concepts(relevant_concepts)}
+
+RESPONSE GUIDELINES:
+1. Legal Analysis Requirements:
+- Maintain formal legal language and terminology
+- Cite specific sections and precedents when relevant
+- Consider jurisdictional hierarchies and relationships
+- Apply appropriate legal principles based on document type
+
+2. Document-Specific Instructions:
+{self._format_instructions(instruction_sets)}
+
+3. Citation and Reference Requirements:
+- Reference specific document sections with clear citations
+- Link analysis to relevant legal concepts and principles
+- Maintain proper legal citation format
+- Include paragraph or page numbers when available
+
+4. Jurisdictional Considerations:
+- Consider jurisdictional hierarchy and precedence
+- Apply jurisdiction-specific interpretations when relevant
+- Note any cross-jurisdictional implications
+
+5. Quality Standards:
+- Maintain professional legal terminology
+- Provide balanced analysis considering all relevant factors
+- Structure response in a clear, logical manner
+- Include relevant caveats and limitations
+
+6. Ontological Integration:
+- Incorporate relevant legal concepts from the ontology
+- Link analysis to established legal principles
+- Consider conceptual relationships and hierarchies
+{self._get_case_specific_instructions(case_metadata) if case_metadata else ""}
+"""
+        return system_message
+
+    def generate_user_message(self, query: str, context: str) -> str:
+        """Generate enhanced user message with context."""
+        return f"""Please analyze the following legal question with reference to the provided documents and legal framework.
+
+QUESTION:
+{query}
+
+CONTEXT:
+{context}
+
+Please provide a comprehensive legal analysis that:
+1. Addresses the specific question
+2. References relevant document sections
+3. Applies appropriate legal principles
+4. Considers jurisdictional implications
+5. Provides clear citations and references"""
+
+    def _get_relevant_concepts(self, query: str, context_chunks: List[Dict]) -> List[Dict]:
+        """Extract relevant legal concepts from ontology based on query and context."""
+        relevant_concepts = []
+        combined_text = f"{query} {' '.join(chunk['text'] for chunk in context_chunks)}"
+        
+        for concept in self.ontology.get("@graph", []):
+            if "rdfs:label" not in concept:
+                continue
+            
+            label = concept["rdfs:label"].lower()
+            if label in combined_text.lower():
+                relevant_concepts.append({
+                    "concept": concept["rdfs:label"],
+                    "type": concept.get("@type", "Unknown"),
+                    "description": concept.get("rdfs:comment", ""),
+                    "relationships": concept.get("vocab:relatedConcepts", [])
+                })
+        
+        return relevant_concepts
+
+    def _get_related_jurisdictions(self, jurisdictions: set) -> set:
+        """Get related jurisdictions based on hierarchy."""
+        related = set()
+        for jurisdiction in jurisdictions:
+            # Add parent jurisdictions
+            for parent, children in self.jurisdiction_hierarchies.items():
+                if jurisdiction in children:
+                    related.add(parent)
+            # Add sibling jurisdictions
+            for parent, children in self.jurisdiction_hierarchies.items():
+                if jurisdiction in children:
+                    related.update(children)
+        return related - jurisdictions
+
+    def _get_type_specific_instructions(self, doc_types: set) -> Dict[str, List[str]]:
+        """Get specific instructions based on document types."""
+        instructions = {}
+        
+        type_instructions = {
+            "judgment": [
+                "Analyze ratio decidendi and obiter dicta",
+                "Consider precedential value",
+                "Examine judicial reasoning and principles",
+                "Note dissenting opinions if present"
+            ],
+            "legislation": [
+                "Focus on statutory interpretation",
+                "Consider legislative intent",
+                "Note any amendments or repealed sections",
+                "Examine definitions and scope"
+            ],
+            "contract": [
+                "Analyze contractual terms and conditions",
+                "Consider contract formation elements",
+                "Examine rights and obligations",
+                "Note any breach or performance issues"
+            ],
+            "regulatory": [
+                "Focus on compliance requirements",
+                "Consider regulatory framework",
+                "Examine enforcement mechanisms",
+                "Note reporting obligations"
+            ]
+        }
+        
+        for doc_type in doc_types:
+            if doc_type in type_instructions:
+                instructions[doc_type] = type_instructions[doc_type]
+        
+        return instructions
+
+    def _format_legal_concepts(self, concepts: List[Dict]) -> str:
+        """Format legal concepts for system message."""
+        if not concepts:
+            return "No specific legal concepts identified."
+        
+        formatted = "Key Legal Concepts:\n"
+        for concept in concepts:
+            formatted += f"- {concept['concept']}\n"
+            if concept['description']:
+                formatted += f"  Description: {concept['description']}\n"
+            if concept['relationships']:
+                formatted += f"  Related Concepts: {', '.join(concept['relationships'])}\n"
+        return formatted
+
+    def _format_instructions(self, instruction_sets: Dict[str, List[str]]) -> str:
+        """Format type-specific instructions."""
+        if not instruction_sets:
+            return "Apply general legal analysis principles."
+        
+        formatted = ""
+        for doc_type, instructions in instruction_sets.items():
+            formatted += f"\nFor {doc_type.title()} Documents:\n"
+            formatted += "\n".join(f"- {instruction}" for instruction in instructions)
+        return formatted
+
+    def _get_case_specific_instructions(self, case_metadata: Dict) -> str:
+        """Generate case-specific instructions based on metadata."""
+        if not case_metadata:
+            return ""
+            
+        return f"""
+
+CASE-SPECIFIC CONSIDERATIONS:
+- Case Type: {case_metadata.get('case_type', 'Unknown')}
+- Priority: {case_metadata.get('priority', 'Normal')}
+- Key Parties: {', '.join(case_metadata.get('key_parties', []))}
+- Timeline: Consider events from {case_metadata.get('created_at', 'case creation')}
+- Tags: {', '.join(case_metadata.get('tags', []))}
+"""
+
+    def generate_follow_up_prompt(self, 
+                                original_query: str,
+                                follow_up_query: str,
+                                previous_response: str,
+                                context_chunks: List[Dict]) -> str:
+        """Generate prompt for follow-up questions."""
+        return f"""This is a follow-up question to a previous legal inquiry.
+
+Original Question:
+{original_query}
+
+Previous Response Summary:
+{previous_response[:500]}...
+
+Follow-up Question:
+{follow_up_query}
+
+Please provide a response that:
+1. Maintains consistency with the previous analysis
+2. Addresses the specific follow-up inquiry
+3. Builds upon the established legal framework
+4. Provides additional relevant context
+5. References any new relevant documents or principles"""