How to use custom components with the Plugin System#

Prerequisites

This guide assumes you are familiar with the following concepts:

Additionally, you need to have Python 3.10+ installed.

Overview#

Agent Spec supports a list of core native components to build your LLM-powered assistants, for instance a list of pre-defined Nodes (Agent Spec Spec, API Reference).

You might however want to go beyond the list of components supported in the base specification and use additional components while being able to use the Agent Spec Assistant Configuration system. The pyagentspec plugin system is designed to address this use case. Example uses of the plugin system include:

  • New Large Language Model (LLM) configurations (by extending the LlmConfig component);

  • New nodes for Flows (by extending the Node component);

  • New tools (by extending the Tool component);

  • Extensions to the Agent and Flow components.

How the plugin system is used

Overview of the Agent Spec plugin system.#

This guide demonstrates how to support a custom PluginRegexNode in Flows to extract information from a raw text using a regular expression (regex). You will:

  1. Create a custom PluginRegexNode Agent Spec node and (de)serialization plugins so your assistant can be correctly saved and loaded;

  2. Use the custom component to create an assistant that can extract information using regex;

  3. Export your assistant configuration to Agent Spec JSON/YAML.

  4. Load and execute your agent in a runtime executor (WayFlow).

Important

The current plugin system enables the serialization and deserialization of custom components for Agent Spec.

To fully support custom components, you also need to:

  1. Support the feature in your Agent Spec runtime implementation.

  2. Support the conversion between the custom Agent Spec plugin component and its corresponding implementation in your runtime implementation.

Basic implementation#

Define custom Agent Spec components#

First you need to create the corresponding Agent Spec components.

from typing import ClassVar, List

from pyagentspec.flows.node import Node
from pyagentspec.property import Property, StringProperty

class PluginRegexNode(Node):
    """Node to extract information from a raw text using a regular expression (regex)."""

    regex_pattern: str
    """Specify the regex pattern for searching in the input text."""

    DEFAULT_INPUT_KEY: ClassVar[str] = "text"
    """Input key for the name to transition to next."""

    DEFAULT_OUTPUT_KEY: ClassVar[str] = "output"
    """Input key for the name to transition to next."""

    def _get_inferred_inputs(self) -> List[Property]:
        input_title = self.inputs[0].title if self.inputs else self.DEFAULT_INPUT_KEY
        return [StringProperty(title=input_title)]

    def _get_inferred_outputs(self) -> List[Property]:
        output_title = self.outputs[0].title if self.outputs else self.DEFAULT_OUTPUT_KEY
        return [StringProperty(title=output_title)]

API Reference: Node | Property

Here, the PluginRegexNode is the custom node that we intent to use in a Flow.

Create an assistant using the custom component#

The decision mechanism of Agents is powered by a Large Language Model. Defining the agent with Agent Spec requires to pass the configuration for the LLM. There are several options such as using OCI GenAI Service or self hosting a model with vLLM.

Start by defining the LLM configuration to be shared across all agents. This example uses a vLLM instance running Llama3.3 70B.

from pyagentspec.llms import OciGenAiConfig
from pyagentspec.llms.ociclientconfig import OciClientConfigWithApiKey

client_config = OciClientConfigWithApiKey(
    name="Oci Client Config",
    service_endpoint="https://url-to-service-endpoint.com",
    auth_profile="DEFAULT",
    auth_file_location="~/.oci/config"
)

llm_config = OciGenAiConfig(
    name="Oci GenAI Config",
    model_id="provider.model-id",
    compartment_id="compartment-id",
    client_config=client_config,
)

Here, you will create a Flow that generates an answer to a query using reasoning thoughts, and then parses the output to return the final answer.

Use the PluginRegexNode defined previously to create your assistant.

from pyagentspec.flows.edges import ControlFlowEdge, DataFlowEdge
from pyagentspec.flows.flow import Flow
from pyagentspec.flows.nodes import EndNode, LlmNode, StartNode
from pyagentspec.property import StringProperty

llmoutput_property = StringProperty(title="llm_output", default="ERROR")
parsedoutput_property = StringProperty(title="output", default="<result>ERROR</result>")
start_node = StartNode(id="start", name="start", inputs=[])
llm_node = LlmNode(
    id="llm",
    name="llm",
    llm_config=llm_config,
    prompt_template=(
        "What is the result of 100+(454-3). Think step by step and then give "
        "your answer between <result>...</result> delimiters"
    ),
    outputs=[llmoutput_property],
)
regex_node = PluginRegexNode(
    id="regex",
    name="regex",
    regex_pattern=r"<result>(.*)</result>",
    outputs=[parsedoutput_property],
)
end_node = EndNode(id="end", name="end", outputs=[parsedoutput_property])

