278 lines
13 KiB
Python
278 lines
13 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, get_type_hints
|
||
from langchain_core.tools import StructuredTool
|
||
from pydantic import BaseModel, Field, create_model
|
||
|
||
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: Инструмент со скриптом + пост-процессинг."""
|
||
|
||
script_dir = os.path.dirname(os.path.abspath(script_path))
|
||
added_to_path = False
|
||
if script_dir not in sys.path:
|
||
sys.path.insert(0, script_dir)
|
||
added_to_path = True
|
||
|
||
try:
|
||
spec = importlib.util.spec_from_file_location("user_dynamic_tool", script_path)
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
func = getattr(module, method_name)
|
||
docstring = inspect.getdoc(func) or f"Выполняет скрипт {method_name}"
|
||
|
||
# ← КЛЮЧЕВОЕ: строим схему ДО того, как потеряем оригинальную сигнатуру
|
||
args_schema = _build_pydantic_schema_from_func(func)
|
||
|
||
except Exception as e:
|
||
raise RuntimeError(f"Ошибка чтения сигнатуры {script_path}: {e}")
|
||
finally:
|
||
# Убираем путь сразу после импорта и получения функции
|
||
if added_to_path:
|
||
sys.path.remove(script_dir)
|
||
|
||
@functools.wraps(func)
|
||
def wrapper(**kwargs: Any) -> str:
|
||
runner_code = f"""
|
||
import json
|
||
import sys
|
||
import importlib.util
|
||
import os
|
||
|
||
# Добавляем папку со скриптом в sys.path, чтобы работали локальные импорты (например, norminv)
|
||
path_to_add = os.path.dirname(os.path.abspath(r'{script_path}'))
|
||
original_path = sys.path.copy()
|
||
sys.path.insert(0, path_to_add)
|
||
|
||
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)
|
||
finally:
|
||
# Возвращаем sys.path в исходное состояние (чистота внутри процесса)
|
||
sys.path = original_path
|
||
"""
|
||
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()
|
||
|
||
raw_stripped = raw_output.strip() if raw_output else ""
|
||
prompt_stripped = hidden_prompt.strip() if hidden_prompt else ""
|
||
|
||
if raw_stripped and prompt_stripped:
|
||
# Случай 1: Есть и результат скрипта, и инструкция — полный вариант
|
||
result = (
|
||
f"Инструмент вернул следующий результат:\n"
|
||
f"<tool_output>\n{raw_output}\n</tool_output>\n\n"
|
||
f"---\n"
|
||
f"Инструкция по оформлению финального ответа пользователю:\n"
|
||
f"<instruction>\n{hidden_prompt}\n</instruction>"
|
||
)
|
||
|
||
elif raw_stripped and not prompt_stripped:
|
||
# Случай 2: Есть результат, нет инструкции
|
||
result = (
|
||
f"Инструмент успешно выполнен. Передай пользователю следующий результат "
|
||
f"ДОСЛОВНО, без изменений, сокращений и дополнений:\n\n"
|
||
f"<tool_output>\n{raw_output}\n</tool_output>"
|
||
)
|
||
|
||
elif not raw_stripped and prompt_stripped:
|
||
# Случай 3: Скрипт вернул пустоту (заглушка/нет данных) — только промпт как инструкция
|
||
result = prompt_stripped
|
||
""" result = (
|
||
f"Скрипт не вернул данных. Выполни следующую инструкцию:\n\n{hidden_prompt}"
|
||
)"""
|
||
|
||
else:
|
||
# Случай 4: Оба пустые — сообщаем агенту честно
|
||
result = (
|
||
"Инструмент не вернул никаких данных и не содержит инструкций. "
|
||
"Сообщи пользователю, что инструмент настроен некорректно."
|
||
)
|
||
|
||
return result
|
||
|
||
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
|
||
|
||
# Явно передаём args_schema — LLM будет видеть реальную типизацию
|
||
return StructuredTool.from_function(
|
||
func=wrapper,
|
||
name=func.__name__,
|
||
description=docstring,
|
||
args_schema=args_schema, # ← вместо 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
|
||
|
||
def _build_pydantic_schema_from_func(func) -> type[BaseModel]:
|
||
"""
|
||
Динамически строит Pydantic BaseModel из сигнатуры функции.
|
||
Используется для корректной передачи схемы в StructuredTool,
|
||
чтобы LLM видела реальные типы аргументов, а не **kwargs.
|
||
"""
|
||
sig = inspect.signature(func)
|
||
|
||
# Пробуем получить аннотации типов (включая from __future__ import annotations)
|
||
try:
|
||
hints = get_type_hints(func)
|
||
except Exception:
|
||
hints = {}
|
||
|
||
fields = {}
|
||
for param_name, param in sig.parameters.items():
|
||
if param_name in ("self", "cls"):
|
||
continue
|
||
if param.kind in (
|
||
inspect.Parameter.VAR_POSITIONAL, # *args
|
||
inspect.Parameter.VAR_KEYWORD, # **kwargs
|
||
):
|
||
continue
|
||
|
||
# Тип параметра
|
||
annotation = hints.get(param_name, Any)
|
||
|
||
# Дефолтное значение
|
||
if param.default is inspect.Parameter.empty:
|
||
# Обязательный параметр
|
||
fields[param_name] = (annotation, Field(...))
|
||
else:
|
||
fields[param_name] = (annotation, Field(default=param.default))
|
||
|
||
if not fields:
|
||
# Фоллбек: если функция не имеет параметров или все **kwargs —
|
||
# создаём схему с одним полем, чтобы LLM хоть что-то передала
|
||
fields["kwargs"] = (dict, Field(default_factory=dict,
|
||
description="Аргументы функции"))
|
||
|
||
model = create_model(
|
||
f"{func.__name__}_Schema",
|
||
**fields
|
||
)
|
||
return model
|