Check this series

it will help you

This video is also great

Spotify is one of sync agent's user

This talk can help you understand how sync agent works.

this doc will help you too

But to start with you can simply write it as a task. The async agent is great for scaling and letting other folks reuse this thing and reduce cost too

agent_service.py

is for remote execution.

base_agent.py

is for local execution. https://github.com/flyteorg/flytekit/tree/master/flytekit/extend/backend You can see the lifecycle of executing agent tasks by reading

base_agent.py

https://github.com/flyteorg/flytekit/blob/master/flytekit/extend/backend/base_agent.py#L277

But you need to implement

AsyncAgentBase

's method. https://github.com/flyteorg/flytekit/blob/master/flytekit/extend/backend/base_agent.py#L127-L175

for example

class ChatGPTTask(PythonTask):
    """
    This is the simplest form of a ChatGPT Task, you can define the model and the input you want.
    """

    _TASK_TYPE = "chatgpt"

    def __init__(self, name: str, openai_organization: str, chatgpt_config: Dict[str, Any], **kwargs):
        """
        Args:
            name: Name of this task, should be unique in the project
            openai_organization: OpenAI Organization. String can be found here. <https://platform.openai.com/docs/api-reference/organization-optional>
            chatgpt_config: ChatGPT job configuration. Config structure can be found here. <https://platform.openai.com/docs/api-reference/completions/create>
        """

        if "model" not in chatgpt_config:
            raise ValueError("The 'model' configuration variable is required in chatgpt_config")

        task_config = {"openai_organization": openai_organization, "chatgpt_config": chatgpt_config}

        inputs = {"message": str}
        outputs = {"o0": str}

        super().__init__(
            task_type=self._TASK_TYPE,
            name=name,
            task_config=task_config,
            interface=Interface(inputs=inputs, outputs=outputs),
            **kwargs,
        )

    def get_custom(self, settings: SerializationSettings) -> Dict[str, Any]:
        return {
            "openai_organization": self.task_config["openai_organization"],
            "chatgpt_config": self.task_config["chatgpt_config"],
        }

this is my

chatgpt/task.py

, it can work remotely.

This is the source code, you can find that I remove

SyncAgentExecutorMixin

and it still works. https://github.com/flyteorg/flytekit/blob/master/plugins/flytekit-openai/flytekitplugins/openai/chatgpt/task.py

local:

pyflyte run task.py xxx

remote:

pyflyte run --remote task.py xxx

thanks, feel free to ask more

My implementation is similar to the BigQuery/ChatGPT examples, and I now want to test that it's working ok - but I'm hitting a few issues from the documentation: 1. I tried seeing if I can just test the task using

pyflyte run workflows.py my_example_task

(eg. this example) but get an error:

SYSTEM:AgentNotFound: error=Cannot find agent for task category: custom_api_v0

So I'm not sure if there might be an extra step needed? Or if this method only works for existing flyte plugins 🤔 2. I tried seeing if I can follow these instructions to run the agent alongside the flyte demo cluster, but hit a few confusion points: •

pyflyte serve agent

doesn't work unless I do an additional

pip install prometheus-client grpcio-health-checking

• I wasn't sure if

flyte-single-binary-local.yaml

is an existing file I should be modifying, or a new file I should be creating? • The instructions mention

make compile

, suggesting an existing Makefile. Where is this located, if I'm running the `flytectl`demo cluster on MacOS from a brew installation? • The note at the bottom of the page mentions "You must build an image that includes the plugin for the task". Should this be one of the main walkthrough steps, or is this only needed in some situations? Thanks, lots of info there - I won't be working on this again til Monday 😄

Straight after posting this I solved question 1 😂 I needed to import

MyCustomAPIAgent

(which included the AgentRegister step) into

workflows.py

, even though it's not used, to ensure the agent is registered at the point of execution 👍 Still working on the demo cluster execution, but I have a basic local test workflow now 🚀