How to Create Custom Operators in DB-GPT's AWEL System for Data Processing

To create custom operators in DB-GPT, subclass a base operator class from the AWEL (Agentic Workflow Expression Language) package—such as MapOperator for one-to-one transformations or StreamifyAbsOperator for streaming data—and implement the required abstract methods like map() or streamify(), then wire them together using the DAG class.

DB-GPT is an open-source AI-native data app development framework that uses AWEL as its internal workflow engine. All data processing steps in DB-GPT are implemented as operators that inherit from base classes defined in the core package. Understanding how to create custom operators allows you to extend DB-GPT's data processing capabilities for specific ETL tasks, LLM interactions, or custom business logic.

Understanding DB-GPT's Operator Architecture

Core Base Classes

The operator system is defined in packages/dbgpt-core/src/dbgpt/core/awel/operators/. The key base classes include:

• MapOperator[IN, OUT] – Processes a single input and returns a single output (one-to-one transformation)
• StreamifyAbsOperator[IN, OUT] – Emits a stream of values from a single input (one-to-many)
• TransformStreamAbsOperator[IN, OUT] – Consumes an async iterator and yields a transformed stream

These classes are implemented in common_operator.py and stream_operator.py respectively.

DAG Orchestration Layer

The DAG class in packages/dbgpt-core/src/dbgpt/core/awel/__init__.py builds directed acyclic graphs of operators. It handles data flow, parallelism, and async execution. Operators are linked using the >> pipe operator, which wires upstream output to downstream input.

How to Create a Custom MapOperator

For simple one-to-one data transformations, inherit from MapOperator and implement the map() method. This pattern is documented in docs/docs/awel/awel_tutorial/getting_started/1.3_custom_operator.md.

import asyncio
from dbgpt.core.awel import DAG, MapOperator

class HelloWorldOperator(MapOperator[str, None]):
    async def map(self, x: str) -> None:
        # Your custom processing here

        print(f"Hello, {x}!")

# Build a DAG that contains the operator

with DAG("awel_hello_world") as dag:
    task = HelloWorldOperator()

# Run the operator (prints "Hello, world!")

asyncio.run(task.call(call_data="world"))

The MapOperator implementation in packages/dbgpt-core/src/dbgpt/core/awel/operators/common_operator.py automatically handles serializability and task metadata capture.

How to Create Custom Streaming Operators

For processing data streams, use the streaming base classes defined in packages/dbgpt-core/src/dbgpt/core/awel/operators/stream_operator.py.

Creating a StreamifyAbsOperator (Producer)

This operator converts a single input into an async stream of outputs:

from typing import AsyncIterator
from dbgpt.core.awel import StreamifyAbsOperator

class NumberProducerOperator(StreamifyAbsOperator[int, int]):
    async def streamify(self, n: int) -> AsyncIterator[int]:
        for i in range(n):
            yield i

Creating a TransformStreamAbsOperator (Consumer)

This operator consumes an async iterator and transforms each element:

from typing import AsyncIterator
from dbgpt.core.awel import TransformStreamAbsOperator

class NumberDoubleOperator(TransformStreamAbsOperator[int, int]):
    async def transform_stream(self, it: AsyncIterator[int]) -> AsyncIterator[int]:
        async for i in it:
            yield i * 2

Wiring Streaming Operators Together

Combine producers and consumers using the DAG and pipe operator:

import asyncio
from dbgpt.core.awel import DAG

# Assemble the workflow

with DAG("numbers_dag") as dag:
    producer = NumberProducerOperator()
    doubler = NumberDoubleOperator()
    producer >> doubler      # pipe output of producer to doubler

# Helper to consume the stream and print results

async def run_example():
    async for v in await doubler.call_stream(call_data=10):
        print(v)

asyncio.run(run_example())

Key Source Files for Custom Operator Development

File	Role	Path
common_operator.py	Defines BaseOperator, MapOperator, and BranchOperator	packages/dbgpt-core/src/dbgpt/core/awel/operators/common_operator.py
stream_operator.py	Implements streaming base classes	packages/dbgpt-core/src/dbgpt/core/awel/operators/stream_operator.py
__init__.py (awel)	Provides DAG class and orchestration utilities	packages/dbgpt-core/src/dbgpt/core/awel/__init__.py
1.3_custom_operator.md	Official tutorial with step-by-step examples	docs/docs/awel/awel_tutorial/getting_started/1.3_custom_operator.md
operators.py (conversation)	Real-world reference implementations	packages/dbgpt-serve/src/dbgpt_serve/conversation/operators.py

Summary

• Choose the right base class: Use MapOperator for one-to-one transformations, StreamifyAbsOperator for producing streams, and TransformStreamAbsOperator for consuming and transforming streams.
• Implement required methods: Override map(), streamify(), or transform_stream() with your custom logic in the respective subclasses.
• Use DAG for orchestration: Wire operators together using the >> pipe operator within a DAG context manager to build executable workflows.
• Reference official examples: Study 1.3_custom_operator.md and stream_operator.py for production patterns and advanced techniques.

Frequently Asked Questions

What is the difference between MapOperator and StreamifyAbsOperator?

MapOperator processes a single input and returns a single output (one-to-one transformation), while StreamifyAbsOperator takes a single input and yields multiple outputs as an async iterator (one-to-many). Use MapOperator for simple transformations like formatting or validation, and StreamifyAbsOperator when you need to expand a single request into a stream of chunks or events.

How do I handle errors in custom operators?

Implement error handling within your map(), streamify(), or transform_stream() methods using standard Python try-except blocks. For async generators in streaming operators, ensure you handle exceptions before yielding values. The AWEL framework propagates exceptions through the DAG context, allowing upstream operators or the execution engine to catch and log errors according to the configuration in DAGContext.

Can I compose multiple custom operators into a complex workflow?

Yes, you can chain any number of operators using the >> pipe operator within a DAG context. For example: operator_a >> operator_b >> operator_c. The framework automatically handles type checking between connected operators and manages data flow through the DAG. You can also branch workflows using BranchOperator defined in common_operator.py or merge streams using specialized join operators.

Where can I find real-world examples of custom operators in DB-GPT?

The packages/dbgpt-serve/src/dbgpt_serve/conversation/operators.py file contains production implementations of operators used for LLM interactions and conversation management. Additionally, the official tutorial at docs/docs/awel/awel_tutorial/getting_started/1.3_custom_operator.md provides step-by-step guides for both simple map operators and streaming operators. These resources demonstrate patterns for handling async I/O, state management, and integration with external APIs.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}