How DB-GPT AWEL Works: A Complete Guide to Creating Custom Workflows

AWEL (Agentic Workflow Expression Language) is DB-GPT's internal workflow engine that enables you to define complex LLM-driven pipelines as directed-acyclic graphs (DAGs) of operators and resources, which are automatically built and executed via the FlowFactory class.

DB-GPT's AWEL serves as the declarative backbone for constructing agentic workflows within the eosphoros-ai/DB-GPT repository. By representing workflows as JSON-serializable FlowPanel objects, AWEL bridges visual pipeline design with executable Python code, allowing you to orchestrate everything from simple data transformations to multi-step LLM reasoning chains.

What Is AWEL?

AWEL represents workflows as directed-acyclic graphs (DAGs) composed of two fundamental primitives: operators (processing nodes) and resources (external dependencies). The central data structure is the FlowPanel, a Pydantic model defined in packages/dbgpt-core/src/dbgpt/core/awel/flow/base.py that contains:

• nodes: A list of FlowNodeData objects representing either operators (ViewMetadata) or resources (ResourceMetadata)
• edges: Connections defining data flow between nodes
• variables: Flow-level variables accessible to all operators
• metadata: UI rendering and description information

The FlowFactory class located in packages/dbgpt-core/src/dbgpt/core/awel/flow/flow_factory.py consumes this description, validates the structure, performs topological sorting, and constructs a runnable DAG object.

Core Architecture and Components

Operators

All processing nodes inherit from BaseOperator (defined in packages/dbgpt-core/src/dbgpt/core/awel/operators/base.py). The engine provides several ready-to-extend base classes in packages/dbgpt-core/src/dbgpt/core/awel/operators/common_operator.py:

• MapOperator: Transforms input to output via the map() method
• BranchOperator: Routes data to different downstream paths
• JoinOperator: Aggregates multiple inputs via the reduce() method

Each operator implements _do_run() or specific abstract methods like map(), and is automatically registered via BaseOperatorMeta when imported.

Resources

Resources are thin wrappers providing external services such as LLM clients, vector stores, or HTTP bodies. They are instantiated once per flow and injected into operators based on ResourceMetadata declarations. The registry lookup occurs in resource/base.py.

Triggers

Triggers initiate flows from external events. The built-in HTTPTrigger in packages/dbgpt-core/src/dbgpt/core/awel/trigger/http_trigger.py exposes flows as REST endpoints, creating temporary DAGs upon each request.

Workflow Runners

The WorkflowRunner abstract base class (line 48 in operators/base.py) executes the DAG. The default DefaultWorkflowRunner processes nodes in topological order within the same process, while custom runners enable distributed execution.

How Flows Are Built

The FlowFactory.build() method orchestrates DAG construction through seven distinct phases:

1. Parse the JSON panel: The FlowPanel Pydantic model validates node and edge structure
2. Separate nodes: Build lookup maps for operators, resources, and their upstream/downstream relationships
3. Topological sort: The _topological_sort() method (line 442 in flow_factory.py) guarantees dependency resolution order
4. Instantiate resources: Map metadata to concrete classes via _get_resource_class()
5. Create operator tasks: Map metadata to classes via _get_operator_class(), injecting resolved resources into constructors
6. Wire the DAG: Connect nodes using the >> operator overload on DAGNode
7. Return the built DAG: A runnable graph ready for execution

Creating Custom Workflows

Step 1: Define a Custom Operator

Create a Python class inheriting from one of the common operator bases and implement the required abstract method:

# my_operator.py

from dbgpt.core.awel.operators.common_operator import MapOperator

class UpperCaseOperator(MapOperator[str, str]):
    """Simple operator that upper-cases a string."""
    
    async def map(self, input_value: str) -> str:
        return input_value.upper()

The class automatically registers upon import because BaseOperatorMeta calls _apply_defaults() and after_define() during class creation.

Step 2: Export Operator Metadata

Define metadata so the factory can render your operator in the UI and validate parameters:

# my_operator_metadata.py

from dbgpt.core.awel.flow.base import OperatorCategory, OperatorType
from my_operator import UpperCaseOperator

UpperCaseOperator.metadata = UpperCaseOperator.Metadata(
    operator_type=OperatorType.MAP,
    operator_category=OperatorCategory.COMMON,
    label="Upper-Case",
    description="Converts a string to upper case",
    inputs=[UpperCaseOperator.Parameter(
        name="input", 
        type_name="str", 
        type_cls="builtins.str", 
        label="Input"
    )],
    outputs=[UpperCaseOperator.Parameter(
        name="output", 
        type_name="str", 
        type_cls="builtins.str", 
        label="Output"
    )]
)

Step 3: Build a FlowPanel JSON

