from __future__ import annotations import json from collections.abc import Sequence from typing import Any, cast from packaging.version import Version from pydantic import ValidationError from sqlalchemy import select from sqlalchemy.orm import Session from core.agent.entities import AgentToolEntity from core.agent.plugin_entities import AgentStrategyParameter from core.memory.token_buffer_memory import TokenBufferMemory from core.model_manager import ModelInstance from core.plugin.entities.request import InvokeCredentials from core.plugin.impl.model_runtime_factory import create_plugin_model_assembly from core.tools.entities.tool_entities import ToolIdentity, ToolParameter, ToolProviderType from core.tools.tool_manager import ToolManager from core.workflow.system_variables import SystemVariableKey, get_system_text from extensions.ext_database import db from graphon.model_runtime.entities.model_entities import AIModelEntity, ModelType from graphon.runtime import VariablePool from models.model import Conversation from .entities import AgentNodeData, AgentOldVersionModelFeatures, ParamsAutoGenerated from .exceptions import AgentInputTypeError, AgentVariableNotFoundError from .strategy_protocols import ResolvedAgentStrategy class AgentRuntimeSupport: def build_parameters( self, *, agent_parameters: Sequence[AgentStrategyParameter], variable_pool: VariablePool, node_data: AgentNodeData, strategy: ResolvedAgentStrategy, tenant_id: str, user_id: str, app_id: str, invoke_from: Any, for_log: bool = False, ) -> dict[str, Any]: agent_parameters_dictionary = {parameter.name: parameter for parameter in agent_parameters} result: dict[str, Any] = {} for parameter_name in node_data.agent_parameters: parameter = agent_parameters_dictionary.get(parameter_name) if not parameter: result[parameter_name] = None continue agent_input = node_data.agent_parameters[parameter_name] match agent_input.type: case "variable": variable = variable_pool.get(agent_input.value) # type: ignore[arg-type] if variable is None: raise AgentVariableNotFoundError(str(agent_input.value)) parameter_value = variable.value case "mixed" | "constant": try: if not isinstance(agent_input.value, str): parameter_value = json.dumps(agent_input.value, ensure_ascii=False) else: parameter_value = str(agent_input.value) except TypeError: parameter_value = str(agent_input.value) segment_group = variable_pool.convert_template(parameter_value) parameter_value = segment_group.log if for_log else segment_group.text try: if not isinstance(agent_input.value, str): parameter_value = json.loads(parameter_value) except json.JSONDecodeError: parameter_value = parameter_value case _: raise AgentInputTypeError(agent_input.type) value = parameter_value if parameter.type == "array[tools]": value = cast(list[dict[str, Any]], value) value = [tool for tool in value if tool.get("enabled", False)] value = self._filter_mcp_type_tool(strategy, value) for tool in value: if "schemas" in tool: tool.pop("schemas") parameters = tool.get("parameters", {}) if all(isinstance(v, dict) for _, v in parameters.items()): params = {} for key, param in parameters.items(): if param.get("auto", ParamsAutoGenerated.OPEN) in ( ParamsAutoGenerated.CLOSE, 0, ): value_param = param.get("value", {}) if value_param and value_param.get("type", "") == "variable": variable_selector = value_param.get("value") if not variable_selector: raise ValueError("Variable selector is missing for a variable-type parameter.") variable = variable_pool.get(variable_selector) if variable is None: raise AgentVariableNotFoundError(str(variable_selector)) params[key] = variable.value else: params[key] = value_param.get("value", "") if value_param is not None else None else: params[key] = None parameters = params tool["settings"] = {k: v.get("value", None) for k, v in tool.get("settings", {}).items()} tool["parameters"] = parameters if not for_log: if parameter.type == "array[tools]": value = cast(list[dict[str, Any]], value) tool_value = [] for tool in value: provider_type = ToolProviderType(tool.get("type", ToolProviderType.BUILT_IN)) setting_params = tool.get("settings", {}) parameters = tool.get("parameters", {}) manual_input_params = [key for key, value in parameters.items() if value is not None] parameters = {**parameters, **setting_params} entity = AgentToolEntity( provider_id=tool.get("provider_name", ""), provider_type=provider_type, tool_name=tool.get("tool_name", ""), tool_parameters=parameters, plugin_unique_identifier=tool.get("plugin_unique_identifier", None), credential_id=tool.get("credential_id", None), ) extra = tool.get("extra", {}) runtime_variable_pool: VariablePool | None = None if node_data.version != "1" or node_data.tool_node_version is not None: runtime_variable_pool = variable_pool tool_runtime = ToolManager.get_agent_tool_runtime( tenant_id, app_id, entity, user_id, invoke_from, runtime_variable_pool, ) if tool_runtime.entity.description: tool_runtime.entity.description.llm = ( extra.get("description", "") or tool_runtime.entity.description.llm ) for tool_runtime_params in tool_runtime.entity.parameters: tool_runtime_params.form = ( ToolParameter.ToolParameterForm.FORM if tool_runtime_params.name in manual_input_params else tool_runtime_params.form ) manual_input_value = {} if tool_runtime.entity.parameters: manual_input_value = { key: value for key, value in parameters.items() if key in manual_input_params } runtime_parameters = { **tool_runtime.runtime.runtime_parameters, **manual_input_value, } tool_value.append( { **tool_runtime.entity.model_dump(mode="json"), "runtime_parameters": runtime_parameters, "credential_id": tool.get("credential_id", None), "provider_type": provider_type.value, } ) value = tool_value if parameter.type == AgentStrategyParameter.AgentStrategyParameterType.MODEL_SELECTOR: value = cast(dict[str, Any], value) model_instance, model_schema = self.fetch_model( tenant_id=tenant_id, user_id=user_id, value=value, ) history_prompt_messages = [] if node_data.memory: memory = self.fetch_memory( variable_pool=variable_pool, app_id=app_id, model_instance=model_instance, ) if memory: prompt_messages = memory.get_history_prompt_messages( message_limit=node_data.memory.window.size or None ) history_prompt_messages = [ prompt_message.model_dump(mode="json") for prompt_message in prompt_messages ] value["history_prompt_messages"] = history_prompt_messages if model_schema: model_schema = self._remove_unsupported_model_features_for_old_version(model_schema) value["entity"] = model_schema.model_dump(mode="json") else: value["entity"] = None result[parameter_name] = value return result def build_credentials(self, *, parameters: dict[str, Any]) -> InvokeCredentials: credentials = InvokeCredentials() credentials.tool_credentials = {} for tool in parameters.get("tools", []): if not tool.get("credential_id"): continue try: identity = ToolIdentity.model_validate(tool.get("identity", {})) except ValidationError: continue credentials.tool_credentials[identity.provider] = tool.get("credential_id", None) return credentials def fetch_memory( self, *, variable_pool: VariablePool, app_id: str, model_instance: ModelInstance, ) -> TokenBufferMemory | None: conversation_id = get_system_text(variable_pool, SystemVariableKey.CONVERSATION_ID) if conversation_id is None: return None with Session(db.engine, expire_on_commit=False) as session: stmt = select(Conversation).where(Conversation.app_id == app_id, Conversation.id == conversation_id) conversation = session.scalar(stmt) if not conversation: return None return TokenBufferMemory(conversation=conversation, model_instance=model_instance) def fetch_model( self, *, tenant_id: str, user_id: str, value: dict[str, Any], ) -> tuple[ModelInstance, AIModelEntity | None]: assembly = create_plugin_model_assembly(tenant_id=tenant_id, user_id=user_id) provider_model_bundle = assembly.provider_manager.get_provider_model_bundle( tenant_id=tenant_id, provider=value.get("provider", ""), model_type=ModelType.LLM, ) model_name = value.get("model", "") model_credentials = provider_model_bundle.configuration.get_current_credentials( model_type=ModelType.LLM, model=model_name, ) provider_name = provider_model_bundle.configuration.provider.provider model_type_instance = provider_model_bundle.model_type_instance model_instance = assembly.model_manager.get_model_instance( tenant_id=tenant_id, provider=provider_name, model_type=ModelType(value.get("model_type", "")), model=model_name, ) model_schema = model_type_instance.get_model_schema(model_name, model_credentials) return model_instance, model_schema @staticmethod def _remove_unsupported_model_features_for_old_version(model_schema: AIModelEntity) -> AIModelEntity: if model_schema.features: for feature in model_schema.features[:]: try: AgentOldVersionModelFeatures(feature.value) except ValueError: model_schema.features.remove(feature) return model_schema @staticmethod def _filter_mcp_type_tool( strategy: ResolvedAgentStrategy, tools: list[dict[str, Any]], ) -> list[dict[str, Any]]: meta_version = strategy.meta_version if meta_version and Version(meta_version) > Version("0.0.1"): return tools return [tool for tool in tools if tool.get("type") != ToolProviderType.MCP]