assistant = Flow(
    id="regex_flow",
    name="regex_flow",
    start_node=start_node,
    nodes=[start_node, llm_node, regex_node, end_node],
    control_flow_connections=[
        ControlFlowEdge(id="start_llm", name="start->llm", from_node=start_node, to_node=llm_node),
        ControlFlowEdge(id="llm_regex", name="llm->regex", from_node=llm_node, to_node=regex_node),
        ControlFlowEdge(id="regex_end", name="regex->end", from_node=regex_node, to_node=end_node),
    ],
    data_flow_connections=[
        DataFlowEdge(
            name="edge",
            source_node=llm_node,
            source_output="llm_output",
            destination_node=regex_node,
            destination_input="text"
        ),
        DataFlowEdge(
            name="edge",
            source_node=regex_node,
            source_output="output",
            destination_node=end_node,
            destination_input="output"
        ),
    ]
)

Register plugins and export the agent configuration#

You need to register (de)serialization plugins so your custom node can be correctly saved and loaded using the Agent Spec serialization format.

from pyagentspec.serialization.pydanticdeserializationplugin import (
    PydanticComponentDeserializationPlugin,
)
from pyagentspec.serialization.pydanticserializationplugin import PydanticComponentSerializationPlugin

example_serialization_plugin = PydanticComponentSerializationPlugin(
    component_types_and_models={
        PluginRegexNode.__name__: PluginRegexNode,
    }
)

example_deserialization_plugin = PydanticComponentDeserializationPlugin(
    component_types_and_models={
        PluginRegexNode.__name__: PluginRegexNode,
    }
)

API Reference: PydanticComponentSerializationPlugin | PydanticComponentDeserializationPlugin.

This enables PyAgentSpec’s serializer and deserializer to recognize your custom classes. In this example, since the components are directly inheriting from the Pydantic model Component, you should use the already existing (de)serialization plugins.

For more advanced use, you can implement custom plugins by inheriting from ComponentSerializationPlugin and ComponentDeserializationPlugin

You can then serialize your assistant to its Agent Spec JSON/YAML configuration using the registered plugins:

from pyagentspec.serialization import AgentSpecSerializer
serialized_assistant = AgentSpecSerializer(plugins=[example_serialization_plugin]).to_json(assistant)

with open("assistant_config.json", "w") as f:
    f.write(serialized_assistant)

API Reference: AgentSpecSerializer

Here is what the Agent Spec representation will look like ↓

