How to stop an agent from terminating?

[karun19]

I am working with Magentic One v0.4.0.dev13. I have a use case to onboard a new employee. I have created 3 agents,

1. Fetch steps to onboard an employee agent
2. Create AD acount agent
3. Create SNOW account agent

Registered all the agent and when I run the flow "Fetch steps to onboard an employee agent" is fetching the isntructions scussfully but terminates after immediately. Instead I expect the flow to continue further and execute the steps (Create AD acount agent, Create SNOW account agent) fetched. I see plan is as intended but it is terminating in the middle.

Below is the code, please suggest how to stop termination and continue further.

example_onboarding_incident_weather_and_coder.py

"""This example demonstrates a human user interacting with a coder agent
which uses a model to generate code snippets. The user and the coder agent
takes turn to write input or generate code snippets, orchestrated by an
round-robin orchestrator agent.
The code snippets are not executed in this example."""

import asyncio
import logging

from autogen_core import EVENT_LOGGER_NAME, AgentId, AgentProxy, SingleThreadedAgentRuntime

# from typing import Any, Dict, List, Tuple, Union
from autogen_magentic_one.agents.weather_agent import WeatherAgent
from autogen_magentic_one.agents.fetch_onboarding_steps import FetchOnboardingSteps
from autogen_magentic_one.agents.create_ad_account import ADAccountRequest
from autogen_magentic_one.agents.create_incident import CreateIncident
# from autogen_magentic_one.agents.orchestrator import RoundRobinOrchestrator
from autogen_magentic_one.agents.orchestrator import LedgerOrchestrator
from autogen_magentic_one.agents.user_proxy import UserProxy
from autogen_magentic_one.messages import RequestReplyMessage
from autogen_magentic_one.utils import LogHandler, create_completion_client_from_env
from autogen_magentic_one.agents.coder import Coder, Executor
from autogen_core.code_executor import CodeBlock

async def confirm_code(code: CodeBlock) -> bool:
    response = await asyncio.to_thread(
        input,
        f"Executor is about to execute code (lang: {code.language}):\n{code.code}\n\nDo you want to proceed? (yes/no): ",
    )
    return response.lower() == "yes"
    
async def save_to_file(content: str, filename: str = "output.html") -> None:
    """Save the given content to an HTML file."""
    try:
        with open(filename, "w", encoding="utf-8") as file:
            file.write(content)
        print(f"Final output saved to {filename}")
    except Exception as e:
        print(f"Failed to save output to {filename}: {e}")
    
async def main() -> None:
    # Create the runtime.
    runtime = SingleThreadedAgentRuntime()

    # Get an appropriate client
    client = create_completion_client_from_env()

    # Register agents.
    await WeatherAgent.register(runtime, "WeatherAgent", lambda: WeatherAgent(model_client=client))
    weather_agent = AgentProxy(AgentId("WeatherAgent", "default"), runtime)
    
    await FetchOnboardingSteps.register(runtime, "FetchOnboardingSteps", lambda: FetchOnboardingSteps(model_client=client))
    fetch_onboarding_steps = AgentProxy(AgentId("FetchOnboardingSteps", "default"), runtime)
    
    await ADAccountRequest.register(runtime, "ADAccountRequest", lambda: ADAccountRequest(model_client=client))
    create_ad_account = AgentProxy(AgentId("ADAccountRequest", "default"), runtime)
    
    await CreateIncident.register(runtime, "CreateIncident", lambda: CreateIncident(model_client=client))
    create_incident = AgentProxy(AgentId("CreateIncident", "default"), runtime)
    
    await Coder.register(runtime, "Coder", lambda: Coder(model_client=client))
    coder = AgentProxy(AgentId("Coder", "default"), runtime)

    await UserProxy.register(
        runtime,
        "UserProxy",
        lambda: UserProxy(description="The current user interacting with you."),
    )
    user_proxy = AgentProxy(AgentId("UserProxy", "default"), runtime)

    # await RoundRobinOrchestrator.register(runtime, "orchestrator", lambda: RoundRobinOrchestrator([weather_agent, coder, user_proxy]))
    agent_list = [coder, weather_agent, fetch_onboarding_steps, create_ad_account, create_incident]
    
    await LedgerOrchestrator.register(
        runtime,
        "Orchestrator",
        lambda: LedgerOrchestrator(
            agents=agent_list,
            model_client=client,
            max_rounds=30,
            max_time=25 * 60,
            return_final_answer=True,
        ),
    )

    runtime.start()

    # Send a message to the UserProxy and await a response
    final_response = await runtime.send_message(RequestReplyMessage(), user_proxy.id)

    # Stop the runtime when idle
    await runtime.stop_when_idle()

    # Save the final response to output.html
    if final_response:
        await save_to_file(str(final_response), "output.html")


