从 HTTP 触发器到 DAG 编排：DB-GPT AWEL 工作流快速上手指南

从 HTTP 触发器到 DAG 编排：DB-GPT AWEL 工作流快速上手指南

【免费下载链接】DB-GPTopen-source agentic AI data assistant for the next generation of AI + Data products.项目地址: https://gitcode.com/GitHub_Trending/db/DB-GPT

本文基于 DB-GPT 仓库中的 AWEL（Agentic Workflow Expression Language）入门文档，完整讲解如何用 HTTP 触发器 + 自定义算子搭建一条最小可运行的工作流：你将学会定义请求体模型、编写自定义 MapOperator、用>>语法组装 DAG，并掌握“生产模式挂服务”与“开发模式本地调试”两种验证方式，最终能把任意 AWEL 编排以 REST 接口形式暴露出来。

AWEL 是什么：为 LLM 应用设计的智能体工作流语言

AWEL（Agentic Workflow Expression Language，智能体工作流表达语言）是 DB-GPT 专门为大模型应用开发设计的编排框架。其官方包文档给出了清晰的定位描述：

AWEL is a set of intelligent agent workflow expression language specially designed for large model application development. It provides great functionality and flexibility. Through the AWEL API, you can focus on the development of business logic for LLMs applications without paying attention to cumbersome model and environment details.

从源码包 awel/init.py 的导出列表可以确认，AWEL 对外暴露的核心构件分为四类：

• DAG 编排层：DAG、DAGContext、DAGVar，定义任务间的依赖关系；
• 算子层（Operator）：BaseOperator、MapOperator、JoinOperator、BranchOperator、ReduceStreamOperator、InputOperator等，对应不同数据流转模式（一对一映射、分支、归并、流式聚合）；
• 触发器层（Trigger）：HttpTrigger、IteratorTrigger，以及可选依赖的RequestHttpTrigger（awel/init.py 中通过 try/except 做可选导入，缺少 starlette 依赖时自动降级）；
• 执行层（Runner）：DefaultWorkflowRunner及TaskContext、TaskOutput、InputSource等任务执行抽象。

理解这四层后，下面这个入门示例就顺理成章了：一个 AWEL 应用 = 触发器（接收输入）+ 若干算子（处理数据）+ DAG（描述流转顺序）+ Runner（负责调度执行）。

示例目标：HTTP 请求 + 输出改写

入门示例的核心功能是“处理一个 HTTP 请求的输入并改写其输出”，因此整条编排只包含两步：

1. HTTP Request：接收请求；
2. Processing HTTP Response Result：处理请求体，返回改写后的结果。

DB-GPT 已将上述依赖的基础算子封装好，可直接引用：

from dbgpt._private.pydantic import BaseModel, Field from dbgpt.core.awel import DAG, HttpTrigger, MapOperator

注意BaseModel与Field的导入路径是dbgpt._private.pydantic而非直接pydantic——这是 DB-GPT 对 pydantic v1/v2 做版本兼容适配的做法，写 AWEL 算子时应统一使用这个路径以保证行为一致。

第一步：定义请求体模型

定义一个接受name和age两个参数的 HTTP 请求体：

class TriggerReqBody(BaseModel): name: str = Field(..., description="User name") age: int = Field(18, description="User age")

这个模型将作为HttpTrigger的request_body参数传入，框架会用它做请求体解析与校验：name为必填字段（...），age缺省时默认取18。

第二步：自定义算子 RequestHandleOperator

定义一个请求处理算子RequestHandleOperator，它继承基础MapOperator并做泛型特化：MapOperator[TriggerReqBody, str]声明输入类型为TriggerReqBody、输出类型为str。算子动作非常直接——解析请求体，取出 name 与 age 字段，拼接成一句话，例如：

"Hello, zhangsan, your age is 18."

class RequestHandleOperator(MapOperator[TriggerReqBody, str]): def __init__(self, **kwargs): super().__init__(**kwargs) async def map(self, input_value: TriggerReqBody) -> str: print(f"Receive input value: {input_value}") return f"Hello, {input_value.name}, your age is {input_value.age}"

这里有两个值得注意的设计点：

• map是异步方法：AWEL 的执行链路整体基于 asyncio，算子的核心处理方法声明为async，Runner 会在事件循环中调度它，这使得同一 DAG 内可以并行处理多个请求；
• 泛型参数即接口契约：MapOperator[I, O]的输入/输出类型在类定义处就确定了，DAG 组装时上下游节点的数据类型匹配关系在编写期就能被表达出来，而不是等到运行期才发现类型不匹配。

