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

> Learn to create custom operators in DB-GPT's AWEL system for advanced data processing. Subclass base operators and implement methods to build powerful data pipelines easily.

- Repository: [eosphoros/DB-GPT](https://github.com/eosphoros-ai/db-gpt)
- Tags: how-to-guide
- Published: 2026-02-23

---

**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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/common_operator.py) and [`stream_operator.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/stream_operator.py) respectively.

### DAG Orchestration Layer

The `DAG` class in [`packages/dbgpt-core/src/dbgpt/core/awel/__init__.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/docs/docs/awel/awel_tutorial/getting_started/1.3_custom_operator.md).

```python
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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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:

```python
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:

```python
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:

```python
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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/common_operator.py) | Defines `BaseOperator`, `MapOperator`, and `BranchOperator` | [`packages/dbgpt-core/src/dbgpt/core/awel/operators/common_operator.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-core/src/dbgpt/core/awel/operators/common_operator.py) |
| [`stream_operator.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/stream_operator.py) | Implements streaming base classes | [`packages/dbgpt-core/src/dbgpt/core/awel/operators/stream_operator.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-core/src/dbgpt/core/awel/operators/stream_operator.py) |
| [`__init__.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/__init__.py) (awel) | Provides `DAG` class and orchestration utilities | [`packages/dbgpt-core/src/dbgpt/core/awel/__init__.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-core/src/dbgpt/core/awel/__init__.py) |
| [`1.3_custom_operator.md`](https://github.com/eosphoros-ai/DB-GPT/blob/main/1.3_custom_operator.md) | Official tutorial with step-by-step examples | [`docs/docs/awel/awel_tutorial/getting_started/1.3_custom_operator.md`](https://github.com/eosphoros-ai/DB-GPT/blob/main/docs/docs/awel/awel_tutorial/getting_started/1.3_custom_operator.md) |
| [`operators.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/operators.py) (conversation) | Real-world reference implementations | [`packages/dbgpt-serve/src/dbgpt_serve/conversation/operators.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/1.3_custom_operator.md) and [`stream_operator.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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.