wehub-resource-sync
875 lines
32 KiB
Python
875 lines
32 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import uuid
|
|
from collections.abc import AsyncGenerator
|
|
from dataclasses import dataclass
|
|
from typing import Any, Literal
|
|
|
|
import dirtyjson # type: ignore[import-untyped]
|
|
from fastapi import HTTPException
|
|
from llmai import get_client # type: ignore[import-not-found]
|
|
from llmai.shared import ( # type: ignore[import-not-found]
|
|
AssistantMessage,
|
|
AssistantToolCall,
|
|
Message,
|
|
SystemMessage,
|
|
TextContentPart,
|
|
ToolResponseMessage,
|
|
UserMessage,
|
|
)
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from models.chat import ChatAttachment
|
|
from models.sql.presentation import PresentationModel
|
|
from services.chat.conversation_store import ChatConversationStore
|
|
from services.chat.presentation_context_store import PresentationContextStore
|
|
from services.chat.prompts import build_system_prompt
|
|
from services.chat.llm_tools import build_chat_llm_tools
|
|
from services.chat.tools import ChatToolMode, ChatTools
|
|
from services.documents_loader import DocumentsLoader
|
|
from services.mem0_presentation_memory_service import MEM0_PRESENTATION_MEMORY_SERVICE
|
|
from services.temp_file_service import TEMP_FILE_SERVICE
|
|
from utils.llm_client_error_handler import handle_llm_client_exceptions
|
|
from utils.llm_config import get_llm_config
|
|
from utils.llm_provider import get_model
|
|
from utils.llm_utils import (
|
|
extract_text,
|
|
get_generate_kwargs,
|
|
stream_generate_events,
|
|
)
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
MAX_TOOL_ROUNDS = 40
|
|
MAX_CHAT_ATTACHMENT_CONTEXT_CHARS = 6000
|
|
MAX_CHAT_ATTACHMENT_FILE_CHARS = 3000
|
|
DOCUMENT_CONTENT_INTENT_PATTERN = re.compile(
|
|
r"\b(read|extract|parse|analy[sz]e|summari[sz]e|review|reference|cite|quote|"
|
|
r"compare|content|from|data|numbers?|metrics?|table|chart|outline|"
|
|
r"presentation|create|generate|build|draft|write|convert)\b|"
|
|
r"\bbased\s+on\b|\baccording\s+to\b|"
|
|
r"\buse\s+(?:the\s+)?(?:attached|provided|this)\s+"
|
|
r"(?:document|pdf|file|attachment)\b",
|
|
re.IGNORECASE,
|
|
)
|
|
DIRECT_FILE_PLACEMENT_PATTERN = re.compile(
|
|
r"\b(place|put|insert|attach|display|show|move|resize|position|replace|add)\b",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ChatTurnResult:
|
|
conversation_id: uuid.UUID
|
|
response_text: str
|
|
tool_calls: list[str]
|
|
|
|
|
|
ChatStreamEventType = Literal["chunk", "complete", "status", "trace"]
|
|
ChatStreamEventValue = str | ChatTurnResult | dict[str, Any]
|
|
|
|
|
|
class PresentationChatService:
|
|
def __init__(
|
|
self,
|
|
sql_session: AsyncSession,
|
|
presentation_id: uuid.UUID,
|
|
conversation_id: uuid.UUID | None,
|
|
chat_mode: ChatToolMode = "presentation",
|
|
):
|
|
self._sql_session = sql_session
|
|
self._presentation_id = presentation_id
|
|
self._conversation_id = conversation_id
|
|
|
|
self._conversation_store = ChatConversationStore(sql_session)
|
|
self._memory = PresentationContextStore(sql_session, presentation_id)
|
|
self._tools = ChatTools(self._memory, mode=chat_mode)
|
|
|
|
async def generate_reply(
|
|
self,
|
|
user_message: str,
|
|
attachments: list[ChatAttachment] | None = None,
|
|
) -> ChatTurnResult:
|
|
self._tools.set_turn_context(user_message)
|
|
conversation_id, messages, persisted_user_message = await self._prepare_turn_context(
|
|
user_message,
|
|
attachments or [],
|
|
)
|
|
response_text, tool_calls = await self._run_llm_with_tools(messages)
|
|
return await self._persist_turn(
|
|
conversation_id=conversation_id,
|
|
user_message=persisted_user_message,
|
|
response_text=response_text,
|
|
tool_calls=tool_calls,
|
|
)
|
|
|
|
async def stream_reply(
|
|
self,
|
|
user_message: str,
|
|
attachments: list[ChatAttachment] | None = None,
|
|
) -> AsyncGenerator[tuple[ChatStreamEventType, ChatStreamEventValue], None]:
|
|
self._tools.set_turn_context(user_message)
|
|
yield "status", "Reading deck context"
|
|
conversation_id, messages, persisted_user_message = await self._prepare_turn_context(
|
|
user_message,
|
|
attachments or [],
|
|
)
|
|
|
|
client = get_client(config=get_llm_config())
|
|
model = get_model()
|
|
tools = build_chat_llm_tools(self._tools.get_tool_definitions())
|
|
|
|
called_tools: list[str] = []
|
|
last_tool_results: list[dict[str, Any]] = []
|
|
response_text: str | None = None
|
|
|
|
for round_index in range(MAX_TOOL_ROUNDS):
|
|
completion_chunk: Any | None = None
|
|
round_content_chunks: list[str] = []
|
|
thinking_chunks: list[str] = []
|
|
|
|
try:
|
|
async for event in stream_generate_events(
|
|
client,
|
|
**get_generate_kwargs(
|
|
model=model,
|
|
messages=messages,
|
|
tools=tools,
|
|
stream=True,
|
|
),
|
|
):
|
|
event_type = getattr(event, "type", None)
|
|
if event_type == "content":
|
|
chunk = getattr(event, "chunk", None)
|
|
if chunk:
|
|
round_content_chunks.append(chunk)
|
|
yield "chunk", chunk
|
|
elif event_type == "thinking":
|
|
thinking_text = self._event_text(event)
|
|
if thinking_text:
|
|
thinking_chunks.append(thinking_text)
|
|
elif event_type == "completion":
|
|
completion_chunk = event
|
|
except Exception as exc:
|
|
raise handle_llm_client_exceptions(exc)
|
|
|
|
thinking_summary = self._summarize_model_note(thinking_chunks)
|
|
if thinking_summary:
|
|
yield "trace", {
|
|
"kind": "model_note",
|
|
"round": round_index + 1,
|
|
"status": "info",
|
|
"message": thinking_summary,
|
|
}
|
|
|
|
completion_tool_calls = list(
|
|
getattr(completion_chunk, "tool_calls", []) or []
|
|
)
|
|
if completion_tool_calls:
|
|
tool_names = [tool_call.name for tool_call in completion_tool_calls]
|
|
called_tools.extend(tool_names)
|
|
yield "trace", {
|
|
"kind": "tool_plan",
|
|
"round": round_index + 1,
|
|
"tools": tool_names,
|
|
"message": f"Using tools: {', '.join(tool_names)}",
|
|
}
|
|
messages = self._append_sanitized_assistant_tool_turn(
|
|
messages,
|
|
content=getattr(completion_chunk, "content", None),
|
|
tool_calls=completion_tool_calls,
|
|
)
|
|
|
|
last_tool_results = []
|
|
for tool_call in completion_tool_calls:
|
|
tool_focus = self._tool_focus_from_arguments(
|
|
tool_name=tool_call.name,
|
|
arguments=tool_call.arguments,
|
|
)
|
|
start_trace: dict[str, Any] = {
|
|
"kind": "tool_call",
|
|
"round": round_index + 1,
|
|
"tool": tool_call.name,
|
|
"status": "start",
|
|
"message": self._tool_start_message(tool_call.name),
|
|
}
|
|
if tool_focus:
|
|
start_trace.update(tool_focus)
|
|
yield "trace", start_trace
|
|
tool_result = await self._tools.execute_tool_call(tool_call)
|
|
last_tool_results.append(tool_result)
|
|
resolved_tool_focus = self._tool_focus_from_result(
|
|
tool_name=tool_call.name,
|
|
tool_result=tool_result,
|
|
)
|
|
complete_trace: dict[str, Any] = {
|
|
"kind": "tool_call",
|
|
"round": round_index + 1,
|
|
"tool": tool_call.name,
|
|
"status": "success" if tool_result.get("ok") else "error",
|
|
"message": self._summarize_tool_result(
|
|
tool_call.name, tool_result
|
|
),
|
|
}
|
|
if resolved_tool_focus:
|
|
complete_trace.update(resolved_tool_focus)
|
|
yield "trace", complete_trace
|
|
tool_response_content = json.dumps(tool_result, ensure_ascii=False)
|
|
messages.append(
|
|
ToolResponseMessage(
|
|
id=tool_call.id,
|
|
content=[TextContentPart(text=tool_response_content)],
|
|
)
|
|
)
|
|
continue
|
|
|
|
response_text = "".join(round_content_chunks)
|
|
if not response_text and completion_chunk:
|
|
response_text = extract_text(getattr(completion_chunk, "content", None))
|
|
if not response_text:
|
|
response_text = "I could not generate a response for that request."
|
|
|
|
if not round_content_chunks:
|
|
yield "chunk", response_text
|
|
break
|
|
else:
|
|
LOGGER.warning("Max tool rounds reached in chat stream flow")
|
|
yield "trace", {
|
|
"kind": "limit",
|
|
"message": (
|
|
"Reached tool-call limit before final answer; "
|
|
"attempting best-effort summary."
|
|
),
|
|
}
|
|
yield "status", "Finalizing response"
|
|
response_text = await self._try_final_response_without_tools(
|
|
client=client,
|
|
model=model,
|
|
messages=messages,
|
|
)
|
|
if not response_text:
|
|
response_text = self._build_tool_limit_fallback(last_tool_results)
|
|
yield "chunk", response_text
|
|
|
|
final_response_text = response_text or "I could not generate a response for that request."
|
|
if response_text is None:
|
|
yield "chunk", final_response_text
|
|
|
|
yield "status", "Saving chat"
|
|
result = await self._persist_turn(
|
|
conversation_id=conversation_id,
|
|
user_message=persisted_user_message,
|
|
response_text=final_response_text,
|
|
tool_calls=called_tools,
|
|
)
|
|
yield "complete", result
|
|
|
|
async def _prepare_turn_context(
|
|
self,
|
|
user_message: str,
|
|
attachments: list[ChatAttachment] | None = None,
|
|
) -> tuple[uuid.UUID, list[Message], str]:
|
|
if not (user_message or "").strip():
|
|
raise HTTPException(status_code=400, detail="Message is required")
|
|
|
|
presentation = await self._sql_session.get(PresentationModel, self._presentation_id)
|
|
if not presentation:
|
|
raise HTTPException(status_code=404, detail="Presentation not found")
|
|
|
|
attachment_context, attachment_memory = await self._build_attachment_context(
|
|
user_message=user_message,
|
|
attachments=attachments or [],
|
|
presentation_language=getattr(presentation, "language", None),
|
|
)
|
|
|
|
conversation_id = await self._conversation_store.ensure_conversation_id(
|
|
self._conversation_id
|
|
)
|
|
history = await self._conversation_store.load_history(
|
|
presentation_id=self._presentation_id,
|
|
conversation_id=conversation_id,
|
|
)
|
|
history_messages = self._convert_history_to_messages(history)
|
|
|
|
normalized_user_message = self._strip_ui_context_prefix(user_message)
|
|
attachment_names = " ".join(
|
|
self._attachment_display_name(attachment) for attachment in attachments or []
|
|
)
|
|
memory_query = " ".join(
|
|
part.strip()
|
|
for part in (normalized_user_message or user_message, attachment_names)
|
|
if part and part.strip()
|
|
)
|
|
presentation_memory = await self._memory.retrieve_context(memory_query)
|
|
chat_memory = await self._conversation_store.retrieve_semantic_context(
|
|
presentation_id=self._presentation_id,
|
|
conversation_id=conversation_id,
|
|
query=memory_query,
|
|
)
|
|
if attachment_memory:
|
|
await MEM0_PRESENTATION_MEMORY_SERVICE.store_generation_context(
|
|
presentation_id=self._presentation_id,
|
|
system_prompt=None,
|
|
user_prompt=None,
|
|
extracted_document_text=attachment_memory,
|
|
source_content=None,
|
|
instructions=None,
|
|
)
|
|
model_user_message = self._compose_user_message_for_model(
|
|
user_message,
|
|
attachment_context,
|
|
)
|
|
messages: list[Message] = [
|
|
SystemMessage(
|
|
content=build_system_prompt(
|
|
presentation_memory_context=presentation_memory,
|
|
chat_memory_context=chat_memory,
|
|
)
|
|
),
|
|
*history_messages,
|
|
UserMessage(content=model_user_message),
|
|
]
|
|
return conversation_id, messages, self._persisted_user_message(
|
|
user_message,
|
|
attachments or [],
|
|
)
|
|
|
|
async def _persist_turn(
|
|
self,
|
|
*,
|
|
conversation_id: uuid.UUID,
|
|
user_message: str,
|
|
response_text: str,
|
|
tool_calls: list[str],
|
|
) -> ChatTurnResult:
|
|
await self._conversation_store.append_turn(
|
|
presentation_id=self._presentation_id,
|
|
conversation_id=conversation_id,
|
|
user_message=self._strip_ui_context_prefix(user_message) or user_message,
|
|
assistant_message=response_text,
|
|
tool_calls=tool_calls,
|
|
)
|
|
await self._sql_session.commit()
|
|
|
|
return ChatTurnResult(
|
|
conversation_id=conversation_id,
|
|
response_text=response_text,
|
|
tool_calls=tool_calls,
|
|
)
|
|
|
|
async def _run_llm_with_tools(self, messages: list[Message]) -> tuple[str, list[str]]:
|
|
client = get_client(config=get_llm_config())
|
|
model = get_model()
|
|
tools = build_chat_llm_tools(self._tools.get_tool_definitions())
|
|
|
|
called_tools: list[str] = []
|
|
last_tool_results: list[dict[str, Any]] = []
|
|
|
|
for _ in range(MAX_TOOL_ROUNDS):
|
|
try:
|
|
response = await asyncio.to_thread(
|
|
client.generate,
|
|
**get_generate_kwargs(
|
|
model=model,
|
|
messages=messages,
|
|
tools=tools,
|
|
),
|
|
)
|
|
except Exception as exc:
|
|
raise handle_llm_client_exceptions(exc)
|
|
|
|
if not response.tool_calls:
|
|
response_text = extract_text(response.content) or (
|
|
"I could not generate a response for that request."
|
|
)
|
|
return response_text, called_tools
|
|
|
|
called_tools.extend([tool_call.name for tool_call in response.tool_calls])
|
|
messages = self._append_sanitized_assistant_tool_turn(
|
|
messages,
|
|
content=getattr(response, "content", None),
|
|
tool_calls=list(response.tool_calls),
|
|
)
|
|
|
|
last_tool_results = []
|
|
for tool_call in response.tool_calls:
|
|
tool_result = await self._tools.execute_tool_call(tool_call)
|
|
last_tool_results.append(tool_result)
|
|
tool_response_content = json.dumps(tool_result, ensure_ascii=False)
|
|
messages.append(
|
|
ToolResponseMessage(
|
|
id=tool_call.id,
|
|
content=[TextContentPart(text=tool_response_content)],
|
|
)
|
|
)
|
|
|
|
LOGGER.warning("Max tool rounds reached in chat flow")
|
|
final_response = await self._try_final_response_without_tools(
|
|
client=client,
|
|
model=model,
|
|
messages=messages,
|
|
)
|
|
if final_response:
|
|
return final_response, called_tools
|
|
|
|
return self._build_tool_limit_fallback(last_tool_results), called_tools
|
|
|
|
async def _try_final_response_without_tools(
|
|
self,
|
|
*,
|
|
client: Any,
|
|
model: str,
|
|
messages: list[Message],
|
|
) -> str | None:
|
|
try:
|
|
response = await asyncio.to_thread(
|
|
client.generate,
|
|
**get_generate_kwargs(
|
|
model=model,
|
|
messages=messages,
|
|
),
|
|
)
|
|
except Exception:
|
|
LOGGER.warning("Final no-tool synthesis call failed", exc_info=True)
|
|
return None
|
|
|
|
return extract_text(response.content)
|
|
|
|
@staticmethod
|
|
def _summarize_model_note(chunks: list[str]) -> str:
|
|
text = "".join(chunks).strip()
|
|
if not text or text in {"{}", "[]"}:
|
|
return ""
|
|
|
|
compact = " ".join(text.split())
|
|
if compact.lower() in {"start", "end"}:
|
|
return ""
|
|
if len(compact) > 600:
|
|
return f"{compact[:600].rstrip()}..."
|
|
return compact
|
|
|
|
@staticmethod
|
|
def _event_text(event: Any) -> str:
|
|
for attr in ("chunk", "delta", "text", "content"):
|
|
value = getattr(event, attr, None)
|
|
if isinstance(value, str):
|
|
return value
|
|
return ""
|
|
|
|
@staticmethod
|
|
def _append_sanitized_assistant_tool_turn(
|
|
messages: list[Message],
|
|
*,
|
|
content: Any,
|
|
tool_calls: list[AssistantToolCall],
|
|
) -> list[Message]:
|
|
response_text = extract_text(content)
|
|
return [
|
|
*messages,
|
|
AssistantMessage(
|
|
content=[response_text] if response_text else None,
|
|
tool_calls=list(tool_calls),
|
|
),
|
|
]
|
|
|
|
@staticmethod
|
|
def _strip_ui_context_prefix(user_message: str) -> str:
|
|
marker = "\nUser message:"
|
|
if not user_message.startswith("UI context:"):
|
|
return user_message
|
|
marker_index = user_message.find(marker)
|
|
if marker_index == -1:
|
|
return user_message
|
|
return user_message[marker_index + len(marker) :].lstrip()
|
|
|
|
@staticmethod
|
|
def _attachment_display_name(attachment: ChatAttachment) -> str:
|
|
name = (attachment.name or "").strip()
|
|
if name:
|
|
return name
|
|
return os.path.basename(attachment.file_path or "").strip() or "attachment"
|
|
|
|
@staticmethod
|
|
def _trim_attachment_text(text: str, limit: int) -> str:
|
|
value = (text or "").strip()
|
|
if len(value) <= limit:
|
|
return value
|
|
return f"{value[:limit].rstrip()}\n[Attachment truncated]"
|
|
|
|
@classmethod
|
|
def _should_parse_attachments(
|
|
cls,
|
|
user_message: str,
|
|
attachments: list[ChatAttachment],
|
|
) -> bool:
|
|
if not attachments:
|
|
return False
|
|
|
|
user_text = cls._strip_ui_context_prefix(user_message).strip()
|
|
if not user_text:
|
|
return True
|
|
if DOCUMENT_CONTENT_INTENT_PATTERN.search(user_text):
|
|
return True
|
|
if DIRECT_FILE_PLACEMENT_PATTERN.search(user_text):
|
|
return False
|
|
return True
|
|
|
|
@classmethod
|
|
def _compose_user_message_for_model(
|
|
cls,
|
|
user_message: str,
|
|
attachment_context: str,
|
|
) -> str:
|
|
if not attachment_context:
|
|
return user_message
|
|
|
|
marker = "\nUser message:"
|
|
if user_message.startswith("UI context:"):
|
|
marker_index = user_message.find(marker)
|
|
if marker_index != -1:
|
|
return (
|
|
f"{user_message[:marker_index].rstrip()}\n"
|
|
f"{attachment_context}\n"
|
|
f"{user_message[marker_index:].lstrip()}"
|
|
)
|
|
|
|
return f"{attachment_context}\n\nUser message: {user_message}"
|
|
|
|
@classmethod
|
|
def _persisted_user_message(
|
|
cls,
|
|
user_message: str,
|
|
attachments: list[ChatAttachment],
|
|
) -> str:
|
|
display_message = cls._strip_ui_context_prefix(user_message) or user_message
|
|
if not attachments:
|
|
return display_message
|
|
|
|
attachment_list = ", ".join(
|
|
cls._attachment_display_name(attachment) for attachment in attachments
|
|
)
|
|
return f"{display_message}\n\nAttached files: {attachment_list}"
|
|
|
|
async def _build_attachment_context(
|
|
self,
|
|
*,
|
|
user_message: str,
|
|
attachments: list[ChatAttachment],
|
|
presentation_language: str | None,
|
|
) -> tuple[str, str]:
|
|
if not attachments:
|
|
return "", ""
|
|
|
|
should_parse = self._should_parse_attachments(user_message, attachments)
|
|
names = [
|
|
f"Document {index + 1}: {self._attachment_display_name(attachment)}"
|
|
for index, attachment in enumerate(attachments)
|
|
]
|
|
|
|
if not should_parse:
|
|
context = "\n".join(
|
|
[
|
|
(
|
|
"UI context: the user attached document file(s), but this "
|
|
"request appears to place or reference the file rather than "
|
|
"read its contents. Use only the file names unless the user "
|
|
"asks to read, extract, summarize, or build from them."
|
|
),
|
|
*names,
|
|
]
|
|
)
|
|
return context, ""
|
|
|
|
temp_dir = TEMP_FILE_SERVICE.create_temp_dir(str(uuid.uuid4()))
|
|
loader = DocumentsLoader(
|
|
file_paths=[attachment.file_path for attachment in attachments],
|
|
presentation_language=presentation_language,
|
|
)
|
|
await loader.load_documents(temp_dir=temp_dir)
|
|
|
|
context_lines = [
|
|
(
|
|
"UI context: the user attached parsed document(s) to this chat "
|
|
"request. Use this content when the user asks to read, extract, "
|
|
"summarize, or build slide/chart/data content from attachments."
|
|
)
|
|
]
|
|
memory_lines: list[str] = []
|
|
remaining_chars = MAX_CHAT_ATTACHMENT_CONTEXT_CHARS
|
|
|
|
for index, attachment in enumerate(attachments):
|
|
if remaining_chars <= 0:
|
|
context_lines.append("[Additional attachment content omitted]")
|
|
break
|
|
|
|
name = self._attachment_display_name(attachment)
|
|
parsed = loader.documents[index] if index < len(loader.documents) else ""
|
|
file_limit = min(MAX_CHAT_ATTACHMENT_FILE_CHARS, remaining_chars)
|
|
trimmed = self._trim_attachment_text(parsed, file_limit)
|
|
context_lines.append(f"Document {index + 1} ({name}):\n{trimmed}")
|
|
memory_lines.append(f"Document {index + 1} ({name}):\n{trimmed}")
|
|
remaining_chars -= len(trimmed)
|
|
|
|
return "\n".join(context_lines), "\n\n".join(memory_lines).strip()
|
|
|
|
@staticmethod
|
|
def _tool_focus_from_arguments(
|
|
*,
|
|
tool_name: str,
|
|
arguments: str | None,
|
|
) -> dict[str, Any] | None:
|
|
if tool_name not in {
|
|
"getSlideAtIndex",
|
|
"addNewSlide",
|
|
"addNewSlideLayout",
|
|
"saveSlide",
|
|
"updateSlide",
|
|
"deleteSlide",
|
|
"addElement",
|
|
"updateElement",
|
|
"deleteElement",
|
|
"addComponent",
|
|
"createComponent",
|
|
"updateComponent",
|
|
"deleteComponent",
|
|
}:
|
|
return None
|
|
|
|
parsed_args: dict[str, Any]
|
|
try:
|
|
parsed_args = dirtyjson.loads(arguments or "{}")
|
|
except Exception:
|
|
try:
|
|
parsed_args = json.loads(arguments or "{}")
|
|
except Exception:
|
|
return None
|
|
if not isinstance(parsed_args, dict):
|
|
return None
|
|
|
|
focus_payload: dict[str, Any] = {}
|
|
index = parsed_args.get("index")
|
|
if isinstance(index, int):
|
|
normalized_index = max(0, index)
|
|
focus_payload["slide_index"] = normalized_index
|
|
focus_payload["slide_number"] = normalized_index + 1
|
|
|
|
component_id = parsed_args.get("componentId") or parsed_args.get("component_id")
|
|
if isinstance(component_id, str) and component_id:
|
|
focus_payload["component_id"] = component_id
|
|
element_path = parsed_args.get("elementPath") or parsed_args.get("element_path")
|
|
if isinstance(element_path, str) and element_path:
|
|
focus_payload["element_path"] = element_path
|
|
|
|
target_slide_indices = PresentationChatService._extract_target_slide_indices(
|
|
parsed_args
|
|
)
|
|
if target_slide_indices:
|
|
focus_payload["target_slide_indices"] = target_slide_indices
|
|
focus_payload["target_slide_numbers"] = [
|
|
index + 1 for index in target_slide_indices
|
|
]
|
|
|
|
return focus_payload or None
|
|
|
|
@staticmethod
|
|
def _tool_focus_from_result(
|
|
*,
|
|
tool_name: str,
|
|
tool_result: dict[str, Any],
|
|
) -> dict[str, Any] | None:
|
|
if tool_name not in {
|
|
"getSlideAtIndex",
|
|
"addNewSlide",
|
|
"addNewSlideLayout",
|
|
"saveSlide",
|
|
"updateSlide",
|
|
"deleteSlide",
|
|
"addElement",
|
|
"updateElement",
|
|
"deleteElement",
|
|
"addComponent",
|
|
"createComponent",
|
|
"updateComponent",
|
|
"deleteComponent",
|
|
}:
|
|
return None
|
|
if not tool_result.get("ok"):
|
|
return None
|
|
|
|
result = tool_result.get("result")
|
|
if not isinstance(result, dict):
|
|
return None
|
|
|
|
focus_payload: dict[str, Any] = {}
|
|
index: int | None = None
|
|
resolved_index = result.get("resolved_index")
|
|
if isinstance(resolved_index, int):
|
|
index = resolved_index
|
|
else:
|
|
direct_index = result.get("index")
|
|
if isinstance(direct_index, int):
|
|
index = direct_index
|
|
else:
|
|
slide = result.get("slide")
|
|
if isinstance(slide, dict) and isinstance(slide.get("index"), int):
|
|
index = slide["index"]
|
|
|
|
if index is not None:
|
|
normalized_index = max(0, index)
|
|
focus_payload["slide_index"] = normalized_index
|
|
focus_payload["slide_number"] = normalized_index + 1
|
|
|
|
component_id = result.get("component_id")
|
|
if isinstance(component_id, str) and component_id:
|
|
focus_payload["component_id"] = component_id
|
|
element_path = result.get("element_path")
|
|
if isinstance(element_path, str) and element_path:
|
|
focus_payload["element_path"] = element_path
|
|
|
|
target_slide_indices = PresentationChatService._extract_target_slide_indices(
|
|
result
|
|
)
|
|
if target_slide_indices:
|
|
focus_payload["target_slide_indices"] = target_slide_indices
|
|
focus_payload["target_slide_numbers"] = [
|
|
index + 1 for index in target_slide_indices
|
|
]
|
|
|
|
return focus_payload or None
|
|
|
|
@staticmethod
|
|
def _extract_target_slide_indices(payload: dict[str, Any]) -> list[int]:
|
|
raw_candidates = []
|
|
for key in (
|
|
"target_slide_indices",
|
|
"targetSlideIndices",
|
|
"target_indices",
|
|
"targetIndices",
|
|
"slide_indices",
|
|
"slideIndices",
|
|
"indices",
|
|
):
|
|
value = payload.get(key)
|
|
if isinstance(value, list):
|
|
raw_candidates.extend(value)
|
|
|
|
normalized_indices: list[int] = []
|
|
seen_indices: set[int] = set()
|
|
for candidate in raw_candidates:
|
|
if not isinstance(candidate, int):
|
|
continue
|
|
normalized_index = max(0, candidate)
|
|
if normalized_index in seen_indices:
|
|
continue
|
|
seen_indices.add(normalized_index)
|
|
normalized_indices.append(normalized_index)
|
|
return normalized_indices
|
|
|
|
@staticmethod
|
|
def _tool_start_message(tool_name: str) -> str:
|
|
labels = {
|
|
"addOutline": "Adding an outline slide",
|
|
"updateOutline": "Updating the outline slide",
|
|
"deleteOutline": "Deleting the outline slide",
|
|
"addNewSlide": "Adding a blank slide",
|
|
"addNewSlideLayout": "Adding slide from layout",
|
|
"getTemplateSummary": "Reading template summary",
|
|
"readSourceDocuments": "Reading source documents",
|
|
"searchSlide": "Searching relevant slides",
|
|
"getSlideAtIndex": "Opening the requested slide",
|
|
"getAvailableLayouts": "Checking available layouts",
|
|
"getContentSchemaFromLayoutId": "Reading layout content schema",
|
|
"generateAssets": "Generating slide assets",
|
|
"saveSlide": "Saving the slide",
|
|
"updateSlide": "Updating the slide",
|
|
"deleteSlide": "Deleting the slide",
|
|
"addElement": "Adding slide element",
|
|
"updateElement": "Updating slide element",
|
|
"deleteElement": "Removing slide element",
|
|
"addComponent": "Adding slide component",
|
|
"createComponent": "Creating slide component",
|
|
"updateComponent": "Updating slide component",
|
|
"deleteComponent": "Removing slide component",
|
|
"getPresentationTheme": "Checking available themes",
|
|
"setPresentationTheme": "Applying presentation theme",
|
|
}
|
|
return labels.get(tool_name, f"Running {tool_name}")
|
|
|
|
@staticmethod
|
|
def _build_tool_limit_fallback(last_tool_results: list[dict[str, Any]]) -> str:
|
|
for entry in reversed(last_tool_results):
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
if not entry.get("ok"):
|
|
continue
|
|
result = entry.get("result")
|
|
if not isinstance(result, dict):
|
|
continue
|
|
message = result.get("message")
|
|
if isinstance(message, str) and message.strip():
|
|
return message.strip()
|
|
|
|
return (
|
|
"I completed several tool operations but could not finalize the response "
|
|
"within the tool limit. Please ask a follow-up and I will continue."
|
|
)
|
|
|
|
@staticmethod
|
|
def _summarize_tool_result(tool_name: str, tool_result: dict[str, Any]) -> str:
|
|
if not tool_result.get("ok"):
|
|
error = tool_result.get("error")
|
|
if isinstance(error, str) and error.strip():
|
|
recovery = tool_result.get("recovery")
|
|
if isinstance(recovery, dict):
|
|
guidance = recovery.get("guidance")
|
|
if isinstance(guidance, list) and guidance:
|
|
first_guidance = str(guidance[0]).strip()
|
|
if first_guidance:
|
|
return (
|
|
f"{tool_name} failed: {error.strip()} "
|
|
f"Recovery: {first_guidance}"
|
|
)
|
|
return f"{tool_name} failed: {error.strip()}"
|
|
return f"{tool_name} failed."
|
|
|
|
result = tool_result.get("result")
|
|
if isinstance(result, dict):
|
|
message = result.get("message")
|
|
if isinstance(message, str) and message.strip():
|
|
return message.strip()
|
|
|
|
note = result.get("note")
|
|
if isinstance(note, str) and note.strip():
|
|
return note.strip()
|
|
|
|
count = result.get("count")
|
|
if isinstance(count, int):
|
|
return f"{tool_name} returned {count} result(s)."
|
|
|
|
found = result.get("found")
|
|
if isinstance(found, bool):
|
|
return (
|
|
f"{tool_name} found requested data."
|
|
if found
|
|
else f"{tool_name} did not find matching data."
|
|
)
|
|
|
|
return f"{tool_name} completed."
|
|
|
|
@staticmethod
|
|
def _convert_history_to_messages(history: list[dict[str, str]]) -> list[Message]:
|
|
messages: list[Message] = []
|
|
for item in history:
|
|
role = item.get("role")
|
|
content = item.get("content")
|
|
if not content:
|
|
continue
|
|
if role == "user":
|
|
content = PresentationChatService._strip_ui_context_prefix(content)
|
|
messages.append(UserMessage(content=content))
|
|
elif role == "assistant":
|
|
messages.append(AssistantMessage(content=[content]))
|
|
return messages
|