The Dynamic Agent Workflow system is a program that evolved from me scripting a simple series of ai agents and connecting to a local ai using ollama to run the output from one ai response into the input of another ai system with added directions on how to handle that input. A simple workflow of
I implemented proc agents with just 5 lines of actual code. These agents give you access to functions to do data processing on the results the ai nodes return. I was getting a block of base64 coded png file from the stable diffusion api I am running locally when I had it make an image for me from a prompt using the web api. I wanted to decode that block of base64 encoded result and save it to disk. I am doing that now.
''' ==== ==== Proc section === === '''
def exec_proc_agent(function_name: str, step_params: Dict[str, Any],function_def: str) -> bytes:
logging.info(f"Starting {function_name}")
logging.debug(f"******** \n step_params{step_params}")
logging.debug(f"Function definition:\n{function_def}")
result = b'' # Initialize result as empty bytes
try:
# Dynamically create abd dispatch to the appropriate processing function
# Check if the function is already defined in the global scope
if function_name not in globals():
# Define the function dynamically
exec(function_def, globals())
func = globals()[function_name]
result = func(**step_params)
except Exception as e:
logging.error(f"An error occurred while executing {function_name}: {str(e)}")
raise
logging.info(f"Completed {function_name}")
return result
This dynamically and lazily creates functions dynamically from the proc agent's definition in the json configuration file, it stores the script in the function_def tag.
"read_file": {
"type": "proc",
"help": "Reads content from a file",
"function": "read_file",
"function_def": "def read_file(file_name: str) -> bytes:
\n\twith open(file_name, 'rb') as f:
\n\t\treturn f.read()\n",
"inputs": [
"file_name"
],
"outputs": [
"file_content"
],
"prompt": "{prompt}",
"steps": []
},
"write_file": {
"type": "proc",
"help": "Writes content to a file",
"function": "write_file",
"function_def": "def write_file(file_name: str, file_content: bytes) -> bytes:\n
\twith open(file_name, 'wb') as f:\n\t\tf.write(file_content)
\n\treturn b\"Successfully wrote data\"\n",
"inputs": [
"file_name",
"file_content"
],
"outputs": [
"success"
],
"prompt": "{prompt}",
"steps": []
},
The beautiful of this approach is that there is no code to maintain in the core system and the system becomes fully configurable. It is hard to make mistakes in just 5 lines of code, and easy to fix it if you did.
And to work on a new function you can just make the function and edit it like normal and then json-ify it later once it is tested. You have to escape the quotes, newlines, double quotes, and tabs. Maybe some other things too, I don't know, maybe there is a tool that can take a string and escape it for you.
It feels like I am onto something when I can introduce a huge upgrade like processing agents to my ai agents workflow system and it only requires a conditional statement, a function call, and a 5 line program to implement it.
I know that json configuration files are not the easiest thing to configure. But once I have this system developed to the point where the configurations and features have slowed down to a crawl, I will start developing a UI to develop and execute workflows. Based on existing agents. And once you have a workflow build, you can add it to the configuration file automatically, making another agent that you can use in more workflows.
No comments:
Post a Comment