feat(config): Adds BaseAgent.config_type field to indicate the config for the current agent and removes if-else branches against LlmAgent/LoopAgent/... in config_agent_utils::from_config

This makes the logic work with any user-defined agent with user-defined XxxAgentConfig.

PiperOrigin-RevId: 789845354

Author: Jacksunwei

src/google/adk/agents/base_agent.py

@@ -19,6 +19,7 @@
 from typing import AsyncGenerator
 from typing import Awaitable
 from typing import Callable
+from typing import ClassVar
 from typing import Dict
 from typing import final
 from typing import Mapping
@@ -75,6 +76,22 @@ class BaseAgent(BaseModel):
   )
   """The pydantic model config."""
 
+  config_type: ClassVar[type[BaseAgentConfig]] = BaseAgentConfig
+  """The config type for this agent.
+
+  Sub-classes should override this to specify their own config type.
+
+  Example:
+
+  ```
+  class MyAgentConfig(BaseAgentConfig):
+    my_field: str = ''
+
+  class MyAgent(BaseAgent):
+    config_type: ClassVar[type[BaseAgentConfig]] = MyAgentConfig
+  ```
+  """
+
   name: str
   """The agent's name.
 

src/google/adk/agents/config_agent_utils.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import importlib
+import inspect
 import os
 from typing import Any
 from typing import List
@@ -24,16 +25,9 @@
 from ..utils.feature_decorator import working_in_progress
 from .agent_config import AgentConfig
 from .base_agent import BaseAgent
+from .base_agent_config import BaseAgentConfig
 from .common_configs import AgentRefConfig
 from .common_configs import CodeConfig
-from .llm_agent import LlmAgent
-from .llm_agent_config import LlmAgentConfig
-from .loop_agent import LoopAgent
-from .loop_agent_config import LoopAgentConfig
-from .parallel_agent import ParallelAgent
-from .parallel_agent import ParallelAgentConfig
-from .sequential_agent import SequentialAgent
-from .sequential_agent import SequentialAgentConfig
 
 
 @working_in_progress("from_config is not ready for use.")
@@ -53,17 +47,36 @@ def from_config(config_path: str) -> BaseAgent:
   """
   abs_path = os.path.abspath(config_path)
   config = _load_config_from_path(abs_path)
-
-  if isinstance(config.root, LlmAgentConfig):
-    return LlmAgent.from_config(config.root, abs_path)
-  elif isinstance(config.root, LoopAgentConfig):
-    return LoopAgent.from_config(config.root, abs_path)
-  elif isinstance(config.root, ParallelAgentConfig):
-    return ParallelAgent.from_config(config.root, abs_path)
-  elif isinstance(config.root, SequentialAgentConfig):
-    return SequentialAgent.from_config(config.root, abs_path)
+  agent_config = config.root
+
+  # pylint: disable=unidiomatic-typecheck Needs exact class matching.
+  if type(agent_config) is BaseAgentConfig:
+    # Resolve the concrete agent config for user-defined agent classes.
+    agent_class = _resolve_agent_class(agent_config.agent_class)
+    agent_config = agent_class.config_type.model_validate(
+        agent_config.model_dump()
+    )
+    return agent_class.from_config(agent_config, abs_path)
   else:
-    raise ValueError("Unsupported config type")
+    # For built-in agent classes, no need to re-validate.
+    agent_class = _resolve_agent_class(agent_config.agent_class)
+    return agent_class.from_config(agent_config, abs_path)
+
+
+def _resolve_agent_class(agent_class: str) -> type[BaseAgent]:
+  """Resolve the agent class from its fully qualified name."""
+  agent_class_name = agent_class or "LlmAgent"
+  if "." not in agent_class_name:
+    agent_class_name = f"google.adk.agents.{agent_class_name}"
+
+  agent_class = _resolve_fully_qualified_name(agent_class_name)
+  if inspect.isclass(agent_class) and issubclass(agent_class, BaseAgent):
+    return agent_class
+
+  raise ValueError(
+      f"Invalid agent class `{agent_class_name}`. It must be a subclass of"
+      " BaseAgent."
+  )
 
 
 @working_in_progress("_load_config_from_path is not ready for use.")