第三步：用 DAG 上下文组装管道

写完算子后，用 DAG 上下文把它们装配成编排图。这条 DAG 共两个节点：第一个是内置的HttpTrigger（负责处理 HTTP 请求），第二个是自定义的RequestHandleOperator（处理请求体）：

with DAG("simple_dag_example") as dag: trigger = HttpTrigger("/examples/hello", request_body=TriggerReqBody) map_node = RequestHandleOperator() trigger >> map_node

with DAG(...)是 DAG 的上下文管理器写法，dag变量保存了整个编排图实例（后文开发模式中会用到）。HttpTrigger的第一个参数是相对端点路径/examples/hello，request_body指定请求体解析模型。

关于trigger >> map_node这条依赖边，源码层面有明确的实现依据：dag/base.py 中的DependencyMixin定义了set_upstream/set_downstream接口，并重载了移位运算符——node >> next_node实际调用node.set_downstream(next_node)，node << input_node则调用set_upstream，且都支持传入节点列表以一次声明多条依赖。也就是说，>>只是“设置下游节点”的语法糖，DAG 内部维护的是显式的上下游依赖关系图，Runner 据此决定执行顺序与并行度。

访问验证：两种运行模式

示例提供了两种验证路径：生产模式（挂到 DB-GPT 服务端）与开发模式（本地独立调试），分别对应不同的端口与启动方式。

生产模式：随 dbgpt_server 启动

进行访问验证前，需要先启动项目服务：

python dbgpt/app/dbgpt_server.py

（文档写作时的入口路径为dbgpt/app/dbgpt_server.py；在当前仓库的 monorepo 布局下，服务端代码位于packages/dbgpt-app/src/dbgpt_app/目录，其中 dbgpt_app/_cli.py 中的from dbgpt_app.dbgpt_server import run_webserver表明dbgpt_server模块仍然存在，可通过 CLI 的 start 子命令启动。）

服务启动后，DAG 定义文件会被自动加载（AWEL 的DAGManager启动时扫描配置的 DAG 目录并注册触发器节点），然后即可用 curl 验证：

curl -X GET http://127.0.0.1:5670/api/v1/awel/trigger/examples/hello\?name\=zhangsan "Hello, zhangsan, your age is 18"

这个 URL 的三段式结构是 AWEL 触发器路由的通用形态，各段含义在源码中均有对应：

• http://127.0.0.1:5670：DB-GPT 服务端默认监听地址；
• /api/v1/awel/trigger：触发器路由前缀。在 trigger_manager.py 中，HttpTriggerManager的构造参数router_prefix默认值就是"/api/v1/awel/trigger"，注册时会把前缀与触发器端点拼接成完整路径；
• examples/hello：HttpTrigger("/examples/hello", ...)中声明的相对端点。

参数name=zhangsan通过查询字符串传入，age缺省，由TriggerReqBody的默认值补全为 18，最终响应即为算子map方法拼接出的字符串。

开发模式：不启动服务直接调试

为了让用户更方便地测试，AWEL 提供了无需启动完整 dbgpt_server 的开发环境。在 DAG 定义之后追加如下代码：

if __name__ == "__main__": if dag.leaf_nodes[0].dev_mode: # Development mode, you can run the dag locally for debugging. from dbgpt.core.awel import setup_dev_environment setup_dev_environment([dag], port=5555) else: # Production mode, DB-GPT will automatically load and execute the current file after startup. pass

然后直接运行python examples/awel/simple_dag_example.py，在不启动项目的情况下测试：

curl -X GET http://127.0.0.1:5555/api/v1/awel/trigger/examples/hello\?name\=zhangsan "Hello, zhangsan, your age is 18"

从 awel/init.py 的setup_dev_environment实现看，它的签名与行为如下：

def setup_dev_environment( dags: List[DAG], host: str = "127.0.0.1", port: int = 5555, logging_level: Optional[str] = None, logger_filename: Optional[str] = None, show_dag_graph: Optional[bool] = True, ) -> None

• dags：待运行的 DAG 列表，示例传的是[dag]；
• host/port：开发服务器的绑定地址，默认127.0.0.1:5555，所以示例中显式传port=5555其实与默认值一致；
• logging_level/logger_filename：日志级别与日志文件名，未指定时默认写入dbgpt_awel_dev.log；
• show_dag_graph：默认True，会调用dag.visualize_dag()把 DAG 图保存为文件并尝试自动打开；若系统未安装 graphviz，会降级为 warning 提示安装（pip install graphviz或sudo apt install graphviz），不会中断运行。

其内部执行流程是：创建（含 HTTP 触发器时）一个 FastAPI 应用 → 构建SystemApp并设置DAGVar上下文 → 创建DefaultTriggerManager并把每个 DAG 的trigger_nodes逐个register_trigger→ 最后用 uvicorn 启动 HTTP 服务。也就是说，开发模式本质上是在本地起一个“精简版 DB-GPT 服务”，只挂载你传入的这些 DAG，路由前缀同样是/api/v1/awel/trigger，因此两个模式下的 curl 命令结构完全相同，只是端口从 5670 换成了 5555。

另外注意if dag.leaf_nodes[0].dev_mode这个分支判断：同一个 DAG 定义文件同时兼容两种模式——由脚本直接运行（开发调试）时走setup_dev_environment；由 DB-GPT 服务端加载执行时（生产模式）则什么都不做，交给服务端的 DAG 加载机制接管。仓库中对应的完整示例文件是 examples/awel/simple_dag_example.py，其文件头 docstring 中给出的调用示例与上文 curl 命令一致，且仓库中已存在可直接查看的 examples/awel/simple_dag_example.py 文件，与文档代码完全对应。

生产模式下 DAG 是如何被自动加载的

文档中提到“DB-GPT 启动后会自动加载执行当前文件”，这一行为背后的调用链在源码中可以完整确认：

1. 服务初始化时，component_configs.py 中的_initialize_awel(system_app, web_config.awel_dirs)被调用：它先取内置的_DAG_DEFINITION_DIR作为基础 DAG 目录，再追加配置文件awel_dirs中指定的目录（逗号分隔），最后调用initialize_awel(system_app, dag_dirs)；
2. initialize_awel 做了三件事：绑定DAGVar的 SystemApp 上下文、注册DefaultTriggerManager组件、创建DAGManager(system_app, dag_dirs)并注册为系统实例，最后initialize_runner(DefaultWorkflowRunner())安装默认执行器；
3. DAGManager 内部使用LocalFileDAGLoader(dag_dirs)扫描指定目录下的 DAG 定义文件（loader.py），加载后注册其中声明的触发器节点；
4. 触发器注册时，HttpTriggerManager.register_trigger 会把router_prefix（/api/v1/awel/trigger）与触发器的真实端点拼接成完整路由，挂载到 FastAPI 应用/路由上，并维护路由表防止路径冲突。

因此“把 DAG 文件放进配置的awel_dirs目录 + 重启服务”就是在生产环境上线一条 AWEL 工作流的标准操作；而从HttpTrigger支持methods、http_response_body、streaming_response等参数（见 http_trigger.py 中HttpTriggerMetadata与请求体体系BaseHttpBody/DictHttpBody/StringHttpBody）可以看出，该触发器机制同样支撑 POST、流式响应等更复杂的 API 形态。

小结与延伸

本文沿 AWEL 入门文档的脉络走完了最小闭环：

1. 用BaseModel定义请求体TriggerReqBody（name 必填、age 默认 18）；
2. 用MapOperator[TriggerReqBody, str]派生自定义算子RequestHandleOperator，核心逻辑写在async def map中；
3. 用with DAG(...) as dag:上下文把HttpTrigger与算子用>>连成两节点 DAG；
4. 生产模式下随 dbgpt_server 启动后访问http://127.0.0.1:5670/api/v1/awel/trigger/examples/hello?name=zhangsan；开发模式下用setup_dev_environment([dag], port=5555)在本地 5555 端口独立验证，二者响应一致。

掌握这个模式后，可以沿着以下仓库入口继续深入 AWEL 的能力边界：完整示例脚本 examples/awel/simple_dag_example.py、AWEL 核心包 packages/dbgpt-core/src/dbgpt/core/awel/、触发器实现 packages/dbgpt-core/src/dbgpt/core/awel/trigger/、DAG 基类与依赖关系实现 packages/dbgpt-core/src/dbgpt/core/awel/dag/base.py，以及官方文档中的进阶章节 docs/docs/awel/awel.md 与 docs/docs/awel/why_use_awel.md，其中涵盖了分支算子（BranchOperator）、流式算子（StreamifyAbsOperator等）和多轮会话场景的编排方法。

【免费下载链接】DB-GPTopen-source agentic AI data assistant for the next generation of AI + Data products.项目地址: https://gitcode.com/GitHub_Trending/db/DB-GPT