项目文件夹

文件
2026-07-13 13:08:41 +08:00

875 行
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