173 lines
8.0 KiB
Python
173 lines
8.0 KiB
Python
import functools
|
||
import importlib.util
|
||
import inspect
|
||
import subprocess
|
||
import json
|
||
import os
|
||
from langchain_core.tools import StructuredTool
|
||
from langgraph.prebuilt import create_react_agent
|
||
from langchain_openai import ChatOpenAI
|
||
from llm_client import MODELS
|
||
|
||
import importlib.util
|
||
import inspect
|
||
import subprocess
|
||
import json
|
||
import os
|
||
import sys
|
||
from typing import Any
|
||
from langchain_core.tools import StructuredTool
|
||
from pydantic import BaseModel, Field
|
||
|
||
def build_react_agent(model_name: str, temperature: float, max_tokens: int, agency_mode: bool, obsidian_settings: dict):
|
||
"""Инициализация ядра на базе LangGraph."""
|
||
m_cfg = MODELS.get(model_name)
|
||
llm = ChatOpenAI(
|
||
api_key=m_cfg["apiKey"],
|
||
base_url=m_cfg["apiBase"],
|
||
model_name=m_cfg["model_name"],
|
||
temperature=temperature,
|
||
max_tokens=max_tokens if max_tokens else None,
|
||
streaming=True
|
||
)
|
||
tools = get_dynamic_tools(obsidian_settings) if agency_mode else []
|
||
return create_react_agent(llm, tools=tools)
|
||
|
||
# Специфическая Pydantic-модель для инструментов без скрипта
|
||
class PurePromptSchema(BaseModel):
|
||
context: str = Field(description="Контекст, данные или текст, к которому нужно применить эту инструкцию/промпт.")
|
||
|
||
|
||
def create_subprocess_tool(script_path: str, method_name: str,
|
||
hidden_prompt: str):
|
||
"""Вид 2: Инструмент со скриптом + пост-процессинг."""
|
||
spec = importlib.util.spec_from_file_location("user_dynamic_tool",
|
||
script_path)
|
||
module = importlib.util.module_from_spec(spec)
|
||
|
||
try:
|
||
spec.loader.exec_module(module)
|
||
func = getattr(module, method_name)
|
||
docstring = inspect.getdoc(func) or f"Выполняет скрипт {method_name}"
|
||
except Exception as e:
|
||
raise RuntimeError(f"Ошибка чтения сигнатуры {script_path}: {e}")
|
||
|
||
@functools.wraps(func)
|
||
def wrapper(**kwargs: Any) -> str:
|
||
runner_code = f"""
|
||
import json
|
||
import sys
|
||
import importlib.util
|
||
|
||
try:
|
||
spec = importlib.util.spec_from_file_location('sub_module', r'{script_path}')
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
|
||
kwargs = json.loads(sys.stdin.read())
|
||
result = getattr(module, '{method_name}')(**kwargs)
|
||
print(result)
|
||
except Exception as e:
|
||
print(f"SCRIPT RUNTIME ERROR: {{e}}", file=sys.stderr)
|
||
sys.exit(1)
|
||
"""
|
||
try:
|
||
process = subprocess.run([sys.executable, "-c", runner_code],
|
||
input=json.dumps(kwargs),
|
||
capture_output=True,
|
||
text=True,
|
||
encoding='utf-8',
|
||
check=True)
|
||
raw_output = process.stdout.strip()
|
||
|
||
# Прилизывание для скрипта
|
||
if hidden_prompt:
|
||
return f"РЕЗУЛЬТАТ ВЫПОЛНЕНИЯ СКРИПТА:\n{raw_output}\n\nИНСТРУКЦИЯ ПО ФОРМАТИРОВАНИЮ ОТВЕТА ДЛЯ АГЕНТА:\n{hidden_prompt}"
|
||
return raw_output
|
||
|
||
except subprocess.CalledProcessError as e:
|
||
# Даже если скрипт упал, мы отдаем ошибку + промпт, чтобы агент мог обработать её по инструкции
|
||
error_msg = f"ОШИБКА ИСПОЛНЕНИЯ (Код {e.returncode}):\nSTDOUT: {e.stdout}\nSTDERR: {e.stderr}"
|
||
if hidden_prompt:
|
||
return f"{error_msg}\n\nИНСТРУКЦИЯ НА СЛУЧАЙ ОШИБКИ:\n{hidden_prompt}"
|
||
return error_msg
|
||
|
||
return StructuredTool.from_function(func=wrapper,
|
||
name=func.__name__,
|
||
description=docstring,
|
||
infer_schema=True)
|
||
|
||
def get_dynamic_tools(obsidian_settings: dict):
|
||
"""Собирает инструменты двух видов на основе настроек Obsidian."""
|
||
tools = []
|
||
custom_tools = obsidian_settings.get("customTools", [])
|
||
|
||
for ct in custom_tools:
|
||
name = ct.get("name")
|
||
desc = ct.get("description")
|
||
full_path = ct.get("scriptPath", "")
|
||
# Поддерживаем оба ключа, если в JS/TS они не раскрылись
|
||
hidden_prompt = ct.get("hiddenPromptExpanded") or ct.get("hiddenPrompt", "")
|
||
|
||
if not name:
|
||
continue # Инструмент без имени создать нельзя
|
||
|
||
# Парсим путь к скрипту (учитываем Windows-пути с буквой диска)
|
||
script_path = ""
|
||
method_name = "main"
|
||
|
||
if full_path:
|
||
parts = full_path.rsplit(":", 1) # Разделяем строку с конца 1 раз
|
||
|
||
if len(parts) == 2:
|
||
# Если слева от разделения ровно 1 символ-буква (например, 'D'),
|
||
# значит это путь без метода, а двоеточие — от диска Windows.
|
||
if len(parts[0]) == 1 and parts[0].isalpha():
|
||
script_path = full_path
|
||
else:
|
||
script_path = parts[0]
|
||
method_name = parts[1]
|
||
else:
|
||
script_path = full_path
|
||
|
||
# ПРОВЕРКА: Есть ли физический скрипт?
|
||
is_valid_script = script_path and os.path.exists(script_path)
|
||
|
||
if not is_valid_script:
|
||
# ----------------------------------------------------
|
||
# ВИД 1: Чистый промпт (Инструмент-инструкция / Системная подсказка)
|
||
# ----------------------------------------------------
|
||
def create_pure_prompt_wrapper(prompt_text: str):
|
||
# Функция принимает context, который ИИ обязан туда положить
|
||
def pure_prompt_wrapper(context: str) -> str:
|
||
return (
|
||
f"ВХОДНЫЕ ДАННЫЕ ОТ ПОЛЬЗОВАТЕЛЯ/МОДЕЛИ:\n{context}\n\n"
|
||
f"ИНСТРУКЦИЯ ПО ОБРАБОТКЕ И ПРИЛИЗЫВАНИЮ (ОБЯЗАТЕЛЬНО К ИСПОЛНЕНИЮ):\n{prompt_text}"
|
||
)
|
||
return pure_prompt_wrapper
|
||
|
||
tool = StructuredTool.from_function(
|
||
func=create_pure_prompt_wrapper(hidden_prompt),
|
||
name=name,
|
||
description=desc or "Инструмент для применения специфических правил форматирования или инструкций.",
|
||
args_schema=PurePromptSchema # Жестко навязываем схему с аргументом context!
|
||
)
|
||
tools.append(tool)
|
||
print(f"🛠 Инструмент-промпт '{tool.name}' (без скрипта) успешно инициализирован.")
|
||
|
||
else:
|
||
# ----------------------------------------------------
|
||
# ВИД 2: Скрипт + Прилизывание
|
||
# ----------------------------------------------------
|
||
try:
|
||
tool = create_subprocess_tool(script_path, method_name, hidden_prompt)
|
||
tool.name = name # Приоритет имени из UI Obsidian
|
||
if desc:
|
||
tool.description = desc
|
||
tools.append(tool)
|
||
print(f"🛠 Инструмент-скрипт '{tool.name}' успешно инициализирован.")
|
||
except Exception as e:
|
||
print(f"❌ Ошибка генерации инструмента-скрипта '{name}': {e}")
|
||
|
||
return tools
|