Add Human Input Mode in AGS (#4210)

* add react-flow

* update tutorial, some ui improvement for agent thread layout

* v1 for interactive viz for agent state

* add Ui visualization of message based transitions

* minor updates

* fix node edges, support visualization of self loops

* improve edges and layout, add support for selector group chat prompt

* minor ui tweaks

* ui and layout updates

* ugrade dependencies to fix dependabot scan errors

* persist sidebar, enable contentbar title mechanism #4191

* add support for user proxy agent, support human in put mode. #4011

* add UI support for human input mode via a userproxy agent  #4011

* version update

* fix db initialization bug

* support for human input mode in UI, fix backend api route minor bugs

* update pyproject toml and uv lock

* readme update, support full screen mode for agent visualiation

* update uv.lock

Author: victordibia

python/packages/autogen-studio/README.md

@@ -3,11 +3,11 @@
 [![PyPI version](https://badge.fury.io/py/autogenstudio.svg)](https://badge.fury.io/py/autogenstudio)
 [![Downloads](https://static.pepy.tech/badge/autogenstudio/week)](https://pepy.tech/project/autogenstudio)
 
-![ARA](./docs/ara_stockprices.png)
+![ARA](./docs/ags_screen.png)
 
 AutoGen Studio is an AutoGen-powered AI app (user interface) to help you rapidly prototype AI agents, enhance them with skills, compose them into workflows and interact with them to accomplish tasks. It is built on top of the [AutoGen](https://microsoft.github.io/autogen) framework, which is a toolkit for building AI agents.
 
-Code for AutoGen Studio is on GitHub at [microsoft/autogen](https://github.com/microsoft/autogen/tree/main/samples/apps/autogen-studio)
+Code for AutoGen Studio is on GitHub at [microsoft/autogen](https://github.com/microsoft/autogen/tree/main/python/packages/autogen-studio)
 
 > **Note**: AutoGen Studio is meant to help you rapidly prototype multi-agent workflows and demonstrate an example of end user interfaces built with AutoGen. It is not meant to be a production-ready app.
 
@@ -16,6 +16,7 @@ Code for AutoGen Studio is on GitHub at [microsoft/autogen](https://github.com/m
 
 **Updates**
 
+> Nov 14: AutoGen Studio is being rewritten to use the updated AutoGen 0.4.0 api AgentChat api.
 > April 17: AutoGen Studio database layer is now rewritten to use [SQLModel](https://sqlmodel.tiangolo.com/) (Pydantic + SQLAlchemy). This provides entity linking (skills, models, agents and workflows are linked via association tables) and supports multiple [database backend dialects](https://docs.sqlalchemy.org/en/20/dialects/) supported in SQLAlchemy (SQLite, PostgreSQL, MySQL, Oracle, Microsoft SQL Server). The backend database can be specified a `--database-uri` argument when running the application. For example, `autogenstudio ui --database-uri sqlite:///database.sqlite` for SQLite and `autogenstudio ui --database-uri postgresql+psycopg://user:password@localhost/dbname` for PostgreSQL.
 
 > March 12: Default directory for AutoGen Studio is now /home/<user>/.autogenstudio. You can also specify this directory using the `--appdir` argument when running the application. For example, `autogenstudio ui --appdir /path/to/folder`. This will store the database and other files in the specified directory e.g. `/path/to/folder/database.sqlite`. `.env` files in that directory will be used to set environment variables for the app.
@@ -49,7 +50,7 @@ There are two ways to install AutoGen Studio - from PyPi or from source. We **re
       pip install -e .
       ```
 
-    - Navigate to the `samples/apps/autogen-studio/frontend` directory, install dependencies, and build the UI:
+    - Navigate to the `python/packages/autogen-studio/frontend` directory, install dependencies, and build the UI:
 
       ```bash
       npm install -g gatsby-cli
@@ -88,16 +89,23 @@ AutoGen Studio also takes several parameters to customize the application:
 Now that you have AutoGen Studio installed and running, you are ready to explore its capabilities, including defining and modifying agent workflows, interacting with agents and sessions, and expanding agent skills.
 
 #### If running from source
+
 When running from source, you need to separately bring up the frontend server.
+
 1. Open a separate terminal and change directory to the frontend
+
 ```bash
 cd frontend
 ```
+
 3. Create a `.env.development` file.
+
 ```bash
 cp .env.default .env.development
 ```
+
 3. Launch frontend server
+
 ```bash
 npm run start
 ```

python/packages/autogen-studio/autogenstudio/components/__init__.py

@@ -0,0 +1 @@
+from .agents.userproxy import UserProxyAgent

python/packages/autogen-studio/autogenstudio/components/agents/__init__.py

@@ -0,0 +1,47 @@
+from typing import Callable, List, Optional, Sequence, Union, Awaitable
+from inspect import iscoroutinefunction
+
+from autogen_agentchat.agents import BaseChatAgent
+from autogen_agentchat.base import Response
+from autogen_agentchat.messages import ChatMessage, TextMessage
+from autogen_core.base import CancellationToken
+import asyncio
+
+
+class UserProxyAgent(BaseChatAgent):
+    """An agent that can represent a human user in a chat."""
+
+    def __init__(
+        self,
+        name: str,
+        description: Optional[str] = "a",
+        input_func: Optional[Union[Callable[..., str],
+                                   Callable[..., Awaitable[str]]]] = None
+    ) -> None:
+        super().__init__(name, description=description)
+        self.input_func = input_func or input
+        self._is_async = iscoroutinefunction(
+            input_func) if input_func else False
+
+    @property
+    def produced_message_types(self) -> List[type[ChatMessage]]:
+        return [TextMessage]
+
+    async def _get_input(self, prompt: str) -> str:
+        """Handle both sync and async input functions"""
+        if self._is_async:
+            return await self.input_func(prompt)
+        else:
+            return await asyncio.get_event_loop().run_in_executor(None, self.input_func, prompt)
+
+    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
+
+        try:
+            user_input = await self._get_input("Enter your response: ")
+            return Response(chat_message=TextMessage(content=user_input, source=self.name))
+        except Exception as e:
+            # Consider logging the error here
+            raise RuntimeError(f"Failed to get user input: {str(e)}") from e
+
+    async def on_reset(self, cancellation_token: CancellationToken) -> None:
+        pass

python/packages/autogen-studio/autogenstudio/components/agents/userproxy.py

@@ -1,3 +1,3 @@
 from .db_manager import DatabaseManager
-from .component_factory import ComponentFactory
+from .component_factory import ComponentFactory, Component
 from .config_manager import ConfigurationManager

python/packages/autogen-studio/autogenstudio/database/__init__.py

@@ -1,6 +1,6 @@
 import os
 from pathlib import Path
-from typing import List, Literal, Union, Optional, Dict, Any, Type
+from typing import Callable, List, Literal, Union, Optional, Dict, Any, Type
 from datetime import datetime
 import json
 from autogen_agentchat.task import MaxMessageTermination, TextMentionTermination, StopMessageTermination
@@ -13,6 +13,7 @@
     TeamTypes, AgentTypes, ModelTypes, ToolTypes,
     ComponentType, ComponentConfig, ComponentConfigInput, TerminationConfig, TerminationTypes, Response
 )
+from ..components import UserProxyAgent
 from autogen_agentchat.agents import AssistantAgent
 from autogen_agentchat.teams import RoundRobinGroupChat, SelectorGroupChat
 from autogen_ext.models import OpenAIChatCompletionClient
@@ -38,6 +39,17 @@
 Component = Union[RoundRobinGroupChat, SelectorGroupChat,
                   AssistantAgent, OpenAIChatCompletionClient, FunctionTool]
 
+DEFAULT_SELECTOR_PROMPT = """You are in a role play game. The following roles are available:
+{roles}.
+Read the following conversation. Then select the next role from {participants} to play. Only return the role.
+
+{history}
+
+Read the above conversation. Then select the next role from {participants} to play. Only return the role.
+"""
+
+CONFIG_RETURN_TYPES = Literal['object', 'dict', 'config']
+
 
 class ComponentFactory:
     """Creates and manages agent components with versioned configuration loading"""
@@ -55,19 +67,22 @@ def __init__(self):
         self._tool_cache: Dict[str, FunctionTool] = {}
         self._last_cache_clear = datetime.now()
 
-    async def load(self, component: ComponentConfigInput, return_type: ReturnType = 'object') -> Union[Component, dict, ComponentConfig]:
+    async def load(
+        self,
+        component: ComponentConfigInput,
+        input_func: Optional[Callable] = None,
+        return_type: ReturnType = 'object'
+    ) -> Union[Component, dict, ComponentConfig]:
         """
         Universal loader for any component type
 
         Args:
             component: Component configuration (file path, dict, or ComponentConfig)
+            input_func: Optional callable for user input handling
             return_type: Type of return value ('object', 'dict', or 'config')
 
         Returns:
             Component instance, config dict, or ComponentConfig based on return_type
-
-        Raises:
-            ValueError: If component type is unknown or version unsupported
         """
         try:
             # Load and validate config
@@ -95,8 +110,8 @@ async def load(self, component: ComponentConfigInput, return_type: ReturnType =
 
             # Otherwise create and return component instance
             handlers = {
-                ComponentType.TEAM: self.load_team,
-                ComponentType.AGENT: self.load_agent,
+                ComponentType.TEAM: lambda c: self.load_team(c, input_func),
+                ComponentType.AGENT: lambda c: self.load_agent(c, input_func),
                 ComponentType.MODEL: self.load_model,
                 ComponentType.TOOL: self.load_tool,
                 ComponentType.TERMINATION: self.load_termination
@@ -113,7 +128,7 @@ async def load(self, component: ComponentConfigInput, return_type: ReturnType =
             logger.error(f"Failed to load component: {str(e)}")
             raise
 
-    async def load_directory(self, directory: Union[str, Path], check_exists: bool = False, return_type: ReturnType = 'object') -> List[Union[Component, dict, ComponentConfig]]:
+    async def load_directory(self, directory: Union[str, Path], return_type: ReturnType = 'object') -> List[Union[Component, dict, ComponentConfig]]:
         """
         Import all component configurations from a directory.
         """
@@ -124,7 +139,7 @@ async def load_directory(self, directory: Union[str, Path], check_exists: bool =
             for path in list(directory.glob("*")):
                 if path.suffix.lower().endswith(('.json', '.yaml', '.yml')):
                     try:
-                        component = await self.load(path, return_type)
+                        component = await self.load(path, return_type=return_type)
                         components.append(component)
                     except Exception as e:
                         logger.info(
@@ -176,22 +191,17 @@ async def load_termination(self, config: TerminationConfig) -> TerminationCompon
             raise ValueError(
                 f"Termination condition creation failed: {str(e)}")
 
-    async def load_team(self, config: TeamConfig) -> TeamComponent:
+    async def load_team(
+        self,
+        config: TeamConfig,
+        input_func: Optional[Callable] = None
+    ) -> TeamComponent:
         """Create team instance from configuration."""
-
-        default_selector_prompt = """You are in a role play game. The following roles are available:
-{roles}.
-Read the following conversation. Then select the next role from {participants} to play. Only return the role.
-
-{history}
-
-Read the above conversation. Then select the next role from {participants} to play. Only return the role.
-"""
         try:
-            # Load participants (agents)
+            # Load participants (agents) with input_func
             participants = []
             for participant in config.participants:
-                agent = await self.load(participant)
+                agent = await self.load(participant, input_func=input_func)
                 participants.append(agent)
 
             # Load model client if specified
@@ -202,7 +212,6 @@ async def load_team(self, config: TeamConfig) -> TeamComponent:
             # Load termination condition if specified
             termination = None
             if config.termination_condition:
-                # Now we can use the universal load() method since termination is a proper component
                 termination = await self.load(config.termination_condition)
 
             # Create team based on type
@@ -215,7 +224,7 @@ async def load_team(self, config: TeamConfig) -> TeamComponent:
                 if not model_client:
                     raise ValueError(
                         "SelectorGroupChat requires a model_client")
-                selector_prompt = config.selector_prompt if config.selector_prompt else default_selector_prompt
+                selector_prompt = config.selector_prompt if config.selector_prompt else DEFAULT_SELECTOR_PROMPT
                 return SelectorGroupChat(
                     participants=participants,
                     model_client=model_client,
@@ -229,24 +238,37 @@ async def load_team(self, config: TeamConfig) -> TeamComponent:
             logger.error(f"Failed to create team {config.name}: {str(e)}")
             raise ValueError(f"Team creation failed: {str(e)}")
 
-    async def load_agent(self, config: AgentConfig) -> AgentComponent:
+    async def load_agent(
+        self,
+        config: AgentConfig,
+        input_func: Optional[Callable] = None
+    ) -> AgentComponent:
         """Create agent instance from configuration."""
         try:
             # Load model client if specified
             model_client = None
             if config.model_client:
                 model_client = await self.load(config.model_client)
+
             system_message = config.system_message if config.system_message else "You are a helpful assistant"
+
             # Load tools if specified
             tools = []
             if config.tools:
                 for tool_config in config.tools:
                     tool = await self.load(tool_config)
                     tools.append(tool)
 
-            if config.agent_type == AgentTypes.ASSISTANT:
+            if config.agent_type == AgentTypes.USERPROXY:
+                return UserProxyAgent(
+                    name=config.name,
+                    description=config.description or "A human user",
+                    input_func=input_func  # Pass through to UserProxyAgent
+                )
+            elif config.agent_type == AgentTypes.ASSISTANT:
                 return AssistantAgent(
                     name=config.name,
+                    description=config.description or "A helpful assistant",
                     model_client=model_client,
                     tools=tools,
                     system_message=system_message

python/packages/autogen-studio/autogenstudio/database/component_factory.py

@@ -1,3 +1,4 @@
+from pathlib import Path
 import threading
 from datetime import datetime
 from typing import Optional
@@ -19,15 +20,34 @@ class DatabaseManager:
 
     _init_lock = threading.Lock()
 
-    def __init__(self, engine_uri: str, auto_upgrade: bool = True):
+    def __init__(
+        self,
+        engine_uri: str,
+        base_dir: Optional[Path | str] = None,
+        auto_upgrade: bool = True
+    ):
+        """
+        Initialize DatabaseManager with optional custom base directory.
+
+        Args:
+            engine_uri: Database connection URI
+            base_dir: Custom base directory for Alembic files. If None, uses current working directory
+            auto_upgrade: Whether to automatically upgrade schema when differences found
+        """
+        # Convert string path to Path object if necessary
+        if isinstance(base_dir, str):
+            base_dir = Path(base_dir)
+
         connection_args = {
-            "check_same_thread": True} if "sqlite" in engine_uri else {}
+            "check_same_thread": True
+        } if "sqlite" in engine_uri else {}
+
         self.engine = create_engine(engine_uri, connect_args=connection_args)
         self.schema_manager = SchemaManager(
             engine=self.engine,
+            base_dir=base_dir,
             auto_upgrade=auto_upgrade,
         )
-
         # Check and upgrade on startup
         upgraded, status = self.schema_manager.check_and_upgrade()
         if upgraded: