Kubernetes Pod Commandline Code Executor
The PodCommandLineCodeExecutor in the autogen.coding.kubernetes
module is designed to execute code blocks using a pod in Kubernetes. It
functions similarly to the DockerCommandLineCodeExecutor, but
specifically creates container within Kubernetes environments.
There are two condition to use PodCommandLineCodeExecutor.
- Access to a Kubernetes cluster
- installation
autogenwith the extra requirements'pyautogen[kubernetes]'
For local development and testing, this document uses a Minikube cluster.
Minikube is a tool that allows you to run a single-node Kubernetes cluster on you local machine. You can refer to the link below for installation and setup of Minikube.
🔗 https://minikube.sigs.k8s.io/docs/start/
Access kubernetes cluster​
There are four options PodCommandLineCodeExecutor to access kubernetes API server.
- default kubeconfig file path:
~/.kube/config - Provide a custom kubeconfig file path using the
kube_config_fileargument ofPodCommandLineCodeExecutor. - Set the kubeconfig file path using the
KUBECONFIGenvironment variable. - Provide token from Kubernetes ServiceAccount with sufficient permissions
Generally, if kubeconfig file is located in ~/.kube/config, there’s no
need to provide kubeconfig file path on parameter or environment
variables.
The tutorial of providing ServiceAccount Token is in the last section
Example​
In order to use kubernetes Pod based code executor, you need to install Kubernetes Python SDK.
You can do this by running the following command:
pip install 'kubernetes>=27'
Alternatively, you can install it with the extra features for Kubernetes:
pip install 'autogen-agentchat[kubernetes]~=0.2'
To provide kubeconfig file path with environment variable, It can be
added with os.environ["KUBECONFIG"]
import os
# Set the KUBECONFIG environment variable
# if the kubeconfig file is not in the default location(~/.kube/config).
os.environ["KUBECONFIG"] = "path/to/your/kubeconfig"
from autogen.coding import CodeBlock
from autogen.coding.kubernetes import PodCommandLineCodeExecutor
with PodCommandLineCodeExecutor(
namespace="default",
# kube_config_file="kubeconfig/file/path" # If you have another kubeconfig file, you can add it on kube_config_file argument
) as executor:
print(
executor.execute_code_blocks(
# Example of executing a simple Python code block within a Kubernetes pod.
code_blocks=[
CodeBlock(language="python", code="print('Hello, World!')"),
]
)
)
exit_code=0 output='Hello, World!\n' code_file='/workspace/tmp_code_07da107bb575cc4e02b0e1d6d99cc204.py'
Using a context manager(the with statement), the pod created by
PodCommandLineCodeExecutor is automatically deleted after the tasks
are completed.
Although the pod is automatically deleted when using a context manager,
you might sometimes need to delete it manually. You can do this using
stop() method, as shown below:
executor = PodCommandLineCodeExecutor(namespace="default")
%%bash
# This command lists all pods in the default namespace.
# The default pod name follows the format autogen-code-exec-{uuid.uuid4()}.
kubectl get pod -n default
NAME READY STATUS RESTARTS AGE
autogen-code-exec-afd217ac-f77b-4ede-8c53-1297eca5ec64 1/1 Running 0 10m
%%bash
# This command shows container's image in the pod.
# The default container image is python:3-slim
kubectl get pod autogen-code-exec-afd217ac-f77b-4ede-8c53-1297eca5ec64 -o jsonpath={.spec.containers[0].image}
python:3-slim
executor.stop()
To use a different container image for code executor pod, specify the
desired image tag using image argument.
PodCommandLineCodeExecutor has a default execution policy that allows
Python and shell script code blocks. You can enable other languages with
execution_policies argument.
with PodCommandLineCodeExecutor(
image="node:22-alpine", # Specifies the runtime environments using a container image
namespace="default",
work_dir="./app", # Directory within the container where code block files are stored
timeout=10, # Timeout in seconds for pod creation and code block execution (default is 60 seconds)
execution_policies={
"javascript": True
}, # Enable execution of Javascript code blocks by updating execution policies
) as executor:
print(
executor.execute_code_blocks(
code_blocks=[
CodeBlock(language="javascript", code="console.log('Hello, World!')"),
]
)
)
exit_code=0 output='Hello, World!\n' code_file='app/tmp_code_8c34c8586cb47943728afe1297b7a51c.js'
If you want to apply custom settings for executor pod, such as
annotations, environment variables, commands, volumes etc., you can
provide a custom pod specification using kubernetes.client.V1Pod
format.
The container_name argument should also be provided because
PodCommandLineCodeExecutor does not automatically recognize the
container where code blocks will be executed.
from kubernetes import client
pod = client.V1Pod(
metadata=client.V1ObjectMeta(name="abcd", namespace="default", annotations={"sidecar.istio.io/inject": "false"}),
spec=client.V1PodSpec(
restart_policy="Never",
containers=[
client.V1Container(
args=["-c", "while true;do sleep 5; done"],
command=["/bin/sh"],
name="abcd", # container name where code blocks will be executed should be provided using `container_name` argument
image="python:3.11-slim",
env=[
client.V1EnvVar(name="TEST", value="TEST"),
client.V1EnvVar(
name="POD_NAME",
value_from=client.V1EnvVarSource(
field_ref=client.V1ObjectFieldSelector(field_path="metadata.name")
),
),
],
)
],
),
)
with PodCommandLineCodeExecutor(
pod_spec=pod, # custom executor pod spec
container_name="abcd", # To use custom executor pod spec, container_name where code block will be executed should be specified
work_dir="/autogen",
timeout=60,
) as executor:
print(
executor.execute_code_blocks(
code_blocks=[
CodeBlock(language="python", code="print('Hello, World!')"),
]
)
)
print(
executor.execute_code_blocks(
code_blocks=[
CodeBlock(
code="echo $TEST $POD_NAME", language="bash"
), # echo environment variables specified in pod_spec
]
)
)
exit_code=0 output='Hello, World!\n' code_file='/autogen/tmp_code_07da107bb575cc4e02b0e1d6d99cc204.py'
exit_code=0 output='TEST abcd\n' code_file='/autogen/tmp_code_202399627ea7fb8d8e816f4910b7f87b.sh'
Integrates with AutoGen Agents​
PodCommandLineCodeExecutor can be integrated with Agents.
from autogen import config_list_from_json
config_list = config_list_from_json(
env_or_file="OAI_CONFIG_LIST",
)
from autogen import ConversableAgent
# The code writer agent's system message is to instruct the LLM on how to
# use the code executor with python or shell script code
code_writer_system_message = """
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.
"""
with PodCommandLineCodeExecutor(namespace="default") as executor:
code_executor_agent = ConversableAgent(
name="code_executor_agent",
llm_config=False,
code_execution_config={
"executor": executor,
},
human_input_mode="NEVER",
)
code_writer_agent = ConversableAgent(
"code_writer",
system_message=code_writer_system_message,
llm_config={"config_list": config_list},
code_execution_config=False, # Turn off code execution for this agent.
max_consecutive_auto_reply=2,
human_input_mode="NEVER",
)
chat_result = code_executor_agent.initiate_chat(
code_writer_agent, message="Write Python code to calculate the moves of disk on tower of hanoi with 10 disks"
)
code_executor_agent (to code_writer):
Write Python code to calculate the moves of disk on tower of hanoi with 3 disks
--------------------------------------------------------------------------------
code_writer (to code_executor_agent):
The problem of the Tower of Hanoi with 3 disks involves moving the disks from one peg to another, following these rules:
1. Only one disk can be moved at a time.
2. Each move consists of taking the upper disk from one of the stacks and placing it on top of another stack or on an empty peg.
3. No disk may be placed on top of a smaller disk.
In the solution, I will use a recursive function to calculate the moves and print them out. Here's the Python code to accomplish this:
```python
def tower_of_hanoi(n, from_rod, to_rod, aux_rod):
if n == 1:
print(f"Move disk 1 from rod {from_rod} to rod {to_rod}")
return
tower_of_hanoi(n-1, from_rod, aux_rod, to_rod)
print(f"Move disk {n} from rod {from_rod} to rod {to_rod}")
tower_of_hanoi(n-1, aux_rod, to_rod, from_rod)
n = 3 # Number of disks
tower_of_hanoi(n, 'A', 'C', 'B') # A, B and C are names of the rods
```
This script defines a function `tower_of_hanoi` that will print out each move necessary to solve the Tower of Hanoi problem with the specified number of disks `n`. This specific setup will solve for 3 disks moving from rod 'A' to rod 'C' with the help of rod 'B'.
--------------------------------------------------------------------------------
>>>>>>>> EXECUTING CODE BLOCK (inferred language is python)...
code_executor_agent (to code_writer):
exitcode: 0 (execution succeeded)
Code output: Move disk 1 from rod A to rod C
Move disk 2 from rod A to rod B
Move disk 1 from rod C to rod B
Move disk 3 from rod A to rod C
Move disk 1 from rod B to rod A
Move disk 2 from rod B to rod C
Move disk 1 from rod A to rod C
--------------------------------------------------------------------------------
code_writer (to code_executor_agent):
The execution of the provided code successfully calculated and printed the moves for solving the Tower of Hanoi with 3 disks. Here are the steps it performed:
1. Move disk 1 from rod A to rod C.
2. Move disk 2 from rod A to rod B.
3. Move disk 1 from rod C to rod B.
4. Move disk 3 from rod A to rod C.
5. Move disk 1 from rod B to rod A.
6. Move disk 2 from rod B to rod C.
7. Move disk 1 from rod A to rod C.
This sequence effectively transfers all disks from rod A to rod C using rod B as an auxiliary, following the rules of the Tower of Hanoi puzzle. If you have any more tasks or need further explanation, feel free to ask!
--------------------------------------------------------------------------------
code_executor_agent (to code_writer):
--------------------------------------------------------------------------------
import pprint
pprint.pprint(chat_result)
ChatResult(chat_id=None,
chat_history=[{'content': 'Write Python code to calculate the moves '
'of disk on tower of hanoi with 3 disks',
'name': 'code_executor_agent',
'role': 'assistant'},
{'content': 'The problem of the Tower of Hanoi with 3 '
'disks involves moving the disks from one '
'peg to another, following these rules:\n'
'1. Only one disk can be moved at a '
'time.\n'
'2. Each move consists of taking the '
'upper disk from one of the stacks and '
'placing it on top of another stack or on '
'an empty peg.\n'
'3. No disk may be placed on top of a '
'smaller disk.\n'
'\n'
'In the solution, I will use a recursive '
'function to calculate the moves and '
"print them out. Here's the Python code "
'to accomplish this:\n'
'\n'
'```python\n'
'def tower_of_hanoi(n, from_rod, to_rod, '
'aux_rod):\n'
' if n == 1:\n'
' print(f"Move disk 1 from rod '
'{from_rod} to rod {to_rod}")\n'
' return\n'
' tower_of_hanoi(n-1, from_rod, '
'aux_rod, to_rod)\n'
' print(f"Move disk {n} from rod '
'{from_rod} to rod {to_rod}")\n'
' tower_of_hanoi(n-1, aux_rod, to_rod, '
'from_rod)\n'
'\n'
'n = 3 # Number of disks\n'
"tower_of_hanoi(n, 'A', 'C', 'B') # A, B "
'and C are names of the rods\n'
'```\n'
'\n'
'This script defines a function '
'`tower_of_hanoi` that will print out '
'each move necessary to solve the Tower '
'of Hanoi problem with the specified '
'number of disks `n`. This specific setup '
'will solve for 3 disks moving from rod '
"'A' to rod 'C' with the help of rod 'B'.",
'name': 'code_writer',
'role': 'user'},
{'content': 'exitcode: 0 (execution succeeded)\n'
'Code output: Move disk 1 from rod A to '
'rod C\n'
'Move disk 2 from rod A to rod B\n'
'Move disk 1 from rod C to rod B\n'
'Move disk 3 from rod A to rod C\n'
'Move disk 1 from rod B to rod A\n'
'Move disk 2 from rod B to rod C\n'
'Move disk 1 from rod A to rod C\n',
'name': 'code_executor_agent',
'role': 'assistant'},
{'content': 'The execution of the provided code '
'successfully calculated and printed the '
'moves for solving the Tower of Hanoi '
'with 3 disks. Here are the steps it '
'performed:\n'
'\n'
'1. Move disk 1 from rod A to rod C.\n'
'2. Move disk 2 from rod A to rod B.\n'
'3. Move disk 1 from rod C to rod B.\n'
'4. Move disk 3 from rod A to rod C.\n'
'5. Move disk 1 from rod B to rod A.\n'
'6. Move disk 2 from rod B to rod C.\n'
'7. Move disk 1 from rod A to rod C.\n'
'\n'
'This sequence effectively transfers all '
'disks from rod A to rod C using rod B as '
'an auxiliary, following the rules of the '
'Tower of Hanoi puzzle. If you have any '
'more tasks or need further explanation, '
'feel free to ask!',
'name': 'code_writer',
'role': 'user'},
{'content': '',
'name': 'code_executor_agent',
'role': 'assistant'}],
summary='',
cost={'usage_excluding_cached_inference': {'total_cost': 0},
'usage_including_cached_inference': {'gpt-4-turbo-2024-04-09': {'completion_tokens': 499,
'cost': 0.0269,
'prompt_tokens': 1193,
'total_tokens': 1692},
'total_cost': 0.0269}},
human_input=[])
Use ServiceAccount token​
If a PodCommandLineCodeExecutor instance runs inside of Kubernetes
Pod, it can use a token generated from a ServiceAccount to access
Kubernetes API server.
The PodCommandLineCodeExecutor requires the following permissions: the
verbs create, get, delete for pods resource, and the verb get
for resources pods/status, pods/exec.
You can create a ServiceAccount, ClusterRole and RoleBinding with
kubectl as shown below:
%%bash
# Create ServiceAccount on default namespace
kubectl create sa autogen-executor-sa
serviceaccount/autogen-executor-sa created
%%bash
# Create ClusterRole that has sufficient permissions
kubectl create clusterrole autogen-executor-role \
--verb=get,create,delete --resource=pods,pods/status,pods/exec
clusterrole.rbac.authorization.k8s.io/autogen-executor-role created
%%bash
# Create RoleBinding that binds ClusterRole and ServiceAccount
kubectl create rolebinding autogen-executor-rolebinding \
--clusterrole autogen-executor-role --serviceaccount default:autogen-executor-sa
rolebinding.rbac.authorization.k8s.io/autogen-executor-rolebinding created
A pod with a previously created ServiceAccount can be launched using the following command.
%%bash
# create pod with serviceaccount
kubectl run autogen-executor --image python:3 \
--overrides='{"spec":{"serviceAccount": "autogen-executor-sa"}}' \
-- bash -c 'pip install pyautogen[kubernetes] && sleep inifinity'
pod/autogen-executor created
You can execute PodCommandLineCodeExecutor inside the Python
interpreter process from autogen-executor Pod.
It creates new pod for code execution using token generated from
autogen-executor-sa ServiceAccount.
%%bash
kubectl exec autogen-executor -it -- python
from autogen.coding import CodeBlock
from autogen.coding.kubernetes import PodCommandLineCodeExecutor
# PodCommandLineCodeExecutor uses token generated from ServiceAccount by kubernetes incluster config
with PodCommandLineCodeExecutor() as executor:
print(
executor.execute_code_blocks(
code_blocks=[
CodeBlock(language="python", code="print('Hello, World!')"),
]
)
)
kube_config_path not provided and default location (~/.kube/config) does not exist. Using inCluster Config. This might not work.
exit_code=0 output='Hello, World!\n' code_file='/workspace/tmp_code_07da107bb575cc4e02b0e1d6d99cc204.py'