UN-3276 support for user defined llm in hocon file with class key

llm_extension/groq_langchain_llm_factory.py

@@ -0,0 +1,58 @@
+
+# Copyright (C) 2023-2025 Cognizant Digital Business, Evolutionary AI.
+# All Rights Reserved.
+# Issued under the Academic Public License.
+#
+# You can be released from the terms, and requirements of the Academic Public
+# License by purchasing a commercial license.
+# Purchase of a commercial license is mandatory for any use of the
+# neuro-san SDK Software in commercial settings.
+#
+# END COPYRIGHT
+
+from typing import Any
+from typing import Dict
+from typing import List
+
+from langchain_core.callbacks.base import BaseCallbackHandler
+from langchain_core.language_models.base import BaseLanguageModel
+from langchain_groq import ChatGroq
+
+from neuro_san.internals.run_context.langchain.llms.langchain_llm_factory import LangChainLlmFactory
+
+
+class GroqLangChainLlmFactory(LangChainLlmFactory):
+    """
+    Factory class for LLM operations
+    """
+
+    def create_base_chat_model(self, config: Dict[str, Any],
+                               callbacks: List[BaseCallbackHandler] = None) -> BaseLanguageModel:
+        """
+        Create a BaseLanguageModel from the fully-specified llm config.
+        :param config: The fully specified llm config which is a product of
+                    _create_full_llm_config() above.
+        :param callbacks: A list of BaseCallbackHandlers to add to the chat model.
+        :return: A BaseLanguageModel (can be Chat or LLM)
+                Can raise a ValueError if the config's class or model_name value is
+                unknown to this method.
+        """
+        # Construct the LLM
+        llm: BaseLanguageModel = None
+        chat_class: str = config.get("class")
+        if chat_class is not None:
+            chat_class = chat_class.lower()
+
+        model_name: str = config.get("model_name")
+
+        if chat_class == "groq":
+            llm = ChatGroq(
+                model=model_name,
+                temperature=config.get("temperature")
+            )
+        elif chat_class is None:
+            raise ValueError(f"Class name {chat_class} for model_name {model_name} is unspecified.")
+        else:
+            raise ValueError(f"Class {chat_class} for model_name {model_name} is unrecognized.")
+
+        return llm

llm_extension/llm_info.hocon

@@ -0,0 +1,26 @@
+# Copyright (C) 2023-2025 Cognizant Digital Business, Evolutionary AI.
+# All Rights Reserved.
+# Issued under the Academic Public License.
+#
+# You can be released from the terms, and requirements of the Academic Public
+# License by purchasing a commercial license.
+# Purchase of a commercial license is mandatory for any use of the
+# neuro-san SDK Software in commercial settings.
+#
+# END COPYRIGHT
+
+# The schema specifications for this file are documented here:
+# https://github.com/cognizant-ai-lab/neuro-san/blob/main/docs/llm_info_hocon_reference.md
+
+{
+
+
+    "classes": {
+        "factories": [ "llm_extension.groq_langchain_llm_factory.GroqLangChainLlmFactory" ]
+        "groq": {
+            # Add arguments like temperature that you want to pass to the llm here.
+            "temperature": 0.7
+        }
+    }
+
+}

neuro_san/internals/run_context/langchain/core/langchain_run_context.py

@@ -23,9 +23,8 @@
 from logging import Logger
 from logging import getLogger
 
-from openai import APIError
-from anthropic import BadRequestError
-from anthropic import AuthenticationError
+from openai import APIError as OpenAI_APIError
+from anthropic import APIError as Anthropic_APIError
 
 from pydantic_core import ValidationError
 
