diff --git a/.github/workflows/pr_agent.yml b/.github/workflows/pr_agent.yml index b5a782c..8fd3f8f 100644 --- a/.github/workflows/pr_agent.yml +++ b/.github/workflows/pr_agent.yml @@ -6,9 +6,6 @@ on: branches: - main - dev - - dev_lfx - - dev_madina - - dev_Mkdocs workflow_dispatch: inputs: branch: diff --git a/README.md b/README.md index c07963f..f8da8ab 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,13 @@ --- +## v1.1.0 Highlights + +- Hardened the interpreter path by isolating LLM-generated Python behind trusted-mode controls and a subprocess runner instead of in-process execution. +- Improved CLI safety with better file staging validation, collision handling, session-scoped logging, and consistent `--api-key` propagation through fallback SPARQL paths. +- Made the Streamlit app easier to launch locally by stabilizing its import path handling and aligning the recommended startup command with the tested repo-root workflow. + +======= ## Demo Try the public MetaboT demonstrator at [metabot.holobiomicslab.eu](https://metabot.holobiomicslab.eu). It is connected to the Experimental Natural Products Knowledge Graph (ENPKG), an open metabolomics knowledge graph built from a chemodiverse collection of [1,600 plant extracts](https://doi.org/10.1093/gigascience/giac124). @@ -99,6 +106,14 @@ conda env create -f environment.yml conda activate metabot ``` +To launch the application through Streamlit, install the dependencies and run the app from the repository root. In your terminal, execute: + +```bash +pip install -r requirements.txt +python -m streamlit run streamlit_webapp/streamlit_app.py +``` + +This repo-root launch path is the recommended setup for local development and matches the Streamlit smoke-tested workflow used in `v1.1.0`. You can provide your OpenAI key in the sidebar once the app starts, or preconfigure contributor/admin keys through environment variables if you use those deployment paths. If you prefer a plain virtual environment instead of Conda: ```bash diff --git a/app/config/langgraph.json b/app/config/langgraph.json index 741203d..baf5ff3 100644 --- a/app/config/langgraph.json +++ b/app/config/langgraph.json @@ -3,14 +3,12 @@ {"name": "Entry_Agent", "path": "app.core.agents.entry.agent", "llm_choice": "llm_o"}, {"name": "ENPKG_agent", "path": "app.core.agents.enpkg.agent", "llm_choice": "llm_o"}, {"name": "Sparql_query_runner", "path": "app.core.agents.sparql.agent", "llm_choice": "llm_o"}, - {"name": "Interpreter_agent", "path": "app.core.agents.interpreter.agent", "llm_choice": "llm_o"}, {"name": "supervisor", "path": "app.core.agents.supervisor.agent", "llm_choice": "llm_o"}, {"name": "Validator", "path": "app.core.agents.validator.agent", "llm_choice": "llm_o"} ], "edges": [ {"source": "ENPKG_agent", "target": "supervisor"}, - {"source": "Sparql_query_runner", "target": "supervisor"}, - {"source": "Interpreter_agent", "target": "supervisor"} + {"source": "Sparql_query_runner", "target": "supervisor"} ], "conditional_edges": [ { @@ -19,7 +17,6 @@ "targets": [ {"condition_value": "ENPKG_agent", "target": "ENPKG_agent"}, {"condition_value": "Sparql_query_runner", "target": "Sparql_query_runner"}, - {"condition_value": "Interpreter_agent", "target": "Interpreter_agent"}, {"condition_value": "FINISH", "target": "__end__"} ] }, @@ -41,6 +38,6 @@ "entry_point": "Entry_Agent", "supervisor": { "name": "supervisor", - "members": ["ENPKG_agent", "Sparql_query_runner", "Interpreter_agent"] + "members": ["ENPKG_agent", "Sparql_query_runner"] } } diff --git a/app/config/params.ini b/app/config/params.ini index c2b9a26..8812aae 100644 --- a/app/config/params.ini +++ b/app/config/params.ini @@ -1,16 +1,16 @@ [llm_preview] -id = gpt-4o +id = gpt-5.4 temperature = 0.3 max_retries = 3 [llm_o] -id = gpt-4o +id = gpt-5.4 temperature = 0 max_retries = 3 [llm_mini] -id = gpt-4o-mini +id = gpt-5.4-mini temperature = 0 max_retries = 3 @@ -48,7 +48,7 @@ max_retries = 3 base_url = https://llama-3-1-70b-instruct.endpoints.kepler.ai.cloud.ovh.net/api/openai_compat/v1 [llm_litellm_openai] -id = gpt-4o +id = gpt-5.4 temperature = 0 diff --git a/app/core/agents/agents_factory.py b/app/core/agents/agents_factory.py index bf93e38..9700ed4 100644 --- a/app/core/agents/agents_factory.py +++ b/app/core/agents/agents_factory.py @@ -10,8 +10,6 @@ logger = setup_logger(__name__) -config = load_config() - def create_all_agents(llms, graph, openai_key=None, session_id=None): """ @@ -27,6 +25,7 @@ def create_all_agents(llms, graph, openai_key=None, session_id=None): dict: A dictionary mapping agent names to their created executor instances. """ + config = load_config() agents = config["agents"] executors = {} diff --git a/app/core/agents/enpkg/tool_chemicals.py b/app/core/agents/enpkg/tool_chemicals.py index 0736ff8..8f126bf 100644 --- a/app/core/agents/enpkg/tool_chemicals.py +++ b/app/core/agents/enpkg/tool_chemicals.py @@ -16,7 +16,7 @@ from langchain_openai import OpenAIEmbeddings -from langchain.pydantic_v1 import BaseModel, Field +from pydantic import BaseModel, Field from langchain.tools import BaseTool from typing import Optional @@ -47,7 +47,7 @@ class ChemicalResolver(BaseTool): Dict[str, str]: a dictionary that contains the output chemical name and corresponding NPC Class URI. """ - args_schema = ChemicalInput + args_schema: type[BaseModel] = ChemicalInput csv_data: List[Document] = None retriever: Any = None openai_key: str = None diff --git a/app/core/agents/enpkg/tool_smiles.py b/app/core/agents/enpkg/tool_smiles.py index 32d1a4b..0c904ed 100644 --- a/app/core/agents/enpkg/tool_smiles.py +++ b/app/core/agents/enpkg/tool_smiles.py @@ -1,7 +1,7 @@ import requests from langchain.tools import BaseTool -from langchain.pydantic_v1 import BaseModel, Field +from pydantic import BaseModel, Field from typing import Optional @@ -36,7 +36,7 @@ class SMILESResolver(BaseTool): smiles_string = "CCC12CCCN3C1C4(CC3)C(CC2)NC5=CC=CC=C45" inchikey = _run(smiles_string) """ - args_schema = SMILESInput + args_schema: type[BaseModel] = SMILESInput openai_key: str = None def __init__(self, openai_key: str = None): diff --git a/app/core/agents/enpkg/tool_target.py b/app/core/agents/enpkg/tool_target.py index 7bf6bc7..07fd11d 100644 --- a/app/core/agents/enpkg/tool_target.py +++ b/app/core/agents/enpkg/tool_target.py @@ -5,7 +5,7 @@ -from langchain.pydantic_v1 import BaseModel, Field +from pydantic import BaseModel, Field from typing import Optional @@ -38,7 +38,7 @@ class TargetResolver(BaseTool): str: A string containing the ChEMBLTarget notation. """ - args_schema = TargetInput + args_schema: type[BaseModel] = TargetInput openai_key: str = None def __init__(self, openai_key: str = None): diff --git a/app/core/agents/enpkg/tool_taxon.py b/app/core/agents/enpkg/tool_taxon.py index 04f8aa6..5b0e971 100644 --- a/app/core/agents/enpkg/tool_taxon.py +++ b/app/core/agents/enpkg/tool_taxon.py @@ -1,11 +1,9 @@ -from typing import Optional +from typing import ClassVar, Optional from SPARQLWrapper import JSON, SPARQLWrapper -from langchain.pydantic_v1 import BaseModel, Field - -from typing import Optional +from pydantic import BaseModel, Field from langchain.callbacks.manager import ( CallbackManagerForToolRun, @@ -34,10 +32,10 @@ class TaxonResolver(BaseTool): Returns: str: A string that contains the Wikidata IRI if found, otherwise `None`. """ - args_schema = TaxonInput + args_schema: type[BaseModel] = TaxonInput - ENDPOINT_URL = "https://query.wikidata.org/sparql" - PREFIXES = """ + ENDPOINT_URL: ClassVar[str] = "https://query.wikidata.org/sparql" + PREFIXES: ClassVar[str] = """ PREFIX prov: PREFIX pr: PREFIX wdt: diff --git a/app/core/agents/entry/tool_filesparser.py b/app/core/agents/entry/tool_filesparser.py index a68e3ed..6f0a103 100644 --- a/app/core/agents/entry/tool_filesparser.py +++ b/app/core/agents/entry/tool_filesparser.py @@ -2,18 +2,22 @@ import pandas as pd from datetime import datetime from langchain.tools import BaseTool -from langchain.pydantic_v1 import BaseModel, Field +from pydantic import BaseModel, Field from pathlib import Path from app.core.session import setup_logger, create_user_session logger = setup_logger(__name__) +class FileAnalyzerInput(BaseModel): + """Input schema for FileAnalyzer - no input fields required.""" + pass + class FileAnalyzer(BaseTool): name: str = "FILE_ANALYZER" description: str = """ Analyzes files in a specified directory and provides a summary of their content. """ - args_schema = BaseModel # Using BaseModel directly since no specific input fields are necessary + args_schema: type[BaseModel] = FileAnalyzerInput folder_path: Path = None openai_key: str = None session_id: str = None diff --git a/app/core/agents/interpreter/agent.py b/app/core/agents/interpreter/agent.py index 3c755b6..5a3d406 100644 --- a/app/core/agents/interpreter/agent.py +++ b/app/core/agents/interpreter/agent.py @@ -18,6 +18,7 @@ def create_agent(llms, graph, openai_key, session_id) -> AgentExecutor: tool_parameters = { "openai_key": openai_key, "session_id": session_id, + "llm_instance": llms.get(MODEL_CHOICE), } tools = import_tools(directory, module_prefix, **tool_parameters) diff --git a/app/core/agents/interpreter/prompt.py b/app/core/agents/interpreter/prompt.py index 5190e5c..98a3dc5 100644 --- a/app/core/agents/interpreter/prompt.py +++ b/app/core/agents/interpreter/prompt.py @@ -23,9 +23,10 @@ For using INTERPRETER_TOOL tool, you have to provide all the information needed for the interpretation of the question.This allows INTERPRETER_TOOL to analyze the information thoroughly and provide a clear, concise answer to the initial question. Your input for the tool should be a dictionary with one key, __arg1, and the value of this key should be a string contain at least all relevant complete filepaths, the user question and your demands for the tool. You should include, if any, the generated SPARQL query or other relevant information. + Wrap every filepath in quotes. This is mandatory for any filepath containing spaces and preferred for all filepaths. Your demand should explicitly say what you want from the tool. Please include Example: - Filepath: Filepath provided by SPARQL_QUERY_RUNNER + Filepath: "Filepath provided by SPARQL_QUERY_RUNNER" User question: User question Demand: Interpret the file to give x and y information and provide a chart Other relevant information: Other information here diff --git a/app/core/agents/interpreter/sandbox_runner.py b/app/core/agents/interpreter/sandbox_runner.py new file mode 100644 index 0000000..011d3b6 --- /dev/null +++ b/app/core/agents/interpreter/sandbox_runner.py @@ -0,0 +1,310 @@ +from __future__ import annotations + +import argparse +import ast +import builtins +import contextlib +import io +import json +import os +import signal +import sys +import tempfile +import traceback +from pathlib import Path + + +ALLOWED_IMPORT_ROOTS = { + "collections", + "csv", + "itertools", + "json", + "math", + "numpy", + "pandas", + "pathlib", + "plotly", + "re", + "statistics", +} + +REAL_IMPORT = builtins.__import__ +REAL_OPEN = builtins.open +REAL_IO_OPEN = io.open +REAL_OS_OPEN = os.open +REAL_PATH_OPEN = Path.open + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--payload", required=True) + parser.add_argument("--result", required=True) + return parser.parse_args() + + +def write_result(result_path: Path, payload: dict) -> None: + with REAL_OPEN(result_path, "w", encoding="utf-8") as handle: + json.dump(payload, handle) + + +def validate_code(code: str) -> None: + tree = ast.parse(code, mode="exec") + blocked_names = {"__import__", "eval", "exec", "compile", "globals", "locals", "vars", "breakpoint"} + + # Defense in depth only: AST validation helps reduce accidental footguns, + # but the subprocess sandbox remains the primary containment layer. + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + if alias.name.split(".", 1)[0] not in ALLOWED_IMPORT_ROOTS: + raise ValueError(f"Import of '{alias.name}' is not allowed.") + elif isinstance(node, ast.ImportFrom): + module = node.module or "" + if module.split(".", 1)[0] not in ALLOWED_IMPORT_ROOTS: + raise ValueError(f"Import from '{module}' is not allowed.") + elif isinstance(node, ast.Name) and node.id in blocked_names: + raise ValueError(f"Use of '{node.id}' is not allowed.") + elif isinstance(node, ast.Attribute): + if node.attr.startswith("__") and node.attr.endswith("__"): + raise ValueError("Dunder attribute access is not allowed.") + + +def apply_resource_limits(config: dict) -> None: + try: + import resource + except ImportError: + return + + cpu_seconds = int(config["cpu_seconds"]) + memory_bytes = int(config["memory_bytes"]) + file_size_bytes = int(config["file_size_bytes"]) + + limits = [ + ("RLIMIT_CPU", (cpu_seconds, cpu_seconds)), + ("RLIMIT_AS", (memory_bytes, memory_bytes)), + ("RLIMIT_DATA", (memory_bytes, memory_bytes)), + ("RLIMIT_FSIZE", (file_size_bytes, file_size_bytes)), + ("RLIMIT_NOFILE", (64, 64)), + ("RLIMIT_NPROC", (1, 1)), + ] + + for limit_name, limit_value in limits: + if not hasattr(resource, limit_name): + continue + try: + resource.setrlimit(getattr(resource, limit_name), limit_value) + except (OSError, ValueError): + continue + + +def configure_timeout(timeout_seconds: int) -> None: + def handle_timeout(signum, frame): + raise TimeoutError(f"Interpreter execution timed out after {timeout_seconds} seconds.") + + signal.signal(signal.SIGALRM, handle_timeout) + signal.alarm(timeout_seconds) + + +def disable_network() -> None: + import socket + + def blocked(*args, **kwargs): + raise PermissionError("Outbound network access is disabled for the interpreter.") + + socket.socket = blocked + socket.create_connection = blocked + socket.fromfd = blocked + socket.socketpair = blocked + socket.getaddrinfo = blocked + socket.gethostbyname = blocked + + +def patch_filesystem_access(session_dir: Path, allowed_input_paths: list[str]): + allowed_reads = {Path(path).resolve(strict=False) for path in allowed_input_paths} + created_paths: set[Path] = set() + + tempfile.tempdir = str(session_dir) + + def resolve_candidate(path_value) -> Path: + if isinstance(path_value, bytes): + path_value = os.fsdecode(path_value) + if isinstance(path_value, int): + raise PermissionError("File descriptor access is not allowed in the interpreter sandbox.") + + candidate = Path(path_value) + if not candidate.is_absolute(): + candidate = session_dir / candidate + + return candidate.resolve(strict=False) + + def ensure_allowed(path_value, mode: str) -> Path: + resolved = resolve_candidate(path_value) + write_mode = any(flag in mode for flag in ("w", "a", "x", "+")) + + if write_mode: + if not is_within_directory(resolved, session_dir): + raise PermissionError("Writes are restricted to the active session directory.") + resolved.parent.mkdir(parents=True, exist_ok=True) + created_paths.add(resolved) + return resolved + + if resolved in allowed_reads or resolved in created_paths: + return resolved + + raise PermissionError("Reads are restricted to validated session input files.") + + def safe_open(file, mode="r", *args, **kwargs): + resolved = ensure_allowed(file, mode) + return REAL_OPEN(resolved, mode, *args, **kwargs) + + def safe_io_open(file, mode="r", *args, **kwargs): + resolved = ensure_allowed(file, mode) + return REAL_IO_OPEN(resolved, mode, *args, **kwargs) + + def safe_os_open(file, flags, mode=0o777, *args, **kwargs): + write_flags = os.O_WRONLY | os.O_RDWR | os.O_APPEND | os.O_CREAT | os.O_TRUNC + open_mode = "r+" if flags & write_flags else "r" + resolved = ensure_allowed(file, open_mode) + return REAL_OS_OPEN(os.fspath(resolved), flags, mode, *args, **kwargs) + + builtins.open = safe_open + io.open = safe_io_open + os.open = safe_os_open + Path.open = lambda self, *args, **kwargs: safe_open(self, *args, **kwargs) + + return safe_open + + +def restore_filesystem_access() -> None: + builtins.open = REAL_OPEN + io.open = REAL_IO_OPEN + os.open = REAL_OS_OPEN + Path.open = REAL_PATH_OPEN + + +def is_within_directory(path: Path, directory: Path) -> bool: + resolved_path = path.resolve(strict=False) + resolved_directory = directory.resolve(strict=False) + return resolved_path == resolved_directory or resolved_directory in resolved_path.parents + + +def guarded_import(name, globals=None, locals=None, fromlist=(), level=0): + caller = (globals or {}).get("__name__", "") + root = name.split(".", 1)[0] + if caller == "__sandbox__" and root not in ALLOWED_IMPORT_ROOTS: + raise ImportError(f"Import of '{root}' is not allowed in the interpreter sandbox.") + return REAL_IMPORT(name, globals, locals, fromlist, level) + + +def build_safe_builtins(safe_open): + return { + "__import__": guarded_import, + "abs": abs, + "all": all, + "any": any, + "bool": bool, + "bytes": bytes, + "dict": dict, + "enumerate": enumerate, + "Exception": Exception, + "FileNotFoundError": FileNotFoundError, + "filter": filter, + "float": float, + "int": int, + "isinstance": isinstance, + "KeyError": KeyError, + "len": len, + "list": list, + "map": map, + "max": max, + "min": min, + "next": next, + "open": safe_open, + "PermissionError": PermissionError, + "print": print, + "range": range, + "round": round, + "RuntimeError": RuntimeError, + "set": set, + "sorted": sorted, + "str": str, + "sum": sum, + "TimeoutError": TimeoutError, + "tuple": tuple, + "ValueError": ValueError, + "zip": zip, + } + + +def execute_user_code(config: dict) -> dict: + session_dir = Path(config["session_dir"]).resolve(strict=False) + session_dir.mkdir(parents=True, exist_ok=True) + os.chdir(session_dir) + + if hasattr(os, "geteuid") and os.geteuid() == 0: + raise PermissionError( + "The interpreter runner must not execute as root. " + "Run MetaboT under a restricted OS user before enabling trusted mode." + ) + + validate_code(config["code"]) + apply_resource_limits(config) + configure_timeout(int(config["timeout_seconds"])) + disable_network() + safe_open = patch_filesystem_access(session_dir, config["allowed_input_paths"]) + + stdout_buffer = io.StringIO() + stderr_buffer = io.StringIO() + safe_builtins = build_safe_builtins(safe_open) + sandbox_globals = {"__builtins__": safe_builtins, "__name__": "__sandbox__"} + + try: + with contextlib.redirect_stdout(stdout_buffer), contextlib.redirect_stderr(stderr_buffer): + try: + exec(compile(config["code"], "", "exec"), sandbox_globals, sandbox_globals) + except BaseException as exc: + return { + "success": False, + "stdout": stdout_buffer.getvalue(), + "stderr": stderr_buffer.getvalue(), + "error": str(exc), + "traceback": traceback.format_exc(), + } + return { + "success": True, + "stdout": stdout_buffer.getvalue(), + "stderr": stderr_buffer.getvalue(), + "error": "", + } + finally: + signal.alarm(0) + restore_filesystem_access() + + +def main() -> int: + args = parse_args() + payload_path = Path(args.payload) + result_path = Path(args.result) + + try: + with REAL_OPEN(payload_path, "r", encoding="utf-8") as handle: + config = json.load(handle) + result = execute_user_code(config) + write_result(result_path, result) + return 0 if result.get("success") else 1 + except Exception as exc: + write_result( + result_path, + { + "success": False, + "stdout": "", + "stderr": "", + "error": str(exc), + "traceback": traceback.format_exc(), + }, + ) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/app/core/agents/interpreter/tool_interpreter.py b/app/core/agents/interpreter/tool_interpreter.py index d51a2e0..b1259f4 100644 --- a/app/core/agents/interpreter/tool_interpreter.py +++ b/app/core/agents/interpreter/tool_interpreter.py @@ -1,29 +1,55 @@ from __future__ import annotations - -from codeinterpreterapi import CodeInterpreterSession, File, settings -from langchain.pydantic_v1 import BaseModel, Field -from langchain.tools import BaseTool - -from typing import Optional - -from langchain.callbacks.manager import ( - CallbackManagerForToolRun, -) import json -from app.core.session import setup_logger, create_user_session - import os import re +import signal +import subprocess +import sys import tempfile from pathlib import Path +from typing import Any, Optional + +import pandas as pd +from langchain.callbacks.manager import CallbackManagerForToolRun +from langchain.tools import BaseTool +from langchain_core.messages import HumanMessage, SystemMessage +from langchain_openai import ChatOpenAI +from pydantic import BaseModel, Field + from app.core.memory.database_manager import tools_database +from app.core.security import ( + get_interpreter_cpu_seconds, + get_interpreter_max_file_bytes, + get_interpreter_memory_bytes, + get_interpreter_timeout_seconds, + is_trusted_mode_enabled, + resolve_session_path, +) +from app.core.session import create_user_session, setup_logger logger = setup_logger(__name__) +_ALLOWED_INPUT_SUFFIXES = {".csv", ".tsv", ".mgf", ".txt", ".xlsx", ".xls", ".json"} +_ALLOWED_INPUT_SUFFIXES_PATTERN = "|".join( + sorted(suffix.removeprefix(".") for suffix in _ALLOWED_INPUT_SUFFIXES) +) +_QUOTED_FILE_PATH_PATTERN = re.compile( + rf"""(?P["'])(?P[^"']+\.(?:{_ALLOWED_INPUT_SUFFIXES_PATTERN}))(?P=quote)""", + re.IGNORECASE, +) +_UNQUOTED_FILE_PATH_PATTERN = re.compile( + rf"""(?P[A-Za-z0-9_./\\-]+\.(?:{_ALLOWED_INPUT_SUFFIXES_PATTERN}))""", + re.IGNORECASE, +) +_MAX_OUTPUT_CHARS = 2000 + class InterpreterInput(BaseModel): - input: str = Field(description="Input from Interpreter Agent containing the user's question, necessary file paths and other information.") + input: str = Field( + description="Input from Interpreter Agent containing the user's question, necessary file paths and other information." + ) + class Interpreter(BaseTool): @@ -37,112 +63,407 @@ class Interpreter(BaseTool): Returns: None: Outputs the response after interpreting files. """ - args_schema = InterpreterInput - openai_key: str = None - session_id: str = None + args_schema: type[BaseModel] = InterpreterInput + openai_key: Optional[str] = None + session_id: Optional[str] = None + llm_instance: Optional[Any] = None - def __init__(self, openai_key: str, session_id: str): + def __init__( + self, + openai_key: Optional[str] = None, + session_id: Optional[str] = None, + llm_instance: Optional[Any] = None, + ): super().__init__() - self.openai_key = openai_key + self.openai_key = openai_key or os.getenv("OPENAI_API_KEY") self.session_id = session_id + self.llm_instance = llm_instance def _run( self, input: str, run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> None: - + ) -> str: logger.info(f"Input: {input}") - file_paths = self.extract_file_paths(input) - logger.info(f"File paths: {file_paths}") + if not is_trusted_mode_enabled(): + logger.warning("Rejected interpreter execution because trusted mode is disabled.") + return ( + "Interpreter execution is disabled in this deployment. " + "Set METABOT_TRUSTED_MODE=true only in a trusted environment with host-level isolation." + ) session_dir = create_user_session(self.session_id, user_session_dir=True) + session_dir.mkdir(parents=True, exist_ok=True) + db_manager = tools_database() + + file_paths = self.collect_allowed_file_paths( + input_text=input, + session_dir=session_dir, + db_manager=db_manager, + ) + logger.info(f"Validated file paths: {file_paths}") + + if not file_paths: + return ( + "Interpreter could not find any readable session files. " + "Only files created or uploaded in the current session are allowed." + ) - settings.OPENAI_API_KEY = self.openai_key - settings.MODEL = "gpt-3.5-turbo" - - with CodeInterpreterSession() as session: + llm = self.llm_instance or ChatOpenAI( + api_key=self.openai_key, + model="gpt-5.4", + temperature=0, + ) - db_manager = tools_database() + system_prompt = self.build_system_prompt(session_dir=session_dir, file_paths=file_paths) + preview_block = self.build_preview_block(file_paths) + user_message = f"File previews:\n{preview_block}\n\nRequest: {input}" - user_request = ( - "You are an interpreter helping to analyze different questions, files and outputs generated from a series of LLMs." - f"The details of the current request: {input}" - "Please interpret the current request to generate a meaningful answer." - "Here's some instructions that you have to follow for acheiving the task:" - "1. For any file provided, analyse if and provide clear and brief information about it unless something else is asked." - "2. Only if a specific visualization (e.g., bar chart, diagram) is requested in the question, use the provided information to generate a .json file containing the JSON code for a Plotly graph. This file should be named identically to the analyzed file." - "3. After you finish your tasks, your answer should contain both the interpretation asked and the full visualization file name if visualization was requested." + logger.info("Requesting analysis code from LLM") + try: + response = llm.invoke( + [SystemMessage(content=system_prompt), HumanMessage(content=user_message)] + ) + except Exception as exc: + logger.error(f"Interpreter LLM invocation failed: {exc}") + return "Interpreter could not generate analysis code for this request." + + response_content = self.normalize_response_content(response.content) + + code = self.extract_python_code(response_content) + if code is None: + logger.error("LLM did not return a Python code block") + return "Interpreter could not generate analysis code for this request." + + logger.info(f"Generated code:\n{code}") + + existing_files = self.capture_session_file_metadata(session_dir) + execution_result = self.execute_in_subprocess( + code=code, + session_dir=session_dir, + allowed_input_paths=file_paths, + ) + execution_output = execution_result.get("stdout", "").strip() + error_output = execution_result.get("error", "").strip() + + if error_output: + logger.warning(f"Interpreter subprocess reported an error: {error_output}") + + combined_output = execution_output or error_output or "Interpreter execution completed." + combined_output = self.truncate_output(combined_output) + logger.info(f"Execution output: {combined_output}") + + current_files = self.capture_session_file_metadata(session_dir) + changed_or_new_files = sorted( + str(path) + for path in self.find_changed_or_new_files(existing_files, current_files) + ) + logger.info(f"Files generated or updated: {changed_or_new_files}") + + output_data = {"output": {"paths": changed_or_new_files}} + try: + db_manager.put(data=json.dumps(output_data), tool_name="tool_interpreter") + except Exception as e: + logger.error(f"Error saving to database: {e}") + + suffix = ( + "\n\nThe full path of the files generated or updated are:\n" + "\n".join(changed_or_new_files) + if changed_or_new_files + else "" + ) + return f"{combined_output}{suffix}" + + def collect_allowed_file_paths(self, input_text: str, session_dir: Path, db_manager) -> list[Path]: + candidate_paths = self.extract_file_paths(input_text) + + if not candidate_paths: + candidate_paths = self.extract_paths_from_db(db_manager, "tool_merge_result") + + if not candidate_paths: + candidate_paths = self.extract_paths_from_db(db_manager, "tool_sparql") + + validated_paths: list[Path] = [] + rejected_paths: list[str] = [] + + for raw_path in candidate_paths: + try: + validated_path = resolve_session_path(raw_path, session_dir=session_dir, must_exist=True) + if not validated_path.is_file(): + raise ValueError(f"Path '{raw_path}' is not a file.") + if validated_path.suffix.lower() not in _ALLOWED_INPUT_SUFFIXES: + raise ValueError( + f"Unsupported file type '{validated_path.suffix}'. " + f"Allowed types are: {sorted(_ALLOWED_INPUT_SUFFIXES)}." + ) + if validated_path not in validated_paths: + validated_paths.append(validated_path) + except ValueError as exc: + rejected_paths.append(str(exc)) + + for rejected_path in rejected_paths: + logger.warning(f"Rejected interpreter input path: {rejected_path}") + + return validated_paths + + def extract_paths_from_db(self, db_manager, tool_name: str) -> list[str]: + payload = db_manager.get(tool_name) + if payload is None: + return [] + + try: + payload_json = json.loads(payload) + except json.JSONDecodeError as exc: + logger.warning(f"Could not decode tool payload for {tool_name}: {exc}") + return [] + + if not isinstance(payload_json, dict): + logger.warning( + "Unexpected payload shape for %s: top-level payload type is %s", + tool_name, + type(payload_json).__name__, + ) + return [] + + output = payload_json.get("output") + if isinstance(output, dict): + output_paths = output.get("paths", []) + if isinstance(output_paths, list): + return output_paths + logger.warning( + "Unexpected payload shape for %s: output.paths type is %s", + tool_name, + type(output_paths).__name__, + ) + return [] + + top_level_paths = payload_json.get("paths") + if isinstance(top_level_paths, list): + return top_level_paths + + logger.warning( + "Unexpected payload shape for %s: output type is %s and top-level paths type is %s", + tool_name, + type(output).__name__, + type(top_level_paths).__name__, + ) + return [] + + def build_system_prompt(self, session_dir: Path, file_paths: list[Path]) -> str: + allowed_paths = "\n".join(f"- {path}" for path in file_paths) + return ( + "You are a Python data analysis assistant.\n" + "Write a self-contained Python script that reads the provided file(s), " + "fulfils the user's request, and prints a concise text summary to stdout.\n\n" + "Rules:\n" + "- pandas, numpy, json, pathlib, plotly, math, statistics, csv, itertools, collections, and re are available.\n" + "- Read only the explicitly provided session files listed below.\n" + "- Save any output files only inside the session directory shown below.\n" + "- Do not attempt network access, subprocess execution, package installation, or filesystem access outside the session directory.\n" + "- Use the exact column names shown in the file previews; do not guess them.\n" + "- Print a SHORT summary (counts, stats, top-N rows); never print an entire dataframe.\n" + f"- Session directory for outputs: {session_dir}\n" + f"- Allowed input files:\n{allowed_paths}\n" + "- For visualizations use Plotly and write JSON with: " + f'fig.write_json(str(Path("{session_dir}") / ".json"))\n' + "- Do NOT call plt.show(), fig.show(), or any interactive display.\n" + "- Return ONLY the Python code inside a single ```python ... ``` block." + ) + + def build_preview_block(self, file_paths: list[Path]) -> str: + file_previews = [] + for file_path in file_paths: + try: + file_previews.append(self.preview_file(file_path)) + except Exception as exc: + logger.warning(f"Could not preview {file_path}: {exc}") + file_previews.append(f"File: {file_path} (could not preview)") + return "\n\n".join(file_previews) if file_previews else "(no files)" + + def preview_file(self, file_path: Path) -> str: + suffix = file_path.suffix.lower() + if suffix in {".csv", ".tsv", ".txt"}: + return self.preview_delimited_file(file_path) + if suffix in {".xls", ".xlsx"}: + preview = pd.read_excel(file_path, nrows=3) + return self.format_dataframe_preview(file_path, preview) + return self.preview_text_file(file_path) + + def preview_delimited_file(self, file_path: Path) -> str: + sep = "\t" if file_path.suffix.lower() in {".tsv", ".txt"} else "," + preview = pd.read_csv(file_path, sep=sep, nrows=3) + return self.format_dataframe_preview(file_path, preview) + + def format_dataframe_preview(self, file_path: Path, preview: pd.DataFrame) -> str: + if hasattr(preview, "map"): + preview = preview.map(self.truncate_cell) + else: + preview = preview.applymap(self.truncate_cell) + return ( + f"File: {file_path}\n" + f"Columns: {preview.columns.tolist()}\n" + f"Dtypes: {preview.dtypes.astype(str).to_dict()}\n" + f"Preview (3 rows):\n{preview.to_string(index=False)}" + ) + + def preview_text_file(self, file_path: Path) -> str: + preview_lines = [] + with file_path.open("r", encoding="utf-8", errors="replace") as handle: + for index, line in enumerate(handle): + if index >= 10: + break + preview_lines.append(line.rstrip("\n")[:200]) + + preview_body = "\n".join(preview_lines) if preview_lines else "(empty file)" + return f"File: {file_path}\nPreview (10 lines max):\n{preview_body}" + + def execute_in_subprocess( + self, + code: str, + session_dir: Path, + allowed_input_paths: list[Path], + ) -> dict[str, Any]: + runner_path = Path(__file__).with_name("sandbox_runner.py") + timeout_seconds = get_interpreter_timeout_seconds() + + payload = { + "code": code, + "session_dir": str(session_dir), + "allowed_input_paths": [str(path) for path in allowed_input_paths], + "timeout_seconds": timeout_seconds, + "cpu_seconds": get_interpreter_cpu_seconds(), + "memory_bytes": get_interpreter_memory_bytes(), + "file_size_bytes": get_interpreter_max_file_bytes(), + } + + with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as payload_file: + json.dump(payload, payload_file) + payload_path = Path(payload_file.name) + + with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as result_file: + result_path = Path(result_file.name) + + process = None + try: + process = subprocess.Popen( + [sys.executable, "-I", str(runner_path), "--payload", str(payload_path), "--result", str(result_path)], + cwd=session_dir, + env={ + "HOME": str(session_dir), + "PYTHONIOENCODING": "utf-8", + "TMPDIR": str(session_dir), + }, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + start_new_session=True, ) - files = [] - - for file in file_paths: - files.append(File.from_path(file)) - logger.info(f"File added to interpreter Agent: {file}") - - if not files: - - logger.info("No files provided from the Interpreter Agent. Manually scrapping the database") - output_merged = db_manager.get("tool_merge_result") # Get all file paths from the database associated with the tool_interpreter - if output_merged is not None: - output_merged_json = json.loads(output_merged) - merged_filepaths = output_merged_json['output']['paths'] - - for merged_filepath in merged_filepaths: - logger.info(f"File added to interpreter Agent from the Output Merged tool: {merged_filepath}") - files.append(File.from_path(merged_filepath)) - - if not files: - - sparql_output = db_manager.get("tool_sparql") # Get all file paths from the database associated with the tool_interpreter - if sparql_output is not None: - sparql_output_json = json.loads(sparql_output) - sparql_filepaths = sparql_output_json['output']['paths'] - - for sparql_filepath in sparql_filepaths: - logger.info(f"File added to interpreter Agent from the SPARQL tool: {sparql_filepath}") - files.append(File.from_path(sparql_filepath)) - - # generate the response - logger.info(f"Files submitted: {files}") - response = session.generate_response(user_request, files=files) - logger.info(f"Interpreter Agent Response: {response}") - - filepaths = [] - - # Handling and saving output files - if response.files: - for file in response.files: - logger.info(f"File: {file.name}") - - generated_file_path = session_dir / file.name - with open(generated_file_path, 'wb') as f: - f.write(file.content) - filepaths.append(str(generated_file_path)) - - logger.info(f"File saved: {file.name}") - else: - logger.info("No files generated by Interpreter Tool.") - - output_data = { - "output": { - "paths": filepaths - } + stdout, stderr = process.communicate(timeout=timeout_seconds + 2) + + result: dict[str, Any] | None = None + if result_path.exists(): + try: + result = json.loads(result_path.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + logger.warning(f"Could not decode sandbox result: {exc}") + + if result is None: + error_message = stderr.strip() or stdout.strip() or "Interpreter execution failed." + return {"success": False, "stdout": "", "error": error_message} + + if process.returncode != 0 and not result.get("error"): + result["error"] = stderr.strip() or "Interpreter subprocess failed." + + return result + except subprocess.TimeoutExpired: + if process is not None: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + process.kill() + process.wait() + return { + "success": False, + "stdout": "", + "error": f"Interpreter execution timed out after {timeout_seconds} seconds.", } - - - try: - db_manager.put(data=json.dumps(output_data), tool_name="tool_interpreter") - except Exception as e: - logger.error(f"Error saving to database: {e}") - - return f"{response}.\n\n The full path of the files generated are:\n" + "\n".join(filepaths) - - def extract_file_paths(self, text: str): - # Regex to find file paths or filenames with extensions, possibly surrounded by quotes - regex = r"['\"]?([a-zA-Z0-9_/\\-]+(?:\.csv|\.tsv|\.mgf|\.txt|\.xlsx|\.xls))['\"]?" - matches = re.finditer(regex, text) - file_paths = [match.group(1).replace("'", "").replace('"', '') for match in matches] - return file_paths \ No newline at end of file + finally: + payload_path.unlink(missing_ok=True) + result_path.unlink(missing_ok=True) + + def extract_python_code(self, text: str) -> str | None: + code_blocks = re.findall(r"```python\s*(.*?)\s*```", text, re.DOTALL) + if not code_blocks: + code_blocks = re.findall(r"```\s*(.*?)\s*```", text, re.DOTALL) + return code_blocks[0] if code_blocks else None + + def normalize_response_content(self, content: Any) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + return "\n".join( + item if isinstance(item, str) else json.dumps(item, ensure_ascii=True, default=str) + for item in content + ) + if isinstance(content, dict): + return json.dumps(content, ensure_ascii=True, default=str) + return str(content) + + def extract_file_paths(self, text: str) -> list[str]: + candidates: list[str] = [] + quoted_spans: list[tuple[int, int]] = [] + + for match in _QUOTED_FILE_PATH_PATTERN.finditer(text): + candidates.append(match.group("path")) + quoted_spans.append(match.span()) + + for match in _UNQUOTED_FILE_PATH_PATTERN.finditer(text): + if self.is_overlapping_span(match.span(), quoted_spans): + continue + candidates.append(match.group("path")) + + return self.deduplicate_paths(candidates) + + def capture_session_file_metadata(self, session_dir: Path) -> dict[Path, tuple[int, int]]: + metadata: dict[Path, tuple[int, int]] = {} + for path in session_dir.rglob("*"): + if not path.is_file(): + continue + stat = path.stat() + metadata[path.resolve(strict=False)] = (stat.st_mtime_ns, stat.st_size) + return metadata + + def find_changed_or_new_files( + self, + previous_metadata: dict[Path, tuple[int, int]], + current_metadata: dict[Path, tuple[int, int]], + ) -> set[Path]: + return { + path + for path, metadata in current_metadata.items() + if previous_metadata.get(path) != metadata + } + + def truncate_output(self, output: str) -> str: + if len(output) <= _MAX_OUTPUT_CHARS: + return output + return output[:_MAX_OUTPUT_CHARS] + f"\n... [output truncated at {_MAX_OUTPUT_CHARS} chars]" + + @staticmethod + def truncate_cell(value: Any) -> Any: + if isinstance(value, str) and len(value) > 80: + return value[:80] + "..." + return value + + @staticmethod + def deduplicate_paths(paths: list[str]) -> list[str]: + deduplicated_paths: list[str] = [] + for path in paths: + if path not in deduplicated_paths: + deduplicated_paths.append(path) + return deduplicated_paths + + @staticmethod + def is_overlapping_span(span: tuple[int, int], other_spans: list[tuple[int, int]]) -> bool: + return any(span[0] < other[1] and other[0] < span[1] for other in other_spans) diff --git a/app/core/agents/interpreter/tool_spectrum.py b/app/core/agents/interpreter/tool_spectrum.py index 5289749..65ae373 100644 --- a/app/core/agents/interpreter/tool_spectrum.py +++ b/app/core/agents/interpreter/tool_spectrum.py @@ -1,7 +1,7 @@ from __future__ import annotations from pathlib import Path from typing import Any, Dict, Optional, Union -from langchain.pydantic_v1 import BaseModel, Field +from pydantic import BaseModel, Field from langchain.tools import BaseTool from langchain.callbacks.manager import CallbackManagerForToolRun from app.core.memory.database_manager import tools_database @@ -29,7 +29,7 @@ class SpectrumPlotter(BaseTool): Returns: An url with spectrum plot. """ - args_schema = SpectrumPlotInput + args_schema: type[BaseModel] = SpectrumPlotInput session_id: str = None openai_key: str = None def __init__(self, openai_key: str, session_id: str): diff --git a/app/core/agents/sparql/agent.py b/app/core/agents/sparql/agent.py index 013e32f..face7a5 100644 --- a/app/core/agents/sparql/agent.py +++ b/app/core/agents/sparql/agent.py @@ -11,7 +11,7 @@ logger = setup_logger(__name__) -def create_agent(llms, graph, session_id, llm_instance=None) -> AgentExecutor: +def create_agent(llms, graph, session_id, openai_key=None, llm_instance=None) -> AgentExecutor: logger.info("Creating agent with tools...") directory = os.path.dirname(__file__) module_prefix = get_module_prefix(__name__) @@ -19,7 +19,8 @@ def create_agent(llms, graph, session_id, llm_instance=None) -> AgentExecutor: tool_parameters = { "graph": graph, "llm": llms, - "session_id": session_id + "session_id": session_id, + "openai_key": openai_key, } tools = import_tools(directory, module_prefix, **tool_parameters) diff --git a/app/core/agents/sparql/tool_merge_result.py b/app/core/agents/sparql/tool_merge_result.py index 0336891..4264102 100644 --- a/app/core/agents/sparql/tool_merge_result.py +++ b/app/core/agents/sparql/tool_merge_result.py @@ -1,4 +1,4 @@ -from langchain.pydantic_v1 import BaseModel, Field +from pydantic import BaseModel, Field from langchain.tools import BaseTool from typing import Optional @@ -40,7 +40,7 @@ class OutputMerger(BaseTool): output_file_path (str): The path to the temporary CSV file where the output with common Wikidata IDs will be saved. """ - args_schema = MergerInput + args_schema: type[BaseModel] = MergerInput session_id: str = None def __init__(self, session_id: str): diff --git a/app/core/agents/sparql/tool_sparql.py b/app/core/agents/sparql/tool_sparql.py index a761a5c..8721f67 100644 --- a/app/core/agents/sparql/tool_sparql.py +++ b/app/core/agents/sparql/tool_sparql.py @@ -10,7 +10,7 @@ from langchain.chains.llm import LLMChain from langchain_core.prompts.prompt import PromptTemplate -from langchain.pydantic_v1 import BaseModel, Field +from pydantic import BaseModel, Field from langchain.tools import BaseTool from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings @@ -207,8 +207,8 @@ class SparqlInput(BaseModel): ##Question-answering against an RDF or OWL graph by generating SPARQL statements. class GraphSparqlQAChain(BaseTool): - name = "SPARQL_QUERY_RUNNER" - description = """ + name: str = "SPARQL_QUERY_RUNNER" + description: str = """ The agent resolve the user's question by querying the knowledge graph database. The two inputs should be a string containing the user's question and a string containing the resolved entities in the question. @@ -226,14 +226,22 @@ class GraphSparqlQAChain(BaseTool): """ verbose: bool = True - args_schema = SparqlInput + args_schema: type[BaseModel] = SparqlInput sparql_generation_select_chain: LLMChain = None sparql_improvement_chain: LLMChain = None - requires_params = True + requires_params: bool = True graph: RdfGraph = None session_id: str = None + openai_key: Optional[str] = None - def __init__(self, llm: dict, graph: RdfGraph, session_id: str, **kwargs): + def __init__( + self, + llm: dict, + graph: RdfGraph, + session_id: str, + openai_key: Optional[str] = None, + **kwargs, + ): super().__init__(**kwargs) try: self.sparql_generation_select_chain = LLMChain( @@ -250,6 +258,7 @@ def __init__(self, llm: dict, graph: RdfGraph, session_id: str, **kwargs): raise self.graph = graph self.session_id = session_id + self.openai_key = openai_key or os.getenv("OPENAI_API_KEY") def _run( self, @@ -385,7 +394,8 @@ def search_nodes(self, query): # Construct the path to the faiss_db directory db_path = os.path.abspath(os.path.join(current_dir, '..', '..', '..', 'data', 'faiss_db')) - embeddings = OpenAIEmbeddings() + embedding_kwargs = {"api_key": self.openai_key} if self.openai_key else {} + embeddings = OpenAIEmbeddings(**embedding_kwargs) db = FAISS.load_local(db_path, embeddings, allow_dangerous_deserialization=True) related_nodes = db.similarity_search(query, 12) return related_nodes diff --git a/app/core/agents/sparql/tool_wikidata_query.py b/app/core/agents/sparql/tool_wikidata_query.py index 8fac383..64a6906 100644 --- a/app/core/agents/sparql/tool_wikidata_query.py +++ b/app/core/agents/sparql/tool_wikidata_query.py @@ -3,13 +3,13 @@ import csv import tempfile from pathlib import Path -from typing import List, Optional +from typing import ClassVar, List, Optional from langchain_core.tools import tool from SPARQLWrapper import JSON, SPARQLWrapper from app.core.session import setup_logger, create_user_session -from langchain.pydantic_v1 import BaseModel, Field +from pydantic import BaseModel, Field from langchain.tools import BaseTool from langchain.callbacks.manager import ( CallbackManagerForToolRun, @@ -33,10 +33,10 @@ class WikidataStructureSearch(BaseTool): Returns: str: A string that contains the path to the file with Wikidata IRIs if found, otherwise `None`. """ - args_schema = WikidataInput + args_schema: type[BaseModel] = WikidataInput - ENDPOINT_URL = "https://query.wikidata.org/sparql" - PREFIXES = """ + ENDPOINT_URL: ClassVar[str] = "https://query.wikidata.org/sparql" + PREFIXES: ClassVar[str] = """ PREFIX wd: PREFIX wdt: """ diff --git a/app/core/agents/supervisor/agent.py b/app/core/agents/supervisor/agent.py index 0dfa69a..e875924 100644 --- a/app/core/agents/supervisor/agent.py +++ b/app/core/agents/supervisor/agent.py @@ -10,13 +10,12 @@ logger = setup_logger(__name__) -config = load_config() - def create_agent(llms, graph, llm_instance=None) -> AgentExecutor: """Configure and return a supervisor agent with decision-making logic for task routing.""" llm = llm_instance if llm_instance is not None else llms[MODEL_CHOICE] + config = load_config() members = config["supervisor"]["members"] options = ["FINISH"] + members diff --git a/app/core/agents/supervisor/prompt.py b/app/core/agents/supervisor/prompt.py index 47915e6..bf18fb9 100644 --- a/app/core/agents/supervisor/prompt.py +++ b/app/core/agents/supervisor/prompt.py @@ -13,21 +13,24 @@ If the Sparql_query_runner does not provide the path to the file, it means that there is no answer to the question. In that case, mark the process as FINISH. -If the Sparql_query_runner provides a SPARQL query and the path to the file containing the SPARQL output without directly providing the answer (implying that the answer is too long to be directly included), then delegate this information to the Interpreter_agent for further analysis and interpretation. Provide the Interpreter_agent with the question, SPARQL query, and the path to the file provided by the Sparql_query_runner. Await the Interpreter_agent's response for the final answer. +Only delegate work to Interpreter_agent when Interpreter_agent appears in the available team members list. +If Interpreter_agent is not available, do not attempt file interpretation or chart generation. Use the SPARQL answer directly when possible, otherwise explain that interpreter features are disabled in this deployment. + +If the Sparql_query_runner provides a SPARQL query and the path to the file containing the SPARQL output without directly providing the answer (implying that the answer is too long to be directly included), then delegate this information to the Interpreter_agent for further analysis and interpretation only when Interpreter_agent is available. Provide the Interpreter_agent with the question, SPARQL query, and the path to the file provided by the Sparql_query_runner. Await the Interpreter_agent's response for the final answer. Once the Interpreter_agent has completed its task mark the process as FINISH. Do not call the Interpreter_agent again. -If the Sparql_query_runner agent provides a SPARQL query, the path to the file containing the SPARQL output and final answer to the question, and there is no immediate need for further interpretation, normally mark the process as FINISH. However, if there is a need to visualize the results (regardless of the length of the SPARQL output), also call the Interpreter_agent to generate the necessary plot, chart, or graph based on the SPARQL output. The need for visualization should be assessed based on the user's request or if the nature of the data implies that visualization would enhance understanding. Once the Interpreter_agent has completed its task mark the process as FINISH. Do not call the Interpreter_agent again. +If the Sparql_query_runner agent provides a SPARQL query, the path to the file containing the SPARQL output and final answer to the question, and there is no immediate need for further interpretation, normally mark the process as FINISH. However, if there is a need to visualize the results (regardless of the length of the SPARQL output), call the Interpreter_agent only when it is available to generate the necessary plot, chart, or graph based on the SPARQL output. The need for visualization should be assessed based on the user's request or if the nature of the data implies that visualization would enhance understanding. If Interpreter_agent is unavailable, explain that visualization features are disabled in this deployment and finish with the available SPARQL answer. -If the question provided by the user does mention a figure, chart or visualization of the mass spectra you should call the Interpreter_agent to generate the necessary plot, chart, graph or visualization of the mass spectra based on the SPARQL output. Once the Interpreter_agent has completed its task mark the process as FINISH. Do not call the Interpreter_agent again. +If the question provided by the user does mention a figure, chart or visualization of the mass spectra you should call the Interpreter_agent to generate the necessary plot, chart, graph or visualization of the mass spectra based on the SPARQL output only when Interpreter_agent is available. If it is unavailable, explain that visualization features are disabled in this deployment and finish with the available non-visual answer. -For example, the user provides the following question: For features from Melochia umbellata in PI mode with SIRIUS annotations, get the ones for which a feature in NI mode with the same retention time has the same SIRIUS annotation. Since the question mentions Melochia umbellata you should firstly delegate it to the ENPKG_agent which would provide wikidata IRI with TAXON_RESOLVER tool, then, you should delegate the question together with the output generated by ENPKG_agent to the Sparql_query_runner agent. Afterwards, if the Sparql_query_runner agent provided the answer to the question, SPARQL query and path to the file containing the SPARQL output and there is no need to visualize the output you should mark the process as FINISH. If the Sparql_query_runner agent provided only SPARQL query and path to the file you should call Interpreter_agent which would interpret the results provided by Sparql_query_runner to generate the final response to the question. +For example, the user provides the following question: For features from Melochia umbellata in PI mode with SIRIUS annotations, get the ones for which a feature in NI mode with the same retention time has the same SIRIUS annotation. Since the question mentions Melochia umbellata you should firstly delegate it to the ENPKG_agent which would provide wikidata IRI with TAXON_RESOLVER tool, then, you should delegate the question together with the output generated by ENPKG_agent to the Sparql_query_runner agent. Afterwards, if the Sparql_query_runner agent provided the answer to the question, SPARQL query and path to the file containing the SPARQL output and there is no need to visualize the output you should mark the process as FINISH. If the Sparql_query_runner agent provided only SPARQL query and path to the file you should call Interpreter_agent only when it is available; otherwise explain that interpreter features are disabled in this deployment and finish with the SPARQL answer. Avoid calling the same agent if this agent has already been called previously and provided the answer. For example, if you have called ENPKG_agent and it provided the output do not call this agent again. Always tell the user the SPARQL query that has been returned by the Sparql_query_runner. -If interpretation of a file is needed, delegate the task directly to interpreter agent unless it needs to resolve any chemical entity or retrieve extra information from sparql_query_runner. In this case, delegate the task to the aproppriate agent. +If interpretation of a file is needed, delegate the task directly to interpreter agent only when it is available, unless it needs to resolve any chemical entity or retrieve extra information from sparql_query_runner. In this case, delegate the task to the appropriate agent. If the agent does not provide the expected output mark the process as FINISH. diff --git a/app/core/agents/validator/tool_validator.py b/app/core/agents/validator/tool_validator.py index 4d57c5f..a4bc652 100644 --- a/app/core/agents/validator/tool_validator.py +++ b/app/core/agents/validator/tool_validator.py @@ -5,7 +5,7 @@ from langchain.callbacks.manager import ( CallbackManagerForToolRun, ) -from langchain.pydantic_v1 import BaseModel, Field +from pydantic import BaseModel, Field from langchain.tools import BaseTool from typing import Optional @@ -32,7 +32,7 @@ class PlantDatabaseChecker(BaseTool): """ - args_schema = PlantInput + args_schema: type[BaseModel] = PlantInput def __init__(self): super().__init__() diff --git a/app/core/main.py b/app/core/main.py index 79dac70..30d4e12 100644 --- a/app/core/main.py +++ b/app/core/main.py @@ -1,15 +1,18 @@ import os import shutil import argparse +import configparser from typing import Optional, List +from pathlib import Path + from dotenv import load_dotenv from langsmith import Client -from pathlib import Path -from app.core.session import create_user_session, initialize_session_context +from langchain_community.chat_models import ChatLiteLLM +from langchain_openai import ChatOpenAI + from app.core.workflow.langraph_workflow import create_workflow, process_workflow +from app.core.session import create_user_session, initialize_session_context from app.core.utils import IntRange, setup_logger -import configparser -from langchain_community.chat_models import ChatOpenAI, ChatLiteLLM from app.core.questions import standard_questions @@ -42,6 +45,14 @@ def __init__(self, source_path: Path, message: str): +class SessionFilePreparationError(ValueError): + """Raised when CLI input files cannot be staged into the session directory.""" + + def __init__(self, source_path: Path, message: str): + super().__init__(message) + self.source_path = source_path + + def get_api_key(provider: str) -> Optional[str]: """ Get API key for specified provider from environment variables. @@ -277,34 +288,54 @@ def _prepare_session_files(session_id: str, file_paths: List[str]) -> Path: return input_dir + def main(): - """Main function to run the workflow.""" - # Define command line arguments - - parser = argparse.ArgumentParser(description="Process a workflow with a predefined question number.") - parser.add_argument('-q', '--question', type=int, choices=IntRange(1, len(standard_questions)), - help=f"Choose a standard question number from 1 to {len(standard_questions)}.") - parser.add_argument('-c', '--custom', type=str, - help="Provide a custom question.") + """ + CLI entry-point for running the MetaboT workflow. + + Usage examples: + python -m app.core.main -q 1 + python -m app.core.main -c "Describe my dataset" -f data.csv + python -m app.core.main -c "Compare files" -f file1.csv file2.tsv + """ + parser = argparse.ArgumentParser( + description="Process a workflow with a predefined question number." + ) + parser.add_argument( + '-q', '--question', type=int, + choices=IntRange(1, len(standard_questions)), + help=f"Choose a standard question number from 1 to {len(standard_questions)}.", + ) + parser.add_argument( + '-c', '--custom', type=str, + help="Provide a custom question.", + ) parser.add_argument( '-f', '--file', type=str, nargs='+', help="One or more local file paths to make available for the FILE_ANALYZER tool.", ) - parser.add_argument('-e', '--evaluation', action='store_true', - help="Enable evaluation mode") - parser.add_argument('--api-key', type=str, - help="OpenAI API key (optional, defaults to environment variable)") - parser.add_argument('--endpoint', type=str, - help="Knowledge graph endpoint URL (optional)") + parser.add_argument( + '-e', '--evaluation', action='store_true', + help="Enable evaluation mode.", + ) + parser.add_argument( + '--api-key', type=str, + help="OpenAI API key (optional, defaults to environment variable).", + ) + parser.add_argument( + '--endpoint', type=str, + help="Knowledge graph endpoint URL (optional).", + ) args = parser.parse_args() + # Resolve the question if args.question: question = standard_questions[args.question - 1] elif args.custom: question = args.custom else: - print("You must provide either a standard question number or a custom question.") + print("You must provide either a standard question number (-q) or a custom question (-c).") return # Create a user session (mirrors the Streamlit session lifecycle) and @@ -314,35 +345,36 @@ def main(): global logger logger = setup_logger(__name__) - if args.file: - try: - _prepare_session_files(session_id, args.file) - except SessionFilePreparationError as exc: - logger.error(str(exc)) - print(f"Error: {exc}") - return - # Initialize LangSmith if available langsmith_setup() - # Get endpoint URL from arguments or environment + # Resolve endpoint URL endpoint_url = ( args.endpoint or os.environ.get("KG_ENDPOINT_URL") or "https://enpkg.commons-lab.org/graphdb/repositories/ENPKG" ) + + # Initialize language models models = llm_creation(api_key=args.api_key) + # Stage user-provided files into the session's input directory + if args.file: + try: + _prepare_session_files(session_id, args.file) + except SessionFilePreparationError as exc: + logger.error(str(exc)) + print(f"Error: {exc}") + return + try: - # Create and process workflow workflow = create_workflow( models=models, session_id=session_id, endpoint_url=endpoint_url, evaluation=False, - api_key=args.api_key + api_key=args.api_key, ) - process_workflow(workflow, question) except Exception as e: diff --git a/app/core/memory/custom_sqlite_file.py b/app/core/memory/custom_sqlite_file.py deleted file mode 100644 index cbb93de..0000000 --- a/app/core/memory/custom_sqlite_file.py +++ /dev/null @@ -1,97 +0,0 @@ -# This is a custom sqlite checkpoint for langgraph based on the orininal one https://github.com/langchain-ai/langgraph/blob/main/langgraph/checkpoint/sqlite.py commit @f2ad930 https://github.com/langchain-ai/langgraph/commit/f2ad930cd4cf383ccaefd18d497aa2e0f459e5bd -# This custom sqlite allows multithreading access, necessary for running the langgraph with memory in Streamlit. - -import os -import pickle -import sqlite3 -import threading -from contextlib import contextmanager -from typing import Optional - -from langchain_core.pydantic_v1 import Field -from langchain_core.runnables import RunnableConfig -from langchain_core.runnables.utils import ConfigurableFieldSpec -from langgraph.checkpoint.base import BaseCheckpointSaver, Checkpoint - -from app.core.session import setup_logger - -logger = setup_logger(__name__) - - -class SqliteCheckpointerSaver(BaseCheckpointSaver): - # Specify the path to the database file - database_path = "langgraph_checkpoint.db" - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) # Ensure proper Pydantic initialization - self.__dict__["_local"] = threading.local() - self._initialize_database() - - def _initialize_database(self): - # Check if the database file exists and delete it if so - if os.path.exists(self.database_path): - os.remove(self.database_path) - # Establish a new database connection to the file - self._get_connection(force_new=True) - - def _get_connection(self, force_new=False): - if not hasattr(self._local, "connection") or force_new: - self._local.connection = sqlite3.connect( - self.database_path, check_same_thread=False - ) - else: - pass - return self._local.connection - - def setup(self): - if hasattr(self._local, "is_setup") and self._local.is_setup: - return - - conn = self._get_connection() - conn.executescript( - """ - CREATE TABLE IF NOT EXISTS checkpoints ( - thread_id TEXT PRIMARY KEY, - checkpoint BLOB - ); - """ - ) - self._local.is_setup = True - - @contextmanager - def cursor(self, transaction: bool = True): - self.setup() - conn = self._get_connection() - cur = conn.cursor() - try: - yield cur - finally: - if transaction: - conn.commit() - cur.close() - - def get(self, config: RunnableConfig) -> Optional[Checkpoint]: - with self.cursor(transaction=False) as cur: - cur.execute( - "SELECT checkpoint FROM checkpoints WHERE thread_id = ?", - (config["configurable"]["thread_id"],), - ) - row = cur.fetchone() - return pickle.loads(row[0]) if row else None - - def put(self, config: RunnableConfig, checkpoint: Checkpoint) -> None: - with self.cursor() as cur: - cur.execute( - "INSERT OR REPLACE INTO checkpoints (thread_id, checkpoint) VALUES (?, ?)", - (config["configurable"]["thread_id"], pickle.dumps(checkpoint)), - ) - - # For debugging - def print_database_contents(self): - with self.cursor(transaction=False) as cur: - cur.execute("SELECT thread_id, checkpoint FROM checkpoints") - rows = cur.fetchall() - for row in rows: - print( - f"Thread ID: {row[0]}, Checkpoint (Raw): {row[1][:100]}..." - ) # Print a portion of the checkpoint to avoid flooding the output diff --git a/app/core/memory/database_manager.py b/app/core/memory/database_manager.py index 6a9a67c..b3bcf05 100644 --- a/app/core/memory/database_manager.py +++ b/app/core/memory/database_manager.py @@ -1,39 +1,40 @@ -import os -from app.core.memory.tools_database import SqliteToolsDatabaseManager -from app.core.memory.custom_sqlite_file import SqliteCheckpointerSaver -import importlib - -def tools_database(): - """ - Get the database manager based on the environment variables - - :return: The database manager - """ - if os.getenv('DATABASE_URL') and os.getenv("TOOLS_DATABASE_MANAGER_CLASS"): - class_path = os.getenv('TOOLS_DATABASE_MANAGER_CLASS') - if class_path: - # Dynamically import the module and class based on the environment variable - module_name, class_name = class_path.rsplit('.', 1) - module = importlib.import_module(module_name) - manager_class = getattr(module, class_name) - return manager_class() - else: - return SqliteToolsDatabaseManager() - -def memory_database(): - """ - Get the database manager based on the environment variables - - :return: The database manager - """ - - if os.getenv('DATABASE_URL') and os.getenv("MEMORY_DATABASE_MANAGER_CLASS"): - class_path = os.getenv('MEMORY_DATABASE_MANAGER_CLASS') - if class_path: - # Dynamically import the module and class based on the environment variable - module_name, class_name = class_path.rsplit('.', 1) - module = importlib.import_module(module_name) - manager_class = getattr(module, class_name) - return manager_class() - else: - return SqliteCheckpointerSaver() \ No newline at end of file +import os +from app.core.memory.tools_database import SqliteToolsDatabaseManager +import importlib + +def tools_database(): + """ + Get the database manager based on the environment variables + + :return: The database manager + """ + if os.getenv('DATABASE_URL') and os.getenv("TOOLS_DATABASE_MANAGER_CLASS"): + class_path = os.getenv('TOOLS_DATABASE_MANAGER_CLASS') + if class_path: + # Dynamically import the module and class based on the environment variable + module_name, class_name = class_path.rsplit('.', 1) + module = importlib.import_module(module_name) + manager_class = getattr(module, class_name) + return manager_class() + else: + return SqliteToolsDatabaseManager() + +def memory_database(): + """ + Get the database manager based on the environment variables + + :return: The database manager + """ + + if os.getenv('DATABASE_URL') and os.getenv("MEMORY_DATABASE_MANAGER_CLASS"): + class_path = os.getenv('MEMORY_DATABASE_MANAGER_CLASS') + if class_path: + # Dynamically import the module and class based on the environment variable + module_name, class_name = class_path.rsplit('.', 1) + module = importlib.import_module(module_name) + manager_class = getattr(module, class_name) + return manager_class() + else: + # Use langgraph's built-in MemorySaver which implements the current checkpoint API + from langgraph.checkpoint.memory import MemorySaver + return MemorySaver() \ No newline at end of file diff --git a/app/core/security.py b/app/core/security.py new file mode 100644 index 0000000..561010f --- /dev/null +++ b/app/core/security.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import os +from pathlib import Path + + +_TRUE_VALUES = {"1", "true", "yes", "on", "y", "t"} + + +def env_flag(name: str, default: bool = False) -> bool: + value = os.getenv(name) + if value is None: + return default + return value.strip().lower() in _TRUE_VALUES + + +def is_trusted_mode_enabled() -> bool: + """Trusted mode opt-in for features that execute generated code.""" + return env_flag("METABOT_TRUSTED_MODE", default=False) + + +def get_interpreter_timeout_seconds() -> int: + return max(1, int(os.getenv("METABOT_INTERPRETER_TIMEOUT_SECONDS", "15"))) + + +def get_interpreter_cpu_seconds() -> int: + timeout_seconds = get_interpreter_timeout_seconds() + return max(1, int(os.getenv("METABOT_INTERPRETER_CPU_SECONDS", str(timeout_seconds)))) + + +def get_interpreter_memory_bytes() -> int: + memory_mb = max(64, int(os.getenv("METABOT_INTERPRETER_MEMORY_MB", "512"))) + return memory_mb * 1024 * 1024 + + +def get_interpreter_max_file_bytes() -> int: + file_mb = max(8, int(os.getenv("METABOT_INTERPRETER_MAX_FILE_MB", "64"))) + return file_mb * 1024 * 1024 + + +def is_path_within_directory(path: Path, directory: Path) -> bool: + resolved_path = path.resolve(strict=False) + resolved_directory = directory.resolve(strict=False) + return resolved_path == resolved_directory or resolved_directory in resolved_path.parents + + +def resolve_session_path(path_value: str | Path, session_dir: Path, must_exist: bool = True) -> Path: + candidate = Path(path_value) + if not candidate.is_absolute(): + candidate = session_dir / candidate + + resolved = candidate.resolve(strict=False) + if not is_path_within_directory(resolved, session_dir): + raise ValueError( + f"Path '{path_value}' is outside the allowed session directory '{session_dir}'." + ) + + if must_exist and not resolved.exists(): + raise ValueError(f"Path '{path_value}' does not exist in the current session.") + + return resolved diff --git a/app/core/tests/evaluation.py b/app/core/tests/evaluation.py index f08ee3c..af56209 100644 --- a/app/core/tests/evaluation.py +++ b/app/core/tests/evaluation.py @@ -6,7 +6,6 @@ from dotenv import load_dotenv from langsmith import Client -from langsmith.utils import LangSmithNotFoundError from langchain_core.messages import HumanMessage from langsmith.evaluation import EvaluationResult, run_evaluator from langchain.evaluation import EvaluatorType, load_evaluator @@ -16,128 +15,134 @@ from app.core.main import llm_creation from app.core.utils import setup_logger +# 1. Load environment variables +load_dotenv() logger = setup_logger(__name__) -dataset_name = "benchmark_metabot" +# 2. Check for required API keys safely +api_key = os.getenv("LANGCHAIN_API_KEY") or os.environ.get("LANGSMITH_API_KEY") +openai_key = os.getenv("OPENAI_API_KEY") + +if not api_key: + raise ValueError("Missing LANGCHAIN_API_KEY (or LANGSMITH_API_KEY). Please add it to your repo-root .env (see docs/user-guide/configuration.md).") +if not openai_key: + raise ValueError("Missing OPENAI_API_KEY. Please copy .env.template to .env and add your key.") + +# Set environment variables for LangSmith +os.environ["LANGCHAIN_TRACING_V2"] = "true" +os.environ["LANGCHAIN_PROJECT"] = os.environ.get("LANGCHAIN_PROJECT", "MetaboT evaluation") +os.environ["LANGCHAIN_ENDPOINT"] = os.environ.get("LANGCHAIN_ENDPOINT", "https://api.smith.langchain.com") current_dir = os.path.dirname(os.path.abspath(__file__)) core_dir = os.path.dirname(current_dir) app_dir = os.path.dirname(core_dir) local_data_path = os.path.join(app_dir, "data", "big_benchmark.csv") +client = Client(api_key=api_key) + +dataset_name = "test_metabot" +# Check if the dataset exists in the current user's workspace +try: + client.read_dataset(dataset_name=dataset_name) + logger.info(f"Dataset '{dataset_name}' found in LangSmith workspace.") +except Exception: + logger.info(f"Dataset '{dataset_name}' not found. Attempting to create it from local file...") + + + if not os.path.exists(local_data_path): + raise FileNotFoundError( + f"Could not find '{local_data_path}'. Ensure the local dataset file is shared alongside this script.") + + + # Load local data and create the dataset in the user's LangSmith account + df = pd.read_csv(local_data_path) + dataset = client.create_dataset(dataset_name=dataset_name, description="MetaboT Benchmark") + + inputs = [] + outputs = [] + + for _, row in df.iterrows(): + try: + # The CSV stores the JSON arrays as strings, so we need to parse them back into Python objects + parsed_messages = json.loads(row["messages"]) + parsed_end_state = json.loads(row["__end__"]) + + inputs.append({"messages": parsed_messages}) + outputs.append({"__end__": parsed_end_state}) + + except json.JSONDecodeError as e: + logger.warning(f"Skipping a row due to JSON parsing error: {e}") + continue + + client.create_examples(inputs=inputs, outputs=outputs, dataset_id=dataset.id) + logger.info(f"Successfully created dataset '{dataset_name}' in LangSmith.") +# Custom criteria for SPARQL query evaluation +custom_criteria = { + "structural similarity of SPARQL queries": + "How similar is the structure of the generated SPARQL query to the reference SPARQL query? Does the generated query correctly match subjects to their corresponding objects as in the reference query" +} + +eval_chain_new = load_evaluator(EvaluatorType.LABELED_CRITERIA, criteria=custom_criteria) + +# Define the evaluation configuration +evaluation_config = RunEvalConfig( + evaluators=[ + EvaluatorType.QA, + RunEvalConfig.LabeledScoreString( + { + "accuracy": """ +Score 1: The answer is completely unrelated to the reference. +Score 3: The answer has minor relevance but does not align with the reference. +Score 5: The answer has moderate relevance but contains inaccuracies. +Score 7: The answer aligns with the reference but has minor errors or omissions. +Score 10: The answer is completely accurate and aligns perfectly with the reference.""" + }, + normalize_by=10, + ), + ], + custom_evaluators=[eval_chain_new], +) +endpoint_url = os.environ.get("KG_ENDPOINT_URL", "https://enpkg.commons-lab.org/graphdb/repositories/ENPKG") -def main(): - load_dotenv() - - api_key = os.getenv("LANGCHAIN_API_KEY") or os.environ.get("LANGSMITH_API_KEY") - openai_key = os.getenv("OPENAI_API_KEY") - - if not api_key: - raise ValueError("Missing LANGCHAIN_API_KEY. Please copy .env.template to .env and add your key.") - if not openai_key: - raise ValueError("Missing OPENAI_API_KEY. Please copy .env.template to .env and add your key.") - - os.environ["LANGCHAIN_TRACING_V2"] = "true" - os.environ["LANGCHAIN_PROJECT"] = os.environ.get("LANGCHAIN_PROJECT", "MetaboT evaluation") - os.environ["LANGCHAIN_ENDPOINT"] = os.environ.get("LANGCHAIN_ENDPOINT", "https://api.smith.langchain.com") - - client = Client(api_key=api_key) - - try: - client.read_dataset(dataset_name=dataset_name) - logger.info(f"Dataset '{dataset_name}' found in LangSmith workspace.") - except LangSmithNotFoundError: - logger.info(f"Dataset '{dataset_name}' not found. Attempting to create it from local file...") - - if not os.path.exists(local_data_path): - raise FileNotFoundError( - f"Could not find '{local_data_path}'. Ensure the local dataset file is shared alongside this script." - ) from None - - df = pd.read_csv(local_data_path) - dataset = client.create_dataset(dataset_name=dataset_name, description="MetaboT Benchmark") - - inputs = [] - outputs = [] - - for _, row in df.iterrows(): - try: - parsed_messages = json.loads(row["messages"]) - parsed_end_state = json.loads(row["__end__"]) +models = llm_creation() - inputs.append({"messages": parsed_messages}) - outputs.append({"__end__": parsed_end_state}) +# Create workflow in evaluation mode +app = create_workflow( + models=models, + endpoint_url=endpoint_url, + evaluation=True, + api_key=openai_key +) - except json.JSONDecodeError as e: - logger.warning(f"Skipping a row due to JSON parsing error: {e}") - continue - client.create_examples(inputs=inputs, outputs=outputs, dataset_id=dataset.id) - logger.info(f"Successfully created dataset '{dataset_name}' in LangSmith.") +def evaluate_result(_input, thread_id: int = 1): + """Evaluate the result based on input.""" + # Note: Adjust the 'question' key below if your dataset uses a different input key + input_text = _input.get("question") or _input["messages"][0]["content"] - custom_criteria = { - "structural similarity of SPARQL queries": - "How similar is the structure of the generated SPARQL query to the reference SPARQL query? Does the generated query correctly match subjects to their corresponding objects as in the reference query" + message = { + "messages": [HumanMessage(content=input_text)] } - eval_chain_new = load_evaluator(EvaluatorType.LABELED_CRITERIA, criteria=custom_criteria) + response = app.invoke(message, {"configurable": {"thread_id": thread_id}}) + return {"output": response} - evaluation_config = RunEvalConfig( - evaluators=[ - EvaluatorType.QA, - RunEvalConfig.LabeledScoreString( - { - "accuracy": """ -Score 1: The answer is completely unrelated to the reference. -Score 3: The answer has minor relevance but does not align with the reference. -Score 5: The answer has moderate relevance but contains inaccuracies. -Score 7: The answer aligns with the reference but has minor errors or omissions. -Score 10: The answer is completely accurate and aligns perfectly with the reference.""" - }, - normalize_by=10, - ), - ], - custom_evaluators=[eval_chain_new], - ) - - endpoint_url = os.environ.get("KG_ENDPOINT_URL", "https://enpkg.commons-lab.org/graphdb/repositories/ENPKG") - - models = llm_creation() - - app = create_workflow( - models=models, - endpoint_url=endpoint_url, - evaluation=True, - api_key=openai_key - ) - - def evaluate_result(_input, thread_id: int = 1): - """Evaluate the result based on input.""" - input_text = _input.get("question") or _input["messages"][0]["content"] - - message = { - "messages": [HumanMessage(content=input_text)] - } - - response = app.invoke(message, {"configurable": {"thread_id": thread_id}}) - return {"__end__": response} - - unique_id = uuid4().hex[0:8] - - chain_results = run_on_dataset( - dataset_name=dataset_name, - llm_or_chain_factory=evaluate_result, - evaluation=evaluation_config, - verbose=True, - project_name=f"Testing the app-{unique_id}", - client=client, - project_metadata={ - "model": "gpt-4o", - }, - ) - - logger.info(f"Evaluation complete. View results in LangSmith under project: Testing the app-{unique_id}") - - -if __name__ == "__main__": - main() \ No newline at end of file +def main(): + load_dotenv() + +unique_id = uuid4().hex[0:8] + +# Run evaluation +chain_results = run_on_dataset( + dataset_name=dataset_name, + llm_or_chain_factory=evaluate_result, + evaluation=evaluation_config, + verbose=True, + project_name=f"Testing the app-{unique_id}", + client=client, + project_metadata={ + "model": "gpt-4o", + }, +) + +logger.info(f"Evaluation complete. View results in LangSmith under project: Testing the app-{unique_id}") diff --git a/app/core/tests/test_interpreter_tool.py b/app/core/tests/test_interpreter_tool.py new file mode 100644 index 0000000..c40b938 --- /dev/null +++ b/app/core/tests/test_interpreter_tool.py @@ -0,0 +1,219 @@ +from __future__ import annotations + +import json +import time +from pathlib import Path + +from app.core.agents.interpreter.tool_interpreter import Interpreter + + +class StubDBManager: + def __init__(self, payloads: dict[str, str] | None = None): + self.payloads = payloads or {} + + def get(self, tool_name: str): + return self.payloads.get(tool_name) + + +class StubLLMResponse: + def __init__(self, content): + self.content = content + + +class StubLLM: + def __init__(self, content=None, error: Exception | None = None): + self.content = content + self.error = error + + def invoke(self, messages): + if self.error is not None: + raise self.error + return StubLLMResponse(self.content) + + +def build_interpreter() -> Interpreter: + return Interpreter(llm_instance=object()) + + +def test_extract_file_paths_allows_dots_in_file_stem(): + interpreter = build_interpreter() + + paths = interpreter.extract_file_paths("Use /tmp/session/foo.v1.csv for the analysis.") + + assert paths == ["/tmp/session/foo.v1.csv"] + + +def test_extract_file_paths_supports_quoted_paths_with_spaces(): + interpreter = build_interpreter() + + paths = interpreter.extract_file_paths('Filepath: "/tmp/session/my file.v1.csv"') + + assert paths == ["/tmp/session/my file.v1.csv"] + + +def test_extract_file_paths_supports_multiple_paths(): + interpreter = build_interpreter() + + paths = interpreter.extract_file_paths( + 'Compare "/tmp/session/my file.v1.csv" with /tmp/session/other_file.tsv' + ) + + assert paths == ["/tmp/session/my file.v1.csv", "/tmp/session/other_file.tsv"] + + +def test_collect_allowed_file_paths_falls_back_to_db_when_no_explicit_paths(tmp_path): + interpreter = build_interpreter() + session_dir = tmp_path / "session" + session_dir.mkdir() + input_file = session_dir / "from_db.csv" + input_file.write_text("a,b\n1,2\n", encoding="utf-8") + db_manager = StubDBManager( + { + "tool_sparql": json.dumps( + {"output": {"paths": [str(input_file)]}} + ) + } + ) + + paths = interpreter.collect_allowed_file_paths( + input_text="Interpret the latest results.", + session_dir=session_dir, + db_manager=db_manager, + ) + + assert paths == [input_file.resolve()] + + +def test_collect_allowed_file_paths_rejects_invalid_path_without_suppressing_valid_explicit_path(tmp_path): + interpreter = build_interpreter() + session_dir = tmp_path / "session" + session_dir.mkdir() + valid_file = session_dir / "valid file.csv" + valid_file.write_text("a,b\n1,2\n", encoding="utf-8") + outside_file = tmp_path / "outside.csv" + outside_file.write_text("a,b\n9,9\n", encoding="utf-8") + + paths = interpreter.collect_allowed_file_paths( + input_text=f'Use "{outside_file}" and "{valid_file}"', + session_dir=session_dir, + db_manager=StubDBManager(), + ) + + assert paths == [valid_file.resolve()] + + +def test_extract_paths_from_db_handles_unexpected_payload_shape(): + interpreter = build_interpreter() + + paths = interpreter.extract_paths_from_db( + StubDBManager({"tool_sparql": json.dumps(["not-a-dict"])}), + "tool_sparql", + ) + + assert paths == [] + + +def test_extract_paths_from_db_supports_top_level_paths_list(): + interpreter = build_interpreter() + + paths = interpreter.extract_paths_from_db( + StubDBManager({"tool_sparql": json.dumps({"paths": ["a.csv"]})}), + "tool_sparql", + ) + + assert paths == ["a.csv"] + + +def test_find_changed_or_new_files_includes_new_file(tmp_path): + interpreter = build_interpreter() + session_dir = tmp_path / "session" + session_dir.mkdir() + before = interpreter.capture_session_file_metadata(session_dir) + + created_file = session_dir / "created.csv" + created_file.write_text("a,b\n1,2\n", encoding="utf-8") + after = interpreter.capture_session_file_metadata(session_dir) + + changed_files = interpreter.find_changed_or_new_files(before, after) + + assert changed_files == {created_file.resolve()} + + +def test_find_changed_or_new_files_includes_overwritten_file(tmp_path): + interpreter = build_interpreter() + session_dir = tmp_path / "session" + session_dir.mkdir() + existing_file = session_dir / "updated.csv" + existing_file.write_text("a,b\n1,2\n", encoding="utf-8") + before = interpreter.capture_session_file_metadata(session_dir) + + time.sleep(0.001) + existing_file.write_text("a,b\n30,400\n", encoding="utf-8") + after = interpreter.capture_session_file_metadata(session_dir) + + changed_files = interpreter.find_changed_or_new_files(before, after) + + assert changed_files == {existing_file.resolve()} + + +def test_find_changed_or_new_files_ignores_unchanged_files(tmp_path): + interpreter = build_interpreter() + session_dir = tmp_path / "session" + session_dir.mkdir() + existing_file = session_dir / "unchanged.csv" + existing_file.write_text("a,b\n1,2\n", encoding="utf-8") + before = interpreter.capture_session_file_metadata(session_dir) + after = interpreter.capture_session_file_metadata(session_dir) + + changed_files = interpreter.find_changed_or_new_files(before, after) + + assert changed_files == set() + + +def test_run_returns_user_friendly_error_when_llm_invoke_fails(tmp_path, monkeypatch): + session_dir = tmp_path / "session" + session_dir.mkdir() + input_file = session_dir / "input.csv" + input_file.write_text("a,b\n1,2\n", encoding="utf-8") + interpreter = Interpreter(llm_instance=StubLLM(error=RuntimeError("boom"))) + + monkeypatch.setenv("METABOT_TRUSTED_MODE", "true") + monkeypatch.setattr( + "app.core.agents.interpreter.tool_interpreter.create_user_session", + lambda session_id, user_session_dir=False, input_dir=False: session_dir, + ) + monkeypatch.setattr( + "app.core.agents.interpreter.tool_interpreter.tools_database", + lambda: StubDBManager(), + ) + + result = interpreter._run(f'Filepath: "{input_file}"') + + assert result == "Interpreter could not generate analysis code for this request." + + +def test_run_normalizes_non_string_llm_content(tmp_path, monkeypatch): + session_dir = tmp_path / "session" + session_dir.mkdir() + input_file = session_dir / "input.csv" + input_file.write_text("a,b\n1,2\n", encoding="utf-8") + interpreter = Interpreter(llm_instance=StubLLM(content=["```python\nprint('ok')\n```"])) + + monkeypatch.setenv("METABOT_TRUSTED_MODE", "true") + monkeypatch.setattr( + "app.core.agents.interpreter.tool_interpreter.create_user_session", + lambda session_id, user_session_dir=False, input_dir=False: session_dir, + ) + monkeypatch.setattr( + "app.core.agents.interpreter.tool_interpreter.tools_database", + lambda: StubDBManager(), + ) + monkeypatch.setattr( + Interpreter, + "execute_in_subprocess", + lambda self, code, session_dir, allowed_input_paths: {"stdout": "ok", "error": ""}, + ) + + result = interpreter._run(f'Filepath: "{input_file}"') + + assert result == "ok" diff --git a/app/core/tests/test_main.py b/app/core/tests/test_main.py new file mode 100644 index 0000000..4af8b90 --- /dev/null +++ b/app/core/tests/test_main.py @@ -0,0 +1,155 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +import app.core.main as main_module + + +class DummyLogger: + def __init__(self): + self.info_messages = [] + self.error_messages = [] + + def info(self, message, *args): + if args: + message = message % args + self.info_messages.append(message) + + def error(self, message, *args): + if args: + message = message % args + self.error_messages.append(message) + + +def test_prepare_session_files_copies_valid_file(tmp_path, monkeypatch): + input_dir = tmp_path / "input_files" + input_dir.mkdir() + source_file = tmp_path / "source.csv" + source_file.write_text("a,b\n1,2\n", encoding="utf-8") + dummy_logger = DummyLogger() + input_dir_path = input_dir + + monkeypatch.setattr(main_module, "create_user_session", lambda session_id, input_dir=False: input_dir_path) + monkeypatch.setattr(main_module, "logger", dummy_logger) + + staged_dir = main_module._prepare_session_files("session-id", [str(source_file)]) + + copied_file = input_dir / source_file.name + assert staged_dir == input_dir + assert copied_file.read_text(encoding="utf-8") == source_file.read_text(encoding="utf-8") + + +def test_prepare_session_files_rejects_directory(tmp_path, monkeypatch): + input_dir = tmp_path / "input_files" + input_dir.mkdir() + source_dir = tmp_path / "source_dir" + source_dir.mkdir() + input_dir_path = input_dir + + monkeypatch.setattr(main_module, "create_user_session", lambda session_id, input_dir=False: input_dir_path) + + with pytest.raises(main_module.SessionFilePreparationError, match="not a file"): + main_module._prepare_session_files("session-id", [str(source_dir)]) + + +def test_prepare_session_files_rejects_colliding_basenames(tmp_path, monkeypatch): + input_dir = tmp_path / "input_files" + input_dir.mkdir() + first_dir = tmp_path / "first" + second_dir = tmp_path / "second" + first_dir.mkdir() + second_dir.mkdir() + first_file = first_dir / "results.csv" + second_file = second_dir / "results.csv" + first_file.write_text("a,b\n1,2\n", encoding="utf-8") + second_file.write_text("a,b\n3,4\n", encoding="utf-8") + dummy_logger = DummyLogger() + input_dir_path = input_dir + + monkeypatch.setattr(main_module, "create_user_session", lambda session_id, input_dir=False: input_dir_path) + monkeypatch.setattr(main_module, "logger", dummy_logger) + + with pytest.raises(main_module.SessionFilePreparationError, match="would overwrite"): + main_module._prepare_session_files("session-id", [str(first_file), str(second_file)]) + + +def test_prepare_session_files_rejects_same_file_destination(tmp_path, monkeypatch): + input_dir = tmp_path / "input_files" + input_dir.mkdir() + staged_file = input_dir / "already_staged.csv" + staged_file.write_text("a,b\n1,2\n", encoding="utf-8") + input_dir_path = input_dir + + monkeypatch.setattr(main_module, "create_user_session", lambda session_id, input_dir=False: input_dir_path) + + with pytest.raises(main_module.SessionFilePreparationError, match="already staged"): + main_module._prepare_session_files("session-id", [str(staged_file)]) + + +def test_main_passes_cli_api_key_and_reconfigures_logger(monkeypatch): + state = {"session_initialized": False, "setup_logger_states": []} + captured = {} + + def fake_setup_logger(name): + state["setup_logger_states"].append(state["session_initialized"]) + return DummyLogger() + + def fake_initialize_session_context(session_id): + state["session_initialized"] = True + + def fake_llm_creation(api_key=None, params_file=None): + captured["llm_api_key"] = api_key + return {"llm_o": object()} + + def fake_create_workflow(models, session_id=None, endpoint_url=None, evaluation=False, api_key=None): + captured["workflow_api_key"] = api_key + captured["session_id"] = session_id + return "workflow" + + def fake_process_workflow(workflow, question): + captured["workflow"] = workflow + captured["question"] = question + + monkeypatch.setattr("sys.argv", ["prog", "-c", "hello", "--api-key", "cli-key"]) + monkeypatch.setattr(main_module, "logger", DummyLogger()) + monkeypatch.setattr(main_module, "setup_logger", fake_setup_logger) + monkeypatch.setattr(main_module, "initialize_session_context", fake_initialize_session_context) + monkeypatch.setattr(main_module, "create_user_session", lambda session_id=None, user_session_dir=False, input_dir=False: "session-123") + monkeypatch.setattr(main_module, "langsmith_setup", lambda: None) + monkeypatch.setattr(main_module, "llm_creation", fake_llm_creation) + monkeypatch.setattr(main_module, "create_workflow", fake_create_workflow) + monkeypatch.setattr(main_module, "process_workflow", fake_process_workflow) + + main_module.main() + + assert captured["llm_api_key"] == "cli-key" + assert captured["workflow_api_key"] == "cli-key" + assert state["setup_logger_states"] == [True] + assert captured["session_id"] == "session-123" + assert captured["workflow"] == "workflow" + assert captured["question"] == "hello" + + +def test_main_prints_user_friendly_error_for_bad_staged_file(monkeypatch, capsys): + monkeypatch.setattr("sys.argv", ["prog", "-c", "hello", "-f", "/missing/file.csv"]) + monkeypatch.setattr(main_module, "logger", DummyLogger()) + monkeypatch.setattr(main_module, "setup_logger", lambda name: DummyLogger()) + monkeypatch.setattr(main_module, "initialize_session_context", lambda session_id: None) + monkeypatch.setattr(main_module, "create_user_session", lambda session_id=None, user_session_dir=False, input_dir=False: "session-123") + monkeypatch.setattr(main_module, "langsmith_setup", lambda: None) + monkeypatch.setattr(main_module, "llm_creation", lambda api_key=None, params_file=None: {"llm_o": object()}) + monkeypatch.setattr( + main_module, + "_prepare_session_files", + lambda session_id, file_paths: (_ for _ in ()).throw( + main_module.SessionFilePreparationError(Path(file_paths[0]), f"File not found: {file_paths[0]}") + ), + ) + monkeypatch.setattr(main_module, "create_workflow", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("workflow should not run"))) + + main_module.main() + + captured = capsys.readouterr() + assert "Error: File not found: /missing/file.csv" in captured.out diff --git a/app/core/tests/test_sandbox_runner.py b/app/core/tests/test_sandbox_runner.py new file mode 100644 index 0000000..8a15d88 --- /dev/null +++ b/app/core/tests/test_sandbox_runner.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from app.core.agents.interpreter import sandbox_runner +from app.core.agents.interpreter.sandbox_runner import validate_code + + +def test_validate_code_rejects_dunder_attribute_access(): + with pytest.raises(ValueError) as excinfo: + validate_code("x = object().__class__") + assert "Dunder attribute access" in str(excinfo.value) + + +def test_execute_user_code_returns_captured_output_on_exception_from_subprocess(tmp_path): + session_dir = tmp_path / "session" + session_dir.mkdir() + input_file = session_dir / "input.csv" + input_file.write_text("a,b\n1,2\n", encoding="utf-8") + + payload_path = tmp_path / "payload.json" + result_path = tmp_path / "result.json" + payload_path.write_text( + json.dumps( + { + "code": "print('hello before error')\nraise ValueError('boom')", + "session_dir": str(session_dir), + "allowed_input_paths": [str(input_file)], + "timeout_seconds": 5, + "cpu_seconds": 5, + "memory_bytes": 128 * 1024 * 1024, + "file_size_bytes": 10 * 1024 * 1024, + } + ), + encoding="utf-8", + ) + + try: + completed = subprocess.run( + [ + sys.executable, + "-I", + str(Path(sandbox_runner.__file__).resolve()), + "--payload", + str(payload_path), + "--result", + str(result_path), + ], + cwd=session_dir, + env={ + "HOME": str(session_dir), + "PYTHONIOENCODING": "utf-8", + "TMPDIR": str(session_dir), + }, + capture_output=True, + text=True, + timeout=8, + ) + except subprocess.TimeoutExpired as exc: + pytest.fail(f"Sandbox runner subprocess hung unexpectedly: {exc}") + + assert completed.returncode == 1 + assert result_path.exists(), completed.stderr or completed.stdout + + result = json.loads(result_path.read_text(encoding="utf-8")) + assert result["success"] is False + assert "hello before error" in result["stdout"] + assert result["error"] == "boom" + assert "ValueError: boom" in result["traceback"] diff --git a/app/core/tests/test_security.py b/app/core/tests/test_security.py new file mode 100644 index 0000000..4691b5c --- /dev/null +++ b/app/core/tests/test_security.py @@ -0,0 +1,43 @@ +import pytest + +from app.core.security import resolve_session_path +from app.core.utils import load_config + + +def test_load_config_excludes_interpreter_by_default(monkeypatch): + monkeypatch.delenv("METABOT_TRUSTED_MODE", raising=False) + + config = load_config() + + assert "Interpreter_agent" not in config["supervisor"]["members"] + assert all(agent["name"] != "Interpreter_agent" for agent in config["agents"]) + + +def test_load_config_includes_interpreter_in_trusted_mode(monkeypatch): + monkeypatch.setenv("METABOT_TRUSTED_MODE", "true") + + config = load_config() + + assert "Interpreter_agent" in config["supervisor"]["members"] + assert any(agent["name"] == "Interpreter_agent" for agent in config["agents"]) + + +def test_resolve_session_path_allows_current_session_files(tmp_path): + session_dir = tmp_path / "session" + session_dir.mkdir() + allowed_file = session_dir / "input.csv" + allowed_file.write_text("a,b\n1,2\n", encoding="utf-8") + + resolved = resolve_session_path(allowed_file, session_dir=session_dir) + + assert resolved == allowed_file.resolve() + + +def test_resolve_session_path_rejects_outside_files(tmp_path): + session_dir = tmp_path / "session" + session_dir.mkdir() + outside_file = tmp_path / "outside.csv" + outside_file.write_text("a,b\n1,2\n", encoding="utf-8") + + with pytest.raises(ValueError): + resolve_session_path(outside_file, session_dir=session_dir) diff --git a/app/core/tests/test_sparql_tool.py b/app/core/tests/test_sparql_tool.py new file mode 100644 index 0000000..6b95ae0 --- /dev/null +++ b/app/core/tests/test_sparql_tool.py @@ -0,0 +1,126 @@ +from app.core.agents.sparql import agent as sparql_agent +from app.core.agents.sparql import tool_sparql + + +def _construct_tool(**kwargs): + return tool_sparql.GraphSparqlQAChain.model_construct(**kwargs) + + +def test_create_agent_passes_openai_key_to_import_tools(monkeypatch): + captured = {} + + def fake_import_tools(directory, module_prefix, **kwargs): + captured["tool_kwargs"] = kwargs + return ["tool"] + + def fake_create_openai_tools_agent(model, tools, prompt): + captured["agent_model"] = model + captured["agent_tools"] = tools + return "agent" + + def fake_agent_executor(agent, tools): + captured["executor_agent"] = agent + captured["executor_tools"] = tools + return "executor" + + monkeypatch.setattr(sparql_agent, "import_tools", fake_import_tools) + monkeypatch.setattr( + sparql_agent, + "create_openai_tools_agent", + fake_create_openai_tools_agent, + ) + monkeypatch.setattr(sparql_agent, "AgentExecutor", fake_agent_executor) + + executor = sparql_agent.create_agent( + llms={sparql_agent.MODEL_CHOICE: "model"}, + graph="graph", + session_id="session-123", + openai_key="cli-key", + ) + + assert executor == "executor" + assert captured["tool_kwargs"] == { + "graph": "graph", + "llm": {sparql_agent.MODEL_CHOICE: "model"}, + "session_id": "session-123", + "openai_key": "cli-key", + } + assert captured["agent_model"] == "model" + assert captured["executor_agent"] == "agent" + assert captured["executor_tools"] == ["tool"] + + +def test_search_nodes_uses_explicit_openai_key(monkeypatch): + captured = {} + + class FakeEmbeddings: + def __init__(self, api_key=None): + captured["api_key"] = api_key + + class FakeVectorStore: + def similarity_search(self, query, limit): + captured["query"] = query + captured["limit"] = limit + return ["related-node"] + + def fake_load_local(path, embeddings, allow_dangerous_deserialization=False): + captured["db_path"] = path + captured["embeddings"] = embeddings + captured["allow_dangerous_deserialization"] = allow_dangerous_deserialization + return FakeVectorStore() + + monkeypatch.setattr(tool_sparql, "OpenAIEmbeddings", FakeEmbeddings) + monkeypatch.setattr(tool_sparql.FAISS, "load_local", fake_load_local) + + tool = _construct_tool(openai_key="cli-key") + result = tool.search_nodes("SELECT * WHERE { ?s ?p ?o }") + + assert captured["api_key"] == "cli-key" + assert captured["query"] == "SELECT * WHERE { ?s ?p ?o }" + assert captured["limit"] == 12 + assert captured["allow_dangerous_deserialization"] is True + assert result == ["related-node"] + + +def test_search_nodes_falls_back_to_env_openai_key(monkeypatch): + captured = {} + + class FakeLLMChain: + def __init__(self, llm, prompt): + self.llm = llm + self.prompt = prompt + + class FakeEmbeddings: + def __init__(self, api_key=None): + captured["api_key"] = api_key + + class FakeVectorStore: + def similarity_search(self, query, limit): + captured["query"] = query + captured["limit"] = limit + return ["related-node"] + + def fake_load_local(path, embeddings, allow_dangerous_deserialization=False): + captured["db_path"] = path + captured["embeddings"] = embeddings + captured["allow_dangerous_deserialization"] = allow_dangerous_deserialization + return FakeVectorStore() + + monkeypatch.setenv("OPENAI_API_KEY", "env-key") + monkeypatch.setattr(tool_sparql, "LLMChain", FakeLLMChain) + monkeypatch.setattr(tool_sparql, "OpenAIEmbeddings", FakeEmbeddings) + monkeypatch.setattr(tool_sparql.FAISS, "load_local", fake_load_local) + + tool = tool_sparql.GraphSparqlQAChain( + llm={"llm_o": object(), "llm_o3_mini": object()}, + graph=object(), + session_id="session-123", + openai_key=None, + ) + result = tool.search_nodes("SELECT * WHERE { ?s ?p ?o }") + + assert captured["api_key"] == "env-key" + assert captured["query"] == "SELECT * WHERE { ?s ?p ?o }" + assert captured["limit"] == 12 + assert captured["allow_dangerous_deserialization"] is True + assert result == ["related-node"] diff --git a/app/core/utils.py b/app/core/utils.py index 59c68b2..4b81a5f 100644 --- a/app/core/utils.py +++ b/app/core/utils.py @@ -3,6 +3,7 @@ import logging import logging.config import os +from copy import deepcopy from pathlib import Path from typing import List, Optional, Type from langchain.tools import BaseTool @@ -10,6 +11,7 @@ import inspect import tempfile from uuid import uuid4 +from app.core.security import is_trusted_mode_enabled from app.core.session import setup_logger logger = setup_logger(__name__) @@ -18,7 +20,42 @@ def load_config(): config_path = Path(__file__).resolve().parent.parent / "config" / "langgraph.json" with open(config_path, "r") as file: - return json.load(file) + config = json.load(file) + + return _with_optional_interpreter(config) + + +def _with_optional_interpreter(config: dict) -> dict: + if not is_trusted_mode_enabled(): + return config + + trusted_config = deepcopy(config) + interpreter_agent = { + "name": "Interpreter_agent", + "path": "app.core.agents.interpreter.agent", + "llm_choice": "llm_o", + } + interpreter_edge = {"source": "Interpreter_agent", "target": "supervisor"} + interpreter_target = { + "condition_value": "Interpreter_agent", + "target": "Interpreter_agent", + } + + if interpreter_agent not in trusted_config["agents"]: + trusted_config["agents"].insert(3, interpreter_agent) + + if interpreter_edge not in trusted_config["edges"]: + trusted_config["edges"].append(interpreter_edge) + + for edge in trusted_config["conditional_edges"]: + if edge["source"] == "supervisor" and interpreter_target not in edge["targets"]: + edge["targets"].insert(-1, interpreter_target) + + members = trusted_config["supervisor"]["members"] + if "Interpreter_agent" not in members: + members.append("Interpreter_agent") + + return trusted_config def import_tools(directory: str, module_prefix: str, **kwargs) -> List: diff --git a/app/core/workflow/langraph_workflow.py b/app/core/workflow/langraph_workflow.py index 7d5ee16..bb93662 100644 --- a/app/core/workflow/langraph_workflow.py +++ b/app/core/workflow/langraph_workflow.py @@ -250,5 +250,5 @@ def process_workflow(app: StateGraph, question: str, thread_id: int = 1) -> NoRe logger.info(s) logger.info("----") except Exception as e: - logger.error(f"An error occurred: {e}") + logger.error(f"An error occurred: {e}", exc_info=True) diff --git a/app/tests/installation_test.py b/app/tests/installation_test.py index abb8d1d..3e2b439 100644 --- a/app/tests/installation_test.py +++ b/app/tests/installation_test.py @@ -1,15 +1,14 @@ import sys from pathlib import Path +from app.core.main import main REPO_ROOT = Path(__file__).resolve().parents[2] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) -from app.core.main import main - if "-q" not in sys.argv and "--question" not in sys.argv: sys.argv.extend(["-q", "1"]) if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/docs/user-guide/configuration.md b/docs/user-guide/configuration.md index 85afe7b..f225e74 100644 --- a/docs/user-guide/configuration.md +++ b/docs/user-guide/configuration.md @@ -142,6 +142,8 @@ LANGCHAIN_ENDPOINT=https://api.smith.langchain.com If no tracing key is present, MetaboT disables tracing automatically. +For running the repository's LangSmith-based automated benchmark evaluation, see [docs/examples/langsmith-evaluation.md](../examples/langsmith-evaluation.md). + ## SPARQL and Schema Configuration `app/config/sparql.ini` contains helper queries used to inspect or work with the graph schema. These are especially important when MetaboT needs to construct schema-aware prompts from an endpoint rather than from assumptions. diff --git a/environment.yml b/environment.yml index c1c9393..387c89e 100644 --- a/environment.yml +++ b/environment.yml @@ -12,37 +12,36 @@ dependencies: - xz==5.4.6 - pip: - app==0.0.1 - - "codeboxapi>=0.0.19,<0.2.0" - chromadb==0.4.24 - - codeinterpreterapi==0.0.14 - faiss-cpu==1.8.0 - ipykernel==6.29.3 - jupyter-kernel-gateway==3.0.1 - kaleido==0.2.1 - - langchain==0.1.11 + - langchain==0.3.27 + - langchain_experimental==0.3.4 - langchainhub==0.1.15 - - langchain_community==0.0.27 - - langchain_core==0.1.30 - - langchain_openai==0.0.8 - - langgraph==0.0.26 - - litellm==1.61.6 - - langsmith==0.1.22 + - langchain_community==0.3.31 + - langchain_core==0.3.84 + - langchain_openai==0.3.35 + - langgraph==0.3.34 + - litellm==1.83.0 + - langsmith==0.3.45 - matplotlib==3.8.3 - networkx==3.2.1 - - openai==1.61.0 + - openai==2.32.0 - pandas==2.2.1 - plotly==5.20.0 - psycopg2-binary - - pydantic==2.7.0 - - python-dotenv==1.0.1 + - pydantic==2.13.3 + - python-dotenv==1.2.2 - rdflib==7.0.0 - - Requests==2.31.0 + - requests==2.33.1 - seaborn==0.13.2 - SPARQLWrapper==2.0.0 - SQLAlchemy==2.0.22 - tiktoken>=0.7.0 - - tqdm==4.66.1 - - numpy==1.26.0 + - tqdm==4.66.6 + - numpy==1.26.4 - httpx==0.27.2 - scikit-learn==1.5.1 - spectrum_utils diff --git a/requirements.txt b/requirements.txt index 9590676..1299a35 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,32 +1,31 @@ -codeboxapi>=0.0.19,<0.2.0 -codeinterpreterapi==0.0.14 faiss-cpu==1.8.0 kaleido==0.2.1 ipykernel==6.29.3 jupyter-kernel-gateway==3.0.1 -langchain==0.1.11 +langchain==0.3.27 +langchain_experimental==0.3.4 langchainhub==0.1.15 -langchain_community==0.0.27 -langchain_core==0.1.30 -langchain_openai==0.0.8 -langgraph==0.0.26 -litellm==1.61.6 -langsmith==0.1.22 +langchain_community==0.3.31 +langchain_core==0.3.84 +langchain_openai==0.3.35 +langgraph==0.3.34 +litellm==1.83.0 +langsmith==0.3.45 matplotlib==3.8.3 networkx==3.2.1 -openai==1.61.0 +openai==2.32.0 pandas==2.2.1 plotly==5.20.0 psycopg2-binary -pydantic==2.7.0 -python-dotenv==1.0.1 +pydantic==2.13.3 +python-dotenv==1.2.2 rdflib==7.0.0 -Requests==2.31.0 +requests==2.33.1 seaborn==0.13.2 -streamlit==1.44.0 +streamlit==1.54.0 SPARQLWrapper==2.0.0 tiktoken>=0.7.0 -tqdm==4.66.1 -numpy==1.26.0 +tqdm==4.66.6 +numpy==1.26.4 httpx==0.27.2 scikit-learn==1.5.1 diff --git a/streamlit_webapp/README.md b/streamlit_webapp/README.md index dcd025c..2b53a58 100644 --- a/streamlit_webapp/README.md +++ b/streamlit_webapp/README.md @@ -16,6 +16,12 @@ MetaboT is an AI-powered system designed to facilitate the exploration of metabo - Session management and data persistence - Automated file cleanup system +## v1.1.0 Notes + +- Streamlit startup is now resilient to the working directory used to launch the app. +- Public-facing deployments keep the interpreter workflow disabled unless trusted mode is explicitly enabled. +- Safer file handling and API-key propagation fixes from the CLI now align better with Streamlit-backed workflows. + ## Project Structure ``` @@ -62,11 +68,13 @@ pip install -r requirements.txt 2. Set up the PostgreSQL database and configure the environment variables. -3. Run the Streamlit application: +3. Run the Streamlit application from the repository root: ```bash -streamlit run streamlit_app.py +python -m streamlit run streamlit_webapp/streamlit_app.py ``` +If you are currently inside `streamlit_webapp/`, run `cd ..` first and then use the command above. + ## Usage 1. Launch the application and configure your API keys in the sidebar: @@ -110,5 +118,4 @@ For contributors with access keys, the application provides: MetaboT is released under the Apache 2.0 License. See [LICENSE.txt](LICENSE.txt). - - +[Include support contact information here] diff --git a/streamlit_webapp/streamlit_app.py b/streamlit_webapp/streamlit_app.py index 5de1760..bffcd1a 100644 --- a/streamlit_webapp/streamlit_app.py +++ b/streamlit_webapp/streamlit_app.py @@ -1,10 +1,20 @@ -# Setting up Session for streamlit app -# It's ugly but it has to be done in the very beginning of the script -import streamlit as st -import os -from uuid import uuid4 - -from app.core.session import create_user_session, initialize_session_context, initialize_thread_id, setup_logger +# Setting up Session for streamlit app +# It's ugly but it has to be done in the very beginning of the script +import streamlit as st +import os +import sys +from pathlib import Path +from uuid import uuid4 + +STREAMLIT_APP_DIR = Path(__file__).resolve().parent +REPO_ROOT = STREAMLIT_APP_DIR.parent + +# Support running via `streamlit run streamlit_webapp/streamlit_app.py` from the +# repo root as well as from inside `streamlit_webapp/`. +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from app.core.session import create_user_session, initialize_session_context, initialize_thread_id, setup_logger if "session_id" not in st.session_state: st.session_state.session_id = create_user_session() @@ -25,19 +35,19 @@ logger = setup_logger(__name__) st.session_state.logger = logger -# Following normal code execution -import os -from pathlib import Path -import logging -import time -from openai import OpenAI -from langchain_core.messages import HumanMessage -from langsmith import Client -from app.core.memory.database_manager import tools_database, memory_database -from langchain.callbacks.manager import tracing_v2_enabled -from streamlit_utils import check_characters_api_key, test_sparql_endpoint, test_openai_key, new_process_langgraph_output, create_zip_buffer, is_true -from app.core.workflow.langraph_workflow import create_workflow -from app.core.main import llm_creation +# Following normal code execution +import logging +import time +from openai import OpenAI +from langchain_core.messages import HumanMessage +from langsmith import Client +from app.core.memory.database_manager import tools_database, memory_database +from app.core.security import is_trusted_mode_enabled +from app.core.security import resolve_session_path +from langchain.callbacks.manager import tracing_v2_enabled +from streamlit_webapp.streamlit_utils import check_characters_api_key, test_sparql_endpoint, test_openai_key, new_process_langgraph_output, create_zip_buffer, is_true +from app.core.workflow.langraph_workflow import create_workflow +from app.core.main import llm_creation def add_videos_to_content(): @@ -115,8 +125,11 @@ def add_videos_to_content(): if "preselected_question" not in st.session_state: st.session_state.preselected_question = None -if "set_standard_endpoint" not in st.session_state: - st.session_state.set_standard_endpoint = True +if "set_standard_endpoint" not in st.session_state: + st.session_state.set_standard_endpoint = True + +if "trusted_mode_enabled" not in st.session_state: + st.session_state.trusted_mode_enabled = is_trusted_mode_enabled() if "langgraph_app_created" not in st.session_state: st.session_state.langgraph_app_created = False @@ -170,7 +183,7 @@ def terms_and_conditions_dialog(): """) - if st.button("✅ I Agree to the Terms", use_container_width=True): + if st.button("✅ I Agree to the Terms", width="stretch"): st.session_state.terms_accepted = True st.success("Thanks! You have accepted the Terms.") st.rerun() @@ -178,7 +191,7 @@ def terms_and_conditions_dialog(): # --- Button to open the Terms dialog --- st.markdown("## 📘 Please read the Terms and Conditions.") - open_terms_dialog = st.button("📘 Terms and Conditions", use_container_width=True) + open_terms_dialog = st.button("📘 Terms and Conditions", width="stretch") if open_terms_dialog: terms_and_conditions_dialog() # OpenAI API Key Input and Validation @@ -293,7 +306,7 @@ def question_instructions_dialog(): if st.button("Close"): st.rerun() - open_question_instructions_dialog = st.button("Question Instructions", use_container_width=True, disabled=st.session_state.is_processing) + open_question_instructions_dialog = st.button("Question Instructions", width="stretch", disabled=st.session_state.is_processing) if open_question_instructions_dialog: question_instructions_dialog() @@ -316,20 +329,20 @@ def clear_dialog(): st.rerun() - open_help_dialog = st.button("Help", use_container_width=True, disabled=st.session_state.is_processing) + open_help_dialog = st.button("Help", width="stretch", disabled=st.session_state.is_processing) if open_help_dialog: help_dialog() - download = st.download_button( - label="Download files", - data=create_zip_buffer(st.session_state.session_id), - file_name="metabot_files.zip", - mime="application/zip", - use_container_width=True, - disabled=st.session_state.is_processing - ) - - open_clear_dialog = st.button("Clear conversation data", use_container_width=True, disabled=st.session_state.is_processing) + download = st.download_button( + label="Download files", + data=create_zip_buffer(st.session_state.session_id), + file_name="metabot_files.zip", + mime="application/zip", + width="stretch", + disabled=st.session_state.is_processing + ) + + open_clear_dialog = st.button("Clear conversation data", width="stretch", disabled=st.session_state.is_processing) if open_clear_dialog: clear_dialog() @@ -346,7 +359,7 @@ def clear_dialog(): st.markdown("---") st.subheader("Holobiomics Lab - CNRS, Université Côte d'Azur, Interdisciplinary Institute for Artificial Intelligence (3iA) Côte d'Azur") holobiomics_logo_path = str(Path(__file__).parent / "misc" / "HolobiomicsLab_graphics_v1_logos_small.png") - st.image(holobiomics_logo_path, use_container_width=True) + st.image(holobiomics_logo_path, width="stretch") if st.session_state.openai_key_success == False and st.session_state.endpoint_url_success == False: st.warning("You haven't provided a valid OpenAI API key and validated the connection to the endpoint. You need to provide a valid OpenAI API Key and connect to an Endpoint server. If you already tried to connect to an endpoint and it was unsucessful, there might be a connection problem. Please investigate further or come back later") @@ -366,11 +379,11 @@ def clear_dialog(): st.write(splash_text) add_videos_to_content() -if st.session_state.openai_key_success == True and st.session_state.endpoint_url_success == True: - if st.session_state.langgraph_app_created == False: - st.warning("Initializing the LangGraph... Please wait") - st.session_state.logger.info("Initializing the LangGraph") - try: +if st.session_state.openai_key_success == True and st.session_state.endpoint_url_success == True: + if st.session_state.langgraph_app_created == False: + st.warning("Initializing the LangGraph... Please wait") + st.session_state.logger.info("Initializing the LangGraph") + try: st.session_state.models = llm_creation(api_key=st.session_state.OPENAI_API_KEY) @@ -379,29 +392,41 @@ def clear_dialog(): st.session_state.logger.info("LangGraph initialized") st.rerun() except Exception as e: - st.error(f"An error occurred while initializing the LangGraph: {e}") - st.session_state.logger.error(f"An error occurred while initializing the LangGraph: {e}") - - else: - # Variables to track submission status of predefined and custom question - select_submit_clicked = False - custom_submit_clicked = False - - with st.expander("Input your own file", expanded=False): - user_files = st.file_uploader("Choose your files", accept_multiple_files=True) - if user_files: - filenames = [file.name for file in user_files] - try: - for file in user_files: - file_path = st.session_state.user_input_dir / file.name - with file_path.open("wb") as f: - f.write(file.getbuffer()) - st.success(f"File(s) uploaded successfully: {filenames}") - st.session_state.logger.info(f"File(s) uploaded successfully: {filenames}") - except Exception as e: - st.error(f"An error occurred while processing the file: {e}") - - with st.expander ("Try one our standard questions:", expanded=False): + st.error(f"An error occurred while initializing the LangGraph: {e}") + st.session_state.logger.error(f"An error occurred while initializing the LangGraph: {e}") + + else: + if st.session_state.trusted_mode_enabled == False: + st.info( + "Trusted mode is disabled in this deployment. " + "Interpreter-based file analysis and LLM-generated visualizations are not exposed to public users." + ) + + # Variables to track submission status of predefined and custom question + select_submit_clicked = False + custom_submit_clicked = False + + if st.session_state.trusted_mode_enabled == True: + with st.expander("Input your own file", expanded=False): + user_files = st.file_uploader("Choose your files", accept_multiple_files=True) + if user_files: + filenames = [Path(file.name).name for file in user_files] + try: + for file in user_files: + safe_filename = Path(file.name).name + file_path = resolve_session_path( + safe_filename, + session_dir=st.session_state.user_input_dir, + must_exist=False, + ) + with file_path.open("wb") as f: + f.write(file.getbuffer()) + st.success(f"File(s) uploaded successfully: {filenames}") + st.session_state.logger.info(f"File(s) uploaded successfully: {filenames}") + except Exception as e: + st.error(f"An error occurred while processing the file: {e}") + + with st.expander ("Try one our standard questions:", expanded=False): # Form for selecting a predefined question with st.form(key="select_form"): # Dynamically generated list of questions for user selection diff --git a/streamlit_webapp/test_streamlit_app.py b/streamlit_webapp/test_streamlit_app.py new file mode 100644 index 0000000..d181309 --- /dev/null +++ b/streamlit_webapp/test_streamlit_app.py @@ -0,0 +1,21 @@ +from streamlit.testing.v1 import AppTest + +import app.core.memory.database_manager as database_manager +from streamlit_webapp import streamlit_utils + + +class DummyToolsDatabase: + pass + + +def test_streamlit_app_renders_without_import_errors(monkeypatch): + monkeypatch.setattr(streamlit_utils, "test_sparql_endpoint", lambda endpoint: False) + monkeypatch.setattr(database_manager, "tools_database", lambda: DummyToolsDatabase()) + monkeypatch.setattr(database_manager, "memory_database", lambda: object()) + + app = AppTest.from_file("streamlit_webapp/streamlit_app.py") + app.run(timeout=30) + + assert not app.exception + assert app.title + assert app.title[0].value == "MetaboT - An AI-system for Metabolomics Data Exploration"