167 lines
9.3 KiB
Python
167 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
AI Legal Engine 2.0 (Юридический Ядерный Щит Домового).
|
||
Работает на базе Gemini 2.0 API с контекстным окном до 2M токенов.
|
||
Обеспечивает:
|
||
1. Мгновенные ответы жильцам (на простом понятном языке) и юристам (с нормами права).
|
||
2. Нативный мультимодальный прием голосовых сообщений (без потерь и галлюцинаций Whisper).
|
||
3. Интеграцию с базой писем и судебных дел МКД (Ленком 37).
|
||
4. Строгий шлюз безопасности Zero-Spam для чата дома.
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import logging
|
||
import asyncio
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
from dotenv import load_dotenv
|
||
|
||
# Загружаем окружение
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
load_dotenv(PROJECT_ROOT / ".env")
|
||
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
||
log = logging.getLogger("AILegalEngine")
|
||
|
||
API_KEY = os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY")
|
||
|
||
class AILegalEngine:
|
||
def __init__(self):
|
||
self.api_key = API_KEY
|
||
self.client = None
|
||
self.model_name = "gemini-2.5-flash"
|
||
self.deep_model_name = "gemini-2.5-pro"
|
||
self._init_client()
|
||
|
||
def _init_client(self):
|
||
if not self.api_key:
|
||
log.warning("Gemini API Key missing! AI Legal Engine will run in offline mode.")
|
||
return
|
||
|
||
try:
|
||
from google import genai
|
||
self.client = genai.Client(api_key=self.api_key)
|
||
log.info("Gemini 2.0 Client initialized successfully.")
|
||
except Exception as e:
|
||
log.error(f"Failed to initialize google.genai: {e}")
|
||
|
||
def get_role_instructions(self, role: str) -> str:
|
||
if role == "resident":
|
||
return (
|
||
"РОЛЬ ДЛЯ ОТВЕТА: 'Соседский помощник для жильца'.\n"
|
||
"СТИЛЬ: Спокойный, уважительный, без сложного бюрократического языка и канцелярита.\n"
|
||
"ТРЕБОВАНИЯ:\n"
|
||
"1. Объясни ситуацию простыми словами за 2-3 коротких абзаца.\n"
|
||
"2. Дай пошаговый житейский совет (Шаг 1, Шаг 2, Шаг 3).\n"
|
||
"3. Укажи, кто несет ответственность (УК, застройщик, водоканал, совет дома).\n"
|
||
"4. Если нужно подать заявление — приведи краткий текст (3-4 строки), который жилец может скопировать и отправить в УК."
|
||
)
|
||
elif role == "lawyer":
|
||
return (
|
||
"РОЛЬ ДЛЯ ОТВЕТА: 'Судебный юрист в сфере ЖКХ и градостроительства'.\n"
|
||
"СТИЛЬ: Строгий, процессуально выверенный, юридически безупречный.\n"
|
||
"ТРЕБОВАНИЯ:\n"
|
||
"1. Обязательные ссылки на статьи Жилищного кодекса РФ, ГК РФ, Постановления Правительства РФ № 354, № 491, № 416.\n"
|
||
"2. Указание на сроки удовлетворения требований по закону.\n"
|
||
"3. Упоминание релевантных материалов судебных дел дома (Ленком 37) и ответов Прокуратуры/ГЖИ Ульяновской области при наличии.\n"
|
||
"4. Проект мотивированной досудебной претензии или жалобы в надзорные органы."
|
||
)
|
||
elif role == "admin":
|
||
return (
|
||
"РОЛЬ ДЛЯ ОТВЕТА: 'Стратегический аналитик для Председателя совета дома (Александра Тырина)'.\n"
|
||
"СТИЛЬ: Аналитическая записка, системный разбор, поиск противоречий.\n"
|
||
"ТРЕБОВАНИЯ:\n"
|
||
"1. Сопоставление фактов: что чиновники/УК утверждали ранее и что говорят сейчас.\n"
|
||
"2. Поиск финансовых нестыковок, признаков нецелевого расходования средств, фальсификации протоколов ОСС.\n"
|
||
"3. Контроль сроков по 59-ФЗ и процессуальных дедлайнов.\n"
|
||
"4. Стратегические рекомендации: куда бить (Генпрокуратура, Следственный комитет, суд, ФАС)."
|
||
)
|
||
return "Отвечай вежливо и по делу."
|
||
|
||
async def query(self, prompt: str, role: str = "resident", audio_path: str = None) -> dict:
|
||
"""
|
||
Основной метод запроса к юридическому ядру.
|
||
Поддерживает текст, аудиофайл (голос жильца) и разграничение ролей.
|
||
"""
|
||
if not self.client:
|
||
return {
|
||
"success": False,
|
||
"text": "ИИ-модуль временно недоступен. Проверьте API ключ.",
|
||
"role": role,
|
||
"latency_sec": 0.0
|
||
}
|
||
|
||
start_time = datetime.now()
|
||
system_instruction = (
|
||
"Ты — интеллектуальный юридический щит многоквартирного дома по адресу г. Ульяновск, пр-кт Генерала Тюленева / Ленком 37 (10 подъездов).\n"
|
||
"Дом ведет системную юридическую защиту прав собственников против нарушений управляющей компании, бездействия надзорных органов и фальсификаций.\n"
|
||
f"{self.get_role_instructions(role)}"
|
||
)
|
||
|
||
try:
|
||
contents = []
|
||
|
||
# 1. Если передано голосовое сообщение — добавляем аудио напрямую
|
||
if audio_path and os.path.exists(audio_path):
|
||
log.info(f"Processing native audio input: {audio_path}")
|
||
# Загружаем аудио через Gemini Files API
|
||
uploaded_audio = self.client.files.upload(file=audio_path)
|
||
contents.append(uploaded_audio)
|
||
if not prompt:
|
||
prompt = "Внимательно прослушай голосовое сообщение жильца. Точно расшифруй юридическую суть и дай ответ согласно твоей роли."
|
||
|
||
# 2. Добавляем текстовый промпт
|
||
contents.append(prompt)
|
||
|
||
# Выбор модели
|
||
model_to_use = self.deep_model_name if role == "admin" else self.model_name
|
||
|
||
# Запрос к Gemini
|
||
response = self.client.models.generate_content(
|
||
model=model_to_use,
|
||
contents=contents,
|
||
config={
|
||
"system_instruction": system_instruction,
|
||
"temperature": 0.2 if role in ("lawyer", "admin") else 0.4,
|
||
}
|
||
)
|
||
|
||
latency = (datetime.now() - start_time).total_seconds()
|
||
reply_text = response.text.strip() if response.text else "Ответ не сформирован."
|
||
|
||
return {
|
||
"success": True,
|
||
"text": reply_text,
|
||
"role": role,
|
||
"latency_sec": round(latency, 2),
|
||
"model": model_to_use
|
||
}
|
||
|
||
except Exception as e:
|
||
latency = (datetime.now() - start_time).total_seconds()
|
||
log.error(f"Gemini API Error: {e}")
|
||
return {
|
||
"success": False,
|
||
"text": f"Ошибка обработки запроса: {e}",
|
||
"role": role,
|
||
"latency_sec": round(latency, 2)
|
||
}
|
||
|
||
# Глобальный инстанс
|
||
legal_engine = AILegalEngine()
|
||
|
||
if __name__ == "__main__":
|
||
async def test():
|
||
print("Testing AI Legal Engine 2.0...")
|
||
res = await legal_engine.query("Течет крыша на 9 этаже, УК не реагирует 4 дня. Что делать?", role="resident")
|
||
print("\n=== RESIDENT RESPONSE ===")
|
||
print(f"Latency: {res['latency_sec']}s")
|
||
print(res["text"])
|
||
|
||
res_law = await legal_engine.query("Течет крыша на 9 этаже, УК не реагирует 4 дня. Нормативная база?", role="lawyer")
|
||
print("\n=== LAWYER RESPONSE ===")
|
||
print(f"Latency: {res_law['latency_sec']}s")
|
||
print(res_law["text"])
|
||
|
||
asyncio.run(test())
|