@@ -495,7 +494,7 @@ async def ainvoke(self, agent_executor: AgentExecutor, inputs: Dict[str, Any], i
         while return_dict is None and retries > 0:
             try:
                 return_dict: Dict[str, Any] = await agent_executor.ainvoke(inputs, invoke_config)
-            except (APIError, BadRequestError, AuthenticationError, ChatGoogleGenerativeAIError) as api_error:
+            except (OpenAI_APIError, Anthropic_APIError, ChatGoogleGenerativeAIError) as api_error:
                 message: str = ApiKeyErrorCheck.check_for_api_key_exception(api_error)
                 if message is not None:
                     raise ValueError(message) from api_error

neuro_san/internals/run_context/langchain/llms/default_llm_factory.py

@@ -34,6 +34,7 @@
 from neuro_san.internals.run_context.langchain.llms.llm_info_restorer import LlmInfoRestorer
 from neuro_san.internals.run_context.langchain.llms.standard_langchain_llm_factory import StandardLangChainLlmFactory
 from neuro_san.internals.run_context.langchain.util.api_key_error_check import ApiKeyErrorCheck
+from neuro_san.internals.run_context.langchain.util.argument_validator import ArgumentValidator
 
 
 class DefaultLlmFactory(ContextTypeLlmFactory, LangChainLlmFactory):
@@ -123,24 +124,13 @@ def resolve_one_llm_factory(self, llm_factory_class_name: str, llm_info_file: st
             raise ValueError(f"The value for the classes.factories key in {llm_info_file} "
                              "must be a list of strings")
 
-        class_split: List[str] = llm_factory_class_name.split(".")
-        if len(class_split) <= 2:
-            raise ValueError(f"Value in the classes.factories in {llm_info_file} must be of the form "
-                             "<package_name>.<module_name>.<ClassName>")
-
-        # Create a list of a single package given the name in the value
-        packages: List[str] = [".".join(class_split[:-2])]
-        class_name: str = class_split[-1]
-        resolver = Resolver(packages)
-
-        # Resolve the class name
-        llm_factory_class: Type[LangChainLlmFactory] = None
-        try:
-            llm_factory_class: Type[LangChainLlmFactory] = \
-                resolver.resolve_class_in_module(class_name, module_name=class_split[-2])
-        except AttributeError as exception:
-            raise ValueError(f"Class {llm_factory_class_name} in {llm_info_file} "
-                             "not found in PYTHONPATH") from exception
+        # Resolve the factory class
+        llm_factory_class: Type[LangChainLlmFactory] = self._resolve_class_from_path(
+            class_path=llm_factory_class_name,
+            expected_base=LangChainLlmFactory,
+            source_file=llm_info_file,
+            description="classes.factories"
+        )
 
         # Instantiate it
         try:
@@ -155,7 +145,43 @@ def resolve_one_llm_factory(self, llm_factory_class_name: str, llm_info_file: st
                              "must be of type LangChainLlmFactory")
         return llm_factory
 
-    def create_llm(self, config: Dict[str, Any], callbacks: List[BaseCallbackHandler] = None) -> BaseLanguageModel:
+    def _resolve_class_from_path(
+            self,
+            class_path: str,
+            expected_base: Type,
+            source_file: str,
+            description: str
+    ) -> Type:
+
+        parts: List[str] = class_path.split(".")
+        if len(parts) <= 2:
+            raise ValueError(
+                f"Value for '{description}' in {source_file} must be of the form "
+                "<package>.<module>.<ClassName>"
+            )
+
+        module_name: str = parts[-2]
+        class_name: str = parts[-1]
+        packages: str = [".".join(parts[:-2])]
+        resolver = Resolver(packages)
+
+        try:
+            cls: Type = resolver.resolve_class_in_module(class_name, module_name=module_name)
+        except AttributeError as e:
+            raise ValueError(f"Class {class_path} in {source_file} not found in PYTHONPATH") from e
+
+        if not issubclass(cls, expected_base):
+            raise ValueError(
+                f"Class {class_path} in {source_file} must be a subclass of {expected_base.__name__}"
+            )
+
+        return cls
+
+    def create_llm(
+            self,
+            config: Dict[str, Any],
+            callbacks: Optional[List[BaseCallbackHandler]] = None
+    ) -> BaseLanguageModel:
         """
         Creates a langchain LLM based on the 'model_name' value of
         the config passed in.
@@ -176,6 +202,28 @@ def create_full_llm_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
         :param config: The llm_config from the user
         :return: The fully specified config with defaults filled in.
         """
+
+        class_from_llm_config: str = config.get("class")
+        if class_from_llm_config:
+            if not isinstance(class_from_llm_config, str):
+                raise ValueError("Value of 'class' has to be string.")
+            # A "class" key in the config indicates the user has specified a particular LLM implementation.
+            # However, the config may only contain partial arguments (e.g., {"arg_1": 0.5}) and omit others.
+            #
+            # In the standard factory, LLM classes are instantiated like:
+            #   ChatOpenAI(arg_1=config.get("arg_1"), arg_2=config.get("arg_2"))
+            # If a required argument like "arg_2" is missing in the config, config.get("arg_2") returns None,
+            # which may raise an error during instantiation if the argument has no default.
+            #
+            # To prevent this, we first fetch the default arguments for the given class from llm_info,
+            # then merge them with the user-provided config. This ensures all expected arguments are present,
+            # and the user’s config values take precedence over the defaults.
+            config_from_class_in_llm_info: Dict[str, Any] = self.get_chat_class_args(class_from_llm_config)
+
+            # Merge the defaults from llm_info with the user-defined config,
+            # giving priority to values in config.
+            return self.overlayer.overlay(config_from_class_in_llm_info, config)
+
         default_config: Dict[str, Any] = self.llm_infos.get("default_config")
         use_config = self.overlayer.overlay(default_config, config)
 
@@ -215,7 +263,7 @@ def create_full_llm_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
 
         return full_config
 
-    def get_chat_class_args(self, chat_class_name: str, use_model_name: str) -> Dict[str, Any]:
+    def get_chat_class_args(self, chat_class_name: str, use_model_name: Optional[str] = None) -> Dict[str, Any]:
         """
         :param chat_class_name: string name of the chat class to look up.
         :param use_model_name: the original model name that prompted the chat class lookups
@@ -227,8 +275,9 @@ def get_chat_class_args(self, chat_class_name: str, use_model_name: str) -> Dict
         chat_classes: Dict[str, Any] = self.llm_infos.get("classes")
         chat_class: Dict[str, Any] = chat_classes.get(chat_class_name)
         if chat_class is None:
-            raise ValueError(f"llm info entry for {use_model_name} uses a 'class' of {chat_class_name} "
-                             "which is not defined in the 'classes' table.")
+            # If chat_class_name is not in "classes" in llm_info,
+            # it could be a user-specified langchain model class
+            return {}
 
         # Get the args from the chat class
         args: Dict[str, Any] = chat_class.get("args")
@@ -242,9 +291,10 @@ def get_chat_class_args(self, chat_class_name: str, use_model_name: str) -> Dict
         return args
 
     def create_base_chat_model(self, config: Dict[str, Any],
-                               callbacks: List[BaseCallbackHandler] = None) -> BaseLanguageModel:
+                               callbacks: Optional[List[BaseCallbackHandler]] = None) -> BaseLanguageModel:
         """
-        Create a BaseLanguageModel from the fully-specified llm config.
+        Create a BaseLanguageModel from the fully-specified llm config either from standard LLM factory,
+        user-defined LLM factory, or user-specified langchain model class.
         :param config: The fully specified llm config which is a product of
                     _create_full_llm_config() above.
         :param callbacks: A list of BaseCallbackHandlers to add to the chat model.
@@ -279,11 +329,55 @@ def create_base_chat_model(self, config: Dict[str, Any],
                 # Let the next model have a crack
                 found_exception = exception
 
+        # Try resolving via 'class' in config if factories failed
+        class_path: str = config.get("class")
+        if llm is None and found_exception is not None and class_path:
+            llm = self.create_base_chat_model_from_user_class(class_path, config)
+            found_exception = None
+
         if found_exception is not None:
             raise found_exception
 
         return llm
 
+    def create_base_chat_model_from_user_class(
+            self,
+            class_path: str,
+            config: Dict[str, Any],
+            callbacks: Optional[List[BaseCallbackHandler]] = None
+    ) -> BaseLanguageModel:
+        """
+        Create a BaseLanguageModel from the user-specified langchain model class.
+        :param class_path: A string in the form of <package>.<module>.<Class>
+        :param config: The fully specified llm config which is a product of
+                    _create_full_llm_config() above.
+        :param callbacks: A list of BaseCallbackHandlers to add to the chat model.
+
+        :return: A BaseLanguageModel
+        """
+
+        if not isinstance(class_path, str):
+            raise ValueError("'class' in llm_config must be a string")
+
+        # Resolve the 'class'
+        llm_class: Type[BaseLanguageModel] = self._resolve_class_from_path(
+            class_path=class_path,
+            expected_base=BaseLanguageModel,
+            source_file="agent network hocon file",
+            description="llm_config"
+        )
+
+        # Copy the config, take 'class' out, and add callbacks
+        # Then unpack into llm constructor
+        user_config: Dict[str, Any] = config.copy()
+        user_config.pop("class")
+        user_config["callbacks"] = callbacks
+
+        # Check for invalid args and throw error if found
+        ArgumentValidator.check_invalid_args(llm_class, user_config)
+
+        return llm_class(**user_config)
+
     def get_max_prompt_tokens(self, config: Dict[str, Any]) -> int:
         """
         :param config: A dictionary which describes which LLM to use.

neuro_san/internals/run_context/langchain/llms/standard_langchain_llm_factory.py

@@ -70,7 +70,10 @@ def create_base_chat_model(self, config: Dict[str, Any],
         if chat_class is not None:
             chat_class = chat_class.lower()
 
-        model_name: str = config.get("model_name")
+        # Check for key "model_name", "model", and "model_id" to use as model name
+        # If the config is from default_llm_info, this is always "model_name"
+        # but with user-specified config, it is possible to have the other keys will be specifed instead.
+        model_name: str = config.get("model_name") or config.get("model") or config.get("model_id")
 
         if chat_class == "openai":
             llm = ChatOpenAI(

neuro_san/internals/run_context/langchain/toolbox/toolbox_factory.py

@@ -10,14 +10,12 @@
 #
 # END COPYRIGHT
 
-from inspect import signature
-from types import MethodType
+
 from typing import Any
 from typing import Callable
 from typing import Dict
 from typing import List
 from typing import Optional
-from typing import Set
 from typing import Type
 from typing import Union
 
@@ -32,6 +30,7 @@
 
 from neuro_san.internals.interfaces.context_type_toolbox_factory import ContextTypeToolboxFactory
 from neuro_san.internals.run_context.langchain.toolbox.toolbox_info_restorer import ToolboxInfoRestorer
+from neuro_san.internals.run_context.langchain.util.argument_validator import ArgumentValidator
 
 
 class ToolboxFactory(ContextTypeToolboxFactory):
@@ -152,7 +151,7 @@ def create_tool_from_toolbox(
             self._get_from_api_wrapper_method(tool_class) or tool_class
 
         # Validate and instantiate
-        self._check_invalid_args(callable_obj, final_args)
+        ArgumentValidator.check_invalid_args(callable_obj, final_args)
         # Instance can be a BaseTool or a BaseToolkit
         instance: Union[BaseTool, BaseToolkit] = callable_obj(**final_args)
 
@@ -178,7 +177,7 @@ def _resolve_args(self, args: Dict[str, Any]) -> Dict[str, Any]:
                 # If the argument is a class definition, resolve and instantiate it
                 nested_class: BaseModel = self._resolve_class(value.get("class"))
                 nested_args: Dict[str, Any] = self._resolve_args(value.get("args", empty))
-                self._check_invalid_args(nested_class, nested_args)
+                ArgumentValidator.check_invalid_args(nested_class, nested_args)
                 resolved_args[key] = nested_class(**nested_args)
             else:
                 # Otherwise, keep primitive values as they are
@@ -210,22 +209,6 @@ def _resolve_class(self, class_path: str) -> Type[BaseTool]:
         except AttributeError as exception:
             raise ValueError(f"Class {class_path} not found in PYTHONPATH") from exception
 
-    def _check_invalid_args(self, method_class: Union[Type, MethodType], args: Dict[str, Any]):
-        """
-        Check for invalid arguments in class or method
-        :param method_class: Class or method to check for the invalid arguments
-        :param args: Arguments to check
-        """
-        class_args_set: Set[str] = set(signature(method_class).parameters.keys())
-        args_set: Set[str] = set(args.keys())
-        invalid_args: Set[str] = args_set - class_args_set
-
-        if invalid_args:
-            raise ValueError(
-                f"Arguments {invalid_args} for '{method_class.__name__}' do not match any attributes "
-                "of the class or any arguments of the method."
-            )
-
     def _get_from_api_wrapper_method(
         self,
         tool_class: Union[Type[BaseTool], Type[BaseToolkit]]