Click here to see the assistant configuration.
{
    "component_type": "Flow",
    "id": "regex_flow",
    "name": "regex_flow",
    "description": null,
    "metadata": {},
    "inputs": [],
    "outputs": [
        {
            "title": "output",
            "default": "<result>ERROR</result>",
            "type": "string"
        }
    ],
    "start_node": {
        "$component_ref": "start"
    },
    "nodes": [
        {
            "$component_ref": "start"
        },
        {
            "$component_ref": "llm"
        },
        {
            "$component_ref": "regex"
        },
        {
            "$component_ref": "end"
        }
    ],
    "state": [],
    "control_flow_connections": [
        {
            "component_type": "ControlFlowEdge",
            "id": "start_llm",
            "name": "start->llm",
            "description": null,
            "metadata": {},
            "from_node": {
                "$component_ref": "start"
            },
            "from_branch": null,
            "to_node": {
                "$component_ref": "llm"
            }
        },
        {
            "component_type": "ControlFlowEdge",
            "id": "llm_regex",
            "name": "llm->regex",
            "description": null,
            "metadata": {},
            "from_node": {
                "$component_ref": "llm"
            },
            "from_branch": null,
            "to_node": {
                "$component_ref": "regex"
            }
        },
        {
            "component_type": "ControlFlowEdge",
            "id": "regex_end",
            "name": "regex->end",
            "description": null,
            "metadata": {},
            "from_node": {
                "$component_ref": "regex"
            },
            "from_branch": null,
            "to_node": {
                "$component_ref": "end"
            }
        }
    ],
    "data_flow_connections": [
        {
            "component_type": "DataFlowEdge",
            "id": "a5811a1e-710d-4c88-927c-6249b3186e95",
            "name": "edge",
            "description": null,
            "metadata": {},
            "source_node": {
                "$component_ref": "llm"
            },
            "source_output": "llm_output",
            "destination_node": {
                "$component_ref": "regex"
            },
            "destination_input": "text"
        },
        {
            "component_type": "DataFlowEdge",
            "id": "14ed74b6-e344-4353-ad10-6afaab9d5dc4",
            "name": "edge",
            "description": null,
            "metadata": {},
            "source_node": {
                "$component_ref": "regex"
            },
            "source_output": "output",
            "destination_node": {
                "$component_ref": "end"
            },
            "destination_input": "output"
        }
    ],
    "$referenced_components": {
        "llm": {
            "component_type": "LlmNode",
            "id": "llm",
            "name": "llm",
            "description": null,
            "metadata": {},
            "inputs": [],
            "outputs": [
                {
                    "title": "llm_output",
                    "default": "ERROR",
                    "type": "string"
                }
            ],
            "branches": [
                "next"
            ],
            "llm_config": {
                "component_type": "VllmConfig",
                "id": "34f14c5d-5af0-4ca5-b3f1-3e61b4280fda",
                "name": "Vllm model",
                "description": null,
                "metadata": {},
                "default_generation_parameters": null,
                "url": "vllm_url",
                "model_id": "model_id"
            },
            "prompt_template": "What is the result of 100+(454-3). Think step by step and then give your answer between <result>...</result> delimiters"
        },
        "start": {
            "component_type": "StartNode",
            "id": "start",
            "name": "start",
            "description": null,
            "metadata": {},
            "inputs": [],
            "outputs": [],
            "branches": [
                "next"
            ]
        },
        "regex": {
            "component_type": "PluginRegexNode",
            "id": "regex",
            "name": "regex",
            "description": null,
            "metadata": {},
            "inputs": [
                {
                    "title": "text",
                    "type": "string"
                }
            ],
            "outputs": [
                {
                    "title": "output",
                    "default": "<result>ERROR</result>",
                    "type": "string"
                }
            ],
            "branches": [
                "next"
            ],
            "regex_pattern": "<result>(.*)</result>",
            "component_plugin_name": "PydanticComponentPlugin",
            "component_plugin_version": "25.3.0.dev2"
        },
        "end": {
            "component_type": "EndNode",
            "id": "end",
            "name": "end",
            "description": null,
            "metadata": {},
            "inputs": [
                {
                    "title": "output",
                    "default": "<result>ERROR</result>",
                    "type": "string"
                }
            ],
            "outputs": [
                {
                    "title": "output",
                    "default": "<result>ERROR</result>",
                    "type": "string"
                }
            ],
            "branches": [],
            "branch_name": "next"
        }
    },
    "agentspec_version": "25.4.1"
}

Load and execute your assistant from a configuration file#

See also

For more details, see the guide on How to Execute an Agent Spec Configuration with WayFlow.

To actually run the assistant, load the JSON or YAML configuration using the deserialization plugins defined above.

from wayflowcore.flow import Flow as RuntimeFlow
from wayflowcore.agentspec import AgentSpecLoader

with open("assistant_config.json") as f:
    agentspec_export = f.read()

# agentspec_loader = AgentSpecLoader(plugins=[example_deserialization_plugin])
assistant: RuntimeFlow = agentspec_loader.load_yaml(agentspec_export)

Now you can run your assistant! The assistant calls an LLM with the pre-defined request, and returns the parsed output.

from wayflowcore.executors.executionstatus import FinishedStatus
inputs = {}
conversation = assistant.start_conversation(inputs)

status = conversation.execute()
if isinstance(status, FinishedStatus):
    outputs = status.output_values
    print(f"Assistant outputs: {outputs['output']}")
else:
    print(f"ERROR: Expected 'FinishedStatus', got {status.__class__.__name__}")

# Assistant outputs: 551

Recap#

This guide covered how to:

  • Create a custom Agent Spec component and its corresponding (de)serialization plugin.

  • Build and export an assistant (Flow) with a custom regex node using the custom component and plugin.

  • Load and execute the assistant in a runtime executor.

Below is the complete code from this guide.
from typing import ClassVar, List

from pyagentspec.flows.node import Node
from pyagentspec.property import Property, StringProperty

