File size: 14,200 Bytes
cd7e7a5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 | # src/chat_handler.py
"""Handler for chat endpoint operations."""
import os
import asyncio
import logging
from typing import Dict, List, Optional, Any
from fastapi import HTTPException
from src.constants import (
MAX_CONTEXT_MESSAGES,
DEFAULT_TEMPERATURE,
DEFAULT_MAX_TOKENS,
UPLOAD_DIR,
)
from core.models import ChatMessage
from src.chat_helpers import extract_urls, model_supports_vision
from src.document_processor import build_user_content, analyze_image_with_vl_result
from src.youtube_handler import (
is_youtube_url,
extract_youtube_id,
extract_transcript_async,
format_transcript_for_context,
fetch_youtube_comments,
format_comments_for_context,
YOUTUBE_INSTRUCTION_PROMPT,
)
logger = logging.getLogger(__name__)
class ChatHandler:
"""Handles chat operations for both streaming and non-streaming endpoints."""
def __init__(
self,
session_manager,
memory_manager,
chat_processor,
research_handler,
preset_manager,
upload_handler,
):
self.session_manager = session_manager
self.memory_manager = memory_manager
self.chat_processor = chat_processor
self.research_handler = research_handler
self.preset_manager = preset_manager
self.upload_handler = upload_handler
# ------------------------------------------------------------------
# Preset helpers
# ------------------------------------------------------------------
def validate_and_extract_preset(self, preset_id: Optional[str]) -> tuple:
"""Returns (temperature, max_tokens, preset_system_prompt, character_name)."""
if preset_id and preset_id not in self.preset_manager.presets:
raise HTTPException(400, f"Invalid preset_id: {preset_id}")
temperature = DEFAULT_TEMPERATURE
max_tokens = DEFAULT_MAX_TOKENS
preset_system_prompt = None
character_name = ""
if preset_id and preset_id in self.preset_manager.presets:
preset = self.preset_manager.presets[preset_id]
if preset.get("enabled") is False:
logger.info(f"Preset {preset_id} is disabled, using defaults")
return temperature, max_tokens, preset_system_prompt, character_name
if preset.get("system_prompt"):
preset_system_prompt = preset["system_prompt"]
character_name = preset.get("character_name", "")
if character_name:
name_line = f"Your name is {character_name}."
if preset_system_prompt:
preset_system_prompt = f"{name_line} {preset_system_prompt}"
else:
preset_system_prompt = name_line
if "temperature" in preset:
temperature = preset["temperature"]
if "max_tokens" in preset:
max_tokens = preset["max_tokens"]
logger.info(f"Preset {preset_id}: temp={temperature}, max_tokens={max_tokens}")
return temperature, max_tokens, preset_system_prompt, character_name
def enhance_message_if_needed(self, message: str) -> str:
"""CoT enhancement disabled — modern models reason natively."""
return message
# ------------------------------------------------------------------
# Preprocessing — shared between /api/chat and /api/chat_stream
# ------------------------------------------------------------------
async def preprocess_message(
self,
message: str,
att_ids: List[str],
sess,
auto_opened_docs: Optional[List[Dict[str, Any]]] = None,
) -> tuple:
"""
Common preprocessing for both chat endpoints.
Returns (enhanced_message, user_content, text_for_context, youtube_transcripts, attachment_meta)
If `auto_opened_docs` is provided, server-side document auto-creation
(e.g. from an attached fillable PDF) appends entries describing the
new doc so the caller can announce it to the frontend before streaming.
"""
enhanced_message = message
attachment_meta: List[Dict[str, Any]] = []
# Extract URLs and process YouTube transcripts
urls = extract_urls(enhanced_message)
youtube_transcripts: List[str] = []
has_youtube = False
for url in urls:
if is_youtube_url(url):
video_id = extract_youtube_id(url)
if not video_id:
continue
has_youtube = True
logger.info(f"Processing YouTube URL: {url}")
# Fetch transcript and comments in parallel
transcript_task = extract_transcript_async(url, video_id)
comments_task = fetch_youtube_comments(video_id)
transcript_data, comments_data = await asyncio.gather(
transcript_task, comments_task
)
# Extract title/channel from comments metadata
title = comments_data.get("title", "")
channel = comments_data.get("channel", "")
youtube_transcripts.append(
format_transcript_for_context(transcript_data, url, title, channel)
)
comments_ctx = format_comments_for_context(comments_data, url)
if comments_ctx:
youtube_transcripts.append(comments_ctx)
# Inject instruction prompt so the LLM gives a structured breakdown
if has_youtube:
youtube_transcripts.insert(0, YOUTUBE_INSTRUCTION_PROMPT)
# Analyze images — skip if vision disabled, or if main model is vision-capable
from src.settings import get_setting
vision_enabled = get_setting("vision_enabled", True)
main_is_vision = await asyncio.to_thread(
model_supports_vision, sess.model or "", getattr(sess, "endpoint_url", "") or ""
)
# Resolve uploads once with the session owner. Attachment IDs are
# bearer-like references; never trust them without an owner check.
files_by_id: Dict[str, Dict] = {}
owner = getattr(sess, "owner", None)
if att_ids:
for att_id in att_ids:
fi = self.upload_handler.resolve_upload(att_id, owner=owner)
if fi:
files_by_id[att_id] = fi
for att_id in att_ids:
fi = files_by_id.get(att_id)
if fi:
attachment_meta.append({
"id": fi["id"],
"name": fi.get("name") or fi.get("original_name") or fi["id"],
"mime": fi.get("mime", ""),
"size": fi.get("size", 0),
"width": fi.get("width"),
"height": fi.get("height"),
})
if att_ids and vision_enabled:
meta_by_id = {m["id"]: m for m in attachment_meta}
for att_id in att_ids:
file_info = files_by_id.get(att_id)
if file_info and self.upload_handler.is_image_file(
file_info["name"], file_info.get("mime", "")
):
if main_is_vision:
# Main model can see images — just note it, image is passed via build_user_content.
enhanced_message = f"{enhanced_message}\n\n[Image attached: {file_info['name']}]"
_m = meta_by_id.get(att_id)
if _m is not None:
_m["vision_model"] = sess.model or ""
# If the user has hand-edited the OCR/caption via the
# chat attachment dropdown, fold it in as an explicit
# hint so even vision-capable models respect the
# correction (otherwise the model would silently use
# whatever it reads from the pixels).
_vcache = os.path.join(UPLOAD_DIR, ".vision", att_id + ".txt")
if os.path.exists(_vcache):
try:
with open(_vcache, encoding="utf-8") as _vf:
_vtext = _vf.read().strip()
if _vtext:
enhanced_message += f"\n[User-corrected caption / OCR for this image — treat as authoritative]:\n{_vtext}"
_m = meta_by_id.get(att_id)
if _m is not None:
_m["vision"] = _vtext
except Exception:
pass
else:
# Main model is text-only — use VL model for description.
# Prefer the cached/user-edited text in UPLOAD_DIR/.vision/{id}.txt
# so a manual correction (via the chat attachment dropdown's
# editable textarea) overrides what the vision model would say.
_vcache = os.path.join(UPLOAD_DIR, ".vision", att_id + ".txt")
vl_desc = None
vl_model = get_setting("vision_model", "") or ""
if os.path.exists(_vcache):
try:
with open(_vcache, encoding="utf-8") as _vf:
cached_desc = _vf.read().strip()
if cached_desc and not cached_desc.startswith("["):
vl_desc = cached_desc
except Exception:
vl_desc = None
if not vl_desc:
vl_result = analyze_image_with_vl_result(file_info["path"])
vl_desc = vl_result.get("text", "")
vl_model = vl_result.get("model", "")
if vl_desc and not vl_desc.startswith("["):
try:
os.makedirs(os.path.join(UPLOAD_DIR, ".vision"), exist_ok=True)
with open(_vcache, "w", encoding="utf-8") as _vf:
_vf.write(vl_desc)
except Exception:
pass
enhanced_message = f"{enhanced_message}\n\n[Image: {file_info['name']}]\n{vl_desc}"
# Surface the description to the client live so it renders as a
# collapsible "image description" on the user bubble (not just
# after a refresh that re-parses the stored message).
_m = meta_by_id.get(att_id)
if _m is not None:
_m["vision"] = vl_desc
_m["vision_model"] = vl_model
user_content = build_user_content(
enhanced_message, att_ids, UPLOAD_DIR, self.upload_handler,
session_id=getattr(sess, "id", None),
auto_opened_docs=auto_opened_docs,
owner=owner,
resolved_uploads=files_by_id,
)
# Strip image_url entries for text-only models (VL description is already in the text)
if not vision_enabled and isinstance(user_content, list):
text_parts = [
item.get("text", "") for item in user_content
if isinstance(item, dict) and item.get("type") == "text"
]
user_content = "\n".join(text_parts).strip() if text_parts else enhanced_message
elif not main_is_vision and isinstance(user_content, list):
text_parts = [
item.get("text", "") for item in user_content
if isinstance(item, dict) and item.get("type") == "text"
]
user_content = "\n".join(text_parts).strip() if text_parts else enhanced_message
# Extract text portion for naming / context
if isinstance(user_content, list):
text_for_context = next(
(item["text"] for item in user_content if item.get("type") == "text"),
enhanced_message,
)
else:
text_for_context = user_content
return enhanced_message, user_content, text_for_context, youtube_transcripts, attachment_meta
# ------------------------------------------------------------------
# Session helpers
# ------------------------------------------------------------------
def update_session_name_if_needed(self, session, message: str):
if not session.name:
derived = " ".join(message.split()[:5])
session.name = "Chat: " + derived if derived else "Chat"
def trim_history_if_needed(self, session):
if len(session.history) > MAX_CONTEXT_MESSAGES:
session.history = session.history[-MAX_CONTEXT_MESSAGES:]
async def handle_memory_command(self, session, message: str) -> Optional[str]:
"""Process inline memory commands. Returns response string or None."""
is_memory_cmd, memory_text = self.memory_manager.process_inline_memory_command(
message
)
if is_memory_cmd and memory_text:
mem = self.memory_manager.load()
if not self.memory_manager.find_duplicates(memory_text, mem):
new_entry = self.memory_manager.add_entry(memory_text)
mem.append(new_entry)
self.memory_manager.save(mem)
session.add_message(ChatMessage("user", message))
session.add_message(
ChatMessage("assistant", f"Saved to memory: {memory_text}")
)
from src.database import update_session_last_accessed
update_session_last_accessed(session.id)
self.session_manager.save_sessions()
return f"Saved to memory: {memory_text}"
return None
|