enhance: Workflow better filenames and custom session id (#3285)

Co-authored-by: Wendong-Fan <133094783+Wendong-Fan@users.noreply.github.com>

Author: hesamsheikh

camel/agents/chat_agent.py

@@ -1247,11 +1247,29 @@ def summarize(
                 result["status"] = status_message
                 return result
 
-            base_filename = (
-                filename
-                if filename
-                else f"context_summary_{datetime.now().strftime('%Y%m%d_%H%M%S')}"  # noqa: E501
-            )
+            # handle structured output if response_format was provided
+            structured_output = None
+            if response_format and response.msgs[-1].parsed:
+                structured_output = response.msgs[-1].parsed
+
+            # determine filename: use provided filename, or extract from
+            # structured output, or generate timestamp
+            if filename:
+                base_filename = filename
+            elif structured_output and hasattr(
+                structured_output, 'task_title'
+            ):
+                # use task_title from structured output for filename
+                task_title = structured_output.task_title
+                clean_title = ContextUtility.sanitize_workflow_filename(
+                    task_title
+                )
+                base_filename = (
+                    f"{clean_title}_workflow" if clean_title else "workflow"
+                )
+            else:
+                base_filename = f"context_summary_{datetime.now().strftime('%Y%m%d_%H%M%S')}"  # noqa: E501
+
             base_filename = Path(base_filename).with_suffix("").name
 
             metadata = context_util.get_session_metadata()
@@ -1262,11 +1280,9 @@ def summarize(
                 }
             )
 
-            # Handle structured output if response_format was provided
-            structured_output = None
-            if response_format and response.msgs[-1].parsed:
-                structured_output = response.msgs[-1].parsed
-                # Convert structured output to custom markdown
+            # convert structured output to custom markdown if present
+            if structured_output:
+                # convert structured output to custom markdown
                 summary_content = context_util.structured_output_to_markdown(
                     structured_data=structured_output, metadata=metadata
                 )

camel/societies/workforce/single_agent_worker.py

@@ -80,6 +80,7 @@ def __init__(
         self._in_use_agents: set = set()
         self._agent_last_used: dict = {}
         self._lock = asyncio.Lock()
+        self._condition = asyncio.Condition(self._lock)
 
         # Statistics
         self._total_borrows = 0
@@ -105,36 +106,31 @@ def _create_fresh_agent(self) -> ChatAgent:
 
     async def get_agent(self) -> ChatAgent:
         r"""Get an agent from the pool, creating one if necessary."""
-        async with self._lock:
+        async with self._condition:
             self._total_borrows += 1
 
-            if self._available_agents:
-                agent = self._available_agents.popleft()
-                self._in_use_agents.add(id(agent))
-                self._pool_hits += 1
-                return agent
-
-            # Check if we can create a new agent
-            if len(self._in_use_agents) < self.max_size or self.auto_scale:
-                agent = self._create_fresh_agent()
-                self._in_use_agents.add(id(agent))
-                return agent
-
-        # Wait for available agent
-        while True:
-            async with self._lock:
+            # Try to get available agent or create new one
+            while True:
                 if self._available_agents:
                     agent = self._available_agents.popleft()
                     self._in_use_agents.add(id(agent))
                     self._pool_hits += 1
                     return agent
-            await asyncio.sleep(0.05)
+
+                # Check if we can create a new agent
+                if len(self._in_use_agents) < self.max_size or self.auto_scale:
+                    agent = self._create_fresh_agent()
+                    self._in_use_agents.add(id(agent))
+                    return agent
+
+                # Wait for an agent to be returned
+                await self._condition.wait()
 
     async def return_agent(self, agent: ChatAgent) -> None:
         r"""Return an agent to the pool."""
         agent_id = id(agent)
 
-        async with self._lock:
+        async with self._condition:
             if agent_id not in self._in_use_agents:
                 return
 
@@ -145,6 +141,8 @@ async def return_agent(self, agent: ChatAgent) -> None:
                 agent.reset()
                 self._agent_last_used[agent_id] = time.time()
                 self._available_agents.append(agent)
+                # Notify one waiting coroutine that an agent is available
+                self._condition.notify()
             else:
                 # Remove tracking for agents not returned to pool
                 self._agent_last_used.pop(agent_id, None)
@@ -154,7 +152,7 @@ async def cleanup_idle_agents(self) -> None:
         if not self.auto_scale:
             return
 
-        async with self._lock:
+        async with self._condition:
             if not self._available_agents:
                 return
 
@@ -428,6 +426,7 @@ async def _process_task(
                     "usage"
                 ) or final_response.info.get("token_usage")
             else:
+                final_response = response
                 usage_info = response.info.get("usage") or response.info.get(
                     "token_usage"
                 )
@@ -562,10 +561,11 @@ async def _periodic_cleanup(self):
         while True:
             try:
                 # Fixed interval cleanup
-                await asyncio.sleep(self.agent_pool.cleanup_interval)
-
                 if self.agent_pool:
+                    await asyncio.sleep(self.agent_pool.cleanup_interval)
                     await self.agent_pool.cleanup_idle_agents()
+                else:
+                    break
             except asyncio.CancelledError:
                 break
             except Exception as e:
@@ -583,7 +583,8 @@ def save_workflow_memories(self) -> Dict[str, Any]:
 
         This method generates a workflow summary from the worker agent's
         conversation history and saves it to a markdown file. The filename
-        is based on the worker's description for easy loading later.
+        is based on either the worker's explicit role_name or the generated
+        task_title from the summary.
 
         Returns:
             Dict[str, Any]: Result dictionary with keys:
@@ -603,13 +604,31 @@ def save_workflow_memories(self) -> Dict[str, Any]:
             self.worker.set_context_utility(context_util)
 
             # prepare workflow summarization components
-            filename = self._generate_workflow_filename()
             structured_prompt = self._prepare_workflow_prompt()
             agent_to_summarize = self._select_agent_for_summarization(
                 context_util
             )
 
+            # check if we should use role_name or let summarize extract
+            # task_title
+            role_name = getattr(self.worker, 'role_name', 'assistant')
+            use_role_name_for_filename = role_name.lower() not in {
+                'assistant',
+                'agent',
+                'user',
+                'system',
+            }
+
             # generate and save workflow summary
+            # if role_name is explicit, use it for filename
+            # if role_name is generic, pass none to let summarize use
+            # task_title
+            filename = (
+                self._generate_workflow_filename()
+                if use_role_name_for_filename
+                else None
+            )
+
             result = agent_to_summarize.summarize(
                 filename=filename,
                 summary_prompt=structured_prompt,
@@ -716,12 +735,23 @@ def _find_workflow_files(
             )
             return []
 
-        # generate filename-safe search pattern from worker description
+        # generate filename-safe search pattern from worker role name
         if pattern is None:
-            # sanitize description: spaces to underscores, remove special chars
-            clean_desc = self.description.lower().replace(" ", "_")
-            clean_desc = re.sub(r'[^a-z0-9_]', '', clean_desc)
-            pattern = f"{clean_desc}_workflow*.md"
+            from camel.utils.context_utils import ContextUtility
+
+            # get role_name (always available, defaults to "assistant")
+            role_name = getattr(self.worker, 'role_name', 'assistant')
+            clean_name = ContextUtility.sanitize_workflow_filename(role_name)
+
+            # check if role_name is generic
+            generic_names = {'assistant', 'agent', 'user', 'system'}
+            if clean_name in generic_names:
+                # for generic role names, search for all workflow files
+                # since filename is based on task_title
+                pattern = "*_workflow*.md"
+            else:
+                # for explicit role names, search for role-specific files
+                pattern = f"{clean_name}_workflow*.md"
 
         # Get the base workforce_workflows directory
         camel_workdir = os.environ.get("CAMEL_WORKDIR")
@@ -816,15 +846,21 @@ def _validate_workflow_save_requirements(self) -> Optional[Dict[str, Any]]:
         return None
 
     def _generate_workflow_filename(self) -> str:
-        r"""Generate a filename for the workflow based on worker description.
+        r"""Generate a filename for the workflow based on worker role name.
+
+        Uses the worker's explicit role_name when available.
 
         Returns:
-            str: Sanitized filename without timestamp (session already has
-                timestamp).
+            str: Sanitized filename without timestamp and without .md
+                extension. Format: {role_name}_workflow
         """
-        clean_desc = self.description.lower().replace(" ", "_")
-        clean_desc = re.sub(r'[^a-z0-9_]', '', clean_desc)
-        return f"{clean_desc}_workflow"
+        from camel.utils.context_utils import ContextUtility
+
+        # get role_name (always available, defaults to "assistant"/"Assistant")
+        role_name = getattr(self.worker, 'role_name', 'assistant')
+        clean_name = ContextUtility.sanitize_workflow_filename(role_name)
+
+        return f"{clean_name}_workflow"
 
     def _prepare_workflow_prompt(self) -> str:
         r"""Prepare the structured prompt for workflow summarization.

camel/societies/workforce/workforce.py

@@ -463,20 +463,27 @@ def __init__(
         # Helper for propagating pause control to externally supplied agents
         # ------------------------------------------------------------------
 
-    def _get_or_create_shared_context_utility(self) -> "ContextUtility":
+    def _get_or_create_shared_context_utility(
+        self,
+        session_id: Optional[str] = None,
+    ) -> "ContextUtility":
         r"""Get or create the shared context utility for workflow management.
 
         This method creates the context utility only when needed, avoiding
         unnecessary session folder creation during initialization.
 
+        Args:
+            session_id (Optional[str]): Custom session ID to use. If None,
+                auto-generates a timestamped session ID. (default: :obj:`None`)
+
         Returns:
             ContextUtility: The shared context utility instance.
         """
         if self._shared_context_utility is None:
             from camel.utils.context_utils import ContextUtility
 
-            self._shared_context_utility = (
-                ContextUtility.get_workforce_shared()
+            self._shared_context_utility = ContextUtility.get_workforce_shared(
+                session_id=session_id
             )
         return self._shared_context_utility
 
@@ -2132,7 +2139,10 @@ def reset(self) -> None:
         else:
             self.metrics_logger = WorkforceLogger(workforce_id=self.node_id)
 
-    def save_workflow_memories(self) -> Dict[str, str]:
+    def save_workflow_memories(
+        self,
+        session_id: Optional[str] = None,
+    ) -> Dict[str, str]:
         r"""Save workflow memories for all SingleAgentWorker instances in the
         workforce.
 
@@ -2142,6 +2152,12 @@ def save_workflow_memories(self) -> Dict[str, str]:
         method.
         Other worker types are skipped.
 
+        Args:
+            session_id (Optional[str]): Custom session ID to use for saving
+                workflows. If None, auto-generates a timestamped session ID.
+                Useful for organizing workflows by project or context.
+                (default: :obj:`None`)
+
         Returns:
             Dict[str, str]: Dictionary mapping worker node IDs to save results.
                 Values are either file paths (success) or error messages
@@ -2150,15 +2166,22 @@ def save_workflow_memories(self) -> Dict[str, str]:
         Example:
             >>> workforce = Workforce("My Team")
             >>> # ... add workers and process tasks ...
-            >>> results = workforce.save_workflows()
+            >>> # save with auto-generated session id
+            >>> results = workforce.save_workflow_memories()
             >>> print(results)
-            {'worker_123': '/path/to/data_analyst_workflow_20250122.md',
+            {'worker_123': '/path/to/developer_agent_workflow.md',
              'worker_456': 'error: No conversation context available'}
+            >>> # save with custom project id
+            >>> results = workforce.save_workflow_memories(
+            ...     session_id="project_123"
+            ... )
         """
         results = {}
 
         # Get or create shared context utility for this save operation
-        shared_context_utility = self._get_or_create_shared_context_utility()
+        shared_context_utility = self._get_or_create_shared_context_utility(
+            session_id=session_id
+        )
 
         for child in self._children:
             if isinstance(child, SingleAgentWorker):