# How DB-GPT AWEL Works: A Complete Guide to Creating Custom Workflows > Discover how DB-GPT AWEL, its powerful workflow engine, creates custom LLM pipelines as directed-acyclic graphs. Learn to build and execute complex workflows. - Repository: [eosphoros/DB-GPT](https://github.com/eosphoros-ai/db-gpt) - Tags: deep-dive - Published: 2026-02-23 --- **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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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: ```python # 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: ```python # 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: ```json { "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: ```python 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()`: ```python 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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/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`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-core/src/dbgpt/core/awel/trigger/http_trigger.py), which automatically exposes flows at `/awel/trigger/`. 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.