dimitriev.gs/llm-agent-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
dimitrievgs 2025-09-12 00:26:19 +03:00
commit d27156d152
14 changed files with 1776 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.gitignore vendored Normal file
View File

@ -0,0 +1,6 @@
venv/
env/
node_modules/
__pycache__

134
Target.md Normal file
View File

@ -0,0 +1,134 @@
https://www.perplexity.ai/search/sozdai-requiremmnts-txt-rkA2KeVJR1y0gtlrZpwecA
---
доработай чтобы удовлетворить всем требованиям
`какие есть открытые агенты, доступные на декстопе вин 10, к которым можно подсоединить как внешние (типа gemini по api) так и внутренние (развернутые в ollama) llm, и чтобы можно было задавать кастомные промпты, типо сделать запрос через curl и получить ссылки на изображения внутри json взять их их и вывести все 4 в чате чтобы визульано было видно что получено, сделать запрос по полученной ссылке на изображения, проанализировать изображение и вывести промпт чтобы сгенерировтаь подобное - а промпт например подать запрос по удаленному эндпоинту. или запросить в некой папке got --diff staged и создать сообщение на основе этого для коммита. или например аналзировать формируемый где-то субтитры видеозвонка и на сонове того брать последние 50000 символов и пропускать через промпт выдавая на выходе предложения по дальнейшему развитию игры в днд`. если нужно задай вначале все необходимые вопросы и не приступай к генерации пока не получишь всю нужную информацию. Если нужно а я не даю, продолжай требовать её
---
1. Ollama не будет использоваться, только внешние модели пока что `MODELS = {
"gemini-2.0-flash": {
"name": "gemini-2.0-flash",
"provider": "openai",
"baseUrl": "https://render-service-gsu7.onrender.com/g/v1beta",
"apiKey": "AIzaSyDpueKFWVqknVKlQn6TdasLmJ2lvAUiBik",
"stream": True,
"capabilities": ["vision"],
},
"gemini-2.5-flash": {
"name": "gemini-2.5-flash",
"provider": "openai",
"baseUrl": "https://render-service-gsu7.onrender.com/g/v1beta",
"apiKey": "AIzaSyDpueKFWVqknVKlQn6TdasLmJ2lvAUiBik",
"stream": True,
"capabilities": ["vision"],
},
"gemini-2.5-flash-preview-05-20": {
"name": "gemini-2.5-flash-preview-05-20",
"provider": "openai",
"baseUrl": "https://render-service-gsu7.onrender.com/g/v1beta",
"apiKey": "AIzaSyDpueKFWVqknVKlQn6TdasLmJ2lvAUiBik",
"stream": True,
"capabilities": ["vision"],
},
"mistral-small-latest": {
"name": "mistral-small-latest",
"provider": "mistralai",
"baseUrl": "https://render-service-gsu7.onrender.com/m",
"apiKey": "Q0m29fvxBY0Cfdj4sjHaKqccy1NjonLW",
"stream": True,
"capabilities": ["vision"],
},
}`, заложи заглушку, которую потом можно развить под ollama (типо пустой словарь соединений с моделями).
1. Веб интерфейс не слабее chatgpt или perlexity, с историями диалогов, с о средней панелью (левая история, правая чат, а средняя - граф) с графом как на flowith, сделай его через react flow если нет лучше альтернативы. суть в том, что активный чат - это выбранная ветка на графе, и новое сообщение в чате порождает новый лист в графе, а граф перестраивается с использованием библиотеки dagre. активный нод в графе - ыбранный последнее сообщение в правой панели, например, если выбрать промужточное на графе (щелкнуть лкм по ноду), то сообщения в правой панели выведутся от корня графа вплоть до щелкнутого сообщения. и если спроить, то будет добален ноду новый ребенок и дальше новое ветвления с постоянным обновлением посдвечиваемого синими цветом активного листа согласно последнему сообщению(сообщения пользователя и llm разными цветми контура нода). я пробовал streamlit и он показался оцень примитивным, возможно, я не прав.
Соответсвтенно хранятся одержимое нодов графа в чем-то и при отправке сообщения нужно использовать память (формировать её?) но не в лоб как в дипсик, чтобы она глупо конкатенировалась, а суммаризацией, возможно либо с отсечкой по кол-ву токенов последних - предложи сначала варианты, какие есть. т.е. для новой ветки формируемый промпт будет разным (отправляемыая история).
1. Нужно бесплатное решение - предложи варианты. и то, что будет работать на gtx 1060 ti .
2. подробное описание и стиль как минимум. предложи, что ещё?
3. сначала предложи вариант платформы для звонка, (варианты), которые позволяю непрерывно извлекать уже сформированные на данный момент звонка субтитры и передавать в систему (промежуточные субтитры).
4. необходимость запустить код в obsidiab ````dataviewjs
const defaultlocations = ["Фаэрун"];
const defaultCurrentDate = "1491/09/29";
const defaultStartDate = "1490/01/01";
const defaultEndDate = "1493/3/30";
const defaultPcsParticipationOnly = false;
const defaultIncludeRegionsParticipants = false;
const fullWidth = true;
const scriptPath = "scripts/timeline-widget";
await dv.view(scriptPath, { arg1: dv, arg2: defaultlocations, arg3: defaultCurrentDate, arg4: defaultStartDate, arg5: defaultEndDate, arg6: defaultPcsParticipationOnly, arg7: defaultIncludeRegionsParticipants, arg8: fullWidth });
```` и считать результат. ВОзможность собирать данные из obsidian.
Также этот эндпоинт на удаленном ресурсе (остаь пока там произвольную константу для url), который имеет этот ресурс для получения изображения по промпту : `def _process_image_generation(prompt: str):
"""
Вспомогательная функция для обработки логики генерации изображений.
Возвращает кортеж (image_urls, error_message).
"""
image_urls = []
error_message = None
# При каждой генерации изображений создаем новый объект BingArt
# с помощью функции load_cookies, чтобы убедиться, что куки актуальны.
# Это позволяет избежать проблем с устаревшими сессиями.
# Важно: close_session() вызывается в finally, чтобы гарантировать закрытие сессии.
local_bing_art = BingArt(
auth_cookie_U=cookies.get('_U'),
auth_cookie_KievRPSSecAuth=cookies.get('KievRPSSecAuth'))
try:
if prompt:
results = local_bing_art.generate_images(prompt)
if results and 'images' in results:
# Фильтруем и обрезаем до 4 изображений
image_urls = [
img['url'] for i, img in enumerate(results['images'])
if i % 2 == 0
][:4]
else:
error_message = "Не удалось получить изображения."
else:
error_message = "Коммандер, введи промпт!"
except Exception as e:
error_message = f"Произошла ошибка: {str(e)}"
app.logger.error(f"Ошибка при генерации изображения: {e}",
exc_info=True)
finally:
# Важно: закрываем сессию BingArt после использования
local_bing_art.close_session()
return image_urls, error_message
@app.route('/generate-images', methods=['GET', 'POST'])
def generate_images():
"""
Генерация изображений с использованием bingart.
Поддерживает как POST (form data), так и GET (query parameter) запросы.
"""
if request.method == 'POST':
prompt = request.form.get('prompt')
else: # GET request
prompt = request.args.get('prompt')
session['prompt'] = prompt # Сохраняем промпт в сессии
image_urls, error_message = _process_image_generation(prompt)
if error_message:
session['error_message'] = error_message
return jsonify({'error': error_message}), 500
session['image_urls'] = image_urls
return jsonify({'image_urls': image_urls})`
7. Думаю, они должны предлагаться как команды по слешу, но эти команды должны предлагаться исходя из контекста. но также должны быть автодополнение - когда слеш начинаешь писать, и предлагается автодополнение. Предлоижи сначала варианты.
храниться они будут в отдельной папке в подпапках, а при начале письма подставляются в виде `<имя папки>/<ввод пользователя><вариант продолжения N>`
и все варианты должны предлагаться в некотором popup-окне или лучше.
8. Русский.
---

217
Target2.md Normal file
View File

