Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Remove system message functionality from code executors #1835

24 changes: 0 additions & 24 deletions autogen/coding/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,30 +42,6 @@ def extract_code_blocks(self, message: Union[str, List[Dict[str, Any]], None]) -
class CodeExecutor(Protocol):
"""(Experimental) A code executor class that executes code blocks and returns the result."""

class UserCapability(Protocol):
"""(Experimental) An AgentCapability class that gives agent ability use this code executor."""

def add_to_agent(self, agent: LLMAgent) -> None:
... # pragma: no cover

@property
def user_capability(self) -> "CodeExecutor.UserCapability":
"""(Experimental) Capability to use this code executor.

The exported capability can be added to an agent to allow it to use this
code executor:

```python
code_executor = CodeExecutor()
agent = ConversableAgent("agent", ...)
code_executor.user_capability.add_to_agent(agent)
```

A typical implementation is to update the system message of the agent with
instructions for how to use this code executor.
"""
... # pragma: no cover

@property
def code_extractor(self) -> CodeExtractor:
"""(Experimental) The code extractor used by this code executor."""
Expand Down
62 changes: 0 additions & 62 deletions autogen/coding/jupyter/embedded_ipython_code_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,67 +35,11 @@ class EmbeddedIPythonCodeExecutor(BaseModel):
kernel_name (str): The kernel name to use. Make sure it is installed.
By default, it is "python3".
output_dir (str): The directory to save output files, by default ".".
system_message_update (str): The system message update to add to the
agent that produces code. By default it is
`EmbeddedIPythonCodeExecutor.DEFAULT_SYSTEM_MESSAGE_UPDATE`.
"""

DEFAULT_SYSTEM_MESSAGE_UPDATE: ClassVar[
str
] = """
# IPython Coding Capability
You have been given coding capability to solve tasks using Python code in a stateful IPython kernel.
You are responsible for writing the code, and the user is responsible for executing the code.

When you write Python code, put the code in a markdown code block with the language set to Python.
For example:
```python
x = 3
```
You can use the variable `x` in subsequent code blocks.
```python
print(x)
```

Write code incrementally and leverage the statefulness of the kernel to avoid repeating code.
Import libraries in a separate code block.
Define a function or a class in a separate code block.
Run code that produces output in a separate code block.
Run code that involves expensive operations like download, upload, and call external APIs in a separate code block.

When your code produces an output, the output will be returned to you.
Because you have limited conversation memory, if your code creates an image,
the output will be a path to the image instead of the image itself.
"""

timeout: int = Field(default=60, ge=1, description="The timeout for code execution.")
kernel_name: str = Field(default="python3", description="The kernel name to use. Make sure it is installed.")
output_dir: str = Field(default=".", description="The directory to save output files.")
system_message_update: str = Field(
default=DEFAULT_SYSTEM_MESSAGE_UPDATE,
description="The system message update to the agent that produces code to be executed by this executor.",
)

class UserCapability:
"""(Experimental) An AgentCapability class that gives agent ability use a stateful
IPython code executor. This capability can be added to an agent using
the `add_to_agent` method which append a system message update to the
agent's system message."""

def __init__(self, system_message_update: str):
self.system_message_update = system_message_update

def add_to_agent(self, agent: LLMAgent) -> None:
"""Add this capability to an agent by appending a system message
update to the agent's system message.

**Currently we do not check for conflicts with existing content in
the agent's system message.**

Args:
agent (LLMAgent): The agent to add the capability to.
"""
agent.update_system_message(agent.system_message + self.system_message_update)

@field_validator("output_dir")
@classmethod
Expand All @@ -121,12 +65,6 @@ def __init__(self, **kwargs: Any):
self._kernel_name = self.kernel_name
self._output_dir = Path(self.output_dir)

@property
def user_capability(self) -> "EmbeddedIPythonCodeExecutor.UserCapability":
"""(Experimental) Export a user capability for this executor that can be added to
an agent using the `add_to_agent` method."""
return EmbeddedIPythonCodeExecutor.UserCapability(self.system_message_update)

@property
def code_extractor(self) -> CodeExtractor:
"""(Experimental) Export a code extractor that can be used by an agent."""
Expand Down
41 changes: 0 additions & 41 deletions autogen/coding/jupyter/jupyter_code_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,48 +22,12 @@


