import os
import sys
import json
import time
import boto3
import httpx
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Any
from anthropic import Anthropic
import openai
from openai import OpenAI
from google import genai
from google.genai import types
from llama_index.core import Settings
from llama_index.llms.openai import OpenAI as LlamaOpenAI
from llama_index.core.llms import ChatMessage
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.retrievers.bm25 import BM25Retriever
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.retrievers import QueryFusionRetriever
from llama_index.core.workflow import Event, Context, Workflow, StartEvent, StopEvent, step
from llama_index.core.schema import NodeWithScore
from config import (
AWS_ACCESS_KEY_ID,
AWS_SECRET_ACCESS_KEY,
ANTHROPIC_API_KEY,
OPENAI_API_KEY,
BUCKET_NAME,
PREFIX_RETRIEVER,
LOCAL_DIR,
SETTINGS,
MAX_TOKENS_CONFIG,
MAX_TOKENS_ANALYSIS,
GENERATION_TEMPERATURE,
LEGAL_POSITION_SCHEMA,
REQUIRED_FILES,
DEBUG_PROMPTS,
ModelProvider,
AnalysisModelName,
DEEPSEEK_API_KEY,
validate_environment
)
from prompts import SYSTEM_PROMPT, LEGAL_POSITION_PROMPT, PRECEDENT_ANALYSIS_TEMPLATE
from utils import (
clean_text,
extract_court_decision_text,
get_links_html,
get_links_html_lp,
extract_json_from_text
)
from embeddings import GeminiEmbedding
# ============ Debug Prompt Logging ============
# DEBUG_PROMPTS is loaded from config (YAML env setting + DEBUG_PROMPTS env var override).
_PROMPT_SEP = "=" * 80
def _log_prompt(provider: str, model: str, system: str, user: str) -> None:
"""Print full system + user prompts when DEBUG_PROMPTS is enabled."""
if not DEBUG_PROMPTS:
return
print(f"\n{_PROMPT_SEP}")
print(f"[PROMPT DEBUG] Provider: {provider} | Model: {model}")
print(f"{_PROMPT_SEP}")
if system:
print("[PROMPT DEBUG] ── SYSTEM PROMPT ──────────────────────────────────")
print(system)
print("[PROMPT DEBUG] ── USER PROMPT ────────────────────────────────────")
print(user)
print(f"{_PROMPT_SEP}\n")
# ============ End Debug Prompt Logging ============
# Initialize embedding model and settings BEFORE importing components
# Priority: OpenAI > Gemini > None
embed_model = None
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
if OPENAI_API_KEY:
embed_model = OpenAIEmbedding(model_name="text-embedding-3-small")
print("OpenAI embedding model initialized successfully")
elif GEMINI_API_KEY:
embed_model = GeminiEmbedding(api_key=GEMINI_API_KEY, model_name="gemini-embedding-001")
print("Gemini embedding model initialized successfully (alternative to OpenAI)")
else:
print("Warning: No embedding API key found (OpenAI or Gemini). Search functionality will be disabled.")
if embed_model:
Settings.embed_model = embed_model
# Set basic LlamaIndex Settings before setting LLM
Settings.chunk_size = SETTINGS["chunk_size"]
Settings.similarity_top_k = SETTINGS["similarity_top_k"]
# Set a default LLM to prevent QueryFusionRetriever from trying to load OpenAI
# Use a mock LLM with minimal initialization to avoid validation issues
# We use DeepSeek but with a gpt-4o-mini model name to pass validation
if DEEPSEEK_API_KEY:
Settings.llm = LlamaOpenAI(
api_key=DEEPSEEK_API_KEY,
api_base="https://api.deepseek.com",
model="gpt-4o-mini" # Use a known model name for validation
)
print("DeepSeek LLM set as default for LlamaIndex (using gpt-4o-mini model name for compatibility)")
elif OPENAI_API_KEY:
Settings.llm = LlamaOpenAI(api_key=OPENAI_API_KEY, model="gpt-4o-mini")
print("OpenAI LLM set as default for LlamaIndex")
# Now we can safely set context_window
Settings.context_window = SETTINGS["context_window"]
# Import components AFTER setting all Settings
from components import search_components
# Initialize S3 client (optional, only if AWS credentials are provided)
s3_client = None
if all([AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY]):
try:
s3_client = boto3.client(
"s3",
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
region_name="eu-north-1"
)
print("AWS S3 client initialized successfully")
except Exception as e:
print(f"Warning: Failed to initialize AWS S3 client: {str(e)}")
s3_client = None
else:
print("AWS credentials not provided. Will use local files only.")
def download_s3_file(bucket_name: str, s3_key: str, local_path: str) -> None:
"""Download a single file from S3."""
if not s3_client:
raise ValueError("S3 client not initialized. Please provide AWS credentials or use local files.")
try:
s3_client.download_file(bucket_name, s3_key, str(local_path))
print(f"Downloaded: {s3_key} -> {local_path}")
except Exception as e:
print(f"Error downloading file {s3_key}: {str(e)}", file=sys.stderr)
raise
def download_s3_folder(bucket_name: str, prefix: str, local_dir: Path) -> None:
"""Download all files from an S3 folder."""
if not s3_client:
raise ValueError("S3 client not initialized. Please provide AWS credentials or use local files.")
try:
response = s3_client.list_objects_v2(Bucket=bucket_name, Prefix=prefix)
if 'Contents' not in response:
raise ValueError(f"No files found in S3 bucket {bucket_name} with prefix {prefix}")
for obj in response['Contents']:
s3_key = obj['Key']
if s3_key.endswith('/'):
continue
local_file_path = local_dir / Path(s3_key).relative_to(prefix)
local_file_path.parent.mkdir(parents=True, exist_ok=True)
s3_client.download_file(bucket_name, s3_key, str(local_file_path))
print(f"Downloaded: {s3_key} -> {local_file_path}")
except Exception as e:
print(f"Error downloading folder {prefix}: {str(e)}", file=sys.stderr)
raise
def initialize_components() -> bool:
"""Initialize all necessary components for the application."""
from index_loader import load_indexes_with_fallback
try:
# Create local directory if it doesn't exist
LOCAL_DIR.mkdir(parents=True, exist_ok=True)
# Check if required files are present
missing_files = [f for f in REQUIRED_FILES if not (LOCAL_DIR / f).exists()]
if missing_files:
print(f"Missing index files: {', '.join(missing_files)}")
print(f"Attempting to load indexes via fallback (local → HF Dataset → S3)...")
indexes_ok = load_indexes_with_fallback(str(LOCAL_DIR))
if not indexes_ok:
# Last resort: try S3 directly if client is available
if s3_client:
print("Fallback failed, trying S3 directly...")
download_s3_folder(BUCKET_NAME, PREFIX_RETRIEVER, LOCAL_DIR)
else:
print(f"Warning: No S3 client and fallback failed for: {', '.join(missing_files)}")
else:
print(f"All required files found locally in {LOCAL_DIR}")
# Final check
missing_files = [f for f in REQUIRED_FILES if not (LOCAL_DIR / f).exists()]
if missing_files:
raise FileNotFoundError(f"Missing required files after all attempts: {', '.join(missing_files)}")
# Initialize search components if any embedding model is available
if embed_model:
success = search_components.initialize_components(LOCAL_DIR)
if not success:
raise RuntimeError("Failed to initialize search components")
print("Search components initialized successfully")
else:
print("Skipping search components initialization (no embedding API key available)")
return True
except Exception as e:
print(f"Error initializing components: {str(e)}", file=sys.stderr)
return False
def deduplicate_nodes(nodes: list[NodeWithScore], key="doc_id"):
"""Видаляє дублікати з результатів пошуку на основі метаданих."""
seen = set()
unique_nodes = []
for node in nodes:
value = node.node.metadata.get(key)
if value and value not in seen:
seen.add(value)
unique_nodes.append(node)
return unique_nodes
def get_text_length_without_spaces(text: str) -> int:
"""Підраховує довжину тексту без пробілів."""
return len(''.join(text.split()))
def get_available_providers() -> Dict[str, bool]:
"""Get status of all AI providers."""
# Use os.getenv directly to ensure we get the latest environment state
# independent of import time
return {
"openai": bool(os.getenv("OPENAI_API_KEY", "")),
"anthropic": bool(os.getenv("ANTHROPIC_API_KEY", "")),
"gemini": bool(os.getenv("GEMINI_API_KEY", "")),
"deepseek": bool(os.getenv("DEEPSEEK_API_KEY", ""))
}
def check_provider_available(provider: str) -> Tuple[bool, str]:
"""
Check if a provider is available.
Returns:
Tuple of (is_available, error_message)
"""
providers = get_available_providers()
provider_key = provider.lower()
if provider_key not in providers:
return False, f"Unknown provider: {provider}"
if not providers[provider_key]:
available = [k.upper() for k, v in providers.items() if v]
if not available:
return False, "No AI provider API keys configured. Please set at least one API key."
return False, f"{provider.upper()} API key not configured. Available providers: {', '.join(available)}"
return True, ""
def normalize_response_keys(response_dict: Dict[str, Any]) -> Dict[str, Any]:
"""
Normalize keys in the response dictionary to match the expected format.
Handles variations like 'text_lp' -> 'text' and 'proceeding_type' -> 'proceeding'.
"""
if not response_dict:
return response_dict
# Map common variations to standard keys
key_mapping = {
"text_lp": "text",
"legal_position_text": "text",
"lp_text": "text",
"proceeding_type": "proceeding",
"type_of_proceeding": "proceeding"
}
normalized = response_dict.copy()
for variant, standard in key_mapping.items():
if variant in normalized and standard not in normalized:
normalized[standard] = normalized.pop(variant)
return normalized
class RetrieverEvent(Event):
"""Event class for retriever operations."""
nodes: list[NodeWithScore]
class LLMAnalyzer:
"""Class for handling different LLM providers."""
def __init__(self, provider: Any, model_name: Any, temperature: float = GENERATION_TEMPERATURE, max_tokens: Optional[int] = None):
self.provider = provider
self.model_name = model_name
self.temperature = temperature
self.max_tokens = max_tokens
if provider == ModelProvider.OPENAI:
if not OPENAI_API_KEY:
raise ValueError(f"OpenAI API key not configured. Please set OPENAI_API_KEY environment variable to use {provider.value} provider.")
# Disable HTTP/2 to avoid 421 Misdirected Request on HF Spaces
self._http_client = httpx.Client(timeout=httpx.Timeout(120.0, connect=30.0), http2=False)
self.client = openai.OpenAI(api_key=OPENAI_API_KEY, http_client=self._http_client)
elif provider == ModelProvider.DEEPSEEK:
if not DEEPSEEK_API_KEY:
raise ValueError(f"DeepSeek API key not configured. Please set DEEPSEEK_API_KEY environment variable to use {provider.value} provider.")
# Disable HTTP/2 for consistent connections
self._http_client = httpx.Client(timeout=httpx.Timeout(120.0, connect=30.0), http2=False)
self.client = openai.OpenAI(api_key=DEEPSEEK_API_KEY, base_url="https://api.deepseek.com", http_client=self._http_client)
elif provider == ModelProvider.ANTHROPIC:
if not ANTHROPIC_API_KEY:
raise ValueError(f"Anthropic API key not configured. Please set ANTHROPIC_API_KEY environment variable to use {provider.value} provider.")
self.client = Anthropic(api_key=ANTHROPIC_API_KEY)
elif provider == ModelProvider.GEMINI:
if not os.environ.get("GEMINI_API_KEY"):
raise ValueError(f"Gemini API key not configured. Please set GEMINI_API_KEY environment variable to use {provider.value} provider.")
# Initialize Gemini client with new API
self.client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
else:
raise ValueError(f"Unsupported provider: {provider}")
async def analyze(self, prompt: str, response_schema: dict) -> str:
"""Analyze text using selected LLM provider."""
if self.provider == ModelProvider.OPENAI:
return await self._analyze_with_openai(prompt, response_schema)
elif self.provider == ModelProvider.DEEPSEEK:
return await self._analyze_with_deepseek(prompt)
elif self.provider == ModelProvider.ANTHROPIC:
return await self._analyze_with_anthropic(prompt, response_schema)
else:
return await self._analyze_with_gemini(prompt, response_schema)
async def _analyze_with_openai(self, prompt: str, response_schema: dict) -> str:
"""Analyze text using OpenAI."""
# Determine model name and if it's a reasoning model
model_val = self.model_name.value if hasattr(self.model_name, "value") else str(self.model_name)
is_reasoning_model = any(m in model_val.lower() for m in ["gpt-4.1", "gpt-4.5", "gpt-5", "o1", "o3"])
# Use developer role for newer models
role = "developer" if is_reasoning_model else "system"
messages = [
ChatMessage(role=role, content=SYSTEM_PROMPT),
ChatMessage(role="user", content=prompt)
]
response_format = {
"type": "json_schema",
"json_schema": {
"name": "relevant_positions_schema",
"schema": response_schema
}
}
try:
completion_params = {
"model": model_val,
"messages": [{"role": m.role, "content": m.content} for m in messages],
"response_format": response_format,
}
# Reasoning models usually require temperature=1.0 or none
if not is_reasoning_model:
completion_params["temperature"] = self.temperature
# Add GPT-5.2 specific parameters
if "gpt-5" in model_val.lower():
completion_params["reasoning_effort"] = "medium"
completion_params["verbosity"] = "medium"
completion_params["store"] = False
# Log full prompts in debug mode
_log_prompt("openai-analyzer", model_val, SYSTEM_PROMPT, prompt)
# Retry logic for OpenAI analysis
max_retries = 3
last_error = None
response = None
for attempt in range(max_retries):
try:
print(f"[DEBUG] OpenAI Analysis call attempt {attempt + 1}/{max_retries}")
response = self.client.chat.completions.create(**completion_params)
break
except Exception as api_err:
last_error = api_err
print(f"[ERROR] OpenAI Analysis attempt {attempt + 1} failed: {type(api_err).__name__}: {str(api_err)}")
if attempt < max_retries - 1:
time.sleep(2 ** attempt)
else:
raise last_error
response_text = response.choices[0].message.content
# Log cache hit stats (automatic caching, no config needed)
if hasattr(response, 'usage') and hasattr(response.usage, 'prompt_tokens_details'):
cached = getattr(response.usage.prompt_tokens_details, 'cached_tokens', 0)
total = response.usage.prompt_tokens
print(f"[CACHE] OpenAI analysis: {cached}/{total} input tokens from cache")
# Verify it's valid JSON
json_data = extract_json_from_text(response_text)
return json.dumps(json_data, ensure_ascii=False) if json_data else response_text
except Exception as e:
raise RuntimeError(f"Error in OpenAI analysis ({model_val}): {str(e)}")
async def _analyze_with_deepseek(self, prompt: str) -> str:
"""Analyze text using DeepSeek."""
model_val = self.model_name.value if hasattr(self.model_name, "value") else str(self.model_name)
is_reasoning = "reasoner" in model_val.lower()
messages = []
if is_reasoning:
# DeepSeek R1 does not support system role, combine with user
messages.append(ChatMessage(role="user", content=f"{SYSTEM_PROMPT}\n\n{prompt}"))
else:
messages.append(ChatMessage(role="system", content=SYSTEM_PROMPT))
messages.append(ChatMessage(role="user", content=prompt))
try:
completion_params = {
"model": model_val,
"messages": [{"role": m.role, "content": m.content} for m in messages],
}
# Use JSON mode and temperature only for non-reasoning models
if not is_reasoning:
completion_params["response_format"] = {'type': 'json_object'}
completion_params["temperature"] = self.temperature
# Log full prompts in debug mode
if is_reasoning:
_log_prompt("deepseek-analyzer", model_val, "", f"{SYSTEM_PROMPT}\n\n{prompt}")
else:
_log_prompt("deepseek-analyzer", model_val, SYSTEM_PROMPT, prompt)
# Retry logic for DeepSeek analysis
max_retries = 3
last_error = None
response = None
for attempt in range(max_retries):
try:
print(f"[DEBUG] DeepSeek Analysis call attempt {attempt + 1}/{max_retries}")
response = self.client.chat.completions.create(**completion_params)
break
except Exception as api_err:
last_error = api_err
print(f"[ERROR] DeepSeek Analysis attempt {attempt + 1} failed: {type(api_err).__name__}: {str(api_err)}")
if attempt < max_retries - 1:
time.sleep(2 ** attempt)
else:
raise last_error
response_text = response.choices[0].message.content
# Verify and clean JSON
json_data = extract_json_from_text(response_text)
return json.dumps(json_data, ensure_ascii=False) if json_data else response_text
except Exception as e:
raise RuntimeError(f"Error in DeepSeek analysis ({model_val}): {str(e)}")
async def _analyze_with_anthropic(self, prompt: str, response_schema: dict) -> str:
"""Analyze text using Anthropic."""
try:
_log_prompt("anthropic-analyzer", str(self.model_name), SYSTEM_PROMPT, prompt)
response = self.client.messages.create(
model=self.model_name,
max_tokens=self.max_tokens or MAX_TOKENS_ANALYSIS,
temperature=self.temperature,
system=[{"type": "text", "text": SYSTEM_PROMPT, "cache_control": {"type": "ephemeral"}}],
messages=[{"role": "user", "content": prompt}]
)
response_text = response.content[0].text
# Extract JSON from potential markdown blocks
json_data = extract_json_from_text(response_text)
if json_data:
return json.dumps(json_data, ensure_ascii=False)
return response_text
except Exception as e:
raise RuntimeError(f"Error in Anthropic analysis: {str(e)}")
async def _analyze_with_gemini(self, prompt: str, response_schema: dict) -> str:
"""Analyze text using Gemini with new API."""
try:
# Форматуємо промпт для отримання відповіді у форматі JSON
json_instruction = """
Твоя відповідь повинна бути в форматі JSON:
{
"relevant_positions": [
{
"lp_id": "ID позиції",
"source_index": "Порядковий номер позиції у списку",
"description": "Детальне обґрунтування релевантності"
}
]
}
"""
formatted_prompt = f"{prompt}\n\n{json_instruction}"
# Log full prompts in debug mode
_log_prompt("gemini-analyzer", str(self.model_name), SYSTEM_PROMPT, formatted_prompt)
# Use new google.genai API
contents = [
types.Content(
role="user",
parts=[
types.Part.from_text(text=formatted_prompt),
],
),
]
generate_content_config = types.GenerateContentConfig(
temperature=self.temperature,
max_output_tokens=self.max_tokens or MAX_TOKENS_ANALYSIS,
system_instruction=[
types.Part.from_text(text=SYSTEM_PROMPT),
],
)
response = self.client.models.generate_content(
model=self.model_name,
contents=contents,
config=generate_content_config,
)
response_text = response.text
if not response_text:
raise RuntimeError("Empty response from Gemini")
# Витягуємо JSON з відповіді за допомогою універсальної функції
json_data = extract_json_from_text(response_text)
if json_data:
if "relevant_positions" not in json_data:
json_data = {
"relevant_positions": [
{
"lp_id": "unknown",
"source_index": "1",
"description": json.dumps(json_data, ensure_ascii=False)
}
]
}
return json.dumps(json_data, ensure_ascii=False)
else:
# Якщо JSON не знайдено, створюємо структурований JSON з тексту
return json.dumps({
"relevant_positions": [
{
"lp_id": "unknown",
"source_index": "1",
"description": response_text
}
]
}, ensure_ascii=False)
except Exception as e:
# Спроба отримати більш детальну інформацію про помилку
error_details = str(e)
if hasattr(e, 'response'):
error_details += f"\nResponse: {e.response}"
raise RuntimeError(f"Error in Gemini analysis: {error_details}")
class PrecedentAnalysisWorkflow(Workflow):
"""Workflow for analyzing legal precedents."""
def __init__(self, provider: Any = ModelProvider.OPENAI,
model_name: Any = AnalysisModelName.GPT4o_MINI,
temperature: float = GENERATION_TEMPERATURE,
max_tokens: Optional[int] = None):
super().__init__()
self.analyzer = LLMAnalyzer(provider, model_name, temperature, max_tokens)
@step
async def analyze(self, ctx: Context, ev: StartEvent) -> StopEvent:
"""Analyze legal precedents."""
try:
query = ev.get("query", "")
question = ev.get("question", "")
nodes = ev.get("nodes", [])
if not query:
return StopEvent(result="Error: No text provided (query)")
if not nodes:
return StopEvent(result="Error: No legal positions provided for analysis (nodes)")
context_parts = []
for i, node in enumerate(nodes, 1):
node_text = node.node.text if hasattr(node, 'node') else node.text
metadata = node.node.metadata if hasattr(node, 'node') else node.metadata
lp_id = metadata.get('lp_id', f'unknown_{i}')
context_parts.append(f"Source {i} (ID: {lp_id}):\n{node_text}")
context_str = "\n\n".join(context_parts)
response_schema = {
"type": "object",
"properties": {
"relevant_positions": {
"type": "array",
"items": {
"type": "object",
"properties": {
"lp_id": {"type": "string"},
"source_index": {"type": "string"},
"description": {"type": "string"}
},
"required": ["lp_id", "source_index", "description"]
}
}
}
}
prompt = PRECEDENT_ANALYSIS_TEMPLATE.format(
query=query,
question=question if question else "Загальний аналіз релевантності",
context_str=context_str
)
response_content = await self.analyzer.analyze(prompt, response_schema)
try:
# Спроба розпарсити JSON
parsed_response = json.loads(response_content)
if "relevant_positions" in parsed_response:
response_lines = []
for position in parsed_response["relevant_positions"]:
position_text = f"* [{position['source_index']}] {position['description']} "
response_lines.append(position_text)
if not response_lines:
return StopEvent(result="Жодної релевантної правової позиції у базі не знайдено.")
response_text = "\n".join(response_lines)
return StopEvent(result=response_text)
else:
# Якщо немає relevant_positions, повертаємо весь текст
return StopEvent(result=f"* [1] {response_content}")
except json.JSONDecodeError as e:
# Якщо не вдалося розпарсити JSON, повертаємо текст як є
return StopEvent(result=f"* [1] {response_content}")
except Exception as e:
return StopEvent(result=f"Error during analysis: {str(e)}")
def generate_legal_position(
input_text: str,
input_type: str,
comment_input: str,
provider: str,
model_name: str,
thinking_enabled: bool = False,
thinking_type: str = "Adaptive",
thinking_level: str = "MEDIUM",
openai_verbosity: str = "medium",
thinking_budget: int = 10000,
temperature: float = GENERATION_TEMPERATURE,
max_tokens: Optional[int] = None,
custom_system_prompt: Optional[str] = None,
custom_lp_prompt: Optional[str] = None
) -> Dict:
"""Generate legal position from input text using specified provider and model."""
try:
# Check if provider is available
is_available, error_msg = check_provider_available(provider)
if not is_available:
return {
"title": "Помилка конфігурації",
"text": error_msg,
"proceeding": "N/A",
"category": "Error"
}
# Use custom prompts if provided, otherwise use defaults
system_prompt = custom_system_prompt if custom_system_prompt else SYSTEM_PROMPT
lp_prompt = custom_lp_prompt if custom_lp_prompt else LEGAL_POSITION_PROMPT
print(f"[DEBUG] RAW input_text length: {len(input_text) if input_text else 0}")
print(f"[DEBUG] RAW input_text preview: {input_text[:300] if input_text else 'Empty'}")
print(f"[DEBUG] Using custom prompts: system={custom_system_prompt is not None}, lp={custom_lp_prompt is not None}")
input_text = clean_text(input_text)
print(f"[DEBUG] AFTER CLEAN input_text length: {len(input_text) if input_text else 0}")
print(f"[DEBUG] AFTER CLEAN input_text preview: {input_text[:300] if input_text else 'Empty'}")
comment_input = clean_text(comment_input)
if input_type == "url":
try:
extracted = extract_court_decision_text(input_text)
print(f"[DEBUG] EXTRACTED text length: {len(extracted) if extracted else 0}")
print(f"[DEBUG] EXTRACTED text preview: {extracted[:300] if extracted else 'Empty'}")
court_decision_text = clean_text(extracted)
print(f"[DEBUG] AFTER CLEAN extracted length: {len(court_decision_text) if court_decision_text else 0}")
print(f"[DEBUG] AFTER CLEAN extracted preview: {court_decision_text[:300] if court_decision_text else 'Empty'}")
except Exception as e:
raise Exception(f"Помилка при отриманні тексту за URL: {str(e)}")
else:
court_decision_text = input_text
# Debug: Check what we have before formatting
print(f"[DEBUG] FINAL court_decision_text length: {len(court_decision_text)}")
print(f"[DEBUG] FINAL court_decision_text preview: {court_decision_text[:300]}")
print(f"[DEBUG] comment_input: {comment_input[:100] if comment_input else 'Empty'}")
# Check if placeholders exist in the prompt, if not - append them to the end
if "{court_decision_text}" not in lp_prompt:
print("[WARNING] {court_decision_text} placeholder missing in prompt! Appending to the end.")
lp_prompt += "\n\n\n{court_decision_text}\n"
if "{comment}" not in lp_prompt:
lp_prompt += "\n\n\n{comment}\n"
content = lp_prompt.format(
court_decision_text=court_decision_text,
comment=comment_input if comment_input else "Коментар відсутній"
)
# Debug: Check formatted content
print(f"[DEBUG] ===== UNIFIED PROMPT FOR ALL PROVIDERS =====")
print(f"[DEBUG] Formatted content length: {len(content)}")
print(f"[DEBUG] Content preview (first 500 chars): {content[:500]}")
print(f"[DEBUG] Provider: {provider}, Model: {model_name}")
print(f"[DEBUG] ==============================================")
# Validation check - ensure court_decision_text is not empty
if not court_decision_text or len(court_decision_text.strip()) < 50:
print(f"[WARNING] court_decision_text is too short or empty! Length: {len(court_decision_text) if court_decision_text else 0}")
raise Exception(f"Текст судового рішення занадто короткий або відсутній (довжина: {len(court_decision_text) if court_decision_text else 0} символів). Будь ласка, перевірте вхідні дані.")
if provider == ModelProvider.OPENAI.value:
# Diagnostic: test raw httpx connection before using OpenAI SDK
http_client = None
response_text = None
try:
print(f"[DEBUG] OpenAI pre-flight check...")
print(f"[DEBUG] openai SDK version: {openai.__version__}")
print(f"[DEBUG] httpx version: {httpx.__version__}")
print(f"[DEBUG] OPENAI_API_KEY length: {len(OPENAI_API_KEY) if OPENAI_API_KEY else 0}")
# Quick raw httpx connectivity test to api.openai.com
try:
with httpx.Client(timeout=10.0, http2=False) as test_client:
test_resp = test_client.get("https://api.openai.com/v1/models",
headers={"Authorization": f"Bearer {OPENAI_API_KEY[:20]}...truncated"})
print(f"[DEBUG] Raw httpx test -> HTTP {test_resp.status_code}")
except Exception as pre_err:
import traceback
print(f"[WARNING] Raw httpx pre-flight failed: {type(pre_err).__name__}: {pre_err}")
print(f"[WARNING] Pre-flight traceback: {traceback.format_exc()}")
http_client = httpx.Client(
timeout=httpx.Timeout(120.0, connect=30.0),
http2=False,
)
client = OpenAI(
api_key=OPENAI_API_KEY,
http_client=http_client
)
print(f"[DEBUG] OpenAI client base_url: {client.base_url}")
# Retry logic for connection errors
max_retries = 3
last_error = None
response = None
print(f"[DEBUG] OpenAI Generation - Model: {model_name}")
# Check for reasoning models (gpt-4.1, gpt-5.2, o1, etc.)
is_reasoning_model = any(m in model_name.lower() for m in ["gpt-4.1", "gpt-4.5", "gpt-5", "o1", "o3"])
# Use developer role for newer models, system for others
role = "developer" if is_reasoning_model else "system"
messages = [
{"role": role, "content": system_prompt},
{"role": "user", "content": content},
]
# Parameters for chat completion
completion_params = {
"model": model_name,
"messages": messages,
}
# Set tokens based on model capabilities
if is_reasoning_model:
completion_params["max_completion_tokens"] = max_tokens or MAX_TOKENS_CONFIG["openai"]
else:
completion_params["max_tokens"] = max_tokens or MAX_TOKENS_CONFIG["openai"]
completion_params["temperature"] = temperature
# Handle thinking/reasoning for GPT-5.2 and other reasoning models
if thinking_enabled and is_reasoning_model:
# GPT-5.2 specific parameters
if "gpt-5" in model_name.lower():
completion_params["reasoning_effort"] = thinking_level.lower()
completion_params["verbosity"] = openai_verbosity.lower()
completion_params["store"] = False
else:
# For other reasoning models (gpt-4.1, o1, etc.)
completion_params["reasoning_effort"] = thinking_level.lower()
# Log full prompts in debug mode
_log_prompt("openai", model_name, system_prompt, content)
# Execute with retries
for attempt in range(max_retries):
try:
print(f"[DEBUG] OpenAI API call attempt {attempt + 1}/{max_retries}")
response = client.chat.completions.create(**completion_params)
break
except Exception as api_err:
import traceback
last_error = api_err
error_type = type(api_err).__name__
error_detail = str(api_err)
# Walk the full exception chain
cause = api_err.__cause__
depth = 0
while cause and depth < 5:
error_detail += f"\n -> Caused by [{depth}]: {type(cause).__name__}: {cause}"
cause = getattr(cause, '__cause__', None) or getattr(cause, '__context__', None)
depth += 1
print(f"[ERROR] OpenAI API attempt {attempt + 1} failed: {error_type}: {error_detail}")
print(f"[ERROR] Full traceback:\n{traceback.format_exc()}")
if attempt < max_retries - 1:
wait_time = 2 ** attempt # 1, 2, 4 seconds
print(f"[DEBUG] Retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise last_error
response_text = response.choices[0].message.content
print(f"[DEBUG] OpenAI response length: {len(response_text) if response_text else 0}")
# Log cache hit stats (automatic caching, no config needed)
if hasattr(response, 'usage') and hasattr(response.usage, 'prompt_tokens_details'):
cached = getattr(response.usage.prompt_tokens_details, 'cached_tokens', 0)
total = response.usage.prompt_tokens
print(f"[CACHE] OpenAI generation: {cached}/{total} input tokens from cache")
json_response = extract_json_from_text(response_text)
if json_response:
json_response = normalize_response_keys(json_response)
if json_response and all(key in json_response for key in ["title", "text", "proceeding", "category"]):
return json_response
else:
print(f"[WARNING] Invalid JSON structure from OpenAI. Text: {response_text[:300] if response_text else 'None'}...")
raise ValueError("Invalid JSON structure")
except Exception as e:
print(f"[ERROR] OpenAI generation/parsing failed: {e}")
return {
"title": "Автоматично сформований заголовок (OpenAI)",
"text": response_text.strip() if response_text else f"Помилка при отриманні відповіді: {str(e)}",
"proceeding": "Не визначено",
"category": "Помилка парсингу"
}
finally:
if http_client:
try:
http_client.close()
except Exception:
pass
if provider == ModelProvider.DEEPSEEK.value:
# Use custom httpx client with HTTP/2 disabled for consistent connections
http_client = None
response_text = None
try:
http_client = httpx.Client(
timeout=httpx.Timeout(120.0, connect=30.0),
http2=False,
)
client = OpenAI(
api_key=DEEPSEEK_API_KEY,
base_url="https://api.deepseek.com",
http_client=http_client
)
# Retry logic for DeepSeek
max_retries = 3
last_error = None
response = None
print(f"[DEBUG] DeepSeek Generation - Model: {model_name}")
# Check for reasoning model (DeepSeek R1)
is_reasoning = "reasoner" in model_name.lower()
messages = []
if is_reasoning:
# R1 does not support system role, combine with user
combined_content = f"{system_prompt}\n\n{content}"
messages.append({"role": "user", "content": combined_content})
else:
messages.append({"role": "system", "content": system_prompt})
completion_params = {
"model": model_name,
"messages": messages,
"max_tokens": max_tokens or MAX_TOKENS_CONFIG["deepseek"],
"frequency_penalty": 0.0,
}
if not is_reasoning:
completion_params["temperature"] = temperature
# Log full prompts in debug mode
if is_reasoning:
_log_prompt("deepseek", model_name, "", combined_content)
else:
_log_prompt("deepseek", model_name, system_prompt, content)
# Execute with retries
for attempt in range(max_retries):
try:
print(f"[DEBUG] DeepSeek API call attempt {attempt + 1}/{max_retries}")
response = client.chat.completions.create(**completion_params)
break
except Exception as api_err:
last_error = api_err
error_type = type(api_err).__name__
print(f"[ERROR] DeepSeek API attempt {attempt + 1} failed: {error_type}: {str(api_err)}")
if attempt < max_retries - 1:
wait_time = 2 ** attempt
print(f"[DEBUG] Retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise last_error
response_text = response.choices[0].message.content
print(f"[DEBUG] DeepSeek response length: {len(response_text) if response_text else 0}")
json_response = extract_json_from_text(response_text)
if json_response:
json_response = normalize_response_keys(json_response)
if json_response and all(key in json_response for key in ["title", "text", "proceeding", "category"]):
return json_response
else:
print(f"[WARNING] Invalid JSON structure from DeepSeek. Text: {response_text[:300] if response_text else 'None'}...")
raise ValueError("Invalid JSON structure")
except Exception as e:
print(f"[ERROR] DeepSeek generation/parsing failed: {e}")
return {
"title": "Автоматично сформований заголовок (DeepSeek)",
"text": response_text.strip() if response_text else f"Помилка при отриманні відповіді від DeepSeek: {str(e)}",
"proceeding": "Не визначено",
"category": "Помилка API/Парсингу"
}
finally:
if http_client:
try:
http_client.close()
except Exception:
pass
elif provider == ModelProvider.ANTHROPIC.value:
client = Anthropic(api_key=ANTHROPIC_API_KEY)
# Debug: check what we're sending to Anthropic
print(f"[DEBUG] Sending to Anthropic - content length: {len(content)}")
print(f"[DEBUG] Content preview: {content[:500]}")
print(f"[DEBUG] ANTHROPIC_API_KEY set: {bool(ANTHROPIC_API_KEY)}, length: {len(ANTHROPIC_API_KEY) if ANTHROPIC_API_KEY else 0}")
messages = [{
"role": "user",
"content": content
}]
# Prepare message creation parameters
message_params = {
"model": model_name,
"max_tokens": max_tokens or MAX_TOKENS_CONFIG["anthropic"],
"system": [{"type": "text", "text": system_prompt, "cache_control": {"type": "ephemeral"}}],
"messages": messages,
"temperature": temperature
}
# Add thinking config if enabled
if thinking_enabled and "claude" in model_name.lower():
# For Claude 4.6 models, we can use Adaptive
if thinking_type.lower() == "adaptive" and getattr(model_name, "find", lambda x: -1)("-4-6") != -1:
message_params["thinking"] = {"type": "adaptive"}
message_params["temperature"] = 1.0
else:
# 'Enabled' type works for both 4.5 and 4.6 models
message_params["thinking"] = {
"type": "enabled",
"budget_tokens": max(1024, int(thinking_budget))
}
message_params["temperature"] = 1.0
# Log full prompts in debug mode
_log_prompt("anthropic", model_name, system_prompt, content)
# Retry logic for connection errors
max_retries = 3
last_error = None
for attempt in range(max_retries):
try:
print(f"[DEBUG] Anthropic API call attempt {attempt + 1}/{max_retries}")
response = client.messages.create(**message_params)
break
except Exception as api_err:
last_error = api_err
error_type = type(api_err).__name__
print(f"[ERROR] Anthropic API attempt {attempt + 1} failed: {error_type}: {str(api_err)}")
if attempt < max_retries - 1:
wait_time = 2 ** attempt # 1, 2, 4 seconds
print(f"[DEBUG] Retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise Exception(f"Помилка з'єднання з Anthropic API після {max_retries} спроб: {error_type}: {str(api_err)}")
try:
# Extract text from response, handling different content block types
response_text = ""
thinking_text = ""
for block in response.content:
if hasattr(block, 'type'):
if block.type == 'thinking':
# Separate thinking blocks (if any)
thinking_text += getattr(block, 'thinking', '')
elif block.type == 'text':
response_text += getattr(block, 'text', '')
elif hasattr(block, 'text'):
# Fallback for simpler response format
response_text += block.text
if thinking_text:
print(f"[DEBUG] Anthropic thinking block length: {len(thinking_text)}")
print(f"[DEBUG] Anthropic response text length: {len(response_text)}")
print(f"[DEBUG] Response preview (first 500 chars): {response_text[:500]}")
# Спробуємо розпарсити JSON за допомогою універсальної функції
json_response = extract_json_from_text(response_text)
if json_response:
json_response = normalize_response_keys(json_response)
# Validate required fields
required = ["title", "text", "proceeding", "category"]
missing = [f for f in required if f not in json_response]
if missing:
print(f"[WARNING] Missing fields in Anthropic JSON: {missing}")
for field in missing:
if field not in json_response:
json_response[field] = "Не вказано"
return json_response
else:
print(f"[ERROR] Could not extract JSON from Anthropic response")
# Fallback: create structured response from raw text
return {
"title": "Автоматично згенерований заголовок",
"text": response_text.strip(),
"proceeding": "Не визначено",
"category": "Помилка парсингу JSON"
}
except Exception as e:
# Скидання помилки для подальшого аналізу
error_details = str(e)
if hasattr(e, 'response'):
error_details += f"\nResponse: {e.response}"
raise RuntimeError(f"Error in Anthropic analysis: {error_details}")
elif provider == ModelProvider.GEMINI.value:
if not os.environ.get("GEMINI_API_KEY"):
raise ValueError("Gemini API key not found in environment variables")
try:
# Debug: Log input parameters
print(f"[DEBUG] Gemini Generation:")
print(f"[DEBUG] Model: {model_name}")
print(f"[DEBUG] Input text length: {len(input_text)}")
print(f"[DEBUG] Court decision text length: {len(court_decision_text)}")
# Use new google.genai API
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
contents = [
types.Content(
role="user",
parts=[
types.Part.from_text(text=content),
],
),
]
# Build config based on model version
config_params = {
"temperature": temperature,
"max_output_tokens": max_tokens or MAX_TOKENS_CONFIG["gemini"],
"system_instruction": [
types.Part.from_text(text=system_prompt),
],
}
# Add thinking config if enabled (only for Gemini 3+ models)
if thinking_enabled and model_name.startswith("gemini-3"):
config_params["thinking_config"] = types.ThinkingConfig(
thinking_level=thinking_level.upper()
)
# Only add response_mime_type for models that support it
# if not model_name.startswith("gemini-3"):
# config_params["response_mime_type"] = "application/json"
generate_content_config = types.GenerateContentConfig(**config_params)
# Log full prompts in debug mode
_log_prompt("gemini", model_name, system_prompt, content)
response = client.models.generate_content(
model=model_name,
contents=contents,
config=generate_content_config,
)
response_text = response.text
# Перевіряємо наявність тексту у відповіді
if not response_text:
raise Exception("Пуста відповідь від моделі Gemini")
print(f"[DEBUG] Gemini response length: {len(response_text)}")
print(f"[DEBUG] Gemini response preview: {response_text[:300]}...")
# Спробуємо розпарсити JSON за допомогою універсальної функції
json_response = extract_json_from_text(response_text)
if json_response:
json_response = normalize_response_keys(json_response)
# Перевіряємо наявність всіх необхідних полів
required_fields = ["title", "text", "proceeding", "category"]
if all(field in json_response for field in required_fields):
return json_response
else:
missing_fields = [field for field in required_fields if field not in json_response]
print(f"[WARNING] Gemini response missing fields: {missing_fields}")
# Fallback for missing fields
for field in required_fields:
if field not in json_response:
json_response[field] = "Не визначено"
return json_response
else:
print(f"[ERROR] Could not extract JSON from Gemini response: {response_text[:300]}...")
return {
"title": "Автоматично сформований заголовок",
"text": response_text.strip(),
"proceeding": "Не визначено",
"category": "Помилка парсингу"
}
except Exception as e:
print(f"Error in Gemini generation: {str(e)}")
return {
"title": "Error in Gemini generation",
"text": str(e),
"proceeding": "Error",
"category": "Error"
}
except Exception as e:
print(f"Error in generate_legal_position: {str(e)}")
return {
"title": "Error",
"text": str(e),
"proceeding": "Unknown",
"category": "Error"
}
async def search_with_ai_action(legal_position_json: Dict) -> Tuple[str, Optional[List[NodeWithScore]]]:
"""Search for relevant legal positions based on input."""
try:
if not embed_model:
return "Помилка: пошук недоступний без налаштованого embedding API ключа (OpenAI або Gemini)", None
retriever = search_components.get_retriever()
if not retriever:
return "Помилка: компоненти пошуку не ініціалізовано", None
query_text = (
f"{legal_position_json['title']}: "
f"{legal_position_json['text']}: "
f"{legal_position_json['proceeding']}: "
f"{legal_position_json['category']}"
)
nodes = await retriever.aretrieve(query_text)
# Видалення дублікатів
unique_nodes = deduplicate_nodes(nodes)
# Обмеження кількості результатів
top_nodes = unique_nodes[:Settings.similarity_top_k]
sources_output = "\n **Результати пошуку (наявні правові позиції Верховного Суду):** \n\n"
for index, node in enumerate(top_nodes, start=1):
source_title = node.node.metadata.get('title')
doc_ids = node.node.metadata.get('doc_id')
lp_ids = node.node.metadata.get('lp_id')
links = get_links_html(doc_ids)
links_lp = get_links_html_lp(lp_ids)
sources_output += f"\n[{index}] *{source_title}* ⚖️ {links_lp} | {links} 👉 Score: {node.score}\n"
return sources_output, top_nodes
except Exception as e:
return f"Помилка при пошуку: {str(e)}", None
async def search_with_raw_text(input_text: str) -> Tuple[str, Optional[List[NodeWithScore]]]:
"""Пошук на основі вхідного тексту з вибором відповідного ретривера."""
try:
if not input_text:
return "Помилка: Порожній текст для пошуку", None
if not embed_model:
return "Помилка: пошук недоступний без налаштованого embedding API ключа (OpenAI або Gemini)", None
retriever = search_components.get_retriever()
if not retriever:
return "Помилка: компоненти пошуку не ініціалізовано", None
# Вибір ретривера залежно від довжини тексту
text_length = get_text_length_without_spaces(input_text)
try:
if text_length < 1024:
nodes = await retriever.aretrieve(input_text)
else:
# Для довгих текстів використовуємо тільки BM25
bm25_retriever = search_components.get_component('bm25_retriever')
if not bm25_retriever:
return "Помилка: BM25 ретривер не ініціалізовано", None
nodes = await bm25_retriever.aretrieve(input_text)
if not nodes:
return "Не знайдено відповідних правових позицій", None
# Видалення дублікатів
unique_nodes = deduplicate_nodes(nodes)
# Обмеження кількості результатів
top_nodes = unique_nodes[:Settings.similarity_top_k]
if not top_nodes:
return "Не знайдено унікальних правових позицій після дедуплікації", None
sources_output = "\n **Результати пошуку (наявні правові позиції Верховного Суду):** \n\n"
for index, node in enumerate(top_nodes, start=1):
source_title = node.node.metadata.get('title', 'Невідомий заголовок')
doc_ids = node.node.metadata.get('doc_id', '')
lp_ids = node.node.metadata.get('lp_id', '')
links = get_links_html(doc_ids)
links_lp = get_links_html_lp(lp_ids)
sources_output += f"\n[{index}] *{source_title}* ⚖️ {links_lp} | {links} 👉 Score: {node.score}\n"
return sources_output, top_nodes
except Exception as e:
return f"Помилка під час виконання пошуку: {str(e)}", None
except Exception as e:
return f"Помилка при пошуку: {str(e)}", None
async def analyze_action(
legal_position_json: Dict,
question: str,
nodes: List[NodeWithScore],
provider: str,
model_name: str,
temperature: float = GENERATION_TEMPERATURE,
max_tokens: Optional[int] = None
) -> str:
"""Analyze search results using AI."""
try:
workflow = PrecedentAnalysisWorkflow(
provider=ModelProvider(provider),
model_name=AnalysisModelName(model_name),
temperature=temperature,
max_tokens=max_tokens
)
query = (
f"{legal_position_json['title']}: "
f"{legal_position_json['text']}: "
f"{legal_position_json['proceeding']}: "
f"{legal_position_json['category']}"
)
response_text = await workflow.run(
query=query,
question=question,
nodes=nodes
)
output = f"**Аналіз ШІ (модель: {model_name}):**\n{response_text}\n\n"
output += "**Наявні в базі правові позицій Верховного Суду:**\n\n"
analysis_lines = response_text.split('\n')
for line in analysis_lines:
if line.startswith('* ['):
index = line[3:line.index(']')]
node = nodes[int(index) - 1]
source_node = node.node
source_title = source_node.metadata.get('title', 'Невідомий заголовок')
source_text_lp = node.text
doc_ids = source_node.metadata.get('doc_id')
lp_id = source_node.metadata.get('lp_id')
links = get_links_html(doc_ids)
links_lp = get_links_html_lp(lp_id)
output += f"[{index}]: *{clean_text(source_title)}* | {clean_text(source_text_lp)} | {links_lp} | {links}\n\n"
return output
except Exception as e:
return f"Помилка при аналізі: {str(e)}"
if __name__ == "__main__":
try:
# Check which providers are available
available_providers = []
if OPENAI_API_KEY:
available_providers.append("OpenAI")
if ANTHROPIC_API_KEY:
available_providers.append("Anthropic")
if os.getenv("GEMINI_API_KEY"):
available_providers.append("Gemini")
if DEEPSEEK_API_KEY:
available_providers.append("DeepSeek")
if not available_providers:
print("Error: No AI provider API keys configured. Please set at least one of:",
file=sys.stderr)
print(" - OPENAI_API_KEY", file=sys.stderr)
print(" - ANTHROPIC_API_KEY", file=sys.stderr)
print(" - GEMINI_API_KEY", file=sys.stderr)
print(" - DEEPSEEK_API_KEY", file=sys.stderr)
sys.exit(1)
print(f"Available AI providers: {', '.join(available_providers)}")
# Check embedding availability for search
if not embed_model:
print("Warning: No embedding model configured. Search functionality will be disabled.")
print(" To enable search, set either OPENAI_API_KEY or GEMINI_API_KEY")
elif GEMINI_API_KEY and not OPENAI_API_KEY:
print("Info: Using Gemini embeddings for search (OpenAI not configured)")
if not all([AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY]):
print("Warning: AWS credentials not configured. Will use local files only.")
# Initialize components
if initialize_components():
print("Components initialized successfully!")
# Import create_gradio_interface here to avoid circular import
from interface import create_gradio_interface
# Create and launch the interface
app = create_gradio_interface()
app.launch(
server_name="0.0.0.0",
server_port=7860,
share=True
)
else:
print("Failed to initialize components. Please check the logs for details.",
file=sys.stderr)
sys.exit(1)
except ImportError as e:
print(f"Error importing required modules: {str(e)}", file=sys.stderr)
sys.exit(1)
except Exception as e:
print(f"Error starting application: {str(e)}", file=sys.stderr)
sys.exit(1)