@ -0,0 +1,217 @@
Для реализации вашего комплексного агента с учётом всех требований предлагаю следующий план и архитектуру. Ниже — ключевые моменты, архитектурные решения и рекомендации, а в конце — пример кода для базового LLM-клиента с заглушкой под Ollama и интеграцией Gemini через LangChain.
## 1. Поддержка только внешних моделей (Gemini и Mistral), заглушка под Ollama
- В `MODELS` оставляем только внешние модели, как вы указали.
- Создаём пустой словарь `LOCAL_MODELS = {}` для будущей интеграции Ollama.
- Для Gemini используем **официальный LangChain-пакет `langchain-google-genai`** (рекомендуется, т.к. поддерживает чат, vision, структурированные ответы).
- Для Mistral — пока простой HTTP-клиент (зависит от API).
## 2. Веб-интерфейс (React + React Flow + Dagre)
- Три панели:
- Левая — история диалогов (ветки графа)
- Средняя — визуализация графа сообщений (React Flow + Dagre для layout)
- Правая — чат с выбранной веткой (последовательность сообщений от корня до выбранного узла)
- Активный чат — выбранный лист графа. Новое сообщение добавляет новый лист (ребёнок) в граф.
- При клике на узел графа показываем в чате цепочку сообщений от корня до этого узла.
- Ноды разного цвета: пользователь — один цвет, LLM — другой.
- Для хранения истории и графа — локальное хранилище (IndexedDB) или backend (например, FastAPI + БД).
- Для суммаризации истории и формирования промптов:
- Использовать **суммаризацию через LLM** (например, Gemini) — для сжатия истории до лимита токенов.
- Альтернативно — скользящее окно последних сообщений с ограничением по токенам.
- Можно комбинировать: последние N сообщений + суммаризация более старых.
## 3. Бесплатное решение и работа на GTX 1060 Ti
- Для LLM — использовать только внешние API (Gemini, Mistral), чтобы не нагружать локальную видеокарту.
- Для визуализации — React + React Flow — лёгкие и бесплатные.
- Для анализа изображений — использовать возможности Gemini Vision или внешние бесплатные API (например, CLIP + локальный сервер, если потребуется).
- Для git — локальные команды через subprocess (git diff) — бесплатно.
- Для субтитров — интеграция с выбранной платформой звонков (см. ниже).
## 4. Описание и стиль
- Подробные комментарии в коде.
- Документация API.
- UI с подсказками и автодополнением.
- Локализация на русский.
## 5. Платформа для видеозвонков с субтитрами
Варианты:
| Платформа | Особенности субтитров | API для извлечения субтитров |
|-------------------|-----------------------------------------------|-----------------------------------------------|
| Zoom | Автоматические субтитры, можно получать через API | Есть Webhooks и API для получения транскриптов |
| Microsoft Teams | Автоматические субтитры, API ограничены | Можно использовать Graph API для записи |
| Google Meet | Субтитры в реальном времени, API нет | Можно использовать сторонние решения для захвата |
| Jitsi | Открытый исходный код, можно интегрировать субтитры | Можно реализовать собственный плагин |
Рекомендация: **Zoom** — наиболее зрелая платформа с API для субтитров.
## 6. Интеграция с Obsidian
- Можно запускать `dataviewjs` скрипты через Obsidian API.
- Для интеграции в агент — сделать отдельный узел, который запускает скрипт и получает результат.
- Для удалённого эндпоинта генерации изображений — сделать HTTP-клиент, который отправляет промпт и получает JSON с 4 ссылками.
## 7. Команды с автодополнением и контекстом
- Ввод команд через слеш `/` в поле ввода.
- Автодополнение реализуется через локальный индекс команд (хранить в папках).
- При вводе `/` показывать popup с командами, фильтровать по введённому тексту.
- Команды могут подставляться в виде `/`.
- Можно использовать библиотеку для автокомплита (например, React Autosuggest).
## 8. Язык — русский
# Резюме и дальнейшие шаги
- Для веб-интерфейса рекомендую React + React Flow + Dagre. Можно использовать готовые шаблоны и расширять их.
- Для суммаризации истории — использовать Gemini для генерации кратких резюме, комбинировать с оконным ограничением по токенам.
- Для субтитров — рекомендую рассмотреть Zoom API или Jitsi с кастомной интеграцией.
- Для git diff — использовать Python `subprocess` с `git diff --staged`.
- Для автодополнения команд — React Autosuggest или аналогичные библиотеки.
- Для генерации и анализа изображений — использовать Gemini Vision API и ваш удалённый эндпоинт с bingart.
- Вся логика с графом сообщений и ветвлениями — на фронтенде и backend (FastAPI или Node.js).
Если хотите, могу помочь с примером React-приложения для визуализации графа и чата или с backend для интеграции git и субтитров.
Жду ваших комментариев или уточнений для следующего шага!
---
1. ОК
2. хранение через indexed db.
3.давай комбинацию
gemini vision - ок пока, добавь заглушку для расширения.
Пока для Google Meet и Microsoft TEams.
7. React Autosuggest - годится
---
Ваши уточнения приняты, предлагаю архитектуру и ключевые технические решения для вашего агента, а также примерный план реализации фронтенда с React Flow + Dagre и IndexedDB, и backend-структуру.
## Архитектура и ключевые решения
| Требование | Решение / Технология |
|-----------------------------------|-----------------------------------------------------------------------------------------------------------|
| **LLM** | Внешние модели Gemini и Mistral через API, заглушка под Ollama (пока пустая). |
| **Веб-интерфейс** | React + React Flow для графа + Dagre для автоматического layout, React Autosuggest для автодополнения. |
| **Хранение истории и графа** | IndexedDB (через библиотеку `idb` или `dexie.js`) для локального хранения диалогов и графа. |
| **Визуализация графа** | React Flow с динамическим пересчётом layout через Dagre (пример из [1][2][4]). |
| **Чат с историей и ветвлениями** | Правая панель — чат, показывающий цепочку сообщений от корня до выбранного узла графа. |
| **Суммаризация истории** | Комбинация: последние N сообщений + вызов Gemini для суммаризации старой истории (с учётом лимита токенов).|
| **Анализ изображений** | Использовать Gemini Vision API, заглушка для расширения. |
| **Платформы для субтитров** | Google Meet и Microsoft Teams (API для получения транскриптов, интеграция через polling или webhook). |
| **Git интеграция** | Локальный запуск `git diff --staged` через Python subprocess, формирование сообщения коммита. |
| **Генерация изображений** | Вызов удалённого эндпоинта с bingart (пример функции из вашего запроса). |
| **Команды с автодополнением** | React Autosuggest, команды хранятся в IndexedDB, показываются в popup при вводе `/`. |
| **Язык интерфейса** | Русский. |
## Краткое описание React Flow + Dagre для графа
- Используйте пример из [1], где через dagre вычисляется layout для узлов и рёбер.
- При добавлении нового сообщения создаётся новый узел и ребро к предыдущему.
- При клике на узел — обновляется чат справа, показывая цепочку сообщений от корня до выбранного узла.
- Используйте разные цвета и стили для нодов пользователя и LLM.
- Для автодополнения команд — React Autosuggest, данные команд загружать из IndexedDB.
## IndexedDB для хранения
- Используйте `dexie.js` или `idb` для удобной работы с IndexedDB.
- Структура БД:
- Таблица `dialogs` — id, название, дата создания.
- Таблица `nodes` — id, dialog_id, parent_id, content, role (user/llm), timestamp.
- Таблица `commands` — id, folder, command_text, description.
## Рекомендации по дальнейшей реализации
- **Frontend**:
- Создайте React-приложение с тремя панелями.
- Используйте React Flow с Dagre для графа (примеры из [1][2][4]).
- Реализуйте IndexedDB для хранения истории и команд.
- Подключите React Autosuggest для автодополнения команд.
- **Backend**:
- API для вызова LLM (Gemini).
- API для git diff, генерации сообщений коммита.
- API для получения субтитров (интеграция с Google Meet / MS Teams).
- API для генерации и анализа изображений (bingart).
- **Суммаризация истории**:
- При отправке нового сообщения формируйте промпт из последних N сообщений + краткое резюме старой истории через LLM.
Если хотите, могу помочь с примером React-компонента для графа на React Flow + Dagre или примером IndexedDB-схемы. Также могу подготовить пример автодополнения команд через React Autosuggest.
Жду ваших пожеланий по следующему шагу!
---
а langgrpah - здесь нужен? потому что я хочу, чтобы по слешу не просто промпты поставлялись, а целые действия выполнялись. Где-то промпт присовокупляется к запросу (но промпт не подставляется, т.е. пользователь не видит), а где неявносначала выполняется скрипт генерации с рандомизацией, а его вывод + запрос подается на вход llm через api
---
LangGraph — это именно тот инструмент, который вам нужен для реализации описанной логики с «целыми действиями» по слеш-командам, где:
- не просто подставляются промпты, а
- запускаются сложные workflow с несколькими узлами (например, генерация с рандомизацией, вызов внешних API, анализ изображений, git diff и т.д.),
- с хранением состояния и ветвлением диалога в графе,
- с возможностью динамически добавлять новые узлы и ветви,
- и с контролем, какие промпты и когда подмешиваются к запросам (причём прозрачно для пользователя).
### Почему LangGraph подходит именно для вашей задачи
- **Графовая модель**: вы описываете логику как набор узлов (actions, LLM вызовы, скрипты), связанных рёбрами — именно это и есть LangGraph[1][3][5].
- **Вызов функций и внешних API**: LangGraph позволяет в узлах выполнять произвольный код (скрипты, запросы), а результат подмешивать к промптам или использовать отдельно[3][4].
- **Состояние и память**: хранит состояние диалога, можно реализовать суммаризацию, ветвления, переключение контекста[1][5].
- **Мультиагентность и сложные сценарии**: можно объединять разные LLM, инструменты, API, что идеально для ваших слеш-команд с разными действиями[2][6].
- **Поддержка потоковой генерации и трассировки**: удобно для интерактивных чатов с визуализацией и отладкой[5].
- **Расширяемость под Ollama и другие LLM**: можно подключать любые модели и переключаться между ними[6].
### Как это будет работать в вашем случае
1. Пользователь вводит в чат команду с `/` — фронтенд (React) через автодополнение предлагает варианты.
2. По выбору команды LangGraph запускает соответствующий узел или цепочку узлов:
- например, сначала запускает скрипт генерации с рандомизацией (не показывая пользователю),
- потом подмешивает результат в промпт,
- вызывает LLM через API,
- получает ответ,
- анализирует изображения,
- при необходимости вызывает git diff, субтитры и т.д.
3. Каждый шаг — отдельный узел графа с состоянием, что позволяет легко отлаживать и расширять логику.
4. Визуализация графа в React Flow отражает структуру диалога и действия агента.
5. Активный узел и ветвь подсвечиваются, чат показывает цепочку сообщений от корня до текущего узла.
### Итог
- **LangGraph — это не просто вызов LLM, а полноценный оркестратор сложных агентных workflow с хранением состояния и ветвлением.**
- Он идеально подходит, если нужно, чтобы по слеш-командам выполнялись не только промпты, а целые сценарии с вызовами функций, API, анализом и генерацией.
- Простые случаи (только промпты) можно реализовать и без него, но для вашей задачи с динамическими действиями и состоянием — LangGraph значительно упростит поддержку и масштабирование.
### Рекомендация
Используйте LangGraph как основу backend-логики для вашего агента. Реализуйте узлы для:
- генерации промптов с рандомизацией,
- вызова LLM (Gemini и Mistral),
- анализа изображений (Gemini Vision или кастомные API),
- git diff и обработки субтитров,
- управления состоянием диалога и ветвлениями.
Фронтенд будет взаимодействовать с LangGraph через API, получать состояние графа и визуализировать его.
Если хотите, могу помочь с примером простого LangGraph workflow, который реализует слеш-команду с генерацией и вызовом LLM, или с архитектурой API для интеграции с React.
**Кратко:** Да, LangGraph нужен и рекомендован для вашей задачи, чтобы управлять сложными действиями, а не просто подставлять промпты.
---
`ConversationSummaryMemory, ConversationSummaryBufferMemory - позволяет задать окно (в токенах), в рамках которого мы сохраняем диалоги в неизменной форме, а при превышении - суммаризируем.
ConversationBufferWindowMemory - сохраняет только последние k диалогов.
ConversationKGMemory - использует граф знаний для сохранения памяти.
ConversationEntityMemory - сохраняет в памяти знания об определенном объекте.` здесь не использовать что-то из этого? как вообще с графом быть?

