AgentScope v2 roadmap for AgentScope Studio as a production client delivery layer? #1909
Replies: 3 comments
|
One thing worth pointing out that wasn't covered in the previous response: AgentScope 2.0's main repository includes a From what we can tell, it provides a real-time conversational interface built on top of the SSE event stream, with streaming rendering of Agent responses, tool call results, and multi-session management. But honestly, the source code will tell you more than we can describe — worth a direct look: 👉 examples/web_ui We'd love to hear your thoughts after taking a look, especially around:
If you're open to sharing more, feel free to continue the discussion here or open a dedicated Discussion thread on client delivery — it's a topic that deserves more focused attention, and your production experience is exactly the kind of input that helps. |
|
Hi, please check this MR in the agent scope studio: I have also added a md file explaining why each changes was done. Changes in the agent scope 1.x version: AsyncAsStudioForwardMessagePrePrintHook.py import httpx
import asyncio
import shortuuid
import logging
from contextvars import ContextVar
from typing import Any, Dict, Final, Optional, Set
from agentscope.agent import AgentBase, UserAgent
logger: logging.Logger = logging.getLogger(__name__)
current_run_id: ContextVar[Optional[str]] = ContextVar("current_run_id", default=None)
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
MUTED_AGENTS: Final[Set[str]] = {
# mute any agent from sending the trace to the as studio
}
async def async_as_studio_forward_message_pre_print_hook(
self: AgentBase,
kwargs: dict[str, Any],
studio_url: str,
run_id: str,
) -> None:
"""The pre-speak hook to forward messages to the studio."""
if getattr(self, "name", "") in MUTED_AGENTS:
return
run_id = current_run_id.get() or run_id
msg = kwargs["msg"]
message_data: Dict[str, Any] = msg.to_dict()
if hasattr(self, "_reply_id"):
reply_id: str = getattr(self, "_reply_id")
else:
reply_id = shortuuid.uuid()
n_retry: int = 0
async with httpx.AsyncClient() as client:
while True:
try:
res = await client.post(
f"{studio_url}/trpc/pushMessage",
json={
"runId": run_id,
"replyId": reply_id,
"replyName": getattr(self, "name", msg.name),
"replyRole": "user"
if isinstance(self, UserAgent)
else "assistant",
"msg": message_data,
},
)
res.raise_for_status()
break
except Exception as e:
if n_retry < 3:
n_retry += 1
# Use asyncio.sleep to avoid blocking the loop during retry backoff
await asyncio.sleep(1)
continue
# Graceful degradation: log warning and return to avoid crashing
logger.warning(
"Failed to forward message to Studio after %d retries: %s. "
"Agent will continue without Studio forwarding.",
n_retry,
e,
)
returnand changes in AsyncStudioUserInput.py """Fully async Studio user input implementation."""
import asyncio
from typing import Any, Type, List
import httpx
import shortuuid
import socketio
from agentscope.agent import UserInputBase, UserInputData
from agentscope.message import TextBlock, ImageBlock, AudioBlock, VideoBlock
from pydantic import BaseModel
from loguru import logger
class AsyncStudioUserInput(UserInputBase):
"""Fully async user input handler for AgentScope Studio.
This implementation avoids blocking the OS thread by using:
- socketio.AsyncClient
- httpx.AsyncClient
- asyncio.Queue
- asyncio.Event
- asyncio.sleep
"""
_websocket_namespace: str = "/python"
def __init__(
self,
studio_url: str,
run_id: str,
max_retries: int = 3,
reconnect_attempts: int = 3,
reconnection_delay: int = 1,
reconnection_delay_max: int = 5,
request_timeout: float = 30.0,
user_input_timeout: float | None = None,
) -> None:
self._is_connected = False
self._is_reconnecting = False
self._connect_lock = asyncio.Lock()
self.studio_url = studio_url.rstrip("/")
self.run_id = run_id
self.max_retries = max_retries
self.request_timeout = request_timeout
self.user_input_timeout = user_input_timeout
self.sio = socketio.AsyncClient(
reconnection_attempts=reconnect_attempts,
reconnection_delay=reconnection_delay,
reconnection_delay_max=reconnection_delay_max,
)
self.http_client = httpx.AsyncClient(
timeout=httpx.Timeout(request_timeout),
)
self.input_queues: dict[str, asyncio.Queue[UserInputData]] = {}
self.input_events: dict[str, asyncio.Event] = {}
self._register_socket_handlers()
def _register_socket_handlers(self) -> None:
@self.sio.on("connect", namespace=self._websocket_namespace)
async def on_connect() -> None:
self._is_connected = True
self._is_reconnecting = False
logger.info(
'Connected to AgentScope Studio at "%s" with run name "%s".',
self.studio_url,
self.run_id,
)
@self.sio.on("disconnect", namespace=self._websocket_namespace)
async def on_disconnect() -> None:
self._is_connected = False
logger.info(
"Disconnected from AgentScope Studio at %s",
self.studio_url,
)
@self.sio.on("reconnect", namespace=self._websocket_namespace)
async def on_reconnect(attempt_number: int) -> None:
self._is_connected = True
self._is_reconnecting = False
logger.info(
"Reconnected to AgentScope Studio at %s with run_id %s after %d attempts",
self.studio_url,
self.run_id,
attempt_number,
)
@self.sio.on("reconnect_attempt", namespace=self._websocket_namespace)
async def on_reconnect_attempt(attempt_number: int) -> None:
self._is_reconnecting = True
logger.info(
"Attempting to reconnect to AgentScope Studio at %s "
"(attempt %d)",
self.studio_url,
attempt_number,
)
@self.sio.on("reconnect_failed", namespace=self._websocket_namespace)
async def on_reconnect_failed() -> None:
self._is_reconnecting = False
logger.error(
"Failed to reconnect to AgentScope Studio at %s",
self.studio_url,
)
@self.sio.on("reconnect_error", namespace=self._websocket_namespace)
async def on_reconnect_error(error: Any) -> None:
logger.error(
"Error while reconnecting to AgentScope Studio at %s: %s",
self.studio_url,
str(error),
)
@self.sio.on("forwardUserInput", namespace=self._websocket_namespace)
async def receive_user_input(
request_id: str,
blocks_input: List[
TextBlock | ImageBlock | AudioBlock | VideoBlock
],
structured_input: dict[str, Any],
) -> None:
queue = self.input_queues.get(request_id)
event = self.input_events.get(request_id)
if queue is None or event is None:
logger.warning(
"Received user input for unknown or expired request_id=%s",
request_id,
)
return
await queue.put(
UserInputData(
blocks_input=blocks_input,
structured_input=structured_input,
),
)
event.set()
async def connect(self) -> None:
"""Connect to AgentScope Studio.
Call this once during application startup, or lazily before the first
user input request.
"""
async with self._connect_lock:
if self.sio.connected and self._is_connected:
return
try:
await self.sio.connect(
self.studio_url,
namespaces=[self._websocket_namespace],
auth={"run_id": self.run_id},
)
except Exception as e:
raise RuntimeError(
f"Failed to connect to AgentScope Studio at {self.studio_url}",
) from e
async def _ensure_connected(
self,
timeout: float = 30.0,
check_interval: float = 1.0,
) -> None:
"""Ensure the socket connection is ready without blocking the OS thread."""
if self._is_connected:
return
if not self.sio.connected:
await self.connect()
if self._is_connected:
return
if self._is_reconnecting:
start_time = asyncio.get_running_loop().time()
while self._is_reconnecting:
elapsed_time = asyncio.get_running_loop().time() - start_time
if elapsed_time > timeout:
raise RuntimeError(
f"Reconnection timeout after {elapsed_time:.1f} seconds",
)
logger.info(
"Waiting for reconnection... (%.1fs / %.1fs)",
elapsed_time,
timeout,
)
await asyncio.sleep(check_interval)
if self._is_connected:
return
raise RuntimeError(
f"Not connected to AgentScope Studio at {self.studio_url}.",
)
async def __call__(
self,
agent_id: str,
agent_name: str,
*args: Any,
structured_model: Type[BaseModel] | None = None,
**kwargs: Any,
) -> UserInputData:
"""Request user input from AgentScope Studio asynchronously."""
await self._ensure_connected()
request_id = shortuuid.uuid()
self.input_queues[request_id] = asyncio.Queue(maxsize=1)
self.input_events[request_id] = asyncio.Event()
if structured_model is None:
structured_input = None
else:
structured_input = structured_model.model_json_schema()
try:
await self._request_user_input(
request_id=request_id,
agent_id=agent_id,
agent_name=agent_name,
structured_input=structured_input,
)
event = self.input_events[request_id]
queue = self.input_queues[request_id]
if self.user_input_timeout is None:
await event.wait()
else:
await asyncio.wait_for(
event.wait(),
timeout=self.user_input_timeout,
)
return await queue.get()
except asyncio.TimeoutError as e:
raise RuntimeError(
f"Timed out waiting for user input for request_id={request_id}",
) from e
finally:
self.input_queues.pop(request_id, None)
self.input_events.pop(request_id, None)
async def _request_user_input(
self,
request_id: str,
agent_id: str,
agent_name: str,
structured_input: dict[str, Any] | None,
) -> None:
"""Notify AgentScope Studio that this run requires user input."""
last_error: Exception | None = None
for attempt in range(1, self.max_retries + 1):
try:
response = await self.http_client.post(
f"{self.studio_url}/trpc/requestUserInput",
json={
"requestId": request_id,
"runId": self.run_id,
"agentId": agent_id,
"agentName": agent_name,
"structuredInput": structured_input,
},
)
response.raise_for_status()
return
except Exception as e:
last_error = e
if attempt >= self.max_retries:
break
logger.warning(
"Failed to request user input from AgentScope Studio. "
"Retrying attempt %d/%d. Error: %s",
attempt,
self.max_retries,
str(e),
)
await asyncio.sleep(0.5 * attempt)
raise RuntimeError(
"Failed to request user input from AgentScope Studio",
) from last_error
async def close(self) -> None:
"""Close socket and HTTP resources."""
try:
if self.sio.connected:
await self.sio.disconnect()
finally:
await self.http_client.aclose()
async def __aenter__(self) -> "AsyncStudioUserInput":
await self.connect()
return self
async def __aexit__(self, exc_type, exc, tb) -> None:
await self.close()Rationale for Separating AS Studio from the Main Agent Service The primary reason for separating AS Studio from the main Agent Service is to establish a clear boundary between agent execution and client interaction/state management. The separation provides two key benefits: Independent and autonomous agent execution Agents should be able to run independently in the background without being coupled to client sessions, connections, or other client-facing concerns. Once a task has been initiated, the Agent Service can continue processing it asynchronously, regardless of whether the client remains connected, disconnects, or reconnects later. This allows agents to focus purely on execution and business logic rather than having to manage client lifecycle, session state, or communication concerns. AS Studio as a durable buffer between agents and clients AS Studio acts as a buffer and persistence layer between the Agent Service and the client. Any meaningful processing, intermediate state, results, events, or outputs produced by the Agent Service can be persisted in AS Studio. The client therefore does not need to maintain a continuous connection to the Agent Service in order to receive the outcome of a task. If a client disconnects while an agent is processing, the Agent Service can continue its work independently. When the client reconnects, AS Studio can provide the relevant state, history, progress, and results without requiring the Agent Service to reconstruct the previous client session or manage the reconnection itself. Architectural Principle In this model: Client ↔ AS Studio ↔ Agent Service The responsibilities are intentionally separated: Client — initiates requests, observes progress, and consumes results. This separation also means that the Agent Service does not need to know whether a client is currently connected, disconnected, or reconnecting. Its responsibility ends with reliably executing the requested work and publishing its state/results back to AS Studio. Additional Benefits This architecture also provides a few longer-term advantages: Fault isolation: Client-facing issues should not directly interrupt agent execution. The key design goal is therefore not merely to introduce another service, but to create a durable boundary between "doing the work" and "managing the client experience." AS Studio becomes the system of record for the client-visible state, while the Agent Service remains an autonomous execution layer. I would really appreciate any suggestions or feedback on the approach above. If I’m missing anything in the scope of Agent 2.0, especially something that could significantly improve the current solution, I’d be happy to hear your thoughts. |
Uh oh!
There was an error while loading. Please reload this page.
Hi AgentScope team,
First, thank you for AgentScope and AgentScope Studio. I really liked AgentScope Studio, not only as a development/debugging UI, but also as a client delivery layer.
In my AgentScope v1-based setup, I customized both AgentScope Studio and my AgentScope service so I could run it in a production-like environment. This worked very well for my use case because Studio already handled many delivery concerns by default: the frontend UI, conversation flow, user input handling, real-time updates, tRPC/WebSocket/Socket.IO-style communication, and the general plumbing between the client and the running agent service. Because of that, I did not need to build a separate client delivery system from scratch.
After checking AgentScope v2, I see that the architecture has changed significantly and the recommended path now seems to be Agent Service, event streaming, and the newer frontend/web UI approach. I understand that v2 is a major rewrite, but I am not fully clear on what this means for AgentScope Studio.
Could you please clarify the intended direction?
My main concern is not only debugging or tracing, but client delivery. Studio previously gave me a working end-to-end client/server interaction model, and I would like to understand whether AgentScope v2 will provide an equivalent official path.
Thanks again for the great work. I would be happy to share more details about my production/customized Studio setup if that helps.
All reactions