if __name__ == "__main__":
    logger = logging.getLogger(EVENT_LOGGER_NAME)
    logger.setLevel(logging.INFO)
    log_handler = LogHandler()
    logger.handlers = [log_handler]
    asyncio.run(main())

fetch_onboarding_steps.py

import re
from typing import List, Literal, Tuple, Union

from autogen_core import CancellationToken, default_subscription
from autogen_core.models import (
    ChatCompletionClient,
    SystemMessage,
    UserMessage,
)

from ..utils import message_content_to_str
from .base_worker import BaseWorker


@default_subscription
class FetchOnboardingSteps(BaseWorker):
    """An agent to provide onboarding steps for new employees and follows the steps to onboard the employee"""

    DEFAULT_DESCRIPTION = "An agent that provides onboarding steps for new employees and follows the steps to onboard the employee"

    DEFAULT_SYSTEM_MESSAGES = [
        SystemMessage(
            content="""You are a helpful agent tasked with providing onboarding steps for new employees.
Ask the user for the employee's details if necessary and provide a list of onboarding steps from a local file.
If the file is missing or cannot be read, inform the user. Follow the steps to onaboard an new employee"""
        )
    ]

    def __init__(
        self,
        model_client: ChatCompletionClient,
        description: str = DEFAULT_DESCRIPTION,
        system_messages: List[SystemMessage] = DEFAULT_SYSTEM_MESSAGES,
        request_terminate: bool = False,
    ) -> None:
        super().__init__(description)
        self._model_client = model_client
        self._system_messages = system_messages
        self._request_terminate = False

    def _fetch_onboarding_steps(self) -> List[str]:
        """Fetch onboarding steps from a local file."""
        file_path = "onboarding_steps.txt"
        try:
            with open(file_path, "r") as file:
                steps = file.readlines()
                return [step.strip() for step in steps if step.strip()]
        except FileNotFoundError:
            return ["Error: onboarding_steps.txt file not found."]
        except Exception as e:
            return [f"Error reading onboarding_steps.txt: {e}"]

    async def _generate_reply(self, cancellation_token: CancellationToken) -> Tuple[bool, str]:
        """Respond to the user's request."""
        onboarding_steps = self._fetch_onboarding_steps()
        steps_message = "\n".join(onboarding_steps)

        for message in self._chat_history:
            if isinstance(message, UserMessage):
                user_message = message.content.strip()
                if user_message:
                    return True, f"Here are the onboarding steps:\n{steps_message}"
        
        return False, f"Unable to process the request.\n\nOnboarding Steps:\n{steps_message}\n\nTERMINATE"

onboarding_steps.txt

Follow below steps to onboard a new employee.

1) Create AD account
2) Create ServiceNow Account
3) Assign a Laptop

create_ad_account.py

import requests
import re
from typing import Awaitable, Callable, List, Literal, Tuple, Union

from autogen_core import CancellationToken, default_subscription
from autogen_core.code_executor import CodeBlock, CodeExecutor
from autogen_core.models import (
    ChatCompletionClient,
    SystemMessage,
    UserMessage,
)

from ..messages import UserContent
from ..utils import message_content_to_str
from .base_worker import BaseWorker


@default_subscription
class ADAccountRequest(BaseWorker):
    """An agent to create an Active Directory (AD) account for a new employee."""

    DEFAULT_DESCRIPTION = "An agent that creates AD accounts for new employees through an external API."

    DEFAULT_SYSTEM_MESSAGES = [
        SystemMessage(
            content="""You are a helpful agent tasked with creating Active Directory accounts.
Ask the user for the new employee's details (e.g., email ID, name, department, and joining date) 
and create an AD account using the provided API.
If you encounter errors, inform the user and provide a clear resolution. End the task with "TERMINATE" when done."""
        )
    ]

    def __init__(
        self,
        model_client: ChatCompletionClient,
        description: str = DEFAULT_DESCRIPTION,
        system_messages: List[SystemMessage] = DEFAULT_SYSTEM_MESSAGES,
        request_terminate: bool = False,
    ) -> None:
        super().__init__(description)
        self._model_client = model_client
        self._system_messages = system_messages
        self._request_terminate = request_terminate

    async def _create_ad_account(self, email: str, name: str, department: str, joining_date: str) -> Tuple[bool, str]:
        """Send a POST request to create an AD account."""
        """api_url = "https://api.example.com/create-ad-account"  # Replace with your API endpoint
        query_params = {
            "api-version": "2023-01-01",  # Replace with the correct API version
        }
        payload = {
            "email": email,
            "name": name,
            "department": department,
            "joining_date": joining_date,
        }"""

        try:
            """response = requests.post(api_url, params=query_params, json=payload)
            response.raise_for_status()
            adAccountResponse = response.json()"""
            return True, f"AD account created successfully"
        except requests.exceptions.RequestException as e:
            return False, f"Failed to create AD account: {e}"
        except ValueError:
            return False, "Error parsing API response. Please check the format."

    async def _generate_reply(self, cancellation_token: CancellationToken) -> Tuple[bool, str]:
        """Respond to the user's request."""
        for message in self._chat_history:
            if isinstance(message, UserMessage):
                user_message = message.content.strip()
                if user_message:
                    #extracted = self._extract_execution_request(message_content_to_str(message.content))
                    if extracted:
                        #email, name, department, joining_date = extracted
                        success, result = await self._create_ad_account("email", "name", "department", "joining_date")
                        if success:
                            return "TERMINATE" in result, result
                        else:
                            return False, result + "\nPlease provide valid employee details or try again."
        return False, "No valid employee details provided. Please enter the required details."

    def _extract_execution_request(self, markdown_text: str) -> Union[Tuple[str, str, str, str], None]:
        """Extract employee details from the user input."""
        email_pattern = r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"
        name_pattern = r"Name:\s*(.+)"
        department_pattern = r"Department:\s*(.+)"
        joining_date_pattern = r"Joining Date:\s*(\d{4}-\d{2}-\d{2})"

        email = re.search(email_pattern, markdown_text)
        name = re.search(name_pattern, markdown_text)
        department = re.search(department_pattern, markdown_text)
        joining_date = re.search(joining_date_pattern, markdown_text)

        if email and name and department and joining_date:
            return (
                email.group(0),
                name.group(1).strip(),
                department.group(1).strip(),
                joining_date.group(1).strip(),
            )
        return None