118
TechSpecification.md Normal file
View File

@ -0,0 +1,118 @@
## ТЗ: Интерактивный Агент с LLM и Визуализацией Графа Диалогов
**Цель:** Создать десктопное приложение-агента, позволяющее взаимодействовать с внешними LLM (Gemini, Mistral) через API, выполнять сложные действия по слеш-командам и визуализировать историю диалогов в виде графа.
**1. LLM:**
* Использовать только внешние модели: Gemini (через LangChain), Mistral (через HTTP API).
* Заглушка для Ollama (пустой словарь `LOCAL_MODELS`).
**2. Веб-интерфейс (React):**
* Три панели:
* История диалогов (различные графы запросов), по умолчанию слева.
* Визуализация активного графа сообщений (React Flow + Dagre), по умолчанию по центру.
* Чат с выбранной веткой активного графа, по умолчанию справа.
* Активный чат = выбранный лист графа.
* Клик по ноду = показ сообщений от корня до нода.
* Разные цвета для нодов пользователя и LLM.
**3. Хранение:**
* Бэкенд хранит историю графов запросов.
* IndexedDB (dexie.js или idb) для временного хранения активного графа на фронтенде (опционально).
**4. Суммаризация истории:**
* Комбинация: последние N сообщений + суммаризация старой истории через LLM (Gemini).
**5. Анализ изображений:**
* Gemini Vision API, заглушка для расширения.
**6. Платформы субтитров:**
* Google Meet и Microsoft Teams: возможность получения текущих, не финальных субтитров (API или polling/webhook).
**7. Генерация изображений:**
* Вызов удаленного эндпоинта bingart.
Эндпоинт генерации изображений
* **URL:** `<URL_удаленного_ресурса>/generate-images`
* **Метод:** GET или POST.
* **Параметры:**
* `prompt` (строка): Текст запроса для генерации изображения. Передается либо как query parameter в GET-запросе, либо как form data в POST-запросе.
* **Результат (JSON):**
* `image_urls` (массив строк): Список URL сгенерированных изображений (до 4 штук).
* `error` (строка, опционально): Сообщение об ошибке, если генерация не удалась.
Пример успешного ответа:
```json
{
"image_urls": [
"url1",
"url2",
"url3",
"url4"
]
}
```
**Пример ответа с ошибкой:**
```json
{
"error": "Не удалось получить изображения."
}
```
**8. Команды:**
* Слеш-команды (`/`) с автодополнением (React Autosuggest).
* Команды хранятся в базе данных на бэкенде.
* Запуск workflow через LangGraph.
**9. LangGraph:**
* Оркестрация сложных workflow по слеш-командам.
* Узлы: генерация, LLM, анализ изображений, субтитры, управление состоянием.
**10. Реализация:**
* **Frontend:** React, React Flow, Dagre, React Autosuggest.
* **Backend:** API для LLM, субтитров, bingart, LangGraph, база данных (хранение истории графов).
**11. Язык:**
* Русский.
**Типичный workflow:**
1. Пользователь вводит `/команда`.
2. Frontend отправляет запрос на бэкенд.
3. Бэкенд запускает LangGraph workflow, соответствующий команде:
* Подставляет релевантный команде промпт (если необходимо).
* Вызывает LLM (Gemini или Mistral).
* Анализирует изображения (если необходимо).
* Запрашивает текущие субтитры (если необходимо в рамках команды, это одна из команд).
* Запрашивает генерацию изображений на удалённом эндпоинте и ждёт ответа.
* Обрабатывает результаты и формирует ответ.
4. Бэкенд сохраняет новый граф запроса в базе данных (или обновляет существующий).
5. Бэкенд отправляет данные активного графа на фронтенд.
6. Frontend отображает граф и чат.
**Приоритеты:**
1. Базовая функциональность чата с Gemini/Mistral и визуализацией графа.
2. Реализация слеш-команд через LangGraph.
3. Интеграция с Google Meet/MS Teams (получение текущих субтитров).
**Использовать:**
* React Flow (с примерами Dagre).
* LangGraph.
* dexie.js или idb (опционально, для временного хранения на фронтенде).
* React Autosuggest.