@@ -90,6 +103,16 @@ def _load_config_from_path(config_path: str) -> AgentConfig:
   return AgentConfig.model_validate(config_data)
 
 
+@working_in_progress("_resolve_fully_qualified_name is not ready for use.")
+def _resolve_fully_qualified_name(name: str) -> Any:
+  try:
+    module_path, obj_name = name.rsplit(".", 1)
+    module = importlib.import_module(module_path)
+    return getattr(module, obj_name)
+  except Exception as e:
+    raise ValueError(f"Invalid fully qualified name: {name}") from e
+
+
 @working_in_progress("resolve_agent_reference is not ready for use.")
 def resolve_agent_reference(
     ref_config: AgentRefConfig, referencing_agent_config_abs_path: str

src/google/adk/agents/llm_agent.py

@@ -22,6 +22,7 @@
 from typing import AsyncGenerator
 from typing import Awaitable
 from typing import Callable
+from typing import ClassVar
 from typing import Literal
 from typing import Optional
 from typing import Type
@@ -55,6 +56,7 @@
 from ..tools.tool_context import ToolContext
 from ..utils.feature_decorator import working_in_progress
 from .base_agent import BaseAgent
+from .base_agent_config import BaseAgentConfig
 from .callback_context import CallbackContext
 from .common_configs import CodeConfig
 from .invocation_context import InvocationContext
@@ -131,6 +133,9 @@ class LlmAgent(BaseAgent):
   When not set, the agent will inherit the model from its ancestor.
   """
 
+  config_type: ClassVar[type[BaseAgentConfig]] = LlmAgentConfig
+  """The config type for this agent."""
+
   instruction: Union[str, InstructionProvider] = ''
   """Instructions for the LLM model, guiding the agent's behavior."""
 

src/google/adk/agents/loop_agent.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 from typing import AsyncGenerator
+from typing import ClassVar
 from typing import Optional
 from typing import Type
 
@@ -26,6 +27,7 @@
 from ..events.event import Event
 from ..utils.feature_decorator import working_in_progress
 from .base_agent import BaseAgent
+from .base_agent_config import BaseAgentConfig
 from .loop_agent_config import LoopAgentConfig
 
 
@@ -36,6 +38,9 @@ class LoopAgent(BaseAgent):
   reached, the loop agent will stop.
   """
 
+  config_type: ClassVar[type[BaseAgentConfig]] = LoopAgentConfig
+  """The config type for this agent."""
+
   max_iterations: Optional[int] = None
   """The maximum number of iterations to run the loop agent.
 

src/google/adk/agents/parallel_agent.py

@@ -18,13 +18,15 @@
 
 import asyncio
 from typing import AsyncGenerator
+from typing import ClassVar
 from typing import Type
 
 from typing_extensions import override
 
 from ..events.event import Event
 from ..utils.feature_decorator import working_in_progress
 from .base_agent import BaseAgent
+from .base_agent_config import BaseAgentConfig
 from .invocation_context import InvocationContext
 from .parallel_agent_config import ParallelAgentConfig
 
@@ -95,6 +97,9 @@ class ParallelAgent(BaseAgent):
   - Generating multiple responses for review by a subsequent evaluation agent.
   """
 
+  config_type: ClassVar[type[BaseAgentConfig]] = ParallelAgentConfig
+  """The config type for this agent."""
+
   @override
   async def _run_async_impl(
       self, ctx: InvocationContext

src/google/adk/agents/sequential_agent.py

@@ -24,6 +24,7 @@
 from ..events.event import Event
 from ..utils.feature_decorator import working_in_progress
 from .base_agent import BaseAgent
+from .base_agent import BaseAgentConfig
 from .invocation_context import InvocationContext
 from .llm_agent import LlmAgent
 from .sequential_agent_config import SequentialAgentConfig
@@ -32,6 +33,9 @@
 class SequentialAgent(BaseAgent):
   """A shell agent that runs its sub-agents in sequence."""
 
+  config_type: Type[BaseAgentConfig] = SequentialAgentConfig
+  """The config type for this agent."""
+
   @override
   async def _run_async_impl(
       self, ctx: InvocationContext