class PluginRegexNode(Node):
    """Node to extract information from a raw text using a regular expression (regex)."""

    regex_pattern: str
    """Specify the regex pattern for searching in the input text."""

    DEFAULT_INPUT_KEY: ClassVar[str] = "text"
    """Input key for the name to transition to next."""

    DEFAULT_OUTPUT_KEY: ClassVar[str] = "output"
    """Input key for the name to transition to next."""

    def _get_inferred_inputs(self) -> List[Property]:
        input_title = self.inputs[0].title if self.inputs else self.DEFAULT_INPUT_KEY
        return [StringProperty(title=input_title)]

    def _get_inferred_outputs(self) -> List[Property]:
        output_title = self.outputs[0].title if self.outputs else self.DEFAULT_OUTPUT_KEY
        return [StringProperty(title=output_title)]

from pyagentspec.flows.edges import ControlFlowEdge, DataFlowEdge
from pyagentspec.flows.flow import Flow
from pyagentspec.flows.nodes import EndNode, LlmNode, StartNode
from pyagentspec.property import StringProperty

llmoutput_property = StringProperty(title="llm_output", default="ERROR")
parsedoutput_property = StringProperty(title="output", default="<result>ERROR</result>")
start_node = StartNode(id="start", name="start", inputs=[])
llm_node = LlmNode(
    id="llm",
    name="llm",
    llm_config=llm_config,
    prompt_template=(
        "What is the result of 100+(454-3). Think step by step and then give "
        "your answer between <result>...</result> delimiters"
    ),
    outputs=[llmoutput_property],
)
regex_node = PluginRegexNode(
    id="regex",
    name="regex",
    regex_pattern=r"<result>(.*)</result>",
    outputs=[parsedoutput_property],
)
end_node = EndNode(id="end", name="end", outputs=[parsedoutput_property])

assistant = Flow(
    id="regex_flow",
    name="regex_flow",
    start_node=start_node,
    nodes=[start_node, llm_node, regex_node, end_node],
    control_flow_connections=[
        ControlFlowEdge(id="start_llm", name="start->llm", from_node=start_node, to_node=llm_node),
        ControlFlowEdge(id="llm_regex", name="llm->regex", from_node=llm_node, to_node=regex_node),
        ControlFlowEdge(id="regex_end", name="regex->end", from_node=regex_node, to_node=end_node),
    ],
    data_flow_connections=[
        DataFlowEdge(
            name="edge",
            source_node=llm_node,
            source_output="llm_output",
            destination_node=regex_node,
            destination_input="text"
        ),
        DataFlowEdge(
            name="edge",
            source_node=regex_node,
            source_output="output",
            destination_node=end_node,
            destination_input="output"
        ),
    ]
)

from pyagentspec.serialization.pydanticdeserializationplugin import (
    PydanticComponentDeserializationPlugin,
)
from pyagentspec.serialization.pydanticserializationplugin import PydanticComponentSerializationPlugin

example_serialization_plugin = PydanticComponentSerializationPlugin(
    component_types_and_models={
        PluginRegexNode.__name__: PluginRegexNode,
    }
)

example_deserialization_plugin = PydanticComponentDeserializationPlugin(
    component_types_and_models={
        PluginRegexNode.__name__: PluginRegexNode,
    }
)

from pyagentspec.serialization import AgentSpecSerializer
serialized_assistant = AgentSpecSerializer(plugins=[example_serialization_plugin]).to_json(assistant)

with open("assistant_config.json", "w") as f:
    f.write(serialized_assistant)


from wayflowcore.flow import Flow as RuntimeFlow
from wayflowcore.agentspec import AgentSpecLoader

with open("assistant_config.json") as f:
    agentspec_export = f.read()

# agentspec_loader = AgentSpecLoader(plugins=[example_deserialization_plugin])
assistant: RuntimeFlow = agentspec_loader.load_yaml(agentspec_export)

from wayflowcore.executors.executionstatus import FinishedStatus
inputs = {}
conversation = assistant.start_conversation(inputs)

status = conversation.execute()
if isinstance(status, FinishedStatus):
    outputs = status.output_values
    print(f"Assistant outputs: {outputs['output']}")
else:
    print(f"ERROR: Expected 'FinishedStatus', got {status.__class__.__name__}")

# Assistant outputs: 551

Next steps#

Having learned how to build assistants with custom Agent Spec components using the plugin system, you may now proceed to How to Build an Orchestrator-Workers Agents Pattern.