73
app/api.py Normal file
View File

@ -0,0 +1,73 @@
"""
Этот файл содержит API endpoints, созданные с использованием Flask.
Он обеспечивает взаимодействие с агентом через HTTP запросы,
включая обработку сообщений, управление графами и получение истории.
"""
from flask import Flask, request, jsonify
from flask_cors import CORS
from workflows import run_agent, graph_history_manager
app2 = Flask(__name__)
CORS(
app2
) # Разрешаем CORS для всех доменов (в production нужно настроить более строго)
@app2.route('/api/chat', methods=['POST'])
def chat():
"""API endpoint для обработки сообщений и возврата ответа."""
data = request.get_json()
message = data.get("message")
graph_id = data.get("graph_id")
parent_node_id = data.get(
"parent_node_id") # Получаем parent_node_id из запроса
if not message:
return jsonify({"error": "Сообщение не может быть пустым."}, 400)
result = run_agent(message, graph_id,
parent_node_id) # Передаем parent_node_id в run_agent
return jsonify(result)
@app2.route('/api/graphs', methods=['GET'])
def get_graphs():
"""API endpoint для получения списка графов."""
graphs = graph_history_manager.get_all_graphs_summary()
print("get_graphs response:", graphs) # Добавлено логирование
return jsonify(graphs)
@app2.route('/api/graphs/<graph_id>', methods=['GET'])
def get_graph_by_id(graph_id):
"""API endpoint для получения данных конкретного графа по ID."""
graph_data = graph_history_manager.get_graph(graph_id)
if graph_data:
return jsonify(graph_data)
else:
return jsonify({"error": f"Граф с ID {graph_id} не найден."}, 404)
@app2.route('/api/graphs/<graph_id>', methods=['DELETE'])
def delete_graph(graph_id):
"""API endpoint для удаления графа."""
if not graph_id:
return jsonify({"error": "Не указан ID графа для удаления."}, 400)
if graph_history_manager.delete_graph(graph_id):
return jsonify({"message": f"Граф {graph_id} успешно удален."})
else:
return jsonify({"error": f"Не удалось удалить граф {graph_id}."}, 500)
@app2.route('/api/messages/<graph_id>/<node_id>', methods=['GET'])
def get_messages_from_root_to_node(graph_id, node_id):
"""API endpoint для получения сообщений от корня до выбранной ноды."""
graph_data = graph_history_manager.get_graph(graph_id)
if not graph_data:
return jsonify({"error": "Граф не найден."}, 404)
messages = graph_history_manager.get_messages_from_root_to_node(
graph_data, node_id)
return jsonify(messages)

190
app/graph_history_manager.py Normal file
View File