class JupyterCodeExecutor(CodeExecutor):
DEFAULT_SYSTEM_MESSAGE_UPDATE: ClassVar[
str
] = """
# IPython Coding Capability
You have been given coding capability to solve tasks using Python code in a stateful IPython kernel.
You are responsible for writing the code, and the user is responsible for executing the code.

When you write Python code, put the code in a markdown code block with the language set to Python.
For example:
```python
x = 3
```
You can use the variable `x` in subsequent code blocks.
```python
print(x)
```

Write code incrementally and leverage the statefulness of the kernel to avoid repeating code.
Import libraries in a separate code block.
Define a function or a class in a separate code block.
Run code that produces output in a separate code block.
Run code that involves expensive operations like download, upload, and call external APIs in a separate code block.

When your code produces an output, the output will be returned to you.
Because you have limited conversation memory, if your code creates an image,
the output will be a path to the image instead of the image itself.
"""

class UserCapability:
def __init__(self, system_message_update: str):
self._system_message_update = system_message_update

def add_to_agent(self, agent: LLMAgent) -> None:
agent.update_system_message(agent.system_message + self._system_message_update)

def __init__(
self,
jupyter_server: Union[JupyterConnectable, JupyterConnectionInfo],
kernel_name: str = "python3",
timeout: int = 60,
output_dir: Union[Path, str] = Path("."),
system_message_update: str = DEFAULT_SYSTEM_MESSAGE_UPDATE,
):
"""(Experimental) A code executor class that executes code statefully using
a Jupyter server supplied to this class.
Expand Down Expand Up @@ -104,11 +68,6 @@ def __init__(
self._jupyter_kernel_client = self._jupyter_client.get_kernel_client(self._kernel_id)
self._timeout = timeout
self._output_dir = output_dir
self._system_message_update = system_message_update

@property
def user_capability(self) -> "JupyterCodeExecutor.UserCapability":
return JupyterCodeExecutor.UserCapability(self._system_message_update)

@property
def code_extractor(self) -> CodeExtractor:
Expand Down
37 changes: 0 additions & 37 deletions autogen/coding/local_commandline_code_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,23 +27,10 @@ class CommandLineCodeResult(CodeResult):


class LocalCommandLineCodeExecutor(CodeExecutor):
DEFAULT_SYSTEM_MESSAGE_UPDATE: ClassVar[
str
] = """
You have been given coding capability to solve tasks using Python code.
In the following cases, suggest python code (in a python coding block) or shell script (in a sh coding block) for the user to execute.
1. When you need to collect info, use the code to output the info you need, for example, browse or search the web, download/read a file, print the content of a webpage or a file, get the current date/time, check the operating system. After sufficient info is printed and the task is ready to be solved based on your language skill, you can solve the task by yourself.
2. When you need to perform some task with code, use the code to perform the task and output the result. Finish the task smartly.
Solve the task step by step if you need to. If a plan is not provided, explain your plan first. Be clear which step uses code, and which step uses your language skill.
When using code, you must indicate the script type in the code block. The user cannot provide any other feedback or perform any other action beyond executing the code you suggest. The user can't modify your code. So do not suggest incomplete code which requires users to modify. Don't use a code block if it's not intended to be executed by the user.
If you want the user to save the code in a file before executing it, put # filename: <filename> inside the code block as the first line. Don't include multiple code blocks in one response. Do not ask users to copy and paste the result. Instead, use 'print' function for the output when relevant. Check the execution result returned by the user.
"""

def __init__(
self,
timeout: int = 60,
work_dir: Union[Path, str] = Path("."),
system_message_update: str = DEFAULT_SYSTEM_MESSAGE_UPDATE,
):
"""(Experimental) A code executor class that executes code through a local command line
environment.
Expand All @@ -66,10 +53,6 @@ def __init__(
work_dir (str): The working directory for the code execution. If None,
a default working directory will be used. The default working
directory is the current directory ".".
system_message_update (str): The system message update for agent that
produces code to run on this executor.
Default is `LocalCommandLineCodeExecutor.DEFAULT_SYSTEM_MESSAGE_UPDATE`.

"""

if timeout < 1:
Expand All @@ -83,26 +66,6 @@ def __init__(

self._timeout = timeout
self._work_dir: Path = work_dir
self._system_message_update = system_message_update

class UserCapability:
"""An AgentCapability class that gives agent ability use a command line
code executor via a system message update. This capability can be added
to an agent using the `add_to_agent` method."""

def __init__(self, system_message_update: str) -> None:
self.system_message_update = system_message_update

def add_to_agent(self, agent: LLMAgent) -> None:
"""Add this capability to an agent by updating the agent's system
message."""
agent.update_system_message(agent.system_message + self.system_message_update)

@property
def user_capability(self) -> "LocalCommandLineCodeExecutor.UserCapability":
"""Export a user capability for this executor that can be added to
an agent that produces code to be executed by this executor."""
return LocalCommandLineCodeExecutor.UserCapability(self._system_message_update)

@property
def timeout(self) -> int:
Expand Down
46 changes: 0 additions & 46 deletions test/coding/test_commandline_code_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -142,52 +142,6 @@ def _test_restart(executor: CodeExecutor) -> None:
executor.restart()


@pytest.mark.skipif(skip_openai, reason="requested to skip openai tests")
def test_local_commandline_executor_conversable_agent_capability() -> None:
with tempfile.TemporaryDirectory() as temp_dir:
executor = LocalCommandLineCodeExecutor(work_dir=temp_dir)
_test_conversable_agent_capability(executor=executor)


def _test_conversable_agent_capability(executor: CodeExecutor) -> None:
KEY_LOC = "notebook"
OAI_CONFIG_LIST = "OAI_CONFIG_LIST"
config_list = config_list_from_json(
OAI_CONFIG_LIST,
file_location=KEY_LOC,
filter_dict={
"model": {
"gpt-3.5-turbo",
"gpt-35-turbo",
},
},
)
llm_config = {"config_list": config_list}
agent = ConversableAgent(
"coding_agent",
llm_config=llm_config,
code_execution_config=False,
)
executor.user_capability.add_to_agent(agent)

# Test updated system prompt.
assert executor.DEFAULT_SYSTEM_MESSAGE_UPDATE in agent.system_message

# Test code generation.
reply = agent.generate_reply(
[{"role": "user", "content": "write a python script to print 'hello world' to the console"}],
sender=ConversableAgent(name="user", llm_config=False, code_execution_config=False),
)

# Test code extraction.
code_blocks = executor.code_extractor.extract_code_blocks(reply) # type: ignore[arg-type]
assert len(code_blocks) == 1 and code_blocks[0].language == "python"

# Test code execution.
code_result = executor.execute_code_blocks(code_blocks)
assert code_result.exit_code == 0 and "hello world" in code_result.output.lower().replace(",", "")


@pytest.mark.parametrize("cls", classes_to_test)
def test_commandline_executor_conversable_agent_code_execution(cls) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
Expand Down
43 changes: 0 additions & 43 deletions test/coding/test_embedded_ipython_code_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -212,49 +212,6 @@ def test_save_html(cls) -> None:
assert f"HTML data saved to {code_result.output_files[0]}" in code_result.output


@pytest.mark.skipif(skip, reason=skip_reason)
@pytest.mark.skipif(skip_openai, reason="openai not installed OR requested to skip")
@pytest.mark.parametrize("cls", classes_to_test)
def test_conversable_agent_capability(cls) -> None:
KEY_LOC = "notebook"
OAI_CONFIG_LIST = "OAI_CONFIG_LIST"
config_list = config_list_from_json(
OAI_CONFIG_LIST,
file_location=KEY_LOC,
filter_dict={
"model": {
"gpt-3.5-turbo",
"gpt-35-turbo",
},
},
)
llm_config = {"config_list": config_list}
agent = ConversableAgent(
"coding_agent",
llm_config=llm_config,
code_execution_config=False,
)
executor = cls()
executor.user_capability.add_to_agent(agent)

# Test updated system prompt.
assert executor.DEFAULT_SYSTEM_MESSAGE_UPDATE in agent.system_message

# Test code generation.
reply = agent.generate_reply(
[{"role": "user", "content": "print 'hello world' to the console in a single python code block"}],
sender=ConversableAgent("user", llm_config=False, code_execution_config=False),
)

# Test code extraction.
code_blocks = executor.code_extractor.extract_code_blocks(reply) # type: ignore[arg-type]
assert len(code_blocks) == 1 and code_blocks[0].language == "python"

# Test code execution.
code_result = executor.execute_code_blocks(code_blocks)
assert code_result.exit_code == 0 and "hello world" in code_result.output.lower()


@pytest.mark.skipif(skip, reason=skip_reason)
@pytest.mark.parametrize("cls", classes_to_test)
def test_conversable_agent_code_execution(cls) -> None:
Expand Down
Loading