""" Определяет логику рабочих процессов агента, используя LangGraph для управления состояниями и переходами между узлами обработки. """ from typing import Dict, Any, Optional, Set from langgraph.graph import StateGraph, END from graph_history_manager import GraphHistoryManager from models import AgentState from nodes import parse_command_node, execute_command_node, call_llm_node, generate_images_node, analyze_image_node, get_meet_subtitles_node, get_teams_subtitles_node, summarize_history_node, handle_error_node, help_node import uuid # Добавлено для генерации UUID from llm_client import DEFAULT_TEMPERATURE, get_llm import os import base64 from pathlib import Path # --- LangGraph: Построение графа --- graph_history_manager = GraphHistoryManager() workflow = StateGraph(AgentState) # Добавляем узлы workflow.add_node("parse_command", parse_command_node) workflow.add_node("execute_command", execute_command_node) # Это по сути роутер workflow.add_node("call_llm", call_llm_node) workflow.add_node("generate_images", generate_images_node) workflow.add_node("analyze_image", analyze_image_node) workflow.add_node("get_meet_subtitles", get_meet_subtitles_node) workflow.add_node("get_teams_subtitles", get_teams_subtitles_node) workflow.add_node("summarize_history", summarize_history_node) workflow.add_node("handle_error", handle_error_node) workflow.add_node("help", help_node) # Устанавливаем точку входа workflow.set_entry_point("parse_command") # Определяем маршрутизацию после parse_command def route_command(state: AgentState) -> str: """Маршрутизирует выполнение на основе распознанной команды.""" if state.error: return "handle_error" # Если была ошибка парсинга команды elif state.command == "imagine": return "generate_images" elif state.command == "analyze": return "analyze_image" elif state.command == "subtitles_meet": return "get_meet_subtitles" elif state.command == "subtitles_teams": return "get_teams_subtitles" elif state.command == "summarize": return "summarize_history" elif state.command == "help": return "help" elif state.command == "chat": return "call_llm" elif state.command == "error": # Команда была, но не известная return "handle_error" else: # Fallback на LLM, если команда не распознана или пуста (хотя parse_command должен был бы ее поймать) return "call_llm" workflow.add_conditional_edges( "parse_command", # Откуда route_command, # Функция, определяющая куда идти { "generate_images": "generate_images", "analyze_image": "analyze_image", "get_meet_subtitles": "get_meet_subtitles", "get_teams_subtitles": "get_teams_subtitles", "summarize_history": "summarize_history", "help": "help", "call_llm": "call_llm", "handle_error": "handle_error", # Для ошибок, найденных в parse_command }, ) # Все узлы операций (кроме ошибок) ведут к END после завершения workflow.add_edge("call_llm", END) workflow.add_edge("generate_images", END) workflow.add_edge("analyze_image", END) workflow.add_edge("get_meet_subtitles", END) workflow.add_edge("get_teams_subtitles", END) workflow.add_edge("summarize_history", END) workflow.add_edge("help", END) workflow.add_edge("handle_error", END) # Ошибка - это тоже конечный пункт для текущего запроса # Компилируем граф app = workflow.compile() def run_agent_streaming(graph_id: str, user_node_id: str, assistant_node_id: str, system_prompt: Optional[str] = None, model: Optional[str] = None, cache_folder: Optional[str] = None, temperature: float = DEFAULT_TEMPERATURE, max_tokens: Optional[int] = None): """Генератор для стримингового ответа LLM с поддержкой вложений.""" try: loaded_graph_data = graph_history_manager.get_graph( graph_id, target_node_id=user_node_id) if not loaded_graph_data: yield {"type": "error", "error": "Граф не найден."} return messages = loaded_graph_data.get("messages", []) from langchain_core.messages import HumanMessage, SystemMessage, AIMessage messages_for_llm = [] if system_prompt: messages_for_llm.append(SystemMessage(content=system_prompt)) # Обрабатываем вложения for i, msg in enumerate(messages): is_last = (i == len(messages) - 1) if msg["role"] == "user": text_content = msg["content"] attachments = msg.get("attachments", []) # Проверяем, есть ли мультимодальный контент has_multimodal = False content_parts = [{"type": "text", "text": text_content}] # Обрабатываем вложения if attachments: for att in attachments: # Раскрываем только permanent или вложения последнего сообщения if att.get("permanent") or is_last: attachment_content = prepare_attachment_for_llm( att, cache_folder) # Если это изображение, то это мультимодальный контент if attachment_content.get("type") == "image_url": has_multimodal = True content_parts.append(attachment_content) # Формируем финальный контент if has_multimodal: # Мультимодальное сообщение messages_for_llm.append( HumanMessage(content=content_parts)) else: # Текстовое сообщение (объединяем все текстовые части) full_text = text_content for part in content_parts[ 1:]: # Пропускаем первую часть (основной текст) if part.get("type") == "text": full_text += part.get("text", "") messages_for_llm.append(HumanMessage(content=full_text)) elif msg["role"] == "assistant": messages_for_llm.append(AIMessage(content=msg["content"])) # Стриминг ответа llm = get_llm(model or "gemini-2.5-flash") for chunk in llm.stream2(messages_for_llm, temperature=temperature, max_tokens=max_tokens): content = chunk.content if hasattr(chunk, 'content') else str(chunk) yield {"type": "chunk", "content": content} except Exception as e: print(f"Ошибка при стриминге: {e}") import traceback traceback.print_exc() yield {"type": "error", "error": str(e)} def prepare_attachment_for_llm(attachment: Dict[str, Any], cache_folder: str) -> Dict[str, Any]: """Подготавливает вложение для отправки в LLM.""" mime_type = attachment.get("mimeType", "application/octet-stream") cached_path = attachment.get("cachedPath") att_cache_folder = attachment.get("cacheFolder", cache_folder) # ИЗМЕНЕНО source_path = attachment.get("sourcePath") att_type = attachment.get("type") name = attachment.get("name") # Получаем полный путь к кешированному файлу full_cached_path = os.path.join(att_cache_folder, cached_path) # Проверяем существование кеша if not os.path.exists(full_cached_path): print(f"❌ Кеш не найден: {full_cached_path}") return {"type": "text", "text": f"\n[Файл {name} недоступен]\n"} # Проверяем тип файла if mime_type.startswith('image/'): # Для изображений отправляем как base64 with open(full_cached_path, 'rb') as f: image_data = base64.b64encode(f.read()).decode('utf-8') return { "type": "image_url", "image_url": { "url": f"data:{mime_type};base64,{image_data}" } } elif mime_type.startswith('text/') or \ mime_type in ['application/json', 'application/xml'] or \ mime_type.startswith('application/x-') or \ mime_type == 'application/octet-stream': # TODO: Спорно, но пока вводим - фронтенд не смог сопоставить расширению mime_type # неразрывно связано с методом getMimeType() на фронтенде # Для текстовых файлов (или потенциально текстовых с неизвестным типом) try: with open(full_cached_path, 'r', encoding='utf-8') as f: content = f.read() except UnicodeDecodeError: # Если не получилось прочитать как текст return { "type": "text", "text": f"\n[Файл {name} не может быть прочитан как текст]\n" } att_type_ru = "промпта" if att_type == "prompt" else "файла" header = f"\n\n--- Содержимое {att_type_ru}: {name} ---\n" footer = f"\n--- Конец {att_type_ru}: {name} ---\n\n" return {"type": "text", "text": header + content.strip() + footer} else: # Для других типов файлов возвращаем информацию return { "type": "text", "text": f"\n[Файл {name} ({mime_type}) не может быть отображен в чате]\n" } def restore_cache_from_source(source_path: str, att_type: str, cache_folder: str, cached_path: str) -> str: """Восстанавливает кеш из исходного файла.""" full_cached_path = os.path.join(cache_folder, cached_path) # Создаем папку кеша, если не существует os.makedirs(cache_folder, exist_ok=True) if att_type == 'external': # Для внешних файлов source_path - абсолютный if not os.path.exists(source_path): raise FileNotFoundError(f"Исходный файл не найден: {source_path}") import shutil shutil.copy2(source_path, full_cached_path) else: # Для внутренних файлов и промптов нужно получить полный путь # Это требует доступа к vault, что на бэкенде недоступно # Поэтому это нужно делать на фронтенде raise NotImplementedError( "Восстановление кеша для внутренних файлов должно происходить на фронтенде" ) return full_cached_path def format_attachment_for_llm(attachment: Dict[str, Any]) -> str: """Форматирует вложение для вставки в сообщение LLM.""" att_type = "промпта" if attachment["type"] == "prompt" else "файла" header = f"\n\n--- Содержимое {att_type}: {attachment['name']} ---\n" footer = f"\n--- Конец {att_type}: {attachment['name']} ---\n\n" return header + attachment.get("content", "").strip() + footer