@ -0,0 +1,190 @@
"""
Этот модуль управляет сохранением и извлечением истории графов
запросов, используя SQLite базу данных для хранения данных.
"""
from typing import Dict, Any, List, Optional
import json
import sqlite3
class GraphHistoryManager:
"""
Менеджер для хранения истории графов запросов в SQLite.
"""
def __init__(self, db_path="graph_history.db"):
self.db_path = db_path
def _get_connection(self):
"""Получает соединение с базой данных."""
return sqlite3.connect(self.db_path)
def _create_table(self, conn):
"""Создает таблицу для хранения графов, если она не существует."""
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS graphs (
id TEXT PRIMARY KEY,
messages TEXT,
graph_nodes TEXT,
graph_edges TEXT,
current_node_id TEXT
)
""")
conn.commit()
def _get_max_graph_id(self, conn):
"""Получает максимальный ID графа из базы данных."""
cursor = conn.cursor()
cursor.execute(
"SELECT MAX(CAST(SUBSTR(id, 7) AS INTEGER)) FROM graphs WHERE id LIKE 'graph_%'"
)
result = cursor.fetchone()[0]
return result if result is not None else 0
def save_graph(self, graph_data: Any) -> str:
"""Сохраняет или обновляет данные графа в базе данных и возвращает ID."""
conn = self._get_connection()
cursor = conn.cursor()
try:
graph_id = graph_data.get("id")
if not graph_id:
max_graph_id = self._get_max_graph_id(conn)
next_graph_id = max_graph_id + 1
graph_id = f"graph_{next_graph_id}"
graph_data["id"] = graph_id
# Преобразуем структуры данных в JSON-строки для хранения в SQLite
messages_json = json.dumps(graph_data.get("messages", []))
graph_nodes_json = json.dumps(graph_data.get("graph_nodes", []))
graph_edges_json = json.dumps(graph_data.get("graph_edges", []))
current_node_id = graph_data.get("current_node_id", "")
# Проверяем, существует ли уже запись с таким ID
cursor.execute("SELECT id FROM graphs WHERE id = ?", (graph_id, ))
existing_record = cursor.fetchone()
if existing_record:
# Обновляем существующую запись
cursor.execute(
"""
UPDATE graphs SET
messages = ?,
graph_nodes = ?,
graph_edges = ?,
current_node_id = ?
WHERE id = ?
""", (messages_json, graph_nodes_json, graph_edges_json,
current_node_id, graph_id))
else:
# Вставляем новую запись
cursor.execute(
"""
INSERT INTO graphs (id, messages, graph_nodes, graph_edges, current_node_id)
VALUES (?, ?, ?, ?, ?)
""", (graph_id, messages_json, graph_nodes_json,
graph_edges_json, current_node_id))
conn.commit()
print(f"Граф сохранен/обновлен: {graph_id}")
return graph_id
finally:
conn.close()
def get_graph(self, graph_id: str) -> Optional[Any]:
"""Получает данные графа по ID из базы данных."""
conn = self._get_connection()
cursor = conn.cursor()
try:
cursor.execute(
"SELECT messages, graph_nodes, graph_edges, current_node_id FROM graphs WHERE id = ?",
(graph_id, ))
result = cursor.fetchone()
if result:
messages_json, graph_nodes_json, graph_edges_json, current_node_id = result
# Преобразуем JSON-строки обратно в структуры данных Python
messages = json.loads(messages_json)
graph_nodes = json.loads(graph_nodes_json)
graph_edges = json.loads(graph_edges_json)
return {
"id": graph_id,
"messages": messages,
"graph_nodes": graph_nodes,
"graph_edges": graph_edges,
"current_node_id": current_node_id
}
else:
return None
finally:
conn.close()
def get_all_graphs_summary(self) -> List[Dict[str, str]]:
"""Возвращает краткий список всех сохраненных графов."""
conn = self._get_connection()
cursor = conn.cursor()
try:
cursor.execute("SELECT id, messages FROM graphs")
graphs_data = cursor.fetchall()
return [{
"id":
gid,
"first_message":
json.loads(messages)[0].get("content", "No content")
if messages else "No content"
} for gid, messages in graphs_data]
finally:
conn.close()
def get_messages_from_root_to_node(
self, graph_data: Dict[str, Any],
target_node_id: str) -> List[Dict[str, str]]:
"""Получает список сообщений от корневого узла до указанного узла."""
nodes = graph_data.get("graph_nodes", [])
edges = graph_data.get("graph_edges", [])
messages = graph_data.get("messages", [])
# Создаем словарь для быстрого поиска родительских узлов
parent_map = {edge['target']: edge['source'] for edge in edges}
# Функция для рекурсивного подъема по дереву до корневого узла
def get_path_to_root(node_id: str) -> List[str]:
path = [node_id]
while node_id in parent_map:
node_id = parent_map[node_id]
path.append(node_id)
return path[::-1] # Инвертируем, чтобы получить путь от корня
# Получаем путь от корня до целевого узла
path_to_root = get_path_to_root(target_node_id)
# Собираем сообщения, соответствующие узлам в пути
messages_for_path = []
node_ids_in_path = set(path_to_root)
for message in messages:
if "node_id" in message and message["node_id"] in node_ids_in_path:
messages_for_path.append({
"role": message["role"],
"content": message["content"]
})
return messages_for_path
def delete_graph(self, graph_id: str) -> bool:
"""Удаляет граф из базы данных."""
conn = self._get_connection()
cursor = conn.cursor()
try:
cursor.execute("DELETE FROM graphs WHERE id = ?", (graph_id, ))
conn.commit()
return True
except Exception as e:
print(f"Ошибка при удалении графа: {e}")
return False
finally:
conn.close()

196
app/llm_client.py Normal file
View File

@ -0,0 +1,196 @@
"""
Этот модуль предоставляет инструменты для взаимодействия с различными LLM,
включая OpenAI (Gemini) и MistralAI, обеспечивая унифицированный интерфейс.
"""
import requests
from typing import Dict, Any, List
from langchain_core.messages import HumanMessage, SystemMessage
# --- Конфигурация LLM ---
# Пустой словарь для локальных моделей (Ollama)
LOCAL_MODELS: Dict[str, Any] = {}
# Конфигурация внешних моделей
MODELS: Dict[str, Dict[str, Any]] = {
"gemini-2.0-flash": {
"name": "gemini-2.0-flash",
"provider": "openai", # Изменено на "openai"
"model_name": "gemini-2.0-flash", # Добавлено имя модели для LangChain
"apiBase": "https://render-service-gsu7.onrender.com/g/v1beta",
"apiKey": "AIzaSyDpueKFWVqknVKlQn6TdasLmJ2lvAUiBik",
"stream": True,
"capabilities": ["vision"],
},
"gemini-2.5-flash": {
"name": "gemini-2.5-flash",
"provider": "openai", # Изменено на "openai"
"model_name": "gemini-2.5-flash", # Добавлено имя модели для LangChain
"apiBase": "https://render-service-gsu7.onrender.com/g/v1beta",
"apiKey": "AIzaSyDpueKFWVqknVKlQn6TdasLmJ2lvAUiBik",
"stream": True,
"capabilities": ["vision"],
},
"gemini-2.5-flash-preview-05-20": {
"name": "gemini-2.5-flash-preview-05-20",
"provider": "openai", # Изменено на "openai"
"model_name":
"gemini-2.5-flash-preview-05-20", # Добавлено имя модели для LangChain
"apiBase": "https://render-service-gsu7.onrender.com/g/v1beta",
"apiKey": "AIzaSyDpueKFWVqknVKlQn6TdasLmJ2lvAUiBik",
"stream": True,
"capabilities": ["vision"],
},
"mistral-small-latest": {
"name": "mistral-small-latest",
"provider": "mistralai",
"baseUrl": "https://render-service-gsu7.onrender.com/m",
"apiKey": "Q0m29fvxBY0Cfdj4sjHaKqccy1NjonLW",
"stream": True,
"capabilities": ["vision"],
},
}
class CustomLLM:
"""
Класс-обертка для взаимодействия с различными LLM.
"""
def __init__(self, config: Dict[str, Any]):
self.name = config["name"]
self.provider = config["provider"]
self.config = config
if self.provider == "openai":
# Инициализация Gemini через OpenAI API
from openai import OpenAI
self.client = OpenAI(api_key=config["apiKey"],
base_url=config["apiBase"])
self.model_name = config["model_name"]
self.stream = config["stream"]
elif self.provider == "mistralai":
# Для Mistral используем прямые HTTP-запросы
self.base_url = config["baseUrl"]
self.api_key = config["apiKey"]
self.stream = config.get("stream", False)
self.capabilities = config.get("capabilities", [])
else:
raise ValueError(f"Неизвестный провайдер LLM: {self.provider}")
def invoke(self, messages: List[Any]) -> str:
"""
Отправляет запрос к LLM и возвращает ответ.
Messages могут быть строкой или списком объектов Langchain Message.
"""
if self.provider == "openai":
# OpenAI ожидает список сообщений в определенном формате
openai_messages = []
if isinstance(messages, str):
openai_messages.append({"role": "user", "content": messages})
elif isinstance(messages, list):
for msg in messages:
if isinstance(msg, HumanMessage):
openai_messages.append({
"role": "user",
"content": msg.content
})
elif isinstance(msg, SystemMessage):
openai_messages.append({
"role": "system",
"content": msg.content
})
else: # Предполагаем, что это другие типы сообщений Langchain или словари
openai_messages.append({
"role":
msg.type if hasattr(msg, 'type') else 'user',
"content":
msg.content
})
try:
response = self.client.chat.completions.create(
model=self.model_name,
messages=openai_messages,
stream=self.stream,
)
if self.stream:
# Handle streaming responses
collected_chunks = []
collected_messages = []
for chunk in response:
collected_chunks.append(chunk)
chunk_message = chunk.choices[0].delta.content
if chunk_message is not None:
collected_messages.append(chunk_message)
full_reply_content = ''.join(collected_messages)
return full_reply_content
else:
return response.choices[0].message.content
except Exception as e:
print(f"Ошибка при вызове OpenAI API: {e}")
raise
elif self.provider == "mistralai":
# Отступы исправлены
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
# Формируем сообщения для Mistral API
mistral_messages = []
if isinstance(messages, str):
mistral_messages.append({"role": "user", "content": messages})
elif isinstance(messages, list):
for msg in messages:
if isinstance(msg, HumanMessage):
mistral_messages.append({
"role": "user",
"content": msg.content
})
elif isinstance(msg, SystemMessage):
mistral_messages.append({
"role": "system",
"content": msg.content
})
else: # Предполагаем, что это другие типы сообщений Langchain или словари
mistral_messages.append({
"role":
msg.type if hasattr(msg, 'type') else 'user',
"content":
msg.content
})
payload = {
"model": self.name,
"messages": mistral_messages,
"stream": self.stream,
}
try:
response = requests.post(self.base_url,
json=payload,
headers=headers)
response.raise_for_status(
) # Вызывает исключение для HTTP ошибок
data = response.json()
# Извлекаем контент ответа
return data["choices"][0]["message"]["content"]
except requests.exceptions.RequestException as e:
print(f"Ошибка при вызове Mistral API: {e}")
raise
except KeyError:
print(f"Неверный формат ответа от Mistral API: {data}")
raise ValueError("Неверный формат ответа от Mistral API")
return ""
def get_llm(name: str) -> CustomLLM:
"""
Возвращает экземпляр CustomLLM для заданной модели.
"""
config = MODELS.get(name)
if not config:
raise ValueError(f"Модель '{name}' не сконфигурирована.")
return CustomLLM(config)

39
app/models.py Normal file
View File

@ -0,0 +1,39 @@
"""
Определения моделей данных (pydantic) для представления состояния агента LangGraph.
Включают схему для отслеживания хода выполнения диалога и визуализации графа.
"""
from typing import Dict, Any, List, Optional
from langchain_core.pydantic_v1 import BaseModel, Field
# Определяем состояние нашего агента (LangGraph)
class AgentState(BaseModel):
"""Состояние графа агента."""
input: str = Field(description="Исходный запрос пользователя.")
command: Optional[str] = Field(None,
description="Распознанная слеш-команда.")
command_args: List[str] = Field([], description="Аргументы команды.")
chat_history: List[Dict[str, str]] = Field(
[],
description=
"Полная история сообщений в текущей ветке диалога. Формат: [{'role': 'user/assistant', 'content': 'message'}]"
)
llm_response: Optional[str] = Field(None, description="Ответ от LLM.")
image_urls: List[str] = Field(
[], description="URL сгенерированных изображений.")
analysis_result: Optional[str] = Field(
None, description="Результат анализа изображения.")
subtitles: Optional[str] = Field(None, description="Полученные субтитры.")
summarized_history_text: Optional[str] = Field(
None, description="Суммированная история.")
error: Optional[str] = Field(None, description="Сообщение об ошибке.")
# Для визуализации графа (имитация узлов и ребер)
graph_nodes: List[Dict[str, Any]] = Field(
[], description="Узлы для визуализации графа.")
graph_edges: List[Dict[str, Any]] = Field(
[], description="Ребра для визуализации графа.")
current_node_id: str = Field(
"start", description="ID текущего узла графа для фронтенда.")
parent_node_id: Optional[str] = Field(None,
description="ID родительского узла.")

469
app/nodes.py Normal file
View File

@ -0,0 +1,469 @@
"""
Этот файл содержит узлы для графа состояний агента.
Каждый узел выполняет определенную функцию, такую как
разбор команд, вызов LLM или выполнение сервисов.
"""
from langchain_core.messages import HumanMessage, SystemMessage
from llm_client import get_llm
from services import ImageGenerationService, SubtitleService, ImageAnalysisService, SummarizationService
from models import AgentState
from typing import Dict, Any, Optional
class CommandManager:
"""
Менеджер для хранения и получения слеш-команд.
В реальном приложении команды будут храниться в БД.
"""
def __init__(self):
self._commands = {
"help": {
"description": "Показывает список доступных команд.",
"workflow": "help_workflow"
},
"imagine": {
"description":
"Генерирует изображения по промпту. Использование: /imagine [prompt]",
"workflow": "image_generation_workflow"
},
"analyze": {
"description":
"Анализирует изображение по URL и промпту. Использование: /analyze [url] [prompt]",
"workflow": "image_analysis_workflow"
},
"subtitles_meet": {
"description": "Получает текущие субтитры из Google Meet.",
"workflow": "get_meet_subtitles_workflow"
},
"subtitles_teams": {
"description": "Получает текущие субтитры из MS Teams.",
"workflow": "get_teams_subtitles_workflow"
},
"summarize": {
"description": "Суммирует текущую историю диалога.",
"workflow": "summarize_history_workflow"
},
"chat": {
"description": "Ведет обычный диалог с LLM.",
"workflow": "chat_workflow"
},
}
def get_commands(self) -> Dict[str, Any]:
"""Возвращает все доступные команды."""
return self._commands
def get_command_workflow(self, command_name: str) -> Optional[str]:
"""Возвращает имя workflow для заданной команды."""
return self._commands.get(command_name, {}).get("workflow")
# Инициализация сервисов (пример, убедитесь, что они доступны в этом файле или импортированы)
command_manager = CommandManager()
def parse_command_node(state: AgentState) -> AgentState:
"""Парсит вход пользователя на предмет слеш-команды."""
print("Выполняется узел: parse_command_node")
user_input = state.input.strip()
# Добавляем узел пользователя в граф для фронтенда
user_node_id = f"user_{len(state.graph_nodes) + 1}"
state.graph_nodes.append({
"id": user_node_id,
"type": "user",
"data": {
"label":
user_input[:30] + "..." if len(user_input) > 30 else user_input
}
})
if state.parent_node_id:
state.graph_edges.append({
"id": f"e{state.parent_node_id}-{user_node_id}",
"source": state.parent_node_id,
"target": user_node_id
})
state.parent_node_id = user_node_id
state.current_node_id = user_node_id
state.chat_history.append({
"role": "user",
"content": user_input,
"node_id": user_node_id # Сохраняем ID узла
})
if user_input.startswith('/'):
parts = user_input[1:].split(' ', 1)
command_name = parts[0].lower()
command_args = parts[1].split(' ') if len(parts) > 1 else []
if command_name in command_manager.get_commands():
state.command = command_name
state.command_args = command_args
print(
f"Команда распознана: /{state.command} с аргументами: {state.command_args}"
)
else:
state.error = f"Неизвестная команда: /{command_name}. Введите /help для списка команд."
print(state.error)
state.command = "error" # Специальный флаг для обработки ошибок команд
else:
state.command = "chat" # Обычный чат
return state
def execute_command_node(state: AgentState) -> AgentState:
"""Выполняет соответствующий workflow для команды."""
print(
f"Выполняется узел: execute_command_node для команды: {state.command}")
# Этот узел будет выступать в роли маршрутизатора
# Фактическая логика команды будет в отдельных узлах, вызываемых по условию
return state # Просто передаем состояние дальше, чтобы маршрутизатор смог выбрать следующий узел
# Предполагаем, что используем Gemini для основного LLM и суммаризации/анализа
# В реальной системе можно выбрать динамически
DEFAULT_LLM_NAME = "gemini-2.5-flash"
main_llm = get_llm(DEFAULT_LLM_NAME)
def call_llm_node(state: AgentState) -> AgentState:
"""Вызывает LLM для обработки обычного диалога или генерации ответа."""
print("Выполняется узел: call_llm_node (для обычного чата)")
try:
# Для обычного чата LLM использует всю историю
messages_for_llm = []
for msg in state.chat_history:
if msg["role"] == "user":
messages_for_llm.append(HumanMessage(content=msg["content"]))
elif msg["role"] == "assistant":
messages_for_llm.append(SystemMessage(content=msg["content"]))
response = main_llm.invoke(messages_for_llm)
# Добавляем узел LLM в граф для фронтенда
llm_node_id = f"llm_{len(state.graph_nodes) + 1}"
state.graph_nodes.append({
"id": llm_node_id,
"type": "llm",
"data": {
"label":
response[:30] + "..." if len(response) > 30 else response
}
})
state.graph_edges.append({
"id": f"e{state.parent_node_id}-{llm_node_id}",
"source": state.parent_node_id,
"target": llm_node_id
})
state.parent_node_id = llm_node_id
state.current_node_id = llm_node_id
state.chat_history.append({
"role": "assistant",
"content": response,
"node_id": llm_node_id # Сохраняем ID узла
})
state.llm_response = response
except Exception as e:
state.error = f"Ошибка при вызове LLM: {e}"
state.llm_response = f"Произошла ошибка: {e}"
print(state.error)
return state
image_gen_service = ImageGenerationService(
base_url="https://render-service-gsu7.onrender.com/g2"
) # TODO: Замените на реальный URL
def generate_images_node(state: AgentState) -> AgentState:
"""Генерирует изображения по промпту."""
print("Выполняется узел: generate_images_node")
prompt = " ".join(state.command_args)
if not prompt:
state.error = "Для команды /imagine требуется промпт. Пример: /imagine кошка на луне"
state.llm_response = state.error
return state
try:
urls = image_gen_service.generate_images(prompt)
state.image_urls = urls
# Формируем текст ответа с использованием Markdown для отображения изображений
image_markdown = "\n".join([
f'<a href="{url}"><img src="{url}" alt="image{i}" style="max-width: 300px; max-height: 300px;"></a>'
for i, url in enumerate(urls)
])
response_text = f"Сгенерировано {len(urls)} изображений:\n{image_markdown}"
state.llm_response = response_text
# Добавляем узел генерации изображений в граф
img_gen_node_id = f"img_gen_{len(state.graph_nodes) + 1}"
state.graph_nodes.append({
"id": img_gen_node_id,
"type": "tool",
"data": {
"label": "Генерация изображений",
"details": prompt
}
})
state.graph_edges.append({
"id": f"e{state.parent_node_id}-{img_gen_node_id}",
"source": state.parent_node_id,
"target": img_gen_node_id
})
state.parent_node_id = img_gen_node_id
state.current_node_id = img_gen_node_id
state.llm_response = "Сгенерированы изображения."
state.chat_history.append({
"role": "assistant",
"content": response_text,
"node_id": img_gen_node_id # Привязываем к ID узла
})
except Exception as e:
state.error = f"Ошибка генерации изображений: {e}"
state.llm_response = f"Произошла ошибка при генерации изображений: {e}"
print(state.error)
return state
image_analysis_service = ImageAnalysisService(
llm=get_llm("gemini-2.5-flash")) # Gemini для анализа изображений
def analyze_image_node(state: AgentState) -> AgentState:
"""Анализирует изображение по URL или данным."""
print("Выполняется узел: analyze_image_node")
if len(state.command_args) < 2:
state.error = "Для команды /analyze требуется URL изображения и промпт. Пример: /analyze [url] 'что на изображении?'"
state.llm_response = state.error
return state
image_url = state.command_args[0]
prompt = " ".join(state.command_args[1:])
try:
analysis_result = image_analysis_service.analyze_image(
image_url, prompt)
state.analysis_result = analysis_result
state.llm_response = f"Результат анализа изображения '{image_url}': {analysis_result}"
state.chat_history.append({
"role": "assistant",
"content": state.llm_response
})
# Добавляем узел анализа изображения в граф
img_analysis_node_id = f"img_analysis_{len(state.graph_nodes) + 1}"
state.graph_nodes.append({
"id": img_analysis_node_id,
"type": "tool",
"data": {
"label": "Анализ изображения",
"details": f"URL: {image_url}, Промпт: {prompt}"
}
})
state.graph_edges.append({
"id": f"e{state.parent_node_id}-{img_analysis_node_id}",
"source": state.parent_node_id,
"target": img_analysis_node_id
})
state.parent_node_id = img_analysis_node_id
state.current_node_id = img_analysis_node_id
except Exception as e:
state.error = f"Ошибка анализа изображения: {e}"
state.llm_response = f"Произошла ошибка при анализе изображения: {e}"
print(state.error)
return state
subtitle_service = SubtitleService()
def get_meet_subtitles_node(state: AgentState) -> AgentState:
"""Получает субтитры из Google Meet."""
print("Выполняется узел: get_meet_subtitles_node")
try:
subtitles = subtitle_service.get_google_meet_subtitles()
state.subtitles = subtitles
state.llm_response = f"Текущие субтитры из Google Meet: {subtitles}"
state.chat_history.append({
"role": "assistant",
"content": state.llm_response
})
# Добавляем узел субтитров Meet в граф
meet_subtitles_node_id = f"meet_subtitles_{len(state.graph_nodes) + 1}"
state.graph_nodes.append({
"id": meet_subtitles_node_id,
"type": "tool",
"data": {
"label": "Субтитры Google Meet",
"details": subtitles[:30] + "..." if subtitles else ""
}
})
state.graph_edges.append({
"id": f"e{state.parent_node_id}-{meet_subtitles_node_id}",
"source": state.parent_node_id,
"target": meet_subtitles_node_id
})
state.parent_node_id = meet_subtitles_node_id
state.current_node_id = meet_subtitles_node_id
except Exception as e:
state.error = f"Ошибка получения субтитров Google Meet: {e}"
state.llm_response = f"Произошла ошибка при получении субтитров Google Meet: {e}"
print(state.error)
return state
def get_teams_subtitles_node(state: AgentState) -> AgentState:
"""Получает субтитры из MS Teams."""
print("Выполняется узел: get_teams_subtitles_node")
try:
subtitles = subtitle_service.get_ms_teams_subtitles()
state.subtitles = subtitles
state.llm_response = f"Текущие субтитры из MS Teams: {subtitles}"
state.chat_history.append({
"role": "assistant",
"content": state.llm_response
})
# Добавляем узел субтитров Teams в граф
teams_subtitles_node_id = f"teams_subtitles_{len(state.graph_nodes) + 1}"
state.graph_nodes.append({
"id": teams_subtitles_node_id,
"type": "tool",
"data": {
"label": "Субтитры MS Teams",
"details": subtitles[:30] + "..." if subtitles else ""
}
})
state.graph_edges.append({
"id": f"e{state.parent_node_id}-{teams_subtitles_node_id}",
"source": state.parent_node_id,
"target": teams_subtitles_node_id
})
state.parent_node_id = teams_subtitles_node_id
state.current_node_id = teams_subtitles_node_id
except Exception as e:
state.error = f"Ошибка получения субтитров MS Teams: {e}"
state.llm_response = f"Произошла ошибка при получении субтитров MS Teams: {e}"
print(state.error)
return state
summarization_service = SummarizationService(
llm=get_llm("gemini-2.5-flash")) # Gemini для суммаризации
def summarize_history_node(state: AgentState) -> AgentState:
"""Суммирует историю диалога."""
print("Выполняется узел: summarize_history_node")
try:
# Для суммаризации используем полную историю, которую ведет AgentState
summarized_text = summarization_service.summarize_history(
state.chat_history)
state.summarized_history_text = summarized_text
state.llm_response = f"История диалога суммирована:\n{summarized_text}"
state.chat_history.append({
"role": "assistant",
"content": state.llm_response
})
# Добавляем узел суммаризации в граф
summary_node_id = f"summary_{len(state.graph_nodes) + 1}"
state.graph_nodes.append({
"id": summary_node_id,
"type": "tool",
"data": {
"label": "Суммаризация истории",
"details":
summarized_text[:30] + "..." if summarized_text else ""
}
})
state.graph_edges.append({
"id": f"e{state.parent_node_id}-{summary_node_id}",
"source": state.parent_node_id,
"target": summary_node_id
})
state.parent_node_id = summary_node_id
state.current_node_id = summary_node_id
except Exception as e:
state.error = f"Ошибка суммаризации истории: {e}"
state.llm_response = f"Произошла ошибка при суммаризации истории: {e}"
print(state.error)
return state
def handle_error_node(state: AgentState) -> AgentState:
"""Обрабатывает ошибки команд."""
print(f"Выполняется узел: handle_error_node. Ошибка: {state.error}")
state.llm_response = state.error
state.chat_history.append({
"role": "assistant",
"content": state.llm_response
})
# Добавляем узел ошибки в граф
error_node_id = f"error_{len(state.graph_nodes) + 1}"
state.graph_nodes.append({
"id": error_node_id,
"type": "error",
"data": {
"label": "Ошибка",
"details": state.error
}
})
state.graph_edges.append({
"id": f"e{state.parent_node_id}-{error_node_id}",
"source": state.parent_node_id,
"target": error_node_id
})
state.parent_node_id = error_node_id
state.current_node_id = error_node_id
return state
def help_node(state: AgentState) -> AgentState:
"""Предоставляет информацию о доступных командах."""
print("Выполняется узел: help_node")
commands_info = command_manager.get_commands()
help_text = "Доступные команды:\n"
for cmd, info in commands_info.items():
help_text += f"/{cmd}: {info['description']}\n"
state.llm_response = help_text
state.chat_history.append({
"role": "assistant",
"content": state.llm_response
})
# Добавляем узел справки в граф
help_node_id = f"help_{len(state.graph_nodes) + 1}"
state.graph_nodes.append({
"id": help_node_id,
"type": "info",
"data": {
"label": "Справка",
"details": "Список команд"
}
})
state.graph_edges.append({
"id": f"e{state.parent_node_id}-{help_node_id}",
"source": state.parent_node_id,
"target": help_node_id
})
state.parent_node_id = help_node_id
state.current_node_id = help_node_id
return state

158
app/services.py Normal file
View File

@ -0,0 +1,158 @@
"""
Этот файл содержит классы сервисов для выполнения различных задач:
- Получение субтитров,
- Анализ изображений,
- Суммаризация текста,
- Генерация изображений.
"""
from llm_client import CustomLLM
from typing import Dict, Any, List, Optional
import requests
# --- Сервисы для внешних операций ---
class SubtitleService:
"""
Сервис для получения текущих субтитров. Заглушки.
"""
def get_google_meet_subtitles(self) -> Optional[str]:
"""
Получает текущие, не финальные субтитры из Google Meet.
(Реализация требует интеграции с Google Workspace API или другим методом.)
"""
print("Заглушка: Получение субтитров из Google Meet...")
# TODO: Реальная интеграция с Google Meet API
return "Это текущие субтитры из Google Meet: 'Привет, как дела?'"
def get_ms_teams_subtitles(self) -> Optional[str]:
"""
Получает текущие, не финальные субтитры из Microsoft Teams.
(Реализация требует интеграции с Microsoft Graph API или другим методом.)
"""
print("Заглушка: Получение субтитров из Microsoft Teams...")
# TODO: Реальная интеграция с Microsoft Teams API
return "Это текущие субтитры из MS Teams: 'Отлично, спасибо!'"
class ImageAnalysisService:
"""
Сервис для анализа изображений с использованием Gemini Vision API.
"""
def __init__(self, llm: CustomLLM):
self.llm = llm # Ожидаем, что это будет экземпляр Gemini LLM
def analyze_image(self, image_url_or_bytes: Any, prompt: str) -> str:
"""
Анализирует изображение с помощью Gemini Vision.
`image_url_or_bytes` может быть URL, путем к файлу или байтами изображения.
LangChain Gemini Vision принимает это как часть сообщения.
"""
if "vision" not in self.llm.config.get("capabilities", []):
raise ValueError(
"Для анализа изображений требуется LLM с поддержкой 'vision' (например, Gemini)."
)
print(f"Анализ изображения: {image_url_or_bytes} с промптом: {prompt}")
# LangChain GeminiVision ожидает контент в определенном формате
# Для простоты, здесь заглушка, имитирующая обработку изображения.
# В реальной реализации, image_url_or_bytes нужно будет соответствующим образом загрузить/передать.
# Пример: [HumanMessage(content=[{"type": "text", "text": prompt}, {"type": "image_url", "image_url": {"url": image_url_or_bytes}}])]
# Или для локальных файлов: [{"type": "image_bytes", "image_bytes": {"data": base64_encoded_bytes}}]
# Поскольку у нас нет реального механизма загрузки изображения здесь,
# имитируем ответ LLM для демо-целей.
# В реальном приложении:
# messages = [
# HumanMessage(
# content=[
# {"type": "text", "text": prompt},
# {"type": "image_url", "image_url": {"url": image_url_or_bytes}} # Или image_bytes
# ]
# )
# ]
# return self.llm.invoke(messages)
return f"Заглушка анализа изображения: Изображение '{image_url_or_bytes}' содержит (по результатам анализа Gemini): 'яркий пейзаж с горами и рекой' по запросу '{prompt}'."
class SummarizationService:
"""
Сервис для суммаризации истории диалога с использованием Gemini LLM.
"""
def __init__(self, llm: CustomLLM):
self.llm = llm # Ожидаем, что это будет экземпляр Gemini LLM
def summarize_history(self,
history: List[Dict[str, str]],
num_recent_messages: int = 5) -> str:
"""
Суммирует старую историю диалога с использованием LLM.
Комбинация: последние N сообщений + суммаризация старой истории.
"""
if self.llm.provider != "google":
print(
"Предупреждение: Суммаризация рекомендуется с Gemini для лучшего качества."
)
# Последние N сообщений
recent_messages = history[-num_recent_messages:]
old_history = history[:-num_recent_messages]
summarized_old_history = ""
if old_history:
# Формируем промпт для суммаризации старой истории
old_messages_text = "\n".join(
[f"{msg['role']}: {msg['content']}" for msg in old_history])
summary_prompt = f"Суммируй следующий диалог максимально кратко, сохранив ключевые моменты и темы:\n---\n{old_messages_text}\n---"
try:
summarized_old_history = self.llm.invoke(summary_prompt)
except Exception as e:
print(f"Ошибка при суммаризации старой истории: {e}")
summarized_old_history = "Ошибка суммаризации старой истории."
# Объединяем суммированную старую историю с последними сообщениями
full_context = []
if summarized_old_history:
full_context.append(
f"Предшествующая история (краткое содержание): {summarized_old_history}"
)
for msg in recent_messages:
full_context.append(f"{msg['role']}: {msg['content']}")
return "\n".join(full_context)
class ImageGenerationService:
"""
Сервис для генерации изображений через удаленный эндпоинт (bingart).
"""
def __init__(self, base_url: str):
self.base_url = base_url
def generate_images(self, prompt: str) -> List[str]:
"""
Отправляет запрос на генерацию изображений.
"""
url = f"{self.base_url}/generate-images"
try:
# Для простоты используем GET, как указано в ТЗ, хотя POST с body более типичен для prompt
response = requests.get(url, params={"prompt": prompt})
response.raise_for_status()
data = response.json()
if "error" in data:
raise Exception(
f"Ошибка генерации изображений: {data['error']}")
return data.get("image_urls", [])
except requests.exceptions.RequestException as e:
print(f"Ошибка при вызове сервиса генерации изображений: {e}")
raise
except Exception as e:
print(f"Не удалось получить изображения: {e}")
raise

147
app/workflows.py Normal file
View File

@ -0,0 +1,147 @@
"""
Определяет логику рабочих процессов агента,
используя LangGraph для управления состояниями
и переходами между узлами обработки.
"""
from typing import Dict, Any, Optional
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
# --- 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(user_input: str,
existing_graph_id: Optional[str] = None,
parent_node_id: Optional[str] = None) -> Dict[str, Any]:
"""
Запускает агента с заданным пользовательским вводом.
Может продолжить существующий граф по graph_id.
Возвращает обновленное состояние и ID графа.
"""
initial_state = AgentState(input=user_input, parent_node_id=parent_node_id)
# Если есть существующий ID графа, загружаем его историю
if existing_graph_id:
existing_graph_data = graph_history_manager.get_graph(
existing_graph_id)
if existing_graph_data:
print(f"Продолжаю существующий граф {existing_graph_id}")
initial_state.chat_history = existing_graph_data.get(
"messages", [])
initial_state.graph_nodes = existing_graph_data.get(
"graph_nodes", [])
initial_state.graph_edges = existing_graph_data.get(
"graph_edges", [])
initial_state.parent_node_id = parent_node_id or existing_graph_data.get(
"current_node_id")
# Запускаем граф
result = app.invoke(initial_state)
final_state = AgentState(**result)
# Сохраняем обновленное состояние графа
graph_data_to_save = {
"id": existing_graph_id,
"messages": final_state.chat_history,
"graph_nodes": final_state.graph_nodes,
"graph_edges": final_state.graph_edges,
"current_node_id": final_state.current_node_id
}
new_graph_id = graph_history_manager.save_graph(graph_data_to_save)
# Формируем ответ для фронтенда
response_data = {
"graph_id": new_graph_id,
"messages": final_state.chat_history,
"llm_response": final_state.llm_response,
"image_urls": final_state.image_urls,
"analysis_result": final_state.analysis_result,
"subtitles": final_state.subtitles,
"summarized_history_text": final_state.summarized_history_text,
"error": final_state.error,
"graph_visualization_data": {
"nodes": final_state.graph_nodes,
"edges": final_state.graph_edges,
"current_node_id": final_state.current_node_id
}
}
return response_data

BIN
graph_history.db Normal file
View File

Binary file not shown.

8
requirements.txt Normal file
View File

@ -0,0 +1,8 @@
langgraph
langchain
langchain_openai
langchain_google_genai
flask
flask_cors

21
run.py Normal file
View File

@ -0,0 +1,21 @@
# ----------------------------------------------- Imports ------------------------------------------------------------
import os
import sys
# ----------------------------------------------- Main ------------------------------------------------------------
if __name__ == "__main__":
# Добавляем папку app в Python path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'app'))
from app.api import app2
print("🚀 Запуск LLM Agent Backend сервера...")
print("📍 API будет доступно на: http://localhost:5000/api")
print("📊 Список графов: http://localhost:5000/api/graphs")
print("💬 Чат эндпоинт: http://localhost:5000/api/chat")
print("\n⚠️ Для остановки нажмите Ctrl+C\n")
app2.run(
debug=True,
port=5000,
host='localhost'
)