Construct the declarative workflow description referencing your custom operator:

{
  "label": "Upper-Case Demo",
  "name": "upper_case_demo",
  "flow_category": "common",
  "flow_data": {
    "nodes": [
      {
        "id": "resource_http_body_0",
        "type": "resource",
        "data": {
          "flow_type": "resource",
          "type_cls": "dbgpt.core.awel.resource.http_body.HTTPBodyResource",
          "label": "HTTP Body"
        },
        "width": 200,
        "height": 80,
        "position": {"x": 100, "y": 100, "zoom": 1}
      },
      {
        "id": "operator_upper_case_0",
        "type": "operator",
        "data": {
          "flow_type": "operator",
          "type_cls": "my_operator.UpperCaseOperator",
          "label": "Upper-Case"
        },
        "width": 200,
        "height": 80,
        "position": {"x": 400, "y": 100, "zoom": 1}
      }
    ],
    "edges": [
      {
        "source": "resource_http_body_0",
        "target": "operator_upper_case_0",
        "source_order": 0,
        "target_order": 0,
        "id": "e0"
      }
    ],
    "viewport": {"x": 0, "y": 0, "zoom": 1}
  }
}

The type_cls field must contain the fully-qualified Python path to your operator class.

Step 4: Load and Execute the DAG

Programmatically build and run the workflow:

import json
from dbgpt.core.awel.flow.flow_factory import FlowFactory
from dbgpt.core.awel.flow.base import FlowPanel

# Load the JSON description

with open("upper_flow.json") as f:
    panel_dict = json.load(f)

panel = FlowPanel(**panel_dict)

# Build the DAG

factory = FlowFactory()
dag = factory.build(panel)

# Execute (async or sync)

result_ctx = await dag.run(initial_data={"text": "hello world"})

# Or synchronously: result_ctx = dag.run_sync(initial_data={"text": "hello world"})

print(result_ctx.output)  # → "HELLO WORLD"

The dag.run() method creates a DAGContext, injects initial data into source nodes, and walks the graph using the DefaultWorkflowRunner.

Extending the Workflow Runner

For distributed execution environments, subclass WorkflowRunner and override execute_workflow():

from dbgpt.core.awel.operators.base import WorkflowRunner, DAGContext

class CeleryRunner(WorkflowRunner):
    async def execute_workflow(self, node, call_data=None, streaming_call=False,
                              exist_dag_ctx=None, dag_variables=None) -> DAGContext:
        # Serialize node and submit to Celery workers

        pass

# Register globally

from dbgpt.core.awel.operators.base import default_runner
default_runner = CeleryRunner()

Summary

• AWEL represents DB-GPT workflows as JSON-serializable FlowPanel objects that describe DAGs of operators and resources.
• The FlowFactory in packages/dbgpt-core/src/dbgpt/core/awel/flow/flow_factory.py converts these panels into executable DAGs via topological sorting and dependency injection.
• Operators inherit from BaseOperator (commonly MapOperator, BranchOperator, or JoinOperator) and implement abstract methods like map() or reduce().
• Resources provide external dependencies and are automatically instantiated and wired into operators during the build phase.
• Custom workflows require defining an operator class, attaching metadata for UI/validation, constructing a FlowPanel JSON, and executing via factory.build() and dag.run().

Frequently Asked Questions

How do I debug a failing AWEL workflow?

Inspect the built DAG Python object using print(dag) to view node IDs and downstream relationships. You can also copy the FlowPanel JSON to the DB-GPT web UI at /awel/flows to visualize the graph and validate connections. Check the FlowFactory logs during the build phase to catch validation errors in node metadata or missing resource dependencies.

Can I trigger AWEL workflows via HTTP endpoints?

Yes. Deploy the built-in HTTPTrigger from packages/dbgpt-core/src/dbgpt/core/awel/trigger/http_trigger.py, which automatically exposes flows at /awel/trigger/<flow_id>. When a POST request hits this endpoint, the trigger creates a temporary DAG instance with the request body as initial data and executes it through the default runner.

What is the difference between dag.run() and dag.run_sync()?

dag.run() executes the workflow asynchronously, allowing non-blocking execution of I/O-bound operators like LLM calls or database queries. dag.run_sync() provides a synchronous wrapper for simple scripts or environments without an async event loop. Both methods accept initial_data to seed input into source nodes and return a DAGContext containing the final output.

How do I share data between operators in a workflow?

Data flows implicitly through the DAG edges defined in the FlowPanel JSON. Each operator receives its input from upstream node outputs and passes results to downstream nodes via the map(), branch(), or reduce() return values. For flow-level shared state, define variables in the variables field of the FlowPanel, which are accessible to all operators during execution.

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