tocmo0nlord/famlaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
7bc0cc8554c8ec23acae52c6adb892efe124259f
famlaw/activeblue_familylaw/models/fl_ai_engine.py
tocmo0nlord b8ab8494c7 Migrate AI engine to Claude API; convert stage field to Many2one Kanban model 
- Replace fl.case.stage Selection field with Many2one → fl.case.stage model,
  enabling Kanban grouping and dynamic stage management
- Add FlCaseStage model (sequence, fold, description) and fl_stage_data.xml
  with all 11 procedural stages seeded with noupdate=1
- Migrate fl_ai_engine.py from Ollama/llama3.1 to Claude API
  (claude-sonnet-4-20250514); key from ir.config_parameter fl_ai.claude_api_key
- Fix stale field references in _rule_based_tagging and _build_case_context:
  employment/income now read from party_ids, timesharing fields corrected
- Add _fallback_complexity() for graceful degradation when API unavailable
- Add Kanban view to fl_case_views.xml; update action view_mode to kanban,tree,form
- Add fl.case.stage ACL entries to ir.model.access.csv

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-28 17:45:26 +00:00

502 lines
24 KiB
Python
Raw Blame History

import json
import logging
from odoo import models
_logger = logging.getLogger(__name__)
CLAUDE_MODEL = 'claude-sonnet-4-20250514'
# ─────────────────────────────────────────────────────────────────────────────
# Rule-based issue tag weights
# Maps (field_name, value_or_True) → list of issue tag names
# Applied to fl.case fields directly (employment-based rules handled separately
# via party_ids lookup in _rule_based_tagging).
# ─────────────────────────────────────────────────────────────────────────────
CASE_FIELD_RULES = [
('threshold_met', True, ['modification_threshold']),
('substantial_timesharing_applies', True, ['timesharing_deviation']),
('domestic_violence_flag', True, ['domestic_violence']),
('fee_waiver_eligible', True, ['fee_waiver']),
('respondent_answered', False, ['default_judgment']),
('residency_requirement_met', False, ['residency']),
('parenting_class_required', True, ['parenting_class']),
('case_type', 'modification', ['post_order']),
]
# ─────────────────────────────────────────────────────────────────────────────
# Caselaw topic → issue tag matching
# ─────────────────────────────────────────────────────────────────────────────
TAG_TO_CASELAW_DOMAINS = {
'modification_threshold': [('issue_tag_ids.name', '=', 'modification_threshold')],
'income_imputation': [('issue_tag_ids.name', '=', 'income_imputation')],
'self_employment_income': [('issue_tag_ids.name', '=', 'self_employment_income')],
'timesharing_deviation': [('issue_tag_ids.name', '=', 'timesharing_deviation')],
'domestic_violence': [('issue_tag_ids.name', '=', 'domestic_violence')],
'fee_waiver': [('issue_tag_ids.name', '=', 'fee_waiver')],
'default_judgment': [('issue_tag_ids.name', '=', 'default_judgment')],
'residency': [('issue_tag_ids.name', '=', 'residency')],
'parenting_class': [('issue_tag_ids.name', '=', 'parenting_class')],
'post_order': [('issue_tag_ids.name', '=', 'post_order')],
}
MAX_CASELAW_IN_PROMPT = 8
class FlAiEngine(models.AbstractModel):
"""
AI Analysis Engine — Claude API (claude-sonnet-4-20250514).
Workflow:
1. _rule_based_tagging — tag issue tags from case field values (fast, deterministic)
2. _match_caselaw — find relevant FL cases from the caselaw library
3. _build_case_context — serialize case data to a JSON dict
4. _build_prompt — compose the Claude prompt
5. _call_claude — Claude API call with error handling
6. _store_analysis — persist fl.analysis record with results
Falls back to _fallback_complexity() when Claude API is unavailable.
AbstractModel — not stored in the database.
"""
_name = 'fl.ai.engine'
_description = 'Family Law AI Analysis Engine (Claude API)'
# ──────────────────────────────────────────────────────────────────────────
# Public entry point
# ──────────────────────────────────────────────────────────────────────────
def analyze_case(self, case_id):
"""Full analysis entry point. Returns fl.analysis record."""
case = self.env['fl.case'].browse(case_id)
if not case.exists():
raise ValueError(f"Case {case_id} not found")
analysis_vals = {
'case_id': case.id,
'state': 'pending',
'model_used': CLAUDE_MODEL,
}
analysis = self.env['fl.analysis'].create(analysis_vals)
try:
triggered_tags = self._rule_based_tagging(case)
if triggered_tags:
existing_tags = case.issue_tag_ids.mapped('name')
new_tag_recs = self.env['fl.issue.tag'].search([
('name', 'in', list(triggered_tags)),
('name', 'not in', existing_tags),
])
if new_tag_recs:
case.write({'issue_tag_ids': [(4, t.id) for t in new_tag_recs]})
matched_cases = self._match_caselaw(triggered_tags)
context = self._build_case_context(case, matched_cases)
complexity = self._assess_complexity(case, triggered_tags)
prompt = self._build_prompt(context, complexity)
result = self._call_claude(prompt)
self._store_analysis(analysis, result, matched_cases, complexity)
except Exception as exc:
_logger.error("AI analysis failed for case %s: %s", case_id, exc, exc_info=True)
fallback_complexity = self._fallback_complexity(case)
analysis.write({
'state': 'failed',
'error_message': str(exc),
'case_complexity': fallback_complexity,
'plain_english_summary': (
"AI analysis could not be completed at this time. "
"Please try again later or contact support."
),
'plain_english_summary_es': (
"El análisis de IA no pudo completarse en este momento. "
"Por favor intente más tarde o contacte soporte."
),
})
return analysis
# ──────────────────────────────────────────────────────────────────────────
# Step 1: Rule-based tagging
# ──────────────────────────────────────────────────────────────────────────
def _rule_based_tagging(self, case):
"""
Apply deterministic rules to identify legal issues in the case.
Returns a set of issue tag name strings that were triggered.
"""
triggered = set()
# Direct case field checks
for field_name, expected, tags in CASE_FIELD_RULES:
val = getattr(case, field_name, None)
if val == expected:
triggered.update(tags)
# Employment-based checks from party_ids
petitioner_party = case.party_ids.filtered(lambda p: p.role == 'petitioner')
respondent_party = case.party_ids.filtered(lambda p: p.role == 'respondent')
pet = petitioner_party[0] if petitioner_party else None
resp = respondent_party[0] if respondent_party else None
if resp:
if resp.employment_type == 'unemployed':
triggered.add('income_imputation')
elif resp.employment_type in ('self_employed', 'underemployed'):
triggered.update(['self_employment_income', 'income_imputation'])
if pet and pet.employment_type == 'self_employed':
triggered.add('self_employment_income')
# Large income disparity → possible income imputation issue
if pet and resp:
pet_income = pet.effective_monthly_income or 0
resp_income = resp.effective_monthly_income or 0
if (pet_income > 0 and resp_income > 0
and resp.employment_type not in ('unemployed', 'self_employed', 'underemployed')):
ratio = max(pet_income, resp_income) / min(pet_income, resp_income)
if ratio > 3.0:
triggered.add('income_imputation')
# Approaching emancipation → post_order concern
for child in case.child_ids:
if child.approaching_emancipation:
triggered.add('post_order')
break
return triggered
# ──────────────────────────────────────────────────────────────────────────
# Step 2: Match caselaw
# ──────────────────────────────────────────────────────────────────────────
def _match_caselaw(self, triggered_tags):
"""
Search the fl.caselaw library for cases matching triggered issue tags.
Prioritizes 3rd DCA (Miami-Dade's appellate court), then FL Supreme Court.
Returns up to MAX_CASELAW_IN_PROMPT records.
"""
if not triggered_tags:
return self.env['fl.caselaw'].search(
[('active', '=', True)],
order='year desc',
limit=4,
)
matched_ids = set()
for tag in triggered_tags:
domain = TAG_TO_CASELAW_DOMAINS.get(tag, [])
if domain:
recs = self.env['fl.caselaw'].search(
[('active', '=', True)] + domain,
limit=6,
)
matched_ids.update(recs.ids)
if not matched_ids:
return self.env['fl.caselaw'].browse()
court_priority = {
'3rd_dca': 0, 'fl_supreme': 1,
'4th_dca': 2, '2nd_dca': 3,
'1st_dca': 4, '5th_dca': 5,
'11th_circuit': 6, 'other': 7,
}
cases = self.env['fl.caselaw'].browse(list(matched_ids))
sorted_cases = sorted(
cases,
key=lambda c: (court_priority.get(c.court, 9), -c.year)
)
return self.env['fl.caselaw'].browse(
[c.id for c in sorted_cases[:MAX_CASELAW_IN_PROMPT]]
)
# ──────────────────────────────────────────────────────────────────────────
# Step 3: Build case context
# ──────────────────────────────────────────────────────────────────────────
def _build_case_context(self, case, matched_cases):
"""
Serialize the case into a JSON-serializable dict for the prompt.
Keeps PII minimal — uses role names, not SSNs etc.
"""
petitioner_party = case.party_ids.filtered(lambda p: p.role == 'petitioner')
respondent_party = case.party_ids.filtered(lambda p: p.role == 'respondent')
pet = petitioner_party[0] if petitioner_party else None
resp = respondent_party[0] if respondent_party else None
context = {
'case_type': case.case_type,
'stage': case.stage_id.name if case.stage_id else 'unknown',
'filing_date': str(case.filing_date) if case.filing_date else None,
'service_date': str(case.service_date) if case.service_date else None,
'petitioner': {
'employment': pet.employment_type if pet else 'unknown',
'monthly_net_income': pet.effective_monthly_income if pet else 0,
},
'respondent': {
'employment': resp.employment_type if resp else 'unknown',
'monthly_net_income': resp.effective_monthly_income if resp else 0,
'answered': case.respondent_answered,
'has_counsel': case.respondent_has_counsel,
},
'support': {
'current_order_amount': case.current_order_amount or 0,
'calculated_support': case.calculated_support or 0,
'support_difference': case.support_difference or 0,
'support_difference_pct': round(case.support_difference_pct or 0, 1),
'threshold_met': case.threshold_met,
'substantial_change_basis': case.substantial_change_basis or None,
},
'children': [
{
'age': c.age,
'approaching_emancipation': c.approaching_emancipation,
'days_until_emancipation': (
c.days_until_emancipation if c.approaching_emancipation else None
),
}
for c in (case.child_ids or [])
if not c.emancipated
],
'timesharing': {
'petitioner_overnights': case.petitioner_overnights or 0,
'respondent_overnights': case.respondent_overnights or 0,
'substantial_timesharing': case.substantial_timesharing_applies,
},
'residency_met': case.residency_requirement_met,
'domestic_violence': case.domestic_violence_flag,
'fee_waiver_eligible': case.fee_waiver_eligible,
'parenting_class_required': case.parenting_class_required,
'parenting_class_completed': (
case.petitioner_parenting_class_done
and case.respondent_parenting_class_done
),
'issue_tags': list(case.issue_tag_ids.mapped('name')),
'caselaw': [
{
'citation': cl.citation,
'holding': (cl.holding or '')[:300],
'favorable_to': cl.favorable_to,
}
for cl in matched_cases
],
}
return context
# ──────────────────────────────────────────────────────────────────────────
# Step 4: Complexity assessment
# ──────────────────────────────────────────────────────────────────────────
def _assess_complexity(self, case, triggered_tags):
"""Simple heuristic complexity scorer. Returns 'simple', 'moderate', or 'complex'."""
score = 0
if case.domestic_violence_flag:
score += 3
if case.respondent_has_counsel:
score += 2
if 'income_imputation' in triggered_tags:
score += 2
if 'self_employment_income' in triggered_tags:
score += 2
if case.child_ids and len(case.child_ids) > 2:
score += 1
if triggered_tags:
score += len(triggered_tags)
if score <= 3:
return 'simple'
elif score <= 7:
return 'moderate'
else:
return 'complex'
def _fallback_complexity(self, case):
"""
Rule-based complexity fallback used when Claude API is unavailable.
Returns 'simple', 'moderate', or 'complex'.
"""
score = 0
if case.domestic_violence_flag:
score += 3
if case.respondent_has_counsel:
score += 2
if case.child_ids and len(case.child_ids) > 2:
score += 1
if score <= 1:
return 'simple'
elif score <= 4:
return 'moderate'
else:
return 'complex'
# ──────────────────────────────────────────────────────────────────────────
# Step 5: Build prompt
# ──────────────────────────────────────────────────────────────────────────
def _build_prompt(self, context, complexity):
"""
Construct the prompt for Claude.
Returns a string (user message content).
The model is instructed to respond with a JSON object only.
"""
context_json = json.dumps(context, indent=2)
attorney_trigger = (
context.get('domestic_violence') or
context.get('respondent', {}).get('has_counsel') or
complexity == 'complex'
)
prompt = f"""You are an AI legal assistant for a Florida family law case management system. \
Your role is to help pro se (self-represented) litigants in Miami-Dade County understand their case. \
You are NOT providing legal advice — you are explaining the legal framework and procedure.
IMPORTANT RULES:
- Always recommend an attorney when the case involves domestic violence, opposing counsel, or high complexity.
- Florida uses the FL 61.30 income-shares model for child support.
- Modification threshold: 15% AND $50 difference required (FL 61.30(1)(b)).
- Timesharing credit applies when either parent has >73 overnights/year (FL 61.30(11)(b)).
- All output must be JSON only — no markdown, no prose outside the JSON object.
CASE DATA:
{context_json}
TASK: Analyze this case and return a JSON object with exactly these fields:
{{
"plain_english_summary": "3-5 sentence plain English explanation of the case situation and key legal issues — no jargon",
"plain_english_summary_es": "Same 3-5 sentences in Spanish",
"petitioner_arguments": ["argument 1", "argument 2", "argument 3"],
"respondent_counterarguments": ["counterargument 1", "counterargument 2"],
"procedural_risks": ["risk 1", "risk 2"],
"attorney_referral_flag": {"true" if attorney_trigger else "false"},
"attorney_referral_reason": "reason if flag is true, else null",
"confidence_level": "high|medium|low",
"case_complexity": "{complexity}",
"next_steps": ["step 1", "step 2", "step 3"]
}}
Respond with the JSON object only. No other text."""
return prompt
# ──────────────────────────────────────────────────────────────────────────
# Step 6: Call Claude API
# ──────────────────────────────────────────────────────────────────────────
def _call_claude(self, prompt):
"""
Call Claude API and return parsed JSON dict.
Raises RuntimeError on connection error or JSON parse failure.
API key read from ir.config_parameter key 'fl_ai.claude_api_key'.
"""
try:
import anthropic
except ImportError:
raise RuntimeError(
'anthropic library not installed. Run: pip install anthropic'
)
api_key = self.env['ir.config_parameter'].sudo().get_param('fl_ai.claude_api_key')
if not api_key:
raise RuntimeError(
'Claude API key not configured. '
'Set fl_ai.claude_api_key in Settings → Technical → System Parameters.'
)
_logger.info("FL AI Engine: calling Claude API (model=%s)", CLAUDE_MODEL)
client = anthropic.Anthropic(api_key=api_key)
try:
message = client.messages.create(
model=CLAUDE_MODEL,
max_tokens=2048,
messages=[{'role': 'user', 'content': prompt}],
)
except anthropic.RateLimitError as exc:
raise RuntimeError(f'Claude API rate limit exceeded: {exc}') from exc
except anthropic.APIConnectionError as exc:
raise RuntimeError(f'Claude API connection error: {exc}') from exc
except anthropic.APIError as exc:
raise RuntimeError(f'Claude API error: {exc}') from exc
raw = message.content[0].text.strip()
_logger.debug("FL AI Engine raw response (first 200 chars): %s", raw[:200])
# Strip markdown code fences if present
if raw.startswith('```'):
parts = raw.split('```')
raw = parts[1] if len(parts) >= 2 else raw
if raw.lower().startswith('json'):
raw = raw[4:]
raw = raw.strip()
# Extract first JSON object if extra prose leaked through
if raw and raw[0] == '{':
brace_depth = 0
end_idx = 0
for i, ch in enumerate(raw):
if ch == '{':
brace_depth += 1
elif ch == '}':
brace_depth -= 1
if brace_depth == 0:
end_idx = i + 1
break
raw = raw[:end_idx]
try:
return json.loads(raw)
except json.JSONDecodeError as exc:
_logger.error("FL AI Engine: JSON parse error: %s\nRaw: %s", exc, raw[:500])
raise RuntimeError(f"Claude returned invalid JSON: {exc}") from exc
# ──────────────────────────────────────────────────────────────────────────
# Step 7: Store analysis
# ──────────────────────────────────────────────────────────────────────────
def _store_analysis(self, analysis, result, matched_cases, complexity):
"""Write parsed Claude result into the fl.analysis record."""
attorney_flag = result.get('attorney_referral_flag', False)
if isinstance(attorney_flag, str):
attorney_flag = attorney_flag.lower() in ('true', '1', 'yes')
petitioner_args = result.get('petitioner_arguments', [])
respondent_counter = result.get('respondent_counterarguments', [])
procedural_risks = result.get('procedural_risks', [])
vals = {
'state': 'complete',
'plain_english_summary': result.get('plain_english_summary', ''),
'plain_english_summary_es': result.get('plain_english_summary_es', ''),
'attorney_referral_flag': bool(attorney_flag),
'attorney_referral_reason': result.get('attorney_referral_reason') or '',
'confidence_level': result.get('confidence_level', 'medium'),
'case_complexity': result.get('case_complexity', complexity),
'petitioner_arguments': json.dumps(petitioner_args, ensure_ascii=False, indent=2),
'respondent_counterarguments': json.dumps(respondent_counter, ensure_ascii=False, indent=2),
'procedural_risks': json.dumps(procedural_risks, ensure_ascii=False, indent=2),
'raw_response': json.dumps(result, ensure_ascii=False, indent=2),
}
if matched_cases:
vals['matched_caselaw_ids'] = [(6, 0, matched_cases.ids)]
analysis.write(vals)
if attorney_flag and analysis.case_id:
analysis.case_id.message_post(
body=(
"<strong>⚠ AI ANALYSIS — ATTORNEY REFERRAL RECOMMENDED</strong><br/>"
f"{result.get('attorney_referral_reason', 'Case complexity warrants legal counsel.')}<br/>"
"FL Volunteer Lawyers Project: <a href='https://www.flvlp.org'>flvlp.org</a> | "
"3-1-1 Legal Info: <a href='https://www.flcourts.gov'>flcourts.gov</a>"
),
message_type='comment',
subtype_xmlid='mail.mt_comment',
)
_logger.info(
"FL AI Engine: analysis %s complete (complexity=%s, attorney_referral=%s)",
analysis.id, complexity, attorney_flag
)
Reference in New Issue View Git Blame Copy Permalink