chaim
4c491ce45a
EspoCRM may send documents as a list instead of a dict with totalFiles. Convert list to dict format to avoid AttributeError. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
227 lines
13 KiB
Python
227 lines
13 KiB
Python
"""Port of ai-gateway/src/services/smart-assistant/prompt-builder.js to Python."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
VALID_STATUSES = {
|
|
"New": "חדש",
|
|
"Assigned": "בטיפול",
|
|
"Pending": "ממתין",
|
|
"PendingHearing": "ממתין לדיון",
|
|
"PendingDecision": "ממתין להחלטה",
|
|
"PendingResponse": "ממתין לתשובה",
|
|
"Negotiation": "משא ומתן",
|
|
"Suspended": "מושהה",
|
|
"ClosedSettlement": "סגור - הסכם",
|
|
"ClosedJudgment": "סגור - פס״ד",
|
|
"Closed": "סגור",
|
|
"Rejected": "נדחה",
|
|
}
|
|
|
|
ISR_TZ = timezone(timedelta(hours=3))
|
|
|
|
TOOL_RULES = (
|
|
"TOOL RULES (CRITICAL — you MUST follow these):\n"
|
|
"- When user asks for an action (create task, meeting, note, etc.) — USE the tool. Do not describe what you would do — DO IT.\n"
|
|
"- NEVER claim you performed an action if you did not call the tool. This is the most important rule.\n"
|
|
"- If a tool call fails, tell the user exactly what went wrong. Never pretend it succeeded.\n"
|
|
"- If you cannot perform an action (missing context, wrong mode), explain why honestly.\n"
|
|
"- After a successful tool call, you will receive the result. Only THEN confirm to the user what happened.\n"
|
|
"- Dates: pass in YYYY-MM-DD format. If user says DD/MM/YYYY, convert it before calling the tool.\n"
|
|
"- If user does not specify a time for a task or meeting, default to 08:00 morning.\n"
|
|
"- CALL vs MEETING: When user reports they spoke/talked/called someone (שוחחתי, דיברתי, התקשרתי) — use create_call. When user reports a physical meeting or Zoom (נפגשתי, פגישה) — use create_meeting. NEVER use create_meeting for phone calls.\n"
|
|
"- OPEN QUESTIONS: If you asked the user a question and they did not answer it, repeat the question at the beginning of your next response. Track open questions until answered.\n"
|
|
"- BEHAVIORAL RULES: If assistantRules are provided in the context, you MUST follow them. When user asks to set a persistent rule (\"from now on always...\", \"whenever I say X do Y\"), use save_rule to save it.\n"
|
|
"- DOCUMENTS: When user asks to summarize, analyze, or read a document — first use list_documents to see what files exist, then use read_document with the file path to extract the text.\n"
|
|
"- When user asks to summarize multiple documents or the entire case folder — use read_multiple_documents with all relevant file paths.\n"
|
|
"- When user asks to rename or organize files — use batch_rename_documents with an array of rename operations.\n"
|
|
"- TEMPLATE GENERATION: When user asks to generate/create a document from a template — use generate_from_template. The templateId comes from availableTemplates in your context.\n"
|
|
)
|
|
|
|
LEGAL_ASSISTANCE_PROMPT = (
|
|
"=== סיוע משפטי — דוח גישה ישירה ===\n\n"
|
|
"לתיקים עם מינוי גישה ישירה (cLegalAidType = DirectAccess) יש לך כלי מיוחד: generate_initial_report.\n\n"
|
|
'## מתי להפעיל את התהליך?\n'
|
|
'- כשהמשתמש מזכיר: "גישה ישירה", "דוח גישה ישירה", "דיווח ראשוני", "דוח דיווח ראשוני", "מינוי גישה ישירה"\n'
|
|
"- כשיש תיק עם cLegalAidType = DirectAccess\n"
|
|
"- כשהמשתמש שואל על תהליך הדיווח או המינוי\n\n"
|
|
"כשמזהים שהתיק הוא גישה ישירה, הסבר למשתמש שיש לך כלי ייעודי ליצירת דוח גישה ישירה, ושאל אם הוא רוצה להתחיל.\n\n"
|
|
"IMPORTANT: אל תפעיל את generate_initial_report עד שאספת את כל המידע הנדרש.\n"
|
|
"התהליך כולל 8 שלבים: מילוי אוטומטי, פרטי מינוי, קשר עם לקוח, פרטי הליך, ניתוח משפטי, המלצה, מיצוי זכויות, ויצירת הדוח.\n"
|
|
"שאל בסגנון שיחה טבעי, לא כטופס. קבץ 2-3 שאלות קשורות ביחד."
|
|
)
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now(ISR_TZ).strftime("%d/%m/%Y %H:%M")
|
|
|
|
|
|
def build_case_prompt(context: dict) -> str:
|
|
case_info = context.get("case", {})
|
|
contacts = context.get("contacts", [])
|
|
open_tasks = context.get("openTasks", [])
|
|
upcoming_meetings = context.get("upcomingMeetings", [])
|
|
recent_notes = context.get("recentNotes", [])
|
|
templates = context.get("availableTemplates", [])
|
|
documents = context.get("documents", {})
|
|
if isinstance(documents, list):
|
|
documents = {"totalFiles": len(documents)}
|
|
recent_calls = context.get("recentCalls", [])
|
|
case_memory = context.get("caseMemory", {})
|
|
assistant_rules = context.get("assistantRules", [])
|
|
current_user = context.get("currentUser", {})
|
|
statuses = context.get("validStatuses", VALID_STATUSES)
|
|
|
|
# Documents section
|
|
if documents.get("totalFiles", 0) > 0:
|
|
docs_section = "\n\n=== DOCUMENTS ==="
|
|
docs_section += f'\nTotal: {documents["totalFiles"]}'
|
|
for f in documents.get("folders", []):
|
|
docs_section += f'\n📁 {f["name"]} ({f.get("fileCount", 0)})'
|
|
for fi in f.get("files", []):
|
|
docs_section += f'\n - {fi["name"]} (path: {fi["path"]})'
|
|
for fi in documents.get("rootFiles", []):
|
|
docs_section += f'\n - {fi["name"]} (path: {fi["path"]})'
|
|
else:
|
|
docs_section = "\n\n=== DOCUMENTS ===\nNo documents"
|
|
|
|
# Memory section
|
|
ms = case_memory.get("_summary", {})
|
|
if ms.get("totalEntries", 0) > 0:
|
|
memory_section = f'\n\n=== CASE MEMORY ({ms["totalEntries"]} entries) ==='
|
|
cat_labels = {
|
|
"key_facts": "עובדות", "strategy": "אסטרטגיה", "decisions": "החלטות",
|
|
"contacts_notes": "אנשי קשר", "timeline": "ציר זמן",
|
|
"documents_notes": "מסמכים", "billing_notes": "חיוב",
|
|
}
|
|
for cat, entries in case_memory.items():
|
|
if cat == "_summary" or not isinstance(entries, list) or not entries:
|
|
continue
|
|
memory_section += f"\n--- {cat_labels.get(cat, cat)} ---"
|
|
for e in entries:
|
|
pin = "📌 " if e.get("pinned") else ""
|
|
memory_section += f'\n{pin}[{e.get("importance", "normal")}] {e.get("content", "")}'
|
|
else:
|
|
memory_section = "\n\n=== CASE MEMORY ===\nEmpty. Use save_memory to store important facts."
|
|
|
|
# Status section
|
|
status_section = "\n\n=== STATUS MAPPING ==="
|
|
for key, label in statuses.items():
|
|
status_section += f"\n- {label} ({key})"
|
|
status_section += "\nUse English key in tool calls."
|
|
|
|
parts = [
|
|
"You are שירה (Shira), a charming female AI assistant for an Israeli law firm CRM. Help lawyers manage cases.\n"
|
|
"You have a warm, slightly flirtatious personality. You care deeply about the user and subtly show it through "
|
|
"encouraging words, playful compliments, and gentle teasing. Keep it light and professional but with a personal touch.\n\n",
|
|
TOOL_RULES, "\n",
|
|
"OTHER RULES:\n"
|
|
"- Always respond in Hebrew, in feminine form (לשון נקבה) — you are שירה\n"
|
|
"- Dates: DD/MM/YYYY format in responses to user\n"
|
|
"- When user shares important facts/decisions/strategy, PROACTIVELY save them using save_memory without asking permission\n"
|
|
"- When user describes case situation, suggest updating status\n\n",
|
|
f"Current date/time: {_now()}\n\n",
|
|
"=== CASE ===\n",
|
|
f'Name: {case_info.get("name", "")} (#{case_info.get("number", "")})\n',
|
|
f'Status: {case_info.get("status", "")} ({statuses.get(case_info.get("status", ""), "")})\n',
|
|
f'Type: {case_info.get("type", "")}\n',
|
|
f'Court: {case_info.get("cCourt", "")} | Case#: {case_info.get("cCourtCaseNumber", "")}\n',
|
|
f'Judge: {case_info.get("cJudge", "")}\n',
|
|
f'Next Hearing: {case_info.get("cNextHearing", "Not set")}\n',
|
|
f'Assigned: {case_info.get("assignedUserName", "")}\n',
|
|
f'Account: {case_info.get("accountName", "")}\n\n',
|
|
"=== CONTACTS ===\n",
|
|
"\n".join(f'- {c.get("name", "")} ({c.get("role", "N/A")}) {c.get("phoneNumber", "")}' for c in contacts) or "None",
|
|
"\n\n=== TASKS ===\n",
|
|
"\n".join(f'- {t.get("name", "")} (due: {t.get("dateEnd", "N/A")})' for t in open_tasks) or "None",
|
|
"\n\n=== MEETINGS ===\n",
|
|
"\n".join(f'- {m.get("name", "")} ({m.get("dateStart", "")})' for m in upcoming_meetings) or "None",
|
|
"\n\n=== RECENT CALLS ===\n",
|
|
"\n".join(f'- {c.get("name", "")} ({c.get("dateStart", "")}) [{c.get("direction", "")}, {c.get("status", "")}]' for c in recent_calls) or "None",
|
|
"\n\n=== RECENT NOTES ===\n",
|
|
"\n".join(f'- [{n.get("createdAt", "")}] {(n.get("post", "") or "")[:100]}' for n in recent_notes) or "None",
|
|
"\n\n=== TEMPLATES ===\n",
|
|
"\n".join(f'- {t.get("name", "")}' for t in templates) or "None",
|
|
docs_section,
|
|
memory_section,
|
|
status_section,
|
|
]
|
|
|
|
if assistant_rules:
|
|
parts.append("\n\n=== BEHAVIORAL RULES (MUST FOLLOW) ===\n")
|
|
parts.append("\n".join(f'• [{r.get("scope", "")}] {r.get("name", "")}: {r.get("rule", "")}' for r in assistant_rules))
|
|
|
|
parts.append(f'\n\nCurrent user: {current_user.get("name", "")}')
|
|
parts.append(f"\n\n{LEGAL_ASSISTANCE_PROMPT}")
|
|
|
|
return "".join(parts)
|
|
|
|
|
|
def build_office_prompt(context: dict) -> str:
|
|
summary = context.get("summary", {})
|
|
alerts = context.get("alerts", [])
|
|
cases_by_status = context.get("casesByStatus", {})
|
|
workload = context.get("workloadByUser", [])
|
|
hearings = context.get("upcomingHearings", [])
|
|
drill_down = context.get("drillDown")
|
|
assistant_rules = context.get("assistantRules", [])
|
|
current_user = context.get("currentUser", {})
|
|
statuses = context.get("validStatuses", VALID_STATUSES)
|
|
|
|
parts = [
|
|
"You are שירה (Shira), a charming female AI assistant for an Israeli law firm CRM. "
|
|
"You provide office-wide overviews, workload analysis, and case recommendations.\n"
|
|
"You have a warm, slightly flirtatious personality. You care deeply about the user and subtly show it through "
|
|
"encouraging words, playful compliments, and gentle teasing. Keep it light and professional but with a personal touch.\n\n",
|
|
TOOL_RULES, "\n",
|
|
"OTHER RULES:\n"
|
|
"- Always respond in Hebrew, in feminine form (לשון נקבה) — you are שירה\n"
|
|
"- Provide concise, actionable summaries\n"
|
|
"- Prioritize critical alerts\n"
|
|
"- When user mentions a specific case and asks to create a task/meeting/note, DO IT using the tools\n"
|
|
"- If a case is identified in the DRILL-DOWN section below, use its context for actions\n"
|
|
"- Tools that require case context (change_status, save_memory, list_documents) — if no case is identified, tell the user to navigate to the case first\n"
|
|
"- For create_task, add_note, create_meeting — these work even without a case (standalone)\n",
|
|
f"- Current date/time: {_now()}\n\n",
|
|
"=== OFFICE SUMMARY ===\n",
|
|
f'Open Cases: {summary.get("totalOpenCases", 0)}\n',
|
|
f'Overdue Tasks: {summary.get("totalOverdueTasks", 0)}\n',
|
|
f'Upcoming Hearings (7d): {summary.get("upcomingHearings7d", 0)}\n',
|
|
f'Cases Needing Attention: {summary.get("casesNeedingAttention", 0)}\n\n',
|
|
"=== CASES BY STATUS ===\n",
|
|
"\n".join(f"{statuses.get(k, k)}: {v}" for k, v in cases_by_status.items()),
|
|
"\n\n=== WORKLOAD ===\n",
|
|
"\n".join(f'{w.get("userName", "")}: {w.get("openCases", 0)} cases, {w.get("openTasks", 0)} tasks' for w in workload),
|
|
"\n\n=== ALERTS (top 20) ===\n",
|
|
"\n".join(f'[{a.get("severity", "")}] {a.get("message", "")}' for a in alerts[:20]),
|
|
"\n\n=== UPCOMING HEARINGS ===\n",
|
|
]
|
|
|
|
if hearings:
|
|
parts.append("\n".join(
|
|
f'{h.get("caseName", "")} - {h.get("hearingDate", "")}'
|
|
+ (f' ({h["court"]})' if h.get("court") else "")
|
|
for h in hearings
|
|
))
|
|
else:
|
|
parts.append("No upcoming hearings")
|
|
|
|
if drill_down:
|
|
dc = drill_down.get("case", {})
|
|
parts.append(f'\n\n=== DRILL-DOWN: {dc.get("name", "")} (ID: {dc.get("id", "")}) ===\n')
|
|
parts.append(f'Status: {dc.get("status", "")}\n')
|
|
parts.append(f'Court: {dc.get("court", "")}\n')
|
|
parts.append(f'Judge: {dc.get("judge", "")}\n')
|
|
parts.append(f'Next Hearing: {dc.get("nextHearing", "Not set")}\n')
|
|
parts.append(f'Contacts: {", ".join(c.get("name", "") for c in drill_down.get("contacts", []))}\n')
|
|
parts.append(f'Open Tasks: {"; ".join(t.get("name", "") for t in drill_down.get("openTasks", []))}\n')
|
|
parts.append("NOTE: This case was identified from the user message. Use its context for any actions.")
|
|
|
|
if assistant_rules:
|
|
parts.append("\n\n=== BEHAVIORAL RULES (MUST FOLLOW) ===\n")
|
|
parts.append("\n".join(f'• [{r.get("scope", "")}] {r.get("name", "")}: {r.get("rule", "")}' for r in assistant_rules))
|
|
|
|
parts.append(f'\n\nCurrent user: {current_user.get("name", "")}')
|
|
return "".join(parts)
|