Execution logs

User input ('exit' to quit):  fetch and follow onboarding steps to on board a new employee [email protected]

---------------------------------------------------------------------------
[2025-01-22T18:25:14.648395], UserProxy:

fetch and follow onboarding steps to on board a new employee [email protected]
C:\Karun\Documents\LUCA\AgenticAI\AG 04\autogen\python\packages\autogen-magentic-one\src\autogen_magentic_one\agents\orchestrator.py:152: UserWarning: Resolved model mismatch: gpt-4o-2024-08-06 != gpt-4o-2024-11-20. Model mapping may be incorrect.
  response = await self._model_client.create(
C:\Karun\Documents\LUCA\AgenticAI\AG 04\autogen\python\packages\autogen-magentic-one\src\autogen_magentic_one\agents\orchestrator.py:165: UserWarning: Resolved model mismatch: gpt-4o-2024-08-06 != gpt-4o-2024-11-20. Model mapping may be incorrect.
  response = await self._model_client.create(

---------------------------------------------------------------------------
[2025-01-22T18:25:25.301828], Orchestrator (thought):

Initial plan:

We are working to address the following user request:

fetch and follow onboarding steps to on board a new employee [email protected]


To answer this request we have assembled the following team:

Coder: A helpful and general-purpose AI assistant that has strong language skills, Python skills, and Linux command line skills.
WeatherAgent: A weather agent that fetches summarized weather details for a location using an external API.
FetchOnboardingSteps: An agent that provides onboarding steps for new employees and follows the steps to onboard the employee
ADAccountRequest: An agent that creates AD accounts for new employees through an external API.
CreateIncident: An agent that creates incident requests through an external API.



Here is an initial fact sheet to consider:

### 1. GIVEN OR VERIFIED FACTS
- There is a request to onboard a new employee with the email address "[email protected]."
- The task involves fetching and following onboarding steps, though no specific process or platform is mentioned.

### 2. FACTS TO LOOK UP
- **Onboarding Process for Digi-WAS**: Specific onboarding steps may need to be retrieved from internal resources (e.g., an employee handbook, HR software, or company intranet).
- **HR or IT Systems**: Any systems (e.g., Workday, BambooHR, etc.) used for onboarding new employees at Digi-WAS.
- **Contact Information or Responsible Parties**: Identifying HR or IT personnel in charge of onboarding at Digi-WAS, in case clarification is needed.
- **Templates or Forms**: Any standard documentation or kits required for the onboarding process.
- **Access Provisioning**: Platforms and systems that the new hire (Mitsuko Robison) needs access to, along with security clearance levels.
- **Company Policies**: Specific policies that apply when onboarding new employees at Digi-WAS.

### 3. FACTS TO DERIVE
- Whether the onboarding process involves specific account setups, training schedules, or completion deadlines for mandatory tasks.
- Which departments may need to be alerted about Mitsuko Robison's arrival (e.g., facilities for workspace setup, IT for hardware and software provisioning).
- If training or orientation sessions are typically group-based or individual, and the appropriate scheduling process for these.
- Given the domain of the email ("digi-was.com"), the organization might focus on digital or tech-related operations, implying that she may require software and tools relevant to her role.

### 4. EDUCATED GUESSES
- The name Mitsuko Robison and the format of the email suggest she is a new employee, but her specific role and department are not provided. Most onboarding processes are role-specific, so this information might be essential.
- If Digi-WAS is a mid-sized or large organization, it is likely using an HR platform (e.g., Workday, SAP, or ADP) for onboarding automation.
- Essential onboarding steps may involve account provisioning, signing HR documents like NDAs or contracts, employee orientation, IT equipment issuance, and initial training.


Here is the plan to follow as best as possible:

### Plan to Address the Request
1. **Fetch Onboarding Steps**
   - Utilize the **FetchOnboardingSteps** agent to retrieve the detailed onboarding steps for a new employee at Digi-WAS.
   - Confirm that the steps are appropriate to onboard "[email protected]" and aligned with Digi-WAS's policies and procedures.

2. **Follow Onboarding Steps**
   - Follow the retrieved steps systematically, ensuring all key tasks (e.g., account creation, hardware provisioning, policy acknowledgment) are completed.
   - Identify specific tools, systems, or departments required for onboarding (e.g., IT, HR systems).

3. **AD Account Creation**
   - If one of the onboarding steps involves creating an Active Directory (AD) account, task the **ADAccountRequest** agent with creating an account for "[email protected]."
   - Record and securely provision login credentials to the appropriate personnel or systems.

4. **Incident or Ticket Creation**
   - If incident requests (e.g., for hardware or technical system setup) are required as part of the onboarding process, use the **CreateIncident** agent to log necessary requests.
   - Ensure proper tracking and follow-up of these incidents, if applicable.

5. **Handle Any Dependencies**
   - If additional context (e.g., role specifics, departmental tools) or cross-functional support is needed, escalate appropriately to internal HR or IT teams as outlined in onboarding procedures.

6. **Verify and Finalize**
   - Confirm completion of all onboarding steps as outlined by **FetchOnboardingSteps.**
   - Verify that Mitsuko Robison has the necessary credentials, equipment, and access to begin her role successfully.

C:\Karun\Documents\LUCA\AgenticAI\AG 04\autogen\python\packages\autogen-magentic-one\src\autogen_magentic_one\agents\orchestrator.py:216: UserWarning: Resolved model mismatch: gpt-4o-2024-08-06 != gpt-4o-2024-11-20. Model mapping may be incorrect.
  ledger_response = await self._model_client.create(

---------------------------------------------------------------------------
[2025-01-22T18:25:29.299468], Orchestrator (thought):

Updated Ledger:
{
  "is_request_satisfied": {
    "reason": "The steps to fetch and follow the onboarding process for Mitsuko Robison have not yet been fully executed or verified.",
    "answer": false
  },
  "is_in_loop": {
    "reason": "The team has yet to repeat previous actions or exhibit looping behavior.",
    "answer": false
  },
  "is_progress_being_made": {
    "reason": "We are in the initial stages of addressing the onboarding request and have outlined a clear plan for moving forward.",
    "answer": true
  },
  "next_speaker": {
    "reason": "The next logical step is to fetch the detailed onboarding steps for Digi-WAS to establish a roadmap for onboarding Mitsuko Robison.",
    "answer": "FetchOnboardingSteps"
  },
  "instruction_or_question": {
    "reason": "Onboarding steps specific to Digi-WAS are required to initiate the employee onboarding process.",
    "answer": "Please fetch the detailed onboarding steps for a new employee at Digi-WAS so we can proceed with onboarding [email protected]."
  }
}

---------------------------------------------------------------------------
[2025-01-22T18:25:29.301016], Orchestrator (-> FetchOnboardingSteps):

Please fetch the detailed onboarding steps for a new employee at Digi-WAS so we can proceed with onboarding [email protected].

---------------------------------------------------------------------------
[2025-01-22T18:25:29.302035], Orchestrator (thought):

Next speaker FetchOnboardingSteps

---------------------------------------------------------------------------
[2025-01-22T18:25:29.317173], FetchOnboardingSteps:

Here are the onboarding steps:
Follow below steps to onboard a new employee.
1) Create AD account
2) Create ServiceNow Account
3) Assign a Laptop

---------------------------------------------------------------------------
[2025-01-22T18:25:29.317173], Orchestrator (termination condition):

FetchOnboardingSteps requested halt.
Press any key to continue . . .

[ekzhu]

Fixed the format for you:

Please use

```python
```

To format code blocks, and four ` to format output blocks.

Thanks.

[ekzhu]

When you request halt you return True, otherwise return False. Based on your code it seems like you got it backward.

Besides, if your workflow has deterministic steps: the agents are invoked in a deterministic order, you should not be using magentic one. Check out sequential workflows or some deterministic workflow patterns

[karun19]

@ekzhu can you please name the a pattern or provide the link.

[ekzhu]

https://microsoft.github.io/autogen/stable/user-guide/core-user-guide/design-patterns/sequential-workflow.html