Refactoring EmailSearchWorkflow and EmailSearchAgent to adapt to the latest version of Agenscope

Author: chenyushuo

docs/sphinx_doc/source/tutorial/example_search_email.md

@@ -1,7 +1,7 @@
 # Email Search Workflow
 
 
-This example shows a multi-turn email search workflow, inspired by [ART](https://openpipe.ai/blog/art-e-mail-agent?refresh=1756431423904). We implement a ReAct Agent and define tools for email search. Note that this example rewquires installing `AgentScope==0.1.6`.
+This example shows a multi-turn email search workflow, inspired by [ART](https://openpipe.ai/blog/art-e-mail-agent?refresh=1756431423904). We implement a ReAct Agent and define tools for email search. Note that this example rewquires installing [AgentScope](https://github.com/agentscope-ai/agentscope).
 
 ## Core Components
 

docs/sphinx_doc/source_zh/tutorial/example_search_email.md

@@ -1,6 +1,6 @@
 # 邮件搜索例子
 
-这个示例展示了一个多轮邮件搜索工作流，内容参考自 [ART](https://openpipe.ai/blog/art-e-mail-agent?refresh=1756431423904)。我们实现了一个 ReAct Agent，并定义了用于邮件搜索的工具。注意：此示例需要安装 `AgentScope==0.1.6`。
+这个示例展示了一个多轮邮件搜索工作流，内容参考自 [ART](https://openpipe.ai/blog/art-e-mail-agent?refresh=1756431423904)。我们实现了一个 ReAct Agent，并定义了用于邮件搜索的工具。注意：此示例需要安装 [AgentScope](https://github.com/agentscope-ai/agentscope)。
 
 ## 核心组件
 

trinity/common/workflows/agentscope/react/react_agent.py

@@ -5,6 +5,7 @@
 from agentscope.formatter import OpenAIChatFormatter
 from agentscope.message import Msg
 from agentscope.model import OpenAIChatModel
+from agentscope.tool import Toolkit
 from pydantic import BaseModel
 
 
@@ -16,6 +17,7 @@ def __init__(
         system_prompt: str,
         generate_kwargs: dict,
         response_structure: Type[BaseModel],
+        toolkit: Toolkit | None = None,
     ):
         """Initialize the AgentScope ReAct agent with specified tools and model.
 
@@ -41,6 +43,7 @@ def __init__(
             formatter=OpenAIChatFormatter(),
             # we enable agentscope's meta tool to allow agent to call tools dynamically without pre-registration
             enable_meta_tool=True,
+            toolkit=toolkit,
         )
         self.response_structure = response_structure
 

trinity/common/workflows/envs/email_searcher/react_agent.py

@@ -1,41 +1,39 @@
+import json
 from dataclasses import asdict
 from datetime import datetime, timedelta
 from typing import Any
 
+from agentscope.message import TextBlock
+from agentscope.tool import Toolkit, ToolResponse
+
+from trinity.common.workflows.agentscope.react.react_agent import AgentScopeReActAgent
 from trinity.common.workflows.envs.email_searcher.utils import (
     read_email_tool,
     search_emails_tool,
 )
 
-BaseAgentClass = object
-try:
-    from agentscope.agents import ReActAgentV2
-
-    BaseAgentClass = ReActAgentV2  # type: ignore[misc]
-except ImportError as e:
-    error_message = f"AgentScope is not installed. Please install the agentscope framework first before running the workflow. Error: {str(e)}"
-    pass
-
 
-class EmailSearchAgent(BaseAgentClass):
+class EmailSearchAgent(AgentScopeReActAgent):
     """
     A customized ReAct agent with pre-defined tools for email search and reading.
     Ref: https://github.com/OpenPipe/ART/blob/main/dev/art-e/art_e/rollout.py#L260
     """
 
     def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-
-        self.service_toolkit.add(self.search_emails)
-        self.service_toolkit.add(self.read_email)
+        self.message_id_list = []
+        self.ever_read_message_ids = []
+        toolkit = Toolkit()
+        toolkit.register_tool_function(self.search_emails)
+        toolkit.register_tool_function(self.read_email)
+        super().__init__(*args, toolkit=toolkit, **kwargs)
 
     def search_emails(
         self,
         inbox_address: str,
         query_date: str,
         keywords: list[str],
         **kwargs: Any,
-    ):
+    ) -> ToolResponse:
         """
         Search the user's email inbox for emails that match the given keywords.
 
@@ -49,7 +47,6 @@ def search_emails(
                 The status field indicates whether the tool call was successful.
                 The content field contains a list of SearchResult objects with message_id and snippet. If no emails are found, it returns an empty list.
         """
-        from agentscope.service import ServiceExecStatus, ServiceResponse
 
         try:
             next_day = (datetime.strptime(query_date, "%Y-%m-%d") + timedelta(days=1)).strftime(
@@ -59,14 +56,26 @@ def search_emails(
 
             self.message_id_list.extend([r.message_id for r in res])
 
-            return ServiceResponse(
-                status=ServiceExecStatus.SUCCESS,
-                content=[asdict(r) for r in res],
+            return ToolResponse(
+                content=[
+                    TextBlock(
+                        type="text",
+                        text=json.dumps([asdict(r) for r in res]),
+                    ),
+                ],
             )
-        except Exception:
-            return ServiceResponse(
-                status=ServiceExecStatus.ERROR,
-                content=[],
+        except Exception as e:
+            print(f"Error in tool: {e}")
+            import traceback
+
+            traceback.print_exc()
+            return ToolResponse(
+                content=[
+                    TextBlock(
+                        type="text",
+                        text="Error: Failed to search emails.",
+                    ),
+                ],
             )
 
     def read_email(self, message_id: str, **kwargs: Any):
@@ -80,25 +89,36 @@ def read_email(self, message_id: str, **kwargs: Any):
                 The status field indicates whether the tool call was successful.
                 The content field contains the email content or an error message if the email is not found.
         """
-        from agentscope.service import ServiceExecStatus, ServiceResponse
 
         try:
             email_content = read_email_tool(message_id)
 
             self.ever_read_message_ids.append(message_id)
 
             if email_content is None:
-                return ServiceResponse(
-                    status=ServiceExecStatus.ERROR,
-                    content={"error": "Email not found"},
+                return ToolResponse(
+                    content=[
+                        TextBlock(
+                            type="text",
+                            text="Error: Email not found.",
+                        ),
+                    ],
                 )
             else:
-                return ServiceResponse(
-                    status=ServiceExecStatus.SUCCESS,
-                    content=email_content.model_dump(),
+                return ToolResponse(
+                    content=[
+                        TextBlock(
+                            type="text",
+                            text=json.dumps(email_content.model_dump()),
+                        ),
+                    ],
                 )
         except Exception:
-            return ServiceResponse(
-                status=ServiceExecStatus.ERROR,
-                content={"error": "Timeout"},
+            return ToolResponse(
+                content=[
+                    TextBlock(
+                        type="text",
+                        text="Error: Timeout to read email.",
+                    ),
+                ],
             )

trinity/common/workflows/envs/email_searcher/workflow.py

@@ -5,7 +5,6 @@
 import openai
 
 from trinity.common.models.model import ModelWrapper
-from trinity.common.workflows.envs.email_searcher.react_agent import EmailSearchAgent
 from trinity.common.workflows.envs.email_searcher.utils import (
     AnswerModel,
     FinalRubric,
@@ -35,44 +34,15 @@ def __init__(
         model: ModelWrapper,
         auxiliary_models: Optional[List[openai.OpenAI]] = None,
     ):
-        try:
-            import agentscope
-            from agentscope.service import ServiceToolkit
-        except ImportError as e:
-            error_message = f"AgentScope is not installed. Please install the agentscope framework first before running the workflow. Error: {str(e)}"
-            self.logger.error(error_message)
-            raise ImportError(error_message)
-
         # get openai client from model
-        self.openai_client = model.get_openai_client()
+        self.openai_client = model.get_openai_async_client()
         self.model_name = self.openai_client.model_path
         super().__init__(
             task=task,
             model=model,
             auxiliary_models=auxiliary_models,
         )
 
-        temperature = self.rollout_args.get("temperature", 1.0)
-        max_tokens = 4096
-
-        agentscope.init(
-            model_configs=[
-                {
-                    "model_type": "openai_chat",
-                    "config_name": "react_model",
-                    "model_name": self.model_name,
-                    "api_key": "EMPTY",
-                    "generate_args": {
-                        "temperature": temperature,
-                        "max_tokens": max_tokens,
-                    },
-                    "use_openai_formatter": True,
-                }
-            ],
-            disable_saving=True,
-        )
-
-        self.toolkit = ServiceToolkit()
         self.reset(task)
 
     @property
@@ -83,6 +53,10 @@ def repeatable(self) -> bool:
     def resettable(self):
         return True
 
+    @property
+    def asynchronous(self):
+        return True
+
     def reset(self, task: Task):
         self.query = QueryModel.model_validate(task.raw_task)
         self.task_desc = task.task_desc  # question
@@ -99,39 +73,34 @@ def reset(self, task: Task):
             query_date=self.query.query_date,
         )
 
-        self.agent = EmailSearchAgent(
-            name="react_agent",
-            sys_prompt=self.system_prompt,
-            model_config_name="react_model",
-            service_toolkit=self.toolkit,
-            max_iters=self.max_turns,
-            verbose=False,
-        )
-        # we set the openai client to the agent's model
-        self.agent.model.client = self.openai_client
-        self.agent.message_id_list = []
-        self.agent.ever_read_message_ids = []
-
-    def run(self):
-        # make sure that we have the correct import
         try:
-            from agentscope.message import Msg
+            from trinity.common.workflows.envs.email_searcher.react_agent import (
+                EmailSearchAgent,
+            )
         except ImportError as e:
             error_message = f"AgentScope is not installed. Please install the agentscope framework first before running the workflow. Error: {str(e)}"
             self.logger.error(error_message)
             raise ImportError(error_message)
 
-        # provide the task to the react agent
-        msg = Msg("user", self.task_desc, role="user")
+        self.agent = EmailSearchAgent(
+            model_name=self.openai_client.model_path,
+            openai_client=self.openai_client,
+            system_prompt=self.system_prompt,
+            generate_kwargs={
+                "temperature": self.rollout_args.get("temperature", 1.0),
+                "max_tokens": self.rollout_args.get("max_tokens", 4096),
+            },
+            response_structure=AnswerModel,
+        )
 
-        response = self.agent.reply(
-            msg,
-            structured_model=AnswerModel,
+    async def run_async(self):
+        response = await self.agent.reply(
+            self.task_desc,
         )
-        if response.metadata is None:
-            answer_and_sources = {"answer": response.content, "sources": []}
+        if response is None:
+            answer_and_sources = {"answer": "", "sources": []}
         else:
-            answer_and_sources = response.metadata
+            answer_and_sources = response
 
         experiences = self.model.extract_experience_from_history(clear_history=True)
         